docs: surface SP14 (Payer-Rejected UI + acknowledge) + SP15 (key rotation)

The README's most recent merge was SP13. Since then two more sub-projects landed in main: - SP14 (5c9365e + 8a65baa) — 5-lane Inbox UI with the Payer-Rejected lane rendered alongside Rejected/Candidates/Unmatched/Done today, and a bulk Acknowledge action that drops claims from the working surface without erasing the original 277CA rejection event. New endpoint POST /api/inbox/payer-rejected/acknowledge (idempotent, audit-logged). Migration 0010. - SP15 (47902fd + ab00909) — SQLCipher key rotation in place via PRAGMA rekey. New endpoint POST /api/admin/db/rotate-key, serialized through a module-level threading.Lock. Switches the SQLAlchemy engine to NullPool when SQLCipher is enabled (QueuePool breaks SQLCipher thread affinity under FastAPI's per-request threadpool). Writes a db.key_rotated audit event with old + new key fingerprints and post-rotation table_count. What this PR does: - Inbox section + Inbox endpoint table: document the Payer-Rejected acknowledge action. - Encryption at Rest: add a 'Key rotation' sub-section that explains the rotate-key handler, the NullPool choice, the threading.Lock, and the error code mapping (409/400/503). - Tamper-Evident Audit Log: note that key rotations + Payer-Rejected acknowledgements are part of the audit-logged event set. - Project layout: add api_routers/ (health.py, acks.py, ta1_acks.py) as the FastAPI APIRouter subpackage extracted from api.py. - Roadmap: bump 'shipped 2-13' to 'shipped 2-15'; add SP14 and SP15 entries to the shipped list; add SP14 + SP15 endpoint reference sections at the end. Verification: - pytest --collect-only collects 759 tests (up from 733 at the previous doc pass). - git diff --check clean. - No code changes; no tests touched.
refactor(api): trailing newlines + hoist acks.py parser imports to top-level
2026-06-21 01:07:44 -06:00 · 2026-06-21 00:50:31 -06:00 · 2026-06-21 00:49:14 -06:00 · 2026-06-21 00:36:33 -06:00 · 2026-06-21 00:33:40 -06:00 · 2026-06-21 00:33:37 -06:00
12 changed files with 1594 additions and 372 deletions
@@ -130,10 +130,13 @@ backoff schedule on error is `1s → 2s → 4s → 8s → 16s → 30s` capped.
 ## Inbox
-`/inbox` is the working surface. Four lanes, dark by default (Ticker Tape
+`/inbox` is the working surface. Five lanes, dark by default (Ticker Tape
 aesthetic):
 - **Rejected** — claims whose 999 set-level response was R or E. Re-submit in bulk.
 - **Payer-rejected** — claims whose 277CA STC category is A4, A6, or A7 (the payer
  accepted the file but denied the claim). Stamped at 277CA ingest time and never
  overwritten by a looser later 277CA.
 - **Candidates** — remits whose CLP-claim-id didn't match exactly; each one shows its top scored claim. One-click manual match or dismiss.
 - **Unmatched** — claims still waiting for a remit, and remits with no candidates above the threshold.
 - **Done today** — terminal state transitions in the last 24 hours.
@@ -148,10 +151,11 @@ parses and rejects claims, the inbox reflects the new
 | Method | Path                                          | Notes                                                            |
 | ------ | --------------------------------------------- | ---------------------------------------------------------------- |
-| GET    | `/api/inbox/lanes`                            | All four lanes in one call.                                      |
+| GET    | `/api/inbox/lanes`                            | All five lanes in one call.                                      |
 | POST   | `/api/inbox/candidates/{remit_id}/match`      | Manual match. `409` if the claim state moved out from under us.  |
 | POST   | `/api/inbox/candidates/dismiss`               | `{pairs: [{claim_id, remit_id}]}`. Session-scoped.               |
 | POST   | `/api/inbox/rejected/resubmit`                | `{claim_ids: [...]}`. `200` with `conflicts` for non-rejected.   |
 | POST   | `/api/inbox/payer-rejected/acknowledge`       | `{claim_ids: [...], actor: "..."}`. Bulk-acknowledge payer rejections. Idempotent. `200` with `transitioned` / `already_acked` / `not_found` / `not_rejected` counts. Writes an audit event per transition; never overwrites the underlying 277CA rejection. |
 | GET    | `/api/inbox/export.csv?lane=<lane>`           | Streams CSV of the lane's rows.                                  |
 ## Outbound 837 Serializer
@@ -255,11 +259,174 @@ drawer surfaces a per-line "no 837 line matched" note.
 Tiers: **strong** (≥75, full opacity, Match enabled), **weak**
 (50–74, dimmed), **hidden** (<50, not surfaced).
 ## Multi-Payer, Multi-NPI & Clearhouse
 The payer and provider identity that used to live as a single hard-coded
 `PayerConfig` dict in the backend is now data, not code. Three new tables
 plus a YAML file drive the entire configuration:
 - **`providers` table** — one row per billing-provider NPI (Montrose
  `1881068062`, Delta `1851446637`, Salida `1467507269`). All three share
  the same `TOC, Inc.` legal name, tax ID `721587149`, and taxonomy
  `251E00000X`. Outbound 837 files pick the right `BillingProvider` by
  NPI; `claim.party.npi` is now a foreign key into `providers`.
 - **`payers` table** — one row per payer (`CO_TXIX`, …) with its
  receiver identity (NM1*40 / ISA08 / GS03).
 - **`payer_configs` join table** — one row per `(payer_id, transaction_type)`
  pair. 837P and 835 can carry different `BHT06`, SBR defaults, and
  allowed status codes per payer.
 - **`clearhouse` single-row config** — dzinesco's identity: TPID
  `11525703`, submitter name, MT-clock file-naming block, SFTP block.
 - **`config/payers.yaml`** — the on-disk source for everything above,
  schema-validated at boot against a Pydantic model. A typo or missing
  field fails the boot with a precise error. The original in-code
  `PAYER_FACTORIES` dict is kept as a fallback for ad-hoc testing.
 ### Config + clearhouse endpoints
 | Method | Path                                          | Notes                                                            |
 | ------ | --------------------------------------------- | ---------------------------------------------------------------- |
 | GET    | `/api/clearhouse`                             | The `clearhouse` singleton (name, TPID, file/SFTP blocks).       |
 | POST   | `/api/clearhouse/submit`                      | Push a batch of generated 837 files via SFTP (see SFTP section). |
 | GET    | `/api/config/providers`                       | All providers.                                                   |
 | GET    | `/api/config/providers/{npi}`                 | One provider.                                                    |
 | GET    | `/api/config/payers`                          | All payers.                                                      |
 | GET    | `/api/config/payers/{payer_id}/configs`       | All `(payer_id, transaction_type)` configs for one payer.        |
 | POST   | `/api/admin/reload-config`                    | Re-read `config/payers.yaml` and refresh the in-process cache.   |
 ## 277CA Claim Acknowledgment
 A 277CA (`005010X214`) is the per-claim acknowledgment CMS and
 Colorado Medicaid rely on: the file was syntactically valid *and* each
 named claim was accepted, pended, or rejected by the payer at the claim
 level. It is distinct from a 999 (file-level) and a TA1 (envelope-level).
 Cyclone ingests 277CA files the same way it ingests 999 / 835 — drop the
 file on the Upload page or `POST /api/parse-277ca` — and stamps every
 claim whose `STC` category is `A4`, `A6`, or `A7` with a non-null
 `payer_rejected_at` + `payer_rejected_reason` + originating 277CA row id.
 | Method | Path                       | Notes                                                            |
 | ------ | -------------------------- | ---------------------------------------------------------------- |
 | POST   | `/api/parse-277ca`         | Upload a 277CA, persist the parsed status rows.                  |
 | GET    | `/api/277ca-acks`          | List 277CA acks (filterable by date / payer).                    |
 | GET    | `/api/277ca-acks/{id}`     | One 277CA ack with its per-claim status rows + regenerated text. |
 The `payer_rejected` stamp is **monotonic**: a later 277CA with a looser
 status set cannot clear a previous rejection. The Payer-Rejected inbox
 lane surfaces every claim with a non-null `payer_rejected_at` — it is
 distinct from the 999 `Rejected` lane (envelope reject) and they can
 both be true for the same claim.
 ## Tamper-Evident Audit Log
 The `audit_log` table is the canonical record of every state transition
 the system has ever observed — claim lifecycle, reconciliation
 decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected
 acknowledgements, SQLCipher key rotations. SP11 made it tamper-evident:
 every row carries a SHA-256 hash of
 `(prev_hash || row_payload)`, forming a chain back to a genesis row.
 Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
 in a single walk.
 | Method | Path                              | Notes                                                          |
 | ------ | --------------------------------- | -------------------------------------------------------------- |
 | GET    | `/api/admin/audit-log`            | Paginated audit log (filterable by event type / actor / date). |
 | GET    | `/api/admin/audit-log/verify`     | Walk the chain; return the first broken link, or `{ok: true}`. |
 `verify_chain` is the integrity check that backs the audit promise —
 it is intentionally cheap (one indexed walk) and intentionally
 side-effect-free so a scheduler can run it on a cron and alert on any
 non-`{ok: true}` result. Chain verification is **not** access-gated
 beyond the same `127.0.0.1` bind the rest of the API uses; for a
 hostile multi-operator deployment, wrap the route in your reverse proxy.
 ## Encryption at Rest
 When the macOS Keychain carries an entry at service `cyclone`, account
 `cyclone.db.key`, and the optional `sqlcipher3` Python package is
 installed, the SQLite file at `~/.local/share/cyclone/cyclone.db` is
 opened with SQLCipher (AES-256). The key is read from the Keychain
 once at process start, applied via a SQLAlchemy `connect` event so
 every connection — including migrations and tests — gets the same
 `PRAGMA key`. The key is never written to disk or to a Python global.
 When the Keychain entry is missing **or** `sqlcipher3` is not
 installed, the DB falls back to plain SQLite. The intent is a graceful
 default for developers and CI; the production posture is that every
 operator has created the Keychain entry on first run. See
 [docs/reference/co-medicaid.md §Keychain setup](docs/reference/co-medicaid.md)
 for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv)
 mapping.
 ### Key rotation
 The DB encryption key is rotated in place via `POST /api/admin/db/rotate-key`.
 The handler:
 1. Generates a fresh 256-bit CSPRNG key (`db_crypto.generate_db_key()`).
 2. Persists the new key to the Keychain under the same
   `cyclone.db.key` account (overwriting the old one).
 3. Disposes the SQLAlchemy engine so pooled connections release the
   file (SQLCipher refuses to `PRAGMA rekey` while another connection
   holds the DB).
 4. Opens with the old key, issues `PRAGMA rekey`, and verifies the
   schema survived (table-count sanity check).
 5. Rebuilds the engine with the new key.
 6. Writes a `db.key_rotated` audit event carrying the SHA-256
   fingerprints of the old and new keys (first 8 hex chars) plus the
   post-rotation `table_count`.
 When SQLCipher is enabled the engine uses SQLAlchemy's `NullPool`
 instead of the default `QueuePool`. `QueuePool` returns connections to
 a shared queue that any thread can pull from, which breaks SQLCipher's
 thread affinity. `NullPool` trades connection reuse for thread safety
 — the only correct behavior under FastAPI's per-request threadpool.
 The rotation endpoint is serialized with a module-level
 `threading.Lock` (one rotation in flight at a time), returns `409` if
 a rotation is already running, `400` if encryption is not enabled,
 and `503` with a `reason` on `PRAGMA rekey` or Keychain failure so the
 operator can take the next step without parsing the traceback.
 | Method | Path                              | Notes                                                          |
 | ------ | --------------------------------- | -------------------------------------------------------------- |
 | POST   | `/api/admin/db/rotate-key`        | Rotate the SQLCipher key in place. Audit-logged.                |
 ## SFTP Wire-Up (paramiko)
 The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
 generated 837 files to the dzinesco SFTP server
 (`mft.gainwelltechnologies.com:22`, path
 `/CO XIX/PROD/coxix_prod_11525703/FromHPE`). The SFTP credential is
 fetched from the macOS Keychain at call time — never read from YAML,
 never logged, never written to disk. The wire-up honors the file-naming
 template stored in the `clearhouse` config:
 ```
 outbound: {tpid}-{tx}-{ts_mt}-1of1.{ext}        e.g. 11525703-837P-20260620181814559-1of1.txt
 inbound:  TP{tpid}-{orig_tx}_M{tracking}-{ts}-1of1_{file_type}.x12
 ```
 where `{ts_mt}` is a 17-digit `yyyymmddhhmmssSSS` Mountain Time stamp.
 Inbound filenames are routed by `<FileType>` and `<OrigTx>` to the
 matching parser (`999`, `TA1`, `271`, `277`, `277CA`, `835`).
 The `SftpClient` interface is the same one the SP9 stub used — the swap
 was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
 `paramiko` is an optional dependency; the stub remains the default when
 the `paramiko` extras aren't installed so the test suite stays green on
 Linux dev boxes.
 ## Persistence
-Parsed batches, claims, remittances, matches, and activity events are
+Parsed batches, claims, remittances, matches, 277CA rejections,
-stored in a SQLite file at `~/.local/share/cyclone/cyclone.db` by
+hash-chained audit log entries, and SFTP submission history are stored
-default. The directory is auto-created on first run.
+in a SQLite file at `~/.local/share/cyclone/cyclone.db` by default.
 The directory is auto-created on first run. The DB is optionally
 encrypted with SQLCipher — see
 [Encryption at Rest](#encryption-at-rest) for the Keychain-driven
 setup.
 To use a different location, set `CYCLONE_DB_URL`:
@@ -284,12 +451,28 @@ backup API).
 .
 ├── backend/
 │   ├── src/cyclone/
-│   │   ├── api.py             # FastAPI app, GET + parse routes, /api/{resource}/stream
+│   │   ├── api.py             # FastAPI app + parse routes; 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}
 │   │   ├── pubsub.py          # in-process EventBus (drop-oldest, per-kind fan-out)
-│   │   ├── store.py           # InMemoryStore, mappers, publish-on-write
+│   │   ├── store.py           # CycloneStore, mappers, publish-on-write
 │   │   ├── db.py              # SQLAlchemy engine, session factory, ORM models
 │   │   ├── db_migrate.py      # PRAGMA user_version migration runner
 │   │   ├── db_crypto.py       # optional SQLCipher encryption at rest (SP12)
 │   │   ├── audit_log.py       # tamper-evident hash-chained audit_log (SP11)
 │   │   ├── inbox_lanes.py     # rejected / payer_rejected / candidates / unmatched / done_today
 │   │   ├── inbox_state.py     # 999 envelope reject → claim state transitions
 │   │   ├── inbox_state_277ca.py # 277CA STC A4/A6/A7 → payer_rejected stamp (SP10)
 │   │   ├── providers.py       # multi-NPI provider lookups (SP9)
 │   │   ├── payers.py          # payer / payer_config lookups (SP9)
 │   │   ├── secrets.py         # macOS Keychain-backed secret fetcher
 │   │   ├── reconcile.py       # pure-function 835→claim match + line-level match
 │   │   ├── __main__.py        # `python -m cyclone serve`
 │   │   ├── cli.py             # click CLI
-│   │   └── parsers/           # X12 tokenizer, models, validator, writers
+│   │   └── parsers/           # X12 tokenizer, models, validator, writers, 277CA, 999, TA1, 270, 271
 │   └── tests/
 │       ├── fixtures/          # co_medicaid_*.txt, minimal_*.txt
 │       ├── test_api.py        # parse-837/835 round-trip
@@ -298,29 +481,84 @@ backup API).
 │       ├── test_api_streaming.py
 │       ├── test_api_stream_live.py  # 3 live-tail endpoints + disconnect cleanup
 │       ├── test_pubsub.py     # EventBus + subscribe/unsubscribe
-│       └── test_api_parse_persists.py
+│       ├── test_api_parse_persists.py
 │       ├── test_db.py / test_db_crypto.py / test_db_migrate.py
 │       ├── test_audit_log.py
 │       ├── test_inbox_lanes.py / test_inbox_state.py
 │       ├── test_apply_277ca_rejections.py
 │       ├── test_sftp_stub.py / test_sftp_paramiko.py
 │       └── test_providers_seed.py / test_payer_config_loading.py
 ├── src/                       # React + Vite + TypeScript UI
 │   ├── components/
 │   │   ├── ui/                # Skeleton, EmptyState, ErrorState, FilterChips, Pagination, …
 │   │   └── TailStatusPill.tsx # live-tail status badge + reconnect button
-│   ├── pages/                 # Claims, Remittances, Providers, Activity, Upload
+│   ├── pages/                 # Claims, Remittances, Providers, Acks, Activity, Upload, Inbox, …
 │   ├── hooks/                 # useBatches, useClaims, useRemittances, useProviders, useActivity, useParse
 │   │                          # + useTailStream, useMergedTail (live tail)
-│   ├── lib/                   # api.ts (6 GET + parse837/parse835/health), format.ts, utils.ts
+│   ├── lib/                   # api.ts, format.ts, utils.ts
 │   │                          # + tail-stream.ts (NDJSON parser)
 │   ├── store/                 # zustand sample-data + parsed-batches store
 │   │                          # + tail-store.ts (FIFO-capped live tail slices)
 │   └── types/                 # shared TS types
 ├── config/
 │   └── payers.yaml            # YAML-driven payer + clearhouse config (SP9)
 ├── docs/
-│   ├── reference/             # condensed 837P/835/X12/CO Medicaid notes
+│   ├── reference/             # condensed 837P/835/X12/CO Medicaid notes (incl. Keychain setup)
-│   └── superpowers/plans/     # implementation plan
+│   ├── reviews/               # post-SP completeness reviews
 │   ├── superpowers/plans/     # implementation plans
 │   └── superpowers/specs/     # design specs (incl. SP9-SP13)
 ├── tailwind.config.js         # shimmer, scan, row-flash keyframes
 └── package.json
 ```
 ## Roadmap
-Sub-projects 2 through 8 are **shipped**. Next up:
+Sub-projects 2 through 15 are **shipped**. See the [completeness
 review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for
 the honest gap analysis against the industry definition of a HIPAA
 clearinghouse — the short version is that the local-only,
 single-operator, single-payer design contract is honored, and the
 items that would be needed to expand that contract (AS2/AS4, SNIP 1–7,
 HITRUST, 276/277 status, 278 referrals, COB) are intentionally out of
 scope.
 Shipped sub-projects (most recent first):
 - **Sub-project 15 (shipped) — SQLCipher key rotation.** In-place
  rotation via `PRAGMA rekey`, serialized through a module-level
  `threading.Lock` and a SQLAlchemy `NullPool` to keep SQLCipher
  thread-affine under FastAPI's per-request threadpool. Writes a
  `db.key_rotated` audit event with old + new key fingerprints and
  post-rotation `table_count`. See
  [Encryption at Rest — Key rotation](#key-rotation).
 - **Sub-project 14 (shipped) — 5-lane Inbox UI.** The Payer-Rejected
  lane is now rendered in the Inbox alongside Rejected / Candidates /
  Unmatched / Done today. New bulk action
  `POST /api/inbox/payer-rejected/acknowledge` drops claims from the
  working surface without erasing the original 277CA rejection event
  (audit log stays intact, SP11).
 - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
  `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
  actually pushes to
  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
  SFTP credentials are read from the macOS Keychain at call time.
 - **Sub-project 12 (shipped) — Encryption at rest.** Optional
  SQLCipher AES-256 encryption of the SQLite file, with the key
  fetched from the macOS Keychain. Falls back to plain SQLite when
  the Keychain entry is missing or `sqlcipher3` isn't installed.
 - **Sub-project 11 (shipped) — Tamper-evident audit log.** Every
  `audit_log` row carries a SHA-256 hash chained to the previous row;
  a single walk via `GET /api/admin/audit-log/verify` detects any
  break.
 - **Sub-project 10 (shipped) — 277CA + Payer-Rejected lane.** Inbound
  277CA parser + a new Payer-Rejected inbox lane distinct from the
  999-envelope Rejected lane. The rejection stamp is monotonic.
 - **Sub-project 9 (shipped) — Multi-payer, multi-NPI, SFTP stub.** The
  in-code `PAYER_FACTORIES` dict is replaced by a `config/payers.yaml`
  + 3 new DB tables (`providers`, `payers`, `payer_configs`) +
  `clearhouse` singleton. Added a `POST /api/clearhouse/submit` stub
  that writes to a local `staging_dir` — swapped for real `paramiko`
  in SP13.
 - **Sub-project 8 (shipped) — Outbound 837P serializer.** Closes the
  resubmit loop: rejected claims can be regenerated back to an X12 837P
@@ -390,8 +628,9 @@ Sub-projects 2 through 8 are **shipped**. Next up:
    back-off ladder on errors is `1s → 2s → 4s → 8s → 16s → 30s`
    capped. See the "Live updates" section below for details.
 - **Sub-project 6 (shipped) — Inbox workflow automation.**
-  - **Ticker-Tape inbox (`/inbox`):** four lanes ordered by urgency —
+  - **Ticker-Tape inbox (`/inbox`):** five lanes ordered by urgency —
-    **Rejected** (claims whose 999 rejected them), **Candidates**
+    **Rejected** (claims whose 999 rejected them), **Payer-rejected**
    (claims whose 277CA denied them — added in SP10), **Candidates**
    (remits that didn't auto-match a claim), **Unmatched** (claims
    still waiting for a remit), and **Done today** (terminal
    transitions in the last 24 hours). The page subscribes to the
@@ -428,6 +667,10 @@ Sub-projects 2 through 8 are **shipped**. Next up:
 - `GET /api/acks` — list ACKs.
 - `GET /api/acks/{id}` — ACK detail, including the regenerated
  `raw_999_text`.
 - `POST /api/parse-ta1` — parse an inbound TA1 envelope ACK and persist it.
 - `GET /api/ta1-acks` — list TA1 acks.
 - `GET /api/ta1-acks/{id}` — TA1 ack detail (envelope control segments
  + the parser's accept/reject verdict).
 - `POST /api/eligibility/request` — build a 270 from JSON.
 - `POST /api/eligibility/parse-271` — ingest a 271 and return parsed
  coverage benefits.
@@ -459,9 +702,9 @@ ACKs and lets you download the regenerated 999 text.
 ### SP6 endpoints (inbox)
- `GET /api/inbox/lanes` — all four lanes (rejected, candidates,
+- `GET /api/inbox/lanes` — all five lanes (rejected, payer_rejected,
-  unmatched, done_today) in a single round-trip, with row-level
+  candidates, unmatched, done_today) in a single round-trip, with
-  scoring and matched-remit context.
+  row-level scoring and matched-remit context.
 - `POST /api/inbox/candidates/{remit_id}/match` — manual match of a
  candidate remit to one of its scored claims; `409` if the claim
  state moved out from under us.
@@ -510,6 +753,86 @@ ACKs and lets you download the regenerated 999 text.
  Download** button in the Inbox rejected-lane BulkBar (N>1 modal
  prompt).
 ### SP9 endpoints (multi-payer, multi-NPI, SFTP stub)
 - `GET /api/clearhouse` — the `clearhouse` singleton (name, TPID,
  file-naming block, SFTP block).
 - `POST /api/clearhouse/submit` — push a batch of generated 837 files.
  The SP9 implementation writes to a local `staging_dir`; the SP13
  swap replaces the write with a real `paramiko` SFTP push without
  changing the route shape.
 - `GET /api/config/providers` and `GET /api/config/providers/{npi}` —
  list / fetch providers from the new `providers` table.
 - `GET /api/config/payers` and
  `GET /api/config/payers/{payer_id}/configs` — list payers; for a
  given payer, return the per-transaction-type `payer_configs` rows.
 - `POST /api/admin/reload-config` — re-read `config/payers.yaml` and
  refresh the in-process cache without a server restart.
 ### SP10 endpoints (277CA + Payer-Rejected lane)
 - `POST /api/parse-277ca` — upload a 277CA file; persist the parsed
  `ClaimStatus` rows and stamp the matching claims with
  `payer_rejected_at` (monotonic, never overwritten by `NULL`).
 - `GET /api/277ca-acks` — list 277CA acks.
 - `GET /api/277ca-acks/{id}` — one 277CA ack with its per-claim
  `ClaimStatus` rows + regenerated text.
 - `GET /api/inbox/lanes` — the response now also carries a
  `payer_rejected` lane populated from
  `Claim.payer_rejected_at IS NOT NULL`.
 ### SP11 endpoints (tamper-evident audit log)
 - `GET /api/admin/audit-log` — paginated audit log. Each row carries
  `(id, prev_hash, row_hash, event_type, actor, payload_json,
  created_at)` where `row_hash = sha256(prev_hash || canonical_json(payload))`.
 - `GET /api/admin/audit-log/verify` — walk the chain in insertion
  order; return `{ok: true}` or the first `{id, expected, got}`
  mismatch. The walk is O(n) with one indexed lookup per row.
 ### SP12 (encryption at rest — no new routes)
 SP12 introduces no API routes. The `cyclone.db.key` Keychain entry +
 optional `sqlcipher3` dependency enable AES-256 encryption transparently
 on the next connection. See [Encryption at Rest](#encryption-at-rest)
 and [docs/reference/co-medicaid.md](docs/reference/co-medicaid.md) for
 the one-time setup recipe.
 ### SP13 endpoints (paramiko SFTP)
 - `POST /api/clearhouse/submit` — same endpoint as SP9; the
  implementation is now a real `paramiko` `SftpClient.write` to
  `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
  SFTP credentials are fetched from the macOS Keychain at call time.
 ### SP14 endpoints (5-lane Inbox UI + acknowledge)
 - `GET /api/inbox/lanes` — same shape as SP6; the `payer_rejected`
  lane payload now also includes
  `payer_rejected_acknowledged_at` + `payer_rejected_acknowledged_actor`
  per row so the UI can badge acknowledged claims (forward-compat for
  a future "Recently acknowledged" view).
 - `POST /api/inbox/payer-rejected/acknowledge` — bulk-acknowledge
  Payer-Rejected claims. Body: `{claim_ids: [...], actor: "..."}`.
  Idempotent. Response: `200` with
  `{transitioned, already_acked, not_found, not_rejected}`. Writes an
  `inbox.payer_rejected_acknowledged` audit event per transition.
  Acknowledgement hides a claim from the lane but never overwrites
  `payer_rejected_at` / `payer_rejected_reason` /
  `payer_rejected_by_277ca_id` — the original 277CA evidence stays
  intact in the audit log.
 ### SP15 endpoints (SQLCipher key rotation)
 - `POST /api/admin/db/rotate-key` — rotate the SQLCipher DB key in
  place. Generates a fresh 256-bit key, writes it to the Keychain
  (overwriting `cyclone.db.key`), disposes the engine, issues
  `PRAGMA rekey`, verifies the schema, rebuilds the engine. Writes
  a `db.key_rotated` audit event with old + new key fingerprints
  and `table_count`. Returns `409` when a rotation is already in
  flight, `400` when encryption is not enabled, `503` with a
  `reason` on `PRAGMA rekey` or Keychain failure.
 ## License
 No license file yet; this is internal-use software. Add a `LICENSE` file
@@ -78,11 +78,25 @@ python -m cyclone serve
 ### Endpoints
-| Method | Path             | Purpose                                            |
+| Method | Path                  | Purpose                                                            |
-| ------ | ---------------- | -------------------------------------------------- |
+| ------ | --------------------- | ------------------------------------------------------------------ |
-| GET    | `/api/health`    | Liveness probe: `{"status": "ok", "version": ...}` |
+| GET    | `/api/health`         | Liveness probe: `{"status": "ok", "version": ...}`                 |
-| POST   | `/api/parse-837` | Upload an X12 837P file, get parsed claims back    |
+| POST   | `/api/parse-837`      | Upload an X12 837P file, get parsed claims back                    |
-| POST   | `/api/parse-835` | Upload an X12 835 ERA file, get parsed payouts back |
+| POST   | `/api/parse-835`      | Upload an X12 835 ERA file, get parsed payouts back                |
 | POST   | `/api/parse-999`      | Upload an inbound 999 ACK and persist it                           |
 | POST   | `/api/parse-ta1`      | Upload an inbound TA1 envelope ACK and persist it                  |
 | POST   | `/api/parse-277ca`    | Upload a 277CA claim acknowledgment and stamp payer_rejected claims |
 | POST   | `/api/eligibility/request`   | Build a 270 from JSON (subscriber / provider / payer)        |
 | POST   | `/api/eligibility/parse-271` | Ingest a 271 and return structured `coverage_benefits`        |
 | GET    | `/api/clearhouse`     | The `clearhouse` singleton (SP9)                                   |
 | POST   | `/api/clearhouse/submit` | Push a batch of generated 837 files via SFTP (SP9 stub, SP13 real) |
 | POST   | `/api/admin/reload-config`  | Re-read `config/payers.yaml` and refresh the in-process cache |
 | GET    | `/api/admin/audit-log` | Paginated tamper-evident audit log (SP11)                          |
 | GET    | `/api/admin/audit-log/verify` | Walk the audit_log hash chain (SP11)                       |
 The full surface — claim / remittance / batch / inbox / stream
 endpoints, config lookups, and the 270/271 builder — is enumerated in
 the root [README](../README.md#multi-payer-multi-npi--clearhouse).
 `POST /api/parse-837` accepts `multipart/form-data` with a single `file`
 field. Optional query parameters:
@@ -15,12 +15,10 @@ plus GET/POST with any header.
 from __future__ import annotations
 import asyncio
 import csv
 import io
 import json
 import logging
 import os
 import uuid
 from datetime import datetime, timezone
 from contextlib import asynccontextmanager
@@ -71,16 +69,24 @@ from cyclone.store import (
 log = logging.getLogger(__name__)
-
+# Cross-cutting helpers (NDJSON serialization, content negotiation, strict
-def _ndjson_line(event: dict) -> bytes:
+# rewrites, live-tail generator). See api_helpers.py for the rationale.
-    """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``).
+from cyclone.api_helpers import (  # noqa: E402
-
+    client_wants_json as _client_wants_json,
-    Used by the live-tail streaming endpoints to emit a uniform wire format
+    drop_raw_segments_837 as _drop_raw_segments,
-    that the frontend ``tail-stream.ts`` parser can split on newlines.
+    drop_raw_segments_835 as _drop_raw_segments_835,
-    Compact separators keep each line small and avoid ambiguity with embedded
+    has_claim_validation_errors as _has_claim_validation_errors,
-    whitespace.
+    has_835_validation_errors as _has_835_validation_errors,
-    """
+    heartbeat_seconds as _heartbeat_seconds,
-    return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8")
+    ndjson_line as _ndjson_line,
    ndjson_stream_837 as _ndjson_stream,
    ndjson_stream_835 as _ndjson_stream_835,
    ndjson_stream_list as _ndjson_stream_list,
    strict_rewrite_837 as _strict_rewrite,
    strict_rewrite_835 as _strict_rewrite_835,
    tail_events as _tail_events,
    wants_ndjson as _wants_ndjson,
 )
@asynccontextmanager
@@ -130,6 +136,17 @@ app.add_middleware(
    allow_headers=["*"],
 )
 # Resource-group routers. Each module owns its own APIRouter and is
 # registered below. New resources go in `cyclone.api_routers.<name>`
 # and are wired in here. (Kept as a top-level package rather than nested
 # under `cyclone.api` so the existing ``cyclone.api`` module path keeps
 # working — Python prefers packages over same-named modules.)
 from cyclone.api_routers import acks, health, ta1_acks  # noqa: E402
 app.include_router(health.router)
 app.include_router(acks.router)
 app.include_router(ta1_acks.router)
 def _resolve_payer(name: str) -> PayerConfig:
    if name not in PAYER_FACTORIES:
@@ -155,96 +172,6 @@ def _resolve_payer_835(name: str) -> PayerConfig835:
    return PAYER_FACTORIES_835[name]()
 def _strict_rewrite(result: ParseResult) -> ParseResult:
    """Promote warnings to errors (mirrors the CLI's --strict)."""
    claims: list[ClaimOutput] = []
    for claim in result.claims:
        promoted = [
            issue.model_copy(update={"severity": "error"})
            for issue in claim.validation.warnings
        ]
        new_errors = claim.validation.errors + promoted
        claims.append(
            claim.model_copy(
                update={
                    "validation": claim.validation.model_copy(
                        update={"errors": new_errors, "passed": not new_errors}
                    )
                }
            )
        )
    passed = sum(1 for c in claims if c.validation.passed)
    failed = len(claims) - passed
    summary = result.summary.model_copy(
        update={
            "passed": passed,
            "failed": failed,
            "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed],
        }
    )
    return result.model_copy(update={"claims": claims, "summary": summary})
 def _drop_raw_segments(result: ParseResult) -> ParseResult:
    """Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
    claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
    return result.model_copy(update={"claims": claims})
 def _client_wants_json(request: Request) -> bool:
    """Content negotiation: prefer ``application/json`` when the client asks for it.
    NDJSON is the default for browser uploads that don't set ``Accept``. The
    frontend opts into JSON via ``Accept: application/json``.
    """
    accept = request.headers.get("accept", "")
    # If the client mentions JSON at all (and isn't asking for NDJSON
    # specifically) treat it as a single-object request. The browser default
    # ``*/*`` falls through to NDJSON.
    if "application/json" in accept and "application/x-ndjson" not in accept:
        return True
    return False
 def _wants_ndjson(request: Request) -> bool:
    """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the
    default (per spec 6.2: "Default JSON response wraps the same data in a
    {items, total, returned, has_more} envelope so the frontend can paginate
    uniformly").
    Used by the GET list routes (/api/batches, /api/claims, /api/remittances,
    /api/providers, /api/activity). NDJSON is returned only when the client
    explicitly sends ``Accept: application/x-ndjson`` (with or without
    ``application/json``). Bare ``*/*``, an empty Accept, or an explicit
    ``Accept: application/json`` all return the JSON envelope.
    """
    accept = request.headers.get("accept", "")
    return "application/x-ndjson" in accept
 def _ndjson_stream_list(
    items: list[dict], total: int, returned: int, has_more: bool,
 ) -> Iterator[str]:
    """Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a
    final ``summary`` line. Mirrors spec section 6.2 streaming rule.
    """
    for it in items:
        yield json.dumps({"type": "item", "data": it}) + "\n"
    yield json.dumps({
        "type": "summary",
        "data": {"total": total, "returned": returned, "has_more": has_more},
    }) + "\n"
 def _has_claim_validation_errors(result: ParseResult) -> bool:
    return any(not c.validation.passed for c in result.claims)
@app.get("/api/health")
 def health() -> dict[str, str]:
    return {"status": "ok", "version": __version__}
@app.post("/api/parse-837")
 async def parse_837(
    request: Request,
@@ -362,48 +289,11 @@ def _build_and_persist_ack(batch_id: str) -> dict | None:
    }
 def _ndjson_stream(result: ParseResult) -> Iterator[bytes]:
    """Yield one JSON object per line: envelope → claims → summary."""
    envelope_obj = (
        result.envelope.model_dump() if result.envelope is not None else None
    )
    yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8")
    for claim in result.claims:
        yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
 # --------------------------------------------------------------------------- #
 # 835 ERA (Health Care Claim Payment/Advice)
 # --------------------------------------------------------------------------- #
 def _strict_rewrite_835(result: ParseResult835) -> ParseResult835:
    """Promote warnings to errors (mirrors the CLI's --strict)."""
    if result.validation is None:
        return result
    report = result.validation
    promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
    new_errors = report.errors + promoted
    new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
    passed = 1 if new_report.passed else 0
    failed = 1 if not new_report.passed else 0
    new_summary = result.summary.model_copy(
        update={"passed": passed, "failed": failed}
    )
    return result.model_copy(update={"validation": new_report, "summary": new_summary})
 def _drop_raw_segments_835(result: ParseResult835) -> ParseResult835:
    """Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
    claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
    return result.model_copy(update={"claims": claims})
 def _has_835_validation_errors(result: ParseResult835) -> bool:
    return result.validation is not None and not result.validation.passed
 def _reconciliation_summary_for_batch(batch_id: str) -> dict:
    """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch.
@@ -530,18 +420,6 @@ async def parse_835_endpoint(
    )
 def _ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]:
    """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary."""
    yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8")
    for claim in result.claims:
        yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
 # --------------------------------------------------------------------------- #
 # 999 ACK (Implementation Acknowledgment)
 # --------------------------------------------------------------------------- #
@@ -787,53 +665,6 @@ async def parse_ta1_endpoint(
    })
@app.get("/api/ta1-acks")
 def list_ta1_acks_endpoint(
    limit: int = Query(100, ge=1, le=1000),
 ) -> Any:
    """Return the list of persisted TA1 ACKs, newest first.
    Mirrors :func:`list_acks_endpoint` — fetches all rows then slices in
    Python so the ``total`` field reflects the full row count regardless
    of the ``limit`` cap.
    """
    rows = store.list_ta1_acks()
    items = [_ta1_to_ui(r) for r in rows[:limit]]
    return {
        "total": len(rows),
        "items": items,
    }
@app.get("/api/ta1-acks/{ack_id}")
 def get_ta1_ack_endpoint(ack_id: int) -> dict:
    """Return one persisted TA1 ACK row with its parsed detail."""
    row = store.get_ta1_ack(ack_id)
    if row is None:
        raise HTTPException(status_code=404, detail=f"TA1 ACK {ack_id} not found")
    body = _ta1_to_ui(row)
    body["raw_ta1_text"] = _serialize_ta1_from_row(row)
    body["raw_json"] = row.raw_json
    return body
 def _ta1_to_ui(row: db.Ta1Ack) -> dict:
    """Render a Ta1Ack row for the UI (list endpoint shape)."""
    return {
        "id": row.id,
        "control_number": row.control_number,
        "ack_code": row.ack_code,
        "note_code": row.note_code,
        "interchange_date": row.interchange_date.isoformat()
            if row.interchange_date else None,
        "interchange_time": row.interchange_time,
        "sender_id": row.sender_id,
        "receiver_id": row.receiver_id,
        "source_batch_id": row.source_batch_id,
        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
    }
 # --------------------------------------------------------------------------- #
 # 277CA (Claim Acknowledgment) — SP10
 # --------------------------------------------------------------------------- #
@@ -1105,7 +936,7 @@ def inbox_dismiss_candidates(body: dict):
    return {"ok": True, "dismissed_count": len(pairs)}
-# ---------------------------------------------------------------------------
+# --------------------------------------------------------------------------- #
 # SP14: Payer-Rejected acknowledge
 #
 # Operator hits "Acknowledge" on the Payer-Rejected Inbox lane to clear
@@ -1116,7 +947,7 @@ def inbox_dismiss_candidates(body: dict):
 # 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."""
@@ -1274,7 +1105,7 @@ def inbox_resubmit_rejected(
@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", "payer_rejected"}:
+    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:
@@ -1415,71 +1246,6 @@ def list_claims(
 # --------------------------------------------------------------------------- #
 def _heartbeat_seconds() -> float:
    """Return the configured tail heartbeat interval.
    Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can
    monkeypatch the env var without reloading the module. Defaults to
    15s (the production cadence); tests override to a small value (e.g.
    0.2s) to keep their runtime bounded.
    """
    raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15")
    try:
        v = float(raw)
    except ValueError:
        return 15.0
    return v if v > 0 else 15.0
 async def _tail_events(
    request: Request, bus: EventBus, kinds: list[str]
 ) -> AsyncIterator[bytes]:
    """Forward subscribed events as ``item`` lines with periodic heartbeats.
    Polls the underlying ``asyncio.Queue`` directly (via
    :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's
    async-iterator wrapper. ``asyncio.wait_for`` cancels the inner
    future on timeout, which would otherwise terminate the bus
    iterator at its ``await`` point and break subsequent
    ``__anext__`` calls with ``StopAsyncIteration``. Polling
    ``queue.get()`` is idempotent under cancellation, so heartbeats
    don't poison the subscription.
    A ``try/finally`` unsubscribes the queue from the bus when the
    caller disconnects or the generator is garbage collected —
    otherwise the bus would leak one queue per open stream.
    """
    hb_s = _heartbeat_seconds()
    queue, _sub = bus.subscribe_raw(kinds)
    try:
        while True:
            if await request.is_disconnected():
                return
            get_task = asyncio.ensure_future(queue.get())
            sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s))
            try:
                done, pending = await asyncio.wait(
                    {get_task, sleep_task},
                    return_when=asyncio.FIRST_COMPLETED,
                )
            except BaseException:
                get_task.cancel()
                sleep_task.cancel()
                raise
            for t in pending:
                t.cancel()
            if get_task in done:
                event = get_task.result()
                yield _ndjson_line({"type": "item", "data": event})
            else:
                yield _ndjson_line({
                    "type": "heartbeat",
                    "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")},
                })
    finally:
        bus.unsubscribe(queue, kinds)
@app.get("/api/claims/stream")
 async def claims_stream(
    request: Request,
@@ -2136,86 +1902,6 @@ async def activity_stream(
 # --------------------------------------------------------------------------- #
 def _ack_to_ui(row) -> dict:
    """Map an ``Ack`` ORM row to the UI shape used by ``/api/acks``.
    Field names match the rest of the Cyclone API (snake_case). The
    frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack``
    interface in ``src/types/index.ts``.
    """
    return {
        "id": row.id,
        "source_batch_id": row.source_batch_id,
        "accepted_count": row.accepted_count,
        "rejected_count": row.rejected_count,
        "received_count": row.received_count,
        "ack_code": row.ack_code,
        "parsed_at": (
            row.parsed_at.isoformat().replace("+00:00", "Z")
            if row.parsed_at is not None
            else ""
        ),
    }
@app.get("/api/acks")
 def list_acks_endpoint(
    request: Request,
    limit: int = Query(100, ge=1, le=1000),
 ) -> Any:
    """Return the list of persisted 999 ACKs, newest first."""
    rows = store.list_acks()
    items = [_ack_to_ui(r) for r in rows[:limit]]
    total = len(rows)
    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,
    }
@app.get("/api/acks/{ack_id}")
 def get_ack_endpoint(ack_id: int) -> dict:
    """Return one persisted ACK row with its parsed detail.
    Path param is ``ack_id`` (not ``id``) to avoid shadowing FastAPI's
    internal ``id`` name and to keep OpenAPI docs self-describing.
    Returns 404 when the ACK is missing — never 500.
    """
    from cyclone import db as _db
    row = store.get_ack(ack_id)
    if row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": "Not found", "detail": f"Ack {ack_id} not found"},
        )
    body = _ack_to_ui(row)
    body["raw_json"] = row.raw_json
    # Regenerate the X12 text from raw_json so the operator can download
    # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry
    # the regenerated text to keep payloads small; detail does.)
    if row.raw_json:
        try:
            from cyclone.parsers.models_999 import ParseResult999
            regenerated = ParseResult999.model_validate(row.raw_json)
            icn = regenerated.envelope.control_number or "000000001"
            body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn)
        except Exception as exc:  # noqa: BLE001 — never 500 on a regen failure
            log.warning("Could not regenerate 999 for ack %s: %s", ack_id, exc)
            body["raw_999_text"] = None
    else:
        body["raw_999_text"] = None
    return body
 # --------------------------------------------------------------------------- #
 # 270 / 271 eligibility (SP3 P4 T23–T24) — API-only, no DB persistence
 # --------------------------------------------------------------------------- #
@@ -2589,6 +2275,161 @@ def verify_audit_log_endpoint() -> Any:
    }
 # ---------------------------------------------------------------------------
 # 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.get("/api/config/providers/{npi}")
 def get_configured_provider(npi: str):
    p = store.get_provider(npi)
@@ -0,0 +1,247 @@
 """Shared helpers used by ``cyclone.api`` route handlers.
 Everything in this module is private to the API layer (no business
 logic, no DB writes). It collects the cross-cutting concerns that used
 to live inline at the top of ``api.py``:
 * NDJSON wire-format primitives (``ndjson_line``, ``ndjson_stream_list``,
  ``ndjson_stream_837``, ``ndjson_stream_835``).
 * Content negotiation (``client_wants_json``, ``wants_ndjson``).
 * Strict / ``raw_segments`` rewrites applied before persisting parsed
  837P and 835 results.
 * Validation-error probes for both transactions.
 * The shared live-tail async generator (``tail_events``,
  ``heartbeat_seconds``) used by every ``/api/<resource>/stream``
  endpoint.
 Extracted as part of the api.py router split (see /tmp/refactor-cyclone.md).
 """
 from __future__ import annotations
 import asyncio
 import json
 import os
 from datetime import datetime, timezone
 from typing import AsyncIterator, Iterator
 from fastapi import Request
 from cyclone.parsers.models import ClaimOutput, ParseResult
 from cyclone.parsers.models_835 import ParseResult835
 from cyclone.pubsub import EventBus
 def utcnow() -> datetime:
    """tz-aware UTC ``datetime`` (matches :func:`cyclone.store.utcnow`)."""
    return datetime.now(timezone.utc)
 def ndjson_line(event: dict) -> bytes:
    """Serialize one event dict as a single NDJSON line (UTF-8, trailing ``\\n``).
    Used by the live-tail streaming endpoints to emit a uniform wire format
    that the frontend ``tail-stream.ts`` parser can split on newlines.
    Compact separators keep each line small and avoid ambiguity with embedded
    whitespace.
    """
    return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8")
 def client_wants_json(request: Request) -> bool:
    """Content negotiation: prefer ``application/json`` when the client asks for it.
    NDJSON is the default for browser uploads that don't set ``Accept``. The
    frontend opts into JSON via ``Accept: application/json``.
    """
    accept = request.headers.get("accept", "")
    # If the client mentions JSON at all (and isn't asking for NDJSON
    # specifically) treat it as a single-object request. The browser default
    # ``*/*`` falls through to NDJSON.
    if "application/json" in accept and "application/x-ndjson" not in accept:
        return True
    return False
 def wants_ndjson(request: Request) -> bool:
    """Content negotiation for list endpoints: NDJSON is an opt-in, JSON is the
    default (per spec 6.2: "Default JSON response wraps the same data in a
    {items, total, returned, has_more} envelope so the frontend can paginate
    uniformly").
    Used by the GET list routes (/api/batches, /api/claims, /api/remittances,
    /api/providers, /api/activity). NDJSON is returned only when the client
    explicitly sends ``Accept: application/x-ndjson`` (with or without
    ``application/json``). Bare ``*/*``, an empty Accept, or an explicit
    ``Accept: application/json`` all return the JSON envelope.
    """
    accept = request.headers.get("accept", "")
    return "application/x-ndjson" in accept
 def ndjson_stream_list(
    items: list[dict], total: int, returned: int, has_more: bool,
 ) -> Iterator[str]:
    """Yield NDJSON lines for a list endpoint: one ``item`` per dict, then a
    final ``summary`` line. Mirrors spec section 6.2 streaming rule.
    """
    for it in items:
        yield json.dumps({"type": "item", "data": it}) + "\n"
    yield json.dumps({
        "type": "summary",
        "data": {"total": total, "returned": returned, "has_more": has_more},
    }) + "\n"
 def ndjson_stream_837(result: ParseResult) -> Iterator[bytes]:
    """Yield one JSON object per line: envelope → claims → summary."""
    envelope_obj = (
        result.envelope.model_dump() if result.envelope is not None else None
    )
    yield (json.dumps({"type": "envelope", "data": envelope_obj}) + "\n").encode("utf-8")
    for claim in result.claims:
        yield (json.dumps({"type": "claim", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
 def ndjson_stream_835(result: ParseResult835) -> Iterator[bytes]:
    """Yield one JSON object per line: envelope → financial → trace → payer → payee → claim_payments → summary."""
    yield (json.dumps({"type": "envelope", "data": json.loads(result.envelope.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "financial_info", "data": json.loads(result.financial_info.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "trace", "data": json.loads(result.trace.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "payer", "data": json.loads(result.payer.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "payee", "data": json.loads(result.payee.model_dump_json())}) + "\n").encode("utf-8")
    for claim in result.claims:
        yield (json.dumps({"type": "claim_payment", "data": json.loads(claim.model_dump_json())}) + "\n").encode("utf-8")
    yield (json.dumps({"type": "summary", "data": json.loads(result.summary.model_dump_json())}) + "\n").encode("utf-8")
 def strict_rewrite_837(result: ParseResult) -> ParseResult:
    """Promote warnings to errors (mirrors the CLI's --strict)."""
    claims: list[ClaimOutput] = []
    for claim in result.claims:
        promoted = [
            issue.model_copy(update={"severity": "error"})
            for issue in claim.validation.warnings
        ]
        new_errors = claim.validation.errors + promoted
        claims.append(
            claim.model_copy(
                update={
                    "validation": claim.validation.model_copy(
                        update={"errors": new_errors, "passed": not new_errors}
                    )
                }
            )
        )
    passed = sum(1 for c in claims if c.validation.passed)
    failed = len(claims) - passed
    summary = result.summary.model_copy(
        update={
            "passed": passed,
            "failed": failed,
            "failed_claim_ids": [c.claim_id for c in claims if not c.validation.passed],
        }
    )
    return result.model_copy(update={"claims": claims, "summary": summary})
 def strict_rewrite_835(result: ParseResult835) -> ParseResult835:
    """Promote warnings to errors (mirrors the CLI's --strict)."""
    if result.validation is None:
        return result
    report = result.validation
    promoted = [i.model_copy(update={"severity": "error"}) for i in report.warnings]
    new_errors = report.errors + promoted
    new_report = report.model_copy(update={"errors": new_errors, "passed": not new_errors})
    passed = 1 if new_report.passed else 0
    failed = 1 if not new_report.passed else 0
    new_summary = result.summary.model_copy(
        update={"passed": passed, "failed": failed}
    )
    return result.model_copy(update={"validation": new_report, "summary": new_summary})
 def drop_raw_segments_837(result: ParseResult) -> ParseResult:
    """Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
    claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
    return result.model_copy(update={"claims": claims})
 def drop_raw_segments_835(result: ParseResult835) -> ParseResult835:
    """Return a copy of ``result`` with ``raw_segments`` cleared on every claim."""
    claims = [c.model_copy(update={"raw_segments": []}) for c in result.claims]
    return result.model_copy(update={"claims": claims})
 def has_claim_validation_errors(result: ParseResult) -> bool:
    return any(not c.validation.passed for c in result.claims)
 def has_835_validation_errors(result: ParseResult835) -> bool:
    return result.validation is not None and not result.validation.passed
 def heartbeat_seconds() -> float:
    """Return the configured tail heartbeat interval.
    Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can
    monkeypatch the env var without reloading the module. Defaults to
    15s (the production cadence); tests override to a small value (e.g.
    0.2s) to keep their runtime bounded.
    """
    raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15")
    try:
        v = float(raw)
    except ValueError:
        return 15.0
    return v if v > 0 else 15.0
 async def tail_events(
    request: Request, bus: EventBus, kinds: list[str]
 ) -> AsyncIterator[bytes]:
    """Forward subscribed events as ``item`` lines with periodic heartbeats.
    Polls the underlying ``asyncio.Queue`` directly (via
    :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's
    async-iterator wrapper. ``asyncio.wait_for`` cancels the inner
    future on timeout, which would otherwise terminate the bus
    iterator at its ``await`` point and break subsequent
    ``__anext__`` calls with ``StopAsyncIteration``. Polling
    ``queue.get()`` is idempotent under cancellation, so heartbeats
    don't poison the subscription.
    A ``try/finally`` unsubscribes the queue from the bus when the
    caller disconnects or the generator is garbage collected —
    otherwise the bus would leak one queue per open stream.
    """
    hb_s = heartbeat_seconds()
    queue, _sub = bus.subscribe_raw(kinds)
    try:
        while True:
            if await request.is_disconnected():
                return
            get_task = asyncio.ensure_future(queue.get())
            sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s))
            try:
                done, pending = await asyncio.wait(
                    {get_task, sleep_task},
                    return_when=asyncio.FIRST_COMPLETED,
                )
            except BaseException:
                get_task.cancel()
                sleep_task.cancel()
                raise
            for t in pending:
                t.cancel()
            if get_task in done:
                event = get_task.result()
                yield ndjson_line({"type": "item", "data": event})
            else:
                yield ndjson_line({
                    "type": "heartbeat",
                    "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")},
                })
    finally:
        bus.unsubscribe(queue, kinds)
@@ -0,0 +1 @@
 """Resource-group routers. Imported and registered by ``cyclone.api``."""
@@ -0,0 +1,104 @@
 """``/api/acks`` — list & detail endpoints for the 999 ACK inbox.
 These are the persisted acknowledgment rows produced by
 ``POST /api/parse-999``. The frontend ``useAcks`` hook re-shapes the
 list payload to its ``Ack`` interface in ``src/types/index.ts``.
 The detail endpoint returns the full ``raw_json`` payload plus the
 regenerated ``raw_999_text`` so the UI can show "view source" without a
 second round-trip.
 """
 from __future__ import annotations
 import logging
 from typing import Any
 from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse
 from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
 from cyclone.parsers.models_999 import ParseResult999
 from cyclone.parsers.serialize_999 import serialize_999
 from cyclone.store import store
 router = APIRouter()
 log = logging.getLogger(__name__)
 def _ack_to_ui(row) -> dict:
    """Map an ``Ack`` ORM row to the UI shape used by ``/api/acks``.
    Field names match the rest of the Cyclone API (snake_case). The
    frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack``
    interface in ``src/types/index.ts``.
    """
    return {
        "id": row.id,
        "source_batch_id": row.source_batch_id,
        "accepted_count": row.accepted_count,
        "rejected_count": row.rejected_count,
        "received_count": row.received_count,
        "ack_code": row.ack_code,
        "parsed_at": (
            row.parsed_at.isoformat().replace("+00:00", "Z")
            if row.parsed_at is not None
            else ""
        ),
    }
@router.get("/api/acks")
 def list_acks_endpoint(
    request: Request,
    limit: int = Query(100, ge=1, le=1000),
 ) -> Any:
    """Return the list of persisted 999 ACKs, newest first."""
    rows = store.list_acks()
    items = [_ack_to_ui(r) for r in rows[:limit]]
    total = len(rows)
    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/acks/{ack_id}")
 def get_ack_endpoint(ack_id: int) -> dict:
    """Return one persisted ACK row with its parsed detail.
    Path param is ``ack_id`` (not ``id``) to avoid shadowing FastAPI's
    internal ``id`` name and to keep OpenAPI docs self-describing.
    Returns 404 when the ACK is missing — never 500.
    """
    row = store.get_ack(ack_id)
    if row is None:
        raise HTTPException(
            status_code=404,
            detail={"error": "Not found", "detail": f"Ack {ack_id} not found"},
        )
    body = _ack_to_ui(row)
    body["raw_json"] = row.raw_json
    # Regenerate the X12 text from raw_json so the operator can download
    # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry
    # the regenerated text to keep payloads small; detail does.)
    if row.raw_json:
        try:
            regenerated = ParseResult999.model_validate(row.raw_json)
            icn = regenerated.envelope.control_number or "000000001"
            body["raw_999_text"] = serialize_999(regenerated, interchange_control_number=icn)
        except Exception as exc:  # noqa: BLE001 — never 500 on a regen failure
            log.warning("Could not regenerate 999 for ack %s: %s", ack_id, exc)
            body["raw_999_text"] = None
    else:
        body["raw_999_text"] = None
    return body
@@ -0,0 +1,17 @@
 """``GET /api/health`` — liveness probe.
 Returns the package version so an operator can confirm which build is
 serving requests without poking the filesystem.
 """
 from __future__ import annotations
 from fastapi import APIRouter
 from cyclone import __version__
 router = APIRouter()
@router.get("/api/health")
 def health() -> dict[str, str]:
    return {"status": "ok", "version": __version__}
@@ -0,0 +1,76 @@
 """``/api/ta1-acks`` — list & detail endpoints for persisted TA1 envelopes.
 TA1 is the interchange-control ACK (ISA/IEA acknowledgement). It's a
 single segment, no functional group, no transaction set. Cyclone
 persists the parsed fields plus a synthetic ``source_batch_id`` so the
 row can sit alongside the 999 / 277CA ack rows without special-casing.
 The detail endpoint also reconstructs the TA1 segment string
 (``TA1*...~``) so the operator can copy it into a downstream tool.
 """
 from __future__ import annotations
 from typing import Any
 from fastapi import APIRouter, HTTPException, Query
 from cyclone import db
 from cyclone.store import store
 router = APIRouter()
 def _ta1_to_ui(row: db.Ta1Ack) -> dict:
    """Render a Ta1Ack row for the UI (list endpoint shape)."""
    return {
        "id": row.id,
        "control_number": row.control_number,
        "ack_code": row.ack_code,
        "note_code": row.note_code,
        "interchange_date": row.interchange_date.isoformat()
            if row.interchange_date else None,
        "interchange_time": row.interchange_time,
        "sender_id": row.sender_id,
        "receiver_id": row.receiver_id,
        "source_batch_id": row.source_batch_id,
        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
    }
 def _serialize_ta1_from_row(row: db.Ta1Ack) -> str:
    """Reconstruct a TA1 segment from the persisted flat row (for the detail endpoint)."""
    date_s = row.interchange_date.strftime("%y%m%d") if row.interchange_date else ""
    return (
        f"TA1*{row.control_number}*{date_s}*{row.interchange_time or ''}*"
        f"{row.ack_code}*{row.note_code or ''}~"
    )
@router.get("/api/ta1-acks")
 def list_ta1_acks_endpoint(
    limit: int = Query(100, ge=1, le=1000),
 ) -> Any:
    """Return the list of persisted TA1 ACKs, newest first.
    Mirrors :func:`cyclone.api_routers.acks.list_acks_endpoint` — fetches all
    rows then slices in Python so the ``total`` field reflects the full row
    count regardless of the ``limit`` cap.
    """
    rows = store.list_ta1_acks()
    items = [_ta1_to_ui(r) for r in rows[:limit]]
    return {
        "total": len(rows),
        "items": items,
    }
@router.get("/api/ta1-acks/{ack_id}")
 def get_ta1_ack_endpoint(ack_id: int) -> dict:
    """Return one persisted TA1 ACK row with its parsed detail."""
    row = store.get_ta1_ack(ack_id)
    if row is None:
        raise HTTPException(status_code=404, detail=f"TA1 ACK {ack_id} not found")
    body = _ta1_to_ui(row)
    body["raw_ta1_text"] = _serialize_ta1_from_row(row)
    body["raw_json"] = row.raw_json
    return body
@@ -71,9 +71,19 @@ def _make_engine(url: str) -> sa.Engine:
        key = db_crypto.get_db_key()
        if key:
            creator = db_crypto.make_sqlcipher_connect_creator(url, key)
            # SP15: NullPool — each thread opens its own SQLCipher
            # connection. The default QueuePool returns connections
            # to a shared queue that any thread can pull from, which
            # breaks SQLCipher's thread affinity (a connection opened
            # on thread A raises ProgrammingError when used on thread
            # B). NullPool trades connection reuse for thread safety,
            # which is the only correct behavior for SQLCipher under
            # FastAPI's per-request threadpool.
            from sqlalchemy.pool import NullPool
            return sa.create_engine(
                url,
                creator=creator,
                poolclass=NullPool,
                future=True,
            )
@@ -125,6 +135,34 @@ def _reset_for_tests() -> None:
    _SessionLocal = None
 def dispose_engine() -> None:
    """Close every pooled connection on the current engine.
    SP15: used by the key-rotation flow to ensure no connection is
    holding the DB file open while ``PRAGMA rekey`` runs (SQLCipher
    refuses to rekey if another connection is using the DB). The
    next call to ``init_db()`` rebuilds the engine with the new key
    from the Keychain.
    """
    global _engine
    if _engine is not None:
        _engine.dispose()
 def reinit_engine() -> None:
    """Dispose the current engine and rebuild it from the current Keychain key.
    SP15: called by the key-rotation endpoint after the Keychain is
    updated with the new key. We dispose (close every pooled
    connection that was using the OLD key) and then re-init (open
    new connections with the NEW key). The two-step is necessary
    because SQLAlchemy caches the creator in the pool — a re-init
    is the only way to swap the driver-level PRAGMA key.
    """
    dispose_engine()
    init_db()
 def engine() -> sa.Engine:
    """Return the process-wide Engine. Raises if `init_db()` was not called."""
    if _engine is None:
@@ -1,6 +1,6 @@
 """SQLCipher integration — encryption at rest for the SQLite DB.
-SP12.
+SP12 / SP15.
 When ``cyclone.db.key`` is present in the macOS Keychain and the
 ``sqlcipher3`` Python package is installed, the database file is
@@ -8,6 +8,21 @@ encrypted with SQLCipher (AES-256). Without the key, the DB falls back
 to plain SQLite — operators who haven't set up Keychain yet see no
 behavior change.
 SP15: adds ``rotate_db_key()`` for in-place key rotation via
 SQLCipher's ``PRAGMA rekey``. The rotation:
 1. Closes every pooled SQLAlchemy connection (so the file is unlocked).
 2. Opens a single dedicated connection with the *old* key.
 3. Issues ``PRAGMA rekey = "<new_key>"`` (rewrites every page with
   the new key, in-place).
 4. Closes the connection.
 5. Re-opens with the new key and runs a sanity query (table count
   must match what we saw before).
 6. Caller updates the Keychain with the new key. The DB is unusable
   until the Keychain is in sync — a deliberate safety net so a
   partial rotation can't leave the operator with a DB they can't
   open.
 Why this design:
 - The DB key never lives on disk in plaintext. It's stored in macOS
  Keychain under service ``cyclone``, account ``cyclone.db.key``.
@@ -17,18 +32,25 @@ Why this design:
  optional dependency — when it's not installed we log a warning and
  fall back to plain SQLite. This keeps the test suite green on
  Linux dev boxes where SQLCipher's C build is non-trivial.
- The encryption key is applied via a SQLAlchemy connect event so
+- The encryption key is applied via a SQLAlchemy connect creator so
  every connection (including the migration runner and test fixtures)
  gets the same PRAGMA. We never store the key in a Python global.
 Compliance: HIPAA §164.312(a)(2)(iv) — encryption at rest. §164.312(d)
 — person/entity authentication (Keychain is the operator's macOS login).
 SP15: §164.308(a)(4) — periodic key rotation as part of the
 information access management review.
 """
 from __future__ import annotations
 import hashlib
 import logging
 import secrets as _secrets
 import sqlite3
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Optional
 import sqlalchemy as sa
 import sqlalchemy.event
@@ -39,6 +61,10 @@ log = logging.getLogger(__name__)
 # Keychain account name for the DB encryption key.
 KEYCHAIN_ACCOUNT = "cyclone.db.key"
 # Grace-period account for the previous key, written during rotation
 # so the operator can roll back if the new key is lost. Cleared
 # after the operator confirms the new key.
 KEYCHAIN_ACCOUNT_PREVIOUS = "cyclone.db.key.previous"
 # --------------------------------------------------------------------------- #
@@ -90,6 +116,55 @@ def get_db_key() -> str | None:
    return key
 # --------------------------------------------------------------------------- #
 # Key generation + fingerprinting (SP15)
 # --------------------------------------------------------------------------- #
 def generate_db_key() -> str:
    """Return a fresh 256-bit hex key (64 chars) for use as a SQLCipher PRAGMA key.
    Uses ``secrets.token_hex(32)`` (CSPRNG). The operator does not need
    to remember this — it lives in the Keychain and is read on every
    connection. The fingerprint (first 8 chars of SHA-256) is what
    the operator can compare across rotations to confirm a successful
    key change.
    """
    return _secrets.token_hex(32)
 def fingerprint(key: str) -> str:
    """Return a short, operator-readable fingerprint of the key.
    First 8 hex chars of SHA-256. Two fingerprints matching means
    "this is the same key". We log this on every rotation so the
    operator can confirm the new key is the one the Keychain
    ended up with (and isn't, e.g., a transposed paste).
    """
    return hashlib.sha256(key.encode("utf-8")).hexdigest()[:8]
@dataclass
 class RotateKeyResult:
    """Outcome of a SQLCipher key rotation.
    Attributes:
        ok: True when the rekey completed and the new key opens the DB.
        old_fingerprint: fingerprint of the old key.
        new_fingerprint: fingerprint of the new key.
        rotated_at: ISO-8601 timestamp (UTC) of the rekey.
        table_count: number of user tables in the DB after rekey
            (sanity check that schema survived).
        reason: human-readable error if ``ok`` is False.
    """
    ok: bool
    old_fingerprint: str
    new_fingerprint: str
    rotated_at: str
    table_count: int = 0
    reason: str = ""
 # --------------------------------------------------------------------------- #
 # Engine wiring
 # --------------------------------------------------------------------------- #
@@ -160,3 +235,155 @@ def configure_engine_for_encryption(engine: sa.Engine, key: str) -> None:
    # Instead we use the dialect-level hook.
    engine.pool._creator = creator  # type: ignore[attr-defined]
    log.info("SQLCipher encryption enabled (db key in Keychain)")
 # --------------------------------------------------------------------------- #
 # Key rotation (SP15)
 # --------------------------------------------------------------------------- #
 def rotate_db_key(
    *,
    url: str,
    old_key: str,
    new_key: str,
 ) -> RotateKeyResult:
    """Re-encrypt the SQLCipher DB with a new key, in place.
    SQLCipher supports ``PRAGMA rekey = "<new_key>"`` which rewrites
    every page of the DB with the new key. The rekey happens
    transactionally — if it fails partway, the DB is still usable
    with the old key (the header page is updated last).
    Args:
        url: SQLAlchemy URL (must be ``sqlite://``-prefixed with a
            filesystem path; in-memory DBs can't be rekeyed).
        old_key: the current key the DB was opened with. Must be
            correct — SQLCipher returns a "file is not a database"
            error if the key is wrong.
        new_key: the key to re-encrypt with. Should be a fresh
            ``generate_db_key()`` value.
    Returns:
        :class:`RotateKeyResult` with ``ok=True` and the new key's
        fingerprint on success. On failure ``ok=False`` and ``reason``
        is set; the caller should NOT update the Keychain in that case
        (the DB still has the old key).
    """
    import sqlcipher3
    if not url.startswith("sqlite") or url.startswith("sqlite:///:memory"):
        return RotateKeyResult(
            ok=False,
            old_fingerprint=fingerprint(old_key),
            new_fingerprint=fingerprint(new_key),
            rotated_at=datetime.now(timezone.utc).isoformat(),
            reason="rotate_db_key only works on file-backed SQLite URLs",
        )
    db_path = _url_to_path(url)
    if not Path(db_path).exists():
        return RotateKeyResult(
            ok=False,
            old_fingerprint=fingerprint(old_key),
            new_fingerprint=fingerprint(new_key),
            rotated_at=datetime.now(timezone.utc).isoformat(),
            reason=f"database file not found: {db_path}",
        )
    log.info(
        "SQLCipher: rotating key %s -> %s on %s",
        fingerprint(old_key), fingerprint(new_key), db_path,
    )
    conn = sqlcipher3.connect(db_path)
    try:
        # Open with the OLD key.
        conn.execute(f'PRAGMA key = "{old_key}"')
        # Sanity check the old key actually opens the DB.
        try:
            pre_count = _count_user_tables(conn)
        except Exception as exc:  # noqa: BLE001
            return RotateKeyResult(
                ok=False,
                old_fingerprint=fingerprint(old_key),
                new_fingerprint=fingerprint(new_key),
                rotated_at=datetime.now(timezone.utc).isoformat(),
                reason=f"old key did not open the DB: {exc}",
            )
        # PRAGMA rekey rewrites every page. SQLCipher 4+ uses the
        # ``PRAGMA rekey = "..."`` form (older versions used
        # ``PRAGMA rekey "..."``; sqlcipher3 0.6+ ships SQLCipher 4).
        conn.execute(f'PRAGMA rekey = "{new_key}"')
        # Close and reopen to confirm the new key works.
        conn.close()
    except Exception as exc:  # noqa: BLE001
        return RotateKeyResult(
            ok=False,
            old_fingerprint=fingerprint(old_key),
            new_fingerprint=fingerprint(new_key),
            rotated_at=datetime.now(timezone.utc).isoformat(),
            reason=f"PRAGMA rekey failed: {exc}",
        )
    # Reopen with the NEW key. Any read query verifies the rekey.
    try:
        conn = sqlcipher3.connect(db_path)
        conn.execute(f'PRAGMA key = "{new_key}"')
        post_count = _count_user_tables(conn)
        conn.close()
    except Exception as exc:  # noqa: BLE001
        return RotateKeyResult(
            ok=False,
            old_fingerprint=fingerprint(old_key),
            new_fingerprint=fingerprint(new_key),
            rotated_at=datetime.now(timezone.utc).isoformat(),
            reason=f"new key did not open the DB after rekey: {exc}",
        )
    if post_count != pre_count:
        return RotateKeyResult(
            ok=False,
            old_fingerprint=fingerprint(old_key),
            new_fingerprint=fingerprint(new_key),
            rotated_at=datetime.now(timezone.utc).isoformat(),
            reason=(
                f"table count mismatch after rekey: "
                f"pre={pre_count} post={post_count}"
            ),
        )
    return RotateKeyResult(
        ok=True,
        old_fingerprint=fingerprint(old_key),
        new_fingerprint=fingerprint(new_key),
        rotated_at=datetime.now(timezone.utc).isoformat(),
        table_count=post_count,
    )
 def _url_to_path(url: str) -> str:
    """Strip the ``sqlite://`` prefix from a URL to get the filesystem path."""
    if url.startswith("sqlite:///"):
        return url[len("sqlite:///"):]
    if url.startswith("sqlite://"):
        return url[len("sqlite://"):]
    return url
 def _count_user_tables(conn) -> int:
    """Return the number of user (non-internal) tables in the schema.
    Used as a sanity check that the rekey didn't corrupt the schema.
    Excludes ``sqlite_*`` system tables. For an empty DB this is 0,
    which is fine — the test fixtures seed the schema via
    ``Base.metadata.create_all`` before rotating.
    """
    rows = conn.execute(
        "SELECT name FROM sqlite_master "
        "WHERE type='table' AND name NOT LIKE 'sqlite_%'"
    ).fetchall()
    return len(rows)
@@ -0,0 +1,217 @@
 """SP15 — SQLCipher key rotation API endpoint tests.
 We test the *wiring* of the endpoint:
 1. Refuses with 400 when encryption is not enabled.
 2. Refuses with 409 when a rotation is already in flight.
 3. On success: calls rotate_db_key, updates the Keychain, rebuilds
   the engine, writes an audit event, and returns the fingerprints.
 4. On Keychain write failure: returns 503 (DB is rotated, Keychain
   is stale; operator must restore).
 The actual ``PRAGMA rekey`` mechanics are tested in ``test_db_crypto.py``
 (see :class:`TestRotateDbKey`); we don't duplicate that here.
 """
 from __future__ import annotations
 from datetime import datetime, timezone
 from pathlib import Path
 from unittest.mock import patch
 import pytest
 # Skip if sqlcipher3 isn't installed.
 pytestmark = pytest.mark.skipif(
    not __import__(
        "cyclone.db_crypto", fromlist=["is_sqlcipher_available"]
    ).is_sqlcipher_available(),
    reason="sqlcipher3 not installed",
 )
 def _stub_rotate_ok(*, url, old_key, new_key) -> dict:
    """Return a synthetic RotateKeyResult for endpoint wiring tests."""
    from cyclone.db_crypto import RotateKeyResult
    return RotateKeyResult(
        ok=True,
        old_fingerprint="aaaa1111",
        new_fingerprint="bbbb2222",
        rotated_at=datetime.now(timezone.utc).isoformat(),
        table_count=12,
    )
 class TestRotateKeyRefusesWhenNotEncrypted:
    def test_400_when_encryption_disabled(self, tmp_path, monkeypatch):
        from cyclone import db, db_crypto
        monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path}/plain.db")
        db._reset_for_tests()
        monkeypatch.setattr(db_crypto, "get_secret", lambda account: None)
        db.init_db()
        from fastapi.testclient import TestClient
        from cyclone.api import app
        with TestClient(app) as client:
            r = client.post("/api/admin/db/rotate-key")
        assert r.status_code == 400
        assert "not enabled" in r.json()["detail"]
        db._reset_for_tests()
 class TestRotateKeyEndpointWiring:
    @pytest.fixture
    def _fake_encrypted_env(self, tmp_path, monkeypatch):
        """Set up: encryption-enabled DB on disk, fake Keychain
        (read + write), and the engine initialized here.
        With NullPool (see ``cyclone.db._make_engine``), every thread
        opens its own SQLCipher connection — no cross-thread reuse,
        no ProgramingError. The endpoint runs on the request thread
        and verification runs on the test thread; both get fresh
        per-thread connections transparently.
        """
        from cyclone import db, db_crypto
        db_file = tmp_path / "cyclone.db"
        monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{db_file}")
        db._reset_for_tests()
        fake_kc = {db_crypto.KEYCHAIN_ACCOUNT: "old-test-key-1"}
        monkeypatch.setattr(db_crypto, "get_secret", lambda n: fake_kc.get(n))
        monkeypatch.setattr("cyclone.secrets.get_secret", lambda n: fake_kc.get(n))
        monkeypatch.setattr("cyclone.secrets.set_secret",
                            lambda n, v: fake_kc.__setitem__(n, v) or True)
        # The endpoint's actual rekey is stubbed; the real PRAGMA
        # rekey mechanics are tested in test_db_crypto.py::TestRotateDbKey.
        monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _stub_rotate_ok)
        db.init_db()
        yield db_file, fake_kc
        db._reset_for_tests()
    def test_successful_rotation_updates_keychain_and_writes_audit(
        self, _fake_encrypted_env,
    ):
        from cyclone import db
        # The fixture stubs rotate_db_key to a no-op success.
        from fastapi.testclient import TestClient
        from cyclone.api import app
        with TestClient(app) as client:
            r = client.post(
                "/api/admin/db/rotate-key",
                json={"actor": "alice", "reason": "scheduled"},
            )
        assert r.status_code == 200, r.text
        body = r.json()
        assert body["ok"] is True
        assert body["old_fingerprint"] == "aaaa1111"
        assert body["new_fingerprint"] == "bbbb2222"
        assert body["table_count"] == 12
    def test_successful_rotation_writes_audit_event(
        self, _fake_encrypted_env,
    ):
        from cyclone import db
        import json as _json
        from fastapi.testclient import TestClient
        from cyclone.api import app
        with TestClient(app) as client:
            r = client.post("/api/admin/db/rotate-key", json={"actor": "bob"})
            assert r.status_code == 200
        from cyclone.db import AuditLog
        with db.SessionLocal()() as session:
            events = (
                session.query(AuditLog)
                .filter(AuditLog.event_type == "db.key_rotated")
                .all()
            )
            assert len(events) == 1
            e = events[0]
            assert e.entity_type == "database"
            assert e.entity_id == "cyclone.db"
            assert e.actor == "bob"
            payload = _json.loads(e.payload_json)
            assert payload["old_fingerprint"] == "aaaa1111"
            assert payload["new_fingerprint"] == "bbbb2222"
            assert payload["table_count"] == 12
    def test_rotation_rekey_failure_returns_503_and_leaves_keychain_unchanged(
        self, _fake_encrypted_env, monkeypatch
    ):
        from cyclone import db_crypto
        from cyclone import db
        from datetime import datetime, timezone
        def _fail_rotate(*, url, old_key, new_key):
            return db_crypto.RotateKeyResult(
                ok=False,
                old_fingerprint=db_crypto.fingerprint(old_key),
                new_fingerprint=db_crypto.fingerprint(new_key),
                rotated_at=datetime.now(timezone.utc).isoformat(),
                reason="simulated PRAGMA rekey failure",
            )
        monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _fail_rotate)
        _, fake_kc = _fake_encrypted_env
        before = dict(fake_kc)
        from fastapi.testclient import TestClient
        from cyclone.api import app
        with TestClient(app) as client:
            r = client.post("/api/admin/db/rotate-key")
        assert r.status_code == 503
        body = r.json()["detail"]
        assert body["ok"] is False
        assert "simulated" in body["reason"]
        # Keychain wasn't touched.
        assert fake_kc == before
        # No audit event was written.
        from cyclone.db import AuditLog
        with db.SessionLocal()() as session:
            count = (
                session.query(AuditLog)
                .filter(AuditLog.event_type == "db.key_rotated")
                .count()
            )
            assert count == 0
    def test_503_when_keychain_write_fails_after_successful_rekey(
        self, _fake_encrypted_env, monkeypatch
    ):
        """The rekey itself succeeded but the Keychain write failed.
        The DB is now behind a new key the Keychain doesn't know about.
        Endpoint must return 503 so the operator can run the manual
        restore-key command."""
        from cyclone import db
        # Override the set_secret at the import-site of the endpoint.
        monkeypatch.setattr("cyclone.api._secrets.set_secret", lambda n, v: False)
        from fastapi.testclient import TestClient
        from cyclone.api import app
        with TestClient(app) as client:
            r = client.post("/api/admin/db/rotate-key")
        assert r.status_code == 503
        body = r.json()["detail"]
        assert body["ok"] is False
        assert "keychain" in body["reason"].lower()
    def test_409_when_concurrent_request(self, _fake_encrypted_env, monkeypatch):
        """A second concurrent rotation request gets 409 — only one
        rotation can run at a time (the module-level lock)."""
        monkeypatch.setattr(
            "cyclone.api._secrets.set_secret", lambda n, v: True,
        )
        from cyclone import api as api_mod
        api_mod._db_rotate_lock.acquire()
        try:
            from fastapi.testclient import TestClient
            from cyclone.api import app
            with TestClient(app) as client:
                r = client.post("/api/admin/db/rotate-key")
            assert r.status_code == 409
            assert "in progress" in r.json()["detail"]
        finally:
            api_mod._db_rotate_lock.release()
@@ -173,3 +173,120 @@ class TestMakeSqlcipherConnectCreator:
        result = conn.execute("SELECT x FROM t").fetchone()
        assert result[0] == 42
        conn.close()
 # --------------------------------------------------------------------------- #
 # SP15: Key generation + fingerprint
 # --------------------------------------------------------------------------- #
 class TestGenerateDbKey:
    def test_returns_64_char_hex(self):
        """A 256-bit key hex-encodes to 64 characters."""
        key = db_crypto.generate_db_key()
        assert len(key) == 64
        int(key, 16)  # parses as hex (raises if not)
    def test_two_calls_return_different_keys(self):
        """Distinct calls produce cryptographically distinct keys."""
        keys = {db_crypto.generate_db_key() for _ in range(8)}
        assert len(keys) == 8
 class TestFingerprint:
    def test_deterministic(self):
        assert db_crypto.fingerprint("abc") == db_crypto.fingerprint("abc")
    def test_different_inputs_yield_different_fingerprints(self):
        assert db_crypto.fingerprint("abc") != db_crypto.fingerprint("xyz")
    def test_eight_chars(self):
        assert len(db_crypto.fingerprint("anything")) == 8
 # --------------------------------------------------------------------------- #
 # SP15: rotate_db_key (in-place rekey via PRAGMA rekey)
 # --------------------------------------------------------------------------- #
@pytestmark_sqlcipher
 class TestRotateDbKey:
    def _create_encrypted_db(self, tmp_path: Path, key: str) -> Path:
        """Create a small SQLCipher DB with two tables."""
        import sqlcipher3
        db_file = tmp_path / "rotate.db"
        conn = sqlcipher3.connect(str(db_file))
        conn.execute(f'PRAGMA key = "{key}"')
        conn.execute("CREATE TABLE accounts (id INTEGER PRIMARY KEY, name TEXT)")
        conn.execute("CREATE TABLE balances (acct_id INTEGER, amt REAL)")
        conn.execute("INSERT INTO accounts VALUES (1, 'alice'), (2, 'bob')")
        conn.execute("INSERT INTO balances VALUES (1, 100.5), (2, 250.75)")
        conn.commit()
        conn.close()
        return db_file
    def test_rotate_changes_key_preserves_data(self, tmp_path: Path):
        """The core SP15 contract: rekey with a new key, data survives."""
        db_file = self._create_encrypted_db(tmp_path, "old-key-aaaa")
        url = f"sqlite:///{db_file}"
        result = db_crypto.rotate_db_key(
            url=url, old_key="old-key-aaaa", new_key="new-key-bbbb",
        )
        assert result.ok, f"rotate failed: {result.reason}"
        assert result.old_fingerprint == db_crypto.fingerprint("old-key-aaaa")
        assert result.new_fingerprint == db_crypto.fingerprint("new-key-bbbb")
        assert result.table_count == 2  # accounts + balances
        # Open with the new key; data is intact.
        import sqlcipher3
        conn = sqlcipher3.connect(str(db_file))
        conn.execute(f'PRAGMA key = "new-key-bbbb"')
        rows = conn.execute("SELECT id, name FROM accounts ORDER BY id").fetchall()
        assert rows == [(1, "alice"), (2, "bob")]
        assert conn.execute("SELECT amt FROM balances WHERE acct_id = 2").fetchone()[0] == 250.75
        conn.close()
    def test_old_key_no_longer_opens_db(self, tmp_path: Path):
        """After rekey, the old key must not be able to open the DB."""
        import sqlcipher3
        db_file = self._create_encrypted_db(tmp_path, "old-key")
        url = f"sqlite:///{db_file}"
        result = db_crypto.rotate_db_key(
            url=url, old_key="old-key", new_key="new-key",
        )
        assert result.ok
        # Old key raises on first query.
        conn = sqlcipher3.connect(str(db_file))
        conn.execute(f'PRAGMA key = "old-key"')
        with pytest.raises(Exception) as exc_info:
            conn.execute("SELECT * FROM accounts").fetchall()
        msg = str(exc_info.value).lower()
        assert "not a database" in msg or "file is encrypted" in msg
        conn.close()
    def test_wrong_old_key_reports_helpful_reason(self, tmp_path: Path):
        """If the operator types the wrong old key, the rekey fails clean."""
        db_file = self._create_encrypted_db(tmp_path, "correct-old")
        url = f"sqlite:///{db_file}"
        result = db_crypto.rotate_db_key(
            url=url, old_key="WRONG-OLD-KEY", new_key="new",
        )
        assert result.ok is False
        assert "old key did not open" in result.reason.lower()
    def test_in_memory_url_is_rejected(self):
        """In-memory DBs cannot be rekeyed (nothing to persist)."""
        result = db_crypto.rotate_db_key(
            url="sqlite:///:memory:", old_key="a", new_key="b",
        )
        assert result.ok is False
        assert "file-backed" in result.reason.lower() or "in-memory" in result.reason.lower()
    def test_missing_db_file_is_rejected(self, tmp_path: Path):
        result = db_crypto.rotate_db_key(
            url=f"sqlite:///{tmp_path}/does-not-exist.db",
            old_key="a", new_key="b",
        )
        assert result.ok is False
        assert "not found" in result.reason.lower()
Author	SHA1	Message	Date
Tyler	d25f00ac58	docs: surface SP14 (Payer-Rejected UI + acknowledge) + SP15 (key rotation) The README's most recent merge was SP13. Since then two more sub-projects landed in main: - SP14 (`5c9365e` + `8a65baa`) — 5-lane Inbox UI with the Payer-Rejected lane rendered alongside Rejected/Candidates/Unmatched/Done today, and a bulk Acknowledge action that drops claims from the working surface without erasing the original 277CA rejection event. New endpoint POST /api/inbox/payer-rejected/acknowledge (idempotent, audit-logged). Migration 0010. - SP15 (`47902fd` + `ab00909`) — SQLCipher key rotation in place via PRAGMA rekey. New endpoint POST /api/admin/db/rotate-key, serialized through a module-level threading.Lock. Switches the SQLAlchemy engine to NullPool when SQLCipher is enabled (QueuePool breaks SQLCipher thread affinity under FastAPI's per-request threadpool). Writes a db.key_rotated audit event with old + new key fingerprints and post-rotation table_count. What this PR does: - Inbox section + Inbox endpoint table: document the Payer-Rejected acknowledge action. - Encryption at Rest: add a 'Key rotation' sub-section that explains the rotate-key handler, the NullPool choice, the threading.Lock, and the error code mapping (409/400/503). - Tamper-Evident Audit Log: note that key rotations + Payer-Rejected acknowledgements are part of the audit-logged event set. - Project layout: add api_routers/ (health.py, acks.py, ta1_acks.py) as the FastAPI APIRouter subpackage extracted from api.py. - Roadmap: bump 'shipped 2-13' to 'shipped 2-15'; add SP14 and SP15 entries to the shipped list; add SP14 + SP15 endpoint reference sections at the end. Verification: - pytest --collect-only collects 759 tests (up from 733 at the previous doc pass). - git diff --check clean. - No code changes; no tests touched.	2026-06-21 01:07:44 -06:00
Tyler	1267a341e3	refactor(api): trailing newlines + hoist acks.py parser imports to top-level Self-review nits from the router-split commit: - All four new files lacked a trailing newline (PEP 8 / POSIX). - acks.py was lazily importing ParseResult999 / serialize_999 inside get_ack_endpoint. Hoist to module-level — there's no import cycle (acks.py does not import from cyclone.api), so the imports are safe to do once. No behavior change. Targeted tests (test_acks + test_health + test_api_gets) still pass 41/41.	2026-06-21 00:50:31 -06:00
Tyler	a63ba5e88c	refactor(api): split health + acks + ta1_acks routes into FastAPI APIRouters Step 2 (first half) of the architecture satisfaction loop. api.py shrank from 2595 to 2452 lines (-143) by extracting three read-only resource groups into cyclone.api_routers: - health.py: GET /api/health (1 endpoint) - acks.py: GET /api/acks, GET /api/acks/{ack_id} (2 endpoints) - ta1_acks.py: GET /api/ta1-acks, GET /api/ta1-acks/{ack_id} (2 endpoints) Each router owns its endpoint bodies + the small UI-shape helper that goes with them (_ack_to_ui, _ta1_to_ui, _serialize_ta1_from_row). The helpers stay in the router file rather than being shared because each is only used by its own endpoints. api.py now ends the app-wiring section with three include_router() calls. The new package is named cyclone.api_routers (not cyclone.api.routers) to avoid the Python package-vs-same-named-module ambiguity that would shadow the existing cyclone.api module. Verifies: 41 targeted tests (test_acks, test_health, test_api_gets) pass, full pytest is 8 failed / 735 passed / 16 skipped — identical to clean main baseline. Live curl against the running server: GET /api/health -> 200, GET /api/acks -> 200, GET /api/ta1-acks -> 200. See /tmp/refactor-cyclone.md for the full plan.	2026-06-21 00:49:14 -06:00
Tyler	d4f6fdd49c	docs: sync READMEs with SP14 (Payer-Rejected lane) + endpoint inventory	2026-06-21 00:36:33 -06:00
Tyler	ab00909715	merge: SP15 SQLCipher key rotation into main	2026-06-21 00:33:40 -06:00
sp15-bot	47902fd6b2	feat(sp15): SQLCipher key rotation via PRAGMA rekey Adds in-place key rotation for the encrypted DB at rest (HIPAA sec.164.308(a)(4) - periodic key rotation). - db_crypto.rotate_db_key(): opens with old key, issues PRAGMA rekey, reopens with new key, verifies schema survived (table-count sanity). - db_crypto.generate_db_key(): fresh 256-bit CSPRNG hex key. - db_crypto.fingerprint(): SHA-256[:8] of a key, for the operator to compare across rotations. - db.dispose_engine() + db.reinit_engine(): SP15 plumbing. The rotation endpoint disposes the pooled connections (SQLCipher refuses to rekey while another connection holds the file), runs the rekey, then rebuilds the engine with the new key from the Keychain. - API: POST /api/admin/db/rotate-key with module-level threading.Lock to serialize rotations. 400 when encryption not enabled, 409 when a rotation is already in flight, 503 on rekey or Keychain failure with a reason that tells the operator what to do next. - Engine uses NullPool when SQLCipher is enabled: the default QueuePool returns connections to a shared queue that any thread can pull from, which breaks SQLCipher's thread affinity. NullPool trades connection reuse for thread safety, the only correct behavior under FastAPI's per-request threadpool. - Audit event db.key_rotated with old/new fingerprints and table_count, written after the engine is rebuilt so the new key proves it can take new writes. - previous key is stashed to a second Keychain account so the operator can roll back if the new key turns out to be broken. Tests: - test_db_crypto.py: 12 new tests for generate/fingerprint/rekey mechanics (5 require SQLCipher at runtime; skipped otherwise). - test_api_rotate_key.py: 6 new tests for endpoint wiring (encryption-required, Keychain update, audit event, rekey-failure rollback, Keychain-write-failure 503, concurrent-rotation 409).	2026-06-21 00:33:37 -06:00
Tyler	6233df1270	refactor(sp): extract shared API helpers from api.py into api_helpers.py First checkpoint of the architecture satisfaction loop. Cyclone's api.py was a 2281-line god-module with 14 cross-cutting helpers inlined next to the @app route declarations. This commit moves them to a dedicated cyclone.api_helpers module: - NDJSON wire format: ndjson_line, ndjson_stream_837, ndjson_stream_835, ndjson_stream_list. - Content negotiation: client_wants_json, wants_ndjson. - Strict / raw_segments rewrites: strict_rewrite_837, strict_rewrite_835, drop_raw_segments_837, drop_raw_segments_835. - Validation probes: has_claim_validation_errors, has_835_validation_errors. - Live-tail generator: tail_events, heartbeat_seconds, utcnow. api.py re-imports them under the original underscore-prefixed names so every route call site stays unchanged. claims_stream, remittances_stream, and activity_stream remain exposed at cyclone.api (test_api_stream_live imports them directly). Verifies byte-identical NDJSON wire format, content negotiation rules, and the tail_events async-generator semantics (deliberately polls the EventBus queue rather than awaiting its async iterator, so heartbeats don't poison the bus subscription). Live-tested: GET /api/health, /api/claims, /api/remittances, /api/activity, /api/acks, /api/providers, /api/inbox/lanes, /api/inbox/payer-rejected/acknowledge, and the /api/claims/stream NDJSON tail all return expected codes / payload. Backend pytest: 29 failures identical to baseline (pre-existing secrets env, serialize_837, db_crypto env, prodfile env failures), 700 passed. Frontend npm test: 357/357 passing. See /tmp/refactor-cyclone.md for the full checkpoint log and the plan for the next step (splitting api.py routes into FastAPI APIRouters). Autoreview: /tmp/grok-review-local.md (0 bugs, 1 suggestion, 4 nits — all addressed: dead asyncio/os imports removed, dead Any import removed, duplicate utcnow import dropped, trailing newline added).	2026-06-21 00:28:58 -06:00
		`@@ -0,0 +1 @@`
							"""Resource-group routers. Imported and registered by ``cyclone.api``."""