3 changed files with 577 additions and 425 deletions
@@ -31,16 +31,9 @@ from cyclone import __version__, db
 from cyclone.db import Claim, ClaimState, Remittance
 from cyclone.inbox_state import apply_999_rejections
 from cyclone.inbox_state_277ca import apply_277ca_rejections
-from cyclone.audit_log import AuditEvent, append_event, verify_chain
+from cyclone.audit_log import AuditEvent, append_event
 from cyclone.parsers.exceptions import CycloneParseError
 from cyclone.parsers.models import BatchSummary, Envelope
 from cyclone.parsers.models_270 import (
    EligibilityBenefitInquiry,
    InformationReceiver270,
    InformationSource270,
    ParseResult270,
    Subscriber270,
 )
 from cyclone.parsers.models_271 import ParseResult271
 from cyclone.parsers.models_835 import ParseResult835
 from cyclone.parsers.payer import PayerConfig, PayerConfig835
@@ -51,7 +44,6 @@ from cyclone.parsers.parse_837 import parse
 from cyclone.parsers.parse_835 import parse as parse_835
 from cyclone.parsers.parse_999 import parse_999_text
 from cyclone.parsers.parse_ta1 import parse_ta1_text
 from cyclone.parsers.serialize_270 import serialize_270
 from cyclone.parsers.serialize_999 import serialize_999
 from cyclone.parsers.batch_ack_builder import build_ack_for_batch
@@ -142,10 +134,12 @@ app.add_middleware(
 from cyclone.api_routers import (  # noqa: E402
    acks,
    activity,
    admin,
    batches,
    claims,
    clearhouse,
    config,
    eligibility,
    health,
    inbox,
    providers,
@@ -166,6 +160,8 @@ app.include_router(claims.router)
 app.include_router(remittances.router)
 app.include_router(inbox.router)
 app.include_router(reconciliation.router)
 app.include_router(eligibility.router)
 app.include_router(admin.router)
 # Backwards-compat re-exports: test_api_stream_live.py imports these
 # names directly from ``cyclone.api`` to invoke the endpoint coroutines
@@ -175,6 +171,24 @@ activity_stream = activity.activity_stream_endpoint
 claims_stream = claims.claims_stream_endpoint
 remittances_stream = remittances.remittances_stream_endpoint
 # Backwards-compat lock re-export: test_api_rotate_key.py does
 # ``api_mod._db_rotate_lock.acquire()`` to simulate a concurrent
 # rotation request. The lock itself lives in the admin router (it
 # guards that router's rotate-key endpoint) but we expose the same
 # object here so the test holds the same lock the endpoint does.
 _db_rotate_lock = admin._db_rotate_lock
 # Backwards-compat module re-exports: test_api_rotate_key.py does
 # ``monkeypatch.setattr("cyclone.api._db_crypto.rotate_db_key", ...)``
 # and ``monkeypatch.setattr("cyclone.api._secrets.set_secret", ...)``
 # to stub crypto/keychain calls. Re-exporting here keeps the
 # ``cyclone.api._db_crypto`` / ``cyclone.api._secrets`` paths alive
 # so the monkeypatch targets the same module objects the admin
 # endpoint actually uses (both are the same ``cyclone.db_crypto`` /
 # ``cyclone.secrets`` module instances).
 _db_crypto = admin.db_crypto
 _secrets = admin.secrets
 def _resolve_payer(name: str) -> PayerConfig:
    if name not in PAYER_FACTORIES:
@@ -926,424 +940,10 @@ def _serialize_ta1_from_row(row: db.Ta1Ack) -> str:
 # --------------------------------------------------------------------------- #
-def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
+# --- /api/eligibility/* lives in cyclone.api_routers.eligibility ---
    """Build a :class:`ParseResult270` from a request body dict.
-    The body shape is the minimum surface needed to build a valid 270
+# --- /api/admin/* lives in cyclone.api_routers.admin ---
    inquiry (per spec section 3.4 — operator-driven, ephemeral):
    ::
        {
            "subscriber": {first_name, last_name, member_id, dob},
            "provider":   {npi, name},
            "payer":      {id, name},
            "service_type_code": "1"
        }
    Returns ``(ParseResult270, service_type_code)``. Raises
    :class:`HTTPException` (400) when the body is missing required
    fields.
    """
    subscriber_in = body.get("subscriber") or {}
    provider_in = body.get("provider") or {}
    payer_in = body.get("payer") or {}
    service_type_code = (body.get("service_type_code") or "").strip()
    # Required-field checks. We surface a single 400 with the first
    # missing field name to match the rest of the API's error contract.
    if not service_type_code:
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "service_type_code is required"},
        )
    if not subscriber_in.get("member_id"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
        )
    if not provider_in.get("npi"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "provider.npi is required"},
        )
    if not payer_in.get("name"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "payer.name is required"},
        )
    # Build the Pydantic models. The serializer handles all envelope
    # generation (sender_id/receiver_id/control_number/transaction_date
    # are filled in by the serializer with sensible defaults).
    from datetime import date as _date
    subscriber_dob_raw = subscriber_in.get("dob")
    subscriber_dob: _date | None = None
    if subscriber_dob_raw:
        try:
            subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
        except (TypeError, ValueError) as exc:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": "Bad request",
                    "detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
                },
            ) from exc
    result = ParseResult270(
        envelope=Envelope(
            sender_id="SUBMITTERID",
            receiver_id=str(payer_in.get("id") or "RECEIVERID"),
            control_number="000000001",
            transaction_date=_date.today(),
            implementation_guide="005010X279A1",
        ),
        information_source=InformationSource270(
            name=str(payer_in["name"]),
            id=str(payer_in.get("id") or "") or None,
        ),
        information_receiver=InformationReceiver270(
            name=str(provider_in.get("name") or ""),
            npi=str(provider_in["npi"]),
        ),
        subscriber=Subscriber270(
            member_id=str(subscriber_in["member_id"]),
            first_name=str(subscriber_in.get("first_name") or "") or None,
            last_name=str(subscriber_in.get("last_name") or "") or None,
            dob=subscriber_dob,
        ),
        inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
        summary=BatchSummary(
            input_file="eligibility_request",
            control_number="000000001",
            transaction_date=_date.today(),
            total_claims=1,
            passed=1,
            failed=0,
        ),
    )
    return result, service_type_code
@app.post("/api/eligibility/request")
 def post_eligibility_request(body: dict) -> Any:
    """Build a 270 eligibility inquiry from a small JSON body.
    Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
    so the operator can either download the raw text (paste into a
    payer portal) or render the parsed fields directly. Per spec
    section 3.4, nothing is persisted to the DB.
    """
    try:
        result, _ = _validate_eligibility_request(body)
    except HTTPException:
        raise
    except (KeyError, TypeError, ValueError) as exc:
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
        ) from exc
    raw_270_text = serialize_270(result)
    return {
        "raw_270_text": raw_270_text,
        "parsed": json.loads(result.model_dump_json()),
    }
@app.post("/api/eligibility/parse-271")
 async def post_eligibility_parse_271(
    file: UploadFile = File(...),
 ) -> Any:
    """Parse a 271 eligibility response and return the structured summary.
    Accepts the raw 271 text as a file upload (multipart/form-data),
    mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
    result is NOT persisted — the operator re-pastes the 271 each
    time they need a fresh read.
    The response body is a JSON object with three top-level keys:
    ``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
    returned on empty / undecodable / malformed EDI; 200 on success.
    """
    raw = await file.read()
    if not raw:
        return JSONResponse(
            status_code=400,
            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
        )
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        return JSONResponse(
            status_code=400,
            content={"error": "Encoding error", "detail": str(exc)},
        )
    try:
        result = parse_271_text(text, input_file=file.filename or "")
    except CycloneParseError as exc:
        return JSONResponse(
            status_code=400,
            content={"error": "Parse error", "detail": str(exc)},
        )
    except Exception as exc:  # pragma: no cover - safety net
        log.exception("Unexpected parser failure on 271")
        return JSONResponse(
            status_code=500,
            content={"error": "Internal server error", "detail": str(exc)},
        )
    return {
        "coverage_benefits": [
            json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
        ],
        "subscriber": json.loads(result.subscriber.model_dump_json()),
        "summary": json.loads(result.summary.model_dump_json()),
        "envelope": json.loads(result.envelope.model_dump_json()),
        "information_source": json.loads(result.information_source.model_dump_json()),
        "information_receiver": json.loads(result.information_receiver.model_dump_json()),
    }
 # --------------------------------------------------------------------------- #
 # SP11: tamper-evident audit log (admin)
 # --------------------------------------------------------------------------- #
@app.get("/api/admin/audit-log")
 def list_audit_log_endpoint(
    entity_type: str | None = Query(default=None),
    entity_id: str | None = Query(default=None),
    event_type: str | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=1000),
 ) -> Any:
    """List audit-log rows, newest first, with optional filters.
    Filters match the (entity_type, entity_id) pair (typical use:
    "show me everything that happened to claim C-123") or a single
    event_type (typical use: "show me all clearhouse.submitted
    events today").
    """
    with db.SessionLocal()() as s:
        q = s.query(db.AuditLog)
        if entity_type:
            q = q.filter(db.AuditLog.entity_type == entity_type)
        if entity_id:
            q = q.filter(db.AuditLog.entity_id == entity_id)
        if event_type:
            q = q.filter(db.AuditLog.event_type == event_type)
        rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
        return {
            "total": len(rows),
            "items": [
                {
                    "id": r.id,
                    "event_type": r.event_type,
                    "entity_type": r.entity_type,
                    "entity_id": r.entity_id,
                    "actor": r.actor,
                    "payload": json.loads(r.payload_json) if r.payload_json else None,
                    "created_at": r.created_at.isoformat() if r.created_at else None,
                    "prev_hash": r.prev_hash,
                    "hash": r.hash,
                }
                for r in rows
            ],
        }
@app.get("/api/admin/audit-log/verify")
 def verify_audit_log_endpoint() -> Any:
    """Walk the audit-log chain and verify every row's hash.
    Returns ``{"ok": true, "checked": N}`` for a clean chain, or
    ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
    for a broken chain. This is the operator's "did anyone tamper?"
    endpoint; run it on demand or via a nightly cron job.
    """
    with db.SessionLocal()() as s:
        result = verify_chain(s)
    return {
        "ok": result.ok,
        "checked": result.checked,
        "first_bad_id": result.first_bad_id,
        "reason": result.reason,
    }
 # ---------------------------------------------------------------------------
 # SP15: SQLCipher key rotation
 #
 # Re-encrypts the DB in place with a fresh key, then updates the
 # Keychain so subsequent connections open with the new key. This is
 # a 1-time operation per rotation; for routine read/write the rest
 # of the API is unchanged.
 #
 # Concurrency: the rotation holds a module-level lock so two
 # concurrent requests can't race and end up with mismatched Keychain
 # + DB. The lock is a simple threading.Lock; a process restart
 # resets it (intentional — the operator's next start-up opens with
 # whatever key is in the Keychain).
 # ---------------------------------------------------------------------------
 import threading as _threading
 from cyclone import db_crypto as _db_crypto
 from cyclone import secrets as _secrets
 _db_rotate_lock = _threading.Lock()
@app.post("/api/admin/db/rotate-key")
 def rotate_db_key_endpoint(body: dict | None = None) -> Any:
    """Generate a fresh DB key, re-encrypt the DB, update the Keychain.
    Request body (optional):
        actor: who initiated the rotation. Defaults to "operator".
        reason: human-readable reason. Written to the audit log.
    Returns:
        ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
        on success. On failure (DB not encrypted, rekey failed,
        Keychain update failed) returns the same shape with
        ``ok=false`` and a ``reason``. HTTP 503 is returned if the
        rekey fails or encryption is not enabled.
    The Keychain write happens *after* the rekey succeeds. If the
    Keychain write fails, the DB has the new key but the Keychain
    still has the old one — the endpoint returns 503 with a
    "keychain update failed" reason and the operator must restore
    the old key manually (``cyclone db restore-key <old_key>``) to
    avoid being locked out.
    """
    body = body or {}
    actor = body.get("actor") or "operator"
    reason = body.get("reason") or ""
    if not _db_crypto.is_encryption_enabled():
        raise HTTPException(
            status_code=400,
            detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
        )
    # Acquire the lock; non-blocking so a stuck rotation doesn't
    # silently hold up other requests.
    if not _db_rotate_lock.acquire(blocking=False):
        raise HTTPException(
            status_code=409,
            detail="another key rotation is in progress",
        )
    try:
        url = db._resolve_url()
        old_key = _db_crypto.get_db_key()
        if not old_key:
            raise HTTPException(
                status_code=400,
                detail="no DB key in Keychain; cannot rotate",
            )
        new_key = _db_crypto.generate_db_key()
        result = _db_crypto.rotate_db_key(
            url=url, old_key=old_key, new_key=new_key,
        )
        if not result.ok:
            # Rekey failed. The DB still has the old key. The
            # Keychain is unchanged. Caller should NOT retry with
            # the same new key (it's lost); generate a fresh one.
            log.error("SQLCipher rotate failed: %s", result.reason)
            raise HTTPException(
                status_code=503,
                detail={
                    "ok": False,
                    "old_fingerprint": result.old_fingerprint,
                    "new_fingerprint": result.new_fingerprint,
                    "rotated_at": result.rotated_at,
                    "reason": result.reason,
                },
            )
        # Rekey succeeded. Now update the Keychain. If this fails
        # the DB is locked behind the new key — operator must
        # restore the old key manually.
        if not _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT, new_key):
            log.error("Keychain update failed after successful rekey!")
            raise HTTPException(
                status_code=503,
                detail={
                    "ok": False,
                    "old_fingerprint": result.old_fingerprint,
                    "new_fingerprint": result.new_fingerprint,
                    "rotated_at": result.rotated_at,
                    "reason": (
                        "rekey succeeded but Keychain update failed — "
                        "the DB is now encrypted with the new key but "
                        "the Keychain still has the old one. "
                        "Restore the old key to the Keychain to recover."
                    ),
                },
            )
        # Store the old key in the "previous" account for a grace
        # period so the operator can roll back if they discover the
        # new key is broken (e.g. the Keychain entry got truncated).
        _secrets.set_secret(_db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
        # Rebuild the engine so subsequent connections use the new
        # key. dispose_engine() closes every pooled connection that
        # was using the old key; init_db() opens new ones with the
        # new key from the (now-updated) Keychain.
        db.reinit_engine()
        # Audit log the rotation. We do this after the engine is
        # rebuilt so the audit event is written with the new key —
        # proving that the new key works for new writes.
        try:
            from cyclone.audit_log import append_event, AuditEvent
            with db.SessionLocal()() as s:
                append_event(s, AuditEvent(
                    event_type="db.key_rotated",
                    entity_type="database",
                    entity_id="cyclone.db",
                    actor=actor,
                    payload={
                        "old_fingerprint": result.old_fingerprint,
                        "new_fingerprint": result.new_fingerprint,
                        "table_count": result.table_count,
                        "reason": reason,
                    },
                ))
                s.commit()
        except Exception as exc:  # noqa: BLE001
            # Audit append is best-effort; rotation already succeeded.
            log.warning("could not write audit event for rotation: %s", exc)
        return {
            "ok": True,
            "old_fingerprint": result.old_fingerprint,
            "new_fingerprint": result.new_fingerprint,
            "rotated_at": result.rotated_at,
            "table_count": result.table_count,
        }
    finally:
        _db_rotate_lock.release()
@app.post("/api/admin/reload-config")
 def reload_config():
    """Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
    from cyclone import payers as payer_loader
    try:
        configs = payer_loader.load_payer_configs()
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return {"ok": True, "loaded": len(configs), "errors": []}
 __all__ = ["app"]
@@ -0,0 +1,332 @@
 """``/api/admin/*`` — operator-facing admin surface (SP11 + SP15).
 Four endpoints:
 * ``GET /api/admin/audit-log`` — list audit-log rows with optional
  ``entity_type`` / ``entity_id`` / ``event_type`` filters (SP11).
 * ``GET /api/admin/audit-log/verify`` — walk the audit-log hash chain
  and verify every row (SP11).
 * ``POST /api/admin/db/rotate-key`` — SP15 SQLCipher key rotation.
  Re-encrypts the DB in place with a fresh key, updates the Keychain.
  Protected by a module-level ``threading.Lock`` so two concurrent
  rotations can't race (a single process restart resets the lock —
  intentional, the next start-up opens with whatever key is in the
  Keychain).
 * ``POST /api/admin/reload-config`` — re-read ``config/payers.yaml``
  and revalidate; returns the number of configs loaded.
 """
 from __future__ import annotations
 import json
 import logging
 import threading
 from typing import Any
 from fastapi import APIRouter, HTTPException, Query
 from cyclone import db, db_crypto, secrets
 from cyclone.audit_log import AuditEvent, append_event, verify_chain
 log = logging.getLogger(__name__)
 router = APIRouter()
 # Module-level lock: protects against concurrent key rotations within
 # the same process. A process restart resets it (intentional — the
 # operator's next start-up opens with whatever key is in the Keychain).
 _db_rotate_lock = threading.Lock()
@router.get("/api/admin/audit-log")
 def list_audit_log_endpoint(
    entity_type: str | None = Query(default=None),
    entity_id: str | None = Query(default=None),
    event_type: str | None = Query(default=None),
    limit: int = Query(default=100, ge=1, le=1000),
 ) -> Any:
    """List audit-log rows, newest first, with optional filters.
    Filters match the (entity_type, entity_id) pair (typical use:
    "show me everything that happened to claim C-123") or a single
    event_type (typical use: "show me all clearhouse.submitted
    events today").
    """
    with db.SessionLocal()() as s:
        q = s.query(db.AuditLog)
        if entity_type:
            q = q.filter(db.AuditLog.entity_type == entity_type)
        if entity_id:
            q = q.filter(db.AuditLog.entity_id == entity_id)
        if event_type:
            q = q.filter(db.AuditLog.event_type == event_type)
        rows = q.order_by(db.AuditLog.id.desc()).limit(limit).all()
        return {
            "total": len(rows),
            "items": [
                {
                    "id": r.id,
                    "event_type": r.event_type,
                    "entity_type": r.entity_type,
                    "entity_id": r.entity_id,
                    "actor": r.actor,
                    "payload": json.loads(r.payload_json) if r.payload_json else None,
                    "created_at": r.created_at.isoformat() if r.created_at else None,
                    "prev_hash": r.prev_hash,
                    "hash": r.hash,
                }
                for r in rows
            ],
        }
@router.get("/api/admin/audit-log/verify")
 def verify_audit_log_endpoint() -> Any:
    """Walk the audit-log chain and verify every row's hash.
    Returns ``{"ok": true, "checked": N}`` for a clean chain, or
    ``{"ok": false, "checked": K, "first_bad_id": X, "reason": "..."}``
    for a broken chain. This is the operator's "did anyone tamper?"
    endpoint; run it on demand or via a nightly cron job.
    """
    with db.SessionLocal()() as s:
        result = verify_chain(s)
    return {
        "ok": result.ok,
        "checked": result.checked,
        "first_bad_id": result.first_bad_id,
        "reason": result.reason,
    }
@router.post("/api/admin/db/rotate-key")
 def rotate_db_key_endpoint(body: dict | None = None) -> Any:
    """Generate a fresh DB key, re-encrypt the DB, update the Keychain.
    Request body (optional):
        actor: who initiated the rotation. Defaults to "operator".
        reason: human-readable reason. Written to the audit log.
    Returns:
        ``{ok, old_fingerprint, new_fingerprint, rotated_at, table_count}``
        on success. On failure (DB not encrypted, rekey failed,
        Keychain update failed) returns the same shape with
        ``ok=false`` and a ``reason``. HTTP 503 is returned if the
        rekey fails or encryption is not enabled.
    The Keychain write happens *after* the rekey succeeds. If the
    Keychain write fails, the DB has the new key but the Keychain
    still has the old one — the endpoint returns 503 with a
    "keychain update failed" reason and the operator must restore
    the old key manually (``cyclone db restore-key <old_key>``) to
    avoid being locked out.
    """
    body = body or {}
    actor = body.get("actor") or "operator"
    reason = body.get("reason") or ""
    if not db_crypto.is_encryption_enabled():
        raise HTTPException(
            status_code=400,
            detail="encryption not enabled (sqlcipher3 missing or no Keychain key)",
        )
    # Acquire the lock; non-blocking so a stuck rotation doesn't
    # silently hold up other requests.
    if not _db_rotate_lock.acquire(blocking=False):
        raise HTTPException(
            status_code=409,
            detail="another key rotation is in progress",
        )
    try:
        url = db._resolve_url()
        old_key = db_crypto.get_db_key()
        if not old_key:
            raise HTTPException(
                status_code=400,
                detail="no DB key in Keychain; cannot rotate",
            )
        new_key = db_crypto.generate_db_key()
        result = db_crypto.rotate_db_key(
            url=url, old_key=old_key, new_key=new_key,
        )
        if not result.ok:
            # Rekey failed. The DB still has the old key. The
            # Keychain is unchanged. Caller should NOT retry with
            # the same new key (it's lost); generate a fresh one.
            log.error("SQLCipher rotate failed: %s", result.reason)
            raise HTTPException(
                status_code=503,
                detail={
                    "ok": False,
                    "old_fingerprint": result.old_fingerprint,
                    "new_fingerprint": result.new_fingerprint,
                    "rotated_at": result.rotated_at,
                    "reason": result.reason,
                },
            )
        # Rekey succeeded. Now update the Keychain. If this fails
        # the DB is locked behind the new key — operator must
        # restore the old key manually.
        if not secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT, new_key):
            log.error("Keychain update failed after successful rekey!")
            raise HTTPException(
                status_code=503,
                detail={
                    "ok": False,
                    "old_fingerprint": result.old_fingerprint,
                    "new_fingerprint": result.new_fingerprint,
                    "rotated_at": result.rotated_at,
                    "reason": (
                        "rekey succeeded but Keychain update failed — "
                        "the DB is now encrypted with the new key but "
                        "the Keychain still has the old one. "
                        "Restore the old key to the Keychain to recover."
                    ),
                },
            )
        # Store the old key in the "previous" account for a grace
        # period so the operator can roll back if they discover the
        # new key is broken (e.g. the Keychain entry got truncated).
        secrets.set_secret(db_crypto.KEYCHAIN_ACCOUNT_PREVIOUS, old_key)
        # Rebuild the engine so subsequent connections use the new
        # key. dispose_engine() closes every pooled connection that
        # was using the old key; init_db() opens new ones with the
        # new key from the (now-updated) Keychain.
        db.reinit_engine()
        # Audit log the rotation. We do this after the engine is
        # rebuilt so the audit event is written with the new key —
        # proving that the new key works for new writes.
        try:
            with db.SessionLocal()() as s:
                append_event(s, AuditEvent(
                    event_type="db.key_rotated",
                    entity_type="database",
                    entity_id="cyclone.db",
                    actor=actor,
                    payload={
                        "old_fingerprint": result.old_fingerprint,
                        "new_fingerprint": result.new_fingerprint,
                        "table_count": result.table_count,
                        "reason": reason,
                    },
                ))
                s.commit()
        except Exception as exc:  # noqa: BLE001
            # Audit append is best-effort; rotation already succeeded.
            log.warning("could not write audit event for rotation: %s", exc)
        return {
            "ok": True,
            "old_fingerprint": result.old_fingerprint,
            "new_fingerprint": result.new_fingerprint,
            "rotated_at": result.rotated_at,
            "table_count": result.table_count,
        }
    finally:
        _db_rotate_lock.release()
@router.post("/api/admin/reload-config")
 def reload_config():
    """Re-read ``config/payers.yaml`` and revalidate. Returns counts."""
    from cyclone import payers as payer_loader
    try:
        configs = payer_loader.load_payer_configs()
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return {"ok": True, "loaded": len(configs), "errors": []}
 # ---------------------------------------------------------------------------
 # SP16: inbound MFT scheduler
 # ---------------------------------------------------------------------------
@router.post("/api/admin/scheduler/start")
 async def scheduler_start():
    """Start the background MFT polling loop (idempotent)."""
    from cyclone import scheduler as sched_mod
    sched = sched_mod.get_scheduler()
    await sched.start()
    return {"ok": True, "status": sched.status().as_dict()}
@router.post("/api/admin/scheduler/stop")
 async def scheduler_stop():
    """Stop the background MFT polling loop. Waits for the current tick."""
    from cyclone import scheduler as sched_mod
    sched = sched_mod.get_scheduler()
    await sched.stop()
    return {"ok": True, "status": sched.status().as_dict()}
@router.get("/api/admin/scheduler/status")
 def scheduler_status():
    """Return the scheduler's running state + last tick summary."""
    from cyclone import scheduler as sched_mod
    sched = sched_mod.get_scheduler()
    return sched.status().as_dict()
@router.post("/api/admin/scheduler/tick")
 async def scheduler_tick():
    """Force a single poll cycle. Returns the TickResult.
    Useful for the operator's "run now" button or for testing. If a
    scheduled tick is already in flight, this call coalesces onto it
    (waits for it to finish, returns its result).
    """
    from cyclone import scheduler as sched_mod
    sched = sched_mod.get_scheduler()
    result = await sched.tick()
    return {"ok": True, "tick": result.as_dict()}
@router.get("/api/admin/scheduler/processed-files")
 def scheduler_processed_files(
    limit: int = 50,
    status: str | None = None,
 ):
    """List recently-processed inbound files (newest first).
    The operator uses this to confirm the scheduler is keeping up
    with MFT traffic. ``status`` filters to one of: ok, error,
    skipped, pending.
    """
    from sqlalchemy import desc
    from cyclone import db
    from cyclone.db import ProcessedInboundFile
    with db.SessionLocal()() as session:
        q = session.query(ProcessedInboundFile).order_by(
            desc(ProcessedInboundFile.processed_at),
        )
        if status:
            q = q.filter(ProcessedInboundFile.status == status)
        q = q.limit(min(limit, 500))
        rows = q.all()
        return {
            "count": len(rows),
            "files": [
                {
                    "id": r.id,
                    "name": r.name,
                    "size": r.size,
                    "file_type": r.file_type,
                    "processed_at": r.processed_at.isoformat() if r.processed_at else None,
                    "parser_used": r.parser_used,
                    "claim_count": r.claim_count,
                    "status": r.status,
                    "error_message": (
                        r.error_message[:200] + "..."
                        if r.error_message and len(r.error_message) > 200
                        else r.error_message
                    ),
                }
                for r in rows
            ],
        }
@@ -0,0 +1,220 @@
 """``/api/eligibility/*`` — 270/271 eligibility surface (SP3 P4 T23–T24).
 API-only — no DB persistence per spec section 3.4. The operator builds
 a 270 inquiry from a small JSON body, downloads the raw text to paste
 into a payer portal, then re-pastes the 271 response and parses it
 back into structured fields on demand.
 Two endpoints:
 * ``POST /api/eligibility/request`` — build a 270 inquiry from a JSON
  body and return ``{raw_270_text, parsed}``.
 * ``POST /api/eligibility/parse-271`` — accept a raw 271 file upload
  and return the parsed ``coverage_benefits`` / ``subscriber`` /
  ``summary`` / envelope triples.
 The ``_validate_eligibility_request`` helper is private to this router
 — it builds the Pydantic ``ParseResult270`` and is only used by the
 ``/request`` endpoint.
 """
 from __future__ import annotations
 import json
 import logging
 from datetime import date as _date
 from typing import Any
 from fastapi import APIRouter, File, HTTPException, UploadFile
 from fastapi.responses import JSONResponse
 from cyclone.parsers.exceptions import CycloneParseError
 from cyclone.parsers.models import BatchSummary, Envelope
 from cyclone.parsers.models_270 import (
    EligibilityBenefitInquiry,
    InformationReceiver270,
    InformationSource270,
    ParseResult270,
    Subscriber270,
 )
 from cyclone.parsers.parse_271 import parse as parse_271_text
 from cyclone.parsers.serialize_270 import serialize_270
 log = logging.getLogger(__name__)
 router = APIRouter()
 def _validate_eligibility_request(body: dict) -> tuple[ParseResult270, str]:
    """Build a :class:`ParseResult270` from a request body dict.
    The body shape is the minimum surface needed to build a valid 270
    inquiry (per spec section 3.4 — operator-driven, ephemeral):
    ::
        {
            "subscriber": {first_name, last_name, member_id, dob},
            "provider":   {npi, name},
            "payer":      {id, name},
            "service_type_code": "1"
        }
    Returns ``(ParseResult270, service_type_code)``. Raises
    :class:`HTTPException` (400) when the body is missing required
    fields.
    """
    subscriber_in = body.get("subscriber") or {}
    provider_in = body.get("provider") or {}
    payer_in = body.get("payer") or {}
    service_type_code = (body.get("service_type_code") or "").strip()
    # Required-field checks. We surface a single 400 with the first
    # missing field name to match the rest of the API's error contract.
    if not service_type_code:
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "service_type_code is required"},
        )
    if not subscriber_in.get("member_id"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "subscriber.member_id is required"},
        )
    if not provider_in.get("npi"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "provider.npi is required"},
        )
    if not payer_in.get("name"):
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": "payer.name is required"},
        )
    subscriber_dob_raw = subscriber_in.get("dob")
    subscriber_dob: _date | None = None
    if subscriber_dob_raw:
        try:
            subscriber_dob = _date.fromisoformat(subscriber_dob_raw)
        except (TypeError, ValueError) as exc:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": "Bad request",
                    "detail": f"subscriber.dob must be YYYY-MM-DD: {exc}",
                },
            ) from exc
    result = ParseResult270(
        envelope=Envelope(
            sender_id="SUBMITTERID",
            receiver_id=str(payer_in.get("id") or "RECEIVERID"),
            control_number="000000001",
            transaction_date=_date.today(),
            implementation_guide="005010X279A1",
        ),
        information_source=InformationSource270(
            name=str(payer_in["name"]),
            id=str(payer_in.get("id") or "") or None,
        ),
        information_receiver=InformationReceiver270(
            name=str(provider_in.get("name") or ""),
            npi=str(provider_in["npi"]),
        ),
        subscriber=Subscriber270(
            member_id=str(subscriber_in["member_id"]),
            first_name=str(subscriber_in.get("first_name") or "") or None,
            last_name=str(subscriber_in.get("last_name") or "") or None,
            dob=subscriber_dob,
        ),
        inquiries=[EligibilityBenefitInquiry(service_type_code=service_type_code)],
        summary=BatchSummary(
            input_file="eligibility_request",
            control_number="000000001",
            transaction_date=_date.today(),
            total_claims=1,
            passed=1,
            failed=0,
        ),
    )
    return result, service_type_code
@router.post("/api/eligibility/request")
 def post_eligibility_request(body: dict) -> Any:
    """Build a 270 eligibility inquiry from a small JSON body.
    Returns ``{"raw_270_text": <X12>, "parsed": <ParseResult270>}``
    so the operator can either download the raw text (paste into a
    payer portal) or render the parsed fields directly. Per spec
    section 3.4, nothing is persisted to the DB.
    """
    try:
        result, _ = _validate_eligibility_request(body)
    except HTTPException:
        raise
    except (KeyError, TypeError, ValueError) as exc:
        raise HTTPException(
            status_code=400,
            detail={"error": "Bad request", "detail": f"Malformed body: {exc}"},
        ) from exc
    raw_270_text = serialize_270(result)
    return {
        "raw_270_text": raw_270_text,
        "parsed": json.loads(result.model_dump_json()),
    }
@router.post("/api/eligibility/parse-271")
 async def post_eligibility_parse_271(
    file: UploadFile = File(...),
 ) -> Any:
    """Parse a 271 eligibility response and return the structured summary.
    Accepts the raw 271 text as a file upload (multipart/form-data),
    mirrors the ``/api/parse-999`` contract. Per spec section 3.4 the
    result is NOT persisted — the operator re-pastes the 271 each
    time they need a fresh read.
    The response body is a JSON object with three top-level keys:
    ``coverage_benefits``, ``subscriber``, and ``summary``. 400 is
    returned on empty / undecodable / malformed EDI; 200 on success.
    """
    raw = await file.read()
    if not raw:
        return JSONResponse(
            status_code=400,
            content={"error": "Empty file", "detail": "Uploaded file contained no bytes."},
        )
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        return JSONResponse(
            status_code=400,
            content={"error": "Encoding error", "detail": str(exc)},
        )
    try:
        result = parse_271_text(text, input_file=file.filename or "")
    except CycloneParseError as exc:
        return JSONResponse(
            status_code=400,
            content={"error": "Parse error", "detail": str(exc)},
        )
    except Exception as exc:  # pragma: no cover - safety net
        log.exception("Unexpected parser failure on 271")
        return JSONResponse(
            status_code=500,
            content={"error": "Internal server error", "detail": str(exc)},
        )
    return {
        "coverage_benefits": [
            json.loads(cb.model_dump_json()) for cb in result.coverage_benefits
        ],
        "subscriber": json.loads(result.subscriber.model_dump_json()),
        "summary": json.loads(result.summary.model_dump_json()),
        "envelope": json.loads(result.envelope.model_dump_json()),
        "information_source": json.loads(result.information_source.model_dump_json()),
        "information_receiver": json.loads(result.information_receiver.model_dump_json()),
    }