merge: SP21 store split into main

Splits the 2,995-LOC backend/src/cyclone/store.py into a
cyclone/store/ subpackage (13 sibling modules + __init__.py
facade) with no public API change. Adds dashboard_kpis,
check_matched_pair_drift, _claim_state_str (SP27) into the
right homes, and surfaces 3 private helpers (_claim_status_from_validation,
_persist_835_remit, _remittance_835_row) from the facade for
existing test imports.

Spec: docs/superpowers/specs/2026-06-21-cyclone-store-split-design.md
Plan: docs/superpowers/plans/2026-06-29-cyclone-store-split-resume.md

Tests at baseline: 1 failed (pre-existing isolation flake),
1176 passed, 10 skipped.

backend/src/cyclone/store/__init__.py

@@ -0,0 +1,333 @@
+"""SQLAlchemy-backed batch store for parsed X12 files.
+
+The package exposes a single ``CycloneStore`` class and a module-level
+singleton (``store``). All persistence flows through SQLAlchemy
+sessions via ``db.SessionLocal()()`` (the double-paren function-style
+accessor).
+
+This ``__init__.py`` is the facade — it re-exports every public name
+plus the 3 private helpers imported by tests (``_claim_status_from_validation``,
+``_persist_835_remit``, ``_remittance_835_row``) from the sibling modules:
+
+    exceptions      AlreadyMatchedError, NotMatchedError, InvalidStateError
+    records         BatchKind, BatchRecord, BatchRecord837, BatchRecord835
+    orm_builders    ORM row builders + the 3 private-helper re-exports
+    ui              UI serializers (to_ui_*)
+    write           add_record + private event/reconcile helpers
+    batches         Batch reads + _BatchesShim (test-cleanup shim)
+    claim_detail    Single-claim reads + matched-pair drift audit
+    kpis            Dashboard aggregation
+    acks            999 / TA1 / 277CA ACK persistence
+    backups         Backup-pending marker inserts
+    inbox           Manual match/unmatch + lane listing
+    providers       Provider/payer/clearhouse config
+
+The ``CycloneStore`` class below keeps its current method signatures as
+thin delegations for back-compat with existing callers and tests.
+
+Public API (preserved verbatim):
+    CycloneStore, store, BatchRecord, BatchRecord837, BatchRecord835,
+    BatchKind, AlreadyMatchedError, NotMatchedError, InvalidStateError,
+    utcnow, dashboard_kpis, check_matched_pair_drift
+
+Backward-compat shims for tests:
+    ``_lock`` — a threading.RLock used by 7 test files for cleanup.
+    ``_batches.clear()`` — wipes all rows from the DB tables; the
+        ``_BatchesShim`` class lives in ``batches.py``.
+"""
+
+from __future__ import annotations
+
+import threading
+from datetime import datetime, timezone
+
+
+def utcnow() -> datetime:
+    """tz-aware UTC `datetime` (replaces the old `utcnow_iso` string helper)."""
+    return datetime.now(timezone.utc)
+
+
+from . import write
+from .batches import (
+    _BatchesShim,
+    _row_to_record,
+    all_batches,
+    get_batch,
+    get_record,
+    list_batches,
+    load_two_for_diff,
+)
+from .claim_detail import (
+    check_matched_pair_drift,
+    count_claims,
+    count_remittances,
+    distinct_providers,
+    get_claim_detail,
+    get_remittance,
+    iter_claims,
+    iter_remittances,
+    recent_activity,
+    summarize_remittances,
+)
+from .acks import (
+    add_277ca_ack,
+    add_999_ack,
+    add_ta1_ack,
+    get_277ca_ack,
+    get_ack,
+    get_ta1_ack,
+    list_277ca_acks,
+    list_acks,
+    list_ta1_acks,
+)
+from .backups import add_backup_pending
+from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
+from .inbox import list_unmatched, manual_match, manual_unmatch
+from .kpis import dashboard_kpis
+from .orm_builders import (
+    _claim_status_from_validation,
+    _persist_835_remit,
+    _remittance_835_row,
+)
+from .providers import (
+    ensure_clearhouse_seeded,
+    get_clearhouse,
+    get_payer_config,
+    get_provider,
+    list_payers,
+    list_providers,
+    update_clearhouse,
+    upsert_provider,
+)
+from .records import BatchKind, BatchRecord, BatchRecord837, BatchRecord835
+from .ui import (
+    to_ui_claim_detail,
+    to_ui_claim_from_orm,
+    to_ui_provider,
+    to_ui_remittance_from_orm,
+    to_ui_remittance_with_adjustments,
+)
+
+
+# ---------------------------------------------------------------------------
+# UI mappers: ORM rows → simpler UI types.
+# ---------------------------------------------------------------------------
+# Backward-compat shim: tests called ``_batches.clear()`` on the in-memory
+# store. The DB-backed store doesn't have an in-memory list, so we expose
+# a tiny shim object whose ``.clear()`` wipes the DB. The class itself
+# lives in ``batches.py`` (imported above as ``_BatchesShim``); the
+# ``CycloneStore.__init__`` below instantiates that imported class.
+# ---------------------------------------------------------------------------
+
+
+# ---------------------------------------------------------------------------
+# CycloneStore: the SQLAlchemy-backed facade.
+# ---------------------------------------------------------------------------
+
+
+class CycloneStore:
+    """SQLAlchemy-backed facade over the parsed X12 store.
+
+    Each public method opens a short-lived session via
+    ``db.SessionLocal()()`` so callers don't have to manage session
+    lifecycles. Concurrency is handled by the SQLAlchemy engine; the
+    ``_lock`` attribute is a no-op ``RLock`` retained for backward
+    compatibility with code that wrapped cleanup in a lock context.
+    """
+
+    def __init__(self) -> None:
+        self._lock = threading.RLock()
+        self._batches = _BatchesShim()
+
+    # -- write path -----------------------------------------------------
+
+    def add(self, record: BatchRecord, *, event_bus=None):
+        """Persist a parsed batch (837P or 835)."""
+        return write.add_record(record, event_bus=event_bus)
+
+    def _publish_events_sync(self, event_bus, record, claim_ids, remit_ids):
+        return write.publish_events_sync(event_bus, record, claim_ids, remit_ids)
+
+    @staticmethod
+    def _sync_publish(event_bus, kind: str, payload: dict) -> None:
+        return write._sync_publish(event_bus, kind, payload)
+
+    # -- read path ------------------------------------------------------
+
+    def get_batch(self, batch_id: str) -> dict | None:
+        return get_batch(batch_id)
+
+    def get(self, batch_id: str) -> BatchRecord | None:
+        return get_record(batch_id)
+
+    def list(self, *, limit: int = 100) -> list[BatchRecord]:
+        return list_batches(limit=limit)
+
+    def get_remittance(self, remittance_id: str) -> dict | None:
+        return get_remittance(remittance_id)
+
+    def get_claim_detail(self, claim_id: str) -> dict | None:
+        return get_claim_detail(claim_id)
+
+    def all(self) -> list[BatchRecord]:
+        return all_batches()
+
+    def load_two_for_diff(
+        self,
+        a_id: str,
+        b_id: str,
+    ) -> tuple[BatchRecord, BatchRecord]:
+        return load_two_for_diff(a_id, b_id)
+
+    def iter_claims(self, **kwargs):
+        return iter_claims(**kwargs)
+
+    def iter_remittances(self, **kwargs):
+        return iter_remittances(**kwargs)
+
+    # -- count helpers (SP27 Task 13b) ---------------------------------
+    #
+    # The list endpoints (``/api/claims`` and ``/api/remittances``)
+    # previously computed ``total`` by calling ``iter_*`` with default
+    # ``limit=100`` and taking ``len(...)``. With 60k+ claims and 800+
+    # remits in production, the reported total silently capped at 100
+    # even when the UI rendered a "100" KPI tile and a 100-row table —
+    # the page looked complete but the population was 600× larger. These
+    # helpers reuse the iter's filter pipeline with an effectively
+    # unbounded ``limit`` so the count reflects the true DB population,
+    # not a 100-row sample. Mirrors Dashboard's "server-aggregated
+    # counts" fix from commit 59c3275.
+    def count_claims(
+        self,
+        *,
+        batch_id: str | None = None,
+        status: str | None = None,
+        provider_npi: str | None = None,
+        payer: str | None = None,
+        date_from: str | None = None,
+        date_to: str | None = None,
+    ) -> int:
+        """Count claims that would be returned by ``iter_claims``."""
+        return count_claims(
+            batch_id=batch_id, status=status, provider_npi=provider_npi,
+            payer=payer, date_from=date_from, date_to=date_to,
+        )
+
+    def count_remittances(
+        self,
+        *,
+        batch_id: str | None = None,
+        payer: str | None = None,
+        claim_id: str | None = None,
+        date_from: str | None = None,
+        date_to: str | None = None,
+    ) -> int:
+        """Count remittances that would be returned by ``iter_remittances``."""
+        return count_remittances(
+            batch_id=batch_id, payer=payer, claim_id=claim_id,
+            date_from=date_from, date_to=date_to,
+        )
+
+    def summarize_remittances(
+        self,
+        *,
+        batch_id: str | None = None,
+        payer: str | None = None,
+        claim_id: str | None = None,
+        date_from: str | None = None,
+        date_to: str | None = None,
+    ) -> dict:
+        """Return ``{count, total_paid, total_adjustments}`` summed over
+        the remittance population that ``iter_remittances`` would return.
+        """
+        return summarize_remittances(
+            batch_id=batch_id, payer=payer, claim_id=claim_id,
+            date_from=date_from, date_to=date_to,
+        )
+
+    def distinct_providers(self) -> list[dict]:
+        return distinct_providers()
+
+    def recent_activity(self, *, limit: int = 200) -> list[dict]:
+        return recent_activity(limit=limit)
+
+    # -- 999 ACKs (SP3 P3 T13) -------------------------------------------
+
+    def add_ack(self, **kwargs):
+        return add_999_ack(**kwargs)
+
+    def list_acks(self):
+        return list_acks()
+
+    def get_ack(self, ack_id):
+        return get_ack(ack_id)
+
+    # -- TA1 (Interchange Acknowledgment) -------------------------------
+
+    def add_ta1_ack(self, **kwargs):
+        return add_ta1_ack(**kwargs)
+
+    def list_ta1_acks(self):
+        return list_ta1_acks()
+
+    def get_ta1_ack(self, ack_id):
+        return get_ta1_ack(ack_id)
+
+    # -- 277CA (SP10) --------------------------------------------------
+
+    def add_277ca_ack(self, **kwargs):
+        return add_277ca_ack(**kwargs)
+
+    def list_277ca_acks(self):
+        return list_277ca_acks()
+
+    def get_277ca_ack(self, ack_id):
+        return get_277ca_ack(ack_id)
+
+    # -- SP17: encrypted DB backups -------------------------------------
+
+    def add_backup_pending(self, *, filename: str, backup_dir: str):
+        return add_backup_pending(filename=filename, backup_dir=backup_dir)
+
+    # -- manual reconciliation (T12) -----------------------------------
+
+    def list_unmatched(self, *, kind="both"):
+        return list_unmatched(kind=kind)
+
+    def manual_match(self, claim_id, remit_id):
+        return manual_match(claim_id, remit_id)
+
+    def manual_unmatch(self, claim_id):
+        return manual_unmatch(claim_id)
+
+    # ------------------------------------------------------------------
+    # SP9: providers / payers / payer_configs / clearhouse
+    # ------------------------------------------------------------------
+
+    def list_providers(self, *, is_active=True):
+        return list_providers(is_active=is_active)
+
+    def get_provider(self, npi):
+        return get_provider(npi)
+
+    def upsert_provider(self, provider):
+        return upsert_provider(provider)
+
+    def list_payers(self, *, is_active=True):
+        return list_payers(is_active=is_active)
+
+    def get_payer_config(self, payer_id, transaction_type):
+        return get_payer_config(payer_id, transaction_type)
+
+    def get_clearhouse(self):
+        return get_clearhouse()
+
+    def update_clearhouse(self, block):
+        return update_clearhouse(block)
+
+    def ensure_clearhouse_seeded(self):
+        return ensure_clearhouse_seeded()
+
+
+# Module-level singleton — same import path the old InMemoryStore used.
+store = CycloneStore()

backend/src/cyclone/store/acks.py

@@ -0,0 +1,174 @@
+"""ACK persistence — 999 / TA1 / 277CA acknowledgements.
+
+Each ACK type has its own ORM row (Ack / Ta1Ack / Two77caAck). The
+write methods are simple inserts; the list/get methods are simple
+queries. Fail-soft: persistence errors are logged but do not block
+parsing.
+"""
+
+from datetime import date
+
+from cyclone import db
+from cyclone.db import Ack
+
+from . import utcnow
+
+
+# -- 999 ACKs (SP3 P3 T13) -------------------------------------------
+
+def add_999_ack(
+    *,
+    source_batch_id: str,
+    accepted_count: int,
+    rejected_count: int,
+    received_count: int,
+    ack_code: str,
+    raw_json: dict,
+) -> db.Ack:
+    """Persist a 999 ACK row and return it.
+
+    ``source_batch_id`` must reference an existing ``batches.id``.
+    For received 999s with no source batch the caller should pass a
+    synthetic id (e.g. ``"999-<interchange_control_number>"``) —
+    see ``/api/parse-999`` for that policy.
+
+    ``raw_json`` is the full ``ParseResult999`` model dump; the
+    detail endpoint surfaces it without re-parsing the original
+    X12 text.
+    """
+    with db.SessionLocal()() as s:
+        row = Ack(
+            source_batch_id=source_batch_id,
+            accepted_count=accepted_count,
+            rejected_count=rejected_count,
+            received_count=received_count,
+            ack_code=ack_code,
+            parsed_at=utcnow(),
+            raw_json=raw_json,
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+
+def list_acks() -> list[db.Ack]:
+    """Return every 999 ACK row, newest first (auto-increment id desc)."""
+    with db.SessionLocal()() as s:
+        return (
+            s.query(Ack)
+            .order_by(Ack.id.desc())
+            .all()
+        )
+
+def get_ack( ack_id: int) -> db.Ack | None:
+    """Return a single ACK row by id, or ``None`` if not found."""
+    with db.SessionLocal()() as s:
+        return s.get(Ack, ack_id)
+
+# -- TA1 (Interchange Acknowledgment) -------------------------------
+
+def add_ta1_ack(
+    *,
+    source_batch_id: str,
+    control_number: str,
+    interchange_date: date | None,
+    interchange_time: str | None,
+    ack_code: str,
+    note_code: str | None,
+    ack_generated_date: date | None,
+    sender_id: str,
+    receiver_id: str,
+    raw_json: dict,
+) -> db.Ta1Ack:
+    """Persist a TA1 (Interchange Acknowledgment) row and return it.
+
+    Mirrors :meth:`add_ack` for the lower-level envelope ack. The
+    flat columns are promoted out of ``raw_json`` so the list
+    endpoint can sort/filter without a JSON parse; the full
+    ``ParseResultTa1`` stays in ``raw_json`` for the detail endpoint.
+    """
+    with db.SessionLocal()() as s:
+        row = db.Ta1Ack(
+            source_batch_id=source_batch_id,
+            control_number=control_number,
+            interchange_date=interchange_date,
+            interchange_time=interchange_time,
+            ack_code=ack_code,
+            note_code=note_code,
+            ack_generated_date=ack_generated_date,
+            sender_id=sender_id,
+            receiver_id=receiver_id,
+            parsed_at=utcnow(),
+            raw_json=raw_json,
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+
+def list_ta1_acks() -> list[db.Ta1Ack]:
+    """Return every TA1 ACK row, newest first (auto-increment id desc).
+
+    Mirrors :meth:`list_acks` — the API endpoint slices to its own
+    ``limit`` so the ``total`` field reflects the full row count.
+    """
+    with db.SessionLocal()() as s:
+        return (
+            s.query(db.Ta1Ack)
+            .order_by(db.Ta1Ack.id.desc())
+            .all()
+        )
+
+def get_ta1_ack( ack_id: int) -> db.Ta1Ack | None:
+    """Return a single TA1 ACK row by id, or ``None`` if not found."""
+    with db.SessionLocal()() as s:
+        return s.get(db.Ta1Ack, ack_id)
+
+# -- 277CA (SP10) --------------------------------------------------
+
+def add_277ca_ack(
+    *,
+    source_batch_id: str,
+    control_number: str,
+    accepted_count: int,
+    rejected_count: int,
+    paid_count: int,
+    pended_count: int,
+    raw_json: dict,
+) -> db.Two77caAck:
+    """Persist a 277CA (Claim Acknowledgment) row and return it.
+
+    Mirrors :meth:`add_ack` but for the claim-level ack. The
+    per-claim status detail stays in ``raw_json``; only the four
+    counts are promoted so the list endpoint stays fast.
+    """
+    with db.SessionLocal()() as s:
+        row = db.Two77caAck(
+            source_batch_id=source_batch_id,
+            control_number=control_number,
+            accepted_count=accepted_count,
+            rejected_count=rejected_count,
+            paid_count=paid_count,
+            pended_count=pended_count,
+            parsed_at=utcnow(),
+            raw_json=raw_json,
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+
+def list_277ca_acks() -> list[db.Two77caAck]:
+    """Return every 277CA ACK row, newest first (auto-increment id desc)."""
+    with db.SessionLocal()() as s:
+        return (
+            s.query(db.Two77caAck)
+            .order_by(db.Two77caAck.id.desc())
+            .all()
+        )
+
+def get_277ca_ack( ack_id: int) -> db.Two77caAck | None:
+    """Return a single 277CA ACK row by id, or ``None`` if not found."""
+    with db.SessionLocal()() as s:
+        return s.get(db.Two77caAck, ack_id)
+

backend/src/cyclone/store/backups.py

@@ -0,0 +1,39 @@
+"""Backup-pending marker inserts — SP17 encrypted DB backups.
+
+``add_backup_pending()`` inserts a ``pending`` row for a backup that is
+about to start; the BackupService updates ``status`` / ``size_bytes``
+/ ``db_fingerprint`` / ``table_count`` / ``completed_at`` after the
+encrypted blob lands on disk.
+"""
+
+from cyclone import db
+
+from . import utcnow
+
+
+# -- SP17: encrypted DB backups -------------------------------------
+
+def add_backup_pending(*, filename: str, backup_dir: str) -> db.DbBackup:
+    """Insert a ``pending`` row for a backup that is about to start.
+
+    The BackupService fills in ``status`` / ``size_bytes`` /
+    ``db_fingerprint`` / ``table_count`` / ``completed_at`` after
+    the encrypted blob lands on disk.
+    """
+    with db.SessionLocal()() as s:
+        row = db.DbBackup(
+            filename=filename,
+            backup_dir=backup_dir,
+            size_bytes=0,
+            db_fingerprint=None,
+            table_count=0,
+            created_at=utcnow(),
+            completed_at=None,
+            status="pending",
+            error_message=None,
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+

backend/src/cyclone/store/batches.py

@@ -0,0 +1,155 @@
+"""Batch read APIs and the legacy _BatchesShim.
+
+The ``_BatchesShim`` class is retained for back-compat with the
+``with store._lock: store._batches.clear()`` test-cleanup idiom used
+in 7 test files. Its ``clear()`` method wipes all rows from the DB
+tables so tests that depended on a fresh in-memory list per-test get
+a fresh DB state per-test.
+"""
+
+from __future__ import annotations
+
+from datetime import timezone
+
+from cyclone import db
+from cyclone.db import (
+    ActivityEvent,
+    Batch,
+    CasAdjustment,
+    Claim,
+    Match,
+    Remittance,
+)
+from cyclone.parsers.models import ParseResult
+from cyclone.parsers.models_835 import ParseResult835
+
+from .records import BatchRecord, BatchRecord837, BatchRecord835
+
+
+class _BatchesShim:
+    """Drop-in replacement for the old in-memory ``_batches`` list.
+
+    ``clear()`` removes every row from the DB tables in FK-safe order.
+    Other list operations are not implemented because the only call site
+    is the ``clear()`` inside the test fixtures (``test_api_gets.py`` and
+    ``test_api_parse_persists.py``).
+    """
+
+    def clear(self) -> None:  # type: ignore[no-untyped-def]
+        with db.SessionLocal()() as s:
+            s.query(ActivityEvent).delete()
+            s.query(Match).delete()
+            s.query(CasAdjustment).delete()
+            s.query(Remittance).delete()
+            s.query(Claim).delete()
+            s.query(Batch).delete()
+            s.commit()
+
+
+def _row_to_record(row: Batch) -> BatchRecord:
+    """Rehydrate a ``BatchRecord`` (837 or 835) from a Batch ORM row.
+
+    The full ``ParseResult`` / ``ParseResult835`` lives in
+    ``raw_result_json`` (stashed at insert time). Re-parsing JSON
+    here means callers get the same typed Pydantic object the old
+    in-memory store handed out, so api.py and tests that do
+    ``rec.result.claims`` keep working unchanged.
+
+    SQLite drops tz info on round-trip even though the column type
+    is ``DateTime(timezone=True)``. We re-attach UTC so the
+    ``BatchRecord`` validator (``parsed_at must be tz-aware``)
+    passes.
+    """
+    if row.kind == "835":
+        result_cls = ParseResult835
+    else:
+        result_cls = ParseResult
+    payload = row.raw_result_json or {}
+    result = result_cls.model_validate(payload)
+    parsed_at = row.parsed_at
+    if parsed_at is not None and parsed_at.tzinfo is None:
+        parsed_at = parsed_at.replace(tzinfo=timezone.utc)
+    record_cls = BatchRecord835 if row.kind == "835" else BatchRecord837
+    return record_cls(
+        id=row.id,
+        kind=row.kind,
+        input_filename=row.input_filename,
+        parsed_at=parsed_at,
+        result=result,
+    )
+
+
+def get_batch(batch_id: str) -> dict | None:
+    """Return a summary dict for ``batch_id`` or ``None`` if missing.
+
+    The dict shape matches what ``/api/batches/{id}`` callers need:
+    ``id``, ``kind``, ``input_filename``, ``parsed_at``, and the
+    full ``result`` (raw_result_json) as a dict.
+    """
+    with db.SessionLocal()() as s:
+        row = s.get(Batch, batch_id)
+        if row is None:
+            return None
+        return {
+            "id": row.id,
+            "kind": row.kind,
+            "input_filename": row.input_filename,
+            "parsed_at": row.parsed_at,
+            "result": row.raw_result_json,
+        }
+
+
+def get_record(batch_id: str) -> BatchRecord | None:
+    """Return the ``BatchRecord`` for ``batch_id`` or ``None``.
+
+    Preserves the in-memory store contract: callers get a Pydantic
+    ``BatchRecord`` (subclass ``BatchRecord837`` / ``BatchRecord835``)
+    with ``.id``, ``.kind``, ``.input_filename``, ``.parsed_at``,
+    and ``.result`` (typed ``ParseResult`` / ``ParseResult835``).
+    """
+    with db.SessionLocal()() as s:
+        row = s.get(Batch, batch_id)
+        if row is None:
+            return None
+        return _row_to_record(row)
+
+
+def list_batches(*, limit: int = 100) -> list[BatchRecord]:
+    """Return up to ``limit`` ``BatchRecord``s, newest first."""
+    with db.SessionLocal()() as s:
+        rows = (
+            s.query(Batch)
+            .order_by(Batch.parsed_at.desc())
+            .limit(limit)
+            .all()
+        )
+        return [_row_to_record(r) for r in rows]
+
+
+def all_batches() -> list[BatchRecord]:
+    """Return every ``BatchRecord``, oldest first (no pagination)."""
+    with db.SessionLocal()() as s:
+        rows = s.query(Batch).order_by(Batch.parsed_at.asc()).all()
+        return [_row_to_record(r) for r in rows]
+
+
+def load_two_for_diff(a_id: str, b_id: str) -> tuple[BatchRecord, BatchRecord]:
+    """Load two batches by id for the side-by-side diff view.
+
+    Returns ``(a, b)`` as ``BatchRecord`` objects. Raises
+    :class:`LookupError` when either id is missing — the API layer
+    catches it and maps it to ``404 Not Found`` (matching the
+    ``GET /api/batches/{id}`` contract). The two loads happen in
+    independent sessions so a transient failure on one side can't
+    poison the other.
+
+    Used exclusively by :mod:`cyclone.batch_diff` via the
+    ``/api/batch-diff`` endpoint.
+    """
+    a = get_record(a_id)
+    if a is None:
+        raise LookupError(f"batch {a_id} not found")
+    b = get_record(b_id)
+    if b is None:
+        raise LookupError(f"batch {b_id} not found")
+    return a, b

backend/src/cyclone/store/claim_detail.py

@@ -0,0 +1,710 @@
+"""Claim- and remittance-detail read APIs.
+
+Includes the only large non-write query in the store:
+``get_claim_detail`` which joins Claim + CasAdjustment + ActivityEvent
+history for the right-drawer UI.
+
+Also hosts ``check_matched_pair_drift()`` — the SP27 startup invariant
+audit that returns the count of Claim↔Remit pairs where
+``matched_remittance_id`` doesn't agree with the FK in ``remittances.claim_id``.
+It lives here (not in its own module) because it reads the same pair of
+tables as ``get_claim_detail``'s matched-remittance summary.
+"""
+
+from __future__ import annotations
+
+from datetime import timezone
+from decimal import Decimal
+
+from cyclone import db
+from cyclone.db import (
+    ActivityEvent,
+    CasAdjustment,
+    Claim,
+    ClaimState,
+    Remittance,
+)
+from .ui import (
+    CLAIM_DETAIL_HISTORY_LIMIT,
+    _date_in_bounds,
+    _iso_z,
+    _svc_to_wire_dict,
+    to_ui_claim_detail,
+    to_ui_claim_from_orm,
+    to_ui_provider,
+    to_ui_remittance_from_orm,
+    to_ui_remittance_with_adjustments,
+)
+
+
+def get_remittance(remittance_id: str) -> dict | None:
+    """Return a UI-shaped remittance dict with ``adjustments`` array.
+
+    Joins the persisted ``CasAdjustment`` rows for ``remittance_id``
+    and labels each via :mod:`cyclone.parsers.cas_codes`. Returns
+    ``None`` when the remittance is not found so the API layer can
+    map that to a 404.
+
+    SP7: also returns the per-line SVC composites
+    (``serviceLinePayments``) and the CLP-level (claim-level) CAS
+    bucket (``claimLevelAdjustments``) so the remit drawer can show
+    per-line payments + adjustments without a second fetch.
+    """
+    with db.SessionLocal()() as s:
+        row = s.get(Remittance, remittance_id)
+        if row is None:
+            return None
+        cas_rows = (
+            s.query(CasAdjustment)
+            .filter(CasAdjustment.remittance_id == remittance_id)
+            .all()
+        )
+        parsed_at = (
+            row.batch.parsed_at if row.batch is not None else row.received_at
+        )
+        if parsed_at is not None and parsed_at.tzinfo is None:
+            parsed_at = parsed_at.replace(tzinfo=timezone.utc)
+        body = to_ui_remittance_with_adjustments(
+            row,
+            batch_id=row.batch_id,
+            parsed_at=parsed_at,
+            cas_rows=cas_rows,
+        )
+        # SP7: per-line SVC composites + claim-level CAS bucket.
+        from cyclone.db import ServiceLinePayment as SLP
+        slps = (
+            s.query(SLP)
+            .filter(SLP.remittance_id == remittance_id)
+            .order_by(SLP.line_number)
+            .all()
+        )
+        body["serviceLinePayments"] = [
+            _svc_to_wire_dict(svc) for svc in slps
+        ]
+        body["claimLevelAdjustments"] = [
+            {
+                "id": c.id,
+                "group_code": c.group_code,
+                "reason_code": c.reason_code,
+                "amount": str(Decimal(str(c.amount))),
+                "quantity": (
+                    str(Decimal(str(c.quantity)))
+                    if c.quantity is not None
+                    else None
+                ),
+            }
+            for c in cas_rows
+            if c.service_line_payment_id is None
+        ]
+        return body
+
+
+def get_claim_detail(claim_id: str) -> dict | None:
+    """Return the SP4 detail-drawer shape for one claim, or ``None``.
+
+    Drives ``GET /api/claims/{claim_id}``. Returns the spec-shaped
+    dict from :func:`to_ui_claim_detail` (header + state + parties +
+    validation + service lines + diagnoses + raw segments) stitched
+    with the claim's recent activity history and, if paired, a
+    matched-remittance summary.
+
+    Returns ``None`` when ``claim_id`` is not in the DB so the API
+    layer can map that to a 404 — the URL-driven drawer
+    distinguishes "claim doesn't exist" from "fetch failed" (the
+    spec §3.4 calls for a distinct 404 state in the drawer).
+
+    The history is capped at :data:`CLAIM_DETAIL_HISTORY_LIMIT`
+    (50, per the spec) and ordered ``ts DESC`` so the most recent
+    event is first. The status string in ``matchedRemittance``
+    follows the same ``reconciled``/``received`` mapping used by
+    :func:`to_ui_remittance_from_orm`.
+    """
+    # Lazy import — same pattern used throughout this module to
+    # avoid a circular store ↔ db import on cold start.
+    from cyclone import db as _db
+
+    with _db.SessionLocal()() as s:
+        row = s.get(Claim, claim_id)
+        if row is None:
+            return None
+
+        history_rows = (
+            s.query(ActivityEvent)
+            .filter(ActivityEvent.claim_id == claim_id)
+            .order_by(ActivityEvent.ts.desc())
+            .limit(CLAIM_DETAIL_HISTORY_LIMIT)
+            .all()
+        )
+
+        # Claim.batch_id is FK NOT NULL with ON DELETE CASCADE, so
+        # ``row.batch`` is always populated in normal flow. Re-attach
+        # UTC only when SQLite drops the tzinfo on read.
+        parsed_at = row.batch.parsed_at
+        if parsed_at.tzinfo is None:
+            parsed_at = parsed_at.replace(tzinfo=timezone.utc)
+
+        detail = to_ui_claim_detail(
+            row,
+            batch_id=row.batch_id,
+            parsed_at=parsed_at,
+        )
+
+        detail["stateHistory"] = [
+            {
+                "kind": ev.kind,
+                # SQLite drops tzinfo on read; rows are stored UTC
+                # at write time (see ``add`` / ``manual_match``),
+                # so re-attach UTC if needed to keep the spec
+                # contract that ``ts`` ends in Z.
+                "ts": _iso_z(ev.ts),
+                "batchId": ev.batch_id,
+                "remittanceId": ev.remittance_id,
+            }
+            for ev in history_rows
+        ]
+
+        if row.matched_remittance_id is not None:
+            remit = s.get(Remittance, row.matched_remittance_id)
+            if remit is not None:
+                status = (
+                    "reconciled"
+                    if remit.status_code in ("21", "22")
+                    else "received"
+                )
+                detail["matchedRemittance"] = {
+                    "id": remit.id,
+                    "totalPaid": float(remit.total_paid or 0),
+                    "status": status,
+                    "receivedAt": _iso_z(remit.received_at),
+                }
+            # If the remittance was deleted out from under the FK
+            # (the FK is ``ON DELETE SET NULL`` so the column is
+            # already cleared in normal flow), the matched_remittance_id
+            # would be None here and we wouldn't enter this branch.
+            # If the FK is non-null but the row is gone (e.g. tests
+            # that bypass the cascade), fall through with the
+            # default ``None`` — the UI shows "no match" rather
+            # than crashing.
+
+        # SP7 §5.2: slim per-line projection so the ServiceLinesTable
+        # can show Paid + Adjustments columns without a second fetch.
+        # The 837 side is keyed by ``claim_service_line_number`` (the
+        # 1-based line number from raw_json) since 837 service lines
+        # are not a separate ORM table.
+        from cyclone.db import (
+            LineReconciliation, ServiceLinePayment, CasAdjustment,
+        )
+        slim_lrs = list(
+            s.query(LineReconciliation)
+            .filter(LineReconciliation.claim_id == claim_id)
+            .all()
+        )
+        svc_ids_for_cas = [
+            lr.service_line_payment_id
+            for lr in slim_lrs
+            if lr.service_line_payment_id is not None
+        ]
+        cas_sums_by_svc: dict = {}
+        svc_by_id_slim: dict = {}
+        if svc_ids_for_cas:
+            cas_rows = (
+                s.query(CasAdjustment.service_line_payment_id, CasAdjustment.amount)
+                .filter(CasAdjustment.service_line_payment_id.in_(svc_ids_for_cas))
+                .all()
+            )
+            from collections import defaultdict
+            agg = defaultdict(lambda: Decimal("0"))
+            for svc_id, amount in cas_rows:
+                agg[svc_id] += Decimal(str(amount))
+            cas_sums_by_svc = {k: str(v) for k, v in agg.items()}
+            for svc in (
+                s.query(ServiceLinePayment)
+                .filter(ServiceLinePayment.id.in_(svc_ids_for_cas))
+                .all()
+            ):
+                svc_by_id_slim[svc.id] = svc
+
+        slim_by_num: dict = {
+            lr.claim_service_line_number: lr
+            for lr in slim_lrs
+            if lr.claim_service_line_number is not None
+        }
+        line_reconciliation_slim: list = []
+        for sl in detail["serviceLines"]:
+            ln = sl.get("lineNumber")
+            lr = slim_by_num.get(ln)
+            if lr is None:
+                line_reconciliation_slim.append({
+                    "lineNumber": ln,
+                    "status": "unmatched_837_only",
+                    "paid": None,
+                    "adjustmentsSum": None,
+                })
+                continue
+            svc = (
+                svc_by_id_slim.get(lr.service_line_payment_id)
+                if lr.service_line_payment_id
+                else None
+            )
+            line_reconciliation_slim.append({
+                "lineNumber": ln,
+                "status": lr.status,
+                "paid": str(Decimal(str(svc.payment))) if svc else None,
+                "adjustmentsSum": (
+                    cas_sums_by_svc.get(lr.service_line_payment_id)
+                    if lr.service_line_payment_id
+                    else None
+                ),
+            })
+        detail["lineReconciliation"] = line_reconciliation_slim
+
+        return detail
+
+
+def iter_claims(
+    *,
+    batch_id: str | None = None,
+    status: str | None = None,
+    provider_npi: str | None = None,
+    payer: str | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    sort: str | None = None,
+    order: str = "desc",
+    limit: int = 100,
+    offset: int = 0,
+) -> list[dict]:
+    """Return UI-shaped claim dicts from the DB.
+
+    Filters mirror the in-memory version. The ``payer`` filter is
+    a case-insensitive substring on the payer's ``name``, recovered
+    from each claim's ``raw_json`` payload (the DB stores it there
+    because ``Claim`` itself only carries ``payer_id``).
+    """
+    with db.SessionLocal()() as s:
+        q = s.query(Claim)
+        if batch_id is not None:
+            q = q.filter(Claim.batch_id == batch_id)
+        if status is not None:
+            q = q.filter(Claim.state == ClaimState(status))
+        if provider_npi is not None:
+            q = q.filter(Claim.provider_npi == provider_npi)
+
+        rows = q.all()
+        # Bulk-load matched-remittance totals so the UI's "Received"
+        # KPI + per-claim received_amount reflect real paid amounts
+        # rather than always-0. One SQL roundtrip for the whole page
+        # rather than per-claim lookups.
+        matched_ids = [
+            r.matched_remittance_id
+            for r in rows
+            if r.matched_remittance_id
+        ]
+        received_by_remit: dict[str, float] = {}
+        if matched_ids:
+            for rid, total_paid in (
+                s.query(Remittance.id, Remittance.total_paid)
+                .filter(Remittance.id.in_(matched_ids))
+                .all()
+            ):
+                received_by_remit[rid] = float(total_paid or 0)
+
+        out: list[dict] = []
+        for r in rows:
+            raw = r.raw_json or {}
+            bp = raw.get("billing_provider", {})
+            payer_obj = raw.get("payer", {})
+            sub = raw.get("subscriber", {})
+            claim_hdr = raw.get("claim", {})
+            service_lines = raw.get("service_lines", [])
+            parsed_at_iso = (
+                r.batch.parsed_at.isoformat().replace("+00:00", "Z")
+                if r.batch is not None
+                else ""
+            )
+            cpt = (
+                service_lines[0].get("procedure", {}).get("code", "")
+                if service_lines
+                else ""
+            )
+            out.append({
+                "id": r.id,
+                "patientName": (
+                    f"{sub.get('first_name', '')} "
+                    f"{sub.get('last_name', '')}".strip()
+                ),
+                "providerNpi": bp.get("npi") or r.provider_npi or "",
+                "payerName": payer_obj.get("name") or "",
+                "cptCode": cpt,
+                "billedAmount": float(r.charge_amount or 0),
+                "receivedAmount": received_by_remit.get(
+                    r.matched_remittance_id, 0.0
+                ),
+                "status": r.state.value if hasattr(r.state, "value") else str(r.state),
+                "state": r.state.value if hasattr(r.state, "value") else str(r.state),
+                "denialReason": None,
+                "submissionDate": parsed_at_iso,
+                "batchId": r.batch_id,
+                "parsedAt": parsed_at_iso,
+                # Keep these so we can sort on them in-memory below.
+                "_sort_billedAmount": float(r.charge_amount or 0),
+                "_sort_submissionDate": parsed_at_iso,
+            })
+
+    if payer is not None:
+        needle = payer.casefold()
+        out = [
+            c for c in out
+            if needle in (c.get("payerName") or "").casefold()
+        ]
+    out = [
+        c for c in out
+        if _date_in_bounds(c, "submissionDate", date_from, date_to)
+    ]
+    if sort is not None:
+        out.sort(
+            key=lambda c: c.get(f"_sort_{sort}", 0) or 0,
+            reverse=(order == "desc"),
+        )
+    # Drop the private sort keys before returning.
+    for c in out:
+        c.pop("_sort_billedAmount", None)
+        c.pop("_sort_submissionDate", None)
+    return out[offset:offset + limit]
+
+
+def iter_remittances(
+    *,
+    batch_id: str | None = None,
+    payer: str | None = None,
+    claim_id: str | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+    sort: str | None = None,
+    order: str = "desc",
+    limit: int = 100,
+    offset: int = 0,
+) -> list[dict]:
+    """Return UI-shaped remittance dicts from the DB."""
+    with db.SessionLocal()() as s:
+        q = s.query(Remittance)
+        if batch_id is not None:
+            q = q.filter(Remittance.batch_id == batch_id)
+        if claim_id is not None:
+            q = q.filter(Remittance.claim_id == claim_id)
+
+        rows = q.all()
+        # Bulk-fetch all CAS rows for these remittances in one query
+        # (SP3 P2 follow-up — fixes the list-view's empty adjustments
+        # expansion). N+1-free.
+        cas_by_remit: dict[str, list] = {}
+        if rows:
+            from cyclone.parsers.cas_codes import reason_label
+            cas_rows = (
+                s.query(CasAdjustment)
+                .filter(CasAdjustment.remittance_id.in_([r.id for r in rows]))
+                .all()
+            )
+            for c in cas_rows:
+                cas_by_remit.setdefault(c.remittance_id, []).append(c)
+
+        out: list[dict] = []
+        for r in rows:
+            raw = r.raw_json or {}
+            parsed_at_iso = (
+                r.batch.parsed_at.isoformat().replace("+00:00", "Z")
+                if r.batch is not None
+                else r.received_at.isoformat().replace("+00:00", "Z")
+            )
+            payer_name = ""
+            if r.batch is not None and r.batch.raw_result_json:
+                payer_name = (
+                    r.batch.raw_result_json.get("payer", {}).get("name", "")
+                )
+            adjustments = [
+                {
+                    "group": c.group_code,
+                    "reason": c.reason_code,
+                    "label": reason_label(c.group_code, c.reason_code),
+                    "amount": float(c.amount),
+                    "quantity": float(c.quantity) if c.quantity is not None else None,
+                }
+                for c in cas_by_remit.get(r.id, [])
+            ]
+            out.append({
+                "id": r.id,
+                "claimId": r.claim_id or "",
+                "payerName": payer_name,
+                "paidAmount": float(r.total_paid or 0),
+                "adjustmentAmount": float(r.adjustment_amount or 0),
+                "status": (
+                    "reconciled" if r.status_code in ("21", "22")
+                    else "received"
+                ),
+                "denialReason": None,
+                "validationWarnings": [],
+                "receivedDate": r.received_at.isoformat().replace("+00:00", "Z"),
+                "batchId": r.batch_id,
+                "parsedAt": parsed_at_iso,
+                "adjustments": adjustments,
+                "_sort_receivedDate": r.received_at.isoformat().replace("+00:00", "Z"),
+            })
+
+    if payer is not None:
+        out = [r for r in out if r.get("payerName") == payer]
+    out = [
+        r for r in out
+        if _date_in_bounds(r, "receivedDate", date_from, date_to)
+    ]
+    if sort is not None:
+        out.sort(
+            key=lambda r: r.get(f"_sort_{sort}", 0) or 0,
+            reverse=(order == "desc"),
+        )
+    for r in out:
+        r.pop("_sort_receivedDate", None)
+    return out[offset:offset + limit]
+
+
+def distinct_providers() -> list[dict]:
+    """Group claims by NPI and return one row per provider."""
+    with db.SessionLocal()() as s:
+        rows = s.query(Claim).all()
+        by_npi: dict[str, dict] = {}
+        for r in rows:
+            npi = r.provider_npi or ""
+            if npi not in by_npi:
+                raw = r.raw_json or {}
+                bp = raw.get("billing_provider", {})
+                by_npi[npi] = to_ui_provider(
+                    npi=npi,
+                    name=bp.get("name") or "",
+                    tax_id=bp.get("tax_id"),
+                    address=None,
+                    city=None,
+                    state=None,
+                    zip=None,
+                    phone=None,
+                    claim_count=0,
+                    outstanding_ar=0.0,
+                )
+            by_npi[npi]["claimCount"] += 1
+        return list(by_npi.values())
+
+
+def recent_activity(*, limit: int = 200) -> list[dict]:
+    """Return recent activity events from the DB, newest first.
+
+    SP21 Task 2.5: each row also carries ``claimId`` and
+    ``remittanceId`` (read from the ORM columns) so the Dashboard's
+    Recent-activity card can route clicks to the right entity
+    drawer via ``src/lib/event-routing.ts``. Both are nullable
+    strings; the wire shape uses camelCase keys to match the
+    existing ``npi`` / ``amount`` fields and the frontend
+    ``Activity`` interface.
+    """
+    with db.SessionLocal()() as s:
+        rows = (
+            s.query(ActivityEvent)
+            .order_by(ActivityEvent.ts.desc())
+            .limit(limit)
+            .all()
+        )
+        return [
+            {
+                "id": f"ae-{r.id}",
+                "kind": r.kind,
+                "message": (r.payload_json or {}).get("message", ""),
+                "timestamp": r.ts.isoformat().replace("+00:00", "Z"),
+                "npi": (r.payload_json or {}).get("npi"),
+                "amount": (r.payload_json or {}).get("amount"),
+                "claimId": r.claim_id,
+                "remittanceId": r.remittance_id,
+            }
+            for r in rows
+        ]
+
+
+def check_matched_pair_drift() -> int:
+    """Audit the ``Claim.matched_remittance_id`` ↔ ``Remittance.claim_id``
+    FK pair at startup. Non-blocking (SP27 Task 11).
+
+    The matched pair is a denormalized FK pair maintained transactionally
+    by ``manual_match``, ``manual_unmatch``, and ``reconcile.run``. A
+    pre-existing mismatch (e.g. a row written before a state migration
+    that added one column but not the other) would otherwise stay
+    invisible until the next operator pair attempt fails confusingly.
+    This check logs the count + up to N examples so operators can
+    investigate without booting the system.
+
+    Returns the number of drifted rows (0 means clean). Does not
+    raise; bootstrap continues even if drift is detected.
+
+    Count semantics: this returns *drifted rows*, not *drifted pairs*.
+    A single broken pair (``Claim.matched_remittance_id = X`` AND
+    ``Remittance.claim_id = Y != nil`` with neither pointing back)
+    can produce TWO drifted rows — one in case A (the claim) and
+    one in case B (the remit). In practice drift is almost always
+    asymmetric (one side NULL), so count == count(pairs); for the
+    fully-symmetric minority, divide by ~2 when alerting. Real drift
+    should be fixed by repairing the writer path, not by counting.
+
+    Cases:
+      A. Claim ``matched_remittance_id = X`` but the paired remit X's
+         ``claim_id`` is either NULL or doesn't point back to the claim.
+      B. Remit ``claim_id = A`` but the paired claim A's
+         ``matched_remittance_id`` is either NULL or doesn't point back.
+    """
+    import logging
+    from sqlalchemy import select
+
+    log = logging.getLogger(__name__)
+
+    with db.SessionLocal()() as s:
+        # Case A: claim says X is paired, but X.claim_id doesn't point back.
+        case_a = list(
+            s.execute(
+                select(
+                    Claim.id.label("claim_id"),
+                    Claim.matched_remittance_id.label("claimed_remit_id"),
+                    Remittance.claim_id.label("remit_points_to"),
+                )
+                .outerjoin(
+                    Remittance,
+                    Remittance.id == Claim.matched_remittance_id,
+                )
+                .where(Claim.matched_remittance_id.is_not(None))
+                .where(
+                    (Remittance.claim_id.is_(None))
+                    | (Remittance.claim_id != Claim.id)
+                )
+            ).all()
+        )
+        # Case B: remit says A is paired, but A.matched_remittance_id
+        # doesn't point back.
+        case_b = list(
+            s.execute(
+                select(
+                    Remittance.id.label("remit_id"),
+                    Remittance.claim_id.label("claimed_claim_id"),
+                    Claim.matched_remittance_id.label("claim_points_to"),
+                )
+                .outerjoin(
+                    Claim,
+                    Claim.id == Remittance.claim_id,
+                )
+                .where(Remittance.claim_id.is_not(None))
+                .where(
+                    (Claim.matched_remittance_id.is_(None))
+                    | (Claim.matched_remittance_id != Remittance.id)
+                )
+            ).all()
+        )
+
+    total = len(case_a) + len(case_b)
+    if total == 0:
+        log.info("matched-pair drift check: 0 mismatches (clean)")
+        return 0
+
+    log.warning(
+        "matched-pair drift check: %d mismatched pair(s) (showing up to 5 "
+        "of each). Investigate via SELECT against claim / remittance; "
+        "manual re-pair via /api/claims/{id}/manual-match will repair.",
+        total,
+    )
+    for r in case_a[:5]:
+        log.warning(
+            "  case A: claim %s -> remit %s, but remit.claim_id=%r",
+            r.claim_id,
+            r.claimed_remit_id,
+            r.remit_points_to,
+        )
+    for r in case_b[:5]:
+        log.warning(
+            "  case B: remit %s -> claim %s, but claim.matched_remittance_id=%r",
+            r.remit_id,
+            r.claimed_claim_id,
+            r.claim_points_to,
+        )
+    return total
+
+# ---------------------------------------------------------------------------
+# Aggregate counters (SP25 / SP27): full-population counts and sums that
+# the /api/* endpoints expose so page-local reductions (25 rows + live-tail
+# delta) can never silently understate the true population.
+# ---------------------------------------------------------------------------
+
+_ITER_UNBOUNDED = 2**31 - 1
+
+
+def count_claims(
+    *,
+    batch_id: str | None = None,
+    status: str | None = None,
+    provider_npi: str | None = None,
+    payer: str | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+) -> int:
+    """Count claims that would be returned by ``iter_claims``.
+
+    Same filter parameters as ``iter_claims`` (excluding
+    ``sort``/``order``/``limit``/``offset``, which don't affect cardinality).
+    """
+    rows = iter_claims(
+        batch_id=batch_id, status=status, provider_npi=provider_npi,
+        payer=payer, date_from=date_from, date_to=date_to,
+        sort=None, order="desc", limit=_ITER_UNBOUNDED, offset=0,
+    )
+    return len(rows)
+
+
+def count_remittances(
+    *,
+    batch_id: str | None = None,
+    payer: str | None = None,
+    claim_id: str | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+) -> int:
+    """Count remittances that would be returned by ``iter_remittances``.
+
+    Same filter parameters as ``iter_remittances`` (excluding
+    ``sort``/``order``/``limit``/``offset``).
+    """
+    rows = iter_remittances(
+        batch_id=batch_id, payer=payer, claim_id=claim_id,
+        date_from=date_from, date_to=date_to,
+        sort=None, order="desc", limit=_ITER_UNBOUNDED, offset=0,
+    )
+    return len(rows)
+
+
+def summarize_remittances(
+    *,
+    batch_id: str | None = None,
+    payer: str | None = None,
+    claim_id: str | None = None,
+    date_from: str | None = None,
+    date_to: str | None = None,
+) -> dict:
+    """Return ``{count, total_paid, total_adjustments}`` summed over the
+    remittance population that ``iter_remittances`` would return under
+    the same filters. Backs ``GET /api/remittances/summary`` — the
+    Remittances page's KPI tiles (added in SP27).
+    """
+    rows = iter_remittances(
+        batch_id=batch_id, payer=payer, claim_id=claim_id,
+        date_from=date_from, date_to=date_to,
+        sort=None, order="desc", limit=_ITER_UNBOUNDED, offset=0,
+    )
+    total_paid = 0.0
+    total_adjustments = 0.0
+    for r in rows:
+        total_paid += float(r.get("paidAmount") or 0)
+        total_adjustments += float(r.get("adjustmentAmount") or 0)
+    return {
+        "count": len(rows),
+        "total_paid": total_paid,
+        "total_adjustments": total_adjustments,
+    }

backend/src/cyclone/store/exceptions.py

@@ -0,0 +1,41 @@
+"""Exception types raised by CycloneStore and its domain modules.
+
+These are part of the public API — callers (API endpoints) catch them
+and translate to HTTP 409 Conflict responses.
+"""
+
+
+class AlreadyMatchedError(Exception):
+    """Raised by ``CycloneStore.manual_match`` when the claim is already paired.
+
+    The claim's ``matched_remittance_id`` is set, so any new pairing would
+    clobber an existing match. Callers (the T15 API endpoint) should surface
+    this as a 409 Conflict.
+    """
+
+
+class NotMatchedError(Exception):
+    """Raised by ``CycloneStore.manual_unmatch`` when the claim has no match.
+
+    Mirrors ``AlreadyMatchedError`` for the unpair operation. Same HTTP
+    treatment: 409 Conflict at the API layer.
+    """
+
+
+class InvalidStateError(Exception):
+    """Raised when an apply_* pure fn returns a skipped ApplyIntent.
+
+    ``reconcile.apply_payment`` / ``apply_reversal`` may return
+    ``skipped=True`` (e.g. claim already in a terminal state, or reversal
+    on a non-paid claim). The store surfaces that as ``InvalidStateError``
+    rather than silently pairing. The T15 API endpoint maps this to a
+    409 Conflict and echoes ``current_state`` and ``activity_kind`` so
+    the UI can render a precise message.
+    """
+
+    def __init__(self, current_state: str, activity_kind: str = "invalid_state"):
+        self.current_state = current_state
+        self.activity_kind = activity_kind
+        super().__init__(
+            f"invalid state {current_state} for apply (kind={activity_kind})"
+        )

backend/src/cyclone/store/inbox.py

@@ -0,0 +1,315 @@
+"""Inbox lane state and manual match/unmatch operations.
+
+``list_unmatched`` returns the 5-lane inbox view (unmatched claims +
+unmatched remits). ``manual_match`` and ``manual_unmatch`` invoke the
+reconcile pure functions and persist the resulting Match row (or
+remove it). Invalid state transitions surface as ``InvalidStateError``.
+"""
+
+from cyclone import db
+from cyclone.db import ActivityEvent, Claim, ClaimState, Match, Remittance
+
+from .exceptions import AlreadyMatchedError, InvalidStateError, NotMatchedError
+from . import utcnow
+from .ui import to_ui_claim_from_orm, to_ui_remittance_from_orm
+
+
+# -- manual reconciliation (T12) -----------------------------------
+
+def list_unmatched(*, kind: str = "both") -> dict:
+    """Return unmatched claims and/or remittances.
+
+    An unmatched claim is one with ``matched_remittance_id IS NULL`` —
+    either auto-match never paired it, or it was unpaired by
+    ``manual_unmatch``. An unmatched remittance is one with
+    ``claim_id IS NULL`` — symmetric FK on the remittance side; we
+    update this in ``manual_match`` so the filter reflects the pair.
+
+    Note: T10's ``reconcile.run`` only writes the claim-side FK
+    (``Claim.matched_remittance_id``) when auto-pairing. Auto-matched
+    remittances therefore still show as "unmatched" here until the
+    pair is touched (re-ingest, manual unmatch + rematch). That gap
+    is intentional for T12 — fixing it requires modifying T10.
+
+    ``kind`` selects which side(s) to return:
+      - "claims": only claims
+      - "remittances": only remittances
+      - "both": both (default)
+
+    Returns ``{"claims": [...], "remittances": [...]}`` with the
+    unused side always an empty list (never absent) so callers can
+    unconditionally index.
+    """
+    if kind not in ("claims", "remittances", "both"):
+        raise ValueError(
+            f"list_unmatched: unknown kind={kind!r} "
+            "(expected 'claims', 'remittances', or 'both')"
+        )
+
+    result: dict = {"claims": [], "remittances": []}
+    with db.SessionLocal()() as s:
+        if kind in ("claims", "both"):
+            rows = (
+                s.query(Claim)
+                .filter(Claim.matched_remittance_id.is_(None))
+                .order_by(Claim.id.asc())
+                .all()
+            )
+            for r in rows:
+                parsed_at = (
+                    r.batch.parsed_at
+                    if r.batch is not None
+                    else r.service_date_from or utcnow()
+                )
+                result["claims"].append(
+                    to_ui_claim_from_orm(
+                        r, batch_id=r.batch_id, parsed_at=parsed_at,
+                        # list_unmatched filters matched_remittance_id IS NULL,
+                        # so every row has no remittance yet.
+                        received_total=0.0,
+                    )
+                )
+
+        if kind in ("remittances", "both"):
+            rows = (
+                s.query(Remittance)
+                .filter(Remittance.claim_id.is_(None))
+                .order_by(Remittance.id.asc())
+                .all()
+            )
+            for r in rows:
+                parsed_at = (
+                    r.batch.parsed_at
+                    if r.batch is not None
+                    else r.received_at
+                )
+                result["remittances"].append(
+                    to_ui_remittance_from_orm(
+                        r, batch_id=r.batch_id, parsed_at=parsed_at,
+                    )
+                )
+    return result
+
+def manual_match(claim_id: str, remit_id: str) -> dict:
+    """Pair a claim with a remittance manually (operator override).
+
+    Steps:
+      1. Load the claim; raise ``AlreadyMatchedError`` if it already
+         has a match (we never silently overwrite an existing pair).
+      2. Load the remittance; raise ``LookupError`` if missing.
+      3. Compute the new claim state via ``reconcile.apply_payment``
+         (or ``apply_reversal`` for status codes 21/22).
+      4. Insert a ``Match`` row with ``strategy="manual"``.
+      5. Update the claim (``state``, ``matched_remittance_id``) AND
+         the remittance (``claim_id``) so the symmetric FK reflects
+         the pair — required for ``list_unmatched`` to drop them.
+      6. Record an ``ActivityEvent(kind="manual_match", ...)``.
+      7. Commit; return ``{"claim": <ui>, "match": <ui>}``.
+
+    ``reconcile.apply_payment`` may return a noop (claim in terminal
+    state); we surface that as ``InvalidStateError`` rather than
+    silently pairing, because the operator clearly intended a state
+    change. The T15 API endpoint maps this to a 409 Conflict.
+    """
+    from cyclone import reconcile as _reconcile
+
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, claim_id)
+        if claim is None:
+            raise LookupError(f"claim {claim_id} not found")
+        if claim.matched_remittance_id is not None:
+            raise AlreadyMatchedError(
+                f"claim {claim_id} already matched to "
+                f"{claim.matched_remittance_id}"
+            )
+
+        remit = s.get(Remittance, remit_id)
+        if remit is None:
+            raise LookupError(f"remittance {remit_id} not found")
+
+        prior_state = claim.state
+        if remit.is_reversal:
+            intent = _reconcile.apply_reversal(claim, remit)
+        else:
+            intent = _reconcile.apply_payment(
+                claim, remit,
+                charge=claim.charge_amount,
+                paid=remit.total_paid,
+                status_code=remit.status_code,
+            )
+
+        if intent.skipped or intent.new_state is None:
+            current = (
+                claim.state.value
+                if hasattr(claim.state, "value")
+                else str(claim.state)
+            )
+            raise InvalidStateError(
+                current_state=current,
+                activity_kind=intent.activity_kind,
+            )
+
+        new_state = intent.new_state
+        now = utcnow()
+
+        s.add(Match(
+            claim_id=claim_id,
+            remittance_id=remit_id,
+            strategy="manual",
+            matched_at=now,
+            prior_claim_state=prior_state,
+            is_reversal=remit.is_reversal,
+        ))
+        claim.state = new_state
+        claim.matched_remittance_id = remit_id
+        # Symmetric FK update — see list_unmatched docstring for why
+        # we don't rely on T10 to do this.
+        remit.claim_id = claim_id
+
+        # SP7: line-level reconciliation + claim-level CAS aggregate.
+        # Skipped for reversals — they don't have SV1↔SVC line pairs.
+        if not remit.is_reversal:
+            _reconcile._reconcile_pair(s, claim, remit)
+
+        s.add(ActivityEvent(
+            ts=now,
+            kind="manual_match",
+            batch_id=remit.batch_id,
+            claim_id=claim_id,
+            remittance_id=remit_id,
+            payload_json={
+                "strategy": "manual",
+                "new_state": new_state.value,
+                "prior_state": prior_state.value,
+                "is_reversal": remit.is_reversal,
+            },
+        ))
+
+        s.commit()
+
+        parsed_at = (
+            claim.batch.parsed_at
+            if claim.batch is not None
+            else now
+        )
+        claim_dict = to_ui_claim_from_orm(
+            claim, batch_id=claim.batch_id, parsed_at=parsed_at,
+            received_total=float(remit.total_paid or 0),
+        )
+        matched_at_iso = now.isoformat().replace("+00:00", "Z")
+        return {
+            "claim": claim_dict,
+            "match": {
+                "strategy": "manual",
+                "claimId": claim_id,
+                "remittanceId": remit_id,
+                "matchedAt": matched_at_iso,
+                "isReversal": remit.is_reversal,
+                "priorState": prior_state.value,
+                "newState": new_state.value,
+            },
+        }
+
+def manual_unmatch(claim_id: str) -> dict:
+    """Unpair a previously matched claim and restore its prior state.
+
+    Reverses ``manual_match`` (and auto-match as a side-effect of
+    clearing the FK). Strategy:
+
+      1. Load the claim; raise ``NotMatchedError`` if it isn't
+         currently matched.
+      2. Delete every ``Match`` row for the claim (there may be more
+         than one — reversals create a 2nd row, see T10 spec).
+      3. Restore ``claim.state`` from the latest Match's
+         ``prior_claim_state``; fall back to ``SUBMITTED`` when
+         ``prior_claim_state`` is NULL (auto-match doesn't set it for
+         non-reversal payments — see reconcile.run line ~278).
+      4. Clear ``claim.matched_remittance_id`` and the symmetric
+         ``remit.claim_id`` so ``list_unmatched`` surfaces the pair
+         again.
+      5. Record ``ActivityEvent(kind="manual_unmatch", ...)`` and
+         commit.
+      6. Return ``{"claim": <ui>, "deletedMatches": <count>}``.
+    """
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, claim_id)
+        if claim is None:
+            raise LookupError(f"claim {claim_id} not found")
+        if claim.matched_remittance_id is None:
+            raise NotMatchedError(
+                f"claim {claim_id} has no active match"
+            )
+
+        matches = (
+            s.query(Match)
+            .filter(Match.claim_id == claim_id)
+            .order_by(Match.matched_at.desc())
+            .all()
+        )
+        if not matches:
+            # Defensive: matched_remittance_id was set but no Match
+            # rows exist. Shouldn't happen, but if it does, fall back
+            # to clearing the FK and starting fresh.
+            latest = None
+            paired_remit = None
+            restored_state = ClaimState.SUBMITTED
+        else:
+            latest = matches[0]
+            paired_remit = s.get(Remittance, latest.remittance_id)
+            restored_state = (
+                latest.prior_claim_state
+                if latest.prior_claim_state is not None
+                else ClaimState.SUBMITTED
+            )
+
+        deleted_count = len(matches)
+        for m in matches:
+            s.delete(m)
+
+        claim.state = restored_state
+        claim.matched_remittance_id = None
+        # Clear the symmetric FK on the remittance so list_unmatched
+        # surfaces the pair again. The remittance may have been
+        # deleted between the match and this call — guard with a
+        # None check so we don't blow up on a stale FK.
+        if paired_remit is not None:
+            paired_remit.claim_id = None
+
+        now = utcnow()
+        s.add(ActivityEvent(
+            ts=now,
+            kind="manual_unmatch",
+            claim_id=claim_id,
+            payload_json={
+                "restored_state": restored_state.value,
+                "deleted_matches": deleted_count,
+            },
+        ))
+
+        s.commit()
+
+        parsed_at = (
+            claim.batch.parsed_at
+            if claim.batch is not None
+            else now
+        )
+        # ``paired_remit`` is the matched remittance we cleared in
+        # the unmatch; use its ``total_paid`` for the response shape
+        # so the UI sees what was paid before the unpair. May be
+        # ``None`` if the remittance was deleted since the match —
+        # default to 0.0 in that case.
+        received_total = (
+            float(paired_remit.total_paid or 0)
+            if paired_remit is not None
+            else 0.0
+        )
+        claim_dict = to_ui_claim_from_orm(
+            claim, batch_id=claim.batch_id, parsed_at=parsed_at,
+            received_total=received_total,
+        )
+        return {
+            "claim": claim_dict,
+            "deletedMatches": deleted_count,
+        }
+
+# ------------------------------------------------------------------

backend/src/cyclone/store/kpis.py

@@ -0,0 +1,307 @@
+"""Dashboard KPI aggregation — cross-table reads consumed by /api/dashboard/kpis.
+
+``dashboard_kpis()`` returns a single dict that the React Dashboard page
+consumes as the source of truth for the 4 KPI tiles + the recent-activity
+panel. It reads from claims + remittances + batches in a single pass.
+
+``_claim_state_str`` is a small mapping helper that translates ORM
+status enum values to UI-friendly strings. It's private to this module.
+"""
+
+from datetime import datetime, timezone
+
+from cyclone import db
+from cyclone.db import Claim, Remittance
+
+
+# ---------------------------------------------------------------------------
+# SP27 Task 13: Dashboard aggregate KPIs.
+#
+# The Dashboard's "Billed / Received / Denial rate / Pending AR" tiles are
+# computed from the *whole* claim population, not a sample. With 60k+ claims
+# in production, fetching ``/api/claims?limit=100`` and reducing in JS would
+# silently produce wrong numbers (denial rate sampled, billed summed from
+# 100 rows). This module-level function does the aggregation server-side in
+# a single session and returns a small structured payload the Dashboard
+# can render directly.
+#
+# Performance: one ``SELECT * FROM claims`` (no pagination) + one
+# ``SELECT id, total_paid FROM remittances WHERE id IN (...)`` for matched
+# remits. SQLite returns 60k claim rows in ~30ms on the development
+# machine; the Python reduce is microseconds. If the dataset grows past
+# ~500k claims we'd want a SQL-side ``GROUP BY month`` instead — but at
+# that volume the dashboard should probably be backed by a materialized
+# view, not a live query.
+# ---------------------------------------------------------------------------
+
+
+def _claim_state_str(claim: Claim) -> str:
+    """Stringify a Claim's ``state`` regardless of enum vs raw str storage."""
+    st = claim.state
+    return st.value if hasattr(st, "value") else str(st)
+
+
+# Claim states counted toward the Dashboard's "pending" tile. SUBMITTED
+# is the initial post-parse state; REJECTED is the 999 envelope-level
+# rejection that means the payer never saw the claim. Both are
+# "outstanding adjudication" from the operator's POV; ``denied`` is
+# payer adjudication and lives in its own tile.
+_DASHBOARD_PENDING_STATES: frozenset[str] = frozenset({"submitted", "rejected"})
+
+
+def dashboard_kpis(
+    *,
+    months: int = 6,
+    top_n_providers: int = 4,
+    top_n_denials: int = 5,
+) -> dict:
+    """Compute Dashboard KPIs over the entire claim/remittance population.
+
+    Parameters
+    ----------
+    months
+        Number of trailing calendar months to include in the ``monthly``
+        sparkline series (default 6 — matches the existing frontend
+        ``MONTHS_BACK`` constant).
+    top_n_providers
+        How many providers to include in the ``topProviders`` array
+        (default 4 — matches the existing Dashboard layout).
+    top_n_denials
+        How many most-recent denied claims to include in the
+        ``topDenials`` array (default 5).
+
+    Returns
+    -------
+    dict with keys:
+
+    - ``totals``: aggregate counts + dollar sums + rates for the whole DB.
+    - ``monthly``: list of ``{month, label, count, billed, received,
+      denied, denialRate, ar}`` dicts, oldest-first, length = ``months``.
+      ``ar`` is the running outstanding accounts-receivable (billed -
+      received) carried forward across months, clamped at zero.
+    - ``topProviders``: list of ``{npi, label, claimCount, billed,
+      denied}`` dicts, sorted by claimCount desc.
+    - ``topDenials``: list of ``{id, patientName, billedAmount,
+      denialReason, submissionDate}`` dicts, sorted by submissionDate
+      desc, capped at ``top_n_denials``.
+
+    Notes
+    -----
+    - Empty DB returns zero-filled aggregates, an empty ``monthly`` array
+      (one entry per requested month), an empty ``topProviders`` array,
+      and an empty ``topDenials`` array.
+    - ``received`` for a claim is sourced from the matched Remittance's
+      ``total_paid`` column. Claims without a matched remittance
+      contribute 0 to the received sum (and thus inflate the
+      outstanding-AR figure, which is the correct operator-visible
+      semantic — "money we haven't been told was paid yet").
+    """
+    # Pre-build the trailing-N-months skeleton so the response always has
+    # exactly ``months`` entries even when the DB is empty or sparse.
+    now = datetime.now(timezone.utc)
+    skeleton: list[dict] = []
+    for i in range(months - 1, -1, -1):
+        d = now.replace(day=1)
+        # Walk backwards N months without ``relativedelta``.
+        for _ in range(i):
+            prev_month = d.month - 1
+            if prev_month == 0:
+                d = d.replace(year=d.year - 1, month=12)
+            else:
+                d = d.replace(month=prev_month)
+        skeleton.append({
+            "month": f"{d.year:04d}-{d.month:02d}",
+            "label": d.strftime("%b"),
+            "count": 0,
+            "billed": 0.0,
+            "received": 0.0,
+            "denied": 0,
+            "ar": 0.0,
+        })
+    skeleton_index = {entry["month"]: entry for entry in skeleton}
+
+    with db.SessionLocal()() as s:
+        # ``Claim.batch`` is a lazy ``relationship`` (default
+        # ``lazy="select"``); without ``selectinload`` each access to
+        # ``r.batch`` in the reduce loop below issues a fresh
+        # ``SELECT ... FROM batches WHERE id=?``. ``selectinload``
+        # pulls every distinct batch in one round-trip instead of N+1
+        # — critical for the 60k-claim dataset.
+        from sqlalchemy.orm import selectinload
+        claims: list[Claim] = (
+            s.query(Claim)
+            .options(selectinload(Claim.batch))
+            .all()
+        )
+
+        # Bulk-load matched-remit total_paid so a 60k-claim DB doesn't
+        # produce a 60k-query N+1.
+        matched_ids = [
+            r.matched_remittance_id
+            for r in claims
+            if r.matched_remittance_id
+        ]
+        received_by_remit: dict[str, float] = {}
+        if matched_ids:
+            for rid, total_paid in (
+                s.query(Remittance.id, Remittance.total_paid)
+                .filter(Remittance.id.in_(matched_ids))
+                .all()
+            ):
+                received_by_remit[rid] = float(total_paid or 0)
+
+        # Per-provider accumulator for the topProviders leaderboard.
+        provider_counts: dict[str, int] = {}
+        provider_billed: dict[str, float] = {}
+        provider_denied: dict[str, int] = {}
+
+        # Per-month accumulator + totals in a single pass.
+        total_count = 0
+        total_billed = 0.0
+        total_received = 0.0
+        denied_count = 0
+        pending_count = 0
+
+        # Collect denied candidates as we walk so we don't issue a
+        # second pass for the topDenials array.
+        denied_candidates: list[dict] = []
+
+        for r in claims:
+            billed = float(r.charge_amount or 0)
+            received = received_by_remit.get(r.matched_remittance_id, 0.0)
+            state_str = _claim_state_str(r)
+
+            total_count += 1
+            total_billed += billed
+            total_received += received
+            if state_str == "denied":
+                denied_count += 1
+                # Drop denied claims with no ``submissionDate`` — they'd
+                # sort first under reverse-lex (empty string < ISO) and
+                # render as "Invalid Date" in the Dashboard. A denial
+                # without a batch is exceptional and operator-irrelevant
+                # for the "recent denials" widget.
+                if r.batch is None or r.batch.parsed_at is None:
+                    pass
+                else:
+                    raw = r.raw_json or {}
+                    sub = raw.get("subscriber", {})
+                    denied_candidates.append({
+                        "id": r.id,
+                        "patientName": (
+                            f"{sub.get('first_name', '')} "
+                            f"{sub.get('last_name', '')}".strip()
+                        ),
+                        "billedAmount": billed,
+                        "denialReason": r.rejection_reason,
+                        "submissionDate": (
+                            r.batch.parsed_at
+                            .isoformat()
+                            .replace("+00:00", "Z")
+                        ),
+                    })
+            if state_str in _DASHBOARD_PENDING_STATES:
+                pending_count += 1
+
+            # Monthly bin. ``submissionDate`` lives on the parent Batch
+            # (all claims in a batch share a parsed_at). Use UTC year-month
+            # to match the skeleton.
+            if r.batch is not None and r.batch.parsed_at is not None:
+                pa = r.batch.parsed_at
+                key = f"{pa.year:04d}-{pa.month:02d}"
+                bucket = skeleton_index.get(key)
+                if bucket is not None:
+                    bucket["count"] += 1
+                    bucket["billed"] += billed
+                    bucket["received"] += received
+                    if state_str == "denied":
+                        bucket["denied"] += 1
+
+            # Provider bin (keyed on NPI).
+            npi = r.provider_npi or ""
+            if npi:
+                provider_counts[npi] = provider_counts.get(npi, 0) + 1
+                provider_billed[npi] = provider_billed.get(npi, 0.0) + billed
+                if state_str == "denied":
+                    provider_denied[npi] = provider_denied.get(npi, 0) + 1
+
+        # Compute denial rate per month + running AR.
+        running_ar = 0.0
+        for entry in skeleton:
+            if entry["count"] > 0:
+                entry["denialRate"] = (entry["denied"] / entry["count"]) * 100.0
+            else:
+                entry["denialRate"] = 0.0
+            running_ar = max(0.0, running_ar + entry["billed"] - entry["received"])
+            entry["ar"] = running_ar
+
+        # Resolve top provider labels from the Provider table in one
+        # round-trip. NPIs without a Provider row still appear in the
+        # leaderboard with an empty label so operators can see
+        # "unknown-NPI" claims are coming from somewhere.
+        # Local import keeps the module-level ``cyclone.providers.Provider``
+        # Pydantic DTO and the SQLAlchemy ``cyclone.db.Provider`` ORM
+        # separate (same pattern as ``list_providers`` / ``upsert_provider``).
+        provider_labels: dict[str, str] = {}
+        if provider_counts:
+            from cyclone.db import Provider as ProviderORM
+            for npi, label in (
+                s.query(ProviderORM.npi, ProviderORM.label).filter(
+                    ProviderORM.npi.in_(provider_counts.keys())
+                ).all()
+            ):
+                provider_labels[npi] = label or ""
+
+        top_providers = sorted(
+            provider_counts.items(),
+            key=lambda kv: kv[1],
+            reverse=True,
+        )[: max(0, top_n_providers)]
+        top_providers_out = [
+            {
+                "npi": npi,
+                "label": provider_labels.get(npi, ""),
+                "claimCount": count,
+                "billed": round(provider_billed.get(npi, 0.0), 2),
+                "denied": provider_denied.get(npi, 0),
+            }
+            for npi, count in top_providers
+        ]
+
+        # Top denials = most recently submitted denied claims, capped at
+        # ``top_n_denials``. submissionDate is ISO-8601 so lex sort ==
+        # chronological sort when timestamps share a tz.
+        denied_candidates.sort(key=lambda d: d["submissionDate"], reverse=True)
+        top_denials = denied_candidates[: max(0, top_n_denials)]
+
+    total_denial_rate = (
+        (denied_count / total_count) * 100.0 if total_count > 0 else 0.0
+    )
+    return {
+        "totals": {
+            "count": total_count,
+            "billed": round(total_billed, 2),
+            "received": round(total_received, 2),
+            "outstandingAr": round(max(0.0, total_billed - total_received), 2),
+            "denied": denied_count,
+            "denialRate": round(total_denial_rate, 4),
+            "pending": pending_count,
+        },
+        "monthly": [
+            {
+                "month": e["month"],
+                "label": e["label"],
+                "count": e["count"],
+                "billed": round(e["billed"], 2),
+                "received": round(e["received"], 2),
+                "denied": e["denied"],
+                "denialRate": round(e["denialRate"], 4),
+                "ar": round(e["ar"], 2),
+            }
+            for e in skeleton
+        ],
+        "topProviders": top_providers_out,
+        "topDenials": top_denials,
+    }
+
+

backend/src/cyclone/store/orm_builders.py

@@ -0,0 +1,172 @@
+"""ORM row builders — convert parser output into SQLAlchemy ORM rows.
+
+Each function returns an unsaved ORM object (or inserts related rows
+within an existing session). They never commit or close the session —
+the caller (``write.add_record``, ``acks.add_ack``, etc.) owns the
+transaction boundary.
+"""
+
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+
+from cyclone import db
+from cyclone.db import Claim, ClaimState, Remittance
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.models_835 import ClaimPayment
+
+from . import utcnow
+
+
+def _service_dates_from_claim(claim: ClaimOutput) -> tuple[date | None, date | None]:
+    """Extract (service_date_from, service_date_to) from a ClaimOutput.
+
+    The 837P model has ``service_lines[*].service_date`` (one per SV1).
+    We use the earliest as ``from`` and the latest as ``to``; if there
+    are no service lines, both are ``None``.
+    """
+    dates: list[date] = []
+    for sl in claim.service_lines:
+        if sl.service_date is not None:
+            dates.append(sl.service_date)
+    if not dates:
+        return None, None
+    return min(dates), max(dates)
+
+
+def _claim_837_row(claim: ClaimOutput, batch_id: str) -> Claim:
+    """Build a Claim ORM row from a ClaimOutput. NOT yet persisted."""
+    d_from, d_to = _service_dates_from_claim(claim)
+    return Claim(
+        id=claim.claim_id,
+        batch_id=batch_id,
+        # SP27 Task 17: Claim.patient_control_number must hold the CLM01
+        # claim_submittr's_identifier the 837 sent — that's the value the
+        # 835 echoes in CLP01, which the reconcile matcher joins on
+        # (reconcile.py:by_pcn), and what 999 / 277CA ACK lookups also
+        # use to cross-reference the original claim. Storing
+        # subscriber.member_id here (the 2010BA NM109) silently broke
+        # every auto-match in production.
+        patient_control_number=claim.claim_id or "",
+        service_date_from=d_from,
+        service_date_to=d_to,
+        charge_amount=Decimal(claim.claim.total_charge or 0),
+        provider_npi=claim.billing_provider.npi,
+        payer_id=claim.payer.id,
+        state=ClaimState.SUBMITTED,
+        raw_json=json.loads(claim.model_dump_json()),
+    )
+
+
+def _remittance_835_row(cp: ClaimPayment, batch_id: str) -> Remittance:
+    """Build a Remittance ORM row from a ClaimPayment. NOT yet persisted."""
+    received_at = utcnow()
+    # Adjustment amount: sum the CAS rows for the first service line.
+    # NOTE: This is a best-effort placeholder used until the reconciliation
+    # pass (T10) overwrites it from the persisted CasAdjustment rows. The
+    # authoritative value comes from `reconcile.run()`, which sums
+    # ``CasAdjustment.amount`` per ``remittance_id`` and writes the result
+    # back to ``Remittance.adjustment_amount``. We keep this stub so the
+    # row has a sane value if reconciliation is disabled or fails.
+    adjustment = Decimal("0")
+    if cp.service_payments:
+        sp = cp.service_payments[0]
+        for adj in sp.adjustments:
+            adjustment += adj.amount
+    # Use the first service line's service_date as the remit service_date.
+    service_date: date | None = None
+    if cp.service_payments and cp.service_payments[0].service_date is not None:
+        service_date = cp.service_payments[0].service_date
+    return Remittance(
+        id=cp.payer_claim_control_number,
+        batch_id=batch_id,
+        payer_claim_control_number=cp.payer_claim_control_number,
+        claim_id=None,
+        status_code=cp.status_code,
+        status_label=cp.status_label,
+        total_charge=Decimal(cp.total_charge or 0),
+        total_paid=Decimal(cp.total_paid or 0),
+        patient_responsibility=cp.patient_responsibility,
+        adjustment_amount=adjustment,
+        received_at=received_at,
+        service_date=service_date,
+        is_reversal=cp.status_code in ("21", "22"),
+        raw_json=json.loads(cp.model_dump_json()),
+    )
+
+
+def _persist_835_remit(session, cp: "ClaimPayment", remittance_id: str) -> None:
+    """SP7: persist ServiceLinePayment + CAS rows for one CLP composite.
+
+    For each 835 SVC composite in ``cp.service_payments``:
+      - insert a ServiceLinePayment row (line_number, procedure, modifiers,
+        charge, payment, units, service_date).
+      - flush to populate slp.id.
+      - insert each per-SVC CAS adjustment with ``service_line_payment_id``
+        set to slp.id.
+
+    For CLP-level CAS adjustments (``cp.claim_adjustments``, a future
+    extension; not produced by today's 835 parser but allowed by the spec):
+      - insert CAS rows with ``service_line_payment_id IS NULL``.
+
+    The caller controls the transaction; this function does not commit.
+    The 835 ingest site calls this after ``_remittance_835_row`` is
+    flushed so the FK target is populated.
+    """
+    import json as _json
+    from cyclone.db import ServiceLinePayment, CasAdjustment
+
+    for svc in cp.service_payments:
+        slp = ServiceLinePayment(
+            remittance_id=remittance_id,
+            line_number=svc.line_number,
+            procedure_qualifier=svc.procedure_qualifier,
+            procedure_code=svc.procedure_code,
+            modifiers_json=_json.dumps(svc.modifiers or []),
+            charge=Decimal(str(svc.charge)),
+            payment=Decimal(str(svc.payment)),
+            units=Decimal(str(svc.units)) if svc.units is not None else None,
+            unit_type=svc.unit_type,
+            service_date=svc.service_date,
+            ref_benefit_plan=svc.ref_benefit_plan,
+        )
+        session.add(slp)
+        session.flush()  # populate slp.id for the FK below
+
+        for adj in svc.adjustments:
+            quantity = getattr(adj, "quantity", None)
+            session.add(CasAdjustment(
+                remittance_id=remittance_id,
+                group_code=adj.group_code,
+                reason_code=adj.reason_code,
+                amount=Decimal(str(adj.amount)),
+                quantity=Decimal(str(quantity)) if quantity is not None else None,
+                service_line_payment_id=slp.id,
+            ))
+
+    # CLP-level CAS (no SVC composite to attach to). Today's parser does
+    # not produce these; the branch is forward-compatible.
+    for adj in getattr(cp, "claim_adjustments", []) or []:
+        quantity = getattr(adj, "quantity", None)
+        session.add(CasAdjustment(
+            remittance_id=remittance_id,
+            group_code=adj.group_code,
+            reason_code=adj.reason_code,
+            amount=Decimal(str(adj.amount)),
+            quantity=Decimal(str(quantity)) if quantity is not None else None,
+            service_line_payment_id=None,
+        ))
+
+
+def _claim_status_from_validation(claim: ClaimOutput) -> str:
+    """Re-implement the in-memory status rules (sub-project 1 §6.2)."""
+    v = claim.validation
+    if not v.passed:
+        has_r050 = any(e.rule == "R050_diagnosis_present" for e in v.errors)
+        return "draft" if has_r050 else "denied"
+    if claim.claim.frequency_code == "1":
+        return "submitted"
+    if v.warnings:
+        return "pending"
+    return "draft"

backend/src/cyclone/store/providers.py

@@ -0,0 +1,307 @@
+"""Provider, payer, and clearhouse configuration reads and upserts.
+
+The ``providers`` and ``payers`` modules in ``cyclone`` provide the
+ORM-row DTOs; this module is the read/upsert surface over them.
+``ensure_clearhouse_seeded`` is called at startup to insert the
+default Clearhouse row if missing.
+"""
+
+import json
+from datetime import datetime, timezone
+
+from cyclone import db
+from cyclone.providers import Clearhouse, Payer, Provider
+
+from . import utcnow
+
+
+# ------------------------------------------------------------------
+# SP9: providers / payers / payer_configs / clearhouse
+# ------------------------------------------------------------------
+
+def list_providers(*, is_active: bool | None = True) -> list[Provider]:
+    """List providers. ``is_active=None`` returns all."""
+    from cyclone.db import Provider as ProviderORM
+    from cyclone.providers import Provider
+    with db.SessionLocal()() as s:
+        q = s.query(ProviderORM)
+        if is_active is not None:
+            q = q.filter(ProviderORM.is_active == (1 if is_active else 0))
+        rows = q.order_by(ProviderORM.label).all()
+        return [Provider.model_validate(_provider_orm_to_dict(r)) for r in rows]
+
+def get_provider(npi: str) -> Provider | None:
+    from cyclone.db import Provider as ProviderORM
+    from cyclone.providers import Provider
+    with db.SessionLocal()() as s:
+        row = s.get(ProviderORM, npi)
+        return Provider.model_validate(_provider_orm_to_dict(row)) if row else None
+
+def upsert_provider(provider: Provider) -> Provider:
+    from cyclone.db import Provider as ProviderORM
+    with db.SessionLocal()() as s:
+        row = s.get(ProviderORM, provider.npi)
+        now = utcnow().isoformat()
+        if row is None:
+            row = ProviderORM(
+                npi=provider.npi, label=provider.label,
+                legal_name=provider.legal_name, tax_id=provider.tax_id,
+                taxonomy_code=provider.taxonomy_code,
+                address_line1=provider.address_line1,
+                address_line2=provider.address_line2,
+                city=provider.city, state=provider.state, zip=provider.zip,
+                is_active=1 if provider.is_active else 0,
+                created_at=provider.created_at.isoformat(),
+                updated_at=now,
+            )
+            s.add(row)
+        else:
+            row.label = provider.label
+            row.legal_name = provider.legal_name
+            row.tax_id = provider.tax_id
+            row.taxonomy_code = provider.taxonomy_code
+            row.address_line1 = provider.address_line1
+            row.address_line2 = provider.address_line2
+            row.city = provider.city
+            row.state = provider.state
+            row.zip = provider.zip
+            row.is_active = 1 if provider.is_active else 0
+            row.updated_at = now
+        s.commit()
+    return get_provider(provider.npi)  # type: ignore[return-value]
+
+def list_payers(*, is_active: bool | None = True) -> list[Payer]:
+    from cyclone.db import Payer as PayerORM
+    from cyclone.providers import Payer
+    with db.SessionLocal()() as s:
+        q = s.query(PayerORM)
+        if is_active is not None:
+            q = q.filter(PayerORM.is_active == (1 if is_active else 0))
+        rows = q.order_by(PayerORM.payer_id).all()
+        return [Payer.model_validate(_payer_orm_to_dict(r)) for r in rows]
+
+def get_payer_config(payer_id: str, transaction_type: str) -> dict | None:
+    from cyclone.db import PayerConfigORM
+    with db.SessionLocal()() as s:
+        row = s.get(PayerConfigORM, (payer_id, transaction_type))
+        return dict(row.config_json) if row else None
+
+def get_clearhouse() -> Clearhouse | None:
+    from cyclone.db import ClearhouseORM
+    from cyclone.providers import Clearhouse
+    with db.SessionLocal()() as s:
+        row = s.get(ClearhouseORM, 1)
+        if row is None:
+            return None
+        return Clearhouse.model_validate({
+            "id": 1,
+            "name": row.name,
+            "tpid": row.tpid,
+            "submitter_name": row.submitter_name,
+            "submitter_id_qual": row.submitter_id_qual,
+            "submitter_contact_name": row.submitter_contact_name,
+            "submitter_contact_email": row.submitter_contact_email,
+            "filename_block": dict(row.filename_block_json),
+            "sftp_block": dict(row.sftp_block_json),
+            "updated_at": row.updated_at,
+        })
+
+def update_clearhouse(block: Clearhouse) -> Clearhouse:
+    """Replace the singleton clearhouse row. SP25.
+
+    Used by ``PATCH /api/clearhouse`` to flip ``sftp_block.stub``
+    and adjust host/port/paths without touching SQLite directly.
+    Raises ``LookupError`` if the singleton row is missing — the
+    caller is expected to run the lifespan seed first.
+    """
+    from cyclone.db import ClearhouseORM
+    with db.SessionLocal()() as s:
+        row = s.get(ClearhouseORM, 1)
+        if row is None:
+            raise LookupError(
+                "clearhouse singleton row missing; run the lifespan "
+                "seed (ensure_clearhouse_seeded) before PATCH /api/clearhouse"
+            )
+        row.name = block.name
+        row.tpid = block.tpid
+        row.submitter_name = block.submitter_name
+        row.submitter_id_qual = block.submitter_id_qual
+        row.submitter_contact_name = block.submitter_contact_name
+        row.submitter_contact_email = block.submitter_contact_email
+        row.filename_block_json = json.loads(
+            json.dumps(block.filename_block.model_dump())
+        )
+        row.sftp_block_json = json.loads(
+            json.dumps(block.sftp_block.model_dump())
+        )
+        row.updated_at = datetime.now(timezone.utc).isoformat()
+        s.commit()
+        return Clearhouse.model_validate({
+            "id": 1,
+            "name": row.name,
+            "tpid": row.tpid,
+            "submitter_name": row.submitter_name,
+            "submitter_id_qual": row.submitter_id_qual,
+            "submitter_contact_name": row.submitter_contact_name,
+            "submitter_contact_email": row.submitter_contact_email,
+            "filename_block": dict(row.filename_block_json),
+            "sftp_block": dict(row.sftp_block_json),
+            "updated_at": row.updated_at,
+        })
+
+def ensure_clearhouse_seeded() -> None:
+    """Insert the default clearhouse singleton + 3 providers + CO_TXIX payer
+    if they don't exist. Idempotent. Called from the API lifespan."""
+    from cyclone.db import ClearhouseORM, Payer as PayerORM, PayerConfigORM, Provider as ProviderORM
+    from cyclone.providers import Clearhouse
+    with db.SessionLocal()() as s:
+        if s.get(ClearhouseORM, 1) is None:
+            ch = Clearhouse(
+                id=1,
+                name="dzinesco",
+                tpid="11525703",
+                submitter_name="Dzinesco",
+                submitter_id_qual="46",
+                submitter_contact_name="Tyler Martinez",
+                submitter_contact_email="tyler@dzinesco.com",
+                filename_block={
+                    "tz": "America/Denver",
+                    "outbound_template": "tp{tpid}-{tx}-{ts_mt}-1of1.{ext}",
+                    "inbound_template": "TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12",
+                },
+                sftp_block={
+                    "host": "mft.gainwelltechnologies.com",
+                    "port": 22,
+                    "username": "colorado-fts\\coxix_prod_11525703",
+                    "paths": {
+                        "outbound": "/CO XIX/PROD/coxix_prod_11525703/ToHPE",
+                        "inbound": "/CO XIX/PROD/coxix_prod_11525703/FromHPE",
+                    },
+                    "stub": True,
+                    "staging_dir": "./var/sftp/staging",
+                    "poll_seconds": 300,
+                    "auth": {"method": "keychain", "secret_ref": "sftp.gainwell.password"},
+                },
+                updated_at=utcnow(),
+            )
+            s.add(ClearhouseORM(
+                id=1,
+                name=ch.name,
+                tpid=ch.tpid,
+                submitter_name=ch.submitter_name,
+                submitter_id_qual=ch.submitter_id_qual,
+                submitter_contact_name=ch.submitter_contact_name,
+                submitter_contact_email=ch.submitter_contact_email,
+                filename_block_json=ch.filename_block.model_dump(),
+                sftp_block_json=ch.sftp_block.model_dump(),
+                updated_at=ch.updated_at.isoformat(),
+            ))
+
+        # Seed 3 providers (idempotent)
+        from cyclone.providers import Provider
+        now = utcnow().isoformat()
+        for npi, label in [
+            ("1881068062", "Montrose"),
+            ("1851446637", "Delta"),
+            ("1467507269", "Salida"),
+        ]:
+            if s.get(ProviderORM, npi) is None:
+                s.add(ProviderORM(
+                    npi=npi,
+                    label=label,
+                    legal_name="TOC, Inc.",
+                    tax_id="721587149",
+                    taxonomy_code="251E00000X",
+                    address_line1="1100 East Main St",
+                    address_line2="Suite A",
+                    city="Montrose",
+                    state="CO",
+                    zip="814014063",
+                    is_active=1,
+                    created_at=now,
+                    updated_at=now,
+                ))
+
+        # Seed CO_TXIX payer (idempotent)
+        if s.get(PayerORM, "CO_TXIX") is None:
+            s.add(PayerORM(
+                payer_id="CO_TXIX",
+                name="Colorado Medical Assistance Program",
+                receiver_name="COLORADO MEDICAL ASSISTANCE PROGRAM",
+                receiver_id="COMEDASSISTPROG",
+                is_active=1,
+                created_at=now,
+                updated_at=now,
+            ))
+            # 837P config block
+            s.add(PayerConfigORM(
+                payer_id="CO_TXIX",
+                transaction_type="837P",
+                config_json={
+                    "submitter_name": "Dzinesco",
+                    "submitter_contact_name": "Tyler Martinez",
+                    "submitter_contact_email": "tyler@dzinesco.com",
+                    "receiver_name": "COLORADO MEDICAL ASSISTANCE PROGRAM",
+                    "receiver_id_qualifier": "46",
+                    "receiver_id": "COMEDASSISTPROG",
+                    "bht06_allowed": ["CH", "RP"],
+                    "bht06_default": "CH",
+                    "sbr09_default": "MC",
+                    "sbr09_allowed": ["MC", "16", "MA", "MB", "ZZ"],
+                    "payer_id_qualifier": "PI",
+                    "payer_id": "CO_TXIX",
+                    "pwk_supported": False,
+                    "cas_2320_group_allowed": False,
+                    "claim_type_codes": {"11": "Office", "12": "Home", "99": "Other"},
+                },
+                updated_at=now,
+            ))
+            # 835 config block
+            s.add(PayerConfigORM(
+                payer_id="CO_TXIX",
+                transaction_type="835",
+                config_json={
+                    "expected_payer_tax_ids": [
+                        "81-1725341", "811725341", "84-0644739",
+                        "840644739", "1811725341",
+                    ],
+                    "expected_payer_health_plan_id": "7912900843",
+                    "payer_name_pattern": "^CO_(TXIX|BHA)$",
+                },
+                updated_at=now,
+            ))
+
+        s.commit()
+
+
+
+
+def _provider_orm_to_dict(row) -> dict:
+    return {
+        "npi": row.npi,
+        "label": row.label,
+        "legal_name": row.legal_name,
+        "tax_id": row.tax_id,
+        "taxonomy_code": row.taxonomy_code,
+        "address_line1": row.address_line1,
+        "address_line2": row.address_line2,
+        "city": row.city,
+        "state": row.state,
+        "zip": row.zip,
+        "is_active": bool(row.is_active),
+        "created_at": row.created_at,
+        "updated_at": row.updated_at,
+    }
+
+
+def _payer_orm_to_dict(row) -> dict:
+    return {
+        "payer_id": row.payer_id,
+        "name": row.name,
+        "receiver_name": row.receiver_name,
+        "receiver_id": row.receiver_id,
+        "is_active": bool(row.is_active),
+        "created_at": row.created_at,
+        "updated_at": row.updated_at,
+    }
+

backend/src/cyclone/store/records.py

@@ -0,0 +1,84 @@
+"""Pydantic models for parsed-batch records.
+
+``BatchRecord`` is the union type; ``BatchRecord837`` and ``BatchRecord835``
+narrow ``result`` to the parser-specific output types. Construction of
+``BatchRecord(kind="837p", ...)`` dispatches to ``BatchRecord837`` via
+``__new__`` so isinstance checks downstream narrow ``result`` correctly.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, model_validator
+
+from cyclone.parsers.models import ParseResult
+from cyclone.parsers.models_835 import ParseResult835
+
+BatchKind = Literal["837p", "835"]
+
+
+# ---------------------------------------------------------------------------
+# BatchRecord: value object preserved from sub-project 1.
+# ---------------------------------------------------------------------------
+
+
+class BatchRecord(BaseModel):
+    """One parsed file, with a stable uuid4 id and the full ParseResult.
+
+    ``result`` is a union: ``ParseResult`` for ``kind="837p"`` and
+    ``ParseResult835`` for ``kind="835"``. The concrete subclasses
+    ``BatchRecord837`` and ``BatchRecord835`` narrow those fields, so
+    callers that want type-checked access should use them and check
+    ``isinstance`` rather than pattern-matching on ``kind``.
+
+    Constructing ``BatchRecord(kind="837p", ...)`` dispatches to
+    ``BatchRecord837``; ``kind="835"`` dispatches to ``BatchRecord835``.
+    This lets the union-member narrowing work transparently.
+    """
+
+    model_config = ConfigDict(extra="ignore")
+
+    id: str
+    kind: BatchKind
+    input_filename: str
+    parsed_at: datetime  # tz-aware UTC
+    result: ParseResult | ParseResult835
+
+    def __new__(
+        cls, *args: Any, **kwargs: Any,
+    ) -> BatchRecord837 | BatchRecord835 | BatchRecord:
+        # Dispatch base-class construction to the right concrete subclass
+        # so isinstance checks downstream narrow `result` correctly.
+        if cls is BatchRecord:
+            kind = kwargs.get("kind")
+            if kind is None and args and isinstance(args[0], dict):
+                kind = args[0].get("kind")
+            if kind == "837p":
+                return BatchRecord837(*args, **kwargs)
+            if kind == "835":
+                return BatchRecord835(*args, **kwargs)
+        return super().__new__(cls)
+
+    @model_validator(mode="after")
+    def _check_parsed_at_tz(self) -> BatchRecord:
+        if self.parsed_at.tzinfo is None:
+            raise ValueError(
+                "parsed_at must be tz-aware (use datetime.now(timezone.utc))"
+            )
+        return self
+
+
+class BatchRecord837(BatchRecord):
+    """A parsed 837P (professional claim) batch."""
+
+    kind: Literal["837p"] = "837p"
+    result: ParseResult
+
+
+class BatchRecord835(BatchRecord):
+    """A parsed 835 (remittance advice) batch."""
+
+    kind: Literal["835"] = "835"
+    result: ParseResult835

backend/src/cyclone/store/ui.py

@@ -0,0 +1,532 @@
+"""UI serialization helpers — convert ORM rows to API-shaped dicts.
+
+These are the boundary between the persistence layer and the wire format
+consumed by the React frontend. Keep them stable: any rename or
+shape change ripples to every API consumer.
+
+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.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from decimal import Decimal
+
+from cyclone import db
+from cyclone.db import Claim, Remittance
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.models_835 import ClaimPayment
+from cyclone.parsers.payer import PayerConfig835
+
+from .orm_builders import _claim_status_from_validation
+
+
+def to_ui_claim(
+    claim: ClaimOutput,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+) -> dict:
+    """Map a 837P ClaimOutput to the UI's `Claim` shape (preserved)."""
+    parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
+    return {
+        "id": claim.claim_id,
+        "patientName": f"{claim.subscriber.first_name} {claim.subscriber.last_name}".strip(),
+        "providerNpi": claim.billing_provider.npi,
+        "payerName": claim.payer.name,
+        "cptCode": (
+            claim.service_lines[0].procedure.code
+            if claim.service_lines
+            else ""
+        ),
+        "billedAmount": float(claim.claim.total_charge or 0.0),
+        "receivedAmount": 0.0,
+        "status": _claim_status_from_validation(claim),
+        "denialReason": None,
+        "submissionDate": parsed_iso,
+        "batchId": batch_id,
+        "parsedAt": parsed_iso,
+    }
+
+
+def to_ui_remittance(
+    cp: ClaimPayment,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+    payer_config: PayerConfig835 | None = None,
+    payer_name: str = "",
+) -> dict:
+    """Map an 835 ClaimPayment to the UI's `Remittance` shape (preserved)."""
+    code = cp.status_code
+    if code in {"21", "22"}:
+        status = "reconciled"
+    else:
+        status = "received"
+
+    denial_reason: str | None = None
+    if code == "4" and cp.service_payments:
+        sp = cp.service_payments[0]
+        if sp.adjustments:
+            adj = sp.adjustments[0]
+            denial_reason = (
+                f"{adj.group_code}-{adj.reason_code}: ${float(adj.amount):.2f}"
+            )
+
+    cfg = payer_config if payer_config is not None else PayerConfig835.generic_835()
+    validation_warnings: list[str] = []
+    if code not in cfg.allowed_status_codes:
+        validation_warnings.append(
+            f"CLP02 code {code} not in payer allowlist"
+        )
+
+    # Aggregate adjustmentAmount across ALL service-line CAS rows, not just
+    # the first line. Mirrors the SUM the reconcile aggregator (T10)
+    # computes against persisted CasAdjustment rows; this inline version
+    # is the write-path equivalent (used when streaming 835 NDJSON
+    # responses before persistence finishes).
+    adjustment_total = Decimal("0")
+    for sp in cp.service_payments:
+        for adj in sp.adjustments:
+            adjustment_total += adj.amount
+
+    parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
+    return {
+        "id": cp.payer_claim_control_number,
+        "claimId": cp.original_claim_id or "",
+        "payerName": payer_name,
+        "paidAmount": float(cp.total_paid or 0.0),
+        "adjustmentAmount": float(adjustment_total),
+        "status": status,
+        "denialReason": denial_reason,
+        "validationWarnings": validation_warnings,
+        "receivedDate": parsed_iso,
+        "batchId": batch_id,
+        "parsedAt": parsed_iso,
+    }
+
+
+def to_ui_claim_from_orm(
+    row: Claim,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+    received_total: float = 0.0,
+) -> dict:
+    """Map an ORM ``Claim`` row to the UI's claim shape.
+
+    ``to_ui_claim`` takes a Pydantic ``ClaimOutput`` (used on the write path
+    during 837 ingest). For read paths — list_unmatched, manual_match return
+    values — we already have the ORM row and the serialized fields it
+    carries in ``raw_json``. Reading from ``raw_json`` keeps the UI shape
+    in sync with the original 837 parse without re-deserializing to a
+    Pydantic model.
+
+    Adds two fields ``to_ui_claim`` doesn't emit: ``state`` (the
+    reconciliation state machine value) and ``matchedRemittanceId`` (the
+    FK to the paired remittance, or None). Both are required by the UI.
+    """
+    raw = row.raw_json or {}
+    bp = raw.get("billing_provider", {})
+    payer_obj = raw.get("payer", {})
+    sub = raw.get("subscriber", {})
+    service_lines = raw.get("service_lines", [])
+    parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
+    cpt = (
+        service_lines[0].get("procedure", {}).get("code", "")
+        if service_lines
+        else ""
+    )
+    state_value = (
+        row.state.value if hasattr(row.state, "value") else str(row.state)
+    )
+    return {
+        "id": row.id,
+        "state": state_value,
+        "billedAmount": float(row.charge_amount or 0),
+        "patientName": (
+            f"{sub.get('first_name', '')} {sub.get('last_name', '')}".strip()
+        ),
+        "providerNpi": bp.get("npi") or row.provider_npi or "",
+        "payerName": payer_obj.get("name") or "",
+        "cptCode": cpt,
+        "submissionDate": parsed_iso,
+        "parsedAt": parsed_iso,
+        "status": state_value,
+        "matchedRemittanceId": row.matched_remittance_id,
+        "batchId": batch_id,
+        # Parity with ``to_ui_claim``'s shape — the UI tolerates extra keys
+        # but expects these on freshly-loaded rows from /api/claims too.
+        # ``received_total`` comes from the matched Remittance row when one
+        # exists; callers that don't pre-compute it (write path, unmatched
+        # list) get the default of 0.0 — which matches the unmapped state.
+        "receivedAmount": float(received_total),
+        "denialReason": None,
+    }
+
+
+# Max number of ActivityEvent rows surfaced in the detail drawer's
+# state history. The spec caps it at 50; a higher claim volume (manual
+# match/unmatch thrash) just shows the 50 most recent. Exposed as a
+# module constant so the endpoint layer can pass it through as a
+# default if it ever supports a `?limit=N` query param.
+CLAIM_DETAIL_HISTORY_LIMIT = 50
+
+
+def _iso_z(value: datetime | None) -> str:
+    """Format a tz-aware-or-naive UTC datetime as ISO-8601 with trailing Z.
+
+    The DB columns are declared ``DateTime(timezone=True)`` and rows are
+    stored UTC at write time, but SQLite drops the tzinfo on read
+    (returning a naive ``datetime``). Re-attach UTC for naive values
+    so the spec contract holds: every ISO datetime field ends in Z.
+    """
+    if value is None:
+        return ""
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.isoformat().replace("+00:00", "Z")
+
+
+def _address_to_ui(addr: dict | None) -> dict:
+    """Render a raw ``Address`` dict in the spec's parties address shape.
+
+    Returns an empty dict when the source is missing so the UI can
+    branch on the field's presence rather than the value. The spec
+    shape is ``{line1, line2|null, city, state, zip}``.
+    """
+    if not addr:
+        return {}
+    return {
+        "line1": addr.get("line1") or "",
+        "line2": addr.get("line2"),
+        "city": addr.get("city") or "",
+        "state": addr.get("state") or "",
+        "zip": addr.get("zip") or "",
+    }
+
+
+def _validation_issues_to_ui(issues: list[dict] | None) -> list[dict]:
+    """Project ValidationIssue dicts onto the spec's per-issue shape.
+
+    Source includes ``segment_index`` (a parser debug aid) which the
+    spec doesn't surface; we drop it. The endpoint contract is
+    ``{rule, severity, message}`` per issue.
+    """
+    if not issues:
+        return []
+    return [
+        {
+            "rule": issue.get("rule", ""),
+            "severity": issue.get("severity", "error"),
+            "message": issue.get("message", ""),
+        }
+        for issue in issues
+    ]
+
+
+def to_ui_claim_detail(
+    row: Claim,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+) -> dict:
+    """Map an ORM ``Claim`` row to the SP4 detail-drawer UI shape.
+
+    A superset of :func:`to_ui_claim_from_orm`: same top-level identity
+    fields, plus the full parties / validation / service-lines /
+    diagnoses / raw-segments / service-date / state-label payload that
+    the drawer needs. ``matchedRemittance`` and ``stateHistory`` are
+    *not* filled in here — they require extra queries and are stitched
+    in by :meth:`CycloneStore.get_claim_detail`.
+
+    The mapper is deliberately a pure function (no DB I/O) so the
+    endpoint layer can call it from a worker thread or swap the
+    history/remittance sources for tests without re-implementing the
+    body.
+    """
+    raw = row.raw_json or {}
+    bp = raw.get("billing_provider", {}) or {}
+    payer_obj = raw.get("payer", {}) or {}
+    sub = raw.get("subscriber", {}) or {}
+    service_lines = raw.get("service_lines", []) or []
+    diagnoses = raw.get("diagnoses", []) or []
+    validation = raw.get("validation", {}) or {}
+    raw_segments = raw.get("raw_segments", []) or []
+
+    parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
+    state_value = (
+        row.state.value if hasattr(row.state, "value") else str(row.state)
+    )
+
+    # Service dates come from the dedicated ORM columns (denormalized at
+    # ingest in _claim_837_row so the list views can sort/filter on
+    # them without a JSON parse). ``isoformat()`` on a ``date`` gives
+    # ``YYYY-MM-DD`` — the spec shape.
+    service_date_from_iso = (
+        row.service_date_from.isoformat() if row.service_date_from else None
+    )
+    service_date_to_iso = (
+        row.service_date_to.isoformat() if row.service_date_to else None
+    )
+
+    return {
+        # -- identity + state -----------------------------------------
+        "id": row.id,
+        "batchId": batch_id,
+        "state": state_value,
+        "stateLabel": state_value.capitalize(),
+        # -- money + dates --------------------------------------------
+        "billedAmount": float(row.charge_amount or 0),
+        "serviceDateFrom": service_date_from_iso,
+        "serviceDateTo": service_date_to_iso,
+        "submissionDate": parsed_iso,
+        "parsedAt": parsed_iso,
+        # -- patient / provider / payer -------------------------------
+        "patientName": (
+            f"{sub.get('first_name', '')} {sub.get('last_name', '')}".strip()
+        ),
+        "providerNpi": bp.get("npi") or row.provider_npi or "",
+        "providerName": bp.get("name") or "",
+        "payerName": payer_obj.get("name") or "",
+        "payerId": payer_obj.get("id") or row.payer_id or "",
+        # -- diagnoses ------------------------------------------------
+        "diagnoses": [
+            {
+                "code": d.get("code", ""),
+                "qualifier": d.get("qualifier"),
+            }
+            for d in diagnoses
+        ],
+        # -- service lines --------------------------------------------
+        # ``service_lines[i].procedure`` is a nested dict in the
+        # serialized raw_json; the spec flattens it into the line.
+        # ``charge`` and ``units`` are stored as Decimal via Pydantic
+        # and serialized to string — coerce defensively. ``modifiers``
+        # defaults to [] so the UI doesn't have to handle null.
+        "serviceLines": [
+            {
+                "lineNumber": sl.get("line_number"),
+                "procedureQualifier": (
+                    sl.get("procedure", {}).get("qualifier", "") or ""
+                ),
+                "procedureCode": (
+                    sl.get("procedure", {}).get("code", "") or ""
+                ),
+                "modifiers": list(
+                    sl.get("procedure", {}).get("modifiers") or []
+                ),
+                "charge": float(sl.get("charge") or 0),
+                "units": (
+                    float(sl["units"])
+                    if sl.get("units") is not None
+                    else None
+                ),
+                "unitType": sl.get("unit_type"),
+                "serviceDate": sl.get("service_date"),
+            }
+            for sl in service_lines
+        ],
+        # -- parties --------------------------------------------------
+        "parties": {
+            "billingProvider": {
+                "name": bp.get("name") or "",
+                "npi": bp.get("npi") or "",
+                "taxId": bp.get("tax_id") or "",
+                "address": _address_to_ui(bp.get("address")),
+            },
+            "subscriber": {
+                "firstName": sub.get("first_name") or "",
+                "lastName": sub.get("last_name") or "",
+                "memberId": sub.get("member_id") or "",
+                "dob": sub.get("dob"),
+                "gender": sub.get("gender"),
+            },
+            "payer": {
+                "name": payer_obj.get("name") or "",
+                "id": payer_obj.get("id") or "",
+            },
+        },
+        # -- validation ----------------------------------------------
+        "validation": {
+            "passed": bool(validation.get("passed", True)),
+            "errors": _validation_issues_to_ui(validation.get("errors")),
+            "warnings": _validation_issues_to_ui(validation.get("warnings")),
+        },
+        # -- raw segments (debug aid) --------------------------------
+        "rawSegments": raw_segments,
+        # -- matched remittance (filled by get_claim_detail) ---------
+        "matchedRemittance": None,
+        # -- state history (filled by get_claim_detail) --------------
+        "stateHistory": [],
+    }
+
+
+def to_ui_remittance_from_orm(
+    row: Remittance,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+) -> dict:
+    """Map an ORM ``Remittance`` row to the UI's remittance shape.
+
+    Same idea as ``to_ui_claim_from_orm``: read the PayerName from the
+    parent batch's ``raw_result_json`` (the ParseResult835 stashed at
+    insert time) since ``Remittance`` itself doesn't carry it.
+    """
+    parsed_iso = parsed_at.isoformat().replace("+00:00", "Z")
+    payer_name = ""
+    if row.batch is not None and row.batch.raw_result_json:
+        payer_obj = row.batch.raw_result_json.get("payer", {}) or {}
+        payer_name = payer_obj.get("name") or ""
+    status = (
+        "reconciled" if row.status_code in ("21", "22") else "received"
+    )
+    return {
+        "id": row.id,
+        "payerClaimControlNumber": row.payer_claim_control_number,
+        "claimId": row.claim_id or "",
+        "payerName": payer_name,
+        "paidAmount": float(row.total_paid or 0),
+        "adjustmentAmount": float(row.adjustment_amount or 0),
+        "status": status,
+        "denialReason": None,
+        "validationWarnings": [],
+        "receivedDate": row.received_at.isoformat().replace("+00:00", "Z"),
+        "batchId": batch_id,
+        "parsedAt": parsed_iso,
+    }
+
+
+def to_ui_remittance_with_adjustments(
+    row: Remittance,
+    *,
+    batch_id: str,
+    parsed_at: datetime,
+    cas_rows: list["db.CasAdjustment"] | None = None,
+) -> dict:
+    """Same shape as :func:`to_ui_remittance_from_orm` plus an ``adjustments`` array.
+
+    Each persisted ``CasAdjustment`` row is rendered as
+    ``{"group", "reason", "label", "amount", "quantity"}``. The ``label``
+    is resolved through :func:`cyclone.parsers.cas_codes.reason_label`
+    so the UI does not have to ship its own CARC dictionary.
+
+    ``cas_rows`` is optional so callers that don't have the rows handy
+    can still get the base dict; in that case ``adjustments`` is ``[]``.
+    Pass ``cas_rows`` to avoid an extra round-trip; the endpoint at
+    ``GET /api/remittances/{id}`` passes them in to keep this mapper a
+    pure function.
+    """
+    base = to_ui_remittance_from_orm(
+        row, batch_id=batch_id, parsed_at=parsed_at,
+    )
+    if not cas_rows:
+        base["adjustments"] = []
+        return base
+
+    # Lazy import to avoid the circular store ↔ parsers import that
+    # happens on cold start; mirrors the same pattern used elsewhere
+    # in this module.
+    from cyclone.parsers.cas_codes import reason_label
+
+    base["adjustments"] = [
+        {
+            "group": c.group_code,
+            "reason": c.reason_code,
+            "label": reason_label(c.group_code, c.reason_code),
+            "amount": float(c.amount or 0),
+            "quantity": (
+                float(c.quantity) if c.quantity is not None else None
+            ),
+        }
+        for c in cas_rows
+    ]
+    return base
+
+
+def _svc_to_wire_dict(svc) -> dict:
+    """Project an ORM ``ServiceLinePayment`` to the wire format used by
+    the remit drawer's ``serviceLinePayments`` array.
+
+    Mirrors the shape produced by the line-reconciliation endpoint so
+    the UI can render the same components from either source.
+    """
+    import json as _json
+    return {
+        "id": svc.id,
+        "line_number": svc.line_number,
+        "procedure_qualifier": svc.procedure_qualifier,
+        "procedure_code": svc.procedure_code,
+        "modifiers": _json.loads(svc.modifiers_json or "[]"),
+        "charge": str(Decimal(str(svc.charge))),
+        "payment": str(Decimal(str(svc.payment))),
+        "units": str(Decimal(str(svc.units))) if svc.units is not None else None,
+        "unit_type": svc.unit_type,
+        "service_date": svc.service_date.isoformat() if svc.service_date else None,
+    }
+
+
+def to_ui_provider(
+    *,
+    npi: str,
+    name: str,
+    tax_id: str | None = None,
+    address: str | None = None,
+    city: str | None = None,
+    state: str | None = None,
+    zip: str | None = None,
+    phone: str | None = None,
+    claim_count: int = 0,
+    outstanding_ar: float = 0.0,
+) -> dict:
+    return {
+        "npi": npi,
+        "name": name,
+        "taxId": tax_id or "",
+        "address": address or "",
+        "city": city or "",
+        "state": state or "",
+        "zip": zip or "",
+        "phone": phone or "",
+        "claimCount": claim_count,
+        "outstandingAr": float(outstanding_ar),
+    }
+
+
+def to_activity_event(
+    *,
+    id: str,
+    kind: str,
+    message: str,
+    timestamp: datetime,
+    npi: str | None = None,
+    amount: float | None = None,
+) -> dict:
+    return {
+        "id": id,
+        "kind": kind,
+        "message": message,
+        "timestamp": timestamp.isoformat().replace("+00:00", "Z"),
+        "npi": npi,
+        "amount": amount,
+    }
+
+
+def _date_in_bounds(
+    item: dict,
+    field: str,
+    date_from: str | None,
+    date_to: str | None,
+) -> bool:
+    """True if ``item[field]`` falls within ``[date_from, date_to]``."""
+    val = item.get(field)
+    if val is None:
+        return date_from is None and date_to is None
+    date_part = val[:10]
+    if date_from is not None and date_part < date_from:
+        return False
+    if date_to is not None and date_part > date_to:
+        return False
+    return True

backend/src/cyclone/store/write.py

@@ -0,0 +1,241 @@
+"""Write path for parsed batches — insert, reconcile, publish.
+
+The single entry point is ``add_record(record, *, event_bus=None)``,
+which replaces the body of the previous ``CycloneStore.add`` method.
+It owns its own SQLAlchemy session, runs idempotency checks, persists
+the batch + child rows, then (for 835 batches) triggers reconciliation
+and (if ``event_bus`` is provided) publishes live-tail events.
+
+Reconciliation runs OUTSIDE the persistence session, fail-soft: errors
+are logged but do not roll back the persisted batch.
+"""
+
+from __future__ import annotations
+
+import json
+
+from cyclone import db
+from cyclone.db import (
+    ActivityEvent,
+    Batch,
+    Claim,
+    Remittance,
+)
+from cyclone.parsers.models import ParseResult
+from cyclone.parsers.models_835 import ParseResult835
+from .orm_builders import _claim_837_row, _persist_835_remit, _remittance_835_row
+from .records import BatchRecord, BatchRecord837, BatchRecord835
+from .ui import to_ui_claim_from_orm, to_ui_remittance_from_orm
+
+
+def add_record(record: BatchRecord, *, event_bus=None) -> None:
+    """Persist a parsed batch (837P or 835) to the DB.
+
+    For 837P batches: inserts the Batch row, one Claim row per
+    claim, and a ``claim_submitted`` ActivityEvent per claim.
+
+    For 835 batches: inserts the Batch row, one Remittance row per
+    ClaimPayment, and a ``remit_received`` ActivityEvent per
+    ClaimPayment. Reconciliation (auto-match + per-pair CAS
+    aggregate) runs IN THE SAME SESSION before commit (SP27
+    Task 10) — a single ``s.commit()`` covers both ingest and
+    reconciliation. If reconcile raises, the whole 835 ingest
+    rolls back; the batch never appears half-reconciled with
+    placeholder ``adjustment_amount`` values.
+
+    Idempotency: ``Claim.id`` and ``Remittance.id`` are PRIMARY KEYS,
+    so a re-ingest of the same fixture (e.g. ``/api/parse-837`` called
+    twice with the same file) would otherwise raise
+    ``IntegrityError``. We do a per-row ``session.get(...)`` check
+    before each insert; if the row already exists, we log a warning
+    and skip. The batch row itself is still inserted (each parse
+    has a fresh ``uuid4`` id from the API). O(n) per row, but
+    acceptable for the small fixture sizes — production load is
+    one batch at a time via the API, not bulk inserts.
+
+    When ``event_bus`` is provided, publishes one ``claim_written``
+    or ``remittance_written`` event per newly-inserted row plus an
+    ``activity_recorded`` event per activity row, after commit. The
+    publish calls are best-effort — failures are logged but do not
+    roll back the persisted batch.
+    """
+    from cyclone.pubsub import EventBus
+
+    import logging
+    log = logging.getLogger(__name__)
+
+    # Track rows we actually inserted so we can publish events for them.
+    inserted_claim_ids: list[str] = []
+    inserted_remit_ids: list[str] = []
+
+    with db.SessionLocal()() as s:
+        batch_row = Batch(
+            id=record.id,
+            kind=record.kind,
+            input_filename=record.input_filename,
+            parsed_at=record.parsed_at,
+            totals_json=None,
+            validation_json=None,
+            raw_result_json=json.loads(record.result.model_dump_json()),
+        )
+        s.add(batch_row)
+
+        if isinstance(record, BatchRecord837):
+            result: ParseResult = record.result
+            for claim in result.claims:
+                if s.get(Claim, claim.claim_id) is not None:
+                    log.warning(
+                        "add: claim %s already exists; skipping (batch=%s)",
+                        claim.claim_id, record.id,
+                    )
+                    continue
+                s.add(_claim_837_row(claim, record.id))
+                s.add(ActivityEvent(
+                    ts=record.parsed_at,
+                    kind="claim_submitted",
+                    batch_id=record.id,
+                    claim_id=claim.claim_id,
+                    payload_json={
+                        "message": (
+                            f"Claim {claim.claim_id} submitted · "
+                            f"{claim.payer.name}"
+                        ),
+                        "npi": claim.billing_provider.npi,
+                        "amount": float(claim.claim.total_charge or 0.0),
+                    },
+                ))
+                inserted_claim_ids.append(claim.claim_id)
+        elif isinstance(record, BatchRecord835):
+            result835: ParseResult835 = record.result
+            payer_name = result835.payer.name
+            for cp in result835.claims:
+                if s.get(Remittance, cp.payer_claim_control_number) is not None:
+                    log.warning(
+                        "add: remittance %s already exists; skipping (batch=%s)",
+                        cp.payer_claim_control_number, record.id,
+                    )
+                    continue
+                remit_row = _remittance_835_row(cp, record.id)
+                s.add(remit_row)
+                # Flush so remit_row.id (FK target of cas_adjustments) is
+                # populated. SQLAlchemy assigns the PK on flush; without
+                # this the CasAdjustment inserts below would reference an
+                # unset id and violate the FK.
+                s.flush()
+                # SP7: persist per-line ServiceLinePayment + linked
+                # SVC-level CAS rows + claim-level CAS bucket. Replaces
+                # the previous per-SVC CAS insert loop so the
+                # service_line_payment_id FK is set correctly.
+                _persist_835_remit(s, cp, remit_row.id)
+                s.add(ActivityEvent(
+                    ts=record.parsed_at,
+                    kind="remit_received",
+                    batch_id=record.id,
+                    remittance_id=cp.payer_claim_control_number,
+                    payload_json={
+                        "message": (
+                            f"Remit {cp.payer_claim_control_number} received"
+                        ),
+                        "payerName": payer_name,
+                        "amount": float(cp.total_paid or 0.0),
+                    },
+                ))
+                inserted_remit_ids.append(cp.payer_claim_control_number)
+        else:
+            raise TypeError(
+                f"Unsupported BatchRecord subclass: {type(record).__name__}"
+            )
+
+        # SP27 Task 10: run reconcile INSIDE the same session, before
+        # commit. The placeholder ``adjustment_amount`` set by
+        # ``_remittance_835_row`` is overwritten by reconcile's
+        # ``_reconcile_pair`` SUM-of-CAS-aggregate pass before the
+        # rows are ever visible. If reconcile raises, the whole
+        # 835 ingest rolls back and the batch never lands.
+        if record.kind == "835":
+            from cyclone import reconcile as _reconcile
+            _reconcile.run(s, record.id)
+
+        s.commit()
+
+    # Publish live-tail events synchronously. EventBus.publish is async
+    # but its body is purely synchronous ``put_nowait`` enqueues; we
+    # bypass the async wrapper and call the internal enqueue directly
+    # so callers (sync FastAPI endpoints, sync test harnesses) don't
+    # need to await.
+    if event_bus is not None and (inserted_claim_ids or inserted_remit_ids):
+        publish_events_sync(
+            event_bus, record, inserted_claim_ids, inserted_remit_ids,
+        )
+
+
+def publish_events_sync(
+    event_bus,
+    record: BatchRecord,
+    claim_ids: list[str],
+    remit_ids: list[str],
+) -> None:
+    """Build UI-shaped payloads for newly-inserted rows and publish.
+
+    Runs after commit so subscribers can immediately re-fetch from
+    the API and see consistent data. Each ``claim_written`` /
+    ``remittance_written`` payload is identical to what the matching
+    list endpoint would return for that row.
+
+    This is sync because EventBus's enqueue path is sync; we don't
+    need a coroutine for ``put_nowait``.
+    """
+    from cyclone.pubsub import EventBus
+
+    import logging
+    log = logging.getLogger(__name__)
+
+    try:
+        with db.SessionLocal()() as s:
+            for cid in claim_ids:
+                row = s.get(Claim, cid)
+                if row is None:
+                    continue
+                ui = to_ui_claim_from_orm(
+                    row, batch_id=row.batch_id or record.id,
+                    parsed_at=record.parsed_at,
+                    # Fresh ingest — no remittance has been paired yet,
+                    # so ``Received`` is necessarily 0.
+                    received_total=0.0,
+                )
+                _sync_publish(event_bus, "claim_written", ui)
+            for rid in remit_ids:
+                row = s.get(Remittance, rid)
+                if row is None:
+                    continue
+                ui = to_ui_remittance_from_orm(
+                    row, batch_id=row.batch_id or record.id,
+                    parsed_at=record.parsed_at,
+                )
+                _sync_publish(event_bus, "remittance_written", ui)
+            # Activity events for this batch.
+            from sqlalchemy import select
+            activity_rows = s.execute(
+                select(ActivityEvent).where(ActivityEvent.batch_id == record.id)
+            ).scalars().all()
+            for arow in activity_rows:
+                ui = {
+                    "kind": arow.kind,
+                    "ts": arow.ts.isoformat().replace("+00:00", "Z"),
+                    "batchId": arow.batch_id,
+                    "claimId": arow.claim_id,
+                    "remittanceId": arow.remittance_id,
+                    "payload": arow.payload_json,
+                }
+                _sync_publish(event_bus, "activity_recorded", ui)
+    except Exception:
+        log.exception("add: event publish failed for batch %s", record.id)
+
+
+def _sync_publish(event_bus, kind: str, payload: dict) -> None:
+    """Synchronous fan-out helper. Mirrors ``EventBus.publish`` but
+    bypasses the async wrapper so callers don't need an event loop.
+    """
+    event = {**payload, "_kind": kind}
+    for queue in list(event_bus._subscribers.get(kind, ())):
+        event_bus._enqueue_or_drop_oldest(queue, event)

docs/superpowers/specs/2026-06-21-cyclone-store-split-design.md

@@ -1,13 +1,26 @@
 # CycloneStore split (Step 4) — Design Spec
 
-**Date:** 2026-06-21
-**Status:** Draft, pending user review
-**Branch:** `main` (split will land as a single atomic commit)
-**Scope:** Behaviour-preserving structural split of `backend/src/cyclone/store.py` (2,412 lines) into a `cyclone/store/` subpackage. Zero public API changes, zero test changes.
+**Date:** 2026-06-21 (original); 2026-06-29 (resumed)
+**Status:** Approved 2026-06-21; resumed 2026-06-29 after SP25-SP27 grew the file
+**Branch:** `sp21-store-split` (per SP-N convention; lands as a single atomic merge commit into `main`)
+**Scope:** Behaviour-preserving structural split of `backend/src/cyclone/store.py` (2,995 lines as of 2026-06-29) into a `cyclone/store/` subpackage. Zero public API changes, zero test changes.
+
+## Delta vs. 2026-06-21 baseline
+
+Resumed on 2026-06-29 because SP25-SP27 added 583 lines to `store.py` and three new top-level symbols (`dashboard_kpis`, `_claim_state_str`, `check_matched_pair_drift`) that the original 13-module layout didn't anticipate. Everything in **Decisions** below is preserved verbatim from the 2026-06-21 spec; only the implementation details (module count, baseline numbers, import surface) are updated.
+
+| | 2026-06-21 | 2026-06-29 |
+|---|---|---|
+| `store.py` size | 2,412 LOC | 2,995 LOC |
+| Target modules | 13 | **14** (new `kpis.py`) |
+| Test baseline | ~712 passed / ~31 failed / ~16 skipped | **1,176 passed / 1 failed (isolation flake) / 10 skipped** |
+| Test files using `_lock`/`_batches` idiom | 17 | **7** |
+| Private-helper leaks via `from cyclone.store import _` | 1 (`_claim_status_from_validation`) | **3** (`_persist_835_remit`, `_remittance_835_row` were imported by tests since SP7 but not audited at spec time) |
+| New top-level symbols to slot | — | `dashboard_kpis`, `_claim_state_str`, `check_matched_pair_drift` |
 
 ## Context
 
-`store.py` is the SQLAlchemy-backed facade over the parsed-X12 store. It sits on the hot path of `parse-999`, `parse-277ca`, reconciliation, and inbox match/unmatch. At 2,412 lines it has outgrown a single file: 41 methods on `CycloneStore`, ~17 module-level helpers, Pydantic models, ORM row builders, UI serializers, and exception types all live in one module.
+`store.py` is the SQLAlchemy-backed facade over the parsed-X12 store. It sits on the hot path of `parse-999`, `parse-277ca`, reconciliation, and inbox match/unmatch. At 2,995 lines it has outgrown a single file: 41 methods on `CycloneStore`, ~20 module-level helpers, Pydantic models, ORM row builders, UI serializers, dashboard aggregators, and exception types all live in one module.
 
 This is "Step 4" of the ongoing refactor series. Steps 1–3 (the api.py splits — commits `931782b`, `fc73075`, `3b5e2af`, `eb674f8`, `ff43f90`, `6ce6385`, `e63be87`, `a63ba5e`) established the pattern: turn a monolithic file into a subpackage with a thin facade, preserving every public import.
 
@@ -32,10 +45,12 @@ backend/src/cyclone/
     │                     _svc_to_wire_dict / _date_in_bounds / _provider_orm_to_dict /
     │                     _payer_orm_to_dict
     ├── write.py        ← add, _publish_events_sync, _sync_publish, _run_reconcile
-    │                     (biggest single module — ~210 lines)
+    │                     (biggest single module — ~280 lines after SP27 growth)
     ├── batches.py      ← get_batch, get, list, all, load_two_for_diff, _BatchesShim, _row_to_record
     ├── claim_detail.py ← get_remittance, get_claim_detail, iter_claims, iter_remittances,
-    │                     distinct_providers, recent_activity
+    │                     distinct_providers, recent_activity,
+    │                     check_matched_pair_drift   ← NEW from SP27 (cross-table invariant)
+    ├── kpis.py         ← dashboard_kpis, _claim_state_str   ← NEW MODULE from SP27
     ├── acks.py         ← add_ack, list_acks, get_ack (999),
     │                     add_ta1_ack, list_ta1_acks, get_ta1_ack,
     │                     add_277ca_ack, list_277ca_acks, get_277ca_ack
@@ -45,7 +60,11 @@ backend/src/cyclone/
                           get_payer_config, get_clearhouse, ensure_clearhouse_seeded
 ```
 
-13 modules total. Largest is `write.py` at ~210 lines; all others are ≤150 lines. Facade `__init__.py` ≈ 80 lines (re-exports + class + singleton).
+14 modules total. Largest is `write.py` at ~280 lines; second-largest is `ui.py` at ~250 lines; all others are ≤180 lines. Facade `__init__.py` ≈ 100 lines (re-exports + class + singleton + the 3 new private-helper re-exports).
+
+**Why a separate `kpis.py`:** `dashboard_kpis` and `_claim_state_str` together are a coherent unit — read-only aggregations across claims + remittances + batches that the UI consumes as a single dashboard payload. They don't fit any existing module's responsibility (`ui.py` is row-formatters, `claim_detail.py` is single-claim reads). A 14th module is one file with two functions; co-locating them in `ui.py` would push `ui.py` past 300 LOC and mix row-shaping with aggregate aggregation.
+
+**Why `check_matched_pair_drift` in `claim_detail.py`:** It's a read-only invariant check that joins `claims` + `remittances` — the same pair-of-tables `get_claim_detail`'s matched-remittance summary reads from. No other module touches both tables for a single read; creating a 15th file for one function would be over-decomposition.
 
 ## CycloneStore class shape
 
@@ -76,7 +95,7 @@ Each module function inlines `with db.SessionLocal()() as s:` at the top — mat
 
 ### Lock + shim stay on the class
 
-`CycloneStore.__init__` keeps `self._lock = threading.RLock()` and `self._batches = _BatchesShim()` because 17 test files use `with store._lock: store._batches.clear()` as the cleanup idiom. The `_BatchesShim` class itself moves to `batches.py` alongside `_row_to_record`.
+`CycloneStore.__init__` keeps `self._lock = threading.RLock()` and `self._batches = _BatchesShim()` because 7 test files still use `with store._lock: store._batches.clear()` as the cleanup idiom (down from 17 at spec time — others moved to `db._reset_for_tests()` over SP23-SP27). The `_BatchesShim` class itself moves to `batches.py` alongside `_row_to_record`.
 
 ### Naming collision table
 
@@ -98,21 +117,34 @@ Each module function inlines `with db.SessionLocal()() as s:` at the top — mat
 
 | Caller | Import | After split |
 |---|---|---|
-| `cyclone/api.py` | `from cyclone.store import (CycloneStore, store, BatchRecord, AlreadyMatchedError, ...)` | unchanged |
-| `cyclone/api.py` (deferred) | `from cyclone.store import NotMatchedError` | unchanged |
-| `cyclone/batch_diff.py` | `from cyclone.store import BatchRecord, BatchRecord835, BatchRecord837, _claim_status_from_validation` | facade re-exports `_claim_status_from_validation` |
-| `cyclone/backup_service.py` (deferred) | `from cyclone.store import store as cycl_store` | unchanged |
+| `cyclone/api.py` (top of file, line 87) | `from cyclone.store import (AlreadyMatchedError, BatchRecord, InvalidStateError, dashboard_kpis, store, utcnow)` | unchanged |
+| `cyclone/api.py` (line 155, in startup hook) | `from cyclone.store import check_matched_pair_drift` | unchanged |
+| `cyclone/api.py` (line 2182, deferred) | `from cyclone.store import NotMatchedError` | unchanged |
+| `cyclone/batch_diff.py` (line 32) | `from cyclone.store import BatchRecord, BatchRecord835, BatchRecord837, _claim_status_from_validation` | facade re-exports `_claim_status_from_validation` |
+| `cyclone/backup_service.py` (line 231, deferred) | `from cyclone.store import store as cycl_store` | unchanged |
 | `cyclone/api_routers/acks.py` | `from cyclone.store import store` | unchanged |
 | `cyclone/api_routers/ta1_acks.py` | `from cyclone.store import store` | unchanged |
 | `cyclone/scheduler.py` | `from cyclone.store import store as cycl_store` (also `BatchRecord` deferred) | unchanged |
+| `cyclone/handlers/handle_999.py` | `from cyclone.store import store as cycl_store` | unchanged (NEW from SP27) |
+| `cyclone/handlers/handle_277ca.py` | `from cyclone.store import store as cycl_store` | unchanged (NEW from SP27) |
+| `cyclone/handlers/handle_ta1.py` | `from cyclone.store import store as cycl_store` | unchanged (NEW from SP27) |
+| `cyclone/handlers/handle_835.py` | `from cyclone.store import BatchRecord, store as cycl_store` | unchanged (NEW from SP27) |
 | `cyclone/parsers/validator.py` (3 deferred) | `from cyclone import store as store_mod` | unchanged |
 | `cyclone/api_helpers.py` | docstring references `cyclone.store.utcnow` | unchanged (facade re-exports `utcnow`) |
 
-**Open decision:** `_claim_status_from_validation` is currently a module-level helper imported by `batch_diff.py` (line 32). Recommended: re-export it from `__init__.py` to preserve the import. This is a small facade pollution but matches the precedent of the api.py split (which re-exports routers).
+**Resolved open item:** `_claim_status_from_validation` is re-exported from `__init__.py` (preserves `batch_diff.py:32`). Two further private-helper leaks surfaced during the SP25-SP27 audit and are re-exported the same way:
+
+| Private helper | Imported by | Lives in |
+|---|---|---|
+| `_claim_status_from_validation` | `batch_diff.py:32` | `orm_builders.py` |
+| `_persist_835_remit` | `test_service_line_payments.py`, `test_api_line_reconciliation.py`, `test_reconcile_line_level.py`, `test_inbox_endpoints_sp7.py` | `orm_builders.py` |
+| `_remittance_835_row` | same 4 test files | `orm_builders.py` |
+
+All three are leading-underscore "internal but reachable" names; re-exporting them from the facade is the same precedent the api.py split set when it re-exported routers. The `_lock` + `_batches.clear()` idiom check is still 7 files (not the original 17); the test files that no longer use it moved to `db._reset_for_tests()` over SP23-SP27.
 
 ### Tests — zero changes
 
-17 test files use `with store._lock: store._batches.clear()` — all continue to work because `CycloneStore.__init__` still sets those attributes and `_BatchesShim` continues to function identically. The 3 dedicated store tests (`test_store.py`, `test_store_claim_detail.py`, `test_store_reconcile.py`) import from `cyclone.store` (still works) or `cyclone` (still works).
+7 test files still use `with store._lock: store._batches.clear()` — all continue to work because `CycloneStore.__init__` still sets those attributes and `_BatchesShim` continues to function identically. The 3 dedicated store tests (`test_store.py`, `test_store_claim_detail.py`, `test_store_reconcile.py`) import from `cyclone.store` (still works) or `cyclone` (still works). The 4 test files that import `_persist_835_remit` / `_remittance_835_row` continue to work because the facade re-exports them.
 
 ## Migration plan
 
@@ -121,10 +153,10 @@ Each module function inlines `with db.SessionLocal()() as s:` at the top — mat
 ```bash
 rm backend/src/cyclone/store.py
 mkdir -p backend/src/cyclone/store
-# write all 13 new files
+# write all 14 new files
 cd backend && python -m pytest tests/ --tb=line -q  # must be green
 git add -A
-git commit -m "refactor(store): split store.py into cyclone/store/ subpackage"
+git commit -m "refactor(sp21): split store.py into cyclone/store/ subpackage (14 modules)"
 ```
 
 **Follow-up commit (only if needed):** Docs update — add "CycloneStore subpackage" section to README following the precedent of commits `81aebf5`, `2718114`, `ea64e6e`, `804e557`.
@@ -145,7 +177,7 @@ The split is structural only. Every test that passes today must pass after.
 cd backend && python -m pytest tests/ --tb=line -q 2>&1 | tee /tmp/cyclone-baseline.txt | tail -5
 ```
 
-Expected baseline: ~712 passed, ~31 failed (pre-existing), ~16 skipped. The 31 failures are documented cross-test pollution (11 `test_serialize_837`, 3 `test_secrets`, 2 `test_sftp_paramiko`, etc., verified in commit `931782b`). Post-split count must match exactly.
+Expected baseline (captured 2026-06-29 on `main`): **1,176 passed, 1 failed, 10 skipped**. The single failure is `tests/test_provider_extended_response.py::test_provider_detail_includes_orphan_remit_received`, which is a known test-isolation flake (passes when run solo, fails after other tests in the suite due to shared state) — verified pre-SP21 and not caused by this refactor. Post-split count must match exactly: 1,176 / 1 / 10.
 
 **Tier 2 — Hot-path focused (14 files).** Exercises every method being split.
 
@@ -190,7 +222,7 @@ Smoke output is documented in the commit message.
 | R4 | `add_record()` body diverges from current `add()` | Medium | High | Mechanical copy of lines 898-1107, strip `self.` prefix, change 3 private-method calls to module-function calls. No logic refactoring. |
 | R5 | Singleton duplicated | Low | High | Declared exactly once, in `__init__.py`. |
 | R6 | TYPE_CHECKING imports break runtime | Low | Low | `EventBus` already used in current `add()` signature; same pattern. |
-| R7 | Test imports of private helpers break | Low | Medium | `grep -rn "from cyclone.store import _" backend/tests/` before impl; re-export anything found. |
+| R7 | Test imports of private helpers break | Low → **Materialized** | Medium | Audited 2026-06-29: 3 leaks found (`_claim_status_from_validation`, `_persist_835_remit`, `_remittance_835_row`). All three are re-exported from the facade. Re-audit before merge. |
 | R8 | Package discovery misses new subpackage | Low | Medium | Verify with `pip install -e .` before commit; if needed, add explicit `packages = [...]`. |
 | R9 | Generator methods leak sessions | Low | Medium | Preserve `with` inside generator body. |
 | R10 | Docstring drift | Low | Low | Preserve existing docstrings verbatim on re-exports. |
@@ -237,7 +269,8 @@ python -c "
 from cyclone.store import (
     CycloneStore, store, BatchRecord, BatchRecord837, BatchRecord835, BatchKind,
     AlreadyMatchedError, NotMatchedError, InvalidStateError, utcnow,
-    _claim_status_from_validation,
+    dashboard_kpis, check_matched_pair_drift,
+    _claim_status_from_validation, _persist_835_remit, _remittance_835_row,
 )
 print('all imports OK')
 "
@@ -254,11 +287,17 @@ print('all imports OK')
 - Adding new tests
 - Modifying DB models in `cyclone.db`
 - Updating SQLCipher key rotation flow
-- Touching any of the 17 test files that use `_lock` / `_batches.clear()`
-- Touching any importer in `cyclone.api`, `cyclone.api_routers.*`, `cyclone.batch_diff`, `cyclone.backup_service`, `cyclone.scheduler`, `cyclone.parsers.validator`, `cyclone.api_helpers`
+- Touching any of the 7 test files that use `_lock` / `_batches.clear()`
+- Touching any of the 4 test files that import `_persist_835_remit` / `_remittance_835_row`
+- Touching any importer in `cyclone.api`, `cyclone.api_routers.*`, `cyclone.batch_diff`, `cyclone.backup_service`, `cyclone.scheduler`, `cyclone.handlers.*`, `cyclone.parsers.validator`, `cyclone.api_helpers`
 
 If any of these surface during implementation, they get deferred to follow-up tasks — never silently included.
 
 ## Open items
 
-1. **`_claim_status_from_validation` re-export** — recommended re-export from `__init__.py`. User to confirm during spec review.
+All open items from the 2026-06-21 draft are resolved:
+
+1. **`_claim_status_from_validation` re-export** — **resolved**: re-exported from `__init__.py` (confirmed by brainstorming 2026-06-29).
+2. **`_persist_835_remit` / `_remittance_835_row` re-export** — **resolved** (NEW during resume): both re-exported from `__init__.py` to preserve the 4 test files that import them.
+3. **`_claim_state_str` placement** — **resolved** (NEW during resume): co-located with `dashboard_kpis` in the new `kpis.py` module.
+4. **`check_matched_pair_drift` placement** — **resolved** (NEW during resume): in `claim_detail.py` (cross-table read with the same pair-of-tables territory as `get_claim_detail`'s matched-remit summary).