7 Commits

Author SHA1 Message Date
Tyler ea64e6e0f0 docs: list all 8 api_routers/ subpackages in Project layout
Four additional routers landed since the last doc pass:
- activity.py (GET /api/activity, /api/activity/stream)
- batches.py (GET /api/batches, /api/batches/{id}, /api/batch-diff)
- providers.py (GET /api/providers)
- clearhouse.py (GET /api/clearhouse, POST /api/clearhouse/submit)
- config.py (/api/config/providers[/...], /api/config/payers[/...],
  POST /api/admin/reload-config)

Update the Project layout block to list all 8 routers with the routes
they own. No new functionality is introduced — the refactor is purely
a code-organization move from api.py into the api_routers/ subpackage.
No code changes; no tests touched.
2026-06-21 03:05:52 -06:00
Tyler 6ce638553b refactor(api): split batches router (list summary + detail)
Extracts GET /api/batches and GET /api/batches/{batch_id} from api.py
into cyclone.api_routers.batches. The _batch_summary_claim_count
helper moves along (only used by list_batches).

api.py: 2257 -> 2201 (-56). Net diff: +93 / -64.
Pytest: 8 failed / 735 passed / 16 skipped — identical to baseline.
Live smoke: /api/batches 200 JSON; /api/batches/{id} 404 on missing;
NDJSON form returns the expected summary envelope.
2026-06-21 02:45:47 -06:00
Tyler e63be87ba9 refactor(api): split activity router (list + live-tail stream)
Extracts GET /api/activity and GET /api/activity/stream from api.py
into cyclone.api_routers.activity. The activity router is small and
self-contained: store.recent_activity() for the snapshot half,
tail_events() (from api_helpers) for the live tail.

activity_stream is re-exported at the cyclone.api module level so
test_api_stream_live.py's direct import keeps working.

api.py: 2310 -> 2257 (-53). Net diff: +110 / -61.
Pytest: 8 failed / 735 passed / 16 skipped — identical to baseline.
Live smoke: /api/activity 200 JSON; /api/activity/stream 200
application/x-ndjson with the expected snapshot lines.
2026-06-21 02:41:20 -06:00
Tyler 6aa440d3e4 refactor(api): trailing newlines + hoist clearhouse.py parser imports
Self-review nits from Checkpoint 2b:
- All three new router files lacked a trailing newline (same nit as 2a).
- clearhouse.py had lazy imports of serialize_837 / parse /
  db as _db inside helper bodies. Hoisted serialize_837 and
  parse to module-level (no import cycle risk: clearhouse.py does
  not import from cyclone.api). Dropped the redundant db as _db
  alias — the top-level from cyclone import db is already in scope.

No behavior change. test_clearhouse_api: 10 / 10 passing.
2026-06-21 02:30:11 -06:00
Tyler 45cafbdb32 refactor(api): split providers + clearhouse + config routers
Checkpoint 2b. Extracts 7 more endpoints (1 + 2 + 4) into three
new routers:

- providers.py: GET /api/providers (list distinct providers from
  claim rows; npi/state filter; NDJSON or paginated JSON).
- clearhouse.py: GET /api/clearhouse + POST /api/clearhouse/submit.
  Carries the two serialize-from-raw helpers (_serialize_claim_for_submit,
  _serialize_claim_from_raw) since they're only used here.
- config.py: GET /api/config/providers + GET /api/config/providers/{npi}
  + GET /api/config/payers + GET /api/config/payers/{payer_id}/configs.
  Payer configs endpoint merges YAML-loaded blocks with any DB-overridden
  live blocks per payer.

api.py shrank from 2474 to 2310 lines (-164) and no longer has any
single resource's full request/response shape — every remaining route
now lives in a dedicated router module.

Verifies:
- Full pytest: 8 failed / 735 passed / 16 skipped — identical to
  clean main baseline (the 8 are pre-existing env/secret/sqlcipher).
- Live smoke: /api/health, /api/providers, /api/clearhouse,
  /api/config/payers, /api/config/payers/co_medicaid/configs all 200.
  /api/config/providers/{npi-not-seeded} returns 404 as expected.

See /tmp/refactor-cyclone.md for the full plan and progress.
2026-06-21 02:28:56 -06:00
Tyler bbf89c9dd8 docs: add Batches, Reconciliation, and Activity endpoint reference
The README's SP-specific endpoint reference blocks (SP3-SP15 in the
Roadmap) cover the per-SP additions, but the pre-existing core operator
surface was never documented as a single block. This pass adds the
missing endpoints:

- GET /api/batches, GET /api/batches/{batch_id} — batch list + detail.
- GET /api/batch-diff?a=<id>&b=<id> — side-by-side diff between two
  batches (added/removed/changed claims + envelope metadata).
- GET /api/reconciliation/unmatched — every claim with no paired
  remit and every remit with no paired claim; powers the
  reconciliation review UI.
- POST /api/reconciliation/match — manual pair (claim_id, remit_id);
  400/404/409 contract.
- POST /api/reconciliation/unmatch — remove a match and reset the
  claim to 'submitted'.
- GET /api/providers — distinct providers from the parsed claim
  stream. Distinct from /api/config/providers/{npi} (the SP9 config
  table endpoint).
- GET /api/activity — recent activity events. Powers the Activity
  page; the streaming counterpart /api/activity/stream is already
  documented under 'Live updates'.

These 7 routes are referenced by the UI (BatchesList, BatchDetail,
BatchDiffView, Reconciliation page, Providers page, Activity page) but
were missing from the README's route inventory. The new section sits
between 'SFTP Wire-Up' and 'Persistence', with a one-paragraph pointer
to the per-SP endpoint reference blocks for SP-specific routes.

No code changes. No tests touched.
2026-06-21 02:07:27 -06:00
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
7 changed files with 538 additions and 297 deletions
+113 -4
View File
@@ -155,6 +155,7 @@ parses and rejects claims, the inbox reflects the new
| POST | `/api/inbox/candidates/{remit_id}/match` | Manual match. `409` if the claim state moved out from under us. | | 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/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/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. | | GET | `/api/inbox/export.csv?lane=<lane>` | Streams CSV of the lane's rows. |
## Outbound 837 Serializer ## Outbound 837 Serializer
@@ -321,8 +322,9 @@ both be true for the same claim.
The `audit_log` table is the canonical record of every state transition The `audit_log` table is the canonical record of every state transition
the system has ever observed — claim lifecycle, reconciliation the system has ever observed — claim lifecycle, reconciliation
decisions, config reloads, SFTP submissions, 277CA rejects. SP11 made decisions, config reloads, SFTP submissions, 277CA rejects, Payer-Rejected
it tamper-evident: every row carries a SHA-256 hash of 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. `(prev_hash || row_payload)`, forming a chain back to a genesis row.
Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable Any `INSERT`, `UPDATE`, or `DELETE` that breaks the chain is detectable
in a single walk. in a single walk.
@@ -357,6 +359,40 @@ operator has created the Keychain entry on first run. See
for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv) for the one-time setup recipe and the HIPAA Security Rule §164.312(a)(2)(iv)
mapping. 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) ## SFTP Wire-Up (paramiko)
The `clearhouse.submit` endpoint uses `paramiko` to push a batch of The `clearhouse.submit` endpoint uses `paramiko` to push a batch of
@@ -382,6 +418,29 @@ was a one-file change (`sftp_paramiko.py` replacing `sftp_stub.py`).
the `paramiko` extras aren't installed so the test suite stays green on the `paramiko` extras aren't installed so the test suite stays green on
Linux dev boxes. Linux dev boxes.
## Batches, Reconciliation, and Activity
These read/write endpoints are the core operator surface for browsing
the parsed record and acting on matches. They predate the per-SP
endpoint reference sections in the **Roadmap** and are listed here in
one place so the route inventory stays discoverable.
| Method | Path | Notes |
| ------ | --------------------------------------------- | ---------------------------------------------------------------- |
| GET | `/api/batches` | All parsed batches, newest first. `?limit=` (11000, default 100). |
| GET | `/api/batches/{batch_id}` | One batch detail (envelope, claims / remits, validation summary). |
| GET | `/api/batch-diff?a=<id>&b=<id>` | Side-by-side diff of two batches: `added` / `removed` / `changed` claims + envelope metadata for each. Both query params required. |
| GET | `/api/reconciliation/unmatched` | `{"claims": [...], "remittances": [...]}` — every Claim with no paired Remittance and vice versa. The two lists are always present (empty list, never absent) so the UI can index unconditionally. |
| POST | `/api/reconciliation/match` | Body `{claim_id, remit_id}`. Manually pair a claim with a remit. `400` on missing ids, `404` on unknown id, `409` on already-matched or terminal-state claim. |
| POST | `/api/reconciliation/unmatch` | Body `{claim_id}`. Remove the current match and reset the claim to `submitted`. `404` on unknown id, `409` on no current match. |
| GET | `/api/providers` | Distinct providers across parsed claims. `?npi=`, `?state=`, `?limit=`, `?offset=`. Distinct from `/api/config/providers/{npi}` (SP9 config table) — this endpoint surfaces providers derived from the parsed claim stream. |
| GET | `/api/activity` | Recent activity events. `?kind=`, `?since=`, `?limit=` (1500, default 200). Powers the Activity page; the streaming counterpart `/api/activity/stream` is documented under **Live updates**. |
The per-SP endpoint blocks at the bottom of the **Roadmap** cover the
SP-specific routes (parse, 999/TA1/277CA, Inbox, claim drawer, line
reconciliation, outbound 837, multi-payer config, audit log, 277CA,
key rotation, payer-rejected acknowledge).
## Persistence ## Persistence
Parsed batches, claims, remittances, matches, 277CA rejections, Parsed batches, claims, remittances, matches, 277CA rejections,
@@ -415,8 +474,17 @@ backup API).
. .
├── backend/ ├── backend/
│ ├── src/cyclone/ │ ├── 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_helpers.py # NDJSON / content-negotiation / live-tail helpers
│ │ ├── api_routers/ # FastAPI APIRouters extracted from api.py
│ │ │ ├── health.py # GET /api/health
│ │ │ ├── acks.py # GET /api/acks, /api/acks/{id}
│ │ │ ├── ta1_acks.py # GET /api/ta1-acks, /api/ta1-acks/{id}
│ │ │ ├── activity.py # GET /api/activity, /api/activity/stream
│ │ │ ├── batches.py # GET /api/batches, /api/batches/{id}, /api/batch-diff
│ │ │ ├── providers.py # GET /api/providers
│ │ │ ├── clearhouse.py # GET /api/clearhouse, POST /api/clearhouse/submit
│ │ │ └── config.py # /api/config/providers[/...], /api/config/payers[/...], /api/admin/reload-config
│ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out) │ │ ├── pubsub.py # in-process EventBus (drop-oldest, per-kind fan-out)
│ │ ├── store.py # CycloneStore, mappers, publish-on-write │ │ ├── store.py # CycloneStore, mappers, publish-on-write
│ │ ├── db.py # SQLAlchemy engine, session factory, ORM models │ │ ├── db.py # SQLAlchemy engine, session factory, ORM models
@@ -473,7 +541,7 @@ backup API).
## Roadmap ## Roadmap
Sub-projects 2 through 13 are **shipped**. See the [completeness Sub-projects 2 through 15 are **shipped**. See the [completeness
review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for review](docs/reviews/2026-06-20-cyclone-completeness-review.md) for
the honest gap analysis against the industry definition of a HIPAA the honest gap analysis against the industry definition of a HIPAA
clearinghouse — the short version is that the local-only, clearinghouse — the short version is that the local-only,
@@ -484,6 +552,19 @@ scope.
Shipped sub-projects (most recent first): 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 - **Sub-project 13 (shipped) — SFTP wire-up.** `paramiko`-backed
`SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint `SftpClient` replaces the SP9 stub. The clearhouse.submit endpoint
actually pushes to actually pushes to
@@ -752,6 +833,34 @@ the one-time setup recipe.
`mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`. `mft.gainwelltechnologies.com:/CO XIX/PROD/coxix_prod_11525703/FromHPE`.
SFTP credentials are fetched from the macOS Keychain at call time. 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 ## License
No license file yet; this is internal-use software. Add a `LICENSE` file No license file yet; this is internal-use software. Add a `LICENSE` file
+22 -293
View File
@@ -141,11 +141,31 @@ app.add_middleware(
# and are wired in here. (Kept as a top-level package rather than nested # 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 # under `cyclone.api` so the existing ``cyclone.api`` module path keeps
# working — Python prefers packages over same-named modules.) # working — Python prefers packages over same-named modules.)
from cyclone.api_routers import acks, health, ta1_acks # noqa: E402 from cyclone.api_routers import ( # noqa: E402
acks,
activity,
batches,
clearhouse,
config,
health,
providers,
ta1_acks,
)
app.include_router(health.router) app.include_router(health.router)
app.include_router(acks.router) app.include_router(acks.router)
app.include_router(ta1_acks.router) app.include_router(ta1_acks.router)
app.include_router(providers.router)
app.include_router(clearhouse.router)
app.include_router(config.router)
app.include_router(activity.router)
app.include_router(batches.router)
# Backwards-compat re-exports: test_api_stream_live.py imports these
# names directly from ``cyclone.api`` to invoke the endpoint coroutines
# in-process. Re-pointing at the router keeps the public surface stable
# while the bodies live in the routers package.
activity_stream = activity.activity_stream_endpoint
def _resolve_payer(name: str) -> PayerConfig: def _resolve_payer(name: str) -> PayerConfig:
@@ -1141,63 +1161,7 @@ def inbox_export_csv(lane: str):
) )
# --------------------------------------------------------------------------- # # --- /api/batches lives in cyclone.api_routers.batches ---
# GET endpoints (read views over the in-memory store)
# --------------------------------------------------------------------------- #
def _batch_summary_claim_count(rec: BatchRecord) -> int:
"""Return the number of claims on a batch, handling both 837P and 835."""
if rec.kind == "837p":
return len(rec.result.claims) # type: ignore[attr-defined]
if rec.kind == "835":
return len(rec.result.claims) # type: ignore[attr-defined]
return 0
@app.get("/api/batches")
def list_batches(
request: Request,
limit: int = Query(100, ge=1, le=1000),
) -> Any:
"""Summary of all parsed batches, newest first."""
records = store.list(limit=limit)
items = [
{
"id": r.id,
"kind": r.kind,
"inputFilename": r.input_filename,
"parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
"claimCount": _batch_summary_claim_count(r),
}
for r in records
]
all_records = store.all()
total = len(all_records)
returned = len(items)
has_more = total > returned
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": total,
"returned": returned,
"has_more": has_more,
}
@app.get("/api/batches/{batch_id}")
def get_batch(batch_id: str) -> Any:
rec = store.get(batch_id)
if rec is None:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
)
return json.loads(rec.result.model_dump_json())
@app.get("/api/claims") @app.get("/api/claims")
@@ -1803,99 +1767,8 @@ def get_remittance(remittance_id: str) -> dict:
return body return body
@app.get("/api/providers")
def list_providers(
request: Request,
npi: str | None = Query(None),
state: str | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
items = store.distinct_providers()
if npi is not None:
items = [p for p in items if p["npi"] == npi]
if state is not None:
items = [p for p in items if p.get("state") == state]
paged = items[offset:offset + limit]
total = len(items)
returned = len(paged)
has_more = total > offset + returned
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(paged, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": paged,
"total": total,
"returned": returned,
"has_more": has_more,
}
@app.get("/api/activity")
def list_activity(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(200, ge=1, le=500),
) -> Any:
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
total = len(events)
has_more = False
if _wants_ndjson(request):
return StreamingResponse(
_ndjson_stream_list(events, total, total, has_more),
media_type="application/x-ndjson",
)
return {
"items": events,
"total": total,
"returned": total,
"has_more": has_more,
}
@app.get("/api/activity/stream")
async def activity_stream(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(50, ge=1, le=500),
) -> StreamingResponse:
"""Stream Activity events as NDJSON: snapshot first, then live events.
Subscribes to ``activity_recorded``. Default ``limit`` is 50
(smaller than the list endpoint's 200) because activity is
high-volume — callers usually want the most recent handful, not a
full replay.
"""
bus: EventBus = request.app.state.event_bus
async def gen() -> AsyncIterator[bytes]:
# Snapshot reuses the same in-memory filter as ``list_activity``
# so the two endpoints are interchangeable for the snapshot
# half.
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
for ev in events:
yield _ndjson_line({"type": "item", "data": ev})
yield _ndjson_line({
"type": "snapshot_end", "data": {"count": len(events)},
})
async for chunk in _tail_events(request, bus, ["activity_recorded"]):
yield chunk
return StreamingResponse(gen(), media_type="application/x-ndjson")
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# 999 ACKs (read views) # 999 ACKs (read views)
@@ -2089,125 +1962,8 @@ async def post_eligibility_parse_271(
} }
# ---------------------------------------------------------------------------
# SP9: providers / payers / clearhouse endpoints
# ---------------------------------------------------------------------------
@app.get("/api/clearhouse")
def get_clearhouse():
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=404, detail="clearhouse not seeded")
return json.loads(ch.model_dump_json())
@app.post("/api/clearhouse/submit")
def submit_to_clearhouse(body: dict):
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
Stub behavior: serializes each claim via the SP7 serializer, builds
an HCPF-compliant outbound filename, and copies the result to
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
real SFTP connection. Returns a receipt per claim.
"""
from cyclone.clearhouse import make_client
from cyclone.edi.filenames import build_outbound_filename
claim_ids = body.get("claim_ids", [])
payer_id = body.get("payer_id")
if not claim_ids:
raise HTTPException(status_code=400, detail="claim_ids required")
if not payer_id:
raise HTTPException(status_code=400, detail="payer_id required")
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=500, detail="clearhouse not seeded")
client = make_client(ch.sftp_block)
results = []
for cid in claim_ids:
try:
x12_text = _serialize_claim_for_submit(cid)
except Exception as exc: # noqa: BLE001
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
continue
filename = build_outbound_filename(ch.tpid, "837P")
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
results.append({
"claim_id": cid,
"ok": True,
"filename": filename,
"staging_path": str(staging_path),
"remote_path": remote,
})
# SP11: audit trail for each successful clearhouse submission.
with db.SessionLocal()() as audit_s:
append_event(audit_s, AuditEvent(
event_type="clearhouse.submitted",
entity_type="claim",
entity_id=cid,
payload={
"filename": filename,
"remote_path": remote,
"tpid": ch.tpid,
"stub": ch.sftp_block.stub,
},
actor="clearhouse-submit",
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _serialize_claim_for_submit(claim_id: str) -> str:
"""Serialize a claim to X12 for SFTP submission. Lazy import of the
serializer to avoid pulling FastAPI machinery at module import time."""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone import db
with db.SessionLocal()() as s:
row = s.get(db.Claim, claim_id)
if row is None:
raise ValueError(f"claim {claim_id!r} not found")
# Re-parse the stored raw_json to get a ClaimOutput
from cyclone.parsers.models import ClaimOutput, Envelope, Subscriber, Payer, BillingProvider
from cyclone.parsers.parse_837 import parse
raw = row.raw_json or {}
# Reconstruct minimal ClaimOutput from raw_json; this is best-effort.
return _serialize_claim_from_raw(row, raw)
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
For SP9 this delegates to the existing serialize_837 helper if the
claim has a complete raw_segments array. Otherwise it returns a
minimal placeholder.
"""
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.parsers.parse_837 import parse
# Re-parse the original batch text (need to re-derive from store).
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
if isinstance(raw, dict) and raw.get("x12_text"):
result = parse(raw["x12_text"])
if result.claims:
return serialize_837(result.claims[0])
# Fallback: raise so the caller sees an error.
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
@app.get("/api/config/providers")
def list_configured_providers(is_active: bool | None = Query(default=True)):
"""List the configured provider rows (3 NPIs for SP9)."""
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
# --------------------------------------------------------------------------- # # --------------------------------------------------------------------------- #
# SP11: tamper-evident audit log (admin) # SP11: tamper-evident audit log (admin)
@@ -2430,35 +2186,8 @@ def rotate_db_key_endpoint(body: dict | None = None) -> Any:
_db_rotate_lock.release() _db_rotate_lock.release()
@app.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
if p is None:
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
return json.loads(p.model_dump_json())
@app.get("/api/config/payers")
def list_configured_payers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
@app.get("/api/config/payers/{payer_id}/configs")
def list_payer_configs(payer_id: str):
"""List all (transaction_type, config_json) blocks for a payer."""
from cyclone import payers as payer_loader
configs = [
{"transaction_type": tx, "config_json": block, "source": "yaml"}
for (pid, tx), block in payer_loader.all_configs().items()
if pid == payer_id
]
# Also check the DB for runtime-overridden configs
for tx in ("837P", "835", "277CA", "999", "TA1"):
live = store.get_payer_config(payer_id, tx)
if live is not None:
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
return configs
@app.post("/api/admin/reload-config") @app.post("/api/admin/reload-config")
def reload_config(): def reload_config():
@@ -0,0 +1,97 @@
"""``/api/activity`` — list & live-tail of recorded Activity events.
The Activity log is the cross-resource audit feed: every parser write,
inbox state change, match/dismiss/resubmit, clearhouse submission, etc.
records one row keyed by ``kind``. The list endpoint returns the most
recent ``limit`` events with optional ``kind`` / ``since`` filters; the
stream endpoint emits a snapshot followed by live updates from the
``EventBus``.
The default ``limit`` on the list endpoint is 200 (matches the prior
inline behavior); the stream endpoint defaults to 50 because activity is
high-volume callers usually want the most recent handful, not a full
replay.
"""
from __future__ import annotations
from typing import Any, AsyncIterator
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import (
ndjson_line,
ndjson_stream_list,
tail_events,
wants_ndjson,
)
from cyclone.pubsub import EventBus
from cyclone.store import store
router = APIRouter()
@router.get("/api/activity")
def list_activity_endpoint(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(200, ge=1, le=500),
) -> Any:
"""Return the recent Activity log, newest first.
Optional ``kind`` filters by ``event["kind"]`` (e.g. ``claim_submitted``).
Optional ``since`` is an inclusive ISO timestamp lower bound applied
to ``event["timestamp"]``.
"""
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
total = len(events)
has_more = False
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(events, total, total, has_more),
media_type="application/x-ndjson",
)
return {
"items": events,
"total": total,
"returned": total,
"has_more": has_more,
}
@router.get("/api/activity/stream")
async def activity_stream_endpoint(
request: Request,
kind: str | None = Query(None),
since: str | None = Query(None),
limit: int = Query(50, ge=1, le=500),
) -> StreamingResponse:
"""Stream Activity events as NDJSON: snapshot first, then live events.
Subscribes to ``activity_recorded``. The snapshot half reuses the
same in-memory filter as ``list_activity`` so the two endpoints are
interchangeable for the snapshot half.
"""
bus: EventBus = request.app.state.event_bus
async def gen() -> AsyncIterator[bytes]:
events = store.recent_activity(limit=limit)
if kind is not None:
events = [e for e in events if e["kind"] == kind]
if since is not None:
events = [e for e in events if e["timestamp"] >= since]
for ev in events:
yield ndjson_line({"type": "item", "data": ev})
yield ndjson_line({
"type": "snapshot_end", "data": {"count": len(events)},
})
async for chunk in tail_events(request, bus, ["activity_recorded"]):
yield chunk
return StreamingResponse(gen(), media_type="application/x-ndjson")
@@ -0,0 +1,75 @@
"""``/api/batches`` — list & detail endpoints for parsed batches.
Each ``store.add(...)`` call (837P, 835, 999, TA1, 277CA) produces a
``BatchRecord`` kept in memory by :class:`CycloneStore`. The list
endpoint returns a lightweight summary (id, kind, filename, parsedAt,
claim count) for the recent batches; the detail endpoint returns the
full parsed payload for one batch.
"""
from __future__ import annotations
import json
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.store import BatchRecord, store
router = APIRouter()
def _batch_summary_claim_count(rec: BatchRecord) -> int:
"""Return the number of claims on a batch, handling both 837P and 835."""
if rec.kind == "837p":
return len(rec.result.claims) # type: ignore[attr-defined]
if rec.kind == "835":
return len(rec.result.claims) # type: ignore[attr-defined]
return 0
@router.get("/api/batches")
def list_batches_endpoint(
request: Request,
limit: int = Query(100, ge=1, le=1000),
) -> Any:
"""Summary of all parsed batches, newest first."""
records = store.list(limit=limit)
items = [
{
"id": r.id,
"kind": r.kind,
"inputFilename": r.input_filename,
"parsedAt": r.parsed_at.isoformat().replace("+00:00", "Z"),
"claimCount": _batch_summary_claim_count(r),
}
for r in records
]
all_records = store.all()
total = len(all_records)
returned = len(items)
has_more = total > returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(items, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": items,
"total": total,
"returned": returned,
"has_more": has_more,
}
@router.get("/api/batches/{batch_id}")
def get_batch_endpoint(batch_id: str) -> Any:
"""Return the full parsed payload for one batch."""
rec = store.get(batch_id)
if rec is None:
raise HTTPException(
status_code=404,
detail={"error": "Not found", "detail": f"Batch {batch_id} not found"},
)
return json.loads(rec.result.model_dump_json())
@@ -0,0 +1,126 @@
"""``/api/clearhouse`` — SP9 single-payer SFTP submit surface.
Two endpoints:
- ``GET /api/clearhouse`` return the singleton clearhouse config
(dzinesco's identity, SFTP block, filename block). 404 when the
config hasn't been seeded yet.
- ``POST /api/clearhouse/submit`` submit a batch of claims to the
clearhouse (SFTP). SP9 ships a stub that copies to ``staging_dir``
instead of opening a real SFTP connection; SP13 wires the real
paramiko-backed client behind the same ``make_client()`` factory.
For each claim the endpoint serializes a fresh 837 via
:func:`_serialize_claim_for_submit`, builds an HCPF-compliant outbound
filename, and records an audit-log row on success. Failures are
captured per-claim so the operator sees exactly which claim_ids
didn't go through.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException
from cyclone import db
from cyclone.audit_log import AuditEvent, append_event
from cyclone.parsers.parse_837 import parse
from cyclone.parsers.serialize_837 import serialize_837
from cyclone.store import store
router = APIRouter()
@router.get("/api/clearhouse")
def get_clearhouse():
"""Return the singleton clearhouse config (dzinesco's identity, SFTP block, filename block)."""
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=404, detail="clearhouse not seeded")
return json.loads(ch.model_dump_json())
@router.post("/api/clearhouse/submit")
def submit_to_clearhouse(body: dict):
"""Submit a batch of claims to the clearhouse (SFTP). SP9: stub.
Body: ``{"claim_ids": [...], "payer_id": "CO_TXIX"}``
Stub behavior: serializes each claim via the SP7 serializer, builds
an HCPF-compliant outbound filename, and copies the result to
``{staging_dir}/{outbound_path}/{filename}`` instead of opening a
real SFTP connection. Returns a receipt per claim.
"""
from cyclone.clearhouse import make_client
from cyclone.edi.filenames import build_outbound_filename
claim_ids = body.get("claim_ids", [])
payer_id = body.get("payer_id")
if not claim_ids:
raise HTTPException(status_code=400, detail="claim_ids required")
if not payer_id:
raise HTTPException(status_code=400, detail="payer_id required")
ch = store.get_clearhouse()
if ch is None:
raise HTTPException(status_code=500, detail="clearhouse not seeded")
client = make_client(ch.sftp_block)
results = []
for cid in claim_ids:
try:
x12_text = _serialize_claim_for_submit(cid)
except Exception as exc: # noqa: BLE001
results.append({"claim_id": cid, "ok": False, "error": str(exc)})
continue
filename = build_outbound_filename(ch.tpid, "837P")
remote = f"{ch.sftp_block.paths['outbound']}/{filename}"
staging_path = client.write_file(remote, x12_text.encode("utf-8"))
results.append({
"claim_id": cid,
"ok": True,
"filename": filename,
"staging_path": str(staging_path),
"remote_path": remote,
})
# SP11: audit trail for each successful clearhouse submission.
with db.SessionLocal()() as audit_s:
append_event(audit_s, AuditEvent(
event_type="clearhouse.submitted",
entity_type="claim",
entity_id=cid,
payload={
"filename": filename,
"remote_path": remote,
"tpid": ch.tpid,
"stub": ch.sftp_block.stub,
},
actor="clearhouse-submit",
))
audit_s.commit()
return {"ok": True, "submitted": results, "stub": ch.sftp_block.stub}
def _serialize_claim_for_submit(claim_id: str) -> str:
"""Serialize a claim to X12 for SFTP submission."""
with db.SessionLocal()() as s:
row = s.get(db.Claim, claim_id)
if row is None:
raise ValueError(f"claim {claim_id!r} not found")
raw = row.raw_json or {}
return _serialize_claim_from_raw(row, raw)
def _serialize_claim_from_raw(claim_row, raw: dict) -> str:
"""Best-effort serializer that uses the stored raw_json to emit a fresh 837.
For SP9 this delegates to the existing serialize_837 helper if the
claim has a complete raw_segments array. Otherwise it raises.
"""
# SP9 stub: if the claim has a `raw_json` with `x12_text`, use that.
if isinstance(raw, dict) and raw.get("x12_text"):
result = parse(raw["x12_text"])
if result.claims:
return serialize_837(result.claims[0])
raise RuntimeError(
f"claim {claim_row.id!r} cannot be re-serialized: no stored x12_text"
)
+56
View File
@@ -0,0 +1,56 @@
"""``/api/config/*`` — configured-provider / configured-payer lookups.
The ``config`` prefix is the operator-facing read-only window into the
``config/payers.yaml`` static config and the seeded providers table.
Used by the frontend Settings page to show what's available.
The payers endpoint at ``/api/config/payers/{payer_id}/configs`` is
slightly different: it returns both the YAML-loaded config and any
DB-overridden config (so a runtime override wins over the static
fallback). 404 / 200 / array shape match the rest of the API.
"""
from __future__ import annotations
import json
from fastapi import APIRouter, HTTPException, Query
from cyclone import payers as payer_loader
from cyclone.store import store
router = APIRouter()
@router.get("/api/config/providers")
def list_configured_providers(is_active: bool | None = Query(default=True)):
"""List the configured provider rows (3 NPIs for SP9)."""
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
@router.get("/api/config/providers/{npi}")
def get_configured_provider(npi: str):
p = store.get_provider(npi)
if p is None:
raise HTTPException(status_code=404, detail=f"provider {npi!r} not found")
return json.loads(p.model_dump_json())
@router.get("/api/config/payers")
def list_configured_payers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_payers(is_active=is_active)]
@router.get("/api/config/payers/{payer_id}/configs")
def list_payer_configs(payer_id: str):
"""List all (transaction_type, config_json) blocks for a payer."""
configs = [
{"transaction_type": tx, "config_json": block, "source": "yaml"}
for (pid, tx), block in payer_loader.all_configs().items()
if pid == payer_id
]
# Also check the DB for runtime-overridden configs
for tx in ("837P", "835", "277CA", "999", "TA1"):
live = store.get_payer_config(payer_id, tx)
if live is not None:
configs.append({"transaction_type": tx, "config_json": live, "source": "db"})
return configs
@@ -0,0 +1,49 @@
"""``GET /api/providers`` — distinct providers observed across claims.
The list is built by ``store.distinct_providers()`` which scans claim
rows and returns one entry per unique ``billing_provider_npi``. The
``npi`` and ``state`` query params filter client-side. ``limit`` /
``offset`` paginate. Accepts ``application/x-ndjson`` like the other
list endpoints see ``api_helpers.ndjson_stream_list``.
"""
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
from cyclone.store import store
router = APIRouter()
@router.get("/api/providers")
def list_providers(
request: Request,
npi: str | None = Query(None),
state: str | None = Query(None),
limit: int = Query(100, ge=1, le=1000),
offset: int = Query(0, ge=0),
) -> Any:
items = store.distinct_providers()
if npi is not None:
items = [p for p in items if p["npi"] == npi]
if state is not None:
items = [p for p in items if p.get("state") == state]
paged = items[offset:offset + limit]
total = len(items)
returned = len(paged)
has_more = total > offset + returned
if wants_ndjson(request):
return StreamingResponse(
ndjson_stream_list(paged, total, returned, has_more),
media_type="application/x-ndjson",
)
return {
"items": paged,
"total": total,
"returned": returned,
"has_more": has_more,
}