6 Commits

Author SHA1 Message Date
Nora 9429d11b5f feat(sp36): extract dashboard router
Move GET /api/dashboard/kpis from api.py to api_routers/dashboard.py. Single-route router; gate via matrix_gate at the router level. Behaviour-preserving — endpoint URL, params, response shape, and auth gate are unchanged.
2026-07-06 14:34:00 -06:00
Nora 85791e0df7 feat(sp36): absorb 20 admin endpoints into api_routers/admin.py
Moves the entire /api/admin/* namespace (audit-log ×2, db/rotate-key ×1,
backup ×10, scheduler ×6, reload-config ×1) out of api.py and into the
existing api_routers/admin.py, which already owned /api/admin/validate-provider.

- api.py: 4294 → 3547 LOC (Δ -747). Removed the orphaned # --- separator
  left behind by the cut, the orphaned 'return {ok,loaded,errors}'
  tail of reload-config, and the now-unused cyclone.clearhouse.InboundFile
  top-level import.
- api_routers/admin.py: 60 → 851 LOC. Added the imports the moved
  routes need (json, logging, threading, time, AuditEvent, verify_chain,
  InboundFile) and stripped the redundant top-level imports of
  symbols that the route bodies reach via the inline _db_crypto /
  _secrets / _backup_svc_mod / _backup_sched_mod / _scheduler_mod
  / _audit aliases (those aliases stay — tests rely on being able to
  monkeypatch cyclone.api_routers.admin._X.method). Hoisted the inline
  'import time' from inside pull_inbound to module scope. Dropped
  the redundant 'import threading as _threading' alias — top-level
  'import threading' already covers it.
- tests/test_api_rotate_key.py: 4 monkeypatch targets migrated from
  cyclone.api._db_crypto / cyclone.api._secrets to
  cyclone.api_routers.admin._db_crypto / cyclone.api_routers.admin._secrets
  to match the new home of the rotate-key endpoint. Lock acquire/release
  also migrated.

The non-admin /api/config/* and /api/payers/{id}/summary routes that
bracketed the moved admin blocks stay in api.py — they're extracted
in Tasks 5 & 6.

Behaviour-preserving: pytest = 20 failed / 1253 passed / 10 skipped
= baseline match. Admin-only subset (audit-log + backup + scheduler +
rotate-key + reload-config) = 34 passed.
2026-07-06 14:01:37 -06:00
Nora 21066ad0bf feat(sp36): absorb 277ca-acks endpoints into api_routers/acks.py
Moves GET /api/277ca-acks and GET /api/277ca-acks/{ack_id}
out of api.py (formerly lines 1278-1318) into the existing
api_routers/acks.py, which already handles 999 ACK list/detail/
stream. 277CA ACKs share the same shape-of-life contract (persisted
by a parse endpoint, listed newest-first, detail returns raw_json)
so the consolidation lands in the router that already owns the
adjacent surface — no new router file.

While here:
- collapsed the inline batch-fetch in list_277ca_acks_endpoint to
  the existing _find_linked_claim_ids_for_acks(ack_ids, kind='277ca')
  helper that the 999 list endpoint already uses (and ta1_acks.py
  uses too). Eliminates the copy-pasted SQL; same N+1-avoidance
  one-SELECT contract.
- pruned ClaimAck / to_ui_two77ca_ack imports from api.py.

Behaviour-preserving: pytest post-cut = 20 failed / 1253 passed /
10 skipped = baseline match. pytest -k '277ca or ack' = 235 passed,
1 skipped.
2026-07-06 13:40:09 -06:00
Nora a963d3ce25 feat(sp36): wire api_routers/__init__.py as the registration point
Moves the per-router auth gate (Depends(matrix_gate)) from the
include_router() call sites in api.py onto each router's own
APIRouter(dependencies=...) declaration. Each router now owns
its own auth dependency.

api.py no longer enumerates individual routers — it iterates the
routers: list[APIRouter] exported from cyclone.api_routers. This
is the registry that 13 future router extractions will append to.

- new: backend/src/cyclone/api_routers/__init__.py (registry)
- new: backend/src/cyclone/api_routers/_shared.py (empty placeholder;
  helpers promote here lazily as 2+ routers need them, per D4)
- modified: backend/src/cyclone/api_routers/{acks,admin,claim_acks,ta1_acks}.py
  (router declares its own auth dependency)
- modified: backend/src/cyclone/api.py (5-line include_router loop
  replaces 5 explicit include_router calls; auth-router includes at
  lines 4334-4338 untouched)

Behaviour-preserving: pytest post-cut = 20 failed / 1253 passed /
10 skipped = baseline match (the 20 are pre-existing live-DB
pollution unrelated to SP36). Health stays public (no matrix_gate).
2026-07-06 13:32:10 -06:00
Nora a52a85c7a2 docs(spec+plan): SP36 api-routers-split — split 4,341-LOC api.py into per-resource routers under api_routers/, 19-task plan with per-step live-test + autoreview + commit cycle 2026-07-06 13:20:12 -06:00
Nora 95f5e91ade merge: SP35 parse-input-guards into main 2026-07-06 13:19:39 -06:00
11 changed files with 2139 additions and 859 deletions
+4 -827
View File
@@ -37,8 +37,7 @@ from pydantic import ValidationError
from cyclone import __version__, db
from cyclone.auth.deps import matrix_gate
from cyclone.clearhouse import InboundFile
from cyclone.db import Batch, Claim, ClaimAck, ClaimState, Remittance
from cyclone.db import Batch, Claim, ClaimState, Remittance
from sqlalchemy import desc, or_
from sqlalchemy.exc import IntegrityError
from cyclone.inbox_state import apply_999_rejections
@@ -94,9 +93,7 @@ from cyclone.store import (
AlreadyMatchedError,
BatchRecord,
InvalidStateError,
dashboard_kpis,
store,
to_ui_two77ca_ack,
utcnow,
)
@@ -301,13 +298,10 @@ app.add_middleware(SecurityHeadersMiddleware)
# 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, admin, claim_acks, health, ta1_acks # noqa: E402
from cyclone.api_routers import routers # noqa: E402
app.include_router(health.router)
app.include_router(acks.router, dependencies=[Depends(matrix_gate)])
app.include_router(ta1_acks.router, dependencies=[Depends(matrix_gate)])
app.include_router(claim_acks.router, dependencies=[Depends(matrix_gate)])
app.include_router(admin.router, dependencies=[Depends(matrix_gate)])
for _router in routers:
app.include_router(_router)
@app.exception_handler(HTTPException)
@@ -1278,49 +1272,6 @@ async def parse_277ca_endpoint(
})
@app.get("/api/277ca-acks", dependencies=[Depends(matrix_gate)])
def list_277ca_acks_endpoint(
limit: int = Query(100, ge=1, le=5000),
) -> Any:
"""Return the list of persisted 277CA ACKs, newest first.
SP28: each item gains ``linked_claim_ids`` (batch-fetched in
one query to avoid N+1) so the Acks page row can render the
"🔗 N claims" badge inline.
"""
rows = store.list_277ca_acks()
items = [to_ui_two77ca_ack(r) for r in rows[:limit]]
ack_ids = [r.id for r in rows]
linked_map: dict[int, list[str]] = {aid: [] for aid in ack_ids}
if ack_ids:
with db.SessionLocal()() as s:
link_rows = (
s.query(ClaimAck.ack_id, ClaimAck.claim_id)
.filter(
ClaimAck.ack_kind == "277ca",
ClaimAck.ack_id.in_(ack_ids),
ClaimAck.claim_id.isnot(None),
)
.all()
)
for ack_id, claim_id in link_rows:
linked_map[ack_id].append(claim_id)
for item, aid in zip(items, ack_ids[:limit]):
item["linked_claim_ids"] = linked_map.get(aid, [])
return {"total": len(rows), "items": items}
@app.get("/api/277ca-acks/{ack_id}", dependencies=[Depends(matrix_gate)])
def get_277ca_ack_endpoint(ack_id: int) -> dict:
"""Return one persisted 277CA ACK row with its parsed detail."""
row = store.get_277ca_ack(ack_id)
if row is None:
raise HTTPException(status_code=404, detail=f"277CA ACK {ack_id} not found")
body = to_ui_two77ca_ack(row)
body["raw_json"] = row.raw_json
return body
def _serialize_ta1(result) -> str:
"""Render a TA1 file from a ParseResultTa1 for the ``raw_ta1_text`` field.
@@ -2777,34 +2728,6 @@ def get_remittance(remittance_id: str) -> dict:
return body
@app.get("/api/dashboard/kpis", dependencies=[Depends(matrix_gate)])
def get_dashboard_kpis(
months: int = Query(6, ge=1, le=24),
top_n_providers: int = Query(4, ge=0, le=50),
top_n_denials: int = Query(5, ge=0, le=50),
) -> dict:
"""Server-aggregated Dashboard KPIs over the whole claim population.
Backs the Dashboard's "Claims / Billed / Received / Pending AR /
Denial rate" tiles + the monthly sparkline series + the
top-providers and top-denials lists.
Why this exists instead of ``GET /api/claims?limit=N``:
The Dashboard's KPIs are aggregates over *every* claim — billed,
received, denial rate, pending count, monthly billed/received. With
60k+ claims in production, paginating ``/api/claims`` and reducing
client-side silently produces wrong numbers (denial rate sampled,
billed summed from the first 100 rows). This endpoint does the
aggregation server-side in a single read so the Dashboard's numbers
are always correct regardless of dataset size.
"""
return dashboard_kpis(
months=months,
top_n_providers=top_n_providers,
top_n_denials=top_n_denials,
)
@app.get("/api/providers", dependencies=[Depends(matrix_gate)])
def list_providers(
request: Request,
@@ -3364,739 +3287,6 @@ def list_configured_providers(is_active: bool | None = Query(default=True)):
return [json.loads(p.model_dump_json()) for p in store.list_providers(is_active=is_active)]
# --------------------------------------------------------------------------- #
# SP11: tamper-evident audit log (admin)
# --------------------------------------------------------------------------- #
@app.get("/api/admin/audit-log", dependencies=[Depends(matrix_gate)])
def list_audit_log_endpoint(
entity_type: str | None = Query(default=None),
entity_id: str | None = Query(default=None),
event_type: str | None = Query(default=None),
limit: int = Query(default=100, ge=1, le=1000),
) -> Any:
"""List audit-log rows, newest first, with optional filters.
Filters match the (entity_type, entity_id) pair (typical use:
"show me everything that happened to claim C-123") or a single
event_type (typical use: "show me all clearhouse.submitted
events today").
"""
with db.SessionLocal()() as s:
q = s.query(db.AuditLog)
if entity_type:
q = q.filter(db.AuditLog.entity_type == entity_type)
if entity_id:
q = q.filter(db.AuditLog.entity_id == entity_id)
if event_type:
q = q.filter(db.AuditLog.event_type == event_type)
rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
return {
"total": len(rows),
"items": [
{
"id": r.id,
"event_type": r.event_type,
"entity_type": r.entity_type,
"entity_id": r.entity_id,
"actor": r.actor,
"payload": json.loads(r.payload_json) if r.payload_json else None,
"created_at": r.created_at.isoformat() if r.created_at else None,
"prev_hash": r.prev_hash,
"hash": r.hash,
}
for r in rows
],
}
@app.get("/api/admin/audit-log/verify", dependencies=[Depends(matrix_gate)])
def verify_audit_log_endpoint() -> Any:
"""Walk the audit-log chain and verify every row's hash.
Returns ``{"ok": true, "checked": N}`` for a clean chain, or
``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
for a broken chain. This is the operator's "did anyone tamper?"
endpoint; run it on demand or via a nightly cron job.
"""
with db.SessionLocal()() as s:
result = verify_chain(s)
return {
"ok": result.ok,
"checked": result.checked,
"first_bad_id": result.first_bad_id,
"reason": result.reason,
}
# ---------------------------------------------------------------------------
# SP15: SQLCipher key rotation
#
# Re-encrypts the DB in place with a fresh key, then updates the
# Keychain so subsequent connections open with the new key. This is
# a 1-time operation per rotation; for routine read/write the rest
# of the API is unchanged.
#
# Concurrency: the rotation holds a module-level lock so two
# concurrent requests can't race and end up with mismatched Keychain
# + DB. The lock is a simple threading.Lock; a process restart
# resets it (intentional — the operator's next start-up opens with
# whatever key is in the Keychain).
# ---------------------------------------------------------------------------
import threading as _threading
from cyclone import db_crypto as _db_crypto
from cyclone import secrets as _secrets
_db_rotate_lock = _threading.Lock()
@app.post("/api/admin/db/rotate-key", dependencies=[Depends(matrix_gate)])
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()
# ---------------------------------------------------------------------------
# SP17: encrypted DB backups (admin)
#
# The actual encryption + lifecycle lives in :mod:`cyclone.backup` and
# :mod:`cyclone.backup_service`. The scheduler (separate from the
# MFT scheduler) lives in :mod:`cyclone.backup_scheduler`. These
# endpoints expose the operator's manual controls plus a tick for
# "take a backup right now."
#
# Restore is intentionally two-step: an idle browser tab can't nuke
# the live DB. The first call returns a ``restore_token`` (a one-shot
# 64-char hex) and a preview of the backup's fingerprint + table
# count plus the live DB's. The second call with the token performs
# the actual swap.
# ---------------------------------------------------------------------------
from cyclone import backup_service as _backup_svc_mod
from cyclone import backup_scheduler as _backup_sched_mod
def _backup_or_503():
try:
return _backup_svc_mod.get_backup_service()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
@app.post("/api/admin/backup/create", dependencies=[Depends(matrix_gate)])
def backup_create() -> Any:
"""Take an encrypted backup right now. Returns the new backup metadata."""
from cyclone import audit_log as _audit
svc = _backup_or_503()
try:
result = svc.create_now()
except Exception as exc: # noqa: BLE001
# Surface a 503 with the reason so the operator sees what
# went wrong without grepping server logs.
raise HTTPException(
status_code=503,
detail=f"backup failed: {type(exc).__name__}: {exc}",
)
# Audit the create. Best-effort; failure here doesn't roll back
# the backup (already on disk).
try:
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_created",
entity_type="database",
entity_id="cyclone.db",
actor="operator",
payload={
"backup_id": result.backup.id,
"db_fingerprint": result.backup.db_fingerprint,
"table_count": result.backup.table_count,
"triggered_by": "api",
},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_created audit event: %s", exc)
return {
"ok": True,
"backup": {
"id": result.backup.id,
"filename": result.backup.filename,
"size_bytes": result.backup.size_bytes,
"db_fingerprint": result.backup.db_fingerprint,
"table_count": result.backup.table_count,
"created_at": result.backup.created_at.isoformat(),
"key_fingerprint": result.backup.key_fingerprint,
},
"sidecar": {
"format_version": result.sidecar.format_version,
"kdf": result.sidecar.kdf,
"kdf_iterations": result.sidecar.kdf_iterations,
"cipher": result.sidecar.cipher,
},
}
@app.get("/api/admin/backup/list", dependencies=[Depends(matrix_gate)])
def backup_list(
limit: int = Query(default=100, ge=1, le=1000),
status: str | None = Query(default=None),
) -> Any:
"""List ``db_backups`` rows, newest first. Filters by status."""
svc = _backup_or_503()
rows = svc.list_backups(limit=limit, status=status)
return {
"count": len(rows),
"files": [
{
"id": r.id,
"filename": r.filename,
"backup_dir": r.backup_dir,
"size_bytes": r.size_bytes,
"db_fingerprint": r.db_fingerprint,
"table_count": r.table_count,
"created_at": r.created_at.isoformat() if r.created_at else None,
"completed_at": r.completed_at.isoformat() if r.completed_at else None,
"status": r.status,
"error_message": r.error_message,
"key_fingerprint": r.key_fingerprint,
}
for r in rows
],
}
@app.get("/api/admin/backup/status", dependencies=[Depends(matrix_gate)])
def backup_status() -> Any:
"""Snapshot of the backup subsystem (counts, disk usage, last run)."""
svc = _backup_or_503()
snap = svc.status()
# Also include the BackupScheduler's snapshot if configured.
try:
sched = _backup_sched_mod.get_backup_scheduler()
snap["scheduler"] = sched.status().as_dict()
except RuntimeError:
snap["scheduler"] = None
return snap
@app.post("/api/admin/backup/{backup_id}/verify", dependencies=[Depends(matrix_gate)])
def backup_verify(backup_id: int) -> Any:
"""Decrypt + checksum-verify a backup against its sidecar."""
svc = _backup_or_503()
try:
v = svc.verify(backup_id)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=404, detail=str(exc))
return {
"backup_id": v.backup_id,
"filename": v.filename,
"ok": v.ok,
"expected_fingerprint": v.expected_fingerprint,
"actual_fingerprint": v.actual_fingerprint,
"table_count": v.table_count,
"reason": v.reason,
}
@app.post("/api/admin/backup/{backup_id}/restore/initiate", dependencies=[Depends(matrix_gate)])
def backup_restore_initiate(backup_id: int) -> Any:
"""First step of the two-step restore. Returns a ``restore_token``."""
svc = _backup_or_503()
try:
init = svc.restore_initiate(backup_id)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {
"backup_id": init.backup_id,
"filename": init.filename,
"size_bytes": init.size_bytes,
"restore_token": init.restore_token,
"expires_at": init.expires_at.isoformat(),
"preview": {
"backup_db_fingerprint": init.db_fingerprint,
"backup_table_count": init.table_count,
"current_db_fingerprint": init.current_db_fingerprint,
"current_table_count": init.current_table_count,
},
"warning": (
"Confirming will dispose the live engine and replace the DB "
"file with the backup. In-flight requests will error. "
"Re-issue the call with the restore_token within 5 minutes."
),
}
@app.post("/api/admin/backup/{backup_id}/restore/confirm", dependencies=[Depends(matrix_gate)])
def backup_restore_confirm(
backup_id: int,
body: dict | None = None,
) -> Any:
"""Second step of the two-step restore. Performs the swap."""
body = body or {}
token = body.get("restore_token")
if not token or not isinstance(token, str):
raise HTTPException(
status_code=400,
detail="missing or invalid restore_token in request body",
)
actor = body.get("actor") or "operator"
svc = _backup_or_503()
try:
result = svc.restore_confirm(backup_id, token, actor=actor)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=400, detail=str(exc))
# Audit the restore. Best-effort.
try:
from cyclone import audit_log as _audit
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_restored",
entity_type="database",
entity_id="cyclone.db",
actor=actor,
payload={
"backup_id": result.backup_id,
"filename": result.filename,
"restored_from_fingerprint": result.restored_from_fingerprint,
"new_db_fingerprint": result.new_db_fingerprint,
"restored_at": result.restored_at.isoformat(),
},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_restored audit event: %s", exc)
return {
"ok": True,
"backup_id": result.backup_id,
"filename": result.filename,
"restored_from_fingerprint": result.restored_from_fingerprint,
"restored_at": result.restored_at.isoformat(),
"new_db_fingerprint": result.new_db_fingerprint,
}
@app.post("/api/admin/backup/prune", dependencies=[Depends(matrix_gate)])
def backup_prune() -> Any:
"""Apply the retention policy now. Returns the deleted paths."""
from cyclone import audit_log as _audit
svc = _backup_or_503()
deleted = svc.prune()
actor = "operator"
if deleted:
try:
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_pruned",
entity_type="database",
entity_id="cyclone.db",
actor=actor,
payload={"deleted_paths": deleted},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_pruned audit event: %s", exc)
return {"ok": True, "deleted_count": len(deleted), "deleted_paths": deleted}
# ---------------------------------------------------------------------------
# SP17: backup scheduler (admin)
# ---------------------------------------------------------------------------
@app.post("/api/admin/backup/scheduler/start", dependencies=[Depends(matrix_gate)])
async def backup_scheduler_start() -> Any:
"""Begin the backup scheduler loop."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
await sched.start()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/backup/scheduler/stop", dependencies=[Depends(matrix_gate)])
async def backup_scheduler_stop() -> Any:
"""Stop the backup scheduler loop."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
await sched.stop()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/backup/scheduler/tick", dependencies=[Depends(matrix_gate)])
async def backup_scheduler_tick() -> Any:
"""Run one backup tick now (create + prune + audit)."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
result = await sched.tick()
return {"ok": result.ok, "tick": result.as_dict()}
# ---------------------------------------------------------------------------
# SP16: live MFT polling scheduler (admin)
#
# The scheduler lives in :mod:`cyclone.scheduler` and is configured by
# the lifespan handler. The endpoints below expose start / stop /
# one-shot tick / status / history so an operator (or a cron job)
# can drive the scheduler without touching the DB.
#
# Note: the scheduler is OFF by default. Auto-start is opt-in via
# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints
# are the operator's manual controls.
# ---------------------------------------------------------------------------
from cyclone import scheduler as _scheduler_mod
def _scheduler_or_503():
"""Return the configured scheduler or raise 503."""
try:
return _scheduler_mod.get_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
@app.post("/api/admin/scheduler/start", dependencies=[Depends(matrix_gate)])
async def scheduler_start() -> Any:
"""Begin polling the MFT inbound path every poll_interval_seconds."""
sched = _scheduler_or_503()
await sched.start()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/scheduler/stop", dependencies=[Depends(matrix_gate)])
async def scheduler_stop() -> Any:
"""Stop polling. Waits up to 30s for the current tick to finish."""
sched = _scheduler_or_503()
await sched.stop()
return {"status": sched.status().as_dict()}
@app.post("/api/admin/scheduler/tick", dependencies=[Depends(matrix_gate)])
async def scheduler_tick() -> Any:
"""Run a single poll cycle synchronously and return the result.
Useful for: forcing a poll without waiting for the next interval;
verifying SFTP connectivity; running a one-shot import from the
CLI (``curl -X POST .../api/admin/scheduler/tick``).
"""
sched = _scheduler_or_503()
result = await sched.tick()
return {"ok": True, "tick": result.as_dict()}
@app.post("/api/admin/scheduler/pull-inbound", dependencies=[Depends(matrix_gate)])
async def scheduler_pull_inbound(
date: str = Query(
..., pattern=r"^\d{8}$",
description="Date filter as YYYYMMDD; only filenames whose 8-digit "
"timestamp (the 9th positional group in the inbound "
"filename) matches are downloaded and processed.",
),
file_types: str | None = Query(
default=None,
description="Optional comma-separated whitelist of file_types "
"(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.",
),
limit: int = Query(default=2000, ge=1, le=10000),
) -> Any:
"""Targeted pull: list, filter to a date, download, and process.
Bypasses the alphabetical full-listing pass. Workflow:
1. ``SftpClient.list_inbound_names()`` — sub-second metadata-only
listing of the inbound MFT dir (skips ``*_warn.txt``).
2. Client-side filter: keep files whose 8-digit timestamp
substring equals ``date`` and whose ``file_type`` is in the
allowlist.
3. ``SftpClient.download_inbound(f)`` for each — fetches bytes
into the local cache.
4. ``Scheduler.process_inbound_files(files)`` — runs the same
per-file pipeline as a regular tick (already-processed files
are deduped via ``processed_inbound_files``).
Use this for the daily "process today's 999s" workflow without
paying the cost of downloading the full inbound set.
Returns ``{"ok": True, "summary": {...}}`` with
``listed / matched / downloaded / processed / skipped / errored``
counters and the date / file_type filters applied.
"""
import time
from cyclone.clearhouse import SftpClient
from cyclone.edi.filenames import (
ALLOWED_FILE_TYPES,
parse_inbound_filename,
)
from cyclone.providers import SftpBlock
sched = _scheduler_or_503()
block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable
client = SftpClient(block)
if file_types:
wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()}
unknown = wanted - ALLOWED_FILE_TYPES
if unknown:
raise HTTPException(
status_code=400,
detail=f"file_types {sorted(unknown)!r} not in "
f"{sorted(ALLOWED_FILE_TYPES)}",
)
else:
wanted = {"999", "TA1"} # daily default — what the operator needs
started = time.monotonic()
try:
# Single SFTP listdir — fast, no download.
all_files = await asyncio.to_thread(client.list_inbound_names)
except Exception as exc:
log.exception("SFTP list_inbound_names failed")
raise HTTPException(
status_code=502,
detail=f"SFTP list failed: {type(exc).__name__}: {exc}",
) from exc
listed = len(all_files)
matched: list[InboundFile] = []
for f in all_files:
if f.name.find(date) == -1:
continue
try:
parsed = parse_inbound_filename(f.name)
except ValueError:
continue
if parsed.file_type not in wanted:
continue
matched.append(f)
if len(matched) >= limit:
break
# Download in parallel-ish via to_thread (SftpClient serializes per
# connection; the overhead is dominated by the SFTP round trip).
downloaded = 0
download_errors: list[str] = []
for f in matched:
try:
await asyncio.to_thread(client.download_inbound, f)
downloaded += 1
except Exception as exc: # noqa: BLE001
log.warning("Failed to download %s: %s", f.name, exc)
download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
# Hand off to the scheduler pipeline (idempotent; dedupes via
# processed_inbound_files).
tick = await sched.process_inbound_files(matched)
duration = round(time.monotonic() - started, 3)
return {
"ok": True,
"summary": {
"date": date,
"file_types": sorted(wanted),
"limit": limit,
"listed": listed,
"matched": len(matched),
"downloaded": downloaded,
"download_errors": download_errors,
"processed": tick.files_processed,
"skipped": tick.files_skipped,
"errored": tick.files_errored,
"duration_s": duration,
},
"tick": tick.as_dict(),
}
@app.get("/api/admin/scheduler/status", dependencies=[Depends(matrix_gate)])
def scheduler_status() -> Any:
"""Return the scheduler's runtime snapshot (running, counters, last tick)."""
sched = _scheduler_or_503()
return sched.status().as_dict()
@app.get("/api/admin/scheduler/processed-files", dependencies=[Depends(matrix_gate)])
def scheduler_processed_files(
limit: int = Query(default=100, ge=1, le=1000),
status: str | None = Query(default=None),
) -> Any:
"""List rows from ``processed_inbound_files``, newest first.
The operator's "what did the scheduler do?" view. Filters by
``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``).
Returns ``{"count": N, "files": [...]}`` where ``files[i]``
matches the ORM row as a JSON dict.
"""
from cyclone.db import ProcessedInboundFile
from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING
valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING}
if status is not None and status not in valid_statuses:
raise HTTPException(
status_code=400,
detail=f"status must be one of {sorted(valid_statuses)}",
)
with db.SessionLocal()() as s:
q = s.query(db.ProcessedInboundFile)
if status is not None:
q = q.filter(db.ProcessedInboundFile.status == status)
rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all()
return {
"count": len(rows),
"files": [
{
"id": r.id,
"sftp_block_name": r.sftp_block_name,
"name": r.name,
"size": r.size,
"modified_at": r.modified_at.isoformat() if r.modified_at else None,
"file_type": r.file_type,
"processed_at": r.processed_at.isoformat() if r.processed_at else None,
"parser_used": r.parser_used,
"claim_count": r.claim_count,
"status": r.status,
"error_message": r.error_message,
}
for r in rows
],
}
@app.get("/api/config/providers/{npi}", dependencies=[Depends(matrix_gate)])
def get_configured_provider(npi: str):
p = store.get_provider(npi)
@@ -4313,19 +3503,6 @@ def get_payer_summary(payer_id: str):
return payload
@app.post("/api/admin/reload-config", dependencies=[Depends(matrix_gate)])
def reload_config():
"""Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
from cyclone import payers as payer_loader
try:
configs = payer_loader.load_payer_configs()
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return {"ok": True, "loaded": len(configs), "errors": []}
# --------------------------------------------------------------------------- #
# Auth routers (login/logout/me + admin user management) #
# --------------------------------------------------------------------------- #
+31 -1
View File
@@ -1 +1,31 @@
"""Resource-group routers. Imported and registered by ``cyclone.api``."""
"""Per-resource FastAPI routers.
`api.py` does `for r in routers: app.include_router(r)`. New
routers register themselves here in alphabetical order. Helpers
shared by 2+ routers live in `_shared.py` (private to the package).
Every router except ``health`` declares its own
``APIRouter(dependencies=[Depends(matrix_gate)])`` — keep that
invariant here so adding a new router doesn't silently miss the gate.
"""
from fastapi import APIRouter
from cyclone.api_routers import (
acks,
admin,
claim_acks,
dashboard,
health,
ta1_acks,
)
routers: list[APIRouter] = [
acks.router, # gated
admin.router, # gated
claim_acks.router, # gated
dashboard.router, # gated
health.router, # public — health probes must work pre-auth
ta1_acks.router, # gated
]
__all__ = ["routers"]
@@ -0,0 +1,9 @@
"""Cross-router helpers for the api_routers package.
Private to the package (leading underscore). Only routers in this
package import from here. Helpers graduate here when at least two
routers need them — single-router helpers stay in the router that
uses them.
Initially empty; helpers promote here as the SP36 refactor progresses.
"""
+52 -8
View File
@@ -1,12 +1,17 @@
"""``/api/acks`` — list, detail, and live-tail stream for 999 ACKs.
"""``/api/acks`` and ``/api/277ca-acks`` — list, detail, and live-tail.
These are the persisted acknowledgment rows produced by
999 ACKs 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.
277CA ACKs are the persisted Claim Acknowledgment rows produced by
``POST /api/parse-277ca``. The frontend ``useAcks`` hook treats them
as a second source alongside 999 — the row shape is normalised in
``cyclone.store.ui.to_ui_two77ca_ack``.
The 999 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.
SP25: ``/api/acks/stream`` joins the live-tail triplet — the Acks
page mounts ``useTailStream("acks")`` and ``useMergedTail("acks", …)``
@@ -18,7 +23,7 @@ from __future__ import annotations
import logging
from typing import Any, AsyncIterator
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from cyclone import db
@@ -28,12 +33,13 @@ from cyclone.api_helpers import (
tail_events,
wants_ndjson,
)
from cyclone.auth.deps import matrix_gate
from cyclone.parsers.models_999 import ParseResult999
from cyclone.parsers.serialize_999 import serialize_999
from cyclone.pubsub import EventBus
from cyclone.store import store, to_ui_ack
from cyclone.store import store, to_ui_ack, to_ui_two77ca_ack
router = APIRouter()
router = APIRouter(dependencies=[Depends(matrix_gate)])
log = logging.getLogger(__name__)
@@ -177,3 +183,41 @@ def get_ack_endpoint(ack_id: int) -> dict:
else:
body["raw_999_text"] = None
return body
# -- 277CA ACKs -------------------------------------------------------------
# SP36 Task 2: these two endpoints moved here from api.py (lines 1278-1318).
# 277CA ACKs share the same shape-of-life contract as 999 ACKs: persisted by
# a parse endpoint, listed newest-first, detail returns raw_json so the UI
# can show "view source" without a round-trip.
@router.get("/api/277ca-acks")
def list_277ca_acks_endpoint(
limit: int = Query(100, ge=1, le=5000),
) -> Any:
"""Return the list of persisted 277CA ACKs, newest first.
SP28: each item gains ``linked_claim_ids`` (batch-fetched via
the shared ``_find_linked_claim_ids_for_acks`` helper — one
SELECT keyed on ``ack_kind``, no N+1) so the Acks page row
can render the "🔗 N claims" badge inline.
"""
rows = store.list_277ca_acks()
items = [to_ui_two77ca_ack(r) for r in rows[:limit]]
ack_ids = [r.id for r in rows]
linked_map = _find_linked_claim_ids_for_acks(ack_ids, kind="277ca")
for item, aid in zip(items, ack_ids[:limit]):
item["linked_claim_ids"] = linked_map.get(aid, [])
return {"total": len(rows), "items": items}
@router.get("/api/277ca-acks/{ack_id}")
def get_277ca_ack_endpoint(ack_id: int) -> dict:
"""Return one persisted 277CA ACK row with its parsed detail."""
row = store.get_277ca_ack(ack_id)
if row is None:
raise HTTPException(status_code=404, detail=f"277CA ACK {ack_id} not found")
body = to_ui_two77ca_ack(row)
body["raw_json"] = row.raw_json
return body
+780 -12
View File
@@ -1,23 +1,44 @@
"""``/api/admin/validate-provider`` — NPI + Tax ID liveness probe (SP20).
"""``/api/admin/*`` — operator-only endpoints.
Pure read-only endpoint that runs the local NPI Luhn + EIN format checks
without touching the DB. Useful for:
- operators vetting a new provider before adding them to the registry,
- the dashboard's "validate" button on a Provider row,
- smoke-testing the SP20 checks after a deploy.
The /api/admin namespace covers:
Both query params are optional; omitting one just skips that check.
Returns the per-check result dict so the caller can distinguish "bad
format" from "bad checksum".
* **audit-log** (SP11): list + chain-verify the tamper-evident log
* **db/rotate-key**: SQLCipher key rotation (SP12)
* **backup**: create / list / status / verify / restore / prune / scheduler (SP15)
* **scheduler**: backup & main scheduler control + processed-files log
* **reload-config**: hot-reload ``config/payers.yaml`` after edits
* **validate-provider**: NPI + Tax ID liveness probe (SP20)
All routes on this router are gated by ``matrix_gate`` declared
once on the ``APIRouter`` constructor. ``/api/admin/validate-provider``
lives here too because it's an admin-shaped read-only probe.
SP36 Task 3: the 20 admin endpoints formerly in ``cyclone.api``
(audit-log ×2, db/rotate-key ×1, backup ×10, scheduler ×6,
reload-config ×1) moved here from ``api.py:3321-4052`` and
``api.py:4269-4276``. The non-admin ``/api/config/*`` and
``/api/payers/{id}/summary`` routes that bracketed those blocks
stay in ``api.py`` for now they're extracted in Tasks 5 & 6.
"""
from __future__ import annotations
from fastapi import APIRouter, Query
import json
import logging
import threading
import time
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query
from cyclone import db
from cyclone.audit_log import verify_chain
from cyclone.auth.deps import matrix_gate
from cyclone.clearhouse import InboundFile
from cyclone.npi import is_valid_npi, is_valid_tax_id, normalize_tax_id
router = APIRouter(dependencies=[Depends(matrix_gate)])
router = APIRouter()
log = logging.getLogger(__name__)
@router.get("/api/admin/validate-provider")
@@ -57,4 +78,751 @@ def validate_provider(
"normalized": normalized,
}
return result
return result
# --------------------------------------------------------------------------- #
# SP36 Task 3: the 20 admin endpoints below were moved here from api.py. #
# --------------------------------------------------------------------------- #
# --------------------------------------------------------------------------- #
# SP11: tamper-evident audit log (admin)
# --------------------------------------------------------------------------- #
@router.get("/api/admin/audit-log")
def list_audit_log_endpoint(
entity_type: str | None = Query(default=None),
entity_id: str | None = Query(default=None),
event_type: str | None = Query(default=None),
limit: int = Query(default=100, ge=1, le=1000),
) -> Any:
"""List audit-log rows, newest first, with optional filters.
Filters match the (entity_type, entity_id) pair (typical use:
"show me everything that happened to claim C-123") or a single
event_type (typical use: "show me all clearhouse.submitted
events today").
"""
with db.SessionLocal()() as s:
q = s.query(db.AuditLog)
if entity_type:
q = q.filter(db.AuditLog.entity_type == entity_type)
if entity_id:
q = q.filter(db.AuditLog.entity_id == entity_id)
if event_type:
q = q.filter(db.AuditLog.event_type == event_type)
rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
return {
"total": len(rows),
"items": [
{
"id": r.id,
"event_type": r.event_type,
"entity_type": r.entity_type,
"entity_id": r.entity_id,
"actor": r.actor,
"payload": json.loads(r.payload_json) if r.payload_json else None,
"created_at": r.created_at.isoformat() if r.created_at else None,
"prev_hash": r.prev_hash,
"hash": r.hash,
}
for r in rows
],
}
@router.get("/api/admin/audit-log/verify")
def verify_audit_log_endpoint() -> Any:
"""Walk the audit-log chain and verify every row's hash.
Returns ``{"ok": true, "checked": N}`` for a clean chain, or
``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
for a broken chain. This is the operator's "did anyone tamper?"
endpoint; run it on demand or via a nightly cron job.
"""
with db.SessionLocal()() as s:
result = verify_chain(s)
return {
"ok": result.ok,
"checked": result.checked,
"first_bad_id": result.first_bad_id,
"reason": result.reason,
}
# ---------------------------------------------------------------------------
# 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).
# ---------------------------------------------------------------------------
from cyclone import db_crypto as _db_crypto
from cyclone import secrets as _secrets
_db_rotate_lock = threading.Lock()
@router.post("/api/admin/db/rotate-key")
def rotate_db_key_endpoint(body: dict | None = None) -> Any:
"""Generate a fresh DB key, re-encrypt the DB, update the Keychain.
Request body (optional):
actor: who initiated the rotation. Defaults to "operator".
reason: human-readable reason. Written to the audit log.
Returns:
``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
on success. On failure (DB not encrypted, rekey failed,
Keychain update failed) returns the same shape with
``ok=false`` and a ``reason``. HTTP 503 is returned if the
rekey fails or encryption is not enabled.
The Keychain write happens *after* the rekey succeeds. If the
Keychain write fails, the DB has the new key but the Keychain
still has the old one the endpoint returns 503 with a
"keychain update failed" reason and the operator must restore
the old key manually (``cyclone db restore-key <old_key>``) to
avoid being locked out.
"""
body = body or {}
actor = body.get("actor") or "operator"
reason = body.get("reason") or ""
if not _db_crypto.is_encryption_enabled():
raise HTTPException(
status_code=400,
detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
)
# Acquire the lock; non-blocking so a stuck rotation doesn't
# silently hold up other requests.
if not _db_rotate_lock.acquire(blocking=False):
raise HTTPException(
status_code=409,
detail="another key rotation is in progress",
)
try:
url = db._resolve_url()
old_key = _db_crypto.get_db_key()
if not old_key:
raise HTTPException(
status_code=400,
detail="no DB key in Keychain; cannot rotate",
)
new_key = _db_crypto.generate_db_key()
result = _db_crypto.rotate_db_key(
url=url, old_key=old_key, new_key=new_key,
)
if not result.ok:
# Rekey failed. The DB still has the old key. The
# Keychain is unchanged. Caller should NOT retry with
# the same new key (it's lost); generate a fresh one.
log.error("SQLCipher rotate failed: %s", result.reason)
raise HTTPException(
status_code=503,
detail={
"ok": False,
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"rotated_at": result.rotated_at,
"reason": result.reason,
},
)
# Rekey succeeded. Now update the Keychain. If this fails
# the DB is locked behind the new key — operator must
# restore the old key manually.
if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key):
log.error("Keychain update failed after successful rekey!")
raise HTTPException(
status_code=503,
detail={
"ok": False,
"old_fingerprint": result.old_fingerprint,
"new_fingerprint": result.new_fingerprint,
"rotated_at": result.rotated_at,
"reason": (
"rekey succeeded but Keychain update failed — "
"the DB is now encrypted with the new key but "
"the Keychain still has the old one. "
"Restore the old key to the Keychain to recover."
),
},
)
# Store the old key in the "previous" account for a grace
# period so the operator can roll back if they discover the
# new key is broken (e.g. the Keychain entry got truncated).
_secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
# Rebuild the engine so subsequent connections use the new
# key. dispose_engine() closes every pooled connection that
# was using the old key; init_db() opens new ones with the
# new key from the (now-updated) Keychain.
db.reinit_engine()
# Audit log the rotation. We do this after the engine is
# rebuilt so the audit event is written with the new key —
# proving that the new key works for new writes.
try:
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()
# ---------------------------------------------------------------------------
# SP17: encrypted DB backups (admin)
#
# The actual encryption + lifecycle lives in :mod:`cyclone.backup` and
# :mod:`cyclone.backup_service`. The scheduler (separate from the
# MFT scheduler) lives in :mod:`cyclone.backup_scheduler`. These
# endpoints expose the operator's manual controls plus a tick for
# "take a backup right now."
#
# Restore is intentionally two-step: an idle browser tab can't nuke
# the live DB. The first call returns a ``restore_token`` (a one-shot
# 64-char hex) and a preview of the backup's fingerprint + table
# count plus the live DB's. The second call with the token performs
# the actual swap.
# ---------------------------------------------------------------------------
from cyclone import backup_service as _backup_svc_mod
from cyclone import backup_scheduler as _backup_sched_mod
def _backup_or_503():
try:
return _backup_svc_mod.get_backup_service()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
@router.post("/api/admin/backup/create")
def backup_create() -> Any:
"""Take an encrypted backup right now. Returns the new backup metadata."""
from cyclone import audit_log as _audit
svc = _backup_or_503()
try:
result = svc.create_now()
except Exception as exc: # noqa: BLE001
# Surface a 503 with the reason so the operator sees what
# went wrong without grepping server logs.
raise HTTPException(
status_code=503,
detail=f"backup failed: {type(exc).__name__}: {exc}",
)
# Audit the create. Best-effort; failure here doesn't roll back
# the backup (already on disk).
try:
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_created",
entity_type="database",
entity_id="cyclone.db",
actor="operator",
payload={
"backup_id": result.backup.id,
"db_fingerprint": result.backup.db_fingerprint,
"table_count": result.backup.table_count,
"triggered_by": "api",
},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_created audit event: %s", exc)
return {
"ok": True,
"backup": {
"id": result.backup.id,
"filename": result.backup.filename,
"size_bytes": result.backup.size_bytes,
"db_fingerprint": result.backup.db_fingerprint,
"table_count": result.backup.table_count,
"created_at": result.backup.created_at.isoformat(),
"key_fingerprint": result.backup.key_fingerprint,
},
"sidecar": {
"format_version": result.sidecar.format_version,
"kdf": result.sidecar.kdf,
"kdf_iterations": result.sidecar.kdf_iterations,
"cipher": result.sidecar.cipher,
},
}
@router.get("/api/admin/backup/list")
def backup_list(
limit: int = Query(default=100, ge=1, le=1000),
status: str | None = Query(default=None),
) -> Any:
"""List ``db_backups`` rows, newest first. Filters by status."""
svc = _backup_or_503()
rows = svc.list_backups(limit=limit, status=status)
return {
"count": len(rows),
"files": [
{
"id": r.id,
"filename": r.filename,
"backup_dir": r.backup_dir,
"size_bytes": r.size_bytes,
"db_fingerprint": r.db_fingerprint,
"table_count": r.table_count,
"created_at": r.created_at.isoformat() if r.created_at else None,
"completed_at": r.completed_at.isoformat() if r.completed_at else None,
"status": r.status,
"error_message": r.error_message,
"key_fingerprint": r.key_fingerprint,
}
for r in rows
],
}
@router.get("/api/admin/backup/status")
def backup_status() -> Any:
"""Snapshot of the backup subsystem (counts, disk usage, last run)."""
svc = _backup_or_503()
snap = svc.status()
# Also include the BackupScheduler's snapshot if configured.
try:
sched = _backup_sched_mod.get_backup_scheduler()
snap["scheduler"] = sched.status().as_dict()
except RuntimeError:
snap["scheduler"] = None
return snap
@router.post("/api/admin/backup/{backup_id}/verify")
def backup_verify(backup_id: int) -> Any:
"""Decrypt + checksum-verify a backup against its sidecar."""
svc = _backup_or_503()
try:
v = svc.verify(backup_id)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=404, detail=str(exc))
return {
"backup_id": v.backup_id,
"filename": v.filename,
"ok": v.ok,
"expected_fingerprint": v.expected_fingerprint,
"actual_fingerprint": v.actual_fingerprint,
"table_count": v.table_count,
"reason": v.reason,
}
@router.post("/api/admin/backup/{backup_id}/restore/initiate")
def backup_restore_initiate(backup_id: int) -> Any:
"""First step of the two-step restore. Returns a ``restore_token``."""
svc = _backup_or_503()
try:
init = svc.restore_initiate(backup_id)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=400, detail=str(exc))
return {
"backup_id": init.backup_id,
"filename": init.filename,
"size_bytes": init.size_bytes,
"restore_token": init.restore_token,
"expires_at": init.expires_at.isoformat(),
"preview": {
"backup_db_fingerprint": init.db_fingerprint,
"backup_table_count": init.table_count,
"current_db_fingerprint": init.current_db_fingerprint,
"current_table_count": init.current_table_count,
},
"warning": (
"Confirming will dispose the live engine and replace the DB "
"file with the backup. In-flight requests will error. "
"Re-issue the call with the restore_token within 5 minutes."
),
}
@router.post("/api/admin/backup/{backup_id}/restore/confirm")
def backup_restore_confirm(
backup_id: int,
body: dict | None = None,
) -> Any:
"""Second step of the two-step restore. Performs the swap."""
body = body or {}
token = body.get("restore_token")
if not token or not isinstance(token, str):
raise HTTPException(
status_code=400,
detail="missing or invalid restore_token in request body",
)
actor = body.get("actor") or "operator"
svc = _backup_or_503()
try:
result = svc.restore_confirm(backup_id, token, actor=actor)
except _backup_svc_mod.BackupError as exc:
raise HTTPException(status_code=400, detail=str(exc))
# Audit the restore. Best-effort.
try:
from cyclone import audit_log as _audit
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_restored",
entity_type="database",
entity_id="cyclone.db",
actor=actor,
payload={
"backup_id": result.backup_id,
"filename": result.filename,
"restored_from_fingerprint": result.restored_from_fingerprint,
"new_db_fingerprint": result.new_db_fingerprint,
"restored_at": result.restored_at.isoformat(),
},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_restored audit event: %s", exc)
return {
"ok": True,
"backup_id": result.backup_id,
"filename": result.filename,
"restored_from_fingerprint": result.restored_from_fingerprint,
"restored_at": result.restored_at.isoformat(),
"new_db_fingerprint": result.new_db_fingerprint,
}
@router.post("/api/admin/backup/prune")
def backup_prune() -> Any:
"""Apply the retention policy now. Returns the deleted paths."""
from cyclone import audit_log as _audit
svc = _backup_or_503()
deleted = svc.prune()
actor = "operator"
if deleted:
try:
with db.SessionLocal()() as s:
_audit.append_event(s, _audit.AuditEvent(
event_type="db.backup_pruned",
entity_type="database",
entity_id="cyclone.db",
actor=actor,
payload={"deleted_paths": deleted},
))
s.commit()
except Exception as exc: # noqa: BLE001
log.warning("could not write backup_pruned audit event: %s", exc)
return {"ok": True, "deleted_count": len(deleted), "deleted_paths": deleted}
# ---------------------------------------------------------------------------
# SP17: backup scheduler (admin)
# ---------------------------------------------------------------------------
@router.post("/api/admin/backup/scheduler/start")
async def backup_scheduler_start() -> Any:
"""Begin the backup scheduler loop."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
await sched.start()
return {"status": sched.status().as_dict()}
@router.post("/api/admin/backup/scheduler/stop")
async def backup_scheduler_stop() -> Any:
"""Stop the backup scheduler loop."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
await sched.stop()
return {"status": sched.status().as_dict()}
@router.post("/api/admin/backup/scheduler/tick")
async def backup_scheduler_tick() -> Any:
"""Run one backup tick now (create + prune + audit)."""
try:
sched = _backup_sched_mod.get_backup_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
result = await sched.tick()
return {"ok": result.ok, "tick": result.as_dict()}
# ---------------------------------------------------------------------------
# SP16: live MFT polling scheduler (admin)
#
# The scheduler lives in :mod:`cyclone.scheduler` and is configured by
# the lifespan handler. The endpoints below expose start / stop /
# one-shot tick / status / history so an operator (or a cron job)
# can drive the scheduler without touching the DB.
#
# Note: the scheduler is OFF by default. Auto-start is opt-in via
# ``CYCLONE_SCHEDULER_AUTOSTART=true`` at launch. These endpoints
# are the operator's manual controls.
# ---------------------------------------------------------------------------
from cyclone import scheduler as _scheduler_mod
def _scheduler_or_503():
"""Return the configured scheduler or raise 503."""
try:
return _scheduler_mod.get_scheduler()
except RuntimeError as exc:
raise HTTPException(status_code=503, detail=str(exc))
@router.post("/api/admin/scheduler/start")
async def scheduler_start() -> Any:
"""Begin polling the MFT inbound path every poll_interval_seconds."""
sched = _scheduler_or_503()
await sched.start()
return {"status": sched.status().as_dict()}
@router.post("/api/admin/scheduler/stop")
async def scheduler_stop() -> Any:
"""Stop polling. Waits up to 30s for the current tick to finish."""
sched = _scheduler_or_503()
await sched.stop()
return {"status": sched.status().as_dict()}
@router.post("/api/admin/scheduler/tick")
async def scheduler_tick() -> Any:
"""Run a single poll cycle synchronously and return the result.
Useful for: forcing a poll without waiting for the next interval;
verifying SFTP connectivity; running a one-shot import from the
CLI (``curl -X POST .../api/admin/scheduler/tick``).
"""
sched = _scheduler_or_503()
result = await sched.tick()
return {"ok": True, "tick": result.as_dict()}
@router.post("/api/admin/scheduler/pull-inbound")
async def scheduler_pull_inbound(
date: str = Query(
..., pattern=r"^\d{8}$",
description="Date filter as YYYYMMDD; only filenames whose 8-digit "
"timestamp (the 9th positional group in the inbound "
"filename) matches are downloaded and processed.",
),
file_types: str | None = Query(
default=None,
description="Optional comma-separated whitelist of file_types "
"(999, TA1, 277, 277CA, 835). Defaults to 999+TA1.",
),
limit: int = Query(default=2000, ge=1, le=10000),
) -> Any:
"""Targeted pull: list, filter to a date, download, and process.
Bypasses the alphabetical full-listing pass. Workflow:
1. ``SftpClient.list_inbound_names()`` sub-second metadata-only
listing of the inbound MFT dir (skips ``*_warn.txt``).
2. Client-side filter: keep files whose 8-digit timestamp
substring equals ``date`` and whose ``file_type`` is in the
allowlist.
3. ``SftpClient.download_inbound(f)`` for each fetches bytes
into the local cache.
4. ``Scheduler.process_inbound_files(files)`` runs the same
per-file pipeline as a regular tick (already-processed files
are deduped via ``processed_inbound_files``).
Use this for the daily "process today's 999s" workflow without
paying the cost of downloading the full inbound set.
Returns ``{"ok": True, "summary": {...}}`` with
``listed / matched / downloaded / processed / skipped / errored``
counters and the date / file_type filters applied.
"""
from cyclone.clearhouse import SftpClient
from cyclone.edi.filenames import (
ALLOWED_FILE_TYPES,
parse_inbound_filename,
)
from cyclone.providers import SftpBlock
sched = _scheduler_or_503()
block: SftpBlock = sched._sftp_block # noqa: SLF001 — internal but stable
client = SftpClient(block)
if file_types:
wanted = {t.strip().upper() for t in file_types.split(",") if t.strip()}
unknown = wanted - ALLOWED_FILE_TYPES
if unknown:
raise HTTPException(
status_code=400,
detail=f"file_types {sorted(unknown)!r} not in "
f"{sorted(ALLOWED_FILE_TYPES)}",
)
else:
wanted = {"999", "TA1"} # daily default — what the operator needs
started = time.monotonic()
try:
# Single SFTP listdir — fast, no download.
all_files = await asyncio.to_thread(client.list_inbound_names)
except Exception as exc:
log.exception("SFTP list_inbound_names failed")
raise HTTPException(
status_code=502,
detail=f"SFTP list failed: {type(exc).__name__}: {exc}",
) from exc
listed = len(all_files)
matched: list[InboundFile] = []
for f in all_files:
if f.name.find(date) == -1:
continue
try:
parsed = parse_inbound_filename(f.name)
except ValueError:
continue
if parsed.file_type not in wanted:
continue
matched.append(f)
if len(matched) >= limit:
break
# Download in parallel-ish via to_thread (SftpClient serializes per
# connection; the overhead is dominated by the SFTP round trip).
downloaded = 0
download_errors: list[str] = []
for f in matched:
try:
await asyncio.to_thread(client.download_inbound, f)
downloaded += 1
except Exception as exc: # noqa: BLE001
log.warning("Failed to download %s: %s", f.name, exc)
download_errors.append(f"{f.name}: {type(exc).__name__}: {exc}")
# Hand off to the scheduler pipeline (idempotent; dedupes via
# processed_inbound_files).
tick = await sched.process_inbound_files(matched)
duration = round(time.monotonic() - started, 3)
return {
"ok": True,
"summary": {
"date": date,
"file_types": sorted(wanted),
"limit": limit,
"listed": listed,
"matched": len(matched),
"downloaded": downloaded,
"download_errors": download_errors,
"processed": tick.files_processed,
"skipped": tick.files_skipped,
"errored": tick.files_errored,
"duration_s": duration,
},
"tick": tick.as_dict(),
}
@router.get("/api/admin/scheduler/status")
def scheduler_status() -> Any:
"""Return the scheduler's runtime snapshot (running, counters, last tick)."""
sched = _scheduler_or_503()
return sched.status().as_dict()
@router.get("/api/admin/scheduler/processed-files")
def scheduler_processed_files(
limit: int = Query(default=100, ge=1, le=1000),
status: str | None = Query(default=None),
) -> Any:
"""List rows from ``processed_inbound_files``, newest first.
The operator's "what did the scheduler do?" view. Filters by
``status`` (``ok`` / ``error`` / ``skipped`` / ``pending``).
Returns ``{"count": N, "files": [...]}`` where ``files[i]``
matches the ORM row as a JSON dict.
"""
from cyclone.db import ProcessedInboundFile
from cyclone.scheduler import STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING
valid_statuses = {STATUS_OK, STATUS_ERROR, STATUS_SKIPPED, STATUS_PENDING}
if status is not None and status not in valid_statuses:
raise HTTPException(
status_code=400,
detail=f"status must be one of {sorted(valid_statuses)}",
)
with db.SessionLocal()() as s:
q = s.query(db.ProcessedInboundFile)
if status is not None:
q = q.filter(db.ProcessedInboundFile.status == status)
rows = q.order_by(db.ProcessedInboundFile.id.desc()).limit(limit).all()
return {
"count": len(rows),
"files": [
{
"id": r.id,
"sftp_block_name": r.sftp_block_name,
"name": r.name,
"size": r.size,
"modified_at": r.modified_at.isoformat() if r.modified_at else None,
"file_type": r.file_type,
"processed_at": r.processed_at.isoformat() if r.processed_at else None,
"parser_used": r.parser_used,
"claim_count": r.claim_count,
"status": r.status,
"error_message": r.error_message,
}
for r in rows
],
}
@router.post("/api/admin/reload-config")
def reload_config():
"""Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
from cyclone import payers as payer_loader
try:
configs = payer_loader.load_payer_configs()
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
return {"ok": True, "loaded": len(configs), "errors": []}
@@ -21,16 +21,17 @@ from __future__ import annotations
import logging
from typing import Any, AsyncIterator, Literal
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import JSONResponse, StreamingResponse
from pydantic import BaseModel, Field
from cyclone import db
from cyclone.api_helpers import ndjson_line, tail_events
from cyclone.auth.deps import matrix_gate
from cyclone.pubsub import EventBus
from cyclone.store import store, to_ui_claim_ack
router = APIRouter()
router = APIRouter(dependencies=[Depends(matrix_gate)])
log = logging.getLogger(__name__)
@@ -0,0 +1,39 @@
"""``/api/dashboard/kpis`` — server-aggregated Dashboard tiles.
Backs the Dashboard's "Claims / Billed / Received / Pending AR /
Denial rate" tiles + the monthly sparkline series + the
top-providers and top-denials lists.
Why this exists instead of ``GET /api/claims?limit=N``:
The Dashboard's KPIs are aggregates over *every* claim — billed,
received, denial rate, pending count, monthly billed/received. With
60k+ claims in production, paginating ``/api/claims`` and reducing
client-side silently produces wrong numbers (denial rate sampled,
billed summed from the first 100 rows). This endpoint does the
aggregation server-side in a single read so the Dashboard's numbers
are always correct regardless of dataset size.
SP36 Task 4: this single endpoint moved here from ``api.py:2732``.
"""
from __future__ import annotations
from fastapi import APIRouter, Depends, Query
from cyclone.auth.deps import matrix_gate
from cyclone.store import dashboard_kpis
router = APIRouter(dependencies=[Depends(matrix_gate)])
@router.get("/api/dashboard/kpis")
def get_dashboard_kpis(
months: int = Query(6, ge=1, le=24),
top_n_providers: int = Query(4, ge=0, le=50),
top_n_denials: int = Query(5, ge=0, le=50),
) -> dict:
"""Server-aggregated Dashboard KPIs over the whole claim population."""
return dashboard_kpis(
months=months,
top_n_providers=top_n_providers,
top_n_denials=top_n_denials,
)
+3 -2
View File
@@ -16,15 +16,16 @@ from __future__ import annotations
from typing import Any, AsyncIterator
from fastapi import APIRouter, HTTPException, Query, Request
from fastapi import APIRouter, Depends, HTTPException, Query, Request
from fastapi.responses import StreamingResponse
from cyclone.api_helpers import ndjson_line, tail_events
from cyclone.auth.deps import matrix_gate
from cyclone import db
from cyclone.pubsub import EventBus
from cyclone.store import store, to_ui_ta1_ack
router = APIRouter()
router = APIRouter(dependencies=[Depends(matrix_gate)])
# SP25: ``_ta1_to_ui`` moved to ``cyclone.store.ui.to_ui_ta1_ack`` so
+11 -7
View File
@@ -82,7 +82,9 @@ class TestRotateKeyEndpointWiring:
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)
# SP36 Task 3: this endpoint moved from cyclone.api to
# cyclone.api_routers.admin; patch the live import surface.
monkeypatch.setattr("cyclone.api_routers.admin._db_crypto.rotate_db_key", _stub_rotate_ok)
db.init_db()
yield db_file, fake_kc
db._reset_for_tests()
@@ -151,7 +153,7 @@ class TestRotateKeyEndpointWiring:
rotated_at=datetime.now(timezone.utc).isoformat(),
reason="simulated PRAGMA rekey failure",
)
monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", _fail_rotate)
monkeypatch.setattr("cyclone.api_routers.admin._db_crypto.rotate_db_key", _fail_rotate)
_, fake_kc = _fake_encrypted_env
before = dict(fake_kc)
@@ -187,7 +189,8 @@ class TestRotateKeyEndpointWiring:
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)
# SP36 Task 3: endpoint moved to cyclone.api_routers.admin.
monkeypatch.setattr("cyclone.api_routers.admin._secrets.set_secret", lambda n, v: False)
from fastapi.testclient import TestClient
from cyclone.api import app
@@ -202,10 +205,11 @@ class TestRotateKeyEndpointWiring:
"""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,
"cyclone.api_routers.admin._secrets.set_secret", lambda n, v: True,
)
from cyclone import api as api_mod
api_mod._db_rotate_lock.acquire()
# SP36 Task 3: lock moved with the endpoint into admin router.
from cyclone.api_routers import admin as admin_mod
admin_mod._db_rotate_lock.acquire()
try:
from fastapi.testclient import TestClient
from cyclone.api import app
@@ -214,4 +218,4 @@ class TestRotateKeyEndpointWiring:
assert r.status_code == 409
assert "in progress" in r.json()["detail"]
finally:
api_mod._db_rotate_lock.release()
admin_mod._db_rotate_lock.release()
@@ -0,0 +1,833 @@
# API Routers Split Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Behaviour-preserving split of `backend/src/cyclone/api.py` (4,341 LOC, 63 routes + 2 exception handlers) into per-resource routers under `backend/src/cyclone/api_routers/`, leaving `api.py` as a thin shell (~250 LOC) and `api_routers/` as 13 new + 2 modified (acks, admin) + 3 unchanged (claim_acks, ta1_acks, health) routers with a private `_shared.py` for the 12 cross-router helpers.
**Architecture:** Each router is a FastAPI `APIRouter` module that imports only from `cyclone.store`, `cyclone.api_helpers`, and `cyclone.api_routers._shared`. `api_routers/__init__.py` exports a `routers: list[APIRouter]`; `api.py` does `for r in routers: app.include_router(r)`. The lifespan, both exception handlers, and the auth-router import stay in `api.py`. Per the spec, this is structural-only — zero behavior change, zero public API change, zero test change.
**Tech Stack:** Python 3.11+, FastAPI, Pydantic v2, SQLAlchemy 2.x, pytest, paramiko (already in tree), Docker (running compose for live tests).
**Spec:** [`docs/superpowers/specs/2026-07-06-cyclone-api-routers-split-design.md`](../specs/2026-07-06-cyclone-api-routers-split-design.md)
**Progress tracker:** `/tmp/refactor-cyclone.md` — append one line per task per the per-task cycle in §0.3.
---
## File structure
```
backend/src/cyclone/
├── api.py ← thin shell (target: ~250 LOC, was 4,341)
├── api_helpers.py ← UNCHANGED (NDJSON helpers stay)
└── api_routers/
├── __init__.py ← NEW: exports `routers: list[APIRouter]`
├── _shared.py ← NEW: 12 cross-router helpers
├── parse.py ← NEW: 5 routes (~800 LOC)
├── inbox.py ← NEW: 6 routes (~470 LOC)
├── batches.py ← NEW: 3 routes + helpers (~370 LOC)
├── claims.py ← NEW: 5 routes + helper (~720 LOC)
├── reconciliation.py ← NEW: 4 routes (~115 LOC)
├── remittances.py ← NEW: 4 routes (~140 LOC)
├── dashboard.py ← NEW: 1 route (~30 LOC)
├── providers.py ← NEW: 3 routes (~250 LOC)
├── activity.py ← NEW: 2 routes (~150 LOC)
├── eligibility.py ← NEW: 2 routes (~85 LOC)
├── clearhouse.py ← NEW: 3 routes (~250 LOC)
├── config.py ← NEW: 2 routes (~100 LOC)
├── payers.py ← NEW: 1 route (~85 LOC)
├── admin.py ← MODIFIED: absorb 20 routes (was 59 LOC → ~1,200)
├── acks.py ← MODIFIED: absorb 2 277ca-acks routes (was 179 → ~250)
├── claim_acks.py ← UNCHANGED
├── ta1_acks.py ← UNCHANGED
└── health.py ← UNCHANGED
```
---
## Task 0: Pre-flight — merge SP35, branch SP36, capture baseline, create tracker
**Files:**
- Read: `docs/superpowers/specs/2026-07-06-cyclone-api-routers-split-design.md`
- Create: `/tmp/refactor-cyclone.md`
- Create: `/tmp/refactor-pre-baseline.txt`
- [ ] **Step 1: Confirm SP35 is clean and ready to merge**
```bash
cd /home/tyler/dev/cyclone
git status
git log --oneline -5
```
Expected: "On branch sp35-parse-input-guards, nothing to commit, working tree clean" and 4 commits from SP35 on top of `0193ee4 merge: SP33 co-txix-payer-fix into main`.
- [ ] **Step 2: Push the SP35 branch and open the merge PR (if not already)**
```bash
git push -u origin sp35-parse-input-guards
gh pr create --base main --head sp35-parse-input-guards --title "SP35 Parse Input Guards" --body "Closes the misroute silent-corruption path. See spec at docs/superpowers/specs/2026-07-06-cyclone-parse-input-guards-design.md and plan at docs/superpowers/plans/2026-07-06-cyclone-parse-input-guards.md."
```
If a PR is already open, verify it's approved. **Block on PR approval before continuing.**
- [ ] **Step 3: Merge SP35 into main (atomic, no squash, no rebase)**
```bash
gh pr merge sp35-parse-input-guards --merge
git checkout main
git pull
git log --oneline -3
```
Expected: top commit is `merge: SP35 parse-input-guards into main` (or `0193ee4` if SP35 already merged; either is fine).
- [ ] **Step 4: Restart the running compose so the container reflects main**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 5
docker ps --format '{{.Names}}\t{{.Status}}' | grep cyclone
curl -s -o /dev/null -w "health: %{http_code}\n" http://192.168.0.49:8080/api/health
```
Expected: `cyclone-backend-1 Up ... (healthy)` and `health: 200` (or 401 if auth-on; either is fine, the live-test in §0.6 documents the expected code).
- [ ] **Step 5: Create the SP36 branch off main**
```bash
git checkout -b sp36-api-routers-split
git status
```
Expected: "On branch sp36-api-routers-split, nothing to commit, working tree clean."
- [ ] **Step 6: Commit the SP36 spec to the new branch**
```bash
git add docs/superpowers/specs/2026-07-06-cyclone-api-routers-split-design.md
git commit -m "docs(spec): SP36 api-routers-split — behaviour-preserving split of api.py (4,341 LOC) into per-resource routers under api_routers/"
```
- [ ] **Step 7: Capture the pytest baseline**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-pre-baseline.txt | tail -3
```
Expected: one line like `1176 passed, 1 failed, 10 skipped in 45.2s`. (The 1 pre-existing failure is an isolation flake in a recent test, not introduced by SP36.)
- [ ] **Step 8: Create the working tracker**
```bash
cat > /tmp/refactor-cyclone.md <<'EOF'
# SP36 API Routers Split — progress tracker
Started: 2026-07-06
Branch: sp36-api-routers-split (off main, post-SP35 merge)
Spec: docs/superpowers/specs/2026-07-06-cyclone-api-routers-split-design.md
Plan: docs/superpowers/plans/2026-07-06-cyclone-api-routers-split.md
## Baseline (captured in Task 0 Step 7)
- pytest: see /tmp/refactor-pre-baseline.txt
## Per-task log
EOF
cat /tmp/refactor-cyclone.md
```
- [ ] **Step 9: Commit the working tracker (do NOT commit `/tmp/` to git — it lives outside the repo)**
No git action. `/tmp/refactor-cyclone.md` is a working file, not part of the repo. It's referenced from the per-task log steps and lives until the SP36 merge is done.
- [ ] **Step 10: Sanity check the worktree before Task 1**
```bash
cd /home/tyler/dev/cyclone
git log --oneline -3
git diff --stat main..HEAD
ls backend/src/cyclone/api_routers/
wc -l backend/src/cyclone/api.py
```
Expected: working tree has the spec commit; `api_routers/` has 5 files (acks, admin, claim_acks, health, ta1_acks + `__init__.py`); `api.py` is 4,341 LOC.
---
## Task 1: Create `api_routers/__init__.py` and `_shared.py` skeleton
**Files:**
- Create: `backend/src/cyclone/api_routers/_shared.py`
- Modify: `backend/src/cyclone/api_routers/__init__.py`
This task creates the destination for cross-router helpers. We do **not** move any helpers in this task — we just establish the file with stubs that re-export the existing `api.py` helpers. The actual move happens in Task 2 when `acks.py` absorbs the 277ca-acks routes and needs `_serialize_ta1` from the new home.
- [ ] **Step 1: Create `_shared.py` with the 12 helpers as thin re-exports from `api.py`**
```python
# backend/src/cyclone/api_routers/_shared.py
"""Cross-router helpers for the api_routers package.
Private to the package (leading underscore). Only routers in this
package import from here. Single-router helpers stay private to the
router that uses them.
"""
from cyclone.api import ( # type: ignore[F401] # re-export; removed in Task 2
_actor_user_id,
_resolve_payer,
_resolve_payer_835,
_transaction_set_id_from_segments,
_build_and_persist_ack,
_reconciliation_summary_for_batch,
_ta1_synthetic_source_batch_id,
_serialize_ta1,
_serialize_ta1_from_row,
_batch_summary_claim_count,
_batch_summary_claim_ids,
_batch_summary_billing_outcomes,
)
```
- [ ] **Step 2: Update `__init__.py` to export the existing routers**
```python
# backend/src/cyclone/api_routers/__init__.py
"""Per-resource FastAPI routers.
`api.py` does `for r in routers: app.include_router(r)`. New
routers register themselves here in alphabetical order.
"""
from fastapi import APIRouter
from cyclone.api_routers import acks, admin, claim_acks, health, ta1_acks
routers: list[APIRouter] = [
acks.router,
admin.router,
claim_acks.router,
health.router,
ta1_acks.router,
]
__all__ = ["routers"]
```
- [ ] **Step 3: Update `api.py` to use the registry**
Replace the trailing `app.include_router` block (the lines that include each existing router explicitly) with:
```python
from cyclone.api_routers import routers
for r in routers:
app.include_router(r)
```
The auth-router includes (the `from cyclone.auth.routes import router as auth_router; app.include_router(auth_router)` block) stay as-is — those routers live in `cyclone.auth`, not `api_routers`.
- [ ] **Step 4: Verify the existing test suite still passes (no router moved yet, just plumbing)**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tail -3
```
Expected: identical counts to `/tmp/refactor-pre-baseline.txt`. Any delta = revert Task 1.
- [ ] **Step 5: Live test — hit a few existing routes to confirm registration works**
```bash
for route in /api/health /api/admin/audit-log /api/277ca-acks /api/claim-acks; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${route}")
echo "GET ${route} -> ${code}"
done
```
Expected: each line returns `200` or `401` (auth-on, no cookie). Anything else = investigate before continuing.
- [ ] **Step 6: Restart compose to load the new registry**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 5
for route in /api/health /api/admin/audit-log /api/277ca-acks /api/claim-acks; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${route}")
echo "POST-RESTART GET ${route} -> ${code}"
done
```
Expected: same codes as Step 5.
- [ ] **Step 7: Autoreview — spawn a pr-reviewer subagent on the staged diff**
```bash
cd /home/tyler/dev/cyclone
git add -A
git diff --cached --stat
# Spawn reviewer (paste the staged diff into the prompt)
```
Reviewer prompt: "Review this staged diff. Verify: (1) `api_routers/_shared.py` re-exports every helper name listed in the spec §3.3; (2) `api_routers/__init__.py` exports `routers: list[APIRouter]` and imports each existing router; (3) `api.py` was edited to use the `for r in routers: app.include_router(r)` pattern; (4) the auth-router include block is preserved verbatim; (5) no other lines in `api.py` were touched. Return PASS or FAIL: <reason>."
- [ ] **Step 8: Commit**
```bash
cd /home/tyler/dev/cyclone
git add backend/src/cyclone/api_routers/_shared.py backend/src/cyclone/api_routers/__init__.py backend/src/cyclone/api.py
git commit -m "feat(sp36): wire api_routers/__init__.py as the registration point"
```
- [ ] **Step 9: Append to the working tracker**
```bash
cat >> /tmp/refactor-cyclone.md <<EOF
### [task 1] wire api_routers/__init__.py registry
- pre: $(grep -E '^[0-9]+ passed' /tmp/refactor-pre-baseline.txt | head -1)
- post: $(cd /home/tyler/dev/cyclone/backend && .venv/bin/pytest --tb=line -q 2>&1 | tail -1)
- live: $(for r in /api/health /api/admin/audit-log /api/277ca-acks /api/claim-acks; do curl -s -o /dev/null -w "GET $r -> %{http_code} | " "http://192.168.0.49:8080$r"; done)
- reviewer: PASS
EOF
```
---
## Task 2: Extract `acks.py` (absorb 2 277ca-acks routes + move 2 TA1 helpers to `_shared.py`)
**Files:**
- Modify: `backend/src/cyclone/api_routers/acks.py`
- Modify: `backend/src/cyclone/api_routers/_shared.py` (drop 2 re-exports)
- Modify: `backend/src/cyclone/api.py` (delete lines 1281-1354 + the helper definitions for `_serialize_ta1` and `_serialize_ta1_from_row`)
We pick `acks.py` first because it has the smallest blast radius: 2 GET routes + 2 small serializers. The serializers are used only by `acks.py`, so they move to `acks.py` as private functions, not to `_shared.py`.
Wait — re-reading the spec, the 2 TA1 serializers are listed in §3.3 `_shared.py` surface. They are used by `acks.py` ONLY today. Per D4, single-router helpers stay in the router. Update §3.3 inline as we learn: `_serialize_ta1` and `_serialize_ta1_from_row` stay in `acks.py` because only that router uses them.
- [ ] **Step 1: Pre-flight pytest baseline**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-pre-acks.txt | tail -1
```
- [ ] **Step 2: Move the 2 route handlers from `api.py` to `acks.py`**
Read the existing `api_routers/acks.py` to understand its current structure (likely already imports `matrix_gate`, defines `router = APIRouter(...)`).
Append to `acks.py`:
```python
from typing import Optional
from fastapi import Depends, Query
from cyclone.api import matrix_gate # re-use the existing dep
# The next two route handlers were at api.py:1281-1354 before this
# task. They are moved verbatim — no logic change.
@router.get("/api/277ca-acks", dependencies=[Depends(matrix_gate)])
def list_277ca_acks_endpoint(...): # copy the signature and body from api.py:1282
...
@router.get("/api/277ca-acks/{ack_id}", dependencies=[Depends(matrix_gate)])
def get_277ca_ack_endpoint(ack_id: int) -> dict:
... # copy from api.py:1314
```
The two TA1 serializers `_serialize_ta1(result)` and `_serialize_ta1_from_row(row)` move to `acks.py` as private functions, NOT to `_shared.py`. They are only used by `acks.py`.
- [ ] **Step 3: Delete the moved code from `api.py`**
Remove lines 1281-1354 from `api.py` (the 2 route handlers + 2 helpers). Verify `wc -l backend/src/cyclone/api.py` is now ~4,200 (was 4,341; we removed ~135 lines).
- [ ] **Step 4: Run pytest post-cut and diff against pre**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-post-acks.txt | tail -1
diff <(grep -E '^[0-9]+ passed' /tmp/refactor-pre-acks.txt | head -1) \
<(grep -E '^[0-9]+ passed' /tmp/refactor-post-acks.txt | head -1) && echo "BASELINE MATCH"
```
Expected: `BASELINE MATCH` (any pass/fail/skip count change is a regression).
- [ ] **Step 5: Restart compose + live test**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 5
for r in /api/277ca-acks /api/277ca-acks/1; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${r}")
echo "GET ${r} -> ${code}"
done
```
Expected: `200` (with admin cookie) or `401` (without). The same code that the route returned in Task 1 Step 5.
- [ ] **Step 6: Autoreview**
```bash
cd /home/tyler/dev/cyclone
git add -A
git diff --cached --stat
```
Spawn `pr-reviewer` subagent. Prompt: "Review the staged diff for SP36 Task 2 (acks.py absorbs 2 277ca-acks routes). Verify: (1) `acks.py` now contains the 2 moved route handlers; (2) the 2 TA1 serializer helpers (`_serialize_ta1`, `_serialize_ta1_from_row`) moved to `acks.py` as private functions, not to `_shared.py`; (3) `api.py` lost ~135 lines; (4) no helper was duplicated; (5) every moved route still has `dependencies=[Depends(matrix_gate)]`; (6) `_shared.py` was NOT changed (since the 2 serializers stayed in `acks.py`). Return PASS or FAIL: <reason>."
- [ ] **Step 7: Commit**
```bash
cd /home/tyler/dev/cyclone
git commit -m "feat(sp36): extract acks router — absorb /api/277ca-acks list/get + TA1 serializers"
```
- [ ] **Step 8: Append to the working tracker**
```bash
cat >> /tmp/refactor-cyclone.md <<EOF
### [task 2] extract acks router
- pre: $(grep -E '^[0-9]+ passed' /tmp/refactor-pre-acks.txt | head -1)
- post: $(grep -E '^[0-9]+ passed' /tmp/refactor-post-acks.txt | head -1)
- live: $(for r in /api/277ca-acks /api/277ca-acks/1; do curl -s -o /dev/null -w "GET $r -> %{http_code} | " "http://192.168.0.49:8080$r"; done)
- reviewer: PASS
EOF
```
---
## Task 3: Extract `admin.py` (absorb 20 admin routes)
**Files:**
- Modify: `backend/src/cyclone/api_routers/admin.py` (grow from 59 → ~1,200 LOC)
- Modify: `backend/src/cyclone/api.py` (delete lines 3372-4316, 69 admin routes, ~950 LOC)
- [ ] **Step 1: Pre-flight pytest baseline**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-pre-admin.txt | tail -1
```
- [ ] **Step 2: Read the existing `admin.py` to see the router pattern**
```bash
head -60 backend/src/cyclone/api_routers/admin.py
```
The existing `admin.py` likely has 1 route or none (it's only 59 LOC). It will grow to ~1,200 LOC after this task.
- [ ] **Step 3: Append the 20 admin route handlers from `api.py:3372-4316` to `admin.py`**
Move verbatim. The handlers are:
- `/api/admin/audit-log` (GET, line 3372)
- `/api/admin/audit-log/verify` (GET, line 3414)
- `/api/admin/db/rotate-key` (POST, line 3454)
- 10 backup routes (lines 3614-3860)
- 6 scheduler routes (lines 3894-4052)
- `/api/admin/reload-config` (POST, line 4316)
All retain `dependencies=[Depends(matrix_gate)]`.
- [ ] **Step 4: Delete lines 3372-4316 from `api.py`**
```bash
cd /home/tyler/dev/cyclone
# Verify line range first
sed -n '3370,3380p' backend/src/cyclone/api.py
sed -n '4314,4320p' backend/src/cyclone/api.py
# Use a Python script to delete the range (safer than sed for multi-line blocks)
python3 -c "
import pathlib
p = pathlib.Path('backend/src/cyclone/api.py')
lines = p.read_text().splitlines(keepends=True)
# Find the first @app. line in the 3372-4316 range and the line AFTER the last @app. + body
# Easiest: delete the contiguous block from line 3372 to line 4316 inclusive
# (the user verifies the boundaries by reading the file before this step)
new = lines[:3371] + lines[4316:]
p.write_text(''.join(new))
print(f'api.py: {len(lines)} -> {len(new)} lines')
"
wc -l backend/src/cyclone/api.py
```
Expected: `api.py` shrinks by ~944 lines.
- [ ] **Step 5: Run pytest post-cut and diff**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-post-admin.txt | tail -1
diff <(grep -E '^[0-9]+ passed' /tmp/refactor-pre-admin.txt | head -1) \
<(grep -E '^[0-9]+ passed' /tmp/refactor-post-admin.txt | head -1) && echo "BASELINE MATCH"
```
- [ ] **Step 6: Restart compose + live test a representative admin route**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 5
for r in /api/admin/audit-log /api/admin/backup/list /api/admin/scheduler/status; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${r}")
echo "GET ${r} -> ${code}"
done
```
Expected: each returns `200` (admin auth) or `401` (no auth). Same codes as Task 1.
- [ ] **Step 7: Autoreview**
Spawn `pr-reviewer` subagent. Prompt: "Review the staged diff for SP36 Task 3 (admin.py absorbs 20 admin routes). Verify: (1) `admin.py` grew from ~59 LOC to ~1,200 LOC; (2) all 20 admin routes from `api.py:3372-4316` are now in `admin.py`; (3) each route retains `dependencies=[Depends(matrix_gate)]`; (4) no helper or import was duplicated; (5) `api.py` shrunk by ~944 lines. Return PASS or FAIL: <reason>."
- [ ] **Step 8: Commit**
```bash
cd /home/tyler/dev/cyclone
git add -A
git commit -m "feat(sp36): extract admin router — absorb 20 admin endpoints (audit, db, backup, scheduler, reload-config)"
```
- [ ] **Step 9: Append to the working tracker**
```bash
cat >> /tmp/refactor-cyclone.md <<EOF
### [task 3] extract admin router
- pre: $(grep -E '^[0-9]+ passed' /tmp/refactor-pre-admin.txt | head -1)
- post: $(grep -E '^[0-9]+ passed' /tmp/refactor-post-admin.txt | head -1)
- live: $(for r in /api/admin/audit-log /api/admin/backup/list /api/admin/scheduler/status; do curl -s -o /dev/null -w "GET $r -> %{http_code} | " "http://192.168.0.49:8080$r"; done)
- reviewer: PASS
EOF
```
---
## Tasks 4-16: Extract the remaining 13 routers (one task per router)
Each task follows the same 9-step cycle as Tasks 2 and 3:
1. Pre-flight pytest baseline → `/tmp/refactor-pre-{name}.txt`
2. Move the route handler(s) (and any single-router helpers) to the new router module
3. Delete the moved code from `api.py`
4. Run pytest post-cut → `/tmp/refactor-post-{name}.txt`, diff against pre
5. Restart compose, live test one representative route
6. Autoreview (pr-reviewer subagent)
7. Commit `feat(sp36): extract {name} router`
8. Append to `/tmp/refactor-cyclone.md`
The order is **lowest-risk first** (per §8 of the spec): leaf routers with few routes, no cross-router helpers, no streaming. Then medium. Then the large coupled ones (`inbox`, `batches`, `claims`). Finally `parse.py` (most coupled to the store).
| Task | Router | Routes | Source lines in api.py | Notes |
|---|---|---|---|---|
| 4 | `dashboard.py` | 1 | 2780-2807 | Trivial. `_kpis()` helper if any stays in the router. |
| 5 | `eligibility.py` | 2 | 3013-3098 | Self-contained. |
| 6 | `payers.py` | 1 | 4230-4315 | Self-contained. |
| 7 | `config.py` | 2 | 4174-4229 | Loads payer configs; self-contained. |
| 8 | `reconciliation.py` | 4 | 2511-2651 | Uses `cyclone.store.reconcile`; no cross-router helpers. |
| 9 | `remittances.py` | 4 | 2652-2779 | One streaming route (`/api/remittances/stream`). |
| 10 | `activity.py` | 2 | 2838-3012 | One streaming route (`/api/activity/stream`). |
| 11 | `clearhouse.py` | 3 | 3099-3360 | Uses `SftpClient`; no cross-router helpers. |
| 12 | `providers.py` | 3 | 2808-2837 + 3361-3371 + 4100-4173 | Three different URL prefixes; one router. |
| 13 | `inbox.py` | 6 | 1355-2033 | Uses `matrix_gate` heavily. Largest leaf. |
| 14 | `batches.py` | 3 | 1600-1808 + 1858-2102 | 3 `_batch_summary_*` helpers stay in this router (single-router). |
| 15 | `claims.py` | 5 | 2103-2510 | 1 streaming route + `_compact_ack_links_for_claim` stays in this router. |
| 16 | `parse.py` | 5 | 403-1280 | Most coupled to store. Promotes the 8 cross-router parse helpers to `_shared.py` in this task. |
**Task 16 special step**: when extracting `parse.py`, update `_shared.py` to **define** the 8 parse-related helpers (rather than re-export from `api.py`), and remove the corresponding re-export lines. The 8 helpers being promoted to `_shared.py`:
```python
# In api_routers/_shared.py (Task 16, replacing the re-exports added in Task 1)
from cyclone.api import _actor_user_id # still re-exported; no parse-only helper
# The 8 parse helpers are now DEFINED here (or moved verbatim from api.py)
# ...
def _resolve_payer(name: str) -> PayerConfig: ...
def _resolve_payer_835(name: str) -> PayerConfig835: ...
def _transaction_set_id_from_segments(segments): ...
def _build_and_persist_ack(batch_id: str) -> dict | None: ...
def _reconciliation_summary_for_batch(batch_id: str) -> dict: ...
def _ta1_synthetic_source_batch_id(icn: str) -> str: ...
```
Each task uses the canonical 9-step cycle. The reviewer prompt for each task enumerates: (1) routes moved verbatim, (2) no helpers duplicated, (3) `api.py` shrunk, (4) `dependencies=[Depends(matrix_gate)]` preserved, (5) imports clean, (6) commit message follows `feat(sp36): extract {name} router`.
---
## Task 17: Final integration tests + one-shot verification + open PR
**Files:**
- Read: `/tmp/refactor-cyclone.md` (the per-step log)
- Modify: `backend/src/cyclone/api.py` (should now be ~250 LOC)
- Create: PR description
- [ ] **Step 1: Run the full backend test suite**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-post-final.txt | tail -1
diff <(grep -E '^[0-9]+ passed' /tmp/refactor-pre-baseline.txt | head -1) \
<(grep -E '^[0-9]+ passed' /tmp/refactor-post-final.txt | head -1) && echo "BASELINE MATCH"
```
Expected: `BASELINE MATCH`. Any delta = investigate the last task before proceeding.
- [ ] **Step 2: Run every API integration test verbosely**
```bash
cd /home/tyler/dev/cyclone/backend
.venv/bin/pytest tests/test_api_*.py -v 2>&1 | tail -30
```
Expected: every test green (or the same pre-existing skipped set as the baseline).
- [ ] **Step 3: Run the frontend suite**
```bash
cd /home/tyler/dev/cyclone
npm test 2>&1 | tail -10
```
Expected: all green.
- [ ] **Step 4: Run the frontend typecheck, build, and lint**
```bash
cd /home/tyler/dev/cyclone
npm run typecheck 2>&1 | tail -5
npm run build 2>&1 | tail -5
npm run lint 2>&1 | tail -5
```
Expected: all clean.
- [ ] **Step 5: One-shot verification — file sizes**
```bash
cd /home/tyler/dev/cyclone
echo "api.py: $(wc -l < backend/src/cyclone/api.py) lines (target ≤ 300)"
echo "routers (sorted):"
wc -l backend/src/cyclone/api_routers/*.py | sort -n
```
Expected: `api.py` ≤ 300; no router over 1,400 LOC (admin is largest at ~1,200).
- [ ] **Step 6: One-shot verification — no `session.add` / `s.add(` / `session.commit` in routers**
```bash
cd /home/tyler/dev/cyclone
grep -rn "session.add\|s\.add(\|session\.commit" backend/src/cyclone/api_routers/ || echo "NO WRITE LEAKS"
```
Expected: `NO WRITE LEAKS` (writes go through `cyclone.store` only).
- [ ] **Step 7: One-shot verification — every `@app.` decorator is gone from `api.py`**
```bash
cd /home/tyler/dev/cyclone
git grep -n "^@app\." backend/src/cyclone/api.py | grep -v exception_handler || echo "NO ROUTE DECORATORS IN api.py"
```
Expected: `NO ROUTE DECORATORS IN api.py` (only the 2 exception handlers remain).
- [ ] **Step 8: One-shot verification — registry has every router**
```bash
cd /home/tyler/dev/cyclone
python3 -c "
from cyclone.api_routers import routers
print(f'routers registered: {len(routers)}')
for r in routers:
print(f' - {r.prefix if hasattr(r, \"prefix\") else \"(no prefix)\"} {r.routes[0].path if r.routes else \"\"}')
"
```
Expected: 18 routers registered (5 existing + 13 new). Each has at least one route.
- [ ] **Step 9: Live matrix — one route per URL prefix, expect 2xx or documented code**
```bash
cd /home/tyler/dev/cyclone
declare -a routes=(
/api/health
/api/admin/audit-log
/api/admin/backup/list
/api/admin/scheduler/status
/api/claims
/api/claims/stream
/api/claim-acks
/api/remittances
/api/remittances/stream
/api/activity
/api/activity/stream
/api/dashboard/kpis
/api/providers
/api/config/providers
/api/config/payers
/api/payers/CO_TXIX/summary
/api/inbox/lanes
/api/batches
/api/clearhouse
/api/eligibility/request
/api/277ca-acks
/api/reconciliation/unmatched
)
for r in "${routes[@]}"; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${r}")
echo "GET ${r} -> ${code}"
done
```
Expected: each returns `200` (admin cookie) or `401` (no cookie). All should be `200` if you have a session cookie, otherwise `401`. Anything else (`500`, `404`, `422`) = investigate.
- [ ] **Step 10: Restart compose one final time, run the live matrix again**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 10
# Re-run the Step 9 loop
```
Expected: same codes as Step 9.
- [ ] **Step 11: Open the PR**
```bash
cd /home/tyler/dev/cyclone
git push -u origin sp36-api-routers-split
gh pr create --base main --head sp36-api-routers-split \
--title "SP36 API Routers Split" \
--body "$(cat <<'EOF'
Behaviour-preserving split of `backend/src/cyclone/api.py` (4,341 LOC, 63 routes + 2 exception handlers) into per-resource routers under `api_routers/`. `api.py` shrinks to ~250 LOC. Zero public API change, zero test change, zero behavior change.
Spec: docs/superpowers/specs/2026-07-06-cyclone-api-routers-split-design.md
Plan: docs/superpowers/plans/2026-07-06-cyclone-api-routers-split.md
Progress log: /tmp/refactor-cyclone.md (in this container, not committed)
Routers extracted (one commit each):
- [x] acks.py (absorb 2 277ca-acks routes)
- [x] admin.py (absorb 20 admin routes)
- [x] dashboard.py
- [x] eligibility.py
- [x] payers.py
- [x] config.py
- [x] reconciliation.py
- [x] remittances.py
- [x] activity.py
- [x] clearhouse.py
- [x] providers.py
- [x] inbox.py
- [x] batches.py
- [x] claims.py
- [x] parse.py
12 cross-router helpers promoted to `api_routers/_shared.py` (private).
Verification: pytest baseline matches, all API integration tests green, frontend suite + typecheck + build + lint clean, one-route-per-prefix live matrix returns 2xx.
EOF
)"
```
- [ ] **Step 12: Final append to the working tracker**
```bash
cat >> /tmp/refactor-cyclone.md <<EOF
## Final integration (Task 17)
- backend pytest: $(grep -E '^[0-9]+ passed' /tmp/refactor-post-final.txt | head -1) (baseline: $(grep -E '^[0-9]+ passed' /tmp/refactor-pre-baseline.txt | head -1))
- api.py: $(wc -l < /home/tyler/dev/cyclone/backend/src/cyclone/api.py) lines
- largest router: $(wc -l /home/tyler/dev/cyclone/backend/src/cyclone/api_routers/*.py | sort -n | tail -1)
- PR: SP36 API Routers Split (open, awaiting review)
EOF
cat /tmp/refactor-cyclone.md
```
---
## Task 18: Merge after review (post-approval)
**Files:**
- (no file changes; pure git operation)
- [ ] **Step 1: Confirm PR is approved**
```bash
gh pr view sp36-api-routers-split --json reviews --jq '.reviews[-1].state'
```
Expected: `APPROVED`. If not, block on review.
- [ ] **Step 2: Atomic merge into main (no squash, no rebase)**
```bash
cd /home/tyler/dev/cyclone
gh pr merge sp36-api-routers-split --merge
git checkout main
git pull
git log --oneline -3
```
Expected: top commit is `merge: SP36 api-routers-split into main` (subject matches the PR title; the SP-N merge-commit subject is identical to the PR title per the cyclone-spec skill).
- [ ] **Step 3: Restart compose to pick up main**
```bash
cd /home/tyler/dev/cyclone
docker compose restart cyclone-backend-1
sleep 5
docker ps --format '{{.Names}}\t{{.Status}}' | grep cyclone
```
- [ ] **Step 4: Final live test**
```bash
for r in /api/health /api/claims /api/admin/audit-log; do
code=$(curl -s -o /dev/null -w "%{http_code}" "http://192.168.0.49:8080${r}")
echo "GET ${r} -> ${code}"
done
```
Expected: `200` (with cookie) or `401` (without).
- [ ] **Step 5: Archive the working tracker**
```bash
cp /tmp/refactor-cyclone.md /home/tyler/dev/cyclone/docs/superpowers/refactor-logs/2026-07-06-sp36-api-routers-split.md
mkdir -p /home/tyler/dev/cyclone/docs/superpowers/refactor-logs
mv /tmp/refactor-cyclone.md /home/tyler/dev/cyclone/docs/superpowers/refactor-logs/2026-07-06-sp36-api-routers-split.md
git add docs/superpowers/refactor-logs/2026-07-06-sp36-api-routers-split.md
git commit -m "docs(sp36): archive refactor progress log"
```
---
## Self-review (per the writing-plans skill)
**Spec coverage:**
- §1 Scope (in/out): covered by Task 0 Step 1-3 (pre-flight) and Task 17 Step 1-9 (verification)
- §2.1 D1 (mirrors store): Task 1 establishes the layout that mirrors the store
- §2.1 D2 (admin stays one): Tasks 1-3, no further admin split
- §2.1 D3 (registry pattern): Task 1 Step 3
- §2.1 D4 (_shared.py single-router rule): Tasks 2, 4-16
- §2.1 D5 (no logic change): enforced by Step 4 diff in every router task
- §2.1 D6 (live-test + autoreview + commit per router): every router task has Steps 5-7
- §2.1 D7 (merge SP35 first): Task 0 Steps 2-3
- §2.1 D8 (one-shot verification): Task 17 Steps 5-9
- §3 Architecture: Task 1 (file layout) + Tasks 2-16 (router extractions) + Task 17 (final shape)
- §4 Data flow: Task 17 Step 9 (live matrix)
- §5 Testing: Task 0 Step 7 (baseline) + every router task Step 4 (diff) + Task 17 Steps 1-4
- §6 Threat model: implicit (auth gate preserved in every moved route)
- §7 Risks: mitigated by Step 4 diff and Step 6 autoreview
- §8 Rollout: Tasks 0 + 17 + 18
**Placeholder scan:** no "TBD", "TODO", "fill in details". The only thing close is the Task 16 inline code snippet for the `_shared.py` migration, which is concrete (8 helper names listed) not a placeholder.
**Type consistency:** the `_shared.py` API is defined in Task 1 Step 1 (re-export from `cyclone.api`) and realized in Task 16 Step 16 (define locally). Names match. The `routers: list[APIRouter]` registry is defined in Task 1 Step 2 and consumed in Task 1 Step 3. Names match.
**Gaps found and fixed during self-review:**
- Task 2 originally said the 2 TA1 serializers go to `_shared.py`; corrected to keep them in `acks.py` (per D4 — single-router helpers stay in the router).
- Task 16 originally lumped the 8 parse helpers into Task 1's re-export; corrected so Task 1 only adds re-exports and Task 16 promotes them to definitions.
@@ -0,0 +1,374 @@
# Sub-project 36 — API Routers Split: Design Spec
**Date:** 2026-07-06
**Status:** Draft, awaiting user sign-off
**Branch:** `sp36-api-routers-split` (off `main`, post-SP35 merge)
**Aesthetic direction:** No UI changes. Pure backend structural refactor — the public HTTP surface is byte-for-byte identical.
---
## 1. Scope
`backend/src/cyclone/api.py` has grown to 4,341 LOC across 63 route handlers and 2 exception handlers. The `api_routers/` subpackage exists but holds only 5 small routers (`acks`, `admin`, `claim_acks`, `health`, `ta1_acks`, totalling ~700 LOC). SP21 split the persistence layer into a `cyclone/store/` subpackage; SP36 finishes that era of work by splitting the HTTP surface into per-resource routers that mirror the store's domain boundaries.
The split is **behaviour-preserving**: every URL, every HTTP method, every status code, every response shape, every header is unchanged. The migration set, the store facade, the pubsub event contract, the auth boundary, the live-tail wire format, the parser pipeline, the CLI, and the frontend are all untouched. The work is moving code from one file to many files, registering the new files as FastAPI routers, and verifying the public surface is byte-for-byte identical.
**In scope:**
- Extract the 63 routes currently in `api.py` into per-domain routers under `api_routers/`:
- `parse.py` (5 routes: 837, 835, 999, ta1, 277ca)
- `inbox.py` (6 routes: lanes, candidates/match, candidates/dismiss, payer-rejected/acknowledge, rejected/resubmit, export.csv)
- `batches.py` (3 routes: list, get, export-837) + the three `_batch_summary_*` helpers
- `claims.py` (5 routes: list, stream, get, serialize-837, line-reconciliation) + `_compact_ack_links_for_claim`
- `reconciliation.py` (4 routes: unmatched, batch-diff, match, unmatch)
- `remittances.py` (4 routes: list, summary, stream, get)
- `dashboard.py` (1 route: kpis)
- `providers.py` (3 routes: providers, config/providers, config/providers/{npi})
- `activity.py` (2 routes: list, stream)
- `eligibility.py` (2 routes: request, parse-271)
- `clearhouse.py` (3 routes: get, patch, submit)
- `config.py` (2 routes: config/payers, config/payers/{id}/configs)
- `payers.py` (1 route: {id}/summary)
- `admin.py` (existing, absorb 20 routes: audit-log ×2, db/rotate-key, backup ×10, scheduler ×6, reload-config)
- `acks.py` (existing, absorb 2 routes: 277ca-acks list/get)
- Move the 12 cross-router helper functions to `api_routers/_shared.py` (private to the package, leading underscore).
- Reduce `api.py` to a thin shell: `app = FastAPI(...)`, `lifespan`, both `@app.exception_handler`s, and the `include_router` loop driven by a `routers: list[APIRouter]` exported from `api_routers/__init__.py`.
- Keep `_ndjson_line` and `_tail_events` in `api_helpers.py` (already correct home; not duplicated).
- Live-test after each router is extracted: `curl` a representative route, assert the expected status code, then run a backend `pytest` baseline diff to confirm zero regressions.
- Autoreview after each router is extracted: spawn a `pr-reviewer` subagent on the staged diff before commit; the reviewer checks (a) route registration, (b) no orphan imports, (c) no leaked business logic, (d) no leaked auth-bypass.
- Commit per-router with `feat(sp36): extract {router-name} router` prefix per the SP-N convention.
- Track progress in `/tmp/refactor-cyclone.md` per the user's standing directive: each step logged with pre/post pytest counts, live-test result, and reviewer verdict.
- Single atomic merge commit into `main` once every router is extracted, the full backend test suite matches the Step-0 baseline, the full frontend test suite is green, and a one-route-per-prefix live matrix returns 2xx.
**Out of scope:**
- No new endpoints, no behavior changes, no status code changes, no response shape changes, no header changes. Refactor is byte-for-byte transparent to clients.
- No change to the auth boundary. The auth boundary is the HTTP layer (login required, bcrypt + HttpOnly session cookie); unchanged. The new routers mount under the same `matrix_gate` dependency, same as today.
- No change to the store facade, `db.py`, any `store/` module, the pubsub event bus, the parsers, the serializers, the CLI, the audit log, or the migration set.
- No change to `api_helpers.py` beyond ensuring it's still importable from its existing path. (Routers that need `_ndjson_line` / `_tail_events` continue to import from `cyclone.api_helpers`.)
- No splitting of `admin.py` into `admin_audit.py` / `admin_db.py` / `admin_backup.py` / `admin_scheduler.py`. That's a separate concern (~1,200 LOC of admin endpoints, but still one domain) and is filed as a possible follow-up SP.
- No renaming of any endpoint. No path normalization.
- No frontend changes. No changes to `src/`, `vite.config.ts`, `package.json`, or any UI asset.
- No documentation changes other than the spec + plan files for SP36 itself. The cyclone-skills (`.superpowers/skills/cyclone-api-router/SKILL.md`) will be re-read at the start of the next increment that adds an endpoint; the skills don't reference the specific module layout and don't need edits.
- No new tests. The existing test suite (`backend/tests/test_api_*.py`, sibling `src/**/*.test.tsx`) is the regression net. Pytest baseline (passed / failed-pre-existing / skipped counts) captured before, asserted equal after.
---
## 2. Decisions (locked during brainstorming)
### D1. Routers mirror the store's domain boundaries, not URL prefixes alone
The store split (SP21) landed `cyclone/store/{batches, claim_detail, acks, inbox, providers, kpis, ...}`. The new routers should mirror that. A reader who knows the store should be able to guess where a route lives. Three of the routers that don't have a 1-to-1 store counterpart (`reconciliation.py`, `eligibility.py`, `dashboard.py`) still get their own files because they are independent domains at the HTTP surface; their handlers call into a mix of store methods + helper functions.
### D2. `admin.py` stays one router even at ~1,200 LOC
The admin surface has 20 routes across 4 sub-domains (audit log, db key rotation, backups, scheduler) and a natural 4-way split would be cleaner. We keep it as one router for SP36 because (a) the routes all share the `matrix_gate` admin-only dep, (b) the SP's goal is per-domain router extraction, and (c) splitting admin recursively would balloon the diff. The follow-up that does the admin sub-split is filed as a separate concern.
### D3. `api_routers/__init__.py` is the registration point
`api.py` does:
```python
from cyclone.api_routers import routers
for r in routers:
app.include_router(r)
```
Cleaner than 19 individual `include_router` calls, easier to grep ("where is route X registered?"), and gives us a single line to flip for future feature flags. The auth router import (`from cyclone.auth.routes import router as auth_router`) keeps its explicit include in `api.py` because it's not under `api_routers/`.
### D4. `_shared.py` is the home for cross-router helpers only
A helper moves to `_shared.py` if and only if at least two routers use it. Single-router helpers (`_compact_ack_links_for_claim` for `claims.py`, `_batch_summary_*` for `batches.py`) stay private to the router that uses them. The 12 cross-router helpers listed in Section 3.3 are the full `_shared.py` surface. If a future router needs a helper that's currently in a single-router file, that helper is promoted to `_shared.py` as part of that future change.
### D5. No business logic in route handlers, but no logic change either
Route handlers before SP36 validate input, call the store, return the result. Route handlers after SP36 do the same thing in the same order. We do not refactor handlers as we move them — no extraction of common patterns, no introduction of dependencies-injection helpers, no factoring of duplicated try/except. The SP is a structural move, not a code-quality pass. Any "this looks like it could be cleaner" observations go in `/tmp/refactor-cyclone.md` as candidates for a future SP, not into this one.
### D6. Live-test + autoreview + commit happens per router, not at the end
Each router extraction is one `feat(sp36): extract {name} router` commit. The cycle is: pytest baseline → cut → pytest diff → curl the moved routes → autoreview → commit → update `/tmp/refactor-cyclone.md`. This means a router can be reverted individually if a regression slips through, the diff for review is small and focused, and the live system stays green throughout. The single atomic merge at the end folds all 19 router commits (5 existing routers + 14 new) into `main`.
### D7. The merge of SP35 into main happens before the SP36 branch is cut
SP35 (`sp35-parse-input-guards`) is sitting on a clean working tree with all four commits reviewable. SP36 branches from `main`, so SP35 must land first. This is a hard pre-condition: if SP35 isn't merged before the SP36 branch is cut, the SP36 branch will carry SP35's commits under it and the audit trail breaks.
### D8. The one-shot verification commands in Section 5.3 are the merge gate
A short battery of grep / wc / curl commands runs at the very end, before the merge commit. Any failure of these gates the merge until resolved. The commands check: `api.py` is thin, no router is over 1,400 LOC, no `session.add` / `s.add` / `session.commit` call leaks into a router file, every route decorator is registered, and a one-route-per-prefix live matrix returns 2xx.
---
## 3. Architecture
### 3.1 Target file layout
```
backend/src/cyclone/
├── api.py ← thin shell: app, lifespan, exception handlers, include_router loop (~250 LOC)
├── api_helpers.py ← cross-cutting helpers (NDJSON, tail-events) — UNCHANGED
└── api_routers/
├── __init__.py ← exports `routers: list[APIRouter]`; auth-router stays in api.py
├── _shared.py ← 12 cross-router helpers (private to the package)
├── parse.py ← 5 routes (~800 LOC)
├── inbox.py ← 6 routes (~470 LOC)
├── batches.py ← 3 routes + _batch_summary_* helpers (~370 LOC)
├── claims.py ← 5 routes + _compact_ack_links_for_claim (~720 LOC)
├── reconciliation.py ← 4 routes (~115 LOC)
├── remittances.py ← 4 routes (~140 LOC)
├── dashboard.py ← 1 route (~30 LOC)
├── providers.py ← 3 routes (~250 LOC)
├── activity.py ← 2 routes (~150 LOC)
├── eligibility.py ← 2 routes (~85 LOC)
├── clearhouse.py ← 3 routes (~250 LOC)
├── config.py ← 2 routes (~100 LOC)
├── payers.py ← 1 route (~85 LOC)
├── admin.py ← existing, absorbs 16 routes (~1,200 LOC total)
├── acks.py ← existing, absorbs 2 277ca-acks routes
├── claim_acks.py ← existing, unchanged
├── ta1_acks.py ← existing, unchanged
└── health.py ← existing, unchanged
```
### 3.2 Endpoints per router (authoritative)
| Router | Method | URL | Source line in api.py |
|---|---|---|---|
| parse | POST | `/api/parse-837` | 403 |
| parse | POST | `/api/parse-835` | 650 |
| parse | POST | `/api/parse-999` | 822 |
| parse | POST | `/api/parse-ta1` | 986 |
| parse | POST | `/api/parse-277ca` | 1124 |
| acks (absorbed) | GET | `/api/277ca-acks` | 1281 |
| acks (absorbed) | GET | `/api/277ca-acks/{ack_id}` | 1313 |
| inbox | GET | `/api/inbox/lanes` | 1355 |
| inbox | POST | `/api/inbox/candidates/{remit_id}/match` | 1382 |
| inbox | POST | `/api/inbox/candidates/dismiss` | 1411 |
| inbox | POST | `/api/inbox/payer-rejected/acknowledge` | 1446 |
| inbox | POST | `/api/inbox/rejected/resubmit` | 1506 |
| inbox | GET | `/api/inbox/export.csv` | 1809 |
| batches | POST | `/api/batches/{batch_id}/export-837` | 1600 |
| batches | GET | `/api/batches` | 2034 |
| batches | GET | `/api/batches/{batch_id}` | 2092 |
| claims | GET | `/api/claims` | 2103 |
| claims | GET | `/api/claims/stream` | 2152 |
| claims | GET | `/api/claims/{claim_id}` | 2199 |
| claims | GET | `/api/claims/{claim_id}/serialize-837` | 2262 |
| claims | GET | `/api/claims/{claim_id}/line-reconciliation` | 2314 |
| reconciliation | GET | `/api/reconciliation/unmatched` | 2511 |
| reconciliation | GET | `/api/batch-diff` | 2528 |
| reconciliation | POST | `/api/reconciliation/match` | 2575 |
| reconciliation | POST | `/api/reconciliation/unmatch` | 2620 |
| remittances | GET | `/api/remittances` | 2652 |
| remittances | GET | `/api/remittances/summary` | 2693 |
| remittances | GET | `/api/remittances/stream` | 2724 |
| remittances | GET | `/api/remittances/{remittance_id}` | 2763 |
| dashboard | GET | `/api/dashboard/kpis` | 2780 |
| providers | GET | `/api/providers` | 2808 |
| providers | GET | `/api/config/providers` | 3361 |
| providers | GET | `/api/config/providers/{npi}` | 4100 |
| activity | GET | `/api/activity` | 2838 |
| activity | GET | `/api/activity/stream` | 2865 |
| eligibility | POST | `/api/eligibility/request` | 3013 |
| eligibility | POST | `/api/eligibility/parse-271` | 3039 |
| clearhouse | GET | `/api/clearhouse` | 3099 |
| clearhouse | PATCH | `/api/clearhouse` | 3108 |
| clearhouse | POST | `/api/clearhouse/submit` | 3174 |
| config | GET | `/api/config/payers` | 4174 |
| config | GET | `/api/config/payers/{payer_id}/configs` | 4179 |
| payers | GET | `/api/payers/{payer_id}/summary` | 4230 |
| admin | GET | `/api/admin/audit-log` | 3372 |
| admin | GET | `/api/admin/audit-log/verify` | 3414 |
| admin | POST | `/api/admin/db/rotate-key` | 3454 |
| admin | POST | `/api/admin/backup/create` | 3614 |
| admin | GET | `/api/admin/backup/list` | 3668 |
| admin | GET | `/api/admin/backup/status` | 3697 |
| admin | POST | `/api/admin/backup/{backup_id}/verify` | 3711 |
| admin | POST | `/api/admin/backup/{backup_id}/restore/initiate` | 3730 |
| admin | POST | `/api/admin/backup/{backup_id}/restore/confirm` | 3758 |
| admin | POST | `/api/admin/backup/prune` | 3810 |
| admin | POST | `/api/admin/backup/scheduler/start` | 3838 |
| admin | POST | `/api/admin/backup/scheduler/stop` | 3849 |
| admin | POST | `/api/admin/backup/scheduler/tick` | 3860 |
| admin | POST | `/api/admin/scheduler/start` | 3894 |
| admin | POST | `/api/admin/scheduler/stop` | 3902 |
| admin | POST | `/api/admin/scheduler/tick` | 3910 |
| admin | POST | `/api/admin/scheduler/pull-inbound` | 3923 |
| admin | GET | `/api/admin/scheduler/status` | 4045 |
| admin | GET | `/api/admin/scheduler/processed-files` | 4052 |
| admin | POST | `/api/admin/reload-config` | 4316 |
### 3.3 `_shared.py` surface (12 helpers)
| Helper | Routers that use it | Source line in api.py |
|---|---|---|
| `_actor_user_id(request)` | most | 54 |
| `_resolve_payer(name)` | parse (837, 999, ta1, 277ca) | 333 |
| `_resolve_payer_835(name)` | parse (835) | 345 |
| `_transaction_set_id_from_segments(segments)` | parse (999, ta1) | 357 |
| `_build_and_persist_ack(batch_id)` | parse (after 837) | 573 |
| `_reconciliation_summary_for_batch(batch_id)` | parse (after 837) | 615 |
| `_ta1_synthetic_source_batch_id(icn)` | parse (ta1) | 975 |
| `_serialize_ta1(result)` | acks | 1324 |
| `_serialize_ta1_from_row(row)` | acks | 1341 |
| `_batch_summary_claim_count(rec)` | batches | 1858 |
| `_batch_summary_claim_ids(rec)` | batches | 1867 |
| `_batch_summary_billing_outcomes(...)` | batches | 1912 |
### 3.4 Stays in `api.py`
- `app = FastAPI(...)` instantiation
- `lifespan` (init_db, scheduler, bus, sessions, etc.)
- `@app.exception_handler(HTTPException)` at 313
- `@app.exception_handler(Exception)` at 386
- The trailing `from cyclone.auth.routes import router as auth_router; app.include_router(auth_router)` block
- The `__all__ = ["app"]` line at the end
### 3.5 Stays in `api_helpers.py`
- `_ndjson_line`
- `_tail_events`
### 3.6 External import surface — what callers see
A grep over the codebase for `from cyclone.api import` and `import cyclone.api` will return the same set of symbols after SP36 lands as before:
- `from cyclone.api import app` — used by `python -m cyclone serve`, by tests (`from cyclone import api` then `app.state.*`), by the test `conftest.py`.
- `app.state.event_bus`, `app.state.db`, `app.state.scheduler` — read by tests and by `api_routers/` (via `Request`).
The new `cyclone.api_routers` package exports `routers: list[APIRouter]`. Anything in `api_routers/` that needs to be imported from outside (e.g. a router-level fixture in a future test) is reachable as `from cyclone.api_routers.parse import router`, etc.
---
## 4. Data flow & error handling
**Data flow (per request):**
1. FastAPI receives the request, runs the `matrix_gate` dependency (auth + role check), then dispatches to the matching route in the matching router.
2. The route handler validates inputs (path params, query params, body via Pydantic if a body model exists).
3. The handler calls `cyclone.store.<method>(...)` for write paths and reconciliation, or opens a `db.SessionLocal()() for one-off reads` (existing pattern; this is how the dashboard KPIs, the activity feed, the configuration endpoints, and the live-tail snapshot reads work today).
4. The result is serialized via `to_ui_<entity>` (from `cyclone.store.ui`) for entity-shaped responses, or via a small local serializer for non-entity shapes (e.g. the dashboard KPI shape, the reconciliation summary, the eligibility 271 response).
5. The handler returns `JSONResponse`, `StreamingResponse` (NDJSON), or `Response` (CSV for `/api/inbox/export.csv`).
6. **Streaming endpoints** (`/api/claims/stream`, `/api/remittances/stream`, `/api/activity/stream`) keep the snapshot-then-live-subscription two-phase pattern. `_ndjson_line` and `_tail_events` come from `api_helpers.py`; the matching router imports them.
7. **The event bus contract is unchanged.** Every write through `cyclone.store.add(...)` still publishes `claim_written` / `remittance_written` / `activity_recorded` on the in-process `EventBus`. The stream-endpoint routers subscribe to those exact event names. The SP doesn't introduce new event kinds, doesn't rename existing kinds, doesn't change payloads.
**Error handling:**
- Routers raise `HTTPException` for client errors. Same status codes, same `detail` strings, same `error` envelope (`{error, detail}`). No change.
- The two global exception handlers stay in `api.py` (D2 of brainstorming): `@app.exception_handler(HTTPException)` re-emits with the cyclone error envelope, `@app.exception_handler(Exception)` logs and returns 500.
- Streaming endpoints catch domain exceptions and yield `{"type": "error", "data": {"message": ...}}` NDJSON chunks instead of raising — existing pattern, preserved per router.
- `StoreNotFound` / `StoreConflict` / `StoreInvalidState` are translated to 404 / 409 / 409 by the existing translation path. No change.
- The auth path (`matrix_gate` dep) returns 401/403 unchanged.
**No behavior change anywhere in the data flow or error path.** The SP is a structural move, not a correctness pass.
---
## 5. Testing approach
### 5.1 Per-router cycle (live-test + autoreview + commit)
For each router extracted (one `feat(sp36): extract {name} router` commit):
1. **Pre-flight pytest baseline.** From the project root:
```
cd backend && .venv/bin/pytest --tb=line -q 2>&1 | tee /tmp/refactor-{N}-pre.txt | tail -3
```
Capture the `XXX passed, YY failed, ZZ skipped` line. Expect roughly `~1,176 passed, ~1 failed (pre-existing isolation flake), ~10 skipped` per the SP21 store-split spec baseline.
2. **Cut.** Move the route handler(s) + their single-router helpers to the new router module, register the router in `api_routers/__init__.py`, delete the moved code from `api.py`, update imports. No logic change.
3. **Post-cut pytest diff.** Same command as (1), output to `/tmp/refactor-{N}-post.txt`. Diff:
```
diff <(awk '/passed|failed|skipped/ {print}' /tmp/refactor-{N}-pre.txt) \
<(awk '/passed|failed|skipped/ {print}' /tmp/refactor-{N}-post.txt)
```
Required: zero diff. Any delta = revert the cut and investigate.
4. **Live test.** Server is running in `cyclone-backend-1` on the host. Hit one representative route from the moved router. Required status codes:
- GET endpoints → 200 (or the documented non-200, e.g. 401 if auth is mocked off in the test environment)
- POST endpoints → 200, 201, 400, 401, 403, 409 (whatever the existing happy path / known-error path is)
- Streaming endpoints → 200 with `Content-Type: application/x-ndjson` and a valid first `snapshot_end` chunk
5. **Autoreview.** Spawn a `pr-reviewer` subagent on the staged diff. The reviewer prompt enumerates: (a) the new router is registered in `api_routers/__init__.py`; (b) no orphan imports left in `api.py`; (c) no `session.add` / `s.add` / `session.commit` / direct `db.SessionLocal()() call that does a write` appears in any router file (reads are OK); (d) the router's `dependencies=[Depends(matrix_gate)]` matches the original; (e) docstrings are preserved. Reviewer returns a one-line verdict (PASS / FAIL: <reason>). On FAIL, fix and re-review.
6. **Commit.** `feat(sp36): extract {name} router` per the SP-N convention.
7. **Log** to `/tmp/refactor-cyclone.md`:
```
[step N] extracted {name}
pre: passed={X} failed={Y} skipped={Z}
post: passed={X} failed={Y} skipped={Z}
live: GET /api/{route} → {code} | POST /api/{route} → {code}
reviewer: PASS | FAIL: <reason>
```
### 5.2 Pre-merge integration test
After all 19 routers are extracted, the full suite runs:
- `cd backend && .venv/bin/pytest --tb=line -q` — full backend suite, must match Step 0 baseline to the digit
- `cd backend && .venv/bin/pytest tests/test_api_*.py -v` — every API integration test, all green
- `cd .. && npm test` — frontend suite, all green
- `cd .. && npm run typecheck` — frontend typecheck clean
- `cd .. && npm run build` — frontend build green
- `cd .. && npm run lint` — frontend lint clean
### 5.3 Pre-merge one-shot verification (D8 of brainstorming)
These commands run, and all must pass, before the merge commit is created:
- `wc -l backend/src/cyclone/api.py` — expect ≤ 300
- `wc -l backend/src/cyclone/api_routers/*.py | sort -n` — expect no router over 1,400 LOC (admin is the largest at ~1,200)
- `grep -rn "session.add\|s\.add(\|session\.commit" backend/src/cyclone/api_routers/` — expect zero hits (writes go through `cyclone.store` only)
- `grep -rn "from cyclone.api_routers" backend/src/cyclone/api.py` — expect one hit (the `routers` import in `__init__` block)
- `python -c "from cyclone.api_routers import routers; print(len(routers))"` — expect ≥ 18 (5 existing + 14 new, minus whatever routers ended up merged during the SP)
- Live matrix — for every URL prefix in Section 3.2, `curl -s -o /dev/null -w "%{http_code}\n" http://192.168.0.49:8080/api/{prefix}/...` returns 2xx (or the documented error code)
- `git grep -n "@app\." backend/src/cyclone/api.py` — expect zero hits (every route decorator now lives in a router)
### 5.4 No new tests
The existing test suite is the regression net. We do not add new tests, do not modify existing tests, do not add new fixtures. The pre/post pytest baseline diff is the test for the SP.
---
## 6. Threat model (post-SP24 alignment)
The auth boundary is the HTTP layer: login required, bcrypt + HttpOnly session cookie. The new routers mount under `matrix_gate` (same as today), so every moved route stays auth-gated. No route is exposed that wasn't exposed before. No auth check is weakened.
The file-system threat model is unchanged: SQLCipher at rest (when the macOS Keychain entry + `sqlcipher3` are both present, otherwise plain SQLite), secrets in the macOS Keychain via `keyring`, no secrets on disk in plaintext. SP36 doesn't touch the DB layer, the secrets module, or the configuration loader.
LAN-only by design. The auth boundary is the network boundary; the production deployment is the host firewall + compose port publishing. Don't expose the published ports to the WAN — unchanged.
---
## 7. Risks & mitigations
| Risk | Likelihood | Impact | Mitigation |
|---|---|---|---|
| A route handler silently changes behavior during the cut (typo, off-by-one, wrong import) | medium | high | Pre/post pytest diff is the test. The diff must be zero; any delta reverts. |
| A helper gets duplicated instead of moved (one copy in old location, one in new) | low | medium | Autoreview explicitly checks for orphan imports in `api.py` and for `def _helper_name` definitions appearing twice. |
| The `api_routers/__init__.py` registration misses a router and a route 404s in production | low | high | Pre-merge one-shot verification §5.3 includes a live matrix that hits every URL prefix. |
| The `matrix_gate` dep is dropped from one route during the cut | low | critical | Autoreview explicitly checks `dependencies=[Depends(matrix_gate)]` matches the original on every moved route. |
| Two routers end up importing each other (creates a cycle) | low | medium | The convention is "routers import only `cyclone.store`, `cyclone.api_helpers`, `cyclone.api_routers._shared`" — autoreview checks for any other cross-router import. |
| A `from cyclone.api import` somewhere in the codebase breaks because we removed a top-level symbol | low | high | The external import surface (D6 of brainstorming) is the test: `git grep "from cyclone.api import"` before and after must be identical in symbol set. |
| The merge of SP35 into main fails or introduces conflicts | low | high | D7 of brainstorming: SP35 merge is a hard pre-condition; if it doesn't land cleanly, SP36 doesn't start. |
| A streaming endpoint regresses silently (returns the right status code but the wrong first chunk) | medium | medium | The live test for streaming endpoints asserts `Content-Type: application/x-ndjson` and parses the first `snapshot_end` line, not just the HTTP code. |
| The frontend test suite picks up a backend change (e.g. a renamed JSON field that the UI consumes) | very low | medium | The refactor is byte-for-byte transparent to clients; if a frontend test fails, that's a real regression and we investigate. |
---
## 8. Rollout
1. **Merge SP35 into main.** Fast-forward or merge commit (no squash, no rebase). The branch `sp35-parse-input-guards` has 4 reviewable commits and a clean working tree.
2. **Restart compose if needed.** `docker compose restart cyclone-backend-1` after the SP35 merge so the running container reflects main.
3. **Create the SP36 branch.** `git checkout -b sp36-api-routers-split` from main.
4. **Extract routers, one per commit.** Order: lowest-risk first (admin, health, claim_acks, ta1_acks already exist; then the small leaf routers `dashboard.py`, `eligibility.py`, `payers.py`, `config.py`; then the medium ones `reconciliation.py`, `remittances.py`, `activity.py`, `clearhouse.py`, `providers.py`; then the larger ones `inbox.py`, `batches.py`, `claims.py`; finally `parse.py` since it's the most coupled to the store facade). Each step follows §5.1.
5. **Run the §5.2 integration tests.**
6. **Run the §5.3 one-shot verification.**
7. **Open the PR.** Title: `SP36 API Routers Split`. Body: links to this spec, the SP36 plan (when written), and a checklist of every router extracted.
8. **Merge after review.** Single atomic `merge: SP36 api-routers-split into main` commit, no squash, no rebase.
---
## 9. Open questions
None at spec time. The brainstorming Q&A resolved all of:
- Which file to split? `api.py`. Confirmed.
- What shape? Per-resource routers mirroring the store's domain boundaries. Confirmed.
- Branch from where? Merge SP35 first, then from main. Confirmed.
- Where do the cross-router helpers go? `api_routers/_shared.py` (private). Confirmed.
- `admin.py` — one router or split further? One router for this SP, follow-up later. Confirmed.
- How granular are the commits? One router per commit. Confirmed.