"""Entity management — rename / move / MERGE / delete-empty for the
operator -> client -> project chain (projects-and-entities.md §manage,
rebuilt per judgment #10: merge is first-class at every level).

Model mapping (02 §2): operators and clients are org parties; the
client relationship is a (operator_party, client_party) EDGE — an
active `direct_client` engagement (∪ pairs implied by project FKs), so
client-level operations are keyed by the edge, not the bare party.

What the FK display model buys us (rebuild delta):
  - rename = ONE party/project row update — no text-cache propagation
    layer (legacy B38 is dead); every entry reachable via the project
    FK shows the new name instantly (T-ENTITY-002).
  - move = re-point the project's party FKs — entries follow.
  - merge = move children + re-point referencing rows + retire the
    husk. Same-name child collisions merge RECURSIVELY (T-ENTITY-013 —
    the PlusROI/Palms incident class), all inside ONE transaction
    (T-ENTITY-014: no partial state, killing the legacy sequential-
    PostgREST landmines B44–B47).

Invoiced rows are snapshot-safe by construction (judgment #9): invoice
lines copy client/project display names at attach time, so re-pointing
an entry's project FK never rewrites an issued invoice.

Husk rule ("delete the husk" vs "parties are never hard-deleted"):
project rows ARE deleted (nothing references a merged-away project);
merged-away PARTIES are retired — their direct_client edges end and,
when no live reference remains, the party is archived
(is_active=false) after re-pointing its invoices at the target.

Permissions: every mutation gates on can_manage_imports (manager+),
per-action message parity with the legacy actions.
"""

from __future__ import annotations

import uuid
from datetime import date

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

from app.models import (
    CcAutoRule,
    CcExpenseBatch,
    CcExpenseLine,
    CompensationTerm,
    Engagement,
    Invoice,
    Party,
    Project,
    RateOverride,
    Tenant,
    TimeEntry,
)
from app.services.audit import write_audit
from app.services.entities.resolve import EntityGraph, load_entity_graph
from app.services.errors import ServiceError


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


def _graph(session: Session, actor) -> EntityGraph:
    if actor.tenant_id is None:
        raise ServiceError(400, "NO_TENANT", "No tenant context resolved.")
    return load_entity_graph(session, actor.tenant_id)


# ---------------------------------------------------------------- counts


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


def _cc_counts(session: Session, tenant_id: uuid.UUID) -> dict[uuid.UUID, int]:
    rows = session.execute(
        select(CcExpenseLine.project_id, func.count())
        .join(CcExpenseBatch, CcExpenseBatch.id == CcExpenseLine.batch_id)
        .where(CcExpenseBatch.tenant_id == tenant_id, CcExpenseLine.project_id.is_not(None))
        .group_by(CcExpenseLine.project_id)
    ).all()
    return dict(rows)


def manage_payload(session: Session, actor) -> dict:
    """The /projects/manage page data: three sections with TRUE
    aggregate counts (T-ENTITY-015 — immune to page caps)."""
    graph = _graph(session, actor)
    entries = _entry_counts(session, actor.tenant_id)
    cc = _cc_counts(session, actor.tenant_id)
    can_see_income = actor.allowed("can_see_income")

    project_rows = session.execute(
        select(Project).where(Project.tenant_id == actor.tenant_id).order_by(Project.name)
    ).scalars()

    projects = []
    per_edge_projects: dict[tuple[uuid.UUID, uuid.UUID], int] = {}
    per_edge_entries: dict[tuple[uuid.UUID, uuid.UUID], int] = {}
    per_op_entries: dict[uuid.UUID, int] = {}
    for p in project_rows:
        n = entries.get(p.id, 0)
        edge = (p.operator_party_id, p.client_party_id)
        per_edge_projects[edge] = per_edge_projects.get(edge, 0) + 1
        per_edge_entries[edge] = per_edge_entries.get(edge, 0) + n
        per_op_entries[p.operator_party_id] = per_op_entries.get(p.operator_party_id, 0) + n
        row = {
            "id": str(p.id),
            "name": p.name,
            "status": p.status,
            "operator_id": str(p.operator_party_id),
            "operator": graph.operators.get(p.operator_party_id),
            "client_id": str(p.client_party_id),
            "client": graph.client_names.get(p.client_party_id),
            "entries": n,
            "cc_lines": cc.get(p.id, 0),
        }
        if can_see_income:
            row["income"] = str(p.income) if p.income is not None else None
        projects.append(row)

    clients = sorted(
        (
            {
                "operator_id": str(op_id),
                "operator": graph.operators.get(op_id, ""),
                "client_id": str(cl_id),
                "name": graph.client_names.get(cl_id, ""),
                "projects": per_edge_projects.get((op_id, cl_id), 0),
                "entries": per_edge_entries.get((op_id, cl_id), 0),
            }
            for op_id, by_name in graph.clients_of.items()
            for cl_id in set(by_name.values())
        ),
        key=lambda c: (c["operator"].lower(), c["name"].lower()),
    )

    operators = sorted(
        (
            {
                "id": str(op_id),
                "name": name,
                "is_tenant": op_id == graph.tenant_party_id,
                "clients": len(set(graph.clients_of.get(op_id, {}).values())),
                "entries": per_op_entries.get(op_id, 0),
            }
            for op_id, name in graph.operators.items()
        ),
        key=lambda o: o["name"].lower(),
    )

    return {
        "operators": operators,
        "clients": clients,
        "projects": projects,
        "can_manage": actor.allowed("can_manage_imports"),
    }


# ---------------------------------------------------------------- rename


def _no_change(name: str) -> dict:
    return {"ok": True, "changed": False, "info": "No change — name was already that."}


def rename_operator(session: Session, actor, party_id: uuid.UUID, new_name: str) -> dict:
    _require_manage(actor, "rename entities")
    graph = _graph(session, actor)
    name = (new_name or "").strip()
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "Name is required.")
    if party_id not in graph.operators:
        raise ServiceError(404, "NOT_FOUND", "Operator not found.")
    if graph.operators[party_id] == name:
        return _no_change(name)
    dup = graph.op_by_lower.get(name.lower())
    if dup is not None and dup != party_id:
        raise ServiceError(
            409, "DUPLICATE_NAME", f'Another operator already has the name "{name}".'
        )
    return _rename_party(session, actor, party_id, name, entity_label="operator")


def rename_client(
    session: Session, actor, operator_id: uuid.UUID, client_id: uuid.UUID, new_name: str
) -> dict:
    _require_manage(actor, "rename entities")
    graph = _graph(session, actor)
    name = (new_name or "").strip()
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "Name is required.")
    if client_id not in set(graph.clients_of.get(operator_id, {}).values()):
        raise ServiceError(404, "NOT_FOUND", "Client not found.")
    if graph.client_names.get(client_id) == name:
        return _no_change(name)
    # the party may hang under several operators — collision-check every
    # scope it participates in (plus the operator set if it wears that hat)
    for by_name in graph.clients_of.values():
        if client_id in set(by_name.values()):
            dup = by_name.get(name.lower())
            if dup is not None and dup != client_id:
                raise ServiceError(
                    409,
                    "DUPLICATE_NAME",
                    f'Another client under this operator already has the name "{name}".',
                )
    if client_id in graph.operators:
        dup = graph.op_by_lower.get(name.lower())
        if dup is not None and dup != client_id:
            raise ServiceError(
                409, "DUPLICATE_NAME", f'Another operator already has the name "{name}".'
            )
    return _rename_party(session, actor, client_id, name, entity_label="client")


def _rename_party(
    session: Session, actor, party_id: uuid.UUID, name: str, *, entity_label: str
) -> dict:
    party = session.get(Party, party_id)
    old = party.name
    party.name = name
    # informational: how many entries now display the new name (FK reach)
    affected = session.scalar(
        select(func.count())
        .select_from(TimeEntry)
        .join(Project, Project.id == TimeEntry.project_id)
        .where(
            TimeEntry.tenant_id == actor.tenant_id,
            (Project.operator_party_id == party_id) | (Project.client_party_id == party_id),
        )
    )
    write_audit(
        session,
        actor,
        action=f"entities.rename_{entity_label}",
        entity_type="party",
        entity_id=party_id,
        before={"name": old},
        after={"name": name},
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Renamed to "{name}". {affected} entr{"y" if affected == 1 else "ies"}'
        " now display it.",
        "entries_affected": affected,
    }


def rename_project(session: Session, actor, project_id: uuid.UUID, new_name: str) -> dict:
    _require_manage(actor, "rename entities")
    graph = _graph(session, actor)
    name = (new_name or "").strip()
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "Name is required.")
    chain = graph.project_chain.get(project_id)
    if chain is None:
        raise ServiceError(404, "NOT_FOUND", "Project not found.")
    old_name, _op_id, cl_id = chain
    if old_name == name:
        return _no_change(name)
    dup = graph.projects_by_client.get(cl_id, {}).get(name.lower())
    if dup is not None and dup != project_id:
        raise ServiceError(
            409,
            "DUPLICATE_NAME",
            f'Another project under this client already has the name "{name}".',
        )
    project = session.get(Project, project_id)
    project.name = name
    affected = session.scalar(
        select(func.count()).where(TimeEntry.project_id == project_id)
    )
    write_audit(
        session,
        actor,
        action="entities.rename_project",
        entity_type="project",
        entity_id=project_id,
        before={"name": old_name},
        after={"name": name},
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Renamed to "{name}". {affected} entr{"y" if affected == 1 else "ies"}'
        " now display it.",
        "entries_affected": affected,
    }


# ---------------------------------------------------------------- move


def move_project(
    session: Session,
    actor,
    project_id: uuid.UUID,
    target_operator_id: uuid.UUID,
    target_client_id: uuid.UUID,
) -> dict:
    """Re-point the project's party FKs (B40–B42 rebuilt): entries
    follow the FK, the project keeps id/overrides/income/adjustments,
    invoices are NOT touched. Name collision -> merge guidance."""
    _require_manage(actor, "move projects")
    graph = _graph(session, actor)
    chain = graph.project_chain.get(project_id)
    if chain is None:
        raise ServiceError(404, "NOT_FOUND", "Project not found.")
    name, cur_op, cur_cl = chain
    if target_client_id not in set(graph.clients_of.get(target_operator_id, {}).values()):
        raise ServiceError(404, "NOT_FOUND", "Target client not found.")
    if (cur_op, cur_cl) == (target_operator_id, target_client_id):
        return {
            "ok": True,
            "changed": False,
            "info": "No change — project is already under that client.",
        }
    dup = graph.projects_by_client.get(target_client_id, {}).get(name.lower())
    if dup is not None and dup != project_id:
        raise ServiceError(
            409,
            "MERGE_INSTEAD",
            f'Couldn\'t move "{name}" — the target client already has a project with'
            " that name. Merge the two projects instead.",
        )
    project = session.get(Project, project_id)
    project.operator_party_id = target_operator_id
    project.client_party_id = target_client_id
    entries = session.scalar(select(func.count()).where(TimeEntry.project_id == project_id))
    write_audit(
        session,
        actor,
        action="entities.move_project",
        entity_type="project",
        entity_id=project_id,
        before={"operator_id": str(cur_op), "client_id": str(cur_cl)},
        after={"operator_id": str(target_operator_id), "client_id": str(target_client_id)},
    )
    session.commit()
    op_name = graph.operators.get(target_operator_id, "?")
    cl_name = graph.client_names.get(target_client_id, "?")
    return {
        "ok": True,
        "changed": True,
        "info": f'Moved "{name}" to {op_name} / {cl_name}.'
        f' {entries} entr{"y" if entries == 1 else "ies"} follow the project.',
        "entries_affected": entries,
    }


# ---------------------------------------------------------------- merge


def _merge_project_rows(
    session: Session, tenant_id: uuid.UUID, graph: EntityGraph,
    source_id: uuid.UUID, target_id: uuid.UUID,
) -> dict:
    """The transactional core of a project merge (no gate, no commit):
    re-point entries / CC lines / auto-rules / project-scoped terms,
    DROP the source's rate overrides (legacy parity — target's rules
    win), delete the source row. Source income/adjustments are lost by
    design."""
    entries = session.scalar(select(func.count()).where(TimeEntry.project_id == source_id))
    cc_lines = session.scalar(
        select(func.count()).where(CcExpenseLine.project_id == source_id)
    )
    session.execute(
        update(TimeEntry).where(TimeEntry.project_id == source_id).values(project_id=target_id)
    )
    session.execute(
        update(CcExpenseLine)
        .where(CcExpenseLine.project_id == source_id)
        .values(project_id=target_id)
    )
    session.execute(
        update(CcAutoRule).where(CcAutoRule.project_id == source_id).values(project_id=target_id)
    )
    # project-scoped compensation terms keep applying to the same work
    session.execute(
        update(CompensationTerm)
        .where(CompensationTerm.project_id == source_id)
        .values(project_id=target_id)
    )
    # overrides dropped, not merged (parity: the target's rules win)
    for override in session.scalars(
        select(RateOverride).where(RateOverride.project_id == source_id)
    ):
        session.delete(override)
    source = session.get(Project, source_id)
    session.delete(source)
    session.flush()

    # keep the in-memory graph honest for recursive callers
    _, _op, cl = graph.project_chain.pop(source_id, (None, None, None))
    if cl is not None:
        by_name = graph.projects_by_client.get(cl, {})
        for key, value in list(by_name.items()):
            if value == source_id:
                del by_name[key]
    return {"entries": entries or 0, "cc_lines": cc_lines or 0}


def merge_project(session: Session, actor, source_id: uuid.UUID, target_id: uuid.UUID) -> dict:
    _require_manage(actor, "merge projects")
    graph = _graph(session, actor)
    if source_id == target_id:
        raise ServiceError(
            409, "SELF_MERGE", "Source and target are the same project — can't merge into self."
        )
    source_chain = graph.project_chain.get(source_id)
    if source_chain is None:
        raise ServiceError(404, "NOT_FOUND", "Source project not found.")
    target_chain = graph.project_chain.get(target_id)
    if target_chain is None:
        raise ServiceError(404, "NOT_FOUND", "Target project not found.")

    counts = _merge_project_rows(session, actor.tenant_id, graph, source_id, target_id)
    write_audit(
        session,
        actor,
        action="entities.merge_project",
        entity_type="project",
        entity_id=target_id,
        before={
            "source": {"id": str(source_id), "name": source_chain[0]},
            "counts": counts,
        },
        after={"target": {"id": str(target_id), "name": target_chain[0]}},
    )
    session.commit()
    tgt_label = f"{graph.client_names.get(target_chain[2], '?')} : {target_chain[0]}"
    cc_note = f" + {counts['cc_lines']} CC expense lines" if counts["cc_lines"] else ""
    return {
        "ok": True,
        "changed": True,
        "info": f'Merged "{source_chain[0]}" into "{tgt_label}".'
        f" Reassigned {counts['entries']} entries{cc_note}.",
        **counts,
    }


def _end_client_edge(
    session: Session, tenant_id: uuid.UUID, operator_id: uuid.UUID, client_id: uuid.UUID
) -> None:
    for engagement in session.scalars(
        select(Engagement).where(
            Engagement.tenant_id == tenant_id,
            Engagement.type == "direct_client",
            Engagement.party_a_id == operator_id,
            Engagement.party_b_id == client_id,
            Engagement.ended_on.is_(None),
        )
    ):
        engagement.ended_on = date.today()


def _ensure_client_edge(
    session: Session, tenant_id: uuid.UUID, operator_id: uuid.UUID, client_id: uuid.UUID
) -> None:
    exists = session.scalar(
        select(Engagement.id).where(
            Engagement.tenant_id == tenant_id,
            Engagement.type == "direct_client",
            Engagement.party_a_id == operator_id,
            Engagement.party_b_id == client_id,
            Engagement.ended_on.is_(None),
        )
    )
    if exists is None:
        session.add(
            Engagement(
                tenant_id=tenant_id,
                type="direct_client",
                party_a_id=operator_id,
                party_b_id=client_id,
                role_a="operator",
                role_b="client",
            )
        )
        session.flush()


def _party_still_referenced(session: Session, party_id: uuid.UUID) -> bool:
    if session.scalar(select(Tenant.id).where(Tenant.party_id == party_id)) is not None:
        return True
    if (
        session.scalar(
            select(Engagement.id).where(
                (Engagement.party_a_id == party_id) | (Engagement.party_b_id == party_id),
                Engagement.ended_on.is_(None),
            )
        )
        is not None
    ):
        return True
    return (
        session.scalar(
            select(Project.id).where(
                (Project.operator_party_id == party_id) | (Project.client_party_id == party_id)
            )
        )
        is not None
    )


def _retire_party_if_unused(
    session: Session, tenant_id: uuid.UUID, party_id: uuid.UUID, target_party_id: uuid.UUID
) -> bool:
    """Archive the merged-away party when nothing live references it,
    re-pointing its invoice linkage at the target first (T-ENTITY-009/
    010: invoice linkage follows the merge)."""
    if _party_still_referenced(session, party_id):
        return False
    session.execute(
        update(Invoice)
        .where(Invoice.tenant_id == tenant_id, Invoice.to_party_id == party_id)
        .values(to_party_id=target_party_id)
    )
    session.execute(
        update(Invoice)
        .where(Invoice.tenant_id == tenant_id, Invoice.from_party_id == party_id)
        .values(from_party_id=target_party_id)
    )
    party = session.get(Party, party_id)
    party.is_active = False
    return True


def _merge_client_edges(
    session: Session, actor, graph: EntityGraph,
    src_op: uuid.UUID, src_cl: uuid.UUID, tgt_op: uuid.UUID, tgt_cl: uuid.UUID,
) -> dict:
    """Transactional core of a client merge: move the source edge's
    projects to the target edge, recursively merging same-named
    projects; end the source edge; retire the source party when it has
    no other life."""
    counts = {"projects_moved": 0, "projects_merged": 0, "entries": 0, "cc_lines": 0}
    source_projects = [
        (pid, chain[0])
        for pid, chain in list(graph.project_chain.items())
        if chain[1] == src_op and chain[2] == src_cl
    ]
    for pid, name in sorted(source_projects, key=lambda t: t[1].lower()):
        collision = graph.projects_by_client.get(tgt_cl, {}).get(name.strip().lower())
        if collision is not None and collision != pid:
            merged = _merge_project_rows(session, actor.tenant_id, graph, pid, collision)
            counts["projects_merged"] += 1
            counts["entries"] += merged["entries"]
            counts["cc_lines"] += merged["cc_lines"]
        else:
            project = session.get(Project, pid)
            project.operator_party_id = tgt_op
            project.client_party_id = tgt_cl
            session.flush()
            counts["projects_moved"] += 1
            # graph bookkeeping: re-home the project under the target
            graph.project_chain[pid] = (name, tgt_op, tgt_cl)
            by_name = graph.projects_by_client.get(src_cl, {})
            for key, value in list(by_name.items()):
                if value == pid:
                    del by_name[key]
            graph.projects_by_client.setdefault(tgt_cl, {})[name.strip().lower()] = pid

    _end_client_edge(session, actor.tenant_id, src_op, src_cl)
    _ensure_client_edge(session, actor.tenant_id, tgt_op, tgt_cl)
    # graph bookkeeping: drop the source edge
    by_name = graph.clients_of.get(src_op, {})
    for key, value in list(by_name.items()):
        if value == src_cl:
            del by_name[key]
    session.flush()
    counts["retired_party"] = (
        src_cl != tgt_cl and _retire_party_if_unused(session, actor.tenant_id, src_cl, tgt_cl)
    )
    return counts


def merge_client(
    session: Session, actor,
    source_operator_id: uuid.UUID, source_client_id: uuid.UUID,
    target_operator_id: uuid.UUID, target_client_id: uuid.UUID,
) -> dict:
    _require_manage(actor, "merge clients")
    graph = _graph(session, actor)
    if source_client_id not in set(graph.clients_of.get(source_operator_id, {}).values()):
        raise ServiceError(404, "NOT_FOUND", "Source client not found.")
    if target_client_id not in set(graph.clients_of.get(target_operator_id, {}).values()):
        raise ServiceError(404, "NOT_FOUND", "Target client not found.")
    if (source_operator_id, source_client_id) == (target_operator_id, target_client_id):
        raise ServiceError(
            409, "SELF_MERGE", "Source and target are the same client — can't merge into self."
        )

    src_name = graph.client_names.get(source_client_id, "?")
    tgt_name = graph.client_names.get(target_client_id, "?")
    counts = _merge_client_edges(
        session, actor, graph,
        source_operator_id, source_client_id, target_operator_id, target_client_id,
    )
    write_audit(
        session,
        actor,
        action="entities.merge_client",
        entity_type="party",
        entity_id=target_client_id,
        before={
            "source": {
                "operator_id": str(source_operator_id),
                "client_id": str(source_client_id),
                "name": src_name,
            },
            "counts": counts,
        },
        after={
            "target": {
                "operator_id": str(target_operator_id),
                "client_id": str(target_client_id),
                "name": tgt_name,
            }
        },
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Merged client "{src_name}" into'
        f' "{graph.operators.get(target_operator_id, "?")} : {tgt_name}".'
        f" Moved {counts['projects_moved']} and merged {counts['projects_merged']}"
        f" projects ({counts['entries']} entries re-pointed).",
        **counts,
    }


def merge_operator(
    session: Session, actor, source_id: uuid.UUID, target_id: uuid.UUID
) -> dict:
    _require_manage(actor, "merge operators")
    graph = _graph(session, actor)
    if source_id not in graph.operators:
        raise ServiceError(404, "NOT_FOUND", "Source operator not found.")
    if target_id not in graph.operators:
        raise ServiceError(404, "NOT_FOUND", "Target operator not found.")
    if source_id == target_id:
        raise ServiceError(
            409, "SELF_MERGE", "Source and target are the same operator — can't merge into self."
        )
    if source_id == graph.tenant_party_id:
        raise ServiceError(
            409,
            "TENANT_PARTY",
            "Can't merge away the tenant's own operator party.",
            "Merge the other operator INTO it instead.",
        )

    src_name = graph.operators[source_id]
    tgt_name = graph.operators[target_id]
    totals = {"clients_merged": 0, "clients_moved": 0, "projects_moved": 0,
              "projects_merged": 0, "entries": 0, "cc_lines": 0}

    source_edges = [
        (cl_id, graph.client_names.get(cl_id, ""))
        for cl_id in set(graph.clients_of.get(source_id, {}).values())
    ]
    target_clients = graph.clients_of.get(target_id, {})
    for cl_id, cl_name in sorted(source_edges, key=lambda t: t[1].lower()):
        same_party = cl_id in set(target_clients.values())
        same_name_party = target_clients.get(cl_name.strip().lower())
        if same_party or (same_name_party is not None and same_name_party != cl_id):
            # collapse into the target's edge (same party or same-named
            # client) — the recursive client merge handles both
            tgt_cl = cl_id if same_party else same_name_party
            counts = _merge_client_edges(
                session, actor, graph, source_id, cl_id, target_id, tgt_cl
            )
            totals["clients_merged"] += 1
        else:
            # re-parent the whole edge under the target operator
            counts = _merge_client_edges(
                session, actor, graph, source_id, cl_id, target_id, cl_id
            )
            totals["clients_moved"] += 1
            graph.clients_of.setdefault(target_id, {})[cl_name.strip().lower()] = cl_id
        for key in ("projects_moved", "projects_merged", "entries", "cc_lines"):
            totals[key] += counts[key]

    # stragglers: projects carrying the source operator FK with no edge
    session.execute(
        update(Project)
        .where(Project.tenant_id == actor.tenant_id, Project.operator_party_id == source_id)
        .values(operator_party_id=target_id)
    )
    session.flush()
    totals["retired_party"] = _retire_party_if_unused(
        session, actor.tenant_id, source_id, target_id
    )

    write_audit(
        session,
        actor,
        action="entities.merge_operator",
        entity_type="party",
        entity_id=target_id,
        before={"source": {"id": str(source_id), "name": src_name}, "counts": totals},
        after={"target": {"id": str(target_id), "name": tgt_name}},
    )
    session.commit()
    return {
        "ok": True,
        "changed": True,
        "info": f'Merged operator "{src_name}" into "{tgt_name}".'
        f" {totals['clients_moved']} clients moved, {totals['clients_merged']} merged"
        f" ({totals['entries']} entries re-pointed).",
        **totals,
    }


# ---------------------------------------------------------------- delete


def delete_project(session: Session, actor, project_id: uuid.UUID) -> dict:
    """Empty-only delete (children -> use merge; the legacy
    delete-with-reassign IS merge now)."""
    _require_manage(actor, "delete projects")
    graph = _graph(session, actor)
    chain = graph.project_chain.get(project_id)
    if chain is None:
        raise ServiceError(404, "NOT_FOUND", "Project not found.")
    name = chain[0]
    entries = session.scalar(select(func.count()).where(TimeEntry.project_id == project_id))
    cc_lines = session.scalar(
        select(func.count()).where(CcExpenseLine.project_id == project_id)
    )
    if entries or cc_lines:
        parts = []
        if entries:
            parts.append(f"{entries} time entr{'y' if entries == 1 else 'ies'}")
        if cc_lines:
            parts.append(f"{cc_lines} CC expense line{'' if cc_lines == 1 else 's'}")
        raise ServiceError(
            409,
            "HAS_CHILDREN",
            f'Can\'t delete "{name}" — {" and ".join(parts)} still point at it.'
            " Merge it into another project instead.",
        )
    for override in session.scalars(
        select(RateOverride).where(RateOverride.project_id == project_id)
    ):
        session.delete(override)
    for term in session.scalars(
        select(CompensationTerm).where(CompensationTerm.project_id == project_id)
    ):
        session.delete(term)
    for rule in session.scalars(select(CcAutoRule).where(CcAutoRule.project_id == project_id)):
        session.delete(rule)
    project = session.get(Project, project_id)
    session.delete(project)
    write_audit(
        session,
        actor,
        action="entities.delete_project",
        entity_type="project",
        entity_id=project_id,
        before={"name": name},
    )
    session.commit()
    return {"ok": True, "changed": True, "info": f'Deleted empty project "{name}".'}


def delete_client(
    session: Session, actor, operator_id: uuid.UUID, client_id: uuid.UUID
) -> dict:
    _require_manage(actor, "delete clients")
    graph = _graph(session, actor)
    if client_id not in set(graph.clients_of.get(operator_id, {}).values()):
        raise ServiceError(404, "NOT_FOUND", "Client not found.")
    name = graph.client_names.get(client_id, "?")
    projects = [
        pid
        for pid, chain in graph.project_chain.items()
        if chain[1] == operator_id and chain[2] == client_id
    ]
    if projects:
        raise ServiceError(
            409,
            "HAS_CHILDREN",
            f'Can\'t delete "{name}" — {len(projects)} project'
            f'{"" if len(projects) == 1 else "s"} still live under it.'
            " Merge it into another client instead.",
        )
    _end_client_edge(session, actor.tenant_id, operator_id, client_id)
    retired = _retire_party_if_unused(session, actor.tenant_id, client_id, client_id)
    write_audit(
        session,
        actor,
        action="entities.delete_client",
        entity_type="party",
        entity_id=client_id,
        before={"operator_id": str(operator_id), "name": name, "retired": retired},
    )
    session.commit()
    return {"ok": True, "changed": True, "info": f'Deleted empty client "{name}".'}


def delete_operator(session: Session, actor, party_id: uuid.UUID) -> dict:
    _require_manage(actor, "delete operators")
    graph = _graph(session, actor)
    if party_id not in graph.operators:
        raise ServiceError(404, "NOT_FOUND", "Operator not found.")
    if party_id == graph.tenant_party_id:
        raise ServiceError(
            409, "TENANT_PARTY", "Can't delete the tenant's own operator party."
        )
    name = graph.operators[party_id]
    clients = set(graph.clients_of.get(party_id, {}).values())
    own_projects = [
        pid for pid, chain in graph.project_chain.items() if chain[1] == party_id
    ]
    if clients or own_projects:
        raise ServiceError(
            409,
            "HAS_CHILDREN",
            f'Can\'t delete "{name}" — {len(clients)} client'
            f'{"" if len(clients) == 1 else "s"} still live under it.'
            " Merge it into another operator instead.",
        )
    retired = _retire_party_if_unused(session, actor.tenant_id, party_id, party_id)
    write_audit(
        session,
        actor,
        action="entities.delete_operator",
        entity_type="party",
        entity_id=party_id,
        before={"name": name, "retired": retired},
    )
    session.commit()
    return {"ok": True, "changed": True, "info": f'Deleted empty operator "{name}".'}