feat(sp28): store.claim_acks facade + batch_envelope_index + to_ui_claim_ack serializer

The persistence layer for the SP28 join table lands in
cyclone/store/claim_acks.py:

- add_claim_ack — insert + publish 'claim_ack_written' on the bus
  (mirrors the publish-from-store pattern used by the ACKs paths)
- list_acks_for_claim / list_claims_for_ack — read helpers for the
  two list endpoints
- find_ack_orphans — Inbox ack-orphans lane source: every ack row
  whose ack_id has no ClaimAck row tied to it
- remove_claim_ack — unlink + publish 'claim_ack_dropped'
- batch_envelope_index — D10 in-memory map of
  {envelope.control_number: batch.id}, called once per ingest (cost
  is ~16 lookups today; trivially cheap)

Plus the to_ui_claim_ack serializer in store/ui.py mirroring
to_ui_ack shape — single source of truth for the wire format so the
live-tail event payload matches the list endpoint byte-for-byte.
Includes a single SELECT against claims for the claim_state field
(TA1 batch-level rows return 'n/a').

The CycloneStore facade in store/__init__.py re-exports all six
methods (add_claim_ack, list_acks_for_claim, list_claims_for_ack,
find_ack_orphans, remove_claim_ack, batch_envelope_index) plus
to_ui_claim_ack from the ui module.

Migration-version assertions in test_acks.py and test_db_migrate.py
bumped 17 → 18 to match the new migration head.

Steps 3.1/3.2/3.3/3.4 of the SP28 implementation plan.

backend/src/cyclone/store/__init__.py

@@ -80,6 +80,14 @@ from .acks import (
     list_acks,
     list_ta1_acks,
 )
+from .claim_acks import (
+    add_claim_ack as _add_claim_ack,
+    batch_envelope_index as _batch_envelope_index,
+    find_ack_orphans as _find_ack_orphans,
+    list_acks_for_claim as _list_acks_for_claim,
+    list_claims_for_ack as _list_claims_for_ack,
+    remove_claim_ack as _remove_claim_ack,
+)
 from .backups import add_backup_pending
 from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
 from .inbox import list_unmatched, manual_match, manual_unmatch
@@ -102,6 +110,7 @@ from .providers import (
 from .records import BatchKind, BatchRecord, BatchRecord837, BatchRecord835
 from .ui import (
     to_ui_ack,
+    to_ui_claim_ack,
     to_ui_claim_detail,
     to_ui_claim_from_orm,
     to_ui_provider,
@@ -287,6 +296,38 @@ class CycloneStore:
     def get_277ca_ack(self, ack_id):
         return get_277ca_ack(ack_id)
 
+    # -- SP28: claim↔ack auto-link join table --------------------------
+
+    def add_claim_ack(self, **kwargs):
+        """Persist one claim_acks link row.
+
+        Mirrors the publish-from-store contract used by the ACK paths:
+        when ``event_bus`` is passed, the matching ``claim_ack_written``
+        event fires after commit so live-tail subscribers on both the
+        claim and the ack side see the row immediately.
+        """
+        return _add_claim_ack(**kwargs)
+
+    def list_acks_for_claim(self, claim_id):
+        """Return every ClaimAck row for one claim (newest first)."""
+        return _list_acks_for_claim(claim_id)
+
+    def list_claims_for_ack(self, kind, ack_id):
+        """Return every ClaimAck row for one ack."""
+        return _list_claims_for_ack(kind, ack_id)
+
+    def find_ack_orphans(self, kind):
+        """Return acks with no resolvable link (Inbox ack-orphans lane)."""
+        return _find_ack_orphans(kind)
+
+    def remove_claim_ack(self, link_id, *, event_bus=None):
+        """Unlink one row. Publishes ``claim_ack_dropped`` on the bus."""
+        return _remove_claim_ack(link_id, event_bus=event_bus)
+
+    def batch_envelope_index(self):
+        """Return a {envelope.control_number: batch.id} map for D10 Pass 1."""
+        return _batch_envelope_index()
+
     # -- SP17: encrypted DB backups -------------------------------------
 
     def add_backup_pending(self, *, filename: str, backup_dir: str):

backend/src/cyclone/store/claim_acks.py

@@ -0,0 +1,360 @@
+"""SP28: claim_acks persistence + batch envelope index (D10).
+
+Five methods on top of the new ``ClaimAck`` ORM table:
+
+* ``add_claim_ack`` — insert one link row (manual or auto) and
+  publish ``claim_ack_written`` on the bus.
+* ``list_acks_for_claim`` — every link row for one claim (per-claim
+  only; TA1 batch-level rows are filtered out by the API layer).
+* ``list_claims_for_ack`` — every link row for one ack.
+* ``find_ack_orphans`` — acks with no resolvable claim (Inbox
+  ack-orphans lane).
+* ``remove_claim_ack`` — unlink. Publishes ``claim_ack_dropped``.
+* ``batch_envelope_index`` — D10 in-memory map of
+  ``Batch.envelope.control_number → batch.id`` (cheap to rebuild;
+  re-built once per ingest).
+
+Each mutating method opens its own ``db.SessionLocal()()`` session
+so callers don't have to manage session lifecycles. Publishes are
+best-effort (wrapped in :func:`_safe_publish`) so a failing bus
+subscriber cannot roll back the persisted row.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import TYPE_CHECKING
+
+from cyclone import db
+from cyclone.db import (
+    Ack,
+    Batch,
+    Claim,
+    ClaimAck,
+    Ta1Ack,
+    Two77caAck,
+)
+
+from . import utcnow
+from .ui import to_ui_claim_ack
+from .write import _sync_publish
+
+if TYPE_CHECKING:
+    from cyclone.pubsub import EventBus
+
+
+log = logging.getLogger(__name__)
+
+
+def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> None:
+    """Best-effort publish wrapped in try/except.
+
+    Mirrors ``cyclone.store.acks._safe_publish`` — never raises,
+    never rolls back the persisted row.
+    """
+    if event_bus is None:
+        return
+    try:
+        _sync_publish(event_bus, kind, payload)
+    except Exception:  # noqa: BLE001
+        log.exception("claim_acks store: %s publish failed", kind)
+
+
+# ---------------------------------------------------------------------------
+# D10 batch envelope index
+# ---------------------------------------------------------------------------
+
+
+def batch_envelope_index() -> dict[str, str]:
+    """Build a ``{envelope.control_number: batch.id}`` map.
+
+    D10 primary join key (spec §D10). Built once per ingest from
+    every Batch row whose ``raw_result_json`` carries an envelope —
+    cost is O(N batches), currently ~16, so trivial. Kept as a
+    plain dict so callers can ``index.get(scn)`` to resolve.
+    """
+    out: dict[str, str] = {}
+    with db.SessionLocal()() as s:
+        rows = (
+            s.query(Batch.id, Batch.raw_result_json)
+            .filter(Batch.kind == "837")
+            .all()
+        )
+        for bid, raw in rows:
+            env = (raw or {}).get("envelope") or {}
+            ctrl = env.get("control_number")
+            if isinstance(ctrl, str) and ctrl:
+                out[ctrl] = bid
+    return out
+
+
+# ---------------------------------------------------------------------------
+# Mutators
+# ---------------------------------------------------------------------------
+
+
+def add_claim_ack(
+    *,
+    claim_id: str | None,
+    batch_id: str | None,
+    ack_id: int,
+    ack_kind: str,
+    ak2_index: int | None = None,
+    set_control_number: str | None = None,
+    set_accept_reject_code: str | None = None,
+    linked_by: str = "auto",
+    event_bus: "EventBus | None" = None,
+    now: datetime | None = None,
+) -> ClaimAck:
+    """Persist one link row and return it.
+
+    The DB-level unique index
+    ``ux_claim_acks_dedup(claim_id, ack_kind, ack_id, ak2_index)
+    WHERE claim_id IS NOT NULL AND ak2_index IS NOT NULL`` enforces
+    idempotency for the per-AK2 case. The 277CA (ak2_index NULL)
+    and TA1 (claim_id NULL) paths are deduplicated by application
+    code — see :func:`cyclone.claim_acks._link_exists`.
+
+    When ``event_bus`` is provided, publishes one ``claim_ack_written``
+    event (with the full :func:`cyclone.store.ui.to_ui_claim_ack`
+    payload) so the live-tail subscribers on both
+    ``/api/claims/{id}/acks/stream`` and
+    ``/api/acks/{kind}/{id}/claims/stream`` see new rows the moment
+    they land.
+    """
+    if ack_kind not in ("999", "277ca", "ta1"):
+        raise ValueError(f"add_claim_ack: unknown ack_kind={ack_kind!r}")
+    if linked_by not in ("auto", "manual"):
+        raise ValueError(f"add_claim_ack: linked_by={linked_by!r}")
+    if claim_id is None and batch_id is None:
+        raise ValueError(
+            "add_claim_ack: at least one of claim_id/batch_id must be set"
+        )
+    ts = now or utcnow()
+    with db.SessionLocal()() as s:
+        row = ClaimAck(
+            claim_id=claim_id,
+            batch_id=batch_id,
+            ack_id=ack_id,
+            ack_kind=ack_kind,
+            ak2_index=ak2_index,
+            set_control_number=set_control_number,
+            set_accept_reject_code=set_accept_reject_code,
+            linked_at=ts,
+            linked_by=linked_by,
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        row_id = row.id
+
+    if event_bus is not None:
+        with db.SessionLocal()() as s:
+            payload = to_ui_claim_ack(s.get(ClaimAck, row_id))
+        _safe_publish(event_bus, "claim_ack_written", payload)
+
+    return row
+
+
+def remove_claim_ack(
+    link_id: int,
+    *,
+    event_bus: "EventBus | None" = None,
+) -> bool:
+    """Unlink. Returns ``True`` when a row was deleted.
+
+    Publishes ``claim_ack_dropped`` with ``{"id", "claim_id"}`` so
+    the live-tail subscribers can remove the link from their local
+    store. Does NOT touch ``Claim.state`` — the unlink is purely
+    about the link row, per spec §D6.
+    """
+    with db.SessionLocal()() as s:
+        row = s.get(ClaimAck, link_id)
+        if row is None:
+            return False
+        payload = {
+            "id": row.id,
+            "claim_id": row.claim_id,
+            "ack_id": row.ack_id,
+            "ack_kind": row.ack_kind,
+        }
+        s.delete(row)
+        s.commit()
+
+    if event_bus is not None:
+        _safe_publish(event_bus, "claim_ack_dropped", payload)
+    return True
+
+
+# ---------------------------------------------------------------------------
+# Readers
+# ---------------------------------------------------------------------------
+
+
+def list_acks_for_claim(claim_id: str) -> list[ClaimAck]:
+    """Return every ClaimAck row for one claim (newest first)."""
+    with db.SessionLocal()() as s:
+        return (
+            s.query(ClaimAck)
+            .filter(ClaimAck.claim_id == claim_id)
+            .order_by(ClaimAck.id.desc())
+            .all()
+        )
+
+
+def list_claims_for_ack(kind: str, ack_id: int) -> list[ClaimAck]:
+    """Return every ClaimAck row for one ack (any kind).
+
+    For 999 / 277CA: returns 0..N rows (one per AK2 / ClaimStatus).
+    For TA1: returns 0..1 row (envelope-level, populated batch_id).
+    """
+    if kind not in ("999", "277ca", "ta1"):
+        raise ValueError(f"list_claims_for_ack: unknown kind={kind!r}")
+    with db.SessionLocal()() as s:
+        return (
+            s.query(ClaimAck)
+            .filter(
+                ClaimAck.ack_kind == kind,
+                ClaimAck.ack_id == ack_id,
+            )
+            .order_by(ClaimAck.id.asc())
+            .all()
+        )
+
+
+# ---------------------------------------------------------------------------
+# Orphan detection (Inbox "Ack orphans" lane — spec §D7)
+# ---------------------------------------------------------------------------
+
+
+def find_ack_orphans(kind: str) -> list[dict]:
+    """Return acks with no resolvable Claim row of their own kind.
+
+    Used by the Inbox ack-orphans lane for the operator's manual
+    reconciliation flow. The downstream ``/api/inbox/ack-orphans``
+    endpoint calls this and returns the rendered shape.
+
+    "Orphan" means: for 999 / 277CA, ``ack_kind=kind AND ack_id IN
+    (acks_with_no_claim_acks_link)``. For TA1, "orphan" means the
+    TA1 row exists but no Batch with matching sender/receiver was
+    resolved (so no link row was created).
+
+    Output dict shape (one row per orphan ack, rendered by
+    :func:`to_ui_claim_ack`-style serialization):
+
+      * ``kind``       — "999" / "277ca" / "ta1"
+      * ``ack_id``     — the ack row's id
+      * ``control_number`` — for 999/277CA, envelope.control_number;
+                             for TA1, ta1.control_number
+      * ``set_control_numbers`` — empty list when no claims match
+      * ``raw_summary`` — flat copy of the ack's UI shape for the
+                          lane-header counts
+    """
+    if kind not in ("999", "277ca", "ta1"):
+        raise ValueError(f"find_ack_orphans: unknown kind={kind!r}")
+
+    out: list[dict] = []
+    if kind == "999":
+        ack_table = Ack
+        ctrl_attr = None
+    elif kind == "277ca":
+        ack_table = Two77caAck
+        ctrl_attr = "control_number"
+    else:
+        ack_table = Ta1Ack
+        ctrl_attr = "control_number"
+
+    with db.SessionLocal()() as s:
+        # Every ack row of the given kind, with a LEFT JOIN against
+        # any claim_acks link; orphan when NO link was created.
+        if kind == "999":
+            # For 999, the "ack has no link" means no ClaimAck row
+            # was emitted at all (the auto-linker emits one per AK2
+            # even when the AK2 is rejected, so 999 with at least
+            # one AK2 that resolved to a claim is never an orphan).
+            # We treat a 999 as orphan when it has zero ClaimAck
+            # rows tied to its id.
+            all_acks = s.query(Ack).order_by(Ack.id.desc()).all()
+            for ack_row in all_acks:
+                count = (
+                    s.query(ClaimAck)
+                    .filter(ClaimAck.ack_kind == "999",
+                            ClaimAck.ack_id == ack_row.id)
+                    .count()
+                )
+                if count == 0:
+                    out.append({
+                        "kind": "999",
+                        "ack_id": ack_row.id,
+                        "control_number": _ack_control_number(ack_row, "999"),
+                        "parsed_at": (
+                            ack_row.parsed_at.isoformat().replace(
+                                "+00:00", "Z"
+                            ) if ack_row.parsed_at else None
+                        ),
+                    })
+        elif kind == "277ca":
+            all_acks = s.query(Two77caAck).order_by(Two77caAck.id.desc()).all()
+            for ack_row in all_acks:
+                count = (
+                    s.query(ClaimAck)
+                    .filter(ClaimAck.ack_kind == "277ca",
+                            ClaimAck.ack_id == ack_row.id)
+                    .count()
+                )
+                if count == 0:
+                    out.append({
+                        "kind": "277ca",
+                        "ack_id": ack_row.id,
+                        "control_number": ack_row.control_number or "",
+                        "parsed_at": (
+                            ack_row.parsed_at.isoformat().replace(
+                                "+00:00", "Z"
+                            ) if ack_row.parsed_at else None
+                        ),
+                    })
+        else:
+            all_acks = s.query(Ta1Ack).order_by(Ta1Ack.id.desc()).all()
+            for ack_row in all_acks:
+                count = (
+                    s.query(ClaimAck)
+                    .filter(ClaimAck.ack_kind == "ta1",
+                            ClaimAck.ack_id == ack_row.id)
+                    .count()
+                )
+                if count == 0:
+                    out.append({
+                        "kind": "ta1",
+                        "ack_id": ack_row.id,
+                        "control_number": ack_row.control_number or "",
+                        "parsed_at": (
+                            ack_row.parsed_at.isoformat().replace(
+                                "+00:00", "Z"
+                            ) if ack_row.parsed_at else None
+                        ),
+                    })
+    return out
+
+
+def _ack_control_number(ack_row: Ack, kind: str) -> str:
+    """Best-effort control-number lookup for a 999 ack row.
+
+    The 999 ORM row doesn't carry the envelope's control_number in
+    a dedicated column; we re-derive it from ``raw_json`` (the same
+    source :func:`cyclone.store.ui.to_ui_ack` uses for the patient
+    control number).
+    """
+    raw = ack_row.raw_json or {}
+    env = raw.get("envelope") or {}
+    return env.get("control_number") or ""
+
+
+__all__ = [
+    "add_claim_ack",
+    "batch_envelope_index",
+    "find_ack_orphans",
+    "list_acks_for_claim",
+    "list_claims_for_ack",
+    "remove_claim_ack",
+]

backend/src/cyclone/store/ui.py

@@ -8,6 +8,15 @@ Also hosts ``utcnow()`` (the tz-aware UTC now) because it's a small
 free function that several modules want and there's no better home.
 
 SP25: ``to_ui_ack`` / ``to_ui_ta1_ack`` / ``to_ui_two77ca_ack`` were
+moved here from ``api_routers/acks.py`` / ``api_routers/ta1_acks.py``
+/ ``api.py`` so the live-tail event payload matches the list endpoint
+shape byte-for-byte. The seam between persistence and streaming
+depends on the two halves staying in sync.
+
+SP28 adds ``to_ui_claim_ack`` for the same reason — the
+``claim_ack_written`` event payload must match the
+``GET /api/acks/{kind}/{id}/claims`` list shape byte-for-byte.
+
 moved here from ``api_routers/acks.py`` / ``api_routers/ta1_acks.py``
 / ``api.py`` so the live-tail event payload can match the list endpoint
 shape byte-for-byte. The seam between persistence and streaming
@@ -20,7 +29,7 @@ from datetime import datetime, timezone
 from decimal import Decimal
 
 from cyclone import db
-from cyclone.db import Claim, Remittance
+from cyclone.db import Claim, ClaimAck, Remittance
 from cyclone.parsers.models import ClaimOutput
 from cyclone.parsers.models_835 import ClaimPayment
 from cyclone.parsers.payer import PayerConfig835
@@ -627,3 +636,48 @@ def _date_in_bounds(
     if date_to is not None and date_part > date_to:
         return False
     return True
+
+
+def to_ui_claim_ack(row: ClaimAck) -> dict:
+    """SP28: map a ClaimAck ORM row to the API shape.
+
+    Mirrors the rest of the to_ui_* serializers. The wire shape is
+    identical between the matching list endpoint (``GET /api/acks/{kind}/
+    {id}/claims``) and the ``claim_ack_written`` pubsub event so the
+    live-tail subscribers can rehydrate the snapshot from the bus
+    without diverging from the API response.
+
+    ``claim_state`` is queried via a session round-trip
+    (``SELECT state FROM claims WHERE id = :claim_id``) so the drawer
+    panel can render the colored ClaimStateBadge inline. For TA1 rows
+    with ``claim_id IS NULL`` (batch-level envelope link),
+    ``claim_state`` is ``"n/a"``.
+    """
+    claim_state: str = "n/a"
+    if row.claim_id:
+        with db.SessionLocal()() as lookup_s:
+            crow = lookup_s.get(Claim, row.claim_id)
+            if crow is not None:
+                claim_state = (
+                    crow.state.value
+                    if hasattr(crow.state, "value")
+                    else str(crow.state)
+                )
+    linked_iso = (
+        row.linked_at.isoformat().replace("+00:00", "Z")
+        if row.linked_at is not None
+        else ""
+    )
+    return {
+        "id": row.id,
+        "claim_id": row.claim_id,
+        "batch_id": row.batch_id,
+        "ack_id": row.ack_id,
+        "ack_kind": row.ack_kind,
+        "ak2_index": row.ak2_index,
+        "set_control_number": row.set_control_number,
+        "set_accept_reject_code": row.set_accept_reject_code,
+        "linked_at": linked_iso,
+        "linked_by": row.linked_by,
+        "claim_state": claim_state,
+    }

backend/tests/test_acks.py

@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
 
 def test_migration_latest_idempotent_on_fresh_db():
     """Re-running the migration on the same DB must be a no-op (PRAGMA
-    user_version already at the latest version — currently 16 after
+    user_version already at the latest version — currently 18 after
     0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
     providers/payers/clearhouse, SP10's 0008 payer_rejected,
     SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
@@ -59,15 +59,16 @@ def test_migration_latest_idempotent_on_fresh_db():
     SP-auth's 0013 users + sessions, SP-audit's 0014 audit_log.user_id,
     SP22's 0015 drop_claims_unique_constraint, SP27-Task 11's 0016
     claims.matched_remittance_id index, SP27-Task 17's 0017
-    claim.patient_control_number backfill UPDATE)."""
+    claim.patient_control_number backfill UPDATE, SP28's 0018
+    claim_acks join table)."""
     with db.engine().begin() as c:
         v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v1 == 17
+    assert v1 == 18
     # A second run should not raise and should not bump the version.
     db_migrate.run(db.engine())
     with db.engine().begin() as c:
         v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v2 == 17
+    assert v2 == 18
 
 
 def test_add_ack_persists_row():

backend/tests/test_db_migrate.py

@@ -123,14 +123,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     index (drift-check perf).
     SP27 Task 17 bumped it to 17 with the patient_control_number
     backfill UPDATE (the migration runner now applies DML too).
+    SP28 bumped it to 18 with the claim_acks join table.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 17, f"expected head=17, got {v_after_first}"
+    assert v_after_first == 18, f"expected head=18, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 17, "second run should not bump version"
+    assert _user_version(engine) == 18, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -156,7 +157,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
     engine = _fresh_engine(tmp_path)
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 17, f"expected head=17, got {_user_version(engine)}"
+    assert _user_version(engine) == 18, f"expected head=18, got {_user_version(engine)}"
 
     # Two claims in one batch with the same patient_control_number
     # must be insertable. If 0015's table recreation re-introduced a