"""services/team_admin — the ADMIN → Team page backend (M7).

This is the R1 rebuild of the legacy team_members CRUD
(01-current-system/features/organizations-and-team.md), mapped onto the
party/engagement/compensation-term model (02-domain-model.md §1–2):

  legacy team_member  ->  a PERSON party
                          + a subcontract worker engagement (tenant party
                            -> person), carrying can_manage_imports and the
                            consolidate_as import-mapping label
                          + two compensation_terms (kind cost / billout) on
                            that engagement — the R1 worker-terms model

The load-bearing invariant is **append-only rate history** (02 §2; 02 §5
invariant 1). A rate change never mutates the in-force term's amount:
it ENDS the current term (stamps effective_to) and starts a NEW term
effective-dated from the change. Historical stamps on time_entries are
NEVER touched here — re-stamping existing entries is an explicit action
that lives on /entries (M5 restamp), deliberately NOT offered on this
page. So a rate change alters what the NEXT import/stamp resolves, and
nothing else.

Permissions: every read and write gates on `can_manage_team` — the
EFFECTIVE rule (02 §4): owner or super-admin (not while viewing-as). The
dead legacy `canManageTeam` any-manager grant is NOT the parity bar;
per-team-owner grants are resource-scoped and land in R2. Billout-rate
fields are additionally gated on `can_transfer` (the owner's private
number — Adi must never see it; the page-level gate already implies it
in R1, but the payload honors the finer rule for R2 forward-safety).
"""

from __future__ import annotations

import re
import uuid
from datetime import date, timedelta
from decimal import Decimal, InvalidOperation

from sqlalchemy import func, select
from sqlalchemy.orm import Session, aliased

from app.models import (
    CompensationTerm,
    Engagement,
    Party,
    Project,
    RateOverride,
    Tenant,
    TimeEntry,
)
from app.services.audit import write_audit
from app.services.entities.parties import create_person_party
from app.services.entities.resolve import slugify
from app.services.errors import ServiceError
from app.services.money.stamping import WORKER_ENGAGEMENT_TYPES

#: The R1 worker engagement created for a new team member. subcontract
#: grants the worker role (02 §4); freelance is the other worker type but
#: the Team page only ever CREATES subcontracts (parity with the legacy
#: team-owner-adds-a-worker flow).
NEW_WORKER_ENGAGEMENT_TYPE = "subcontract"

#: Email shape gate, verbatim from the legacy server action (B39).
_EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$")


# --------------------------------------------------------------- gates


def _require_manage(actor, verb: str) -> None:
    if not actor.allowed("can_manage_team"):
        raise ServiceError(
            403, "NOT_ALLOWED", f"You do not have permission to {verb}."
        )


def _require_tenant(actor) -> uuid.UUID:
    if actor.tenant_id is None:
        raise ServiceError(400, "NO_TENANT", "No tenant context resolved.")
    return actor.tenant_id


def _tenant_party_id(session: Session, tenant_id: uuid.UUID) -> uuid.UUID:
    party_id = session.scalar(select(Tenant.party_id).where(Tenant.id == tenant_id))
    if party_id is None:
        raise ServiceError(400, "NO_TENANT", "Tenant has no party.")
    return party_id


def _tenant_currency(session: Session, tenant_id: uuid.UUID) -> str:
    settings = session.scalar(select(Tenant.settings).where(Tenant.id == tenant_id))
    if isinstance(settings, dict):
        code = settings.get("default_currency")
        if isinstance(code, str) and code.strip():
            return code.strip().upper()
    return "CAD"  # 07 #21 all-CAD default


# --------------------------------------------------------------- parsing


def _clean_text(raw: str | None) -> str | None:
    if raw is None:
        return None
    trimmed = raw.strip()
    return trimmed or None


def _clean_email(raw: str | None) -> str:
    email = (raw or "").strip().lower()
    if not _EMAIL_RE.match(email):
        raise ServiceError(422, "BAD_EMAIL", "A valid email is required.")
    return email


def parse_rate(raw, field_label: str) -> Decimal | None:
    """A money rate: blank/None -> None (NULL = never priced, judgment
    #19); otherwise a non-negative number rounded to cents. 0 is a real
    zero-rate (Gary). Mirrors the legacy cost/billout parse (B40)."""
    if raw is None:
        return None
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return None
    try:
        value = Decimal(str(raw))
    except (InvalidOperation, ValueError) as exc:
        raise ServiceError(
            422,
            "BAD_RATE",
            f"{field_label} must be a non-negative number or blank.",
        ) from exc
    if not value.is_finite() or value < 0:
        raise ServiceError(
            422,
            "BAD_RATE",
            f"{field_label} must be a non-negative number or blank.",
        )
    return value.quantize(Decimal("0.01"))


# --------------------------------------------------------------- lookups


def _allocate_slug(session: Session, name: str) -> str:
    base = slugify(name)
    slug, n = base, 1
    while session.scalar(select(Party.id).where(Party.slug == slug)) is not None:
        n += 1
        slug = f"{base}-{n}"
    return slug


def _worker_engagement(
    session: Session, tenant_id: uuid.UUID, party_id: uuid.UUID, *, active_only: bool
) -> Engagement | None:
    """The person's worker engagement in the tenant (active preferred /
    required). Earliest-created active wins, matching stamping.resolve_worker."""
    stmt = select(Engagement).where(
        Engagement.tenant_id == tenant_id,
        Engagement.party_b_id == party_id,
        Engagement.type.in_(WORKER_ENGAGEMENT_TYPES),
    )
    if active_only:
        stmt = stmt.where(Engagement.ended_on.is_(None))
    return session.scalars(stmt.order_by(Engagement.created_at)).first()


def _person_by_email(session: Session, email: str) -> Party | None:
    return session.scalars(
        select(Party).where(
            Party.kind == "person", func.lower(Party.email) == email
        )
    ).first()


def _open_term(
    session: Session,
    engagement_id: uuid.UUID,
    kind: str,
    project_id: uuid.UUID | None = None,
) -> CompensationTerm | None:
    """The engagement-default term currently open (effective_to IS NULL)
    for a kind. Latest effective_from wins among any survivors — there
    should only ever be one open row per (engagement, kind, project)."""
    terms = (
        session.scalars(
            select(CompensationTerm).where(
                CompensationTerm.engagement_id == engagement_id,
                CompensationTerm.kind == kind,
                CompensationTerm.effective_to.is_(None),
                (
                    CompensationTerm.project_id == project_id
                    if project_id is not None
                    else CompensationTerm.project_id.is_(None)
                ),
            )
        )
        .unique()
        .all()
    )
    if not terms:
        return None
    return max(terms, key=lambda t: t.effective_from)


# --------------------------------------------------------------- term history (append-only)


def change_rate(
    session: Session,
    engagement_id: uuid.UUID,
    kind: str,
    new_rate: Decimal | None,
    currency: str,
    *,
    effective_from: date | None = None,
    project_id: uuid.UUID | None = None,
) -> str | None:
    """Append-only rate change (02 §2, invariant 1). Returns a short
    disposition string, or None when nothing changed.

    - new_rate is None -> clear the rate: END the open term, add nothing
      (NULL = never priced; the NEXT stamp resolves to no rate).
    - open term already at (new_rate, currency) -> no-op.
    - a change effective TODAY (or later) -> stamp the open term's
      effective_to = effective_from - 1 day and INSERT a fresh term.
    - a same-day correction (effective_from <= the open term's own
      start, so no historical stamp could have resolved through it) ->
      update the open term's amount in place; still append-only w.r.t.
      HISTORY, which is the actual invariant.

    NEVER touches time_entries — existing stamps are frozen; re-stamping
    is the explicit /entries action (M5), deliberately not offered here.
    """
    on = effective_from or date.today()
    open_term = _open_term(session, engagement_id, kind, project_id)

    if new_rate is None:
        if open_term is None:
            return None
        open_term.effective_to = on - timedelta(days=1)
        if open_term.effective_to < open_term.effective_from:
            # never priced beyond a single (today) day -> drop it entirely
            session.delete(open_term)
        return "cleared"

    if open_term is not None:
        same = (
            open_term.rate_amount is not None
            and Decimal(open_term.rate_amount) == new_rate
            and (open_term.currency or "").upper() == currency.upper()
        )
        if same:
            return None
        if on <= open_term.effective_from:
            # same-day (or backdated-to-start) correction: no history
            # resolves through this term yet — amend in place.
            open_term.rate_amount = new_rate
            open_term.currency = currency
            return "corrected"
        open_term.effective_to = on - timedelta(days=1)

    session.add(
        CompensationTerm(
            engagement_id=engagement_id,
            kind=kind,
            model="hourly",
            rate_amount=new_rate,
            currency=currency,
            project_id=project_id,
            effective_from=on,
        )
    )
    return "opened"


# --------------------------------------------------------------- payload


def _entry_counts(session: Session, tenant_id: uuid.UUID) -> dict[uuid.UUID, int]:
    rows = session.execute(
        select(TimeEntry.worker_party_id, func.count())
        .where(
            TimeEntry.tenant_id == tenant_id,
            TimeEntry.worker_party_id.is_not(None),
        )
        .group_by(TimeEntry.worker_party_id)
    ).all()
    return dict(rows)


def _term_dto(term: CompensationTerm | None) -> dict | None:
    if term is None:
        return None
    return {
        "rate": str(term.rate_amount) if term.rate_amount is not None else None,
        "currency": term.currency,
        "effective_from": term.effective_from.isoformat(),
    }


def list_workers(session: Session, actor) -> dict:
    """The Team page payload: one row per worker engagement in the tenant
    with the person's identity, active status, current cost/billout terms
    (billout owner-only), entry count, and the consolidate_as label.
    Plus the project-scoped rate-override rules and the project catalog
    for the override editor."""
    _require_manage(actor, "manage the team")
    tenant_id = _require_tenant(actor)
    can_see_billout = actor.allowed("can_transfer")

    counts = _entry_counts(session, tenant_id)

    rows = session.execute(
        select(Engagement, Party)
        .join(Party, Party.id == Engagement.party_b_id)
        .where(
            Engagement.tenant_id == tenant_id,
            Engagement.type.in_(WORKER_ENGAGEMENT_TYPES),
            Party.kind == "person",
        )
        .order_by(Party.name)
    ).all()

    workers = []
    for eng, party in rows:
        cost = _open_term(session, eng.id, "cost")
        billout = _open_term(session, eng.id, "billout")
        worker = {
            "engagement_id": str(eng.id),
            "party_id": str(party.id),
            "name": party.name,
            "email": party.email,
            "active": eng.ended_on is None,
            "ended_on": eng.ended_on.isoformat() if eng.ended_on else None,
            "consolidate_as": eng.consolidate_as,
            "can_manage_imports": eng.can_manage_imports,
            "notes": eng.notes,
            "entries": counts.get(party.id, 0),
            "cost_term": _term_dto(cost),
        }
        if can_see_billout:
            worker["billout_term"] = _term_dto(billout)
        workers.append(worker)

    # active first, then name (matches the legacy is_active DESC, name ASC)
    workers.sort(key=lambda w: (not w["active"], w["name"].lower()))

    return {
        "workers": workers,
        "rate_overrides": _list_overrides(session, tenant_id, can_see_billout),
        "projects": _project_catalog(session, tenant_id),
        "can_manage": True,
        "can_see_billout": can_see_billout,
    }


# --------------------------------------------------------------- add / edit worker


def add_worker(session: Session, actor, data: dict) -> dict:
    """Create a person party (or reuse the existing one for this email —
    parties are never duplicated) + a subcontract worker engagement +
    cost/billout terms. Rejects an email that is already an ACTIVE worker
    (the R1 tenant-wide uniqueness; teams dissolved so it is no longer
    per-team — T-TEAM-01)."""
    _require_manage(actor, "add team members")
    tenant_id = _require_tenant(actor)
    tenant_party_id = _tenant_party_id(session, tenant_id)

    name = _clean_text(data.get("name"))
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "Display name is required.")
    email = _clean_email(data.get("email"))
    currency = (_clean_text(data.get("currency")) or _tenant_currency(session, tenant_id)).upper()
    cost_rate = parse_rate(data.get("cost_rate"), "Cost rate")
    billout_rate = parse_rate(data.get("billout_rate"), "Billout rate")
    consolidate_as = _clean_text(data.get("consolidate_as"))
    notes = _clean_text(data.get("notes"))
    can_manage_imports = bool(data.get("can_manage_imports"))

    existing = _person_by_email(session, email)
    if existing is not None:
        active = _worker_engagement(session, tenant_id, existing.id, active_only=True)
        if active is not None:
            raise ServiceError(
                409, "ALREADY_MEMBER", f"{email} is already a team member."
            )
        party = existing
        party.name = name  # keep display name current on re-add
        if notes and not party.notes:
            party.notes = notes
    else:
        party = create_person_party(
            session,
            name=name,
            slug=_allocate_slug(session, name),
            email=email,
            notes=notes,
        )

    engagement = Engagement(
        tenant_id=tenant_id,
        type=NEW_WORKER_ENGAGEMENT_TYPE,
        party_a_id=tenant_party_id,
        party_b_id=party.id,
        role_a="prime",
        role_b="subcontractor",
        can_manage_imports=can_manage_imports,
        consolidate_as=consolidate_as,
        notes=notes,
    )
    session.add(engagement)
    session.flush()

    for kind, rate in (("cost", cost_rate), ("billout", billout_rate)):
        change_rate(session, engagement.id, kind, rate, currency)

    write_audit(
        session,
        actor,
        action="team.add_worker",
        entity_type="engagement",
        entity_id=engagement.id,
        after={
            "party_id": str(party.id),
            "name": name,
            "email": email,
            "cost_rate": str(cost_rate) if cost_rate is not None else None,
            "billout_rate": str(billout_rate) if billout_rate is not None else None,
        },
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Added {name} to the team.',
        "party_id": str(party.id),
        "engagement_id": str(engagement.id),
    }


def update_worker(session: Session, actor, party_id: uuid.UUID, patch: dict) -> dict:
    """Field-level edit. Identity/label fields (name, email,
    consolidate_as, can_manage_imports, notes) update directly; cost_rate
    / billout_rate — WHEN PRESENT in the patch — go through the
    append-only term machinery (a manager-render form omitting billout
    must not clobber the stored rate — T-TEAM-06). Existing entry stamps
    are never touched (re-stamp is a separate /entries action)."""
    _require_manage(actor, "edit team members")
    tenant_id = _require_tenant(actor)

    engagement = _worker_engagement(session, tenant_id, party_id, active_only=False)
    if engagement is None:
        raise ServiceError(404, "WORKER_NOT_FOUND", "Team member not found.")
    party = session.get(Party, party_id)

    before = {
        "name": party.name,
        "email": party.email,
        "consolidate_as": engagement.consolidate_as,
        "can_manage_imports": engagement.can_manage_imports,
        "notes": engagement.notes,
    }
    changed_terms: list[str] = []

    if "name" in patch:
        name = _clean_text(patch.get("name"))
        if not name:
            raise ServiceError(422, "NAME_REQUIRED", "Display name is required.")
        party.name = name
    if "email" in patch:
        new_email = _clean_email(patch.get("email"))
        if new_email != (party.email or "").lower():
            clash = _person_by_email(session, new_email)
            if clash is not None and clash.id != party.id:
                raise ServiceError(
                    409, "EMAIL_IN_USE", f"{new_email} is already used by another person."
                )
            party.email = new_email
    if "consolidate_as" in patch:
        engagement.consolidate_as = _clean_text(patch.get("consolidate_as"))
    if "notes" in patch:
        engagement.notes = _clean_text(patch.get("notes"))
    if "can_manage_imports" in patch:
        engagement.can_manage_imports = bool(patch.get("can_manage_imports"))

    currency = (
        _clean_text(patch.get("currency")) or _tenant_currency(session, tenant_id)
    ).upper()
    if "cost_rate" in patch:
        disp = change_rate(
            session,
            engagement.id,
            "cost",
            parse_rate(patch.get("cost_rate"), "Cost rate"),
            currency,
        )
        if disp:
            changed_terms.append("cost")
    if "billout_rate" in patch:
        # only when the owner-visible field was actually submitted (B40)
        disp = change_rate(
            session,
            engagement.id,
            "billout",
            parse_rate(patch.get("billout_rate"), "Billout rate"),
            currency,
        )
        if disp:
            changed_terms.append("billout")

    write_audit(
        session,
        actor,
        action="team.update_worker",
        entity_type="engagement",
        entity_id=engagement.id,
        before=before,
        after={
            "name": party.name,
            "email": party.email,
            "consolidate_as": engagement.consolidate_as,
            "can_manage_imports": engagement.can_manage_imports,
            "notes": engagement.notes,
            "rate_changes": changed_terms,
        },
    )
    session.commit()
    note = ""
    if changed_terms:
        note = (
            f" New {' and '.join(changed_terms)} rate"
            f"{'s' if len(changed_terms) > 1 else ''} take effect on the next import"
            " — existing entries keep their stamps."
        )
    return {
        "ok": True,
        "changed": True,
        "info": f"Updated {party.name}.{note}",
        "rate_changes": changed_terms,
    }


def set_active(session: Session, actor, party_id: uuid.UUID, *, active: bool) -> dict:
    """Deactivate = END the worker engagement (ended_on = today); the
    person party, its terms (incl. Gary's cost-0/billout-25 shape), and
    every historical entry stay put — parties are never hard-deleted
    (rebuild delta; fixes the legacy SET-NULL orphaning wart). An ended
    worker no longer resolves on import. Reactivate clears ended_on."""
    verb = "deactivate team members" if not active else "reactivate team members"
    _require_manage(actor, verb)
    tenant_id = _require_tenant(actor)

    engagement = _worker_engagement(session, tenant_id, party_id, active_only=False)
    if engagement is None:
        raise ServiceError(404, "WORKER_NOT_FOUND", "Team member not found.")
    party = session.get(Party, party_id)

    already = (engagement.ended_on is None) == active
    if already:
        state = "active" if active else "inactive"
        return {"ok": True, "changed": False, "info": f"{party.name} is already {state}."}

    before = {"ended_on": engagement.ended_on.isoformat() if engagement.ended_on else None}
    engagement.ended_on = None if active else date.today()

    write_audit(
        session,
        actor,
        action="team.reactivate_worker" if active else "team.deactivate_worker",
        entity_type="engagement",
        entity_id=engagement.id,
        before=before,
        after={"ended_on": engagement.ended_on.isoformat() if engagement.ended_on else None},
    )
    session.commit()
    verbed = "Reactivated" if active else "Deactivated"
    return {"ok": True, "changed": True, "info": f"{verbed} {party.name}."}


# --------------------------------------------------------------- rate overrides


def _project_catalog(session: Session, tenant_id: uuid.UUID) -> list[dict]:
    op = aliased(Party, name="op")
    cp = aliased(Party, name="cp")
    rows = session.execute(
        select(Project.id, Project.name, op.name, cp.name)
        .select_from(Project)
        .outerjoin(op, op.id == Project.operator_party_id)
        .outerjoin(cp, cp.id == Project.client_party_id)
        .where(Project.tenant_id == tenant_id)
        .order_by(Project.name)
    ).all()
    return [
        {
            "id": str(pid),
            "name": name,
            "operator": operator,
            "client": client,
            "label": f"{client} : {name}" if client else name,
        }
        for pid, name, operator, client in rows
    ]


def _list_overrides(
    session: Session, tenant_id: uuid.UUID, can_see_billout: bool
) -> list[dict]:
    """Project-scoped billout override rules (judgment #25). Owner-only —
    an override IS a billout number. Non-owners get an empty list."""
    if not can_see_billout:
        return []
    client = aliased(Party, name="proj_client")
    worker = aliased(Party, name="ovr_worker")
    rows = session.execute(
        select(RateOverride, Project.name, client.name, worker.name)
        .select_from(RateOverride)
        .join(Project, Project.id == RateOverride.project_id)
        .outerjoin(client, client.id == Project.client_party_id)
        .outerjoin(worker, worker.id == RateOverride.worker_party_id)
        .where(RateOverride.tenant_id == tenant_id)
        .order_by(Project.name)
    ).all()
    out = []
    for override, project_name, client_name, worker_name in rows:
        out.append(
            {
                "id": str(override.id),
                "project_id": str(override.project_id),
                "project": project_name,
                "client": client_name,
                "worker_party_id": (
                    str(override.worker_party_id) if override.worker_party_id else None
                ),
                "worker": worker_name,  # None => project-wide (all workers)
                "override_rate": (
                    str(override.override_rate) if override.override_rate is not None else None
                ),
                "override_pct": (
                    str(override.override_pct) if override.override_pct is not None else None
                ),
                "currency": override.currency,
                "notes": override.notes,
            }
        )
    return out


def _validate_override_kind(
    override_rate, override_pct, currency: str | None
) -> tuple[Decimal | None, Decimal | None, str | None]:
    """Absolute rate XOR signed pct (carried verbatim from the legacy
    table). An absolute rate carries a currency; a pct never does (the
    currency-pair CHECK)."""
    has_rate = override_rate is not None and str(override_rate).strip() != ""
    has_pct = override_pct is not None and str(override_pct).strip() != ""
    if has_rate == has_pct:
        raise ServiceError(
            422,
            "OVERRIDE_ONE_KIND",
            "Give either an absolute rate OR a percentage adjustment — not both, not neither.",
        )
    if has_rate:
        rate = parse_rate(override_rate, "Override rate")
        code = (currency or "").strip().upper()
        if not code:
            raise ServiceError(422, "CURRENCY_REQUIRED", "An absolute rate needs a currency.")
        return rate, None, code
    try:
        pct = Decimal(str(override_pct)).quantize(Decimal("0.0001"))
    except (InvalidOperation, ValueError) as exc:
        raise ServiceError(422, "BAD_PCT", "Percentage must be a number (e.g. -20).") from exc
    # T-ADJ-009: a discount past -100% would make billout negative — reject
    # it (−100% is the floor = free). The legacy app enforced the same bound.
    if pct < Decimal("-100"):
        raise ServiceError(
            422,
            "PCT_TOO_LOW",
            "Percentage discount cannot be less than -100% (which is free).",
        )
    return None, pct, None


def _override_project(session: Session, tenant_id: uuid.UUID, project_id: uuid.UUID) -> Project:
    project = session.scalar(
        select(Project).where(Project.id == project_id, Project.tenant_id == tenant_id)
    )
    if project is None:
        raise ServiceError(404, "PROJECT_NOT_FOUND", "Project not found.")
    return project


def _worker_id_or_none(raw) -> uuid.UUID | None:
    if raw is None or str(raw).strip() == "":
        return None
    return uuid.UUID(str(raw))


def create_rate_override(session: Session, actor, data: dict) -> dict:
    """One override rule per (project, worker) — a NULL worker is the
    project-wide rule that applies to everyone incl. workers added later.
    Worker-specific beats project-wide at stamp time (judgment #25 /
    stamping._billout_override)."""
    _require_manage(actor, "manage rate overrides")
    tenant_id = _require_tenant(actor)

    project_id = uuid.UUID(str(data["project_id"]))
    project = _override_project(session, tenant_id, project_id)
    worker_party_id = _worker_id_or_none(data.get("worker_party_id"))
    if worker_party_id is not None and _worker_engagement(
        session, tenant_id, worker_party_id, active_only=False
    ) is None:
        raise ServiceError(404, "WORKER_NOT_FOUND", "Override worker is not a team member.")
    rate, pct, currency = _validate_override_kind(
        data.get("override_rate"), data.get("override_pct"), data.get("currency")
    )

    clash = session.scalar(
        select(RateOverride.id).where(
            RateOverride.project_id == project_id,
            (
                RateOverride.worker_party_id == worker_party_id
                if worker_party_id is not None
                else RateOverride.worker_party_id.is_(None)
            ),
        )
    )
    if clash is not None:
        scope = "project-wide" if worker_party_id is None else "worker-specific"
        raise ServiceError(
            409,
            "OVERRIDE_EXISTS",
            f'A {scope} override already exists on "{project.name}". Edit it instead.',
        )

    override = RateOverride(
        tenant_id=tenant_id,
        project_id=project_id,
        worker_party_id=worker_party_id,
        override_rate=rate,
        currency=currency,
        override_pct=pct,
        notes=_clean_text(data.get("notes")),
    )
    session.add(override)
    session.flush()
    write_audit(
        session,
        actor,
        action="team.create_rate_override",
        entity_type="rate_override",
        entity_id=override.id,
        after={
            "project_id": str(project_id),
            "worker_party_id": str(worker_party_id) if worker_party_id else None,
            "override_rate": str(rate) if rate is not None else None,
            "override_pct": str(pct) if pct is not None else None,
        },
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Added a billout override on "{project.name}".',
        "id": str(override.id),
    }


def update_rate_override(
    session: Session, actor, override_id: uuid.UUID, data: dict
) -> dict:
    """Change the rate/pct or notes of an existing override. The scope
    (project + worker) is fixed — to re-target, delete and recreate."""
    _require_manage(actor, "manage rate overrides")
    tenant_id = _require_tenant(actor)
    override = session.scalar(
        select(RateOverride).where(
            RateOverride.id == override_id, RateOverride.tenant_id == tenant_id
        )
    )
    if override is None:
        raise ServiceError(404, "OVERRIDE_NOT_FOUND", "Rate override not found.")

    before = {
        "override_rate": (
            str(override.override_rate) if override.override_rate is not None else None
        ),
        "override_pct": (
            str(override.override_pct) if override.override_pct is not None else None
        ),
        "currency": override.currency,
        "notes": override.notes,
    }
    rate, pct, currency = _validate_override_kind(
        data.get("override_rate"), data.get("override_pct"), data.get("currency")
    )
    override.override_rate = rate
    override.override_pct = pct
    override.currency = currency
    if "notes" in data:
        override.notes = _clean_text(data.get("notes"))

    write_audit(
        session,
        actor,
        action="team.update_rate_override",
        entity_type="rate_override",
        entity_id=override.id,
        before=before,
        after={
            "override_rate": str(rate) if rate is not None else None,
            "override_pct": str(pct) if pct is not None else None,
            "currency": currency,
        },
    )
    session.commit()
    return {"ok": True, "changed": True, "info": "Rate override updated."}


def delete_rate_override(session: Session, actor, override_id: uuid.UUID) -> dict:
    _require_manage(actor, "manage rate overrides")
    tenant_id = _require_tenant(actor)
    override = session.scalar(
        select(RateOverride).where(
            RateOverride.id == override_id, RateOverride.tenant_id == tenant_id
        )
    )
    if override is None:
        raise ServiceError(404, "OVERRIDE_NOT_FOUND", "Rate override not found.")
    write_audit(
        session,
        actor,
        action="team.delete_rate_override",
        entity_type="rate_override",
        entity_id=override.id,
        before={
            "project_id": str(override.project_id),
            "worker_party_id": (
                str(override.worker_party_id) if override.worker_party_id else None
            ),
            "override_rate": (
                str(override.override_rate) if override.override_rate is not None else None
            ),
            "override_pct": (
                str(override.override_pct) if override.override_pct is not None else None
            ),
        },
    )
    session.delete(override)
    session.commit()
    return {"ok": True, "changed": True, "info": "Rate override removed."}