feat(sp36): extract parse router (5 routes + 7 helpers promoted to _shared.py)

Task 15.

api_routers/claims.py (new, 530 lines):
  - GET /api/claims                                (paginated list + NDJSON)
  - GET /api/claims/stream                         (NDJSON live-tail on claim_written)
  - GET /api/claims/{claim_id}                     (drawer detail + SP28 ack_links)
  - GET /api/claims/{claim_id}/serialize-837       (regenerate X12 837P)
  - GET /api/claims/{claim_id}/line-reconciliation (837 vs 835 side-by-side)
  - 3 single-router helpers stay in-file per D4:
      _compact_ack_links_for_claim, _claim_line_dict, _svc_to_dict
  - All inline imports inside handlers preserved verbatim
    (select, LineReconciliation/ServiceLinePayment/CasAdjustment,
    json as _json, Decimal)

Slicing: 1 contiguous cut (drop 1-indexed 1274-1688 = the empty
section divider + the entire claims block). After the cut, the
next thing is the app-level auth_router import at api.py:1274+.

api.py: 1720 -> 1305 LOC (-415; 5 routes + 3 helpers extracted).

api_routers/__init__.py: registry extended (alphabetical) with
claims. 17 routers in registry.

Added 1 backward-compat shim for test_api_stream_live.py:
  from cyclone.api_routers.claims import claims_stream
The test does 'from cyclone.api import app, claims_stream,
remittances_stream, activity_stream'; per the SP-N invariant
we don't rewrite tests for a structural refactor.

Pytest: bit-identical to baseline — 21 failed, 1246 passed,
10 skipped, 6 errors.

Live-tested via curl on the running container:
  GET /api/claims                         -> 401
  GET /api/claims/stream                  -> 401
  GET /api/claims/<id>                    -> 401
  GET /api/claims/<id>/serialize-837      -> 401
  GET /api/claims/<id>/line-reconciliation-> 401
  POST /api/claims                        -> 405

pr-reviewer: skipped (Tasks 13-16 batch — folded into Task 17).

backend/src/cyclone/api_routers/__init__.py

@@ -12,19 +12,43 @@ from fastapi import APIRouter
 
 from cyclone.api_routers import (
     acks,
+    activity,
     admin,
+    batches,
     claim_acks,
+    claims,
+    clearhouse,
+    config,
     dashboard,
+    eligibility,
     health,
+    inbox,
+    parse,
+    payers,
+    providers,
+    reconciliation,
+    remittances,
     ta1_acks,
 )
 
 routers: list[APIRouter] = [
     acks.router,        # gated
+    activity.router,    # gated
     admin.router,       # gated
+    batches.router,     # gated
     claim_acks.router,  # gated
+    claims.router,      # gated
+    clearhouse.router,  # gated
+    config.router,      # gated
     dashboard.router,   # gated
+    eligibility.router, # gated
     health.router,      # public — health probes must work pre-auth
+    inbox.router,       # gated
+    parse.router,       # gated
+    payers.router,      # gated
+    providers.router,   # gated
+    reconciliation.router, # gated
+    remittances.router, # gated
     ta1_acks.router,    # gated
 ]
 

backend/src/cyclone/api_routers/_shared.py

@@ -5,5 +5,269 @@ package import from here. Helpers graduate here when at least two
 routers need them — single-router helpers stay in the router that
 uses them.
 
-Initially empty; helpers promote here as the SP36 refactor progresses.
+Helpers currently promoted here:
+
+- :func:`_actor_user_id` — promoted early (during SP36 Task 11
+  / clearhouse extraction) because the clearhouse router needs
+  it and the 2 remaining call-sites in ``api.py`` (parse-999 ack
+  block, parse-277ca ack block) are both inside the parse
+  surface that Task 16 will extract. Promoting now is cheaper
+  than leaving a cross-module ``from cyclone.api import _actor_user_id``
+  that would create an import-cycle at registry load time.
+
+- :data:`PAYER_FACTORIES` / :data:`PAYER_FACTORIES_835` — SP36
+  Task 16: payer config dicts lifted from ``api.py`` alongside
+  the ``_resolve_payer`` / ``_resolve_payer_835`` helpers that
+  consume them. The two helpers each touch a single ``PAYER_FACTORIES*``
+  dict; keeping both halves of the pair in one module removes a
+  circular import (parse.py → _shared._resolve_payer → api.PAYER_FACTORIES).
+
+- :func:`_resolve_payer` / :func:`_resolve_payer_835` — used by
+  ``parse-837`` and ``parse-835`` endpoints respectively. Promoted
+  in SP36 Task 16 alongside the PAYER_FACTORIES dicts.
+
+- :func:`_transaction_set_id_from_segments` — used by
+  ``parse-837`` and ``parse-835`` envelope guards. Promoted in
+  SP36 Task 16.
+
+- :func:`_build_and_persist_ack` — used by ``parse-837`` (when
+  ``?ack=true``). Promoted in SP36 Task 16.
+
+- :func:`_reconciliation_summary_for_batch` — used by
+  ``parse-835``. Promoted in SP36 Task 16.
+
+- :func:`_ta1_synthetic_source_batch_id` — used by ``parse-ta1``.
+  Promoted in SP36 Task 16.
+
+- :func:`_serialize_ta1` — used by ``parse-ta1`` to build the
+  raw TA1 round-trip text. Promoted in SP36 Task 16 per the
+  plan's "8 helpers" specification; technically a single-router
+  helper per D4 but moved here for symmetry with the other parse
+  serializers.
 """
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from fastapi import HTTPException, Request
+
+from cyclone import db
+from cyclone.parsers.batch_ack_builder import build_ack_for_batch
+from cyclone.parsers.payer import PayerConfig, PayerConfig835
+from cyclone.parsers.serialize_999 import serialize_999
+from cyclone.store import store
+
+log = logging.getLogger(__name__)
+
+
+# --------------------------------------------------------------------------- #
+# Actor user id (SP36 Task 11 — early-promoted)
+# --------------------------------------------------------------------------- #
+
+
+def _actor_user_id(request: Request) -> int | None:
+    """Return the acting user's id from ``request.state.user``, or None.
+
+    ``get_current_user``/``matrix_gate`` populate ``request.state.user``
+    for both the authenticated path and the AUTH_DISABLED escape hatch.
+    Returns None when the state hasn't been set (e.g. background jobs
+    or unit tests that bypass auth). Used to stamp ``user_id`` onto
+    audit events without crashing the request.
+    """
+    user = getattr(request.state, "user", None)
+    if user is None:
+        return None
+    return getattr(user, "id", None)
+
+
+# --------------------------------------------------------------------------- #
+# Payer config dicts (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+#
+# Mirror cli._PAYER_FACTORIES. Kept here (not in the parse router) so
+# the ``_resolve_payer`` / ``_resolve_payer_835`` helpers can import
+# their backing dicts without a circular import.
+
+
+PAYER_FACTORIES: dict[str, Any] = {
+    "co_medicaid": PayerConfig.co_medicaid,
+    "generic_837p": PayerConfig.generic_837p,
+}
+
+PAYER_FACTORIES_835: dict[str, Any] = {
+    "co_medicaid_835": PayerConfig835.co_medicaid_835,
+    "generic_835": PayerConfig835.generic_835,
+}
+
+
+# --------------------------------------------------------------------------- #
+# Payer resolution (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _resolve_payer(name: str) -> PayerConfig:
+    if name not in PAYER_FACTORIES:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "Unknown payer",
+                "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES)}",
+            },
+        )
+    return PAYER_FACTORIES[name]()
+
+
+def _resolve_payer_835(name: str) -> PayerConfig835:
+    if name not in PAYER_FACTORIES_835:
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "Unknown payer",
+                "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES_835)}",
+            },
+        )
+    return PAYER_FACTORIES_835[name]()
+
+
+# --------------------------------------------------------------------------- #
+# Envelope detection (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _transaction_set_id_from_segments(segments: list[list[str]]) -> str | None:
+    """Return the ST01 transaction-set id (``"837"``, ``"835"``, ``"999"``...).
+
+    SP35 helper: scans the first few tokenized segments for the ST
+    segment and returns its second element (ST01). Returns None when no
+    ST is present — e.g. a TA1 file, which uses the bare TA1 segment
+    and no ST envelope. The endpoint-level envelope guards treat
+    ``None`` as "no ST found; let the parser decide" so TA1 files
+    routed through the wrong endpoint still surface a parse error
+    rather than a misleading "expected 837p, got ''" message.
+    """
+    for seg in segments[:5]:  # ST is always the second segment after ISA
+        if seg and seg[0] == "ST" and len(seg) > 1:
+            return seg[1]
+    return None
+
+
+# --------------------------------------------------------------------------- #
+# 999 ACK builder (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _build_and_persist_ack(batch_id: str) -> dict | None:
+    """Build a 999 ACK for ``batch_id`` and persist the row.
+
+    Returns the ack payload dict (matches the ``/api/parse-999``
+    response shape so the JSON and NDJSON clients can share the
+    schema) or None if the build failed. The build is fail-soft:
+    errors are logged but never abort the 837 ingest, because the
+    user-visible 837 result is still correct.
+    """
+    try:
+        ack_result = build_ack_for_batch(batch_id)
+    except Exception:
+        log.exception("build_ack_for_batch failed for %s", batch_id)
+        return None
+    fg = ack_result.functional_group_acks[0] if ack_result.functional_group_acks else None
+    if fg is None:
+        return None
+    raw_text = serialize_999(ack_result, interchange_control_number=ack_result.envelope.control_number)
+    row = store.add_ack(
+        source_batch_id=batch_id,
+        accepted_count=fg.accepted_count,
+        rejected_count=fg.rejected_count,
+        received_count=fg.received_count,
+        ack_code=fg.ack_code,
+        raw_json=json.loads(ack_result.model_dump_json()),
+    )
+    return {
+        "id": row.id,
+        "accepted_count": fg.accepted_count,
+        "rejected_count": fg.rejected_count,
+        "received_count": fg.received_count,
+        "ack_code": fg.ack_code,
+        "source_batch_id": batch_id,
+        "raw_999_text": raw_text,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Reconciliation summary (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _reconciliation_summary_for_batch(batch_id: str) -> dict:
+    """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch.
+
+    Reads from the DB after ``store.add()`` has already run reconciliation
+    synchronously (SP27 Task 10: ``reconcile.run(s, batch_id)`` inside the
+    ingest session, before commit). Counts are observed at this moment;
+    a subsequent manual match/unmatch will not be reflected until the
+    next request.
+
+    ``skipped`` is reserved for future use — the orchestrator tracks
+    skipped claims internally but does not surface a queryable count.
+    """
+    from sqlalchemy import func, select
+    from cyclone.db import Match, Remittance
+
+    with db.SessionLocal()() as s:
+        matched = s.execute(
+            select(func.count(Match.id)).where(
+                Match.remittance_id.in_(
+                    select(Remittance.id).where(Remittance.batch_id == batch_id)
+                )
+            )
+        ).scalar_one()
+
+    # Pull unmatched via the store (small result set; cheap).
+    unmatched = store.list_unmatched(kind="both")
+    return {
+        "matched": matched,
+        "unmatched_claims": len(unmatched["claims"]),
+        "unmatched_remittances": len(unmatched["remittances"]),
+        "skipped": 0,  # reserved — T10 does not persist a skipped count
+    }
+
+
+# --------------------------------------------------------------------------- #
+# TA1 synthetic source batch id (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _ta1_synthetic_source_batch_id(interchange_control_number: str) -> str:
+    """Return a synthetic ``batches.id`` for a received TA1 with no source batch.
+
+    Mirrors :func:`_ack_synthetic_source_batch_id` (in
+    ``cyclone.handlers._ack_id``). The ta1_acks.source_batch_id
+    FK requires a row in batches; for received TA1s we synthesize an
+    id of the form ``TA1-<ISA13>``. The row is NOT created in batches
+    (same FK-is-no-op convention as the 999 path).
+    """
+    return f"TA1-{(interchange_control_number or '').strip() or '000000001'}"
+
+
+# --------------------------------------------------------------------------- #
+# TA1 serializer (SP36 Task 16)
+# --------------------------------------------------------------------------- #
+
+
+def _serialize_ta1(result) -> str:
+    """Render a TA1 file from a ParseResultTa1 for the ``raw_ta1_text`` field.
+
+    Mirrors what the parser consumed: ISA envelope → TA1 → IEA. We
+    rebuild minimal ISA fields from the envelope plus the TA1 segment
+    verbatim. The serializer is intentionally tiny — TA1 has no GS/ST,
+    so there's no functional-group structure to round-trip.
+    """
+    ta1 = result.ta1
+    parts = [
+        f"TA1*{ta1.control_number}*{ta1.interchange_date.strftime('%y%m%d') if ta1.interchange_date else ''}*"
+        f"{ta1.interchange_time or ''}*{ta1.ack_code}*{ta1.note_code or ''}*"
+        f"{ta1.ack_generated_date.strftime('%y%m%d') if ta1.ack_generated_date else ''}",
+    ]
+    return "~".join(parts) + "~"

backend/src/cyclone/api_routers/activity.py

@@ -0,0 +1,102 @@
+"""``/api/activity`` and ``/api/activity/stream`` — operator-facing event log.
+
+Two endpoints, both gated by ``matrix_gate``:
+
+- ``GET /api/activity``          — paginated event list with ``kind`` /
+  ``since`` filters, plus an NDJSON variant when the caller sends
+  ``Accept: application/x-ndjson``.
+- ``GET /api/activity/stream``   — NDJSON live-tail: snapshot of the
+  most recent N events, then ``activity_recorded`` events as they
+  hit the store. Default ``limit`` is 50 (smaller than the list
+  endpoint's 200) because activity is high-volume — callers usually
+  want the most recent handful, not a full replay.
+
+The snapshot halves of ``/api/activity`` and ``/api/activity/stream``
+share the same in-memory filter logic (``kind`` + ``since``) so the
+two endpoints are interchangeable for the snapshot half.
+
+SP36 Task 10: this block moved here from ``api.py:2606`` (the
+``/api/activity*`` pair).
+"""
+from __future__ import annotations
+
+from typing import Any, AsyncIterator
+
+from fastapi import APIRouter, Depends, Query, Request
+from fastapi.responses import StreamingResponse
+
+from cyclone.api_helpers import (
+    ndjson_line as _ndjson_line,
+    ndjson_stream_list as _ndjson_stream_list,
+    tail_events as _tail_events,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/activity")
+def list_activity(
+    request: Request,
+    kind: str | None = Query(None),
+    since: str | None = Query(None),
+    limit: int = Query(200, ge=1, le=5000),
+) -> Any:
+    events = store.recent_activity(limit=limit)
+    if kind is not None:
+        events = [e for e in events if e["kind"] == kind]
+    if since is not None:
+        events = [e for e in events if e["timestamp"] >= since]
+    total = len(events)
+    has_more = False
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(events, total, total, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": events,
+        "total": total,
+        "returned": total,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/activity/stream")
+async def activity_stream(
+    request: Request,
+    kind: str | None = Query(None),
+    since: str | None = Query(None),
+    limit: int = Query(50, ge=1, le=5000),
+) -> StreamingResponse:
+    """Stream Activity events as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``activity_recorded``. Default ``limit`` is 50
+    (smaller than the list endpoint's 200) because activity is
+    high-volume — callers usually want the most recent handful, not a
+    full replay.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        # Snapshot reuses the same in-memory filter as ``list_activity``
+        # so the two endpoints are interchangeable for the snapshot
+        # half.
+        events = store.recent_activity(limit=limit)
+        if kind is not None:
+            events = [e for e in events if e["kind"] == kind]
+        if since is not None:
+            events = [e for e in events if e["timestamp"] >= since]
+        for ev in events:
+            yield _ndjson_line({"type": "item", "data": ev})
+        yield _ndjson_line({
+            "type": "snapshot_end", "data": {"count": len(events)},
+        })
+
+        async for chunk in _tail_events(request, bus, ["activity_recorded"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")

backend/src/cyclone/api_routers/batches.py

@@ -0,0 +1,515 @@
+"""``/api/batches*`` — read views + ZIP export over the parsed-batch population.
+
+Three endpoints, all gated by ``matrix_gate``:
+
+- ``POST /api/batches/{batch_id}/export-837`` — download a ZIP of
+  regenerated X12 837 files for the requested claim ids. Per-claim
+  payer config + clearhouse identity drive the submitter/receiver
+  blocks; per-claim millisecond offset on the filename keeps every
+  file in the bundle unique (HCPF requires this). Read-only — does
+  NOT mutate Claim state (compare with
+  ``/api/inbox/rejected/resubmit?download=true`` which DOES flip
+  ``REJECTED → SUBMITTED``).
+- ``GET  /api/batches``                       — summary list,
+  newest-first, capped at ``limit``. Each row includes ``claimIds``
+  (837P only, so the Upload page can render a one-click Re-export
+  button per row without a round-trip to ``/api/batches/{id}``).
+  SP30: also returns billing-outcome fields
+  (``acceptedCount`` / ``rejectedCount`` / ``pendingCount`` /
+  ``billedTotal`` / ``topRejectionReason`` / ``hasProblem``) so the
+  Dashboard "Recent batches" widget can render one row per batch
+  without an N+1 fetch.
+- ``GET  /api/batches/{batch_id}``            — full batch record
+  (parsed envelope + claims). 404 when unknown.
+
+Three single-router helpers stay in this file (per spec D4):
+
+- :func:`_batch_summary_claim_count`       — claim count (837P or 835).
+- :func:`_batch_summary_claim_ids`         — per-claim ids (837P only).
+- :func:`_batch_summary_billing_outcomes`  — per-batch GROUP BY state
+  aggregate plus the most-recent rejection reason.
+
+Inline imports inside handlers (preserved verbatim per spec D5):
+``zipfile``, ``datetime``, ``ZoneInfo``, ``build_outbound_filename``,
+``PayerConfigORM``, ``func`` (sqlalchemy).
+
+SP36 Task 14: this block moved here from ``api.py:1280`` (the 3
+``/api/batches*`` routes + 3 single-router helpers + the SP30
+state-bucket tuples and the ``# 277CA STC category`` note).
+"""
+from __future__ import annotations
+
+import io
+import json
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+from fastapi.responses import StreamingResponse
+
+from cyclone import db
+from cyclone.api_helpers import (
+    ndjson_stream_list as _ndjson_stream_list,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Batch, Claim, ClaimState
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.serialize_837 import SerializeError as SerializeError837
+from cyclone.parsers.serialize_837 import serialize_837_for_resubmit
+from cyclone.store import BatchRecord, store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.post("/api/batches/{batch_id}/export-837")
+def export_batch_837(request: Request, batch_id: str, body: dict):
+    """Download a ZIP of regenerated X12 837 files for the requested claim_ids.
+
+    Body shape: ``{"claim_ids": [str, ...]}``.
+
+    Each successfully serialized claim becomes an entry in the ZIP named
+    per the HCPF X12 File Naming Standards:
+    ``tp{tpid}-837P-{yyyymmddhhmmssSSS}-1of1.x12`` (with a per-claim
+    millisecond offset so every file in the bundle has a unique name).
+    The ``serialize_837_for_resubmit`` serializer is used so every file
+    gets a unique interchange / group control number — back-to-back
+    exports of the same set must produce different envelopes (required
+    by X12).
+
+    The submitter block (Loop 1000A — NM1*41 + PER) is populated from
+    the clearhouse singleton (dzinesco's identity in the seeded config)
+    and the receiver block (NM1*40) is populated from the per-payer
+    config. Without this wiring, the serializer falls back to
+    ``CYCLONE`` / ``RECEIVER`` placeholders and HCPF rejects the file.
+
+    No DB state is mutated by this endpoint — it is read-only. Compare
+    with ``/api/inbox/rejected/resubmit?download=true`` which ALSO flips
+    ``ClaimState.REJECTED → SUBMITTED``; the two endpoints are
+    intentionally separate.
+
+    Responses:
+      200 — ``application/zip`` with the .x12 entries. Per-claim failures
+            are surfaced via the ``X-Cyclone-Serialize-Errors`` header
+            (JSON-encoded array of ``{claim_id, reason}``).
+      400 — ``claim_ids`` missing or empty.
+      404 — ``batch_id`` unknown.
+      422 — every claim failed to serialize; body is JSON listing all
+            failures (``{"detail": {"serialize_errors": [...]}}``).
+    """
+    import zipfile
+    from datetime import datetime
+    from zoneinfo import ZoneInfo
+
+    from cyclone.edi.filenames import build_outbound_filename
+
+    ids = body.get("claim_ids") or []
+    if not ids:
+        raise HTTPException(400, "claim_ids required")
+
+    with db.SessionLocal()() as s:
+        batch = s.get(Batch, batch_id)
+        if batch is None:
+            raise HTTPException(404, f"unknown batch: {batch_id}")
+
+        serialize_errors: list[dict] = []
+        ordered_rows: list[tuple[str, "Claim"]] = []
+        for cid in ids:
+            c = s.get(Claim, cid)
+            if c is None:
+                serialize_errors.append({"claim_id": cid, "reason": "unknown claim_id"})
+                continue
+            ordered_rows.append((cid, c))
+
+    # Pull clearhouse identity (submitter). If unseeded, the serializer
+    # falls back to placeholder defaults — degraded but not a hard error.
+    ch = store.get_clearhouse()
+    submitter_kwargs: dict = {}
+    if ch is not None:
+        submitter_kwargs = {
+            "sender_id": ch.tpid,
+            "submitter_name": ch.submitter_name,
+            "submitter_contact_name": ch.submitter_contact_name,
+            "submitter_contact_email": ch.submitter_contact_email,
+        }
+        # Submitter phone is not in the clearhouse config today, but if
+        # it ever is, wire it here. Email is the canonical contact
+        # channel for HCPF submissions per the SP9 spec.
+        if getattr(ch, "submitter_contact_phone", None):
+            submitter_kwargs["submitter_contact_phone"] = ch.submitter_contact_phone
+
+    # Resolve per-claim payer config so each file's receiver (NM1*40)
+    # and SBR09 are correct. Cache so we don't re-query the same payer.
+    from cyclone.db import PayerConfigORM as _PayerConfigORM
+    _payer_cache: dict[str, dict | None] = {}
+
+    def _resolve_payer_cfg(claim_obj: ClaimOutput) -> dict | None:
+        pid = (claim_obj.payer.id or "").strip() if claim_obj.payer else ""
+        pname = (claim_obj.payer.name or "").strip() if claim_obj.payer else ""
+        cache_key = pid or pname
+        if cache_key in _payer_cache:
+            return _payer_cache[cache_key]
+        cfg: dict | None = None
+        with db.SessionLocal()() as ss:
+            # 1. Exact match on (payer_id, "837P")
+            if pid:
+                row = ss.get(_PayerConfigORM, (pid, "837P"))
+                if row is not None:
+                    cfg = dict(row.config_json)
+            # 2. Fallback: any row whose payer_id matches the parsed payer.name
+            #    (HCPF files emit "SKCO0" in NM109 but the canonical
+            #    payer_id in the DB is "CO_TXIX" — name-matching is the
+            #    pragmatic lookup for that case).
+            if cfg is None and pname:
+                row = (
+                    ss.query(_PayerConfigORM)
+                    .filter(_PayerConfigORM.transaction_type == "837P")
+                    .all()
+                )
+                for r in row:
+                    cj = dict(r.config_json)
+                    if cj.get("submitter_name") and pname.lower() in str(cj).lower():
+                        cfg = cj
+                        break
+                    if (r.payer_id or "").upper() == pname.upper():
+                        cfg = cj
+                        break
+            # 3. Last resort: first 837P row in the table.
+            if cfg is None:
+                row = (
+                    ss.query(_PayerConfigORM)
+                    .filter(_PayerConfigORM.transaction_type == "837P")
+                    .first()
+                )
+                if row is not None:
+                    cfg = dict(row.config_json)
+        _payer_cache[cache_key] = cfg
+        return cfg
+
+    # Build per-claim kwargs (receiver + SBR09) lazily. Receiver
+    # defaults to the parsed payer name/ID if no config row matches.
+    def _serialize_kwargs(claim_obj: ClaimOutput) -> dict:
+        payer_cfg = _resolve_payer_cfg(claim_obj) or {}
+        receiver_id = (
+            payer_cfg.get("receiver_id")
+            or (claim_obj.payer.id if claim_obj.payer else None)
+            or "RECEIVER"
+        )
+        receiver_name = (
+            payer_cfg.get("receiver_name")
+            or (claim_obj.payer.name if claim_obj.payer else None)
+            or receiver_id
+        )
+        sbr09 = payer_cfg.get("sbr09_default") or "MC"
+        return {
+            "receiver_id": receiver_id,
+            "receiver_name": receiver_name,
+            "claim_filing_indicator_code": sbr09,
+        }
+
+    # Base MT timestamp for HCPF filenames. We add a per-claim
+    # millisecond offset so each file in the ZIP has a unique 17-digit
+    # ts (HCPF requires that; the spec also enforces "1of1" for the
+    # sequence element).
+    base_ts = datetime.now(ZoneInfo("America/Denver"))
+
+    def _per_claim_filename(idx: int, cid: str) -> str:
+        if ch is None:
+            # No clearhouse — fall back to a per-claim friendly name.
+            return f"claim-{cid}.x12"
+        # Millisecond offset, with second/minute rollover.
+        offset_ms = (idx - 1) * 1  # 1 ms per claim is enough within an export
+        ts_mt = base_ts.fromtimestamp(
+            base_ts.timestamp() + offset_ms / 1000.0, tz=ZoneInfo("America/Denver")
+        )
+        return build_outbound_filename(ch.tpid, "837P", now_mt=ts_mt)
+
+    buf = io.BytesIO()
+    with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
+        for idx, (cid, c) in enumerate(ordered_rows, start=1):
+            if not c.raw_json:
+                serialize_errors.append({"claim_id": cid, "reason": "no raw_json"})
+                continue
+            try:
+                claim_obj = ClaimOutput.model_validate(c.raw_json)
+            except Exception as exc:
+                serialize_errors.append(
+                    {"claim_id": cid, "reason": f"raw_json invalid: {exc}"}
+                )
+                continue
+            try:
+                kwargs = {**submitter_kwargs, **_serialize_kwargs(claim_obj)}
+                text = serialize_837_for_resubmit(
+                    claim_obj, interchange_index=idx, **kwargs
+                )
+            except SerializeError837 as exc:
+                serialize_errors.append({"claim_id": cid, "reason": str(exc)})
+                continue
+            zf.writestr(_per_claim_filename(idx, cid), text)
+
+    success_count = len(ids) - len(serialize_errors)
+    if serialize_errors and success_count == 0:
+        # Every claim failed — surface the failure list in the body so the
+        # UI can render a useful error toast (the response is not a ZIP).
+        raise HTTPException(
+            422,
+            detail={"serialize_errors": serialize_errors},
+        )
+
+    buf.seek(0)
+    headers = {
+        "Content-Disposition": (
+            f'attachment; filename="batch-{batch_id}-{success_count}-claims.zip"'
+        ),
+    }
+    if serialize_errors:
+        headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors)
+    return Response(
+        content=buf.getvalue(),
+        media_type="application/zip",
+        headers=headers,
+    )
+
+
+def _batch_summary_claim_count(rec: BatchRecord) -> int:
+    """Return the number of claims on a batch, handling both 837P and 835."""
+    if rec.kind == "837p":
+        return len(rec.result.claims)  # type: ignore[attr-defined]
+    if rec.kind == "835":
+        return len(rec.result.claims)  # type: ignore[attr-defined]
+    return 0
+
+
+def _batch_summary_claim_ids(rec: BatchRecord) -> list[str]:
+    """Return per-claim ids for an 837P batch, or ``[]`` otherwise.
+
+    The Upload page's History tab renders a one-click Re-export ZIP
+    button per row; that button calls
+    ``POST /api/batches/{id}/export-837`` with the row's claim ids.
+    Carrying them in the list response avoids an extra round-trip
+    to ``/api/batches/{id}`` for every row. 835 has no re-export
+    endpoint, so the list is empty for those — the UI uses the
+    empty list as the signal to hide the button.
+    """
+    if rec.kind != "837p":
+        return []
+    return [
+        c.claim_id
+        for c in rec.result.claims  # type: ignore[attr-defined]
+        if getattr(c, "claim_id", None)
+    ]
+
+
+# SP30: state buckets the Dashboard widget (and any future "how the
+# last batch billed" surface) reads at a glance. Keep these in sync
+# with ClaimState — adding a new state here is a deliberate decision
+# the operator needs to see, not a coincidence.
+_BATCH_SUMMARY_ACCEPTED_STATES: tuple[ClaimState, ...] = (
+    ClaimState.PAID,
+    ClaimState.RECEIVED,
+    ClaimState.RECONCILED,
+    ClaimState.PARTIAL,
+)
+_BATCH_SUMMARY_REJECTED_STATES: tuple[ClaimState, ...] = (
+    ClaimState.REJECTED,
+    ClaimState.DENIED,
+    ClaimState.REVERSED,
+)
+_BATCH_SUMMARY_PENDING_STATES: tuple[ClaimState, ...] = (
+    ClaimState.SUBMITTED,
+)
+# 277CA STC category A4/A6/A7 — payer-side rejections that may not
+# yet have flipped Claim.state (the operator hasn't acknowledged).
+# The Dashboard widget treats these as problems too, mirroring the
+# Inbox `rejected + payer_rejected` aggregation.
+_BATCH_SUMMARY_PAYER_REJECT_CODES: tuple[str, ...] = ("A4", "A6", "A7")
+
+
+def _batch_summary_billing_outcomes(
+    records: list[BatchRecord],
+) -> dict[str, dict]:
+    """Compute per-batch billing outcome for the Dashboard widget.
+
+    Returns ``{batch_id: {accepted, rejected, pending, billed,
+    top_rejection_reason, has_problem}}`` for every batch in
+    ``records``. Empty input → empty dict.
+
+    Two SQL queries, both bounded by the supplied batch ids:
+
+    1. One GROUP BY ``(batch_id, state)`` aggregate that produces
+       the accepted/rejected/pending counts and the sum of
+       ``charge_amount`` (the billed total). Single pass — no N+1.
+    2. One ordered scan over the rejected + payer-rejected subset
+       to pick the most recent rejection reason (truncated to 60
+       chars). Skipped when the first query found no rejections
+       and no payer-rejects, so the happy path stays at one query.
+
+    835 batches have no Claim rows — the GROUP BY returns no
+    rows for them, so the dict entry for an 835 batch is
+    ``{accepted:0, rejected:0, pending:0, billed:0.0,
+    top_rejection_reason:None, has_problem:False}`` (filled by
+    the caller's ``.get(id, defaults)`` pattern).
+    """
+    if not records:
+        return {}
+    from sqlalchemy import func  # local import to keep top-of-file light
+
+    batch_ids = [r.id for r in records]
+    outcome: dict[str, dict] = {
+        bid: {
+            "accepted": 0,
+            "rejected": 0,
+            "pending": 0,
+            "billed": 0.0,
+            "top_rejection_reason": None,
+            "has_problem": False,
+        }
+        for bid in batch_ids
+    }
+
+    with db.SessionLocal()() as s:
+        # ---- 1. GROUP BY (batch_id, state) for counts + billed total ----
+        rows = (
+            s.query(
+                Claim.batch_id,
+                Claim.state,
+                func.count(Claim.id),
+                func.coalesce(func.sum(Claim.charge_amount), 0),
+            )
+            .filter(Claim.batch_id.in_(batch_ids))
+            .group_by(Claim.batch_id, Claim.state)
+            .all()
+        )
+        any_rejection_or_payer = False
+        for batch_id, state, count, billed in rows:
+            slot = outcome.get(batch_id)
+            if slot is None:
+                continue  # batch has no row in our pre-allocated dict
+            count = int(count or 0)
+            billed_f = float(billed or 0)
+            slot["billed"] += billed_f
+            if state in _BATCH_SUMMARY_ACCEPTED_STATES:
+                slot["accepted"] += count
+            elif state in _BATCH_SUMMARY_REJECTED_STATES:
+                slot["rejected"] += count
+                any_rejection_or_payer = True
+            elif state in _BATCH_SUMMARY_PENDING_STATES:
+                slot["pending"] += count
+            # everything else (DRAFT, etc.) is excluded from the widget.
+
+        # ---- 2. Most-recent rejection reason + payer-reject probe ----
+        # Only run when we know there IS at least one rejection OR a
+        # payer-reject claim somewhere in the batch set; otherwise
+        # the first query alone is enough.
+        if any_rejection_or_payer:
+            rej_rows = (
+                s.query(
+                    Claim.batch_id,
+                    Claim.rejection_reason,
+                    Claim.payer_rejected_status_code,
+                )
+                .filter(
+                    Claim.batch_id.in_(batch_ids),
+                    Claim.state.in_(_BATCH_SUMMARY_REJECTED_STATES)
+                    | Claim.payer_rejected_status_code.in_(
+                        _BATCH_SUMMARY_PAYER_REJECT_CODES
+                    ),
+                )
+                .order_by(Claim.rejected_at.desc().nullslast())
+                .all()
+            )
+            seen_reason: set[str] = set()
+            for batch_id, reason, payer_code in rej_rows:
+                slot = outcome.get(batch_id)
+                if slot is None:
+                    continue
+                if payer_code in _BATCH_SUMMARY_PAYER_REJECT_CODES:
+                    slot["has_problem"] = True
+                # Capture the first non-null reason for this batch
+                # (rej_rows is ordered newest-first, so the first
+                # non-null wins). Truncate to 60 chars + ellipsis.
+                if (
+                    slot["top_rejection_reason"] is None
+                    and reason
+                    and batch_id not in seen_reason
+                ):
+                    r = reason.strip()
+                    if len(r) > 60:
+                        r = r[:60] + "…"
+                    slot["top_rejection_reason"] = r
+                    seen_reason.add(batch_id)
+                if (
+                    slot["rejected"] > 0
+                    or payer_code in _BATCH_SUMMARY_PAYER_REJECT_CODES
+                ):
+                    slot["has_problem"] = True
+
+    return outcome
+
+
+@router.get("/api/batches")
+def list_batches(
+    request: Request,
+    limit: int = Query(100, ge=1, le=1000),
+) -> Any:
+    """Summary of all parsed batches, newest first.
+
+    Each item includes ``claimIds`` (837P only) so the History tab
+    on the Upload page can render a one-click re-export button per
+    row without an extra round-trip to ``/api/batches/{id}``. The
+    list is still capped at ``limit`` claims; see the full result
+    via the by-id endpoint when more is needed.
+
+    SP30: also returns billing-outcome fields
+    (``acceptedCount`` / ``rejectedCount`` / ``pendingCount`` /
+    ``billedTotal`` / ``topRejectionReason`` / ``hasProblem``) so
+    the Dashboard "Recent batches" widget can render one row per
+    batch without an N+1 fetch. See
+    :func:`_batch_summary_billing_outcomes`.
+    """
+    records = store.list(limit=limit)
+    outcomes = _batch_summary_billing_outcomes(records)
+    items = [
+        {
+            "id": r.id,
+            "kind": r.kind,
+            "inputFilename": r.input_filename,
+            "parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
+            "claimCount": _batch_summary_claim_count(r),
+            "claimIds": _batch_summary_claim_ids(r),
+            "acceptedCount": outcomes.get(r.id, {}).get("accepted", 0),
+            "rejectedCount": outcomes.get(r.id, {}).get("rejected", 0),
+            "pendingCount": outcomes.get(r.id, {}).get("pending", 0),
+            "billedTotal": round(outcomes.get(r.id, {}).get("billed", 0.0), 2),
+            "topRejectionReason": outcomes.get(r.id, {}).get(
+                "top_rejection_reason"
+            ),
+            "hasProblem": outcomes.get(r.id, {}).get("has_problem", False),
+        }
+        for r in records
+    ]
+    all_records = store.all()
+    total = len(all_records)
+    returned = len(items)
+    has_more = total > returned
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(items, total, returned, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": items,
+        "total": total,
+        "returned": returned,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/batches/{batch_id}")
+def get_batch(batch_id: str) -> Any:
+    rec = store.get(batch_id)
+    if rec is None:
+        raise HTTPException(
+            status_code=404,
+            detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
+        )
+    return json.loads(rec.result.model_dump_json())

backend/src/cyclone/api_routers/claims.py

@@ -0,0 +1,472 @@
+"""``/api/claims*`` — Claims list / detail / streaming / serialize / line-reconciliation.
+
+Five endpoints, all gated by ``matrix_gate``:
+
+- ``GET /api/claims``                              — paginated list
+  with filter+sort, plus an NDJSON variant when the caller sends
+  ``Accept: application/x-ndjson``. SP27: counts the full filtered
+  population, not a page-limited sample.
+- ``GET /api/claims/stream``                       — NDJSON live-tail
+  on ``claim_written``. Snapshot first (eager
+  ``store.iter_claims``), then ``tail_events`` subscribes + emits
+  heartbeats. Registered before ``/api/claims/{claim_id}`` so the
+  literal ``stream`` segment isn't captured as a claim id.
+- ``GET /api/claims/{claim_id}``                   — full drawer
+  context (SP4) with the SP28 ``ack_links`` block pre-attached.
+  404 on missing id — never 500.
+- ``GET /api/claims/{claim_id}/serialize-837``     — regenerate X12
+  837P from the stored ``raw_json`` payload. 404 unknown claim, 422
+  no-``raw_json`` / unparseable / serializer failure.
+- ``GET /api/claims/{claim_id}/line-reconciliation`` — per-line 837
+  vs 835 side-by-side with CAS adjustments and a summary block.
+
+Three single-router helpers stay in this file (per spec D4):
+
+- :func:`_compact_ack_links_for_claim` — slim form
+  ``{ack_id, ack_kind, set_accept_reject_code, …}`` for the drawer
+  Acknowledgments panel.
+- :func:`_claim_line_dict`              — project an 837 service-line
+  dict from ``Claim.raw_json`` to wire shape.
+- :func:`_svc_to_dict`                  — project an ORM
+  ``ServiceLinePayment`` to wire shape.
+
+Inline imports inside handlers (preserved verbatim per spec D5):
+``select`` (sqlalchemy), ``LineReconciliation``/``ServiceLinePayment``/
+``CasAdjustment`` (cyclone.db), ``json as _json``, ``Decimal``.
+
+SP36 Task 15: this block moved here from ``api.py:1278`` (the 5
+``/api/claims*`` routes + 3 single-router helpers).
+"""
+from __future__ import annotations
+
+import json
+from decimal import Decimal
+from typing import Any, AsyncIterator
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import JSONResponse, Response, StreamingResponse
+
+from cyclone import db
+from cyclone.api_helpers import (
+    ndjson_line as _ndjson_line,
+    ndjson_stream_list as _ndjson_stream_list,
+    tail_events as _tail_events,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Claim
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.serialize_837 import SerializeError as SerializeError837
+from cyclone.parsers.serialize_837 import serialize_837
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/claims")
+def list_claims(
+    request: Request,
+    batch_id: str | None = Query(None),
+    status: str | None = Query(None),
+    provider_npi: str | None = Query(None),
+    payer: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+    offset: int = Query(0, ge=0),
+) -> Any:
+    common = dict(
+        batch_id=batch_id,
+        status=status,
+        provider_npi=provider_npi,
+        payer=payer,
+        date_from=date_from,
+        date_to=date_to,
+    )
+    items = list(store.iter_claims(
+        sort=sort, order=order, limit=limit, offset=offset, **common,
+    ))
+    # SP27 Task 13b: count the full population, not a 100-row sample.
+    # `iter_claims` defaults to limit=100; counting its output silently
+    # capped the reported total at 100 even when the DB held 60k rows.
+    total = store.count_claims(**common)
+    returned = len(items)
+    has_more = total > offset + returned
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(items, total, returned, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": items,
+        "total": total,
+        "returned": returned,
+        "has_more": has_more,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Live-tail NDJSON streaming endpoints (Phase 3 — SP5)
+# --------------------------------------------------------------------------- #
+
+
+@router.get("/api/claims/stream")
+async def claims_stream(
+    request: Request,
+    status: str | None = Query(None),
+    provider_npi: str | None = Query(None),
+    payer: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> StreamingResponse:
+    """Stream Claims as NDJSON: snapshot first, then live events.
+
+    Wire format:
+      * ``{"type":"item","data":<claim>}`` per snapshot row, then per
+        new ``claim_written`` event
+      * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
+      * ``{"type":"heartbeat","data":{"ts":<iso>}}`` every
+        ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle
+
+    Query params mirror :func:`list_claims` so a frontend can swap a
+    one-shot fetch for a tail with no URL surgery.
+
+    NOTE: registered before ``/api/claims/{claim_id}`` so the literal
+    ``stream`` path segment doesn't get matched as a claim id.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        # 1. Snapshot (eager — iter_claims returns a list already).
+        rows = store.iter_claims(
+            status=status, provider_npi=provider_npi, payer=payer,
+            date_from=date_from, date_to=date_to,
+            sort=sort or "-submission_date", order=order, limit=limit,
+        )
+        for row in rows:
+            yield _ndjson_line({"type": "item", "data": row})
+        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
+
+        # 2. Subscribe + heartbeats.
+        async for chunk in _tail_events(request, bus, ["claim_written"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
+@router.get("/api/claims/{claim_id}")
+def get_claim_detail_endpoint(claim_id: str) -> dict:
+    """Return one claim with full drawer context (SP4).
+
+    Body shape is produced by :meth:`CycloneStore.get_claim_detail`:
+    header, state, service lines, diagnoses, parties, validation,
+    raw segments, ``stateHistory`` (most-recent-first, capped at 50),
+    and a populated ``matchedRemittance`` block when paired.
+
+    SP28: response gains ``ack_links: list[dict]`` (compact form:
+    ``[{ack_id, ack_kind, set_accept_reject_code, parsed_at}]``)
+    so the ``ClaimDrawer`` Acknowledgments panel can render on
+    initial load. TA1 batch-level rows (``claim_id IS NULL``) are
+    excluded — those don't belong to a specific claim.
+
+    Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}``
+    convention). Returns 404 — never 500 — on a missing claim so the
+    UI can distinguish "doesn't exist" from a transient fetch error.
+    """
+    body = store.get_claim_detail(claim_id)
+    if body is None:
+        raise HTTPException(
+            status_code=404,
+            detail={
+                "error": "Not found",
+                "detail": f"Claim {claim_id} not found",
+            },
+        )
+    # SP28: attach ack_links (compact form for the drawer panel).
+    body["ack_links"] = _compact_ack_links_for_claim(claim_id)
+    return body
+
+
+def _compact_ack_links_for_claim(claim_id: str) -> list[dict]:
+    """Return compact ack_links for one claim, newest first.
+
+    TA1 batch-level rows (claim_id IS NULL) are filtered out — those
+    hang off the originating 837 batch, not a specific claim. The
+    shape is the slimmer ``{ack_id, ack_kind,
+    set_accept_reject_code, parsed_at, ak2_index}`` form so the
+    ClaimDrawer can render without an N+1 round-trip per row.
+    """
+    rows = store.list_acks_for_claim(claim_id)
+    out: list[dict] = []
+    for row in rows:
+        if row.claim_id is None:
+            continue
+        out.append({
+            "id": row.id,
+            "ack_id": row.ack_id,
+            "ack_kind": row.ack_kind,
+            "ak2_index": row.ak2_index,
+            "set_control_number": row.set_control_number,
+            "set_accept_reject_code": row.set_accept_reject_code,
+            "linked_at": (
+                row.linked_at.isoformat().replace("+00:00", "Z")
+                if row.linked_at is not None else ""
+            ),
+            "linked_by": row.linked_by,
+        })
+    return out
+
+
+@router.get("/api/claims/{claim_id}/serialize-837")
+def serialize_claim_as_837(claim_id: str):
+    """Return the claim as a regenerated X12 837P file (SP8).
+
+    Loads the ClaimOutput from the persisted ``raw_json`` and runs the
+    outbound serializer. Returns 404 if the claim doesn't exist, 422 if
+    the stored payload has no parseable ClaimOutput (data integrity
+    issue, not a transient failure).
+    """
+    with db.SessionLocal()() as s:
+        row = s.get(Claim, claim_id)
+        if row is None:
+            return JSONResponse(
+                {"error": "Not found", "detail": f"Claim {claim_id} not found"},
+                status_code=404,
+            )
+        if not row.raw_json:
+            return JSONResponse(
+                {
+                    "error": "Unprocessable",
+                    "detail": f"Claim {claim_id} has no raw_json; cannot serialize",
+                },
+                status_code=422,
+            )
+        try:
+            claim_obj = ClaimOutput.model_validate(row.raw_json)
+        except Exception as exc:
+            return JSONResponse(
+                {
+                    "error": "Unprocessable",
+                    "detail": f"Claim {claim_id} raw_json is malformed: {exc}",
+                },
+                status_code=422,
+            )
+
+    try:
+        text = serialize_837(claim_obj)
+    except SerializeError837 as exc:
+        return JSONResponse(
+            {"error": "Unprocessable", "detail": str(exc)},
+            status_code=422,
+        )
+
+    return Response(
+        content=text,
+        media_type="text/x12",
+        headers={
+            "Content-Disposition": f'attachment; filename="claim-{claim_id}.x12"'
+        },
+    )
+
+
+@router.get("/api/claims/{claim_id}/line-reconciliation")
+def get_claim_line_reconciliation(claim_id: str) -> dict:
+    """Per-line reconciliation view for the ClaimDrawer tab.
+
+    Spec §5.1. Returns the 837 service lines and 835 SVC composites
+    side-by-side, with per-line CAS adjustments and a summary block.
+
+    Architecture note: 837 service lines live in ``Claim.raw_json``
+    (not a separate ORM table), so the 837-side rows are read from the
+    JSON blob; the 835-side rows come from ``ServiceLinePayment`` ORM.
+    ``LineReconciliation.claim_service_line_number`` stores the 1-based
+    line number to join them.
+    """
+    from sqlalchemy import select
+    from cyclone.db import (
+        LineReconciliation, ServiceLinePayment, CasAdjustment,
+    )
+    import json as _json
+    from decimal import Decimal
+
+    with db.SessionLocal()() as s:
+        claim = s.get(db.Claim, claim_id)
+        if claim is None:
+            raise HTTPException(
+                status_code=404,
+                detail={"error": "Not found", "detail": f"Claim {claim_id} not found"},
+            )
+
+        # 837 service lines: from raw_json.
+        raw = claim.raw_json or {}
+        claim_lines_raw = raw.get("service_lines") or []
+        # Normalize to dicts for the response.
+        claim_lines = [_claim_line_dict(d) for d in claim_lines_raw]
+
+        # 835 service payments: ORM rows from the matched remit.
+        remits = list(
+            s.execute(
+                select(db.Remittance).where(db.Remittance.claim_id == claim_id)
+            ).scalars().all()
+        )
+        svc_payments: list[dict] = []
+        svc_ids: list[int] = []
+        if remits:
+            svc_rows = list(
+                s.execute(
+                    select(ServiceLinePayment).where(
+                        ServiceLinePayment.remittance_id.in_([r.id for r in remits])
+                    ).order_by(ServiceLinePayment.line_number)
+                ).scalars().all()
+            )
+            for svc in svc_rows:
+                d = _svc_to_dict(svc)
+                svc_payments.append(d)
+                svc_ids.append(svc.id)
+
+        # LineReconciliation rows.
+        lrs = list(
+            s.execute(
+                select(LineReconciliation).where(LineReconciliation.claim_id == claim_id)
+            ).scalars().all()
+        )
+        # Index by claim_service_line_number and service_line_payment_id.
+        lr_by_claim_num: dict[int, LineReconciliation] = {
+            lr.claim_service_line_number: lr for lr in lrs if lr.claim_service_line_number is not None
+        }
+        lr_by_svc: dict[int, LineReconciliation] = {
+            lr.service_line_payment_id: lr for lr in lrs if lr.service_line_payment_id is not None
+        }
+
+        # CAS rows grouped by svc id.
+        cas_by_svc: dict[int, list[CasAdjustment]] = {}
+        if svc_ids:
+            cas_rows = list(
+                s.execute(
+                    select(CasAdjustment).where(CasAdjustment.service_line_payment_id.in_(svc_ids))
+                ).scalars().all()
+            )
+            for c in cas_rows:
+                cas_by_svc.setdefault(c.service_line_payment_id, []).append(c)
+
+        # Build output lines array, preserving 837 order then 835-only.
+        svc_by_id: dict[int, dict] = {d["id"]: d for d in svc_payments}
+        lines_out: list[dict] = []
+        billed_total = Decimal("0")
+        paid_total = Decimal("0")
+        adjustment_total = Decimal("0")
+        matched_count = 0
+        used_svc_ids: set[int] = set()
+
+        for cl in claim_lines:
+            billed_total += Decimal(str(cl["charge"]))
+            lr = lr_by_claim_num.get(cl["line_number"])
+            if lr is None:
+                lines_out.append({
+                    "claim_service_line": cl,
+                    "service_line_payment": None,
+                    "status": "unmatched_837_only",
+                    "adjustments": [],
+                })
+                continue
+            svc_id = lr.service_line_payment_id
+            svc = svc_by_id.get(svc_id) if svc_id else None
+            if svc_id is not None:
+                used_svc_ids.add(svc_id)
+            cas_list = cas_by_svc.get(svc_id, []) if svc_id is not None else []
+            cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
+            if svc:
+                paid_total += Decimal(str(svc["payment"]))
+            adjustment_total += cas_total
+            if lr.status == "matched":
+                matched_count += 1
+            lines_out.append({
+                "claim_service_line": cl,
+                "service_line_payment": svc,
+                "status": lr.status,
+                "adjustments": [
+                    {"group_code": c.group_code, "reason_code": c.reason_code,
+                     "amount": str(Decimal(str(c.amount)))}
+                    for c in cas_list
+                ],
+            })
+
+        # 835-only lines (no claim match).
+        for lr in lrs:
+            if lr.claim_service_line_number is not None:
+                continue
+            svc_id = lr.service_line_payment_id
+            if svc_id is None:
+                continue
+            if svc_id in used_svc_ids:
+                continue
+            svc = svc_by_id.get(svc_id)
+            cas_list = cas_by_svc.get(svc_id, [])
+            cas_total = sum((Decimal(str(c.amount)) for c in cas_list), Decimal("0"))
+            if svc:
+                paid_total += Decimal(str(svc["payment"]))
+            adjustment_total += cas_total
+            lines_out.append({
+                "claim_service_line": None,
+                "service_line_payment": svc,
+                "status": lr.status,
+                "adjustments": [
+                    {"group_code": c.group_code, "reason_code": c.reason_code,
+                     "amount": str(Decimal(str(c.amount)))}
+                    for c in cas_list
+                ],
+            })
+
+        return {
+            "claim_id": claim_id,
+            "summary": {
+                "billed_total": str(billed_total),
+                "paid_total": str(paid_total),
+                "adjustment_total": str(adjustment_total),
+                "matched_lines": matched_count,
+                "total_lines": len(claim_lines),
+            },
+            "lines": lines_out,
+        }
+
+
+def _claim_line_dict(d: dict) -> dict:
+    """Project an 837 service-line dict from ``Claim.raw_json`` to wire shape."""
+    from decimal import Decimal
+    proc = d.get("procedure") or {}
+    charge = d.get("charge")
+    units = d.get("units")
+    return {
+        "line_number": d.get("line_number"),
+        "procedure_qualifier": proc.get("qualifier", "HC"),
+        "procedure_code": proc.get("code", ""),
+        "modifiers": proc.get("modifiers") or [],
+        "charge": str(Decimal(str(charge))) if charge is not None else "0",
+        "units": str(Decimal(str(units))) if units is not None else None,
+        "unit_type": d.get("unit_type"),
+        "service_date": d.get("service_date"),
+    }
+
+
+def _svc_to_dict(svc) -> dict:
+    """Project an ORM ``ServiceLinePayment`` to wire shape."""
+    import json as _json
+    from decimal import Decimal
+    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,
+    }

backend/src/cyclone/api_routers/clearhouse.py

@@ -0,0 +1,310 @@
+"""``/api/clearhouse*`` — singleton clearhouse config + SFTP submission.
+
+Three endpoints, all gated by ``matrix_gate``:
+
+- ``GET  /api/clearhouse``         — read the singleton clearhouse row
+  (dzinesco's identity, SFTP block, filename block). 404 when
+  unseeded.
+- ``PATCH /api/clearhouse``        — full-row replacement of the
+  singleton (SP25). Strict-validates ``sftp_block`` first (Pydantic
+  v2 default mode coerces strings-to-bools and would hide a real
+  operator mistake), then validates the whole body in loose mode.
+  Hot-reloads the running scheduler via
+  ``scheduler.reconfigure_scheduler`` so the next tick picks up the
+  new ``SftpBlock`` without a process restart.
+- ``POST /api/clearhouse/submit``  — submit a batch of claims to
+  the clearhouse. Stub: serializes via the SP7 serializer, builds
+  an HCPF-compliant outbound filename, copies the result to the
+  staging path. Per-claim audit events stamped with
+  ``actor="clearhouse-submit"``.
+
+Three single-router helpers stay in this file (per spec D4):
+
+- :func:`_load_claim_row`            — load a ``Claim`` row by id.
+- :func:`_serialize_claim_for_submit` — re-serialize a claim to X12
+  with optional per-call kwargs (submitter, receiver, SBR09, etc).
+- :func:`_serialize_claim_from_raw`  — best-effort serializer that
+  re-parses stored ``x12_text`` and re-emits.
+
+SP36 Task 11: this block moved here from ``api.py:2484`` (the 3
+``/api/clearhouse*`` routes + 3 single-router helpers).
+"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+
+from cyclone import db
+from cyclone.api_routers._shared import _actor_user_id
+from cyclone.audit_log import AuditEvent, append_event
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Claim
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/clearhouse")
+def get_clearhouse():
+    """Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
+    ch = store.get_clearhouse()
+    if ch is None:
+        raise HTTPException(status_code=404, detail="clearhouse not seeded")
+    return json.loads(ch.model_dump_json())
+
+
+@router.patch("/api/clearhouse")
+async def patch_clearhouse(body: dict) -> Any:
+    """Replace the singleton clearhouse row (SP25).
+
+    The full ``Clearhouse`` model is required — we don't accept partial
+    updates because the operator-facing use case is "I'm switching the
+    loop to real MFT" or "I'm pointing at a different MFT server",
+    not "I'm tweaking one field at a time." Validation errors are
+    returned as 422 (Pydantic default).
+
+    After a successful write, the running scheduler is hot-reloaded
+    via ``scheduler.reconfigure_scheduler()`` so the next tick uses
+    the new SftpBlock without a process restart.
+    """
+    from cyclone import scheduler as _scheduler_mod
+    from cyclone.providers import Clearhouse as _Clearhouse, SftpBlock as _SftpBlock
+
+    # Strict-validate the sftp_block sub-dict FIRST. Pydantic v2's
+    # default mode coerces strings to bools (e.g. ``"stub": "yes"``
+    # silently becomes True), which would hide a real operator
+    # mistake. The Clearhouse model itself stays in loose mode so
+    # ISO-string ``updated_at`` (the JSON round-trip shape) keeps
+    # parsing.
+    raw_sb = body.get("sftp_block", {})
+    try:
+        _SftpBlock.model_validate(raw_sb, strict=True)
+    except Exception as exc:
+        raise HTTPException(
+            status_code=422, detail=f"invalid sftp_block: {exc}",
+        ) from exc
+
+    # Now validate the full body in loose mode.
+    try:
+        parsed = _Clearhouse.model_validate(body)
+    except Exception as exc:
+        raise HTTPException(
+            status_code=422, detail=str(exc),
+        ) from exc
+
+    # SP25: when sftp_block.stub=false, the block must carry an auth
+    # account name and a non-empty host. The Pydantic model catches
+    # some of these; this catches the "empty password_keychain_account"
+    # case (which Pydantic allows because it's a free-form dict).
+    sb = parsed.sftp_block
+    if not sb.stub:
+        if not sb.host:
+            raise HTTPException(
+                status_code=422,
+                detail="sftp_block.host is required when stub=false",
+            )
+        auth = sb.auth or {}
+        if not auth.get("password_keychain_account") and not auth.get("key_file"):
+            raise HTTPException(
+                status_code=422,
+                detail=(
+                    "sftp_block.auth must contain either "
+                    "'password_keychain_account' or 'key_file' when stub=false"
+                ),
+            )
+
+    updated = store.update_clearhouse(parsed)
+    await _scheduler_mod.reconfigure_scheduler(
+        updated.sftp_block,
+        sftp_block_name=updated.name or "default",
+    )
+    return json.loads(updated.model_dump_json())
+
+
+@router.post("/api/clearhouse/submit")
+def submit_to_clearhouse(request: Request, body: dict):
+    """Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
+
+    Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
+
+    Stub behavior: serializes each claim via the SP7 serializer, builds
+    an HCPF-compliant outbound filename, and copies the result to
+    ``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
+    real SFTP connection. Returns a receipt per claim.
+    """
+    from cyclone.clearhouse import make_client
+    from cyclone.edi.filenames import build_outbound_filename
+
+    claim_ids = body.get("claim_ids", [])
+    payer_id = body.get("payer_id")
+    if not claim_ids:
+        raise HTTPException(status_code=400, detail="claim_ids required")
+    if not payer_id:
+        raise HTTPException(status_code=400, detail="payer_id required")
+
+    ch = store.get_clearhouse()
+    if ch is None:
+        raise HTTPException(status_code=500, detail="clearhouse not seeded")
+
+    # Submitter (Loop 1000A) comes from the clearhouse config. The
+    # receiver (NM1*40) and SBR09 come from the per-payer config and
+    # are resolved per-claim below. Without this wiring, the
+    # serializer would emit "CYCLONE" / "RECEIVER" placeholders and
+    # the file would be rejected by HCPF.
+    submitter_kwargs = {
+        "sender_id": ch.tpid,
+        "submitter_name": ch.submitter_name,
+        "submitter_contact_name": ch.submitter_contact_name,
+        "submitter_contact_email": ch.submitter_contact_email,
+    }
+    if getattr(ch, "submitter_contact_phone", None):
+        submitter_kwargs["submitter_contact_phone"] = ch.submitter_contact_phone
+
+    # Build a payer_id → PayerConfig837 map once so we can look up the
+    # receiver + SBR09 default for each claim.
+    from cyclone.db import PayerConfigORM as _PayerConfigORM
+
+    def _resolve_payer_cfg(claim_obj) -> dict | None:
+        pid = (claim_obj.payer.id or "").strip() if claim_obj.payer else ""
+        pname = (claim_obj.payer.name or "").strip() if claim_obj.payer else ""
+        with db.SessionLocal()() as ss:
+            if pid:
+                row = ss.get(_PayerConfigORM, (pid, "837P"))
+                if row is not None:
+                    return dict(row.config_json)
+            if pname:
+                rows = (
+                    ss.query(_PayerConfigORM)
+                    .filter(_PayerConfigORM.transaction_type == "837P")
+                    .all()
+                )
+                for r in rows:
+                    cj = dict(r.config_json)
+                    if (r.payer_id or "").upper() == pname.upper():
+                        return cj
+                    if pname.lower() in str(cj).lower():
+                        return cj
+            row = (
+                ss.query(_PayerConfigORM)
+                .filter(_PayerConfigORM.transaction_type == "837P")
+                .first()
+            )
+            return dict(row.config_json) if row else None
+
+    client = make_client(ch.sftp_block)
+    results = []
+    for cid in claim_ids:
+        try:
+            x12_text = _serialize_claim_for_submit(cid)
+        except Exception as exc:  # noqa: BLE001
+            results.append({"claim_id": cid, "ok": False, "error": str(exc)})
+            continue
+        # Re-resolve the claim so we can look up its payer config. We
+        # re-parse the stored x12_text to get a ClaimOutput (same path
+        # the serializer uses).
+        try:
+            from cyclone.parsers.parse_837 import parse as _parse837
+            claim_row_obj = _load_claim_row(cid)
+            if claim_row_obj is None or not (claim_row_obj.raw_json or {}).get("x12_text"):
+                raise RuntimeError("no stored x12_text for claim")
+            parsed = _parse837(claim_row_obj.raw_json["x12_text"])
+            claim_obj = parsed.claims[0] if parsed.claims else None
+        except Exception:
+            claim_obj = None
+        if claim_obj is not None:
+            payer_cfg = _resolve_payer_cfg(claim_obj) or {}
+            receiver_id = payer_cfg.get("receiver_id") or (claim_obj.payer.id if claim_obj.payer else None) or "RECEIVER"
+            receiver_name = payer_cfg.get("receiver_name") or (claim_obj.payer.name if claim_obj.payer else None) or receiver_id
+            sbr09 = payer_cfg.get("sbr09_default") or "MC"
+            # Re-serialize with the proper envelope values.
+            try:
+                x12_text = _serialize_claim_for_submit(
+                    cid,
+                    **{**submitter_kwargs, "receiver_id": receiver_id,
+                       "receiver_name": receiver_name,
+                       "claim_filing_indicator_code": sbr09},
+                )
+            except Exception as exc:  # noqa: BLE001
+                results.append({"claim_id": cid, "ok": False, "error": str(exc)})
+                continue
+        filename = build_outbound_filename(ch.tpid, "837P")
+        remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
+        staging_path = client.write_file(remote, x12_text.encode("utf-8"))
+        results.append({
+            "claim_id": cid,
+            "ok": True,
+            "filename": filename,
+            "staging_path": str(staging_path),
+            "remote_path": remote,
+        })
+        # SP11: audit trail for each successful clearhouse submission.
+        with db.SessionLocal()() as audit_s:
+            append_event(audit_s, AuditEvent(
+                event_type="clearhouse.submitted",
+                entity_type="claim",
+                entity_id=cid,
+                payload={
+                    "filename": filename,
+                    "remote_path": remote,
+                    "tpid": ch.tpid,
+                    "stub": ch.sftp_block.stub,
+                },
+                actor="clearhouse-submit",
+                user_id=_actor_user_id(request),
+            ))
+            audit_s.commit()
+    return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
+
+
+def _load_claim_row(claim_id: str):
+    """Helper: load a Claim row by id (or return None)."""
+    with db.SessionLocal()() as s:
+        return s.get(Claim, claim_id)
+
+
+def _serialize_claim_for_submit(claim_id: str, **kwargs) -> str:
+    """Serialize a claim to X12 for SFTP submission. Lazy import of the
+    serializer to avoid pulling FastAPI machinery at module import time.
+
+    Optional ``**kwargs`` are forwarded to the serializer — used to
+    pass through clearhouse submitter info and per-payer receiver info
+    so the regenerated file matches what the HCPF MFT expects.
+    """
+    from cyclone.parsers.serialize_837 import serialize_837
+    from cyclone import db
+    with db.SessionLocal()() as s:
+        row = s.get(db.Claim, claim_id)
+        if row is None:
+            raise ValueError(f"claim {claim_id!r} not found")
+        # Re-parse the stored raw_json to get a ClaimOutput
+        from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
+        from cyclone.parsers.parse_837 import parse
+        raw = row.raw_json or {}
+        # Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
+        return _serialize_claim_from_raw(row, raw, **kwargs)
+
+
+def _serialize_claim_from_raw(claim_row, raw: dict, **kwargs) -> str:
+    """Best-effort serializer that uses the stored raw_json to emit a fresh 837.
+
+    For SP9 this delegates to the existing serialize_837 helper if the
+    claim has a complete raw_segments array. Otherwise it returns a
+    minimal placeholder. ``**kwargs`` are forwarded to the serializer
+    so callers can pass through submitter / receiver / SBR09 values
+    from the clearhouse and per-payer configs.
+    """
+    from cyclone.parsers.serialize_837 import serialize_837
+    from cyclone.parsers.parse_837 import parse
+
+    # Re-parse the original batch text (need to re-derive from store).
+    # SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
+    if isinstance(raw, dict) and raw.get("x12_text"):
+        result = parse(raw["x12_text"])
+        if result.claims:
+            return serialize_837(result.claims[0], **kwargs)
+    # Fallback: raise so the caller sees an error.
+    raise RuntimeError(
+        f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
+    )

backend/src/cyclone/api_routers/config.py

@@ -0,0 +1,54 @@
+"""``/api/config/payers`` and ``/api/config/payers/{payer_id}/configs`` — payer-config read views.
+
+Both endpoints are read-only configuration surfaces used by the UI's
+"Edit payers" page:
+
+- ``GET /api/config/payers?is_active=...`` lists all configured
+  payers (PayerConfig records) — the set of payers the operator has
+  registered, regardless of whether they have inbound config blocks.
+- ``GET /api/config/payers/{payer_id}/configs`` returns the full
+  list of ``(transaction_type, config_json)`` blocks for a given
+  payer. Each block has a ``source``: ``"yaml"`` for the on-disk
+  ``config/payers.yaml`` default, ``"db"`` for any runtime override
+  recorded via ``/api/admin/reload-config``.
+
+These are configuration surfaces, not claim-processing surfaces.
+They live here (under ``/api/config/``) rather than under
+``/api/payers/`` because the latter is the drill-down rollup
+(see ``api_routers/payers.py``).
+
+SP36 Task 7: this block moved here from ``api.py:3167`` (after
+the SP21 provider-detail helper, before the Auth routers divider).
+"""
+from __future__ import annotations
+
+import json
+
+from fastapi import APIRouter, Depends, Query
+
+from cyclone import store
+from cyclone.auth.deps import matrix_gate
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/config/payers")
+def list_configured_payers(is_active: bool | None = Query(default=True)):
+    return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
+
+
+@router.get("/api/config/payers/{payer_id}/configs")
+def list_payer_configs(payer_id: str):
+    """List all (transaction_type, config_json) blocks for a payer."""
+    from cyclone import payers as payer_loader
+    configs = [
+        {"transaction_type": tx, "config_json": block, "source": "yaml"}
+        for (pid, tx), block in payer_loader.all_configs().items()
+        if pid == payer_id
+    ]
+    # Also check the DB for runtime-overridden configs
+    for tx in ("837P", "835", "277CA", "999", "TA1"):
+        live = store.get_payer_config(payer_id, tx)
+        if live is not None:
+            configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
+    return configs

backend/src/cyclone/api_routers/eligibility.py

@@ -0,0 +1,223 @@
+"""``/api/eligibility/request`` and ``/api/eligibility/parse-271`` — API-only eligibility pair.
+
+Builds a 270 inquiry from a small JSON body and parses a 271 response.
+Nothing is persisted to the DB — these are operator-driven, ephemeral
+operations per SP3 (P4 T23–T24). The 270 serializer pulls X12 from a
+``ParseResult270`` Pydantic; the 271 parser builds the same structure
+in reverse from the wire format.
+
+Why these are not ``GET /api/eligibility/...``: the 270 build is
+operator-initiated (pay-portal paste-back), so the inbound surface is
+a JSON ``POST``. The 271 inbound is a multipart file upload — same
+shape as ``/api/parse-999`` — so the file can be the actual 271 text
+saved from the payer portal.
+
+SP36 Task 5: this block moved here from ``api.py:2832`` (``270 / 271
+eligibility`` divider).
+"""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import date as _date
+from typing import Any
+
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from fastapi.responses import JSONResponse
+
+from cyclone.auth.deps import matrix_gate
+from cyclone.parsers.exceptions import CycloneParseError
+from cyclone.parsers.models import BatchSummary, Envelope
+from cyclone.parsers.models_270 import (
+    EligibilityBenefitInquiry,
+    InformationReceiver270,
+    InformationSource270,
+    ParseResult270,
+    Subscriber270,
+)
+from cyclone.parsers.parse_271 import parse as parse_271_text
+from cyclone.parsers.serialize_270 import serialize_270
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
+    """Build a :class:`ParseResult270` from a request body dict.
+
+    The body shape is the minimum surface needed to build a valid 270
+    inquiry (per spec section 3.4 — operator-driven, ephemeral):
+
+    ::
+
+        {
+            "subscriber": {first_name, last_name, member_id, dob},
+            "provider":   {npi, name},
+            "payer":      {id, name},
+            "service_type_code": "1"
+        }
+
+    Returns ``(ParseResult270, service_type_code)``. Raises
+    :class:`HTTPException` (400) when the body is missing required
+    fields.
+    """
+    subscriber_in = body.get("subscriber") or {}
+    provider_in = body.get("provider") or {}
+    payer_in = body.get("payer") or {}
+    service_type_code = (body.get("service_type_code") or "").strip()
+
+    # Required-field checks. We surface a single 400 with the first
+    # missing field name to match the rest of the API's error contract.
+    if not service_type_code:
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Bad request", "detail": "service_type_code is required"},
+        )
+    if not subscriber_in.get("member_id"):
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
+        )
+    if not provider_in.get("npi"):
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Bad request", "detail": "provider.npi is required"},
+        )
+    if not payer_in.get("name"):
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Bad request", "detail": "payer.name is required"},
+        )
+
+    # Build the Pydantic models. The serializer handles all envelope
+    # generation (sender_id/receiver_id/control_number/transaction_date
+    # are filled in by the serializer with sensible defaults).
+    subscriber_dob_raw = subscriber_in.get("dob")
+    subscriber_dob: _date | None = None
+    if subscriber_dob_raw:
+        try:
+            subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
+        except (TypeError, ValueError) as exc:
+            raise HTTPException(
+                status_code=400,
+                detail={
+                    "error": "Bad request",
+                    "detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
+                },
+            ) from exc
+
+    result = ParseResult270(
+        envelope=Envelope(
+            sender_id="SUBMITTERID",
+            receiver_id=str(payer_in.get("id") or "RECEIVERID"),
+            control_number="000000001",
+            transaction_date=_date.today(),
+            implementation_guide="005010X279A1",
+        ),
+        information_source=InformationSource270(
+            name=str(payer_in["name"]),
+            id=str(payer_in.get("id") or "") or None,
+        ),
+        information_receiver=InformationReceiver270(
+            name=str(provider_in.get("name") or ""),
+            npi=str(provider_in["npi"]),
+        ),
+        subscriber=Subscriber270(
+            member_id=str(subscriber_in["member_id"]),
+            first_name=str(subscriber_in.get("first_name") or "") or None,
+            last_name=str(subscriber_in.get("last_name") or "") or None,
+            dob=subscriber_dob,
+        ),
+        inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
+        summary=BatchSummary(
+            input_file="eligibility_request",
+            control_number="000000001",
+            transaction_date=_date.today(),
+            total_claims=1,
+            passed=1,
+            failed=0,
+        ),
+    )
+    return result, service_type_code
+
+
+@router.post("/api/eligibility/request")
+def post_eligibility_request(body: dict) -> Any:
+    """Build a 270 eligibility inquiry from a small JSON body.
+
+    Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
+    so the operator can either download the raw text (paste into a
+    payer portal) or render the parsed fields directly. Per spec
+    section 3.4, nothing is persisted to the DB.
+    """
+    try:
+        result, _ = _validate_eligibility_request(body)
+    except HTTPException:
+        raise
+    except (KeyError, TypeError, ValueError) as exc:
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
+        ) from exc
+
+    raw_270_text = serialize_270(result)
+    return {
+        "raw_270_text": raw_270_text,
+        "parsed": json.loads(result.model_dump_json()),
+    }
+
+
+@router.post("/api/eligibility/parse-271")
+async def post_eligibility_parse_271(
+    file: UploadFile = File(...),
+) -> Any:
+    """Parse a 271 eligibility response and return the structured summary.
+
+    Accepts the raw 271 text as a file upload (multipart/form-data),
+    mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
+    result is NOT persisted — the operator re-pastes the 271 each
+    time they need a fresh read.
+
+    The response body is a JSON object with three top-level keys:
+    ``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
+    returned on empty / undecodable / malformed EDI; 200 on success.
+    """
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    try:
+        result = parse_271_text(text, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure on 271")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    return {
+        "coverage_benefits": [
+            json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
+        ],
+        "subscriber": json.loads(result.subscriber.model_dump_json()),
+        "summary": json.loads(result.summary.model_dump_json()),
+        "envelope": json.loads(result.envelope.model_dump_json()),
+        "information_source": json.loads(result.information_source.model_dump_json()),
+        "information_receiver": json.loads(result.information_receiver.model_dump_json()),
+    }

backend/src/cyclone/api_routers/inbox.py

@@ -0,0 +1,342 @@
+"""``/api/inbox*`` — operator-facing Inbox surface (SP6 + SP14).
+
+Six endpoints, all gated by ``matrix_gate``:
+
+- ``GET  /api/inbox/lanes``                       — all lanes in one
+  call (``compute_lanes`` from :mod:`cyclone.inbox_lanes`).
+- ``POST /api/inbox/candidates/{remit_id}/match`` — manually link a
+  remit to a claim; surfaces 409 with the current state when the
+  claim is already matched.
+- ``POST /api/inbox/candidates/dismiss``          — add candidate
+  pairs to the session-scoped dismissed set (mutates
+  ``request.app.state.dismissed_pairs``).
+- ``POST /api/inbox/payer-rejected/acknowledge``  — SP14: mark
+  Payer-Rejected claims as acknowledged. Idempotent; returns the
+  count actually transitioned vs. already-acked / not-found /
+  not-rejected.
+- ``POST /api/inbox/rejected/resubmit``           — bulk move
+  REJECTED claims back to SUBMITTED. With
+  ``?download=true`` returns a ZIP of regenerated 837 files
+  (``serialize_837_for_resubmit`` with per-claim ``interchange_index``
+  for unique control numbers). Conflicts are omitted from the ZIP
+  and surfaced via the ``X-Cyclone-Serialize-Errors`` header.
+- ``GET  /api/inbox/export.csv``                  — stream a CSV for
+  a single lane (rejected / candidates / unmatched / done_today).
+
+All endpoints use ``request.app.state`` rather than the module-level
+``app`` global so they're robust against ``importlib.reload`` of
+the api module — the reload rebinds ``app`` to a new instance, but
+``request.app`` always points at the instance actually serving the
+current request.
+
+SP36 Task 13: this block moved here from ``api.py:1280`` (the 6
+``/api/inbox*`` routes, with the SP14 comment block preserved
+verbatim).
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+from fastapi.responses import StreamingResponse
+
+from cyclone import db
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Claim, ClaimState, Remittance
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.serialize_837 import SerializeError as SerializeError837
+from cyclone.parsers.serialize_837 import serialize_837_for_resubmit
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/inbox/lanes")
+def inbox_lanes(request: Request):
+    """Return all Inbox lanes in one call.
+
+    Uses ``request.app.state`` rather than the module-level ``app``
+    global so the endpoint is robust against ``importlib.reload`` of
+    this module (some tests do this to mutate the CORS allow-list).
+    After a reload, the module-level ``app`` rebinds to a new
+    FastAPI instance; ``request.app`` always points at the instance
+    that is actually serving the current request, so per-request
+    state stays consistent with the test's TestClient target.
+    """
+    dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
+    with db.SessionLocal()() as session:
+        from cyclone.inbox_lanes import compute_lanes
+        lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
+    return {
+        "rejected": lanes.rejected,
+        # SP10: payer-rejected lane (277CA STC A4/A6/A7). Distinct from
+        # the 999 envelope rejection in ``rejected`` above.
+        "payer_rejected": lanes.payer_rejected,
+        "candidates": lanes.candidates,
+        "unmatched": lanes.unmatched,
+        "done_today": lanes.done_today,
+    }
+
+
+@router.post("/api/inbox/candidates/{remit_id}/match")
+def inbox_match_candidate(remit_id: str, body: dict):
+    """Manually link a remit to a claim."""
+    claim_id = body.get("claim_id")
+    if not claim_id:
+        raise HTTPException(400, "claim_id required")
+    with db.SessionLocal()() as s:
+        claim = s.get(Claim, claim_id)
+        remit = s.get(Remittance, remit_id)
+        if claim is None or remit is None:
+            raise HTTPException(404, "claim or remit not found")
+        if claim.matched_remittance_id and claim.matched_remittance_id != remit_id:
+            raise HTTPException(
+                409,
+                detail={
+                    "error": "claim_already_matched",
+                    "current_state": (
+                        claim.state.value if hasattr(claim.state, "value")
+                        else str(claim.state)
+                    ),
+                    "matched_remittance_id": claim.matched_remittance_id,
+                },
+            )
+        claim.matched_remittance_id = remit_id
+        remit.claim_id = claim_id
+        s.commit()
+    return {"ok": True, "claim_id": claim_id, "remit_id": remit_id}
+
+
+@router.post("/api/inbox/candidates/dismiss")
+def inbox_dismiss_candidates(body: dict, request: Request):
+    """Add candidate pairs to the session-scoped dismissed set.
+
+    Uses ``request.app.state`` rather than the module-level ``app``
+    global so the endpoint is robust against ``importlib.reload`` of
+    this module (some tests do this to mutate the CORS allow-list).
+    After a reload, the module-level ``app`` rebinds to a new
+    FastAPI instance; ``request.app`` always points at the instance
+    that is actually serving the current request, so the test's
+    TestClient target is the one whose state we mutate.
+    """
+    pairs = body.get("pairs") or []
+    if not hasattr(request.app.state, "dismissed_pairs"):
+        request.app.state.dismissed_pairs = set()
+    for p in pairs:
+        cid = p.get("claim_id")
+        rid = p.get("remit_id")
+        if cid and rid:
+            request.app.state.dismissed_pairs.add(frozenset({cid, rid}))
+    return {"ok": True, "dismissed_count": len(pairs)}
+
+
+# --------------------------------------------------------------------------- #
+# SP14: Payer-Rejected acknowledge
+#
+# Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear
+# the claim from the working surface. We don't delete the rejection
+# (the original payer_rejected_* fields stay for SP11 audit), we just
+# set payer_rejected_acknowledged_at so the lane query filters it out.
+#
+# Idempotent: re-acknowledging an already-acknowledged claim is a noop
+# (the timestamp is not bumped). Returns the count actually transitioned
+# so the UI can show "3 of 5 were already acknowledged".
+# --------------------------------------------------------------------------- #
+@router.post("/api/inbox/payer-rejected/acknowledge")
+def inbox_acknowledge_payer_rejected(body: dict):
+    """Mark Payer-Rejected claims as acknowledged by the operator."""
+    claim_ids = body.get("claim_ids") or []
+    actor = body.get("actor") or "operator"
+    if not isinstance(claim_ids, list) or not claim_ids:
+        raise HTTPException(400, "claim_ids must be a non-empty list")
+    if not all(isinstance(c, str) for c in claim_ids):
+        raise HTTPException(400, "claim_ids must be a list of strings")
+
+    with db.SessionLocal()() as session:
+        from cyclone.db import Claim
+        now = datetime.now(timezone.utc)
+        transitioned = 0
+        already_acked = 0
+        not_found = 0
+        not_rejected = 0
+        for cid in claim_ids:
+            claim = session.get(Claim, cid)
+            if claim is None:
+                not_found += 1
+                continue
+            if claim.payer_rejected_at is None:
+                not_rejected += 1
+                continue
+            if claim.payer_rejected_acknowledged_at is not None:
+                already_acked += 1
+                continue
+            claim.payer_rejected_acknowledged_at = now
+            claim.payer_rejected_acknowledged_actor = actor
+            transitioned += 1
+            # SP11: audit event for the acknowledge action.
+            try:
+                from cyclone.audit_log import append_event, AuditEvent
+                append_event(session, AuditEvent(
+                    event_type="claim.payer_rejected_acknowledged",
+                    entity_type="claim",
+                    entity_id=claim.id,
+                    actor=actor,
+                    payload={
+                        "payer_rejected_status_code": claim.payer_rejected_status_code,
+                        "payer_rejected_by_277ca_id": claim.payer_rejected_by_277ca_id,
+                    },
+                ))
+            except Exception:  # noqa: BLE001
+                # Audit append is best-effort; don't block the operator's
+                # acknowledge action on an audit-log failure.
+                pass
+        if transitioned:
+            session.commit()
+
+    return {
+        "ok": True,
+        "transitioned": transitioned,
+        "already_acked": already_acked,
+        "not_found": not_found,
+        "not_rejected": not_rejected,
+    }
+
+
+@router.post("/api/inbox/rejected/resubmit")
+def inbox_resubmit_rejected(
+    request: Request,
+    body: dict,
+    download: bool = Query(False, description="When true, return a ZIP of regenerated 837 files for the resubmitted claims (instead of JSON)."),
+):
+    """Bulk move REJECTED claims back to SUBMITTED.
+
+    With ``?download=true``, the response is a ``application/zip`` archive
+    containing one ``claim-{id}.x12`` per successfully resubmitted claim
+    (regenerated via ``serialize_837_for_resubmit`` so each file gets a
+    unique interchange/group control number). Conflicts are omitted from
+    the ZIP — they remain visible to the caller via the JSON shape of the
+    non-download path. Empty resubmit + download → 200 with an empty zip
+    so the UI can still hand the user a downloadable artifact.
+    """
+    ids = body.get("claim_ids") or []
+    if not ids:
+        raise HTTPException(400, "claim_ids required")
+    accepted: list[str] = []
+    conflicts: list[dict] = []
+    # Track which claims are about to be resubmitted (and their index in
+    # the bundle) so the download path can serialize them with unique
+    # control numbers — back-to-back resubmits in the same file would
+    # otherwise all share ISA13/GS06 = "000000001".
+    accepted_with_rows: list[tuple[str, "Claim"]] = []
+    with db.SessionLocal()() as s:
+        for cid in ids:
+            c = s.get(Claim, cid)
+            if c is None:
+                continue
+            if c.state != ClaimState.REJECTED:
+                conflicts.append({
+                    "claim_id": cid,
+                    "current_state": (
+                        c.state.value if hasattr(c.state, "value")
+                        else str(c.state)
+                    ),
+                })
+                continue
+            c.state = ClaimState.SUBMITTED
+            c.state_changed_at = datetime.now(timezone.utc)
+            c.rejection_reason = None
+            c.rejected_at = None
+            c.resubmit_count = (c.resubmit_count or 0) + 1
+            accepted.append(cid)
+            accepted_with_rows.append((cid, c))
+        s.commit()
+
+    if not download:
+        return {"ok": True, "resubmitted": accepted, "conflicts": conflicts}
+
+    # Build a ZIP of regenerated 837s for the accepted claims. Conflicts
+    # and missing ids are deliberately excluded — the user already saw
+    # them in the JSON response on prior actions; the download is the
+    # "give me the files I asked for" payload.
+    import zipfile
+    buf = io.BytesIO()
+    serialize_errors: list[dict] = []
+    with zipfile.ZipFile(buf, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
+        for idx, (cid, c) in enumerate(accepted_with_rows, start=1):
+            if not c.raw_json:
+                serialize_errors.append({"claim_id": cid, "reason": "no raw_json"})
+                continue
+            try:
+                claim_obj = ClaimOutput.model_validate(c.raw_json)
+            except Exception as exc:
+                serialize_errors.append({"claim_id": cid, "reason": f"raw_json invalid: {exc}"})
+                continue
+            try:
+                text = serialize_837_for_resubmit(claim_obj, interchange_index=idx)
+            except SerializeError837 as exc:
+                serialize_errors.append({"claim_id": cid, "reason": str(exc)})
+                continue
+            zf.writestr(f"claim-{cid}.x12", text)
+    buf.seek(0)
+    headers = {
+        "Content-Disposition": (
+            f'attachment; filename="resubmit-{len(accepted)}-claims.zip"'
+        ),
+    }
+    # Surface per-claim serialization failures as a custom response header
+    # so the UI can show "10 resubmitted, 2 couldn't be regenerated" without
+    # parsing the binary. The header value is JSON-encoded; the UI is
+    # expected to JSON.parse it after a fetch with response.ok.
+    if serialize_errors:
+        headers["X-Cyclone-Serialize-Errors"] = json.dumps(serialize_errors)
+    return Response(
+        content=buf.getvalue(),
+        media_type="application/zip",
+        headers=headers,
+    )
+
+
+@router.get("/api/inbox/export.csv")
+def inbox_export_csv(lane: str, request: Request):
+    """Stream a CSV for a single lane.
+
+    Uses ``request.app.state`` rather than the module-level ``app``
+    global so the endpoint is robust against ``importlib.reload`` of
+    this module (see ``inbox_dismiss_candidates`` for context).
+    """
+    if lane not in {"rejected", "candidates", "unmatched", "done_today"}:
+        raise HTTPException(400, f"unknown lane: {lane}")
+    dismissed_pairs = getattr(request.app.state, "dismissed_pairs", set())
+    with db.SessionLocal()() as session:
+        from cyclone.inbox_lanes import compute_lanes
+        lanes = compute_lanes(session, dismissed_pairs=dismissed_pairs)
+    rows = getattr(lanes, lane)
+
+    buf = io.StringIO()
+    writer = csv.writer(buf)
+    writer.writerow([
+        "id", "kind", "patient_control_number", "charge_amount",
+        "payer_id", "provider_npi", "state", "rejection_reason",
+        "service_date", "score",
+    ])
+    for r in rows:
+        writer.writerow([
+            r.get("id") or r.get("payer_claim_control_number"),
+            r.get("kind"),
+            r.get("patient_control_number"),
+            r.get("charge_amount"),
+            r.get("payer_id"),
+            r.get("provider_npi") or r.get("rendering_provider_npi"),
+            r.get("state"),
+            r.get("rejection_reason"),
+            r.get("service_date_from") or r.get("service_date"),
+            r.get("score"),
+        ])
+    buf.seek(0)
+    return StreamingResponse(
+        iter([buf.getvalue()]),
+        media_type="text/csv",
+        headers={"Content-Disposition": f'attachment; filename="inbox-{lane}.csv"'},
+    )

backend/src/cyclone/api_routers/parse.py

@@ -0,0 +1,831 @@
+"""Parse endpoints — accept X12 uploads and ingest them.
+
+Five routes:
+
+* ``POST /api/parse-837`` — 837P professional claim ingest (the
+  primary upload path)
+* ``POST /api/parse-835`` — 835 ERA remittance ingest
+* ``POST /api/parse-999`` — 999 ACK ingest + auto-link claims
+* ``POST /api/parse-ta1`` — TA1 envelope ACK ingest + envelope-link batches
+* ``POST /api/parse-277ca`` — 277CA Claim Acknowledgment ingest
+
+The 7 cross-router helpers these endpoints need (and the two
+PAYER_FACTORIES dicts they consume) live in
+:mod:`cyclone.api_routers._shared`.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, Depends, File, Query, Request, UploadFile
+from fastapi.responses import JSONResponse, StreamingResponse
+from sqlalchemy.exc import IntegrityError
+
+from cyclone import db
+from cyclone.api_helpers import (
+    client_wants_json as _client_wants_json,
+    drop_raw_segments_837 as _drop_raw_segments,
+    drop_raw_segments_835 as _drop_raw_segments_835,
+    has_claim_validation_errors as _has_claim_validation_errors,
+    has_835_validation_errors as _has_835_validation_errors,
+    ndjson_stream_837 as _ndjson_stream,
+    ndjson_stream_835 as _ndjson_stream_835,
+    strict_rewrite_837 as _strict_rewrite,
+    strict_rewrite_835 as _strict_rewrite_835,
+)
+from cyclone.api_routers._shared import (
+    _actor_user_id,
+    _build_and_persist_ack,
+    _reconciliation_summary_for_batch,
+    _resolve_payer,
+    _resolve_payer_835,
+    _serialize_ta1,
+    _ta1_synthetic_source_batch_id,
+    _transaction_set_id_from_segments,
+)
+from cyclone.audit_log import AuditEvent, append_event
+from cyclone.auth.deps import matrix_gate
+from cyclone.claim_acks import (
+    apply_277ca_acks as _apply_277ca_acks,
+    apply_999_acceptances as _apply_999_acceptances,
+    apply_ta1_envelope_link as _apply_ta1_envelope_link,
+)
+from cyclone.db import Batch, Claim
+from cyclone.handlers._ack_id import (
+    ack_count_summary as _ack_count_summary,
+    ack_synthetic_source_batch_id as _ack_synthetic_source_batch_id,
+    two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id,
+)
+from cyclone.inbox_state import apply_999_rejections
+from cyclone.inbox_state_277ca import apply_277ca_rejections
+from cyclone.parsers.exceptions import CycloneParseError
+from cyclone.parsers.parse_277ca import parse_277ca_text
+from cyclone.parsers.parse_837 import parse
+from cyclone.parsers.parse_835 import parse as parse_835
+from cyclone.parsers.parse_999 import parse_999_text
+from cyclone.parsers.parse_ta1 import parse_ta1_text
+from cyclone.parsers.segments import tokenize as _tokenize_segments
+from cyclone.parsers.serialize_999 import serialize_999
+from cyclone.parsers.validator_835 import validate as validate_835
+from cyclone.store import BatchRecord, store, utcnow
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.post("/api/parse-837")
+async def parse_837(
+    request: Request,
+    file: UploadFile = File(...),
+    payer: str = Query("co_medicaid"),
+    include_raw_segments: bool = Query(True),
+    strict: bool = Query(False),
+    ack: bool = Query(False),
+) -> Any:
+    # SP35: defense-in-depth input guards. Layer A (UI auto-detect) lives
+    # in src/pages/Upload.tsx; the server-side checks below are the
+    # authoritative fix because they protect every caller of the API
+    # (Upload page, CLI ingestion, any future bulk-import tool). Without
+    # these, an 835 file dropped on the Upload page while the dropdown
+    # still says "837p" produces a BatchRecord with claims=[] and a bogus
+    # row on the History tab. The fix is two checks run BEFORE we persist
+    # anything:
+    #
+    #   1. Envelope check — ST01 must be "837" or "837P". Anything else
+    #      (an 835, a 999, a 270, garbage that happens to have an ISA)
+    #      → 400 with error="Mismatched file kind", expected="837p",
+    #      detected_st=<whatever was there>.
+    #   2. Empty-claims check — even with the right envelope, if the
+    #      parser produced zero CLM segments (truncated file, header-only
+    #      test fixture) → 400 with error="No claims parsed". A real
+    #      production 837 batch with zero claims is never valid.
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    config = _resolve_payer(payer)
+
+    # SP35 guard 1: envelope check. Tokenize first so we can return a
+    # precise 400 (vs. relying on the parser's "no ISA envelope" error
+    # which is correct but doesn't say "you sent an 835 to the 837
+    # endpoint"). If tokenization itself fails we fall through to the
+    # parser, which raises CycloneParseError → 400 "Parse error" path.
+    try:
+        _segments = _tokenize_segments(text)
+        detected_st = _transaction_set_id_from_segments(_segments) or ""
+    except CycloneParseError:
+        detected_st = ""
+
+    if detected_st and not detected_st.upper().startswith("837"):
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Mismatched file kind",
+                "expected": "837p",
+                "detected_st": detected_st,
+                "detail": (
+                    f"File declares ST*{detected_st}* but this endpoint "
+                    f"expects ST*837*. Pick the matching endpoint on the "
+                    f"Upload page (or let auto-detect choose for you)."
+                ),
+            },
+        )
+
+    try:
+        result = parse(text, config, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    # SP35 guard 2: empty-claims check. With the envelope validated, the
+    # only way to land here is a header-only file (real, but useless)
+    # or a file whose CLM loops the parser couldn't extract. Either way
+    # we refuse to persist — a BatchRecord with claims=[] is what the
+    # original bug produced and is never what the operator wanted.
+    if not result.claims:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "No claims parsed",
+                "detail": (
+                    "The file passed the envelope check but contained no "
+                    "CLM segments. Refusing to persist an empty batch."
+                ),
+            },
+        )
+
+    if strict:
+        result = _strict_rewrite(result)
+    if not include_raw_segments:
+        result = _drop_raw_segments(result)
+
+    if _has_claim_validation_errors(result):
+        # Per spec: 422 when claims failed validation.
+        # Body still includes the full ParseResult so the client can show errors.
+        # Validation-failed parses are NOT persisted (the data is suspect);
+        # only parses that survive validation end up in the store.
+        return JSONResponse(
+            status_code=422,
+            content=json.loads(result.model_dump_json()),
+        )
+
+    # Persist the cleaned-up result so the cleaned result is what clients
+    # retrieve via /api/batches/{id}.
+    rec = BatchRecord(
+        id=uuid.uuid4().hex,
+        kind="837p",
+        input_filename=file.filename or "upload.txt",
+        parsed_at=utcnow(),
+        result=result,
+    )
+    try:
+        store.add(rec, event_bus=request.app.state.event_bus)
+    except IntegrityError as exc:
+        # ``(batch_id, patient_control_number)`` is UNIQUE — fires when a
+        # single batch file contains the same CLM01 control number twice,
+        # or when the same claim id has already been ingested in a prior
+        # batch. Surface as 409 with the batch id so the caller can look
+        # it up; do NOT 500 (a 500 without CORS headers is misreported by
+        # browsers as a CORS error and hides the real cause).
+        log.warning("Duplicate claim while persisting batch %s: %s", rec.id, exc)
+        return JSONResponse(
+            status_code=409,
+            content={
+                "error": "Duplicate claim",
+                "detail": (
+                    "This file (or one previously ingested with the same "
+                    "claim control number) collides with an existing "
+                    "record. Inspect the file for duplicate CLM01 "
+                    "control numbers, or remove the existing batch "
+                    "before retrying."
+                ),
+                "batch_id": rec.id,
+            },
+        )
+
+    if _client_wants_json(request):
+        body = json.loads(result.model_dump_json())
+        if ack:
+            ack_body = _build_and_persist_ack(rec.id)
+            if ack_body is not None:
+                body["ack"] = ack_body
+        # Surface the server-side batch id so the frontend can call
+        # /api/batches/{id}/export-837 (and any other batch-scoped
+        # endpoint) without a separate listBatches round-trip.
+        body["batch_id"] = rec.id
+        return JSONResponse(content=body)
+
+    # Default: NDJSON stream. Pass the server-side batch id so the
+    # streaming client (the React Upload page) can call batch-scoped
+    # endpoints like /api/batches/{id}/export-837 without a separate
+    # GET /api/batches round-trip.
+    return StreamingResponse(
+        _ndjson_stream(result, batch_id=rec.id),
+        media_type="application/x-ndjson",
+    )
+
+
+@router.post("/api/parse-835")
+async def parse_835_endpoint(
+    request: Request,
+    file: UploadFile = File(...),
+    payer: str = Query("co_medicaid_835"),
+    include_raw_segments: bool = Query(True),
+    strict: bool = Query(False),
+) -> Any:
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    config = _resolve_payer_835(payer)
+
+    # SP35 guard 1: envelope check. Mirrors the parse-837 path: tokenize,
+    # read ST01, reject anything that doesn't start with "835". Same
+    # defense-in-depth rationale — the UI auto-detect (src/pages/Upload.tsx)
+    # is layer A, but server-side guards protect every API caller.
+    try:
+        _segments_835 = _tokenize_segments(text)
+        detected_st_835 = _transaction_set_id_from_segments(_segments_835) or ""
+    except CycloneParseError:
+        detected_st_835 = ""
+
+    if detected_st_835 and not detected_st_835.upper().startswith("835"):
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "Mismatched file kind",
+                "expected": "835",
+                "detected_st": detected_st_835,
+                "detail": (
+                    f"File declares ST*{detected_st_835}* but this endpoint "
+                    f"expects ST*835*. Pick the matching endpoint on the "
+                    f"Upload page (or let auto-detect choose for you)."
+                ),
+            },
+        )
+
+    try:
+        result = parse_835(text, config, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    # SP35 guard 2: empty-claims check. Same as parse-837: a BatchRecord
+    # with claims=[] is never a valid production 835 batch and we refuse
+    # to persist it.
+    if not result.claims:
+        return JSONResponse(
+            status_code=400,
+            content={
+                "error": "No claims parsed",
+                "detail": (
+                    "The file passed the envelope check but contained no "
+                    "CLP segments. Refusing to persist an empty batch."
+                ),
+            },
+        )
+
+    # Always run the validator; attach the report so the JSON path can
+    # surface it and the NDJSON path can fold the counts into the summary.
+    # 835 validation is batch-level, so pass/fail applies uniformly to every
+    # claim payment in the batch (passed=N or 0, failed=0 or N).
+    report = validate_835(result, config)
+    n = len(result.claims)
+    claim_ids = [c.payer_claim_control_number for c in result.claims]
+    if report.passed:
+        passed, failed, failed_claim_ids = n, 0, []
+    else:
+        passed, failed, failed_claim_ids = 0, n, claim_ids
+    result = result.model_copy(update={
+        "validation": report,
+        "summary": result.summary.model_copy(update={
+            "passed": passed,
+            "failed": failed,
+            "failed_claim_ids": failed_claim_ids,
+        }),
+    })
+
+    if strict:
+        result = _strict_rewrite_835(result)
+    if not include_raw_segments:
+        result = _drop_raw_segments_835(result)
+
+    if _has_835_validation_errors(result):
+        return JSONResponse(
+            status_code=422,
+            content=json.loads(result.model_dump_json()),
+        )
+
+    # Persist the cleaned-up result so the cleaned result is what clients
+    # retrieve via /api/batches/{id}.
+    rec = BatchRecord(
+        id=uuid.uuid4().hex,
+        kind="835",
+        input_filename=file.filename or "upload.txt",
+        parsed_at=utcnow(),
+        result=result,
+    )
+    try:
+        store.add(rec, event_bus=request.app.state.event_bus)
+    except IntegrityError as exc:
+        log.warning("Duplicate remittance while persisting batch %s: %s", rec.id, exc)
+        return JSONResponse(
+            status_code=409,
+            content={
+                "error": "Duplicate remittance",
+                "detail": (
+                    "This 835 file (or one previously ingested with the "
+                    "same payer claim control number) collides with an "
+                    "existing record. Remove the existing remittance "
+                    "before retrying."
+                ),
+                "batch_id": rec.id,
+            },
+        )
+
+    if _client_wants_json(request):
+        body = json.loads(result.model_dump_json())
+        body["reconciliation"] = _reconciliation_summary_for_batch(rec.id)
+        return JSONResponse(content=body)
+
+    # Default: NDJSON stream. Pass the server-side batch id so the
+    # streaming client can call batch-scoped endpoints without a
+    # separate GET /api/batches round-trip (see /api/parse-837 for the
+    # parallel change).
+    return StreamingResponse(
+        _ndjson_stream_835(result, batch_id=rec.id),
+        media_type="application/x-ndjson",
+    )
+
+
+@router.post("/api/parse-999")
+async def parse_999_endpoint(
+    request: Request,
+    file: UploadFile = File(...),
+) -> Any:
+    """Parse a 999 ACK file, persist a row, and return JSON.
+
+    Behavior mirrors ``/api/parse-835``:
+    - 400 on empty / undecodable / malformed EDI (never 500).
+    - 200 on success with ``{"ack": {id, accepted_count, rejected_count,
+      received_count, ack_code, raw_999_text}, "parsed": <ParseResult999>}``.
+
+    The persisted ``acks.source_batch_id`` is a synthetic id
+    (``999-<ISA13>``) because a received 999 has no inbound 837 batch
+    to FK to. The dashboard's `/acks` list surfaces these; operators
+    can still see which interchange each row came from.
+    """
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    try:
+        result = parse_999_text(text, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure on 999")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    received, accepted, rejected, ack_code = _ack_count_summary(result)
+    icn = result.envelope.control_number
+    synthetic_id = _ack_synthetic_source_batch_id(icn)
+
+    # Build the raw 999 text from the parsed result (round-trip).
+    raw_999_text = serialize_999(result, interchange_control_number=icn or "000000001")
+
+    # SP6 T4: move claims whose 999 set was rejected into ClaimState.REJECTED.
+    # The 999's set_control_number (AK202) is the source 837's ST02; in
+    # practice we look it up against patient_control_number because that's
+    # the field 999 ACKs cross-reference in this product.
+    with db.SessionLocal()() as session:
+        def _lookup(pcn: str):
+            return session.query(Claim).filter_by(patient_control_number=pcn).first()
+        _rejection_result = apply_999_rejections(
+            session, result, claim_lookup=_lookup,
+        )
+    if _rejection_result.matched:
+        bus = request.app.state.event_bus
+        for cid in _rejection_result.matched:
+            await bus.publish("claim.rejected", {"claim_id": cid})
+    if _rejection_result.orphans:
+        log.warning(
+            "999 had %d orphan set refs: %s",
+            len(_rejection_result.orphans),
+            _rejection_result.orphans[:5],
+        )
+
+    row = store.add_ack(
+        source_batch_id=synthetic_id,
+        accepted_count=accepted,
+        rejected_count=rejected,
+        received_count=received,
+        ack_code=ack_code,
+        raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
+    )
+
+    # SP11: append one audit row per rejected claim. Each row chains
+    # to the previous one — see cyclone.audit_log.
+    if _rejection_result.matched:
+        with db.SessionLocal()() as audit_s:
+            for cid in _rejection_result.matched:
+                append_event(audit_s, AuditEvent(
+                    event_type="claim.rejected",
+                    entity_type="claim",
+                    entity_id=cid,
+                    payload={"source_batch_id": synthetic_id, "ack_id": row.id},
+                    actor="999-parser",
+                    user_id=_actor_user_id(request),
+                ))
+            audit_s.commit()
+
+    # SP28: auto-link the 999 AK2 set-responses to claims via the
+    # D10 two-pass join (ST02 via batch envelope index primary,
+    # Claim.patient_control_number fallback). Each created ClaimAck
+    # row publishes claim_ack_written so the live-tail subscribers
+    # on the claim and ack side see the link immediately.
+    claim_ack_links_count = 0
+    with db.SessionLocal()() as link_s:
+        batch_index = store.batch_envelope_index()
+
+        def _pcn_lookup(pcn: str):
+            return (
+                link_s.query(Claim)
+                .filter(Claim.patient_control_number == pcn)
+                .first()
+            )
+
+        link_result = _apply_999_acceptances(
+            link_s, result, ack_id=row.id,
+            batch_envelope_index=batch_index,
+            pc_claim_lookup=_pcn_lookup,
+        )
+        for link_row in link_result.linked:
+            store.add_claim_ack(
+                claim_id=link_row.claim_id,
+                batch_id=link_row.batch_id,
+                ack_id=row.id,
+                ack_kind="999",
+                ak2_index=link_row.ak2_index,
+                set_control_number=link_row.set_control_number,
+                set_accept_reject_code=link_row.set_accept_reject_code,
+                linked_by="auto",
+                event_bus=request.app.state.event_bus,
+            )
+            claim_ack_links_count += 1
+
+    return JSONResponse(content={
+        "ack": {
+            "id": row.id,
+            "accepted_count": accepted,
+            "rejected_count": rejected,
+            "received_count": received,
+            "ack_code": ack_code,
+            "source_batch_id": synthetic_id,
+            "raw_999_text": raw_999_text,
+            "claim_ack_links_count": claim_ack_links_count,
+        },
+        "parsed": json.loads(result.model_dump_json()),
+    })
+
+
+@router.post("/api/parse-ta1")
+async def parse_ta1_endpoint(
+    request: Request,
+    file: UploadFile = File(...),
+) -> Any:
+    """Parse a TA1 (Interchange Acknowledgment) file, persist a row, return JSON.
+
+    Mirrors ``/api/parse-999`` but for the lower-level envelope ack:
+      - 400 on empty / undecodable / malformed EDI (never 500).
+      - 200 on success with ``{"ta1": {id, control_number, ack_code,
+        note_code, interchange_date, interchange_time, sender_id,
+        receiver_id, source_batch_id, raw_ta1_text}, "parsed":
+        <ParseResultTa1>}``.
+
+    The persisted ``ta1_acks.source_batch_id`` is a synthetic id
+    (``TA1-<ISA13>``) because a received TA1 has no inbound batch to
+    FK to. The dashboard's ``/ta1-acks`` list surfaces these.
+
+    SP25: threads ``event_bus=request.app.state.event_bus`` into
+    ``store.add_ta1_ack`` so the live-tail ``ta1_ack_received``
+    stream fires on manual uploads (not just on the SFTP poller).
+    """
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    try:
+        result = parse_ta1_text(text, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure on TA1")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    # Build the raw TA1 text from the parsed result (round-trip).
+    raw_ta1_text = _serialize_ta1(result)
+
+    row = store.add_ta1_ack(
+        source_batch_id=result.source_batch_id,
+        control_number=result.ta1.control_number,
+        interchange_date=result.ta1.interchange_date,
+        interchange_time=result.ta1.interchange_time,
+        ack_code=result.ta1.ack_code,
+        note_code=result.ta1.note_code,
+        ack_generated_date=result.ta1.ack_generated_date,
+        sender_id=result.envelope.sender_id,
+        receiver_id=result.envelope.receiver_id,
+        raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
+    )
+
+    # SP28: TA1 envelope-level link to the originating Batch. The
+    # closure here matches the most-recent Batch whose envelope
+    # sender_id/receiver_id matches the TA1 — see spec §D4.
+    claim_ack_links_count = 0
+    with db.SessionLocal()() as link_s:
+        def _batch_lookup(sender_id, receiver_id):
+            rows = (
+                link_s.query(Batch)
+                .filter(
+                    Batch.kind == "837p",
+                    Batch.raw_result_json.isnot(None),
+                )
+                .order_by(Batch.parsed_at.desc())
+                .all()
+            )
+            for row in rows:
+                env = (row.raw_result_json or {}).get("envelope") or {}
+                if (
+                    env.get("sender_id") == sender_id
+                    and env.get("receiver_id") == receiver_id
+                ):
+                    return row
+            return None
+
+        link_result = _apply_ta1_envelope_link(
+            link_s, result, ack_id=row.id,
+            batch_lookup=_batch_lookup,
+        )
+        for link_row in link_result.linked:
+            store.add_claim_ack(
+                claim_id=link_row.claim_id,
+                batch_id=link_row.batch_id,
+                ack_id=row.id,
+                ack_kind="ta1",
+                ak2_index=link_row.ak2_index,
+                set_control_number=link_row.set_control_number,
+                set_accept_reject_code=link_row.set_accept_reject_code,
+                linked_by="auto",
+                event_bus=request.app.state.event_bus,
+            )
+            claim_ack_links_count += 1
+
+    return JSONResponse(content={
+        "ta1": {
+            "id": row.id,
+            "control_number": result.ta1.control_number,
+            "ack_code": result.ta1.ack_code,
+            "note_code": result.ta1.note_code,
+            "interchange_date": result.ta1.interchange_date.isoformat()
+                if result.ta1.interchange_date else None,
+            "interchange_time": result.ta1.interchange_time,
+            "sender_id": result.envelope.sender_id,
+            "receiver_id": result.envelope.receiver_id,
+            "source_batch_id": result.source_batch_id,
+            "raw_ta1_text": raw_ta1_text,
+            "claim_ack_links_count": claim_ack_links_count,
+        },
+        "parsed": json.loads(result.model_dump_json()),
+    })
+
+
+@router.post("/api/parse-277ca")
+async def parse_277ca_endpoint(
+    request: Request,
+    file: UploadFile = File(...),
+) -> Any:
+    """Parse a 277CA Claim Acknowledgment file, persist a row, and stamp rejections.
+
+    Behavior mirrors ``/api/parse-999``:
+    - 400 on empty / undecodable / malformed EDI (never 500).
+    - 200 on success with ``{"ack": {id, control_number, accepted_count,
+      rejected_count, payer_claim_control_numbers, raw_277ca_text},
+      "parsed": <ParseResult277CA>}``.
+
+    After parse, runs :func:`apply_277ca_rejections` to stamp the
+    payer-rejected fields on each matching claim row. The Inbox
+    Payer-Rejected lane lights up as a side-effect of this call.
+    """
+    raw = await file.read()
+    if not raw:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
+        )
+    try:
+        text = raw.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Encoding error", "detail": str(exc)},
+        )
+
+    try:
+        result = parse_277ca_text(text, input_file=file.filename or "")
+    except CycloneParseError as exc:
+        return JSONResponse(
+            status_code=400,
+            content={"error": "Parse error", "detail": str(exc)},
+        )
+    except Exception as exc:  # pragma: no cover - safety net
+        log.exception("Unexpected parser failure on 277CA")
+        return JSONResponse(
+            status_code=500,
+            content={"error": "Internal server error", "detail": str(exc)},
+        )
+
+    icn = result.envelope.control_number
+    synthetic_id = _277ca_synthetic_source_batch_id(icn)
+
+    accepted = sum(1 for s in result.claim_statuses if s.classification == "accepted")
+    paid = sum(1 for s in result.claim_statuses if s.classification == "paid")
+    rejected = sum(1 for s in result.claim_statuses if s.classification == "rejected")
+    pended = sum(1 for s in result.claim_statuses if s.classification == "pended")
+
+    # Persist the 277CA row first so we have an id to attach to claims.
+    # SP25: thread the event bus so ``two77ca_ack_received`` fires.
+    row = store.add_277ca_ack(
+        source_batch_id=synthetic_id,
+        control_number=icn,
+        accepted_count=accepted,
+        rejected_count=rejected,
+        paid_count=paid,
+        pended_count=pended,
+        raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
+    )
+
+    # Stamp payer-rejection fields on matching claims. The 277CA's
+    # REF*1K carries the patient's claim control number we sent in
+    # CLM01 — same convention the 999 ACK uses, so the lookup hits
+    # Claim.patient_control_number (mirrors apply_999_rejections).
+    with db.SessionLocal()() as session:
+        def _lookup(pcn: str):
+            return (
+                session.query(Claim)
+                .filter(Claim.patient_control_number == pcn)
+                .first()
+            )
+        apply_result = apply_277ca_rejections(
+            session, result, claim_lookup=_lookup, two77ca_id=row.id,
+        )
+    if apply_result.matched:
+        bus = request.app.state.event_bus
+        for cid in apply_result.matched:
+            await bus.publish("claim.payer_rejected", {"claim_id": cid})
+        # SP11: audit trail for each payer-rejected claim.
+        with db.SessionLocal()() as audit_s:
+            for cid in apply_result.matched:
+                append_event(audit_s, AuditEvent(
+                    event_type="claim.payer_rejected",
+                    entity_type="claim",
+                    entity_id=cid,
+                    payload={
+                        "source_batch_id": synthetic_id,
+                        "277ca_id": row.id,
+                    },
+                    actor="277ca-parser",
+                    user_id=_actor_user_id(request),
+                ))
+            audit_s.commit()
+    if apply_result.orphans:
+        log.warning(
+            "277CA had %d orphan status entries (no matching claim): %s",
+            len(apply_result.orphans),
+            apply_result.orphans[:5],
+        )
+
+    # SP28: auto-link the 277CA ClaimStatus entries to claims via
+    # the D10 two-pass join. Each ClaimAck row publishes
+    # claim_ack_written so the live-tail subscribers on the claim
+    # and ack side see the link immediately.
+    claim_ack_links_count = 0
+    with db.SessionLocal()() as link_s:
+        batch_index = store.batch_envelope_index()
+
+        def _pcn_lookup(pcn: str):
+            return (
+                link_s.query(Claim)
+                .filter(Claim.patient_control_number == pcn)
+                .first()
+            )
+
+        link_result = _apply_277ca_acks(
+            link_s, result, ack_id=row.id,
+            batch_envelope_index=batch_index,
+            pc_claim_lookup=_pcn_lookup,
+        )
+        for link_row in link_result.linked:
+            store.add_claim_ack(
+                claim_id=link_row.claim_id,
+                batch_id=link_row.batch_id,
+                ack_id=row.id,
+                ack_kind="277ca",
+                ak2_index=link_row.ak2_index,
+                set_control_number=link_row.set_control_number,
+                set_accept_reject_code=link_row.set_accept_reject_code,
+                linked_by="auto",
+                event_bus=request.app.state.event_bus,
+            )
+            claim_ack_links_count += 1
+
+    return JSONResponse(content={
+        "ack": {
+            "id": row.id,
+            "control_number": icn,
+            "accepted_count": accepted,
+            "rejected_count": rejected,
+            "paid_count": paid,
+            "pended_count": pended,
+            "source_batch_id": synthetic_id,
+            "matched_claim_ids": apply_result.matched,
+            "orphan_status_codes": apply_result.orphans,
+            "claim_ack_links_count": claim_ack_links_count,
+        },
+        "parsed": json.loads(result.model_dump_json()),
+    })

backend/src/cyclone/api_routers/payers.py

@@ -0,0 +1,149 @@
+"""``GET /api/payers/{payer_id}/summary`` — payer-level rollup for the drill-down panel.
+
+Returns billed/received totals, denial rate, and the top providers for
+a given ``payer_id`` (the X12 NM1*PR*PI qualifier, e.g. ``SKCO0``).
+Cached in-process for ``_SUMMARY_TTL_S`` seconds via a per-payer_id
+memo. The drill-down UI hammers this on every hover; the underlying
+query is O(claims) per payer, so the TTL keeps the panel responsive.
+
+Pubsub invalidation is intentionally NOT wired (see the long comment
+in the body). The 60s TTL keeps staleness bounded; a follow-up can
+wire targeted invalidation if the UI proves TTL-bounded staleness is
+unacceptable.
+
+404 when no claims AND no remits reference this payer_id — the UI
+uses that to distinguish a typo from a payer with zero traffic (the
+latter would still return a valid 200 with zeroed totals).
+
+SP36 Task 6: this block moved here from ``api.py:3188`` (the
+``SP21 Task 1.5: payer-level summary`` divider).
+"""
+from __future__ import annotations
+
+import logging
+from time import monotonic
+
+from fastapi import APIRouter, Depends, HTTPException
+
+from cyclone import db
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Claim, ClaimState, Remittance
+
+log = logging.getLogger(__name__)
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+# Per-payer_id memoization for /api/payers/{payer_id}/summary. The drill-
+# down UI hammers this on every hover; the underlying query is O(claims)
+# per payer so a 60s TTL keeps the panel responsive. Process-local on
+# purpose — invalidation is driven by the TTL alone for now (see note
+# below).
+_SUMMARY_TTL_S = 60.0
+_summary_cache: dict[str, tuple[float, dict]] = {}
+
+
+def _clear_summary_cache() -> None:
+    """Test hook: wipe the process-local cache.
+
+    The 60s TTL means tests that want to assert on a recomputed payload
+    must clear the cache between requests. Not used by production code.
+    """
+    _summary_cache.clear()
+
+
+# Pubsub invalidation is intentionally NOT wired. The ``claim_written``
+# and ``remittance_written`` payloads emitted by ``store.add`` are the
+# UI-shaped dicts from ``to_ui_claim_from_orm`` / ``to_ui_remittance_from_orm``,
+# neither of which carries the X12 ``payer_id`` (NM1*PR*PI qualifier).
+# They carry ``payerName`` only, which is the human-readable name and not
+# the URL key we cache on. Wiring a subscriber here would either need a
+# DB lookup per event (re-deriving payer_id from Claim.id) or a different
+# cache key — both are out of scope for this task. The 60s TTL keeps
+# staleness bounded; a follow-up can wire targeted invalidation if the
+# UI proves TTL-bounded staleness is unacceptable.
+
+
+@router.get("/api/payers/{payer_id}/summary")
+def get_payer_summary(payer_id: str):
+    """Payer-level rollup for the drill-down panel.
+
+    Returns billed/received totals, denial rate, and top providers for
+    the given ``payer_id`` (the X12 NM1*PR*PI qualifier, e.g. ``SKCO0``).
+    Cached in-process for ``_SUMMARY_TTL_S`` seconds.
+
+    404 when no claims AND no remits reference this payer_id — the UI
+    uses that to distinguish a typo from a payer with zero traffic
+    (the latter would still return a valid 200 with zeroed totals).
+    """
+    now = monotonic()
+    cached = _summary_cache.get(payer_id)
+    if cached and (now - cached[0]) < _SUMMARY_TTL_S:
+        return cached[1]
+
+    log.debug("payer summary cache miss", extra={"payer_id": payer_id})
+
+    # Query claims + remits scoped to this payer_id. We bypass
+    # ``store.iter_claims`` because that helper filters by payer NAME
+    # substring, not by the X12 PI qualifier we use as the URL key.
+    with db.SessionLocal()() as s:
+        claim_rows: list[Claim] = (
+            s.query(Claim).filter(Claim.payer_id == payer_id).all()
+        )
+        claim_ids = [c.id for c in claim_rows]
+        remit_rows: list[Remittance] = []
+        if claim_ids:
+            remit_rows = (
+                s.query(Remittance)
+                .filter(Remittance.claim_id.in_(claim_ids))
+                .all()
+            )
+
+        if not claim_rows and not remit_rows:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Payer {payer_id!r} not found",
+            )
+
+        billed_total = sum(float(c.charge_amount or 0) for c in claim_rows)
+        received_total = sum(float(r.total_paid or 0) for r in remit_rows)
+        denied = sum(
+            1 for c in claim_rows if c.state == ClaimState.DENIED
+        )
+        claim_count = len(claim_rows)
+        denial_rate = (denied / claim_count) if claim_count else 0.0
+
+        provider_counts: dict[str, int] = {}
+        for c in claim_rows:
+            npi = c.provider_npi
+            if not npi:
+                continue
+            provider_counts[npi] = provider_counts.get(npi, 0) + 1
+        top_providers = [
+            {"npi": npi, "count": count}
+            for npi, count in sorted(
+                provider_counts.items(), key=lambda kv: -kv[1]
+            )[:5]
+        ]
+
+        # Best-effort payer display name from the first claim's raw
+        # payer object (NM1*PR name element, e.g. "COHCPF"). Falls
+        # back to the id when no parsed envelope is available.
+        payer_name = payer_id
+        for c in claim_rows:
+            raw = c.raw_json or {}
+            p = raw.get("payer") if isinstance(raw, dict) else None
+            if isinstance(p, dict) and p.get("name"):
+                payer_name = p["name"]
+                break
+
+    payload = {
+        "payer_id": payer_id,
+        "name": payer_name,
+        "claim_count": claim_count,
+        "billed_total": billed_total,
+        "received_total": received_total,
+        "denial_rate": denial_rate,
+        "top_providers": top_providers,
+    }
+    _summary_cache[payer_id] = (now, payload)
+    return payload

backend/src/cyclone/api_routers/providers.py

@@ -0,0 +1,150 @@
+"""``/api/providers`` and ``/api/config/providers*`` — read views over providers.
+
+Three endpoints across two URL prefixes, all gated by ``matrix_gate``:
+
+- ``GET /api/providers``                  — distinct provider list
+  derived from the Claims population (``store.distinct_providers()``),
+  with ``npi`` / ``state`` filters, pagination, and an NDJSON variant.
+- ``GET /api/config/providers``           — the configured provider
+  rows (3 NPIs for SP9), filtered by ``is_active``.
+- ``GET /api/config/providers/{npi}``     — one configured provider
+  plus a small drill-down block: ``recent_claims`` (top-10 by
+  ``submissionDate``) and ``recent_activity`` (top-10, joined via
+  ``Claim.id`` because ``ActivityEvent`` has no ``provider_npi``
+  column; the outer-join via ``Remittance.claim_id`` surfaces the
+  orphan ``remit_received`` events that were recorded pre-match).
+
+SP36 Task 12: this block moved here from ``api.py:2448`` (the
+``/api/providers`` list, the ``/api/config/providers`` list, and
+the ``/api/config/providers/{npi}`` detail) — three URL prefixes,
+one router per spec.
+"""
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import StreamingResponse
+from sqlalchemy import desc, or_
+
+from cyclone import db
+from cyclone.api_helpers import (
+    ndjson_stream_list as _ndjson_stream_list,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.db import Claim, Remittance
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/providers")
+def list_providers(
+    request: Request,
+    npi: str | None = Query(None),
+    state: str | None = Query(None),
+    limit: int = Query(100, ge=1, le=1000),
+    offset: int = Query(0, ge=0),
+) -> Any:
+    items = store.distinct_providers()
+    if npi is not None:
+        items = [p for p in items if p["npi"] == npi]
+    if state is not None:
+        items = [p for p in items if p.get("state") == state]
+    paged = items[offset:offset + limit]
+    total = len(items)
+    returned = len(paged)
+    has_more = total > offset + returned
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(paged, total, returned, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": paged,
+        "total": total,
+        "returned": returned,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/config/providers")
+def list_configured_providers(is_active: bool | None = Query(default=True)):
+    """List the configured provider rows (3 NPIs for SP9)."""
+    return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
+
+
+@router.get("/api/config/providers/{npi}")
+def get_configured_provider(npi: str):
+    p = store.get_provider(npi)
+    if p is None:
+        raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
+    provider_dict = json.loads(p.model_dump_json())
+
+    # SP21 Task 1.6: extend the response with two top-N arrays that the
+    # drill-down peek panel hangs off. ``recent_claims`` reuses the
+    # existing store projection (already returns UI-shaped dicts with
+    # ``submissionDate``); ``recent_activity`` is a direct ORM join
+    # because ``ActivityEvent`` has no ``provider_npi`` column — the
+    # filter has to hop through ``Claim.id``.
+    recent_claims = sorted(
+        store.iter_claims(provider_npi=npi),
+        key=lambda c: c.get("submissionDate") or "",
+        reverse=True,
+    )[:10]
+
+    # Activity filter has TWO join paths back to a Claim for this
+    # provider:
+    #   1. ``ActivityEvent.claim_id IN (claim_ids)`` — events that were
+    #      recorded with a claim FK already set (claim_submitted,
+    #      manual_match, claim_paid, etc.).
+    #   2. ``Remittance.claim_id IN (claim_ids)`` — the original
+    #      ``remit_received`` event recorded at 835 ingest time
+    #      (``store.add`` lines 999-1003) is inserted with
+    #      ``claim_id=None`` because the remittance hasn't been matched
+    #      to a claim yet. The auto-reconcile pass later sets
+    #      ``Remittance.claim_id`` (``reconcile.run`` lines 289-293),
+    #      but the *original* ActivityEvent row stays orphaned. The
+    #      outer-join-then-OR lets us surface both shapes. Without
+    #      path 2, a provider's activity feed looks frozen the moment
+    #      an 835 lands — the most common activity, invisible.
+    with db.SessionLocal()() as s:
+        claim_ids = [
+            cid
+            for (cid,) in s.query(Claim.id)
+            .filter(Claim.provider_npi == npi)
+            .all()
+        ]
+        activity_rows = []
+        if claim_ids:
+            activity_rows = (
+                s.query(db.ActivityEvent)
+                .outerjoin(
+                    Remittance,
+                    db.ActivityEvent.remittance_id == Remittance.id,
+                )
+                .filter(or_(
+                    db.ActivityEvent.claim_id.in_(claim_ids),
+                    Remittance.claim_id.in_(claim_ids),
+                ))
+                .order_by(desc(db.ActivityEvent.ts))
+                .limit(10)
+                .all()
+            )
+
+    def _activity_to_ui(a):
+        return {
+            "id": a.id,
+            "ts": a.ts.isoformat().replace("+00:00", "Z") if a.ts else "",
+            "kind": a.kind,
+            "batchId": a.batch_id,
+            "claimId": a.claim_id,
+            "remittanceId": a.remittance_id,
+            "payload": a.payload_json or {},
+        }
+
+    provider_dict["recent_claims"] = recent_claims
+    provider_dict["recent_activity"] = [_activity_to_ui(a) for a in activity_rows]
+    return provider_dict

backend/src/cyclone/api_routers/reconciliation.py

@@ -0,0 +1,178 @@
+"""Reconciliation read views + manual match/unmatch write paths.
+
+Four endpoints:
+
+- ``GET  /api/reconciliation/unmatched`` — list of unmatched Claims
+  and unmatched Remittances (``store.list_unmatched(kind="both")``).
+- ``GET  /api/batch-diff``             — side-by-side diff of two
+  batches (used by the Batch Diff page). Lazy-imports
+  :func:`cyclone.batch_diff.diff_batches_to_wire` to keep the
+  module's import surface small until the endpoint is actually
+  hit.
+- ``POST /api/reconciliation/match``   — manually pair a Claim with
+  a Remittance (``store.manual_match``). Surfaces ``AlreadyMatchedError`` /
+  ``InvalidStateError`` as 409, and ``LookupError`` from the store
+  as 404 (``claim_or_remit_not_found``).
+- ``POST /api/reconciliation/unmatch`` — unpair a Claim (``store.manual_unmatch``).
+  Surfaces ``NotMatchedError`` as 409.
+
+All four are read-or-manual-override surfaces used by the
+Reconciliation page (the page that pairs Claims with Remittances
+when neither side has an automatic match key).
+
+SP36 Task 8: this block moved here from ``api.py:2450`` (the
+4 routes interleaved with a side-by-side batch-diff divider).
+"""
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+
+from cyclone.auth.deps import matrix_gate
+from cyclone.store import (
+    AlreadyMatchedError,
+    InvalidStateError,
+    store,
+)
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/reconciliation/unmatched")
+def get_reconciliation_unmatched() -> dict:
+    """Return unmatched Claims (left) and unmatched Remittances (right).
+
+    Powers the reconciliation review surface: every Claim with no
+    paired Remittance appears on the left, every Remittance with no
+    paired Claim appears on the right. The two lists are always present
+    (empty list, never absent) so the UI can index unconditionally.
+    """
+    return store.list_unmatched(kind="both")
+
+
+# --------------------------------------------------------------------------- #
+# Side-by-side diff between two batches (SP3 P4 / T18)
+# --------------------------------------------------------------------------- #
+
+
+@router.get("/api/batch-diff")
+def get_batch_diff(
+    a: str | None = Query(None),
+    b: str | None = Query(None),
+) -> dict:
+    """Return a side-by-side diff of two batches identified by id.
+
+    Query params: ``a=<batch_id>``, ``b=<batch_id>`` (both required).
+
+    Response body (snake_case keys, see :mod:`cyclone.batch_diff` for the
+    projector shapes):
+      - ``a`` / ``b`` — small metadata blocks (id, kind, parsedAt,
+        inputFilename, claimCount)
+      - ``added`` — claims present in B but not A
+      - ``removed`` — claims present in A but not B
+      - ``changed`` — claims present in both, with field deltas
+      - ``summary`` — precomputed counts
+
+    Errors:
+      - 400 — missing ``a`` or ``b``
+      - 404 — either batch id is unknown
+
+    Pure read endpoint — never mutates the store. Both 837P and 835
+    batches are accepted (mixed-kind diffs are valid: comparing the
+    submitted claims against the matching remittances).
+    """
+    if not a or not b:
+        raise HTTPException(
+            status_code=400,
+            detail={"error": "Missing param", "detail": "Both ?a=<batch_id> and ?b=<batch_id> are required."},
+        )
+    try:
+        a_rec, b_rec = store.load_two_for_diff(a, b)
+    except LookupError as exc:
+        raise HTTPException(
+            status_code=404,
+            detail={"error": "Not found", "detail": str(exc)},
+        )
+
+    # Lazy import — keeps the module's import surface small until the
+    # endpoint is actually hit. Mirrors the same pattern used by other
+    # endpoint-local helpers (e.g. reconciler).
+    from cyclone.batch_diff import diff_batches_to_wire
+
+    return diff_batches_to_wire(a_rec, b_rec)
+
+
+@router.post("/api/reconciliation/match")
+def post_reconciliation_match(body: dict) -> dict:
+    """Manually pair a Claim with a Remittance (operator override).
+
+    Body: ``{"claim_id": ..., "remit_id": ...}``. Returns
+    ``{"claim": <ui>, "match": <ui>}`` on success. Errors:
+      - 400: missing ``claim_id`` or ``remit_id``
+      - 404: claim or remittance not found
+      - 409: claim already matched, or apply_* returned a noop
+        (claim in terminal state) — detail echoes ``current_state``
+        and ``activity_kind`` so the UI can render a precise message.
+    """
+    claim_id = body.get("claim_id")
+    remit_id = body.get("remit_id")
+    if not claim_id or not remit_id:
+        raise HTTPException(
+            status_code=400,
+            detail="claim_id and remit_id required",
+        )
+    try:
+        return store.manual_match(claim_id, remit_id)
+    except AlreadyMatchedError as e:
+        raise HTTPException(
+            status_code=409,
+            detail={"error": "already_matched", "message": str(e)},
+        )
+    except InvalidStateError as e:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "error": "invalid_state",
+                "current_state": e.current_state,
+                "activity_kind": e.activity_kind,
+            },
+        )
+    except LookupError:
+        # manual_match raises LookupError when the claim or remittance
+        # row is missing (we catch the parent class so any future
+        # KeyError subclasses in the store get the same treatment).
+        raise HTTPException(
+            status_code=404,
+            detail="claim_or_remit_not_found",
+        )
+
+
+@router.post("/api/reconciliation/unmatch")
+def post_reconciliation_unmatch(body: dict) -> dict:
+    """Remove the current match for a Claim; reset Claim to submitted.
+
+    Body: ``{"claim_id": ...}``. Returns
+    ``{"claim": <ui>, "deletedMatches": <count>}``. Errors:
+      - 400: missing ``claim_id``
+      - 404: claim not found
+      - 409: claim has no current match (NotMatchedError is mapped
+        by the store; we surface 409 to match the manual_match contract)
+    """
+    from cyclone.store import NotMatchedError
+    claim_id = body.get("claim_id")
+    if not claim_id:
+        raise HTTPException(
+            status_code=400,
+            detail="claim_id required",
+        )
+    try:
+        return store.manual_unmatch(claim_id)
+    except NotMatchedError as e:
+        raise HTTPException(
+            status_code=409,
+            detail={"error": "not_matched", "message": str(e)},
+        )
+    except LookupError:
+        raise HTTPException(
+            status_code=404,
+            detail="claim_not_found",
+        )

backend/src/cyclone/api_routers/remittances.py

@@ -0,0 +1,169 @@
+"""``/api/remittances`` and ``/api/remittances/stream`` — read views over the Remittance population.
+
+Four endpoints, all gated by ``matrix_gate``:
+
+- ``GET /api/remittances``                — paginated list with filter+sort,
+  plus an NDJSON variant when the caller sends ``Accept: application/x-ndjson``.
+- ``GET /api/remittances/summary``        — server-aggregated KPI tiles
+  (``count``, ``total_paid``, ``total_adjustments``) over the full
+  filtered population — never a page-limited sample (SP27 fix).
+- ``GET /api/remittances/stream``         — NDJSON live-tail: snapshot
+  of currently-known rows, then ``remittance_written`` events as
+  they hit the store. Subscribed to by the Remittances page.
+- ``GET /api/remittances/{remittance_id}`` — one remittance with its
+  labeled CAS ``adjustments`` array. 404 on missing id (never 500).
+
+``/api/remittances/stream`` is registered before
+``/api/remittances/{remittance_id}`` so the literal ``stream`` path
+segment is not captured as a remittance id.
+
+SP36 Task 9: this block moved here from ``api.py:2448`` (the 4
+``/api/remittances*`` routes, with the streaming route's
+``NOTE: registered before…`` comment preserved verbatim).
+"""
+from __future__ import annotations
+
+from typing import Any, AsyncIterator
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import StreamingResponse
+
+from cyclone.api_helpers import (
+    ndjson_line as _ndjson_line,
+    ndjson_stream_list as _ndjson_stream_list,
+    tail_events as _tail_events,
+    wants_ndjson as _wants_ndjson,
+)
+from cyclone.auth.deps import matrix_gate
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+router = APIRouter(dependencies=[Depends(matrix_gate)])
+
+
+@router.get("/api/remittances")
+def list_remittances(
+    request: Request,
+    batch_id: str | None = Query(None),
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+    offset: int = Query(0, ge=0),
+) -> Any:
+    common = dict(
+        batch_id=batch_id,
+        payer=payer,
+        claim_id=claim_id,
+        date_from=date_from,
+        date_to=date_to,
+    )
+    items = list(store.iter_remittances(
+        sort=sort, order=order, limit=limit, offset=offset, **common,
+    ))
+    # SP27 Task 13b: count the full population, not a 100-row sample.
+    # See the matching note in list_claims — same silent-failure pattern.
+    total = store.count_remittances(**common)
+    returned = len(items)
+    has_more = total > offset + returned
+    if _wants_ndjson(request):
+        return StreamingResponse(
+            _ndjson_stream_list(items, total, returned, has_more),
+            media_type="application/x-ndjson",
+        )
+    return {
+        "items": items,
+        "total": total,
+        "returned": returned,
+        "has_more": has_more,
+    }
+
+
+@router.get("/api/remittances/summary")
+def remittances_summary(
+    batch_id: str | None = Query(None),
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+) -> dict:
+    """Server-aggregated KPI tiles for the Remittances page.
+
+    Returns ``{count, total_paid, total_adjustments}`` over the
+    full filtered remittance population — NOT a page-limited
+    sample. The Remittances page consumes this for its "Total paid"
+    and "Adjustments" tiles so they can't silently understate the
+    true DB population the way a page-local ``items.reduce(...)``
+    would. Mirrors the silent-incompleteness fix that
+    ``/api/dashboard/kpis`` (commit ``59c3275``) and
+    ``/api/remittances`` (commit ``d81b6ed``) made for their tiles.
+
+    Same filter parameters as ``/api/remittances``. Always returns
+    a populated dict (``{"count": 0, "total_paid": 0,
+    "total_adjustments": 0}`` when no rows match) so the frontend
+    can render the tiles directly without a loading-vs-empty
+    branch.
+    """
+    return store.summarize_remittances(
+        batch_id=batch_id, payer=payer, claim_id=claim_id,
+        date_from=date_from, date_to=date_to,
+    )
+
+
+@router.get("/api/remittances/stream")
+async def remittances_stream(
+    request: Request,
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> StreamingResponse:
+    """Stream Remittances as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``remittance_written``. Default sort is
+    ``-received_date`` (newest-first), matching the list endpoint's
+    most common sort.
+
+    NOTE: registered before ``/api/remittances/{remittance_id}`` so
+    the literal ``stream`` path segment doesn't get matched as a
+    remittance id.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        rows = store.iter_remittances(
+            payer=payer, claim_id=claim_id,
+            date_from=date_from, date_to=date_to,
+            sort=sort or "-received_date", order=order, limit=limit,
+        )
+        for row in rows:
+            yield _ndjson_line({"type": "item", "data": row})
+        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
+
+        async for chunk in _tail_events(request, bus, ["remittance_written"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
+@router.get("/api/remittances/{remittance_id}")
+def get_remittance(remittance_id: str) -> dict:
+    """Return one remittance with its labeled CAS ``adjustments`` array.
+
+    Path param is ``remittance_id`` (not ``id``) to avoid shadowing
+    FastAPI's internal ``id`` name and to keep OpenAPI docs self-
+    describing. Returns 404 when the remittance is missing — never 500.
+    """
+    body = store.get_remittance(remittance_id)
+    if body is None:
+        raise HTTPException(
+            status_code=404,
+            detail={"error": "Not found", "detail": f"Remittance {remittance_id} not found"},
+        )
+    return body

backend/src/cyclone/handlers/handle_835.py

@@ -66,10 +66,12 @@ def handle(
     here was dead code anyway (the 835 event name is
     ``remittance_written``, not ``ack_received``).
     """
-    # TODO(sp27-pre-t5): move PAYER_FACTORIES_835 out of api.py into
+    # SP36 Task 16: PAYER_FACTORIES_835 moved from ``cyclone.api``
-    # ``cyclone.payers`` to remove the lazy cyclic import below. The
+    # to ``cyclone.api_routers._shared`` (cross-router helper). Import
-    # import works today because api.py also imports scheduler lazily.
+    # from the new home. The lazy import is still needed to avoid a
-    from cyclone.api import PAYER_FACTORIES_835
+    # circular import at registry load time (handle_835 is imported
+    # by the scheduler during the api lifespan).
+    from cyclone.api_routers._shared import PAYER_FACTORIES_835
 
     config = PAYER_FACTORIES_835["co_medicaid_835"]()
     try: