"""services/money/invoices.py — THE invoice service (04 "Money paths":
invoice work always goes through here; attaching a line locks the
underlying rows).

R1 semantics = the parity bar (invoices.md B1–B76) plus the rebuild
deltas the plan pins:

  - invoices keep the current shape verbatim (name unique per tenant
    case-insensitively, open/sent/paid with independent first-crossing
    sent_at/paid_at stamps, invoice_date defaulting to the last day of
    the previous month, created_by = the REAL human, notes,
    manual_total override) — 02 §3 / judgment #24;
  - the lock moved to invoice_lines: applying creates one `time`-kind
    line per entry with SNAPSHOT display fields (client/project names,
    description, hours, amount as of attach — judgment #9) and stamps
    `invoice_line_id` on the entry;
  - eligibility is THE ONE predicate (`status_condition("pending")`) —
    the four divergent legacy copies do not port (T-INV-010: display
    and apply agree by construction);
  - apply/detach ride the M5 bulk engine (same-filter scope,
    expected_count safety stop, one transaction, verified counts,
    audit before-images). Detach is EXPLICIT-selection only (B62).
  - delete is GUARDED (only zero-line invoices delete) — the legacy
    detach-everything-on-delete behavior destroyed the apply audit
    trail (B71 wart); the rebuild makes "detach, then delete" two
    deliberate steps (ADR #007);
  - invoices are single-currency documents: applying an entry whose
    billout stamp is in another currency aborts loudly (02 §3 R1 rule);
  - aggregates are SQL-side and scope()-transformed, so a viewer with
    partial visibility sums only what they can see (T-INV-023) and the
    unbounded client-side reduction (B17/B42 wart) does not port.

Undo policy (ADR #007): apply/detach are NOT replayed by the generic
audit undo — the inverse operation IS the undo (detach undoes apply;
re-apply undoes detach), each verified and audited in its own right.
The audit rows still carry complete before-images for forensics.
"""

from __future__ import annotations

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

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

from app.models import Invoice, InvoiceLine, Party, Tenant, TimeEntry
from app.services.audit import snapshot_entry, write_audit
from app.services.authz.scope import scope
from app.services.bulk import BulkSelection, collect_scope
from app.services.errors import ServiceError
from app.services.reporting.filters import EntryJoins, status_condition
from app.services.reporting.money import (
    MixedCurrencyError,
    Money,
    money_from_sum,
    quantize2,
)
from app.services.tenancy import BOOKS_OWNER_SHEET, effective_books_owner

INVOICE_STATUSES = ("open", "sent", "paid")

#: PlusROI-lifecycle statuses this BW module must never touch (ADR
#: #024, extended by ADR #025). ``clean_status`` coerces anything
#: outside open|sent|paid to ``open``, ``apply_entries`` would attach a
#: line to a receivable meant to carry none, and ``delete_invoice``
#: would destroy it outright — all correct for BW documents, all
#: destructive to a PlusROI receivable mid-lifecycle. Every mutation
#: (``update_invoice``/``apply_entries``/``delete_invoice``) refuses via
#: ``_reject_plus_lifecycle`` when the CURRENT status is PlusROI-shaped.
#: ``sent``/``paid`` are shared vocabulary and stay editable/appliable
#: here (BW's own documents live in them). The PlusROI surface is
#: ``PATCH /api/plus-invoices/{id}``.
PLUS_LIFECYCLE_STATUSES = frozenset(
    {"pending_approval", "ready", "cancelled", "delayed"}
)

#: How many lines the detail endpoint returns inline (the legacy
#: "Legacy transfers" invoice carries 2,387 — the UI shows a page of
#: snapshots plus the true total).
DETAIL_LINE_CAP = 500


# ------------------------------------------------------------- helpers


def default_invoice_date(today: date | None = None) -> date:
    """Last day of the previous month (B4: "create the April 30 invoice
    anytime in May"). The pytest twin of the legacy defaultInvoiceDate."""
    today = today or date.today()
    return today.replace(day=1) - timedelta(days=1)


_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def parse_invoice_date(raw: str | None) -> date | None:
    """YYYY-MM-DD with REAL calendar validation — the legacy regex let
    `2026-99-99` through to a raw PG error (B4 wart); the rebuild treats
    impossible dates as malformed (-> default on create, NULL on update,
    exactly T-INV-012)."""
    if raw is None:
        return None
    text = raw.strip()
    if not _DATE_RE.match(text):
        return None
    try:
        return date.fromisoformat(text)
    except ValueError:
        return None


def clean_status(raw: str | None) -> str:
    """B3: trim + lowercase; anything outside open|sent|paid coerces to
    open (including absent)."""
    value = (raw or "").strip().lower()
    return value if value in INVOICE_STATUSES else "open"


def parse_manual_total(raw) -> Decimal | None:
    """parseMoneyOrNull parity (B6): ''/None/non-numeric -> None, else
    2dp half-up."""
    if raw is None:
        return None
    text = str(raw).strip()
    if not text:
        return None
    try:
        return quantize2(Decimal(text))
    except InvalidOperation:
        return None


def _require_transfer(actor, message: str) -> None:
    if not actor.allowed("can_transfer"):
        raise ServiceError(403, "NOT_ALLOWED", message)


def _require_writable_books(session: Session, actor) -> None:
    """Sheet-owned write-block (ADR #019, re-keyed by ADR #021). While a
    tenant's books are SHEET-owned (``books_owner == "sheet"``, which the
    nightly rebuilds from the sheet export — the legacy boolean
    ``parallel_run`` reads as sheet-owned too), EVERY invoice mutation is
    refused: an edit would look saved, then silently vanish at the next
    rebuild — a partner-facing trust hazard (adversarial-review finding,
    2026-07-11). The sheet stays the source of truth; the block lifts
    when the tenant is flipped app-owned (scripts/flip_books_owner.py) —
    no code change."""
    tenant = session.get(Tenant, actor.tenant_id)
    if tenant is not None and (
        effective_books_owner(tenant.settings) == BOOKS_OWNER_SHEET
    ):
        raise ServiceError(
            409,
            "SHEET_OWNED_READ_ONLY",
            "These books are sheet-owned — the sheet is the source of truth.",
            "The app rebuilds these books nightly from the Google Sheet, so an"
            " edit made here would be overwritten tonight. Make the change in"
            " the sheet; invoice editing unlocks when the books are flipped"
            " app-owned.",
        )


def _reject_plus_lifecycle(invoice: Invoice) -> None:
    """Refuse this BW endpoint on a PlusROI-lifecycle row (ADR #024 §5,
    extended by ADR #025 to every mutation, not just ``update_invoice``):
    ``clean_status`` would silently coerce its status to ``open``,
    ``apply_entries`` would attach a `time` line to a receivable that is
    supposed to carry none, and ``delete_invoice`` would destroy it
    outright — all three assume a BW open/sent/paid document.
    ``sent``/``paid`` are shared vocabulary and never reach here."""
    if invoice.status in PLUS_LIFECYCLE_STATUSES:
        raise ServiceError(
            409,
            "PLUS_LIFECYCLE_INVOICE",
            "This invoice is on the PlusROI lifecycle — manage it from the"
            " Books → Invoices page.",
            f"Its current status ({invoice.status}) belongs to the PlusROI"
            " lifecycle; this Bowden Works endpoint's status vocabulary,"
            " invoice_lines, and delete flow all assume a BW document. Use"
            f" PATCH /api/plus-invoices/{invoice.id} instead.",
        )


def _get_invoice(session: Session, actor, invoice_id: uuid.UUID) -> Invoice:
    invoice = session.scalars(
        select(Invoice).where(
            Invoice.id == invoice_id, Invoice.tenant_id == actor.tenant_id
        )
    ).first()
    if invoice is None:
        raise ServiceError(404, "INVOICE_NOT_FOUND", "Invoice not found.")
    return invoice


def _assert_name_free(
    session: Session, actor, name: str, *, exclude: uuid.UUID | None = None
) -> None:
    stmt = select(Invoice.id).where(
        Invoice.tenant_id == actor.tenant_id,
        func.lower(Invoice.name) == name.lower(),
    )
    if exclude is not None:
        stmt = stmt.where(Invoice.id != exclude)
    if session.scalar(stmt) is not None:
        raise ServiceError(
            409,
            "DUPLICATE_INVOICE_NAME",
            f'An invoice named "{name}" already exists.',
        )


def _validate_party(session: Session, raw) -> uuid.UUID | None:
    if raw in (None, ""):
        return None
    try:
        party_id = uuid.UUID(str(raw))
    except ValueError as exc:
        raise ServiceError(422, "BAD_PARTY", "to_party_id must be a UUID.") from exc
    if session.scalar(select(Party.id).where(Party.id == party_id)) is None:
        raise ServiceError(404, "PARTY_NOT_FOUND", "Selected party not found.")
    return party_id


def _tenant_party_id(session: Session, actor) -> uuid.UUID | None:
    return session.scalar(select(Tenant.party_id).where(Tenant.id == actor.tenant_id))


def _tenant_currency(session: Session, actor) -> str:
    settings = (
        session.scalar(select(Tenant.settings).where(Tenant.id == actor.tenant_id))
        or {}
    )
    return settings.get("default_currency", "CAD")


def _apply_status_stamps(invoice: Invoice, new_status: str, now: datetime) -> None:
    """First-crossing stamps (B9/B12): sent_at when crossing into
    sent/paid, paid_at when crossing into paid; NEVER cleared or
    overwritten ("auditing hates that")."""
    if new_status in ("sent", "paid") and invoice.sent_at is None:
        invoice.sent_at = now
    if new_status == "paid" and invoice.paid_at is None:
        invoice.paid_at = now
    invoice.status = new_status


def _snapshot_line(line: InvoiceLine) -> dict:
    out: dict = {}
    for column in InvoiceLine.__table__.columns:
        value = getattr(line, column.name)
        if isinstance(value, uuid.UUID):
            value = str(value)
        elif isinstance(value, datetime | date):
            value = value.isoformat()
        elif isinstance(value, Decimal):
            value = str(value)
        out[column.name] = value
    return out


def _scalar_image(invoice: Invoice) -> dict:
    out: dict = {}
    for column in Invoice.__table__.columns:
        value = getattr(invoice, column.name)
        if isinstance(value, uuid.UUID):
            value = str(value)
        elif isinstance(value, datetime | date):
            value = value.isoformat()
        elif isinstance(value, Decimal):
            value = str(value)
        out[column.name] = value
    return out


# ------------------------------------------------------------ aggregates


def _aggregate_stmt(actor, *group_cols):
    """count/hours/cost/billout over LINES joined to their scoped
    entries — SQL-side, scope()-transformed (T-INV-023: partial
    visibility sums only visible rows)."""
    stmt = (
        select(
            *group_cols,
            func.count(TimeEntry.id).label("entry_count"),
            func.coalesce(func.sum(TimeEntry.duration_seconds), 0).label("seconds"),
            func.sum(TimeEntry.cost_amount).label("cost_sum"),
            func.count(func.distinct(TimeEntry.cost_currency)).label("cost_currencies"),
            func.min(TimeEntry.cost_currency).label("cost_currency"),
            func.sum(InvoiceLine.amount).label("billout_sum"),
        )
        .select_from(TimeEntry)
        .join(InvoiceLine, InvoiceLine.id == TimeEntry.invoice_line_id)
    )
    if group_cols:
        stmt = stmt.group_by(*group_cols)
    return scope(actor, "entries:read", stmt)


def _margin_block(
    *,
    cost: Money,
    summed_billout: Decimal,
    manual_total: Decimal | None,
    currency: str,
) -> dict:
    """B18 math on Decimals: effective billout = manual override or the
    entry sum; margin = effective - cost; pct to 1dp when effective > 0.
    Margin needs one currency on both sides (02 §3) — R1 data is
    all-one-currency and the cost aggregation already errored loudly if
    not; a cost stamped in a foreign currency against this invoice is
    the same data bug."""
    if cost.currency != currency:
        raise MixedCurrencyError(f"invoice margin ({cost.currency} vs {currency})", 2)
    effective = quantize2(manual_total if manual_total is not None else summed_billout)
    margin = quantize2(effective - Decimal(cost.amount))
    margin_pct = (
        str((margin / effective * 100).quantize(Decimal("0.1")))
        if effective > 0
        else None
    )
    summed = quantize2(summed_billout)
    return {
        "cost": cost.as_dict(),
        "billout": Money(str(effective), currency).as_dict(),
        "summed_billout": Money(str(summed), currency).as_dict(),
        "margin": Money(str(margin), currency).as_dict(),
        "margin_pct": margin_pct,
        "manual_override": manual_total is not None,
    }


def _serialize_invoice(
    invoice: Invoice, party_names: dict[uuid.UUID, str], agg_row, actor
) -> dict:
    seconds = int(agg_row.seconds) if agg_row is not None else 0
    entry_count = int(agg_row.entry_count) if agg_row is not None else 0
    cost = money_from_sum(
        agg_row.cost_sum if agg_row is not None else None,
        agg_row.cost_currencies if agg_row is not None else 0,
        agg_row.cost_currency if agg_row is not None else None,
        context=f"invoice {invoice.name} cost",
        zero_currency=invoice.currency,
    )
    manual = (
        quantize2(Decimal(str(invoice.manual_total)))
        if invoice.manual_total is not None
        else None
    )
    summed = (
        Decimal(str(agg_row.billout_sum))
        if agg_row is not None and agg_row.billout_sum is not None
        else Decimal(0)
    )
    row = {
        "id": str(invoice.id),
        "name": invoice.name,
        "status": invoice.status,
        "currency": invoice.currency,
        "invoice_date": invoice.invoice_date.isoformat() if invoice.invoice_date else None,
        "created_at": invoice.created_at.isoformat() if invoice.created_at else None,
        "sent_at": invoice.sent_at.isoformat() if invoice.sent_at else None,
        "paid_at": invoice.paid_at.isoformat() if invoice.paid_at else None,
        "notes": invoice.notes,
        "manual_total": str(manual) if manual is not None else None,
        "from_party": _party_ref(invoice.from_party_id, party_names),
        "to_party": _party_ref(invoice.to_party_id, party_names),
        "entry_count": entry_count,
        "source_seconds": seconds,
    }
    row.update(
        _margin_block(
            cost=cost,
            summed_billout=summed,
            manual_total=manual,
            currency=invoice.currency,
        )
    )
    return row


def _party_ref(party_id: uuid.UUID | None, names: dict[uuid.UUID, str]) -> dict | None:
    if party_id is None:
        return None
    return {"id": str(party_id), "name": names.get(party_id)}


def _party_names(session: Session, invoices: list[Invoice]) -> dict[uuid.UUID, str]:
    ids = {
        pid
        for inv in invoices
        for pid in (inv.from_party_id, inv.to_party_id)
        if pid is not None
    }
    if not ids:
        return {}
    return dict(
        session.execute(select(Party.id, Party.name).where(Party.id.in_(ids))).all()
    )


# --------------------------------------------------------------- reads


def list_invoices(session: Session, actor) -> list[dict]:
    """Newest-created first (B16), per-invoice aggregates in ONE grouped
    SQL query (kills the unbounded B17 reduction)."""
    if actor.tenant_id is None:
        return []
    invoices = list(
        session.scalars(
            select(Invoice)
            .where(Invoice.tenant_id == actor.tenant_id)
            .order_by(Invoice.created_at.desc(), Invoice.id)
        ).all()
    )
    aggs = {
        row.invoice_id: row
        for row in session.execute(
            _aggregate_stmt(actor, InvoiceLine.invoice_id.label("invoice_id"))
        ).all()
    }
    names = _party_names(session, invoices)
    return [_serialize_invoice(inv, names, aggs.get(inv.id), actor) for inv in invoices]


def invoice_breakdown(session: Session, actor, invoice_id: uuid.UUID) -> list[dict]:
    """Per (client × project) over the SNAPSHOT names (judgment #9 —
    attach-time display, immune to renames). Sort: client
    case-insensitive, then project (B44)."""
    invoice = _get_invoice(session, actor, invoice_id)
    rows = session.execute(
        _aggregate_stmt(
            actor,
            InvoiceLine.snapshot_client_name.label("client"),
            InvoiceLine.snapshot_project_name.label("project"),
        ).where(InvoiceLine.invoice_id == invoice.id)
    ).all()
    out = []
    for row in rows:
        cost = money_from_sum(
            row.cost_sum,
            row.cost_currencies,
            row.cost_currency,
            context=f"invoice {invoice.name} breakdown cost",
            zero_currency=invoice.currency,
        )
        summed = (
            Decimal(str(row.billout_sum)) if row.billout_sum is not None else Decimal(0)
        )
        entry = {
            "client": row.client,
            "project": row.project,
            "row_count": int(row.entry_count),
            "source_seconds": int(row.seconds),
        }
        entry.update(
            _margin_block(
                cost=cost,
                summed_billout=summed,
                manual_total=None,
                currency=invoice.currency,
            )
        )
        out.append(entry)
    out.sort(key=lambda r: ((r["client"] or "").lower(), r["project"] or ""))
    return out


def invoice_detail(session: Session, actor, invoice_id: uuid.UUID) -> dict:
    invoice = _get_invoice(session, actor, invoice_id)
    agg = session.execute(
        _aggregate_stmt(actor).where(InvoiceLine.invoice_id == invoice.id)
    ).one()
    names = _party_names(session, [invoice])
    line_total = session.scalar(
        select(func.count())
        .select_from(InvoiceLine)
        .where(InvoiceLine.invoice_id == invoice.id)
    )
    lines = session.scalars(
        select(InvoiceLine)
        .where(InvoiceLine.invoice_id == invoice.id)
        .order_by(InvoiceLine.attached_at, InvoiceLine.created_at, InvoiceLine.id)
        .limit(DETAIL_LINE_CAP)
    ).all()
    return {
        "invoice": _serialize_invoice(invoice, names, agg, actor),
        "breakdown": invoice_breakdown(session, actor, invoice_id),
        "lines": {
            "total": int(line_total or 0),
            "rows": [
                {
                    "id": str(line.id),
                    "kind": line.kind,
                    "description": line.description,
                    "quantity": str(line.quantity) if line.quantity is not None else None,
                    "rate": str(line.rate) if line.rate is not None else None,
                    "amount": str(line.amount) if line.amount is not None else None,
                    "client": line.snapshot_client_name,
                    "project": line.snapshot_project_name,
                    "attached_at": line.attached_at.isoformat()
                    if line.attached_at
                    else None,
                }
                for line in lines
            ],
        },
    }


# -------------------------------------------------------------- writes


def create_invoice(session: Session, actor, payload: dict) -> dict:
    _require_transfer(actor, "Only an owner can create invoices.")
    _require_writable_books(session, actor)
    name = (payload.get("name") or "").strip()
    if not name:
        raise ServiceError(422, "NAME_REQUIRED", "Invoice name is required.")
    _assert_name_free(session, actor, name)

    status = clean_status(payload.get("status"))
    invoice_date = parse_invoice_date(payload.get("invoice_date")) or default_invoice_date()
    now = datetime.now(UTC)

    invoice = Invoice(
        id=uuid.uuid4(),
        tenant_id=actor.tenant_id,
        name=name,
        status="open",
        currency=_tenant_currency(session, actor),
        invoice_date=invoice_date,
        from_party_id=_tenant_party_id(session, actor),
        to_party_id=_validate_party(session, payload.get("to_party_id")),
        manual_total=parse_manual_total(payload.get("manual_total")),
        notes=(payload.get("notes") or "").strip() or None,
        created_by=actor.real_user_id,  # ALWAYS the real human (B8)
    )
    _apply_status_stamps(invoice, status, now)  # B9 back-dating on create
    session.add(invoice)
    session.flush()

    write_audit(
        session,
        actor,
        action="invoices.create",
        entity_type="invoice",
        entity_id=invoice.id,
        after={"invoice": _scalar_image(invoice)},
    )
    session.commit()
    names = _party_names(session, [invoice])
    return _serialize_invoice(invoice, names, None, actor)


def update_invoice(session: Session, actor, invoice_id: uuid.UUID, patch: dict) -> dict:
    """PATCH-style: only the provided keys change (the router builds the
    dict from the request's set fields)."""
    _require_transfer(actor, "Only an owner can edit invoices.")
    _require_writable_books(session, actor)
    invoice = _get_invoice(session, actor, invoice_id)
    _reject_plus_lifecycle(invoice)
    before = _scalar_image(invoice)

    if "name" in patch:
        name = (patch.get("name") or "").strip()
        if not name:
            raise ServiceError(422, "NAME_REQUIRED", "Invoice name is required.")
        _assert_name_free(session, actor, name, exclude=invoice.id)
        invoice.name = name
    if "status" in patch:
        _apply_status_stamps(invoice, clean_status(patch.get("status")), datetime.now(UTC))
    if "invoice_date" in patch:
        # B66 parity: an empty/malformed posted value writes NULL
        invoice.invoice_date = parse_invoice_date(patch.get("invoice_date"))
    if "to_party_id" in patch:
        invoice.to_party_id = _validate_party(session, patch.get("to_party_id"))
    if "manual_total" in patch:
        invoice.manual_total = parse_manual_total(patch.get("manual_total"))
    if "notes" in patch:
        invoice.notes = (patch.get("notes") or "").strip() or None

    write_audit(
        session,
        actor,
        action="invoices.update",
        entity_type="invoice",
        entity_id=invoice.id,
        before={"invoice": before},
        after={"invoice": _scalar_image(invoice)},
    )
    session.commit()
    agg = session.execute(
        _aggregate_stmt(actor).where(InvoiceLine.invoice_id == invoice.id)
    ).one()
    return _serialize_invoice(invoice, _party_names(session, [invoice]), agg, actor)


def delete_invoice(session: Session, actor, invoice_id: uuid.UUID) -> dict:
    """GUARDED delete (rebuild delta / ADR #007): only a zero-line
    invoice deletes. The legacy delete detached everything silently and
    wiped the apply audit (B70/B71) — here detach is its own audited,
    counted step first."""
    _require_transfer(actor, "Only an owner can delete invoices.")
    _require_writable_books(session, actor)
    invoice = _get_invoice(session, actor, invoice_id)
    _reject_plus_lifecycle(invoice)
    line_count = session.scalar(
        select(func.count())
        .select_from(InvoiceLine)
        .where(InvoiceLine.invoice_id == invoice.id)
    )
    if line_count:
        raise ServiceError(
            409,
            "INVOICE_HAS_LINES",
            f"This invoice still has {line_count} attached "
            f"line{'s' if line_count != 1 else ''}. Detach all entries first —"
            " deleting won't silently unlock them.",
        )
    image = _scalar_image(invoice)
    session.delete(invoice)
    write_audit(
        session,
        actor,
        action="invoices.delete",
        entity_type="invoice",
        entity_id=invoice_id,
        before={"invoice": image},
    )
    session.commit()
    return {"ok": True, "deleted": str(invoice_id)}


# ------------------------------------------------------- apply / detach


def apply_entries(
    session: Session, actor, invoice_id: uuid.UUID, selection: BulkSelection
) -> dict:
    """Attach the selection's PENDING rows to the invoice: one `time`
    line per entry (snapshot display fields as of attach), one
    transaction, verified counts (B23–B33 rebuilt on the bulk engine)."""
    _require_transfer(
        actor, "Only an organization owner can apply entries to invoices."
    )
    _require_writable_books(session, actor)
    invoice = _get_invoice(session, actor, invoice_id)
    _reject_plus_lifecycle(invoice)
    if invoice.status == "paid":
        raise ServiceError(
            409,
            "INVOICE_PAID",
            f'"{invoice.name}" is already paid. Flip it back to sent or open'
            " before applying more entries.",
        )

    rows = collect_scope(session, actor, selection)
    matched = len(rows)
    already = [r for r in rows if r.invoice_line_id is not None]
    candidates = [r for r in rows if r.invoice_line_id is None]

    # THE one eligibility predicate, evaluated in SQL (T-INV-009/010):
    # display status and apply agree by construction.
    eligible_ids: set[uuid.UUID] = set()
    if candidates:
        eligible_ids = {
            row[0]
            for row in session.execute(
                select(TimeEntry.id).where(
                    TimeEntry.id.in_([r.id for r in candidates]),
                    status_condition("pending"),
                )
            ).all()
        }
    eligible = [r for r in candidates if r.id in eligible_ids]
    blocked = [r for r in candidates if r.id not in eligible_ids]
    if matched and not eligible:
        raise ServiceError(
            409,
            "NOTHING_TO_APPLY",
            "No entries left to apply — every selected row is already on an"
            " invoice or blocked.",
        )

    # Single-currency document rule (02 §3): a billout stamp in another
    # currency is a data bug — abort loudly before any write.
    mismatched = [
        r
        for r in eligible
        if r.billout_currency is not None and r.billout_currency != invoice.currency
    ]
    if mismatched:
        raise ServiceError(
            409,
            "CURRENCY_MISMATCH",
            f"{len(mismatched)} entries are stamped in a different currency"
            f" than this {invoice.currency} invoice. Nothing was applied.",
        )

    # Snapshot names from the entity chain (eligible rows always have a
    # project FK) — the SAME display expressions /entries renders.
    joins = EntryJoins()
    name_rows = session.execute(
        joins.select_from(TimeEntry.id, joins.client_expr, joins.project_expr).where(
            TimeEntry.id.in_([r.id for r in eligible])
        )
    ).all()
    names = {row[0]: (row[1], row[2]) for row in name_rows}

    now = datetime.now(UTC)
    before = [snapshot_entry(r) for r in eligible]
    line_ids: list[uuid.UUID] = []
    for r in eligible:
        client_name, project_name = names.get(r.id, (None, None))
        line = InvoiceLine(
            id=uuid.uuid4(),
            invoice_id=invoice.id,
            kind="time",
            description=r.description,
            quantity=(
                quantize2(Decimal(r.duration_seconds) / Decimal(3600))
                if r.duration_seconds is not None
                else None
            ),
            rate=None,  # unused by any R1 read path (migration parity)
            amount=r.billout_amount,
            snapshot_client_name=client_name,
            snapshot_project_name=project_name,
            attached_at=now,
            attached_by=actor.real_user_id,  # the REAL human (B25)
        )
        session.add(line)
        r.invoice_line_id = line.id
        line_ids.append(line.id)

    write_audit(
        session,
        actor,
        action="invoices.apply_entries",
        entity_type="invoice",
        entity_id=invoice.id,
        before={"rows": before},
        after={
            "invoice_id": str(invoice.id),
            "invoice_name": invoice.name,
            "matched": matched,
            "applied": len(eligible),
            "skipped_already_applied": len(already),
            "skipped_blocked": len(blocked),
            "line_ids": [str(i) for i in line_ids],
        },
    )
    session.commit()
    return {
        "invoice_id": str(invoice.id),
        "invoice_name": invoice.name,
        "matched": matched,
        "applied": len(eligible),
        "skipped_already_applied": len(already),
        "skipped_blocked": len(blocked),
        "applied_ids": [str(r.id) for r in eligible],
    }


def detach_entries(
    session: Session,
    actor,
    *,
    ids: list[uuid.UUID],
    expected_count: int,
    invoice_id: uuid.UUID | None = None,
) -> dict:
    """Un-apply: remove the lines, clear the locks, report exact counts.
    EXPLICIT selection only (B62/T-INV-015 — the router refuses
    all_matching before it ever gets here). When `invoice_id` is given,
    only rows on THAT invoice detach (the ?invoice= view's scope);
    otherwise any locked selected row detaches (legacy parity)."""
    _require_transfer(actor, "Only an owner can detach entries.")
    _require_writable_books(session, actor)
    if not ids:
        raise ServiceError(422, "NO_SELECTION", "No entries selected.")

    rows = collect_scope(
        session, actor, BulkSelection(expected_count=expected_count, ids=list(ids))
    )
    matched = len(rows)

    line_ids = [r.invoice_line_id for r in rows if r.invoice_line_id is not None]
    lines: dict[uuid.UUID, InvoiceLine] = {}
    if line_ids:
        lines = {
            line.id: line
            for line in session.scalars(
                select(InvoiceLine).where(InvoiceLine.id.in_(line_ids))
            ).all()
        }

    target: list[tuple[TimeEntry, InvoiceLine | None]] = []
    skipped_unlocked = 0
    skipped_other_invoice = 0
    for r in rows:
        if r.invoice_line_id is None:
            skipped_unlocked += 1  # no-op on already-detached rows (B60)
            continue
        line = lines.get(r.invoice_line_id)
        if invoice_id is not None and (line is None or line.invoice_id != invoice_id):
            skipped_other_invoice += 1
            continue
        target.append((r, line))

    if matched and not target:
        raise ServiceError(
            409,
            "NOTHING_TO_DETACH",
            "No entries left to detach — none of the selected rows are attached"
            + (" to this invoice." if invoice_id is not None else " to an invoice."),
        )

    before_rows = [snapshot_entry(r) for r, _ in target]
    before_lines = [_snapshot_line(line) for _, line in target if line is not None]
    touched_invoices = {line.invoice_id for _, line in target if line is not None}

    for r, line in target:
        r.invoice_line_id = None
        if line is not None:
            session.delete(line)

    write_audit(
        session,
        actor,
        action="invoices.detach_entries",
        entity_type="invoice",
        entity_id=invoice_id or (next(iter(touched_invoices), None)),
        before={"rows": before_rows, "lines": before_lines},
        after={
            "matched": matched,
            "detached": len(target),
            "skipped_unlocked": skipped_unlocked,
            "skipped_other_invoice": skipped_other_invoice,
            "invoice_ids": sorted(str(i) for i in touched_invoices),
        },
    )
    session.commit()
    return {
        "matched": matched,
        "detached": len(target),
        "skipped_unlocked": skipped_unlocked,
        "skipped_other_invoice": skipped_other_invoice,
        "detached_ids": [str(r.id) for r, _ in target],
        # re-apply (the manual undo) is offered only when the detach came
        # off exactly one invoice
        "invoice_id": str(next(iter(touched_invoices)))
        if len(touched_invoices) == 1
        else None,
    }