docs: SP16 Inbound MFT Scheduler section + project-layout updates

Documents the SP16 scheduler control surface (5 endpoints under
/api/admin/scheduler/*) and brings the Project layout tree in sync
with the latest api_routers/ split state:

- new 'Inbound MFT Scheduler (SP16)' section between SFTP Wire-Up
  and Batches/Reconciliation/Activity, covering start/stop/status/
  tick/processed-files with a forward-compat note that
  cyclone.scheduler + cyclone.db.ProcessedInboundFile are in flight
- api_routers/ tree now lists all 14 routers, including the four
  parse_* routers (parse_835/837/999/ta1) and admin.py carrying the
  scheduler endpoints
- api.py comment updated to 'app + middleware; mounts api_routers/
  sub-apps' (no inline route bodies remain)

README.md

@@ -418,6 +418,35 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
 the `paramiko` extras aren't installed so the test suite stays green on
 Linux dev boxes.
 
+## Inbound MFT Scheduler (SP16)
+
+The dzinesco SFTP server accepts inbound files (835 remits, 999/TA1
+acks, 271 eligibility responses, 277/277CA claim acknowledgments) that
+need to be pulled on a cadence, parsed, and persisted. SP16 ships the
+**operator-facing control surface** for the polling loop, even though
+the actual background scheduler is still in flight:
+
+| Method | Path                                          | Notes                                                            |
+| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
+| POST   | `/api/admin/scheduler/start`                  | Start the background MFT polling loop (idempotent).              |
+| POST   | `/api/admin/scheduler/stop`                   | Stop the background loop. Waits for the current tick to finish.  |
+| GET    | `/api/admin/scheduler/status`                 | Running state + last tick summary.                               |
+| POST   | `/api/admin/scheduler/tick`                   | Force a single poll cycle. Coalesces with an in-flight tick.     |
+| GET    | `/api/admin/scheduler/processed-files`        | Recently processed inbound files. `?status=ok|error|skipped|pending`, `?limit=` (1–500, default 50). |
+
+The endpoints read from a `cyclone.scheduler` module and a
+`cyclone.db.ProcessedInboundFile` ORM model that are being landed
+alongside SP16 — the admin router's import surface is in place, the
+implementation is in progress. The control surface is forward-compatible
+with the eventual scheduler: starting / stopping / ticking / listing
+processed files will all work as soon as the loop lands.
+
+The `paramiko` SFTP client from [SP13](#sftp-wire-up-paramiko) is the
+underlying transport; the scheduler reuses it to list + download files
+under the inbound SFTP path, then dispatches each downloaded file to
+the matching parser based on the `<FileType>` token in the filename
+(see [SP13 file-naming template](#sftp-wire-up-paramiko)).
+
 ## Batches, Reconciliation, and Activity
 
 These read/write endpoints are the core operator surface for browsing
@@ -474,20 +503,31 @@ backup API).
 .
 ├── backend/
 │   ├── src/cyclone/
-│   │   ├── api.py             # FastAPI app + parse routes; mounts api_routers/* sub-apps
+│   │   ├── api.py             # FastAPI app + middleware; mounts api_routers/* sub-apps
 │   │   ├── api_helpers.py     # NDJSON / content-negotiation / live-tail helpers
 │   │   ├── api_routers/       # FastAPI APIRouters extracted from api.py
 │   │   │   ├── health.py      #   GET /api/health
 │   │   │   ├── acks.py        #   GET /api/acks, /api/acks/{id}
 │   │   │   ├── ta1_acks.py    #   GET /api/ta1-acks, /api/ta1-acks/{id}
 │   │   │   ├── activity.py    #   GET /api/activity, /api/activity/stream
-│   │   │   ├── batches.py     #   GET /api/batches, /api/batches/{id}, /api/batch-diff
+│   │   │   ├── batches.py     #   GET /api/batches, /api/batches/{id}
 │   │   │   ├── claims.py      #   GET /api/claims, /api/claims/stream, /api/claims/{id},
 │   │   │   │                  #   /api/claims/{id}/serialize-837, /api/claims/{id}/line-reconciliation
 │   │   │   ├── remittances.py #   GET /api/remittances, /api/remittances/stream, /api/remittances/{id}
 │   │   │   ├── providers.py   #   GET /api/providers
+│   │   │   ├── inbox.py       #   GET /api/inbox/lanes, POST /api/inbox/* (match/dismiss/acknowledge/resubmit),
+│   │   │   │                  #   GET /api/inbox/export.csv
+│   │   │   ├── reconciliation.py #   GET /api/reconciliation/unmatched, /api/batch-diff,
+│   │   │   │                     #   POST /api/reconciliation/match, /api/reconciliation/unmatch
+│   │   │   ├── eligibility.py #   POST /api/eligibility/request, /api/eligibility/parse-271
 │   │   │   ├── clearhouse.py  #   GET /api/clearhouse, POST /api/clearhouse/submit
-│   │   │   └── config.py      #   /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
+│   │   │   ├── config.py      #   /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
+│   │   │   ├── parse_835.py   #   POST /api/parse-835
+│   │   │   ├── parse_837.py   #   POST /api/parse-837
+│   │   │   ├── parse_999.py   #   POST /api/parse-999
+│   │   │   ├── parse_ta1.py   #   POST /api/parse-ta1
+│   │   │   └── admin.py       #   /api/admin/audit-log[/verify], /api/admin/db/rotate-key,
+│   │   │                      #   /api/admin/scheduler/* (SP16)
 │   │   ├── pubsub.py          # in-process EventBus (drop-oldest, per-kind fan-out)
 │   │   ├── store.py           # CycloneStore, mappers, publish-on-write
 │   │   ├── db.py              # SQLAlchemy engine, session factory, ORM models

backend/src/cyclone/api.py

@@ -15,8 +15,6 @@ plus GET/POST with any header.
 
 from __future__ import annotations
 
-import csv
-import io
 import json
 import logging
 import uuid
@@ -26,23 +24,16 @@ from typing import Any, AsyncIterator
 
 from fastapi import FastAPI, File, HTTPException, Query, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, Response, StreamingResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import ValidationError
 
 from cyclone import __version__, db
 from cyclone.db import Claim, ClaimState, Remittance
 from cyclone.inbox_state import apply_999_rejections
 from cyclone.inbox_state_277ca import apply_277ca_rejections
-from cyclone.audit_log import AuditEvent, append_event, verify_chain
+from cyclone.audit_log import AuditEvent, append_event
 from cyclone.parsers.exceptions import CycloneParseError
-from cyclone.parsers.models import BatchSummary, ClaimOutput, Envelope, ParseResult
+from cyclone.parsers.models import BatchSummary, Envelope
-from cyclone.parsers.models_270 import (
-    EligibilityBenefitInquiry,
-    InformationReceiver270,
-    InformationSource270,
-    ParseResult270,
-    Subscriber270,
-)
 from cyclone.parsers.models_271 import ParseResult271
 from cyclone.parsers.models_835 import ParseResult835
 from cyclone.parsers.payer import PayerConfig, PayerConfig835
@@ -53,9 +44,8 @@ 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.serialize_270 import serialize_270
 from cyclone.parsers.serialize_999 import serialize_999
-from cyclone.parsers.serialize_837 import SerializeError as SerializeError837, serialize_837, serialize_837_for_resubmit
+
 from cyclone.parsers.batch_ack_builder import build_ack_for_batch
 from cyclone.parsers.validator_835 import validate as validate_835
 from cyclone.pubsub import EventBus
@@ -144,12 +134,16 @@ app.add_middleware(
 from cyclone.api_routers import (  # noqa: E402
     acks,
     activity,
+    admin,
     batches,
     claims,
     clearhouse,
     config,
+    eligibility,
     health,
+    inbox,
     providers,
+    reconciliation,
     remittances,
     ta1_acks,
 )
@@ -164,6 +158,10 @@ app.include_router(activity.router)
 app.include_router(batches.router)
 app.include_router(claims.router)
 app.include_router(remittances.router)
+app.include_router(inbox.router)
+app.include_router(reconciliation.router)
+app.include_router(eligibility.router)
+app.include_router(admin.router)
 
 # Backwards-compat re-exports: test_api_stream_live.py imports these
 # names directly from ``cyclone.api`` to invoke the endpoint coroutines
@@ -173,6 +171,24 @@ activity_stream = activity.activity_stream_endpoint
 claims_stream = claims.claims_stream_endpoint
 remittances_stream = remittances.remittances_stream_endpoint
 
+# Backwards-compat lock re-export: test_api_rotate_key.py does
+# ``api_mod._db_rotate_lock.acquire()`` to simulate a concurrent
+# rotation request. The lock itself lives in the admin router (it
+# guards that router's rotate-key endpoint) but we expose the same
+# object here so the test holds the same lock the endpoint does.
+_db_rotate_lock = admin._db_rotate_lock
+
+# Backwards-compat module re-exports: test_api_rotate_key.py does
+# ``monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", ...)``
+# and ``monkeypatch.setattr("cyclone.api._secrets.set_secret", ...)``
+# to stub crypto/keychain calls. Re-exporting here keeps the
+# ``cyclone.api._db_crypto`` / ``cyclone.api._secrets`` paths alive
+# so the monkeypatch targets the same module objects the admin
+# endpoint actually uses (both are the same ``cyclone.db_crypto`` /
+# ``cyclone.secrets`` module instances).
+_db_crypto = admin.db_crypto
+_secrets = admin.secrets
+
 
 def _resolve_payer(name: str) -> PayerConfig:
     if name not in PAYER_FACTORIES:
@@ -901,416 +917,10 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str:
 # --------------------------------------------------------------------------- #
 
 
-@app.get("/api/inbox/lanes")
+# --- /api/inbox/* lives in cyclone.api_routers.inbox ---
-def inbox_lanes():
-    """Return all Inbox lanes in one call."""
-    dismissed_pairs = getattr(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,
-    }
 
+# --- /api/reconciliation/* + /api/batch-diff live in cyclone.api_routers.reconciliation ---
 
-@app.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}
-
-
-@app.post("/api/inbox/candidates/dismiss")
-def inbox_dismiss_candidates(body: dict):
-    """Add candidate pairs to the session-scoped dismissed set."""
-    pairs = body.get("pairs") or []
-    if not hasattr(app.state, "dismissed_pairs"):
-        app.state.dismissed_pairs = set()
-    for p in pairs:
-        cid = p.get("claim_id")
-        rid = p.get("remit_id")
-        if cid and rid:
-            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".
-# --------------------------------------------------------------------------- #
-@app.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,
-    }
-
-
-@app.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,
-    )
-
-
-@app.get("/api/inbox/export.csv")
-def inbox_export_csv(lane: str):
-    """Stream a CSV for a single lane."""
-    if lane not in {"rejected", "candidates", "unmatched", "done_today"}:
-        raise HTTPException(400, f"unknown lane: {lane}")
-    dismissed_pairs = getattr(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"'},
-    )
-
-
-# --- /api/batches lives in cyclone.api_routers.batches ---
-
-
-# --- /api/claims lives in cyclone.api_routers.claims ---
-
-@app.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)
-# --------------------------------------------------------------------------- #
-
-
-@app.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)
-
-
-@app.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",
-        )
-
-
-@app.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",
-        )
 
 
 # --- /api/remittances lives in cyclone.api_routers.remittances ---
@@ -1330,424 +940,10 @@ def post_reconciliation_unmatch(body: dict) -> dict:
 # --------------------------------------------------------------------------- #
 
 
-def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
+# --- /api/eligibility/* lives in cyclone.api_routers.eligibility ---
-    """Build a :class:`ParseResult270` from a request body dict.
 
-    The body shape is the minimum surface needed to build a valid 270
+# --- /api/admin/* lives in cyclone.api_routers.admin ---
-    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).
-    from datetime import date as _date
-
-    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
-
-
-@app.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()),
-    }
-
-
-@app.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()),
-    }
-
-
-
-
-
-# --------------------------------------------------------------------------- #
-# SP11: tamper-evident audit log (admin)
-# --------------------------------------------------------------------------- #
-
-
-@app.get("/api/admin/audit-log")
-def list_audit_log_endpoint(
-    entity_type: str | None = Query(default=None),
-    entity_id: str | None = Query(default=None),
-    event_type: str | None = Query(default=None),
-    limit: int = Query(default=100, ge=1, le=1000),
-) -> Any:
-    """List audit-log rows, newest first, with optional filters.
-
-    Filters match the (entity_type, entity_id) pair (typical use:
-    "show me everything that happened to claim C-123") or a single
-    event_type (typical use: "show me all clearhouse.submitted
-    events today").
-    """
-    with db.SessionLocal()() as s:
-        q = s.query(db.AuditLog)
-        if entity_type:
-            q = q.filter(db.AuditLog.entity_type == entity_type)
-        if entity_id:
-            q = q.filter(db.AuditLog.entity_id == entity_id)
-        if event_type:
-            q = q.filter(db.AuditLog.event_type == event_type)
-        rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
-        return {
-            "total": len(rows),
-            "items": [
-                {
-                    "id": r.id,
-                    "event_type": r.event_type,
-                    "entity_type": r.entity_type,
-                    "entity_id": r.entity_id,
-                    "actor": r.actor,
-                    "payload": json.loads(r.payload_json) if r.payload_json else None,
-                    "created_at": r.created_at.isoformat() if r.created_at else None,
-                    "prev_hash": r.prev_hash,
-                    "hash": r.hash,
-                }
-                for r in rows
-            ],
-        }
-
-
-@app.get("/api/admin/audit-log/verify")
-def verify_audit_log_endpoint() -> Any:
-    """Walk the audit-log chain and verify every row's hash.
-
-    Returns ``{"ok": true, "checked": N}`` for a clean chain, or
-    ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
-    for a broken chain. This is the operator's "did anyone tamper?"
-    endpoint; run it on demand or via a nightly cron job.
-    """
-    with db.SessionLocal()() as s:
-        result = verify_chain(s)
-    return {
-        "ok": result.ok,
-        "checked": result.checked,
-        "first_bad_id": result.first_bad_id,
-        "reason": result.reason,
-    }
-
-
-# ---------------------------------------------------------------------------
-# SP15: SQLCipher key rotation
-#
-# Re-encrypts the DB in place with a fresh key, then updates the
-# Keychain so subsequent connections open with the new key. This is
-# a 1-time operation per rotation; for routine read/write the rest
-# of the API is unchanged.
-#
-# Concurrency: the rotation holds a module-level lock so two
-# concurrent requests can't race and end up with mismatched Keychain
-# + DB. The lock is a simple threading.Lock; a process restart
-# resets it (intentional — the operator's next start-up opens with
-# whatever key is in the Keychain).
-# ---------------------------------------------------------------------------
-import threading as _threading
-from cyclone import db_crypto as _db_crypto
-from cyclone import secrets as _secrets
-
-_db_rotate_lock = _threading.Lock()
-
-
-@app.post("/api/admin/db/rotate-key")
-def rotate_db_key_endpoint(body: dict | None = None) -> Any:
-    """Generate a fresh DB key, re-encrypt the DB, update the Keychain.
-
-    Request body (optional):
-        actor: who initiated the rotation. Defaults to "operator".
-        reason: human-readable reason. Written to the audit log.
-
-    Returns:
-        ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
-        on success. On failure (DB not encrypted, rekey failed,
-        Keychain update failed) returns the same shape with
-        ``ok=false`` and a ``reason``. HTTP 503 is returned if the
-        rekey fails or encryption is not enabled.
-
-    The Keychain write happens *after* the rekey succeeds. If the
-    Keychain write fails, the DB has the new key but the Keychain
-    still has the old one — the endpoint returns 503 with a
-    "keychain update failed" reason and the operator must restore
-    the old key manually (``cyclone db restore-key <old_key>``) to
-    avoid being locked out.
-    """
-    body = body or {}
-    actor = body.get("actor") or "operator"
-    reason = body.get("reason") or ""
-
-    if not _db_crypto.is_encryption_enabled():
-        raise HTTPException(
-            status_code=400,
-            detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
-        )
-
-    # Acquire the lock; non-blocking so a stuck rotation doesn't
-    # silently hold up other requests.
-    if not _db_rotate_lock.acquire(blocking=False):
-        raise HTTPException(
-            status_code=409,
-            detail="another key rotation is in progress",
-        )
-    try:
-        url = db._resolve_url()
-        old_key = _db_crypto.get_db_key()
-        if not old_key:
-            raise HTTPException(
-                status_code=400,
-                detail="no DB key in Keychain; cannot rotate",
-            )
-
-        new_key = _db_crypto.generate_db_key()
-        result = _db_crypto.rotate_db_key(
-            url=url, old_key=old_key, new_key=new_key,
-        )
-        if not result.ok:
-            # Rekey failed. The DB still has the old key. The
-            # Keychain is unchanged. Caller should NOT retry with
-            # the same new key (it's lost); generate a fresh one.
-            log.error("SQLCipher rotate failed: %s", result.reason)
-            raise HTTPException(
-                status_code=503,
-                detail={
-                    "ok": False,
-                    "old_fingerprint": result.old_fingerprint,
-                    "new_fingerprint": result.new_fingerprint,
-                    "rotated_at": result.rotated_at,
-                    "reason": result.reason,
-                },
-            )
-
-        # Rekey succeeded. Now update the Keychain. If this fails
-        # the DB is locked behind the new key — operator must
-        # restore the old key manually.
-        if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key):
-            log.error("Keychain update failed after successful rekey!")
-            raise HTTPException(
-                status_code=503,
-                detail={
-                    "ok": False,
-                    "old_fingerprint": result.old_fingerprint,
-                    "new_fingerprint": result.new_fingerprint,
-                    "rotated_at": result.rotated_at,
-                    "reason": (
-                        "rekey succeeded but Keychain update failed — "
-                        "the DB is now encrypted with the new key but "
-                        "the Keychain still has the old one. "
-                        "Restore the old key to the Keychain to recover."
-                    ),
-                },
-            )
-
-        # Store the old key in the "previous" account for a grace
-        # period so the operator can roll back if they discover the
-        # new key is broken (e.g. the Keychain entry got truncated).
-        _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
-
-        # Rebuild the engine so subsequent connections use the new
-        # key. dispose_engine() closes every pooled connection that
-        # was using the old key; init_db() opens new ones with the
-        # new key from the (now-updated) Keychain.
-        db.reinit_engine()
-
-        # Audit log the rotation. We do this after the engine is
-        # rebuilt so the audit event is written with the new key —
-        # proving that the new key works for new writes.
-        try:
-            from cyclone.audit_log import append_event, AuditEvent
-            with db.SessionLocal()() as s:
-                append_event(s, AuditEvent(
-                    event_type="db.key_rotated",
-                    entity_type="database",
-                    entity_id="cyclone.db",
-                    actor=actor,
-                    payload={
-                        "old_fingerprint": result.old_fingerprint,
-                        "new_fingerprint": result.new_fingerprint,
-                        "table_count": result.table_count,
-                        "reason": reason,
-                    },
-                ))
-                s.commit()
-        except Exception as exc:  # noqa: BLE001
-            # Audit append is best-effort; rotation already succeeded.
-            log.warning("could not write audit event for rotation: %s", exc)
-
-        return {
-            "ok": True,
-            "old_fingerprint": result.old_fingerprint,
-            "new_fingerprint": result.new_fingerprint,
-            "rotated_at": result.rotated_at,
-            "table_count": result.table_count,
-        }
-    finally:
-        _db_rotate_lock.release()
-
-
-
-
-
-@app.post("/api/admin/reload-config")
-def reload_config():
-    """Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
-    from cyclone import payers as payer_loader
-    try:
-        configs = payer_loader.load_payer_configs()
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    return {"ok": True, "loaded": len(configs), "errors": []}
 
 
 __all__ = ["app"]

backend/src/cyclone/api_routers/admin.py

@@ -0,0 +1,332 @@
+"""``/api/admin/*`` — operator-facing admin surface (SP11 + SP15).
+
+Four endpoints:
+
+* ``GET /api/admin/audit-log`` — list audit-log rows with optional
+  ``entity_type`` / ``entity_id`` / ``event_type`` filters (SP11).
+* ``GET /api/admin/audit-log/verify`` — walk the audit-log hash chain
+  and verify every row (SP11).
+* ``POST /api/admin/db/rotate-key`` — SP15 SQLCipher key rotation.
+  Re-encrypts the DB in place with a fresh key, updates the Keychain.
+  Protected by a module-level ``threading.Lock`` so two concurrent
+  rotations can't race (a single process restart resets the lock —
+  intentional, the next start-up opens with whatever key is in the
+  Keychain).
+* ``POST /api/admin/reload-config`` — re-read ``config/payers.yaml``
+  and revalidate; returns the number of configs loaded.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import threading
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query
+
+from cyclone import db, db_crypto, secrets
+from cyclone.audit_log import AuditEvent, append_event, verify_chain
+
+log = logging.getLogger(__name__)
+
+router = APIRouter()
+
+# Module-level lock: protects against concurrent key rotations within
+# the same process. A process restart resets it (intentional — the
+# operator's next start-up opens with whatever key is in the Keychain).
+_db_rotate_lock = threading.Lock()
+
+
+@router.get("/api/admin/audit-log")
+def list_audit_log_endpoint(
+    entity_type: str | None = Query(default=None),
+    entity_id: str | None = Query(default=None),
+    event_type: str | None = Query(default=None),
+    limit: int = Query(default=100, ge=1, le=1000),
+) -> Any:
+    """List audit-log rows, newest first, with optional filters.
+
+    Filters match the (entity_type, entity_id) pair (typical use:
+    "show me everything that happened to claim C-123") or a single
+    event_type (typical use: "show me all clearhouse.submitted
+    events today").
+    """
+    with db.SessionLocal()() as s:
+        q = s.query(db.AuditLog)
+        if entity_type:
+            q = q.filter(db.AuditLog.entity_type == entity_type)
+        if entity_id:
+            q = q.filter(db.AuditLog.entity_id == entity_id)
+        if event_type:
+            q = q.filter(db.AuditLog.event_type == event_type)
+        rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
+        return {
+            "total": len(rows),
+            "items": [
+                {
+                    "id": r.id,
+                    "event_type": r.event_type,
+                    "entity_type": r.entity_type,
+                    "entity_id": r.entity_id,
+                    "actor": r.actor,
+                    "payload": json.loads(r.payload_json) if r.payload_json else None,
+                    "created_at": r.created_at.isoformat() if r.created_at else None,
+                    "prev_hash": r.prev_hash,
+                    "hash": r.hash,
+                }
+                for r in rows
+            ],
+        }
+
+
+@router.get("/api/admin/audit-log/verify")
+def verify_audit_log_endpoint() -> Any:
+    """Walk the audit-log chain and verify every row's hash.
+
+    Returns ``{"ok": true, "checked": N}`` for a clean chain, or
+    ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
+    for a broken chain. This is the operator's "did anyone tamper?"
+    endpoint; run it on demand or via a nightly cron job.
+    """
+    with db.SessionLocal()() as s:
+        result = verify_chain(s)
+    return {
+        "ok": result.ok,
+        "checked": result.checked,
+        "first_bad_id": result.first_bad_id,
+        "reason": result.reason,
+    }
+
+
+@router.post("/api/admin/db/rotate-key")
+def rotate_db_key_endpoint(body: dict | None = None) -> Any:
+    """Generate a fresh DB key, re-encrypt the DB, update the Keychain.
+
+    Request body (optional):
+        actor: who initiated the rotation. Defaults to "operator".
+        reason: human-readable reason. Written to the audit log.
+
+    Returns:
+        ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
+        on success. On failure (DB not encrypted, rekey failed,
+        Keychain update failed) returns the same shape with
+        ``ok=false`` and a ``reason``. HTTP 503 is returned if the
+        rekey fails or encryption is not enabled.
+
+    The Keychain write happens *after* the rekey succeeds. If the
+    Keychain write fails, the DB has the new key but the Keychain
+    still has the old one — the endpoint returns 503 with a
+    "keychain update failed" reason and the operator must restore
+    the old key manually (``cyclone db restore-key <old_key>``) to
+    avoid being locked out.
+    """
+    body = body or {}
+    actor = body.get("actor") or "operator"
+    reason = body.get("reason") or ""
+
+    if not db_crypto.is_encryption_enabled():
+        raise HTTPException(
+            status_code=400,
+            detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
+        )
+
+    # Acquire the lock; non-blocking so a stuck rotation doesn't
+    # silently hold up other requests.
+    if not _db_rotate_lock.acquire(blocking=False):
+        raise HTTPException(
+            status_code=409,
+            detail="another key rotation is in progress",
+        )
+    try:
+        url = db._resolve_url()
+        old_key = db_crypto.get_db_key()
+        if not old_key:
+            raise HTTPException(
+                status_code=400,
+                detail="no DB key in Keychain; cannot rotate",
+            )
+
+        new_key = db_crypto.generate_db_key()
+        result = db_crypto.rotate_db_key(
+            url=url, old_key=old_key, new_key=new_key,
+        )
+        if not result.ok:
+            # Rekey failed. The DB still has the old key. The
+            # Keychain is unchanged. Caller should NOT retry with
+            # the same new key (it's lost); generate a fresh one.
+            log.error("SQLCipher rotate failed: %s", result.reason)
+            raise HTTPException(
+                status_code=503,
+                detail={
+                    "ok": False,
+                    "old_fingerprint": result.old_fingerprint,
+                    "new_fingerprint": result.new_fingerprint,
+                    "rotated_at": result.rotated_at,
+                    "reason": result.reason,
+                },
+            )
+
+        # Rekey succeeded. Now update the Keychain. If this fails
+        # the DB is locked behind the new key — operator must
+        # restore the old key manually.
+        if not secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT, new_key):
+            log.error("Keychain update failed after successful rekey!")
+            raise HTTPException(
+                status_code=503,
+                detail={
+                    "ok": False,
+                    "old_fingerprint": result.old_fingerprint,
+                    "new_fingerprint": result.new_fingerprint,
+                    "rotated_at": result.rotated_at,
+                    "reason": (
+                        "rekey succeeded but Keychain update failed — "
+                        "the DB is now encrypted with the new key but "
+                        "the Keychain still has the old one. "
+                        "Restore the old key to the Keychain to recover."
+                    ),
+                },
+            )
+
+        # Store the old key in the "previous" account for a grace
+        # period so the operator can roll back if they discover the
+        # new key is broken (e.g. the Keychain entry got truncated).
+        secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
+
+        # Rebuild the engine so subsequent connections use the new
+        # key. dispose_engine() closes every pooled connection that
+        # was using the old key; init_db() opens new ones with the
+        # new key from the (now-updated) Keychain.
+        db.reinit_engine()
+
+        # Audit log the rotation. We do this after the engine is
+        # rebuilt so the audit event is written with the new key —
+        # proving that the new key works for new writes.
+        try:
+            with db.SessionLocal()() as s:
+                append_event(s, AuditEvent(
+                    event_type="db.key_rotated",
+                    entity_type="database",
+                    entity_id="cyclone.db",
+                    actor=actor,
+                    payload={
+                        "old_fingerprint": result.old_fingerprint,
+                        "new_fingerprint": result.new_fingerprint,
+                        "table_count": result.table_count,
+                        "reason": reason,
+                    },
+                ))
+                s.commit()
+        except Exception as exc:  # noqa: BLE001
+            # Audit append is best-effort; rotation already succeeded.
+            log.warning("could not write audit event for rotation: %s", exc)
+
+        return {
+            "ok": True,
+            "old_fingerprint": result.old_fingerprint,
+            "new_fingerprint": result.new_fingerprint,
+            "rotated_at": result.rotated_at,
+            "table_count": result.table_count,
+        }
+    finally:
+        _db_rotate_lock.release()
+
+
+@router.post("/api/admin/reload-config")
+def reload_config():
+    """Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
+    from cyclone import payers as payer_loader
+    try:
+        configs = payer_loader.load_payer_configs()
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    return {"ok": True, "loaded": len(configs), "errors": []}
+
+
+# ---------------------------------------------------------------------------
+# SP16: inbound MFT scheduler
+# ---------------------------------------------------------------------------
+
+
+@router.post("/api/admin/scheduler/start")
+async def scheduler_start():
+    """Start the background MFT polling loop (idempotent)."""
+    from cyclone import scheduler as sched_mod
+    sched = sched_mod.get_scheduler()
+    await sched.start()
+    return {"ok": True, "status": sched.status().as_dict()}
+
+
+@router.post("/api/admin/scheduler/stop")
+async def scheduler_stop():
+    """Stop the background MFT polling loop. Waits for the current tick."""
+    from cyclone import scheduler as sched_mod
+    sched = sched_mod.get_scheduler()
+    await sched.stop()
+    return {"ok": True, "status": sched.status().as_dict()}
+
+
+@router.get("/api/admin/scheduler/status")
+def scheduler_status():
+    """Return the scheduler's running state + last tick summary."""
+    from cyclone import scheduler as sched_mod
+    sched = sched_mod.get_scheduler()
+    return sched.status().as_dict()
+
+
+@router.post("/api/admin/scheduler/tick")
+async def scheduler_tick():
+    """Force a single poll cycle. Returns the TickResult.
+
+    Useful for the operator's "run now" button or for testing. If a
+    scheduled tick is already in flight, this call coalesces onto it
+    (waits for it to finish, returns its result).
+    """
+    from cyclone import scheduler as sched_mod
+    sched = sched_mod.get_scheduler()
+    result = await sched.tick()
+    return {"ok": True, "tick": result.as_dict()}
+
+
+@router.get("/api/admin/scheduler/processed-files")
+def scheduler_processed_files(
+    limit: int = 50,
+    status: str | None = None,
+):
+    """List recently-processed inbound files (newest first).
+
+    The operator uses this to confirm the scheduler is keeping up
+    with MFT traffic. ``status`` filters to one of: ok, error,
+    skipped, pending.
+    """
+    from sqlalchemy import desc
+    from cyclone import db
+    from cyclone.db import ProcessedInboundFile
+    with db.SessionLocal()() as session:
+        q = session.query(ProcessedInboundFile).order_by(
+            desc(ProcessedInboundFile.processed_at),
+        )
+        if status:
+            q = q.filter(ProcessedInboundFile.status == status)
+        q = q.limit(min(limit, 500))
+        rows = q.all()
+        return {
+            "count": len(rows),
+            "files": [
+                {
+                    "id": r.id,
+                    "name": r.name,
+                    "size": r.size,
+                    "file_type": r.file_type,
+                    "processed_at": r.processed_at.isoformat() if r.processed_at else None,
+                    "parser_used": r.parser_used,
+                    "claim_count": r.claim_count,
+                    "status": r.status,
+                    "error_message": (
+                        r.error_message[:200] + "..."
+                        if r.error_message and len(r.error_message) > 200
+                        else r.error_message
+                    ),
+                }
+                for r in rows
+            ],
+        }

backend/src/cyclone/api_routers/eligibility.py

@@ -0,0 +1,220 @@
+"""``/api/eligibility/*`` — 270/271 eligibility surface (SP3 P4 T23–T24).
+
+API-only — no DB persistence per spec section 3.4. The operator builds
+a 270 inquiry from a small JSON body, downloads the raw text to paste
+into a payer portal, then re-pastes the 271 response and parses it
+back into structured fields on demand.
+
+Two endpoints:
+* ``POST /api/eligibility/request`` — build a 270 inquiry from a JSON
+  body and return ``{raw_270_text, parsed}``.
+* ``POST /api/eligibility/parse-271`` — accept a raw 271 file upload
+  and return the parsed ``coverage_benefits`` / ``subscriber`` /
+  ``summary`` / envelope triples.
+
+The ``_validate_eligibility_request`` helper is private to this router
+— it builds the Pydantic ``ParseResult270`` and is only used by the
+``/request`` endpoint.
+"""
+from __future__ import annotations
+
+import json
+import logging
+from datetime import date as _date
+from typing import Any
+
+from fastapi import APIRouter, File, HTTPException, UploadFile
+from fastapi.responses import JSONResponse
+
+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()
+
+
+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"},
+        )
+
+    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,309 @@
+"""``/api/inbox/*`` — operator-facing inbox surface.
+
+Six endpoints powering the Inbox page (lanes, candidates, payer-rejected
+acknowledge, rejected resubmit, CSV export). All endpoints run inside a
+``db.SessionLocal()`` scope; the lanes / dismiss endpoints also read
+``app.state.dismissed_pairs`` (a session-scoped set so dismissed items
+don't re-appear until process restart).
+
+Endpoints:
+* ``GET /api/inbox/lanes`` — return all four lanes in one call
+  (rejected / payer_rejected / candidates / unmatched / done_today).
+* ``POST /api/inbox/candidates/{remit_id}/match`` — manually link a
+  remit to a claim from the candidates lane.
+* ``POST /api/inbox/candidates/dismiss`` — add candidate pairs to the
+  dismissed set so they stop appearing in the candidates lane.
+* ``POST /api/inbox/payer-rejected/acknowledge`` — clear SP14
+  payer-rejected claims from the working surface without deleting the
+  rejection (idempotent: re-acknowledging is a noop).
+* ``POST /api/inbox/rejected/resubmit`` — bulk move REJECTED claims
+  back to SUBMITTED; with ``?download=true`` returns a ZIP of
+  regenerated 837s with unique interchange control numbers.
+* ``GET /api/inbox/export.csv?lane=…`` — stream a CSV for a single lane.
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+import zipfile
+from datetime import datetime, timezone
+
+from fastapi import APIRouter, HTTPException, Query, Request, Response
+from fastapi.responses import StreamingResponse
+
+from cyclone import db
+from cyclone.audit_log import AuditEvent, append_event
+from cyclone.db import Claim, ClaimState, Remittance
+from cyclone.parsers.models import ClaimOutput
+from cyclone.parsers.serialize_837 import (
+    SerializeError as SerializeError837,
+    serialize_837_for_resubmit,
+)
+
+router = APIRouter()
+
+
+@router.get("/api/inbox/lanes")
+def inbox_lanes(request: Request) -> dict:
+    """Return all Inbox lanes in one call.
+
+    Reads the session-scoped ``dismissed_pairs`` set so the
+    candidates lane excludes pairs the operator already dismissed in
+    this process lifetime.
+    """
+    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(request: Request, body: dict):
+    """Add candidate pairs to the session-scoped dismissed set."""
+    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)}
+
+
+@router.post("/api/inbox/payer-rejected/acknowledge")
+def inbox_acknowledge_payer_rejected(body: dict):
+    """Mark Payer-Rejected claims as acknowledged by the operator.
+
+    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".
+    """
+    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:
+        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. Best-effort:
+            # an audit-log failure must not block the operator's action.
+            try:
+                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
+                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(
+    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.
+    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(request: Request, lane: str):
+    """Stream a CSV for a single lane."""
+    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/reconciliation.py

@@ -0,0 +1,164 @@
+"""``/api/reconciliation/*`` + ``/api/batch-diff`` — operator review surface.
+
+Four endpoints powering the reconciliation page:
+
+* ``GET /api/reconciliation/unmatched`` — return unmatched Claims
+  (left) and unmatched Remittances (right).
+* ``POST /api/reconciliation/match`` — manually pair a Claim with a
+  Remittance (operator override).
+* ``POST /api/reconciliation/unmatch`` — remove the current match and
+  reset the Claim to submitted.
+* ``GET /api/batch-diff`` — side-by-side diff of two batches (SP3 P4 /
+  T18).
+
+The two POST endpoints delegate to ``store.manual_match`` /
+``store.manual_unmatch``; the router is responsible for translating the
+store's exception hierarchy (``AlreadyMatchedError``, ``InvalidStateError``,
+``NotMatchedError``, ``LookupError``) into the HTTP error contract.
+"""
+from __future__ import annotations
+
+from fastapi import APIRouter, HTTPException, Query
+
+from cyclone.store import (
+    AlreadyMatchedError,
+    InvalidStateError,
+    store,
+)
+
+router = APIRouter()
+
+
+@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")
+
+
+@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.
+    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",
+        )