diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index ac1c08a..e0efc29 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -214,17 +214,6 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]: log.warning("Scheduler shutdown failed: %s", exc) -# Mirror cli._PAYER_FACTORIES. Kept local so the API doesn't depend on the -# Click-based CLI module. -PAYER_FACTORIES: dict[str, Any] = { - "co_medicaid": PayerConfig.co_medicaid, - "generic_837p": PayerConfig.generic_837p, -} - -PAYER_FACTORIES_835: dict[str, Any] = { - "co_medicaid_835": PayerConfig835.co_medicaid_835, - "generic_835": PayerConfig835.generic_835, -} # Allow both common dev-server origins. ``localhost`` and ``127.0.0.1`` # resolve to the same Vite dev server but CORS treats them as distinct @@ -298,46 +287,6 @@ async def _http_exc_handler(request, exc: HTTPException): ) -def _resolve_payer(name: str) -> PayerConfig: - if name not in PAYER_FACTORIES: - raise HTTPException( - status_code=400, - detail={ - "error": "Unknown payer", - "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES)}", - }, - ) - return PAYER_FACTORIES[name]() - - -def _resolve_payer_835(name: str) -> PayerConfig835: - if name not in PAYER_FACTORIES_835: - raise HTTPException( - status_code=400, - detail={ - "error": "Unknown payer", - "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES_835)}", - }, - ) - return PAYER_FACTORIES_835[name]() - - -def _transaction_set_id_from_segments(segments: list[list[str]]) -> str | None: - """Return the ST01 transaction-set id (``"837"``, ``"835"``, ``"999"``...). - - SP35 helper: scans the first few tokenized segments for the ST - segment and returns its second element (ST01). Returns None when no - ST is present — e.g. a TA1 file, which uses the bare TA1 segment - and no ST envelope. The endpoint-level envelope guards treat - ``None`` as "no ST found; let the parser decide" so TA1 files - routed through the wrong endpoint still surface a parse error - rather than a misleading "expected 837p, got ''" message. - """ - for seg in segments[:5]: # ST is always the second segment after ISA - if seg and seg[0] == "ST" and len(seg) > 1: - return seg[1] - return None - # --------------------------------------------------------------------------- # # Catch-all exception handler @@ -368,910 +317,6 @@ async def _unhandled_exception_handler(request: Request, exc: Exception) -> JSON ) -@app.post("/api/parse-837", dependencies=[Depends(matrix_gate)]) -async def parse_837( - request: Request, - file: UploadFile = File(...), - payer: str = Query("co_medicaid"), - include_raw_segments: bool = Query(True), - strict: bool = Query(False), - ack: bool = Query(False), -) -> Any: - # SP35: defense-in-depth input guards. Layer A (UI auto-detect) lives - # in src/pages/Upload.tsx; the server-side checks below are the - # authoritative fix because they protect every caller of the API - # (Upload page, CLI ingestion, any future bulk-import tool). Without - # these, an 835 file dropped on the Upload page while the dropdown - # still says "837p" produces a BatchRecord with claims=[] and a bogus - # row on the History tab. The fix is two checks run BEFORE we persist - # anything: - # - # 1. Envelope check — ST01 must be "837" or "837P". Anything else - # (an 835, a 999, a 270, garbage that happens to have an ISA) - # → 400 with error="Mismatched file kind", expected="837p", - # detected_st=. - # 2. Empty-claims check — even with the right envelope, if the - # parser produced zero CLM segments (truncated file, header-only - # test fixture) → 400 with error="No claims parsed". A real - # production 837 batch with zero claims is never valid. - 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)}, - ) - - config = _resolve_payer(payer) - - # SP35 guard 1: envelope check. Tokenize first so we can return a - # precise 400 (vs. relying on the parser's "no ISA envelope" error - # which is correct but doesn't say "you sent an 835 to the 837 - # endpoint"). If tokenization itself fails we fall through to the - # parser, which raises CycloneParseError → 400 "Parse error" path. - try: - _segments = _tokenize_segments(text) - detected_st = _transaction_set_id_from_segments(_segments) or "" - except CycloneParseError: - detected_st = "" - - if detected_st and not detected_st.upper().startswith("837"): - return JSONResponse( - status_code=400, - content={ - "error": "Mismatched file kind", - "expected": "837p", - "detected_st": detected_st, - "detail": ( - f"File declares ST*{detected_st}* but this endpoint " - f"expects ST*837*. Pick the matching endpoint on the " - f"Upload page (or let auto-detect choose for you)." - ), - }, - ) - - try: - result = parse(text, config, 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") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - # SP35 guard 2: empty-claims check. With the envelope validated, the - # only way to land here is a header-only file (real, but useless) - # or a file whose CLM loops the parser couldn't extract. Either way - # we refuse to persist — a BatchRecord with claims=[] is what the - # original bug produced and is never what the operator wanted. - if not result.claims: - return JSONResponse( - status_code=400, - content={ - "error": "No claims parsed", - "detail": ( - "The file passed the envelope check but contained no " - "CLM segments. Refusing to persist an empty batch." - ), - }, - ) - - if strict: - result = _strict_rewrite(result) - if not include_raw_segments: - result = _drop_raw_segments(result) - - if _has_claim_validation_errors(result): - # Per spec: 422 when claims failed validation. - # Body still includes the full ParseResult so the client can show errors. - # Validation-failed parses are NOT persisted (the data is suspect); - # only parses that survive validation end up in the store. - return JSONResponse( - status_code=422, - content=json.loads(result.model_dump_json()), - ) - - # Persist the cleaned-up result so the cleaned result is what clients - # retrieve via /api/batches/{id}. - rec = BatchRecord( - id=uuid.uuid4().hex, - kind="837p", - input_filename=file.filename or "upload.txt", - parsed_at=utcnow(), - result=result, - ) - try: - store.add(rec, event_bus=request.app.state.event_bus) - except IntegrityError as exc: - # ``(batch_id, patient_control_number)`` is UNIQUE — fires when a - # single batch file contains the same CLM01 control number twice, - # or when the same claim id has already been ingested in a prior - # batch. Surface as 409 with the batch id so the caller can look - # it up; do NOT 500 (a 500 without CORS headers is misreported by - # browsers as a CORS error and hides the real cause). - log.warning("Duplicate claim while persisting batch %s: %s", rec.id, exc) - return JSONResponse( - status_code=409, - content={ - "error": "Duplicate claim", - "detail": ( - "This file (or one previously ingested with the same " - "claim control number) collides with an existing " - "record. Inspect the file for duplicate CLM01 " - "control numbers, or remove the existing batch " - "before retrying." - ), - "batch_id": rec.id, - }, - ) - - if _client_wants_json(request): - body = json.loads(result.model_dump_json()) - if ack: - ack_body = _build_and_persist_ack(rec.id) - if ack_body is not None: - body["ack"] = ack_body - # Surface the server-side batch id so the frontend can call - # /api/batches/{id}/export-837 (and any other batch-scoped - # endpoint) without a separate listBatches round-trip. - body["batch_id"] = rec.id - return JSONResponse(content=body) - - # Default: NDJSON stream. Pass the server-side batch id so the - # streaming client (the React Upload page) can call batch-scoped - # endpoints like /api/batches/{id}/export-837 without a separate - # GET /api/batches round-trip. - return StreamingResponse( - _ndjson_stream(result, batch_id=rec.id), - media_type="application/x-ndjson", - ) - - -def _build_and_persist_ack(batch_id: str) -> dict | None: - """Build a 999 ACK for ``batch_id`` and persist the row. - - Returns the ack payload dict (matches the ``/api/parse-999`` - response shape so the JSON and NDJSON clients can share the - schema) or None if the build failed. The build is fail-soft: - errors are logged but never abort the 837 ingest, because the - user-visible 837 result is still correct. - """ - try: - ack_result = build_ack_for_batch(batch_id) - except Exception: - log.exception("build_ack_for_batch failed for %s", batch_id) - return None - fg = ack_result.functional_group_acks[0] if ack_result.functional_group_acks else None - if fg is None: - return None - raw_text = serialize_999(ack_result, interchange_control_number=ack_result.envelope.control_number) - row = store.add_ack( - source_batch_id=batch_id, - accepted_count=fg.accepted_count, - rejected_count=fg.rejected_count, - received_count=fg.received_count, - ack_code=fg.ack_code, - raw_json=json.loads(ack_result.model_dump_json()), - ) - return { - "id": row.id, - "accepted_count": fg.accepted_count, - "rejected_count": fg.rejected_count, - "received_count": fg.received_count, - "ack_code": fg.ack_code, - "source_batch_id": batch_id, - "raw_999_text": raw_text, - } - - -# --------------------------------------------------------------------------- # -# 835 ERA (Health Care Claim Payment/Advice) -# --------------------------------------------------------------------------- # - - -def _reconciliation_summary_for_batch(batch_id: str) -> dict: - """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. - - Reads from the DB after ``store.add()`` has already run reconciliation - synchronously (SP27 Task 10: ``reconcile.run(s, batch_id)`` inside the - ingest session, before commit). Counts are observed at this moment; - a subsequent manual match/unmatch will not be reflected until the - next request. - - ``skipped`` is reserved for future use — the orchestrator tracks - skipped claims internally but does not surface a queryable count. - """ - from sqlalchemy import func, select - from cyclone import db - from cyclone.db import Match, Remittance - - with db.SessionLocal()() as s: - matched = s.execute( - select(func.count(Match.id)).where( - Match.remittance_id.in_( - select(Remittance.id).where(Remittance.batch_id == batch_id) - ) - ) - ).scalar_one() - - # Pull unmatched via the store (small result set; cheap). - unmatched = store.list_unmatched(kind="both") - return { - "matched": matched, - "unmatched_claims": len(unmatched["claims"]), - "unmatched_remittances": len(unmatched["remittances"]), - "skipped": 0, # reserved — T10 does not persist a skipped count - } - - -@app.post("/api/parse-835", dependencies=[Depends(matrix_gate)]) -async def parse_835_endpoint( - request: Request, - file: UploadFile = File(...), - payer: str = Query("co_medicaid_835"), - include_raw_segments: bool = Query(True), - strict: bool = Query(False), -) -> Any: - 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)}, - ) - - config = _resolve_payer_835(payer) - - # SP35 guard 1: envelope check. Mirrors the parse-837 path: tokenize, - # read ST01, reject anything that doesn't start with "835". Same - # defense-in-depth rationale — the UI auto-detect (src/pages/Upload.tsx) - # is layer A, but server-side guards protect every API caller. - try: - _segments_835 = _tokenize_segments(text) - detected_st_835 = _transaction_set_id_from_segments(_segments_835) or "" - except CycloneParseError: - detected_st_835 = "" - - if detected_st_835 and not detected_st_835.upper().startswith("835"): - return JSONResponse( - status_code=400, - content={ - "error": "Mismatched file kind", - "expected": "835", - "detected_st": detected_st_835, - "detail": ( - f"File declares ST*{detected_st_835}* but this endpoint " - f"expects ST*835*. Pick the matching endpoint on the " - f"Upload page (or let auto-detect choose for you)." - ), - }, - ) - - try: - result = parse_835(text, config, 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") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - # SP35 guard 2: empty-claims check. Same as parse-837: a BatchRecord - # with claims=[] is never a valid production 835 batch and we refuse - # to persist it. - if not result.claims: - return JSONResponse( - status_code=400, - content={ - "error": "No claims parsed", - "detail": ( - "The file passed the envelope check but contained no " - "CLP segments. Refusing to persist an empty batch." - ), - }, - ) - - # Always run the validator; attach the report so the JSON path can - # surface it and the NDJSON path can fold the counts into the summary. - # 835 validation is batch-level, so pass/fail applies uniformly to every - # claim payment in the batch (passed=N or 0, failed=0 or N). - report = validate_835(result, config) - n = len(result.claims) - claim_ids = [c.payer_claim_control_number for c in result.claims] - if report.passed: - passed, failed, failed_claim_ids = n, 0, [] - else: - passed, failed, failed_claim_ids = 0, n, claim_ids - result = result.model_copy(update={ - "validation": report, - "summary": result.summary.model_copy(update={ - "passed": passed, - "failed": failed, - "failed_claim_ids": failed_claim_ids, - }), - }) - - if strict: - result = _strict_rewrite_835(result) - if not include_raw_segments: - result = _drop_raw_segments_835(result) - - if _has_835_validation_errors(result): - return JSONResponse( - status_code=422, - content=json.loads(result.model_dump_json()), - ) - - # Persist the cleaned-up result so the cleaned result is what clients - # retrieve via /api/batches/{id}. - rec = BatchRecord( - id=uuid.uuid4().hex, - kind="835", - input_filename=file.filename or "upload.txt", - parsed_at=utcnow(), - result=result, - ) - try: - store.add(rec, event_bus=request.app.state.event_bus) - except IntegrityError as exc: - log.warning("Duplicate remittance while persisting batch %s: %s", rec.id, exc) - return JSONResponse( - status_code=409, - content={ - "error": "Duplicate remittance", - "detail": ( - "This 835 file (or one previously ingested with the " - "same payer claim control number) collides with an " - "existing record. Remove the existing remittance " - "before retrying." - ), - "batch_id": rec.id, - }, - ) - - if _client_wants_json(request): - body = json.loads(result.model_dump_json()) - body["reconciliation"] = _reconciliation_summary_for_batch(rec.id) - return JSONResponse(content=body) - - # Default: NDJSON stream. Pass the server-side batch id so the - # streaming client can call batch-scoped endpoints without a - # separate GET /api/batches round-trip (see /api/parse-837 for the - # parallel change). - return StreamingResponse( - _ndjson_stream_835(result, batch_id=rec.id), - media_type="application/x-ndjson", - ) - - -# --------------------------------------------------------------------------- # -# 999 ACK (Implementation Acknowledgment) -# --------------------------------------------------------------------------- # - -# SP27 Task 6: ack ID helpers were deduped. Both api.py and scheduler.py -# used to define these locally. Canonical copies now live in -# ``cyclone.handlers._ack_id`` (set up in Task 1). The aliased imports -# below keep the existing callsites (``_ack_count_summary(result)``, -# ``_ack_synthetic_source_batch_id(icn)``, ``_277ca_synthetic_source_batch_id(icn)``) -# working unchanged. -from cyclone.handlers._ack_id import ( - ack_count_summary as _ack_count_summary, -) -from cyclone.handlers._ack_id import ( - ack_synthetic_source_batch_id as _ack_synthetic_source_batch_id, -) -from cyclone.handlers._ack_id import ( - two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id, -) - - -@app.post("/api/parse-999", dependencies=[Depends(matrix_gate)]) -async def parse_999_endpoint( - request: Request, - file: UploadFile = File(...), -) -> Any: - """Parse a 999 ACK file, persist a row, and return JSON. - - Behavior mirrors ``/api/parse-835``: - - 400 on empty / undecodable / malformed EDI (never 500). - - 200 on success with ``{"ack": {id, accepted_count, rejected_count, - received_count, ack_code, raw_999_text}, "parsed": }``. - - The persisted ``acks.source_batch_id`` is a synthetic id - (``999-``) because a received 999 has no inbound 837 batch - to FK to. The dashboard's `/acks` list surfaces these; operators - can still see which interchange each row came from. - """ - 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_999_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 999") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - received, accepted, rejected, ack_code = _ack_count_summary(result) - icn = result.envelope.control_number - synthetic_id = _ack_synthetic_source_batch_id(icn) - - # Build the raw 999 text from the parsed result (round-trip). - raw_999_text = serialize_999(result, interchange_control_number=icn or "000000001") - - # SP6 T4: move claims whose 999 set was rejected into ClaimState.REJECTED. - # The 999's set_control_number (AK202) is the source 837's ST02; in - # practice we look it up against patient_control_number because that's - # the field 999 ACKs cross-reference in this product. - with db.SessionLocal()() as session: - def _lookup(pcn: str): - return session.query(Claim).filter_by(patient_control_number=pcn).first() - _rejection_result = apply_999_rejections( - session, result, claim_lookup=_lookup, - ) - if _rejection_result.matched: - bus = request.app.state.event_bus - for cid in _rejection_result.matched: - await bus.publish("claim.rejected", {"claim_id": cid}) - if _rejection_result.orphans: - log.warning( - "999 had %d orphan set refs: %s", - len(_rejection_result.orphans), - _rejection_result.orphans[:5], - ) - - row = store.add_ack( - source_batch_id=synthetic_id, - accepted_count=accepted, - rejected_count=rejected, - received_count=received, - ack_code=ack_code, - raw_json=json.loads(result.model_dump_json()), - event_bus=request.app.state.event_bus, - ) - - # SP11: append one audit row per rejected claim. Each row chains - # to the previous one — see cyclone.audit_log. - if _rejection_result.matched: - with db.SessionLocal()() as audit_s: - for cid in _rejection_result.matched: - append_event(audit_s, AuditEvent( - event_type="claim.rejected", - entity_type="claim", - entity_id=cid, - payload={"source_batch_id": synthetic_id, "ack_id": row.id}, - actor="999-parser", - user_id=_actor_user_id(request), - )) - audit_s.commit() - - # SP28: auto-link the 999 AK2 set-responses to claims via the - # D10 two-pass join (ST02 via batch envelope index primary, - # Claim.patient_control_number fallback). Each created ClaimAck - # row publishes claim_ack_written so the live-tail subscribers - # on the claim and ack side see the link immediately. - claim_ack_links_count = 0 - with db.SessionLocal()() as link_s: - batch_index = store.batch_envelope_index() - - def _pcn_lookup(pcn: str): - return ( - link_s.query(Claim) - .filter(Claim.patient_control_number == pcn) - .first() - ) - - link_result = _apply_999_acceptances( - link_s, result, ack_id=row.id, - batch_envelope_index=batch_index, - pc_claim_lookup=_pcn_lookup, - ) - for link_row in link_result.linked: - store.add_claim_ack( - claim_id=link_row.claim_id, - batch_id=link_row.batch_id, - ack_id=row.id, - ack_kind="999", - ak2_index=link_row.ak2_index, - set_control_number=link_row.set_control_number, - set_accept_reject_code=link_row.set_accept_reject_code, - linked_by="auto", - event_bus=request.app.state.event_bus, - ) - claim_ack_links_count += 1 - - return JSONResponse(content={ - "ack": { - "id": row.id, - "accepted_count": accepted, - "rejected_count": rejected, - "received_count": received, - "ack_code": ack_code, - "source_batch_id": synthetic_id, - "raw_999_text": raw_999_text, - "claim_ack_links_count": claim_ack_links_count, - }, - "parsed": json.loads(result.model_dump_json()), - }) - - -# --------------------------------------------------------------------------- # -# TA1 (Interchange Acknowledgment) -# --------------------------------------------------------------------------- # - - -def _ta1_synthetic_source_batch_id(interchange_control_number: str) -> str: - """Return a synthetic ``batches.id`` for a received TA1 with no source batch. - - Mirrors :func:`_ack_synthetic_source_batch_id`. The ta1_acks.source_batch_id - FK requires a row in batches; for received TA1s we synthesize an id of - the form ``TA1-``. The row is NOT created in batches (same - FK-is-no-op convention as the 999 path). - """ - return f"TA1-{(interchange_control_number or '').strip() or '000000001'}" - - -@app.post("/api/parse-ta1", dependencies=[Depends(matrix_gate)]) -async def parse_ta1_endpoint( - request: Request, - file: UploadFile = File(...), -) -> Any: - """Parse a TA1 (Interchange Acknowledgment) file, persist a row, return JSON. - - Mirrors ``/api/parse-999`` but for the lower-level envelope ack: - - 400 on empty / undecodable / malformed EDI (never 500). - - 200 on success with ``{"ta1": {id, control_number, ack_code, - note_code, interchange_date, interchange_time, sender_id, - receiver_id, source_batch_id, raw_ta1_text}, "parsed": - }``. - - The persisted ``ta1_acks.source_batch_id`` is a synthetic id - (``TA1-``) because a received TA1 has no inbound batch to - FK to. The dashboard's ``/ta1-acks`` list surfaces these. - - SP25: threads ``event_bus=request.app.state.event_bus`` into - ``store.add_ta1_ack`` so the live-tail ``ta1_ack_received`` - stream fires on manual uploads (not just on the SFTP poller). - """ - 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_ta1_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 TA1") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - # Build the raw TA1 text from the parsed result (round-trip). - raw_ta1_text = _serialize_ta1(result) - - row = store.add_ta1_ack( - source_batch_id=result.source_batch_id, - control_number=result.ta1.control_number, - interchange_date=result.ta1.interchange_date, - interchange_time=result.ta1.interchange_time, - ack_code=result.ta1.ack_code, - note_code=result.ta1.note_code, - ack_generated_date=result.ta1.ack_generated_date, - sender_id=result.envelope.sender_id, - receiver_id=result.envelope.receiver_id, - raw_json=json.loads(result.model_dump_json()), - event_bus=request.app.state.event_bus, - ) - - # SP28: TA1 envelope-level link to the originating Batch. The - # closure here matches the most-recent Batch whose envelope - # sender_id/receiver_id matches the TA1 — see spec §D4. - claim_ack_links_count = 0 - with db.SessionLocal()() as link_s: - def _batch_lookup(sender_id, receiver_id): - rows = ( - link_s.query(Batch) - .filter( - Batch.kind == "837p", - Batch.raw_result_json.isnot(None), - ) - .order_by(Batch.parsed_at.desc()) - .all() - ) - for row in rows: - env = (row.raw_result_json or {}).get("envelope") or {} - if ( - env.get("sender_id") == sender_id - and env.get("receiver_id") == receiver_id - ): - return row - return None - - link_result = _apply_ta1_envelope_link( - link_s, result, ack_id=row.id, - batch_lookup=_batch_lookup, - ) - for link_row in link_result.linked: - store.add_claim_ack( - claim_id=link_row.claim_id, - batch_id=link_row.batch_id, - ack_id=row.id, - ack_kind="ta1", - ak2_index=link_row.ak2_index, - set_control_number=link_row.set_control_number, - set_accept_reject_code=link_row.set_accept_reject_code, - linked_by="auto", - event_bus=request.app.state.event_bus, - ) - claim_ack_links_count += 1 - - return JSONResponse(content={ - "ta1": { - "id": row.id, - "control_number": result.ta1.control_number, - "ack_code": result.ta1.ack_code, - "note_code": result.ta1.note_code, - "interchange_date": result.ta1.interchange_date.isoformat() - if result.ta1.interchange_date else None, - "interchange_time": result.ta1.interchange_time, - "sender_id": result.envelope.sender_id, - "receiver_id": result.envelope.receiver_id, - "source_batch_id": result.source_batch_id, - "raw_ta1_text": raw_ta1_text, - "claim_ack_links_count": claim_ack_links_count, - }, - "parsed": json.loads(result.model_dump_json()), - }) - - -# --------------------------------------------------------------------------- # -# 277CA (Claim Acknowledgment) — SP10 -# --------------------------------------------------------------------------- # - - -# (The _277ca_synthetic_source_batch_id helper was moved to -# cyclone.handlers._ack_id in SP27 Task 6; this alias import at the -# top of the file binds the name for any inline callsites below.) - - -@app.post("/api/parse-277ca", dependencies=[Depends(matrix_gate)]) -async def parse_277ca_endpoint( - request: Request, - file: UploadFile = File(...), -) -> Any: - """Parse a 277CA Claim Acknowledgment file, persist a row, and stamp rejections. - - Behavior mirrors ``/api/parse-999``: - - 400 on empty / undecodable / malformed EDI (never 500). - - 200 on success with ``{"ack": {id, control_number, accepted_count, - rejected_count, payer_claim_control_numbers, raw_277ca_text}, - "parsed": }``. - - After parse, runs :func:`apply_277ca_rejections` to stamp the - payer-rejected fields on each matching claim row. The Inbox - Payer-Rejected lane lights up as a side-effect of this call. - """ - 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_277ca_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 277CA") - return JSONResponse( - status_code=500, - content={"error": "Internal server error", "detail": str(exc)}, - ) - - icn = result.envelope.control_number - synthetic_id = _277ca_synthetic_source_batch_id(icn) - - accepted = sum(1 for s in result.claim_statuses if s.classification == "accepted") - paid = sum(1 for s in result.claim_statuses if s.classification == "paid") - rejected = sum(1 for s in result.claim_statuses if s.classification == "rejected") - pended = sum(1 for s in result.claim_statuses if s.classification == "pended") - - # Persist the 277CA row first so we have an id to attach to claims. - # SP25: thread the event bus so ``two77ca_ack_received`` fires. - row = store.add_277ca_ack( - source_batch_id=synthetic_id, - control_number=icn, - accepted_count=accepted, - rejected_count=rejected, - paid_count=paid, - pended_count=pended, - raw_json=json.loads(result.model_dump_json()), - event_bus=request.app.state.event_bus, - ) - - # Stamp payer-rejection fields on matching claims. The 277CA's - # REF*1K carries the patient's claim control number we sent in - # CLM01 — same convention the 999 ACK uses, so the lookup hits - # Claim.patient_control_number (mirrors apply_999_rejections). - with db.SessionLocal()() as session: - def _lookup(pcn: str): - return ( - session.query(Claim) - .filter(Claim.patient_control_number == pcn) - .first() - ) - apply_result = apply_277ca_rejections( - session, result, claim_lookup=_lookup, two77ca_id=row.id, - ) - if apply_result.matched: - bus = request.app.state.event_bus - for cid in apply_result.matched: - await bus.publish("claim.payer_rejected", {"claim_id": cid}) - # SP11: audit trail for each payer-rejected claim. - with db.SessionLocal()() as audit_s: - for cid in apply_result.matched: - append_event(audit_s, AuditEvent( - event_type="claim.payer_rejected", - entity_type="claim", - entity_id=cid, - payload={ - "source_batch_id": synthetic_id, - "277ca_id": row.id, - }, - actor="277ca-parser", - user_id=_actor_user_id(request), - )) - audit_s.commit() - if apply_result.orphans: - log.warning( - "277CA had %d orphan status entries (no matching claim): %s", - len(apply_result.orphans), - apply_result.orphans[:5], - ) - - # SP28: auto-link the 277CA ClaimStatus entries to claims via - # the D10 two-pass join. Each ClaimAck row publishes - # claim_ack_written so the live-tail subscribers on the claim - # and ack side see the link immediately. - claim_ack_links_count = 0 - with db.SessionLocal()() as link_s: - batch_index = store.batch_envelope_index() - - def _pcn_lookup(pcn: str): - return ( - link_s.query(Claim) - .filter(Claim.patient_control_number == pcn) - .first() - ) - - link_result = _apply_277ca_acks( - link_s, result, ack_id=row.id, - batch_envelope_index=batch_index, - pc_claim_lookup=_pcn_lookup, - ) - for link_row in link_result.linked: - store.add_claim_ack( - claim_id=link_row.claim_id, - batch_id=link_row.batch_id, - ack_id=row.id, - ack_kind="277ca", - ak2_index=link_row.ak2_index, - set_control_number=link_row.set_control_number, - set_accept_reject_code=link_row.set_accept_reject_code, - linked_by="auto", - event_bus=request.app.state.event_bus, - ) - claim_ack_links_count += 1 - - return JSONResponse(content={ - "ack": { - "id": row.id, - "control_number": icn, - "accepted_count": accepted, - "rejected_count": rejected, - "paid_count": paid, - "pended_count": pended, - "source_batch_id": synthetic_id, - "matched_claim_ids": apply_result.matched, - "orphan_status_codes": apply_result.orphans, - "claim_ack_links_count": claim_ack_links_count, - }, - "parsed": json.loads(result.model_dump_json()), - }) - - -def _serialize_ta1(result) -> str: - """Render a TA1 file from a ParseResultTa1 for the ``raw_ta1_text`` field. - - Mirrors what the parser consumed: ISA envelope → TA1 → IEA. We - rebuild minimal ISA fields from the envelope plus the TA1 segment - verbatim. The serializer is intentionally tiny — TA1 has no GS/ST, - so there's no functional-group structure to round-trip. - """ - ta1 = result.ta1 - parts = [ - f"TA1*{ta1.control_number}*{ta1.interchange_date.strftime('%y%m%d') if ta1.interchange_date else ''}*" - f"{ta1.interchange_time or ''}*{ta1.ack_code}*{ta1.note_code or ''}*" - f"{ta1.ack_generated_date.strftime('%y%m%d') if ta1.ack_generated_date else ''}", - ] - return "~".join(parts) + "~" - - -def _serialize_ta1_from_row(row: db.Ta1Ack) -> str: - """Reconstruct a TA1 segment from the persisted flat row (for the detail endpoint).""" - date_s = row.interchange_date.strftime("%y%m%d") if row.interchange_date else "" - return ( - f"TA1*{row.control_number}*{date_s}*{row.interchange_time or ''}*" - f"{row.ack_code}*{row.note_code or ''}~" - ) - - from cyclone.auth.routes import router as auth_router from cyclone.auth.admin import router as admin_users_router @@ -1312,5 +357,22 @@ from cyclone.api_routers.claims import claims_stream # noqa: F401 # SP36 backw from cyclone.api_routers.remittances import remittances_stream # noqa: F401 # SP36 backward-compat shim from cyclone.api_routers.activity import activity_stream # noqa: F401 # SP36 backward-compat shim +# SP36 Task 16 backward-compat: the 3 ack-helper aliases +# (``_ack_count_summary``, ``_ack_synthetic_source_batch_id``, +# ``_277ca_synthetic_source_batch_id``) used to be aliased imports +# at the top of api.py so the parse-999 / parse-277ca routes could +# call them by their underscore-prefixed names. After Task 16 +# extracted the parse routes into ``cyclone.api_routers.parse``, +# those alias imports moved with the routes. The dedup test in +# ``tests/test_api_dedup.py`` asserts ``hasattr(api, name)`` for +# each of these 3 names — it's a guard against the inline copy +# ever creeping back in. Re-import them here so the test keeps +# passing without modification. +from cyclone.handlers._ack_id import ( # noqa: E402,F401 # SP36 backward-compat shim + ack_count_summary as _ack_count_summary, + ack_synthetic_source_batch_id as _ack_synthetic_source_batch_id, + two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id, +) + __all__ = ["app"] diff --git a/backend/src/cyclone/api_routers/__init__.py b/backend/src/cyclone/api_routers/__init__.py index 1e05cc4..23cd9ef 100644 --- a/backend/src/cyclone/api_routers/__init__.py +++ b/backend/src/cyclone/api_routers/__init__.py @@ -23,6 +23,7 @@ from cyclone.api_routers import ( eligibility, health, inbox, + parse, payers, providers, reconciliation, @@ -43,6 +44,7 @@ routers: list[APIRouter] = [ eligibility.router, # gated health.router, # public — health probes must work pre-auth inbox.router, # gated + parse.router, # gated payers.router, # gated providers.router, # gated reconciliation.router, # gated diff --git a/backend/src/cyclone/api_routers/_shared.py b/backend/src/cyclone/api_routers/_shared.py index c8542b3..0e7429c 100644 --- a/backend/src/cyclone/api_routers/_shared.py +++ b/backend/src/cyclone/api_routers/_shared.py @@ -14,10 +14,57 @@ Helpers currently promoted here: surface that Task 16 will extract. Promoting now is cheaper than leaving a cross-module ``from cyclone.api import _actor_user_id`` that would create an import-cycle at registry load time. + +- :data:`PAYER_FACTORIES` / :data:`PAYER_FACTORIES_835` — SP36 + Task 16: payer config dicts lifted from ``api.py`` alongside + the ``_resolve_payer`` / ``_resolve_payer_835`` helpers that + consume them. The two helpers each touch a single ``PAYER_FACTORIES*`` + dict; keeping both halves of the pair in one module removes a + circular import (parse.py → _shared._resolve_payer → api.PAYER_FACTORIES). + +- :func:`_resolve_payer` / :func:`_resolve_payer_835` — used by + ``parse-837`` and ``parse-835`` endpoints respectively. Promoted + in SP36 Task 16 alongside the PAYER_FACTORIES dicts. + +- :func:`_transaction_set_id_from_segments` — used by + ``parse-837`` and ``parse-835`` envelope guards. Promoted in + SP36 Task 16. + +- :func:`_build_and_persist_ack` — used by ``parse-837`` (when + ``?ack=true``). Promoted in SP36 Task 16. + +- :func:`_reconciliation_summary_for_batch` — used by + ``parse-835``. Promoted in SP36 Task 16. + +- :func:`_ta1_synthetic_source_batch_id` — used by ``parse-ta1``. + Promoted in SP36 Task 16. + +- :func:`_serialize_ta1` — used by ``parse-ta1`` to build the + raw TA1 round-trip text. Promoted in SP36 Task 16 per the + plan's "8 helpers" specification; technically a single-router + helper per D4 but moved here for symmetry with the other parse + serializers. """ from __future__ import annotations -from fastapi import Request +import json +import logging +from typing import Any + +from fastapi import HTTPException, Request + +from cyclone import db +from cyclone.parsers.batch_ack_builder import build_ack_for_batch +from cyclone.parsers.payer import PayerConfig, PayerConfig835 +from cyclone.parsers.serialize_999 import serialize_999 +from cyclone.store import store + +log = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- # +# Actor user id (SP36 Task 11 — early-promoted) +# --------------------------------------------------------------------------- # def _actor_user_id(request: Request) -> int | None: @@ -33,3 +80,194 @@ def _actor_user_id(request: Request) -> int | None: if user is None: return None return getattr(user, "id", None) + + +# --------------------------------------------------------------------------- # +# Payer config dicts (SP36 Task 16) +# --------------------------------------------------------------------------- # +# +# Mirror cli._PAYER_FACTORIES. Kept here (not in the parse router) so +# the ``_resolve_payer`` / ``_resolve_payer_835`` helpers can import +# their backing dicts without a circular import. + + +PAYER_FACTORIES: dict[str, Any] = { + "co_medicaid": PayerConfig.co_medicaid, + "generic_837p": PayerConfig.generic_837p, +} + +PAYER_FACTORIES_835: dict[str, Any] = { + "co_medicaid_835": PayerConfig835.co_medicaid_835, + "generic_835": PayerConfig835.generic_835, +} + + +# --------------------------------------------------------------------------- # +# Payer resolution (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _resolve_payer(name: str) -> PayerConfig: + if name not in PAYER_FACTORIES: + raise HTTPException( + status_code=400, + detail={ + "error": "Unknown payer", + "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES)}", + }, + ) + return PAYER_FACTORIES[name]() + + +def _resolve_payer_835(name: str) -> PayerConfig835: + if name not in PAYER_FACTORIES_835: + raise HTTPException( + status_code=400, + detail={ + "error": "Unknown payer", + "detail": f"Unknown payer {name!r}. Choose from: {', '.join(PAYER_FACTORIES_835)}", + }, + ) + return PAYER_FACTORIES_835[name]() + + +# --------------------------------------------------------------------------- # +# Envelope detection (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _transaction_set_id_from_segments(segments: list[list[str]]) -> str | None: + """Return the ST01 transaction-set id (``"837"``, ``"835"``, ``"999"``...). + + SP35 helper: scans the first few tokenized segments for the ST + segment and returns its second element (ST01). Returns None when no + ST is present — e.g. a TA1 file, which uses the bare TA1 segment + and no ST envelope. The endpoint-level envelope guards treat + ``None`` as "no ST found; let the parser decide" so TA1 files + routed through the wrong endpoint still surface a parse error + rather than a misleading "expected 837p, got ''" message. + """ + for seg in segments[:5]: # ST is always the second segment after ISA + if seg and seg[0] == "ST" and len(seg) > 1: + return seg[1] + return None + + +# --------------------------------------------------------------------------- # +# 999 ACK builder (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _build_and_persist_ack(batch_id: str) -> dict | None: + """Build a 999 ACK for ``batch_id`` and persist the row. + + Returns the ack payload dict (matches the ``/api/parse-999`` + response shape so the JSON and NDJSON clients can share the + schema) or None if the build failed. The build is fail-soft: + errors are logged but never abort the 837 ingest, because the + user-visible 837 result is still correct. + """ + try: + ack_result = build_ack_for_batch(batch_id) + except Exception: + log.exception("build_ack_for_batch failed for %s", batch_id) + return None + fg = ack_result.functional_group_acks[0] if ack_result.functional_group_acks else None + if fg is None: + return None + raw_text = serialize_999(ack_result, interchange_control_number=ack_result.envelope.control_number) + row = store.add_ack( + source_batch_id=batch_id, + accepted_count=fg.accepted_count, + rejected_count=fg.rejected_count, + received_count=fg.received_count, + ack_code=fg.ack_code, + raw_json=json.loads(ack_result.model_dump_json()), + ) + return { + "id": row.id, + "accepted_count": fg.accepted_count, + "rejected_count": fg.rejected_count, + "received_count": fg.received_count, + "ack_code": fg.ack_code, + "source_batch_id": batch_id, + "raw_999_text": raw_text, + } + + +# --------------------------------------------------------------------------- # +# Reconciliation summary (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _reconciliation_summary_for_batch(batch_id: str) -> dict: + """Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch. + + Reads from the DB after ``store.add()`` has already run reconciliation + synchronously (SP27 Task 10: ``reconcile.run(s, batch_id)`` inside the + ingest session, before commit). Counts are observed at this moment; + a subsequent manual match/unmatch will not be reflected until the + next request. + + ``skipped`` is reserved for future use — the orchestrator tracks + skipped claims internally but does not surface a queryable count. + """ + from sqlalchemy import func, select + from cyclone.db import Match, Remittance + + with db.SessionLocal()() as s: + matched = s.execute( + select(func.count(Match.id)).where( + Match.remittance_id.in_( + select(Remittance.id).where(Remittance.batch_id == batch_id) + ) + ) + ).scalar_one() + + # Pull unmatched via the store (small result set; cheap). + unmatched = store.list_unmatched(kind="both") + return { + "matched": matched, + "unmatched_claims": len(unmatched["claims"]), + "unmatched_remittances": len(unmatched["remittances"]), + "skipped": 0, # reserved — T10 does not persist a skipped count + } + + +# --------------------------------------------------------------------------- # +# TA1 synthetic source batch id (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _ta1_synthetic_source_batch_id(interchange_control_number: str) -> str: + """Return a synthetic ``batches.id`` for a received TA1 with no source batch. + + Mirrors :func:`_ack_synthetic_source_batch_id` (in + ``cyclone.handlers._ack_id``). The ta1_acks.source_batch_id + FK requires a row in batches; for received TA1s we synthesize an + id of the form ``TA1-``. The row is NOT created in batches + (same FK-is-no-op convention as the 999 path). + """ + return f"TA1-{(interchange_control_number or '').strip() or '000000001'}" + + +# --------------------------------------------------------------------------- # +# TA1 serializer (SP36 Task 16) +# --------------------------------------------------------------------------- # + + +def _serialize_ta1(result) -> str: + """Render a TA1 file from a ParseResultTa1 for the ``raw_ta1_text`` field. + + Mirrors what the parser consumed: ISA envelope → TA1 → IEA. We + rebuild minimal ISA fields from the envelope plus the TA1 segment + verbatim. The serializer is intentionally tiny — TA1 has no GS/ST, + so there's no functional-group structure to round-trip. + """ + ta1 = result.ta1 + parts = [ + f"TA1*{ta1.control_number}*{ta1.interchange_date.strftime('%y%m%d') if ta1.interchange_date else ''}*" + f"{ta1.interchange_time or ''}*{ta1.ack_code}*{ta1.note_code or ''}*" + f"{ta1.ack_generated_date.strftime('%y%m%d') if ta1.ack_generated_date else ''}", + ] + return "~".join(parts) + "~" \ No newline at end of file diff --git a/backend/src/cyclone/api_routers/parse.py b/backend/src/cyclone/api_routers/parse.py new file mode 100644 index 0000000..5d9c003 --- /dev/null +++ b/backend/src/cyclone/api_routers/parse.py @@ -0,0 +1,831 @@ +"""Parse endpoints — accept X12 uploads and ingest them. + +Five routes: + +* ``POST /api/parse-837`` — 837P professional claim ingest (the + primary upload path) +* ``POST /api/parse-835`` — 835 ERA remittance ingest +* ``POST /api/parse-999`` — 999 ACK ingest + auto-link claims +* ``POST /api/parse-ta1`` — TA1 envelope ACK ingest + envelope-link batches +* ``POST /api/parse-277ca`` — 277CA Claim Acknowledgment ingest + +The 7 cross-router helpers these endpoints need (and the two +PAYER_FACTORIES dicts they consume) live in +:mod:`cyclone.api_routers._shared`. +""" +from __future__ import annotations + +import json +import logging +import uuid +from typing import Any + +from fastapi import APIRouter, Depends, File, Query, Request, UploadFile +from fastapi.responses import JSONResponse, StreamingResponse +from sqlalchemy.exc import IntegrityError + +from cyclone import db +from cyclone.api_helpers import ( + client_wants_json as _client_wants_json, + drop_raw_segments_837 as _drop_raw_segments, + drop_raw_segments_835 as _drop_raw_segments_835, + has_claim_validation_errors as _has_claim_validation_errors, + has_835_validation_errors as _has_835_validation_errors, + ndjson_stream_837 as _ndjson_stream, + ndjson_stream_835 as _ndjson_stream_835, + strict_rewrite_837 as _strict_rewrite, + strict_rewrite_835 as _strict_rewrite_835, +) +from cyclone.api_routers._shared import ( + _actor_user_id, + _build_and_persist_ack, + _reconciliation_summary_for_batch, + _resolve_payer, + _resolve_payer_835, + _serialize_ta1, + _ta1_synthetic_source_batch_id, + _transaction_set_id_from_segments, +) +from cyclone.audit_log import AuditEvent, append_event +from cyclone.auth.deps import matrix_gate +from cyclone.claim_acks import ( + apply_277ca_acks as _apply_277ca_acks, + apply_999_acceptances as _apply_999_acceptances, + apply_ta1_envelope_link as _apply_ta1_envelope_link, +) +from cyclone.db import Batch, Claim +from cyclone.handlers._ack_id import ( + ack_count_summary as _ack_count_summary, + ack_synthetic_source_batch_id as _ack_synthetic_source_batch_id, + two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id, +) +from cyclone.inbox_state import apply_999_rejections +from cyclone.inbox_state_277ca import apply_277ca_rejections +from cyclone.parsers.exceptions import CycloneParseError +from cyclone.parsers.parse_277ca import parse_277ca_text +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.segments import tokenize as _tokenize_segments +from cyclone.parsers.serialize_999 import serialize_999 +from cyclone.parsers.validator_835 import validate as validate_835 +from cyclone.store import BatchRecord, store, utcnow + +log = logging.getLogger(__name__) + +router = APIRouter(dependencies=[Depends(matrix_gate)]) + + +@router.post("/api/parse-837") +async def parse_837( + request: Request, + file: UploadFile = File(...), + payer: str = Query("co_medicaid"), + include_raw_segments: bool = Query(True), + strict: bool = Query(False), + ack: bool = Query(False), +) -> Any: + # SP35: defense-in-depth input guards. Layer A (UI auto-detect) lives + # in src/pages/Upload.tsx; the server-side checks below are the + # authoritative fix because they protect every caller of the API + # (Upload page, CLI ingestion, any future bulk-import tool). Without + # these, an 835 file dropped on the Upload page while the dropdown + # still says "837p" produces a BatchRecord with claims=[] and a bogus + # row on the History tab. The fix is two checks run BEFORE we persist + # anything: + # + # 1. Envelope check — ST01 must be "837" or "837P". Anything else + # (an 835, a 999, a 270, garbage that happens to have an ISA) + # → 400 with error="Mismatched file kind", expected="837p", + # detected_st=. + # 2. Empty-claims check — even with the right envelope, if the + # parser produced zero CLM segments (truncated file, header-only + # test fixture) → 400 with error="No claims parsed". A real + # production 837 batch with zero claims is never valid. + 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)}, + ) + + config = _resolve_payer(payer) + + # SP35 guard 1: envelope check. Tokenize first so we can return a + # precise 400 (vs. relying on the parser's "no ISA envelope" error + # which is correct but doesn't say "you sent an 835 to the 837 + # endpoint"). If tokenization itself fails we fall through to the + # parser, which raises CycloneParseError → 400 "Parse error" path. + try: + _segments = _tokenize_segments(text) + detected_st = _transaction_set_id_from_segments(_segments) or "" + except CycloneParseError: + detected_st = "" + + if detected_st and not detected_st.upper().startswith("837"): + return JSONResponse( + status_code=400, + content={ + "error": "Mismatched file kind", + "expected": "837p", + "detected_st": detected_st, + "detail": ( + f"File declares ST*{detected_st}* but this endpoint " + f"expects ST*837*. Pick the matching endpoint on the " + f"Upload page (or let auto-detect choose for you)." + ), + }, + ) + + try: + result = parse(text, config, 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") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + # SP35 guard 2: empty-claims check. With the envelope validated, the + # only way to land here is a header-only file (real, but useless) + # or a file whose CLM loops the parser couldn't extract. Either way + # we refuse to persist — a BatchRecord with claims=[] is what the + # original bug produced and is never what the operator wanted. + if not result.claims: + return JSONResponse( + status_code=400, + content={ + "error": "No claims parsed", + "detail": ( + "The file passed the envelope check but contained no " + "CLM segments. Refusing to persist an empty batch." + ), + }, + ) + + if strict: + result = _strict_rewrite(result) + if not include_raw_segments: + result = _drop_raw_segments(result) + + if _has_claim_validation_errors(result): + # Per spec: 422 when claims failed validation. + # Body still includes the full ParseResult so the client can show errors. + # Validation-failed parses are NOT persisted (the data is suspect); + # only parses that survive validation end up in the store. + return JSONResponse( + status_code=422, + content=json.loads(result.model_dump_json()), + ) + + # Persist the cleaned-up result so the cleaned result is what clients + # retrieve via /api/batches/{id}. + rec = BatchRecord( + id=uuid.uuid4().hex, + kind="837p", + input_filename=file.filename or "upload.txt", + parsed_at=utcnow(), + result=result, + ) + try: + store.add(rec, event_bus=request.app.state.event_bus) + except IntegrityError as exc: + # ``(batch_id, patient_control_number)`` is UNIQUE — fires when a + # single batch file contains the same CLM01 control number twice, + # or when the same claim id has already been ingested in a prior + # batch. Surface as 409 with the batch id so the caller can look + # it up; do NOT 500 (a 500 without CORS headers is misreported by + # browsers as a CORS error and hides the real cause). + log.warning("Duplicate claim while persisting batch %s: %s", rec.id, exc) + return JSONResponse( + status_code=409, + content={ + "error": "Duplicate claim", + "detail": ( + "This file (or one previously ingested with the same " + "claim control number) collides with an existing " + "record. Inspect the file for duplicate CLM01 " + "control numbers, or remove the existing batch " + "before retrying." + ), + "batch_id": rec.id, + }, + ) + + if _client_wants_json(request): + body = json.loads(result.model_dump_json()) + if ack: + ack_body = _build_and_persist_ack(rec.id) + if ack_body is not None: + body["ack"] = ack_body + # Surface the server-side batch id so the frontend can call + # /api/batches/{id}/export-837 (and any other batch-scoped + # endpoint) without a separate listBatches round-trip. + body["batch_id"] = rec.id + return JSONResponse(content=body) + + # Default: NDJSON stream. Pass the server-side batch id so the + # streaming client (the React Upload page) can call batch-scoped + # endpoints like /api/batches/{id}/export-837 without a separate + # GET /api/batches round-trip. + return StreamingResponse( + _ndjson_stream(result, batch_id=rec.id), + media_type="application/x-ndjson", + ) + + +@router.post("/api/parse-835") +async def parse_835_endpoint( + request: Request, + file: UploadFile = File(...), + payer: str = Query("co_medicaid_835"), + include_raw_segments: bool = Query(True), + strict: bool = Query(False), +) -> Any: + 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)}, + ) + + config = _resolve_payer_835(payer) + + # SP35 guard 1: envelope check. Mirrors the parse-837 path: tokenize, + # read ST01, reject anything that doesn't start with "835". Same + # defense-in-depth rationale — the UI auto-detect (src/pages/Upload.tsx) + # is layer A, but server-side guards protect every API caller. + try: + _segments_835 = _tokenize_segments(text) + detected_st_835 = _transaction_set_id_from_segments(_segments_835) or "" + except CycloneParseError: + detected_st_835 = "" + + if detected_st_835 and not detected_st_835.upper().startswith("835"): + return JSONResponse( + status_code=400, + content={ + "error": "Mismatched file kind", + "expected": "835", + "detected_st": detected_st_835, + "detail": ( + f"File declares ST*{detected_st_835}* but this endpoint " + f"expects ST*835*. Pick the matching endpoint on the " + f"Upload page (or let auto-detect choose for you)." + ), + }, + ) + + try: + result = parse_835(text, config, 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") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + # SP35 guard 2: empty-claims check. Same as parse-837: a BatchRecord + # with claims=[] is never a valid production 835 batch and we refuse + # to persist it. + if not result.claims: + return JSONResponse( + status_code=400, + content={ + "error": "No claims parsed", + "detail": ( + "The file passed the envelope check but contained no " + "CLP segments. Refusing to persist an empty batch." + ), + }, + ) + + # Always run the validator; attach the report so the JSON path can + # surface it and the NDJSON path can fold the counts into the summary. + # 835 validation is batch-level, so pass/fail applies uniformly to every + # claim payment in the batch (passed=N or 0, failed=0 or N). + report = validate_835(result, config) + n = len(result.claims) + claim_ids = [c.payer_claim_control_number for c in result.claims] + if report.passed: + passed, failed, failed_claim_ids = n, 0, [] + else: + passed, failed, failed_claim_ids = 0, n, claim_ids + result = result.model_copy(update={ + "validation": report, + "summary": result.summary.model_copy(update={ + "passed": passed, + "failed": failed, + "failed_claim_ids": failed_claim_ids, + }), + }) + + if strict: + result = _strict_rewrite_835(result) + if not include_raw_segments: + result = _drop_raw_segments_835(result) + + if _has_835_validation_errors(result): + return JSONResponse( + status_code=422, + content=json.loads(result.model_dump_json()), + ) + + # Persist the cleaned-up result so the cleaned result is what clients + # retrieve via /api/batches/{id}. + rec = BatchRecord( + id=uuid.uuid4().hex, + kind="835", + input_filename=file.filename or "upload.txt", + parsed_at=utcnow(), + result=result, + ) + try: + store.add(rec, event_bus=request.app.state.event_bus) + except IntegrityError as exc: + log.warning("Duplicate remittance while persisting batch %s: %s", rec.id, exc) + return JSONResponse( + status_code=409, + content={ + "error": "Duplicate remittance", + "detail": ( + "This 835 file (or one previously ingested with the " + "same payer claim control number) collides with an " + "existing record. Remove the existing remittance " + "before retrying." + ), + "batch_id": rec.id, + }, + ) + + if _client_wants_json(request): + body = json.loads(result.model_dump_json()) + body["reconciliation"] = _reconciliation_summary_for_batch(rec.id) + return JSONResponse(content=body) + + # Default: NDJSON stream. Pass the server-side batch id so the + # streaming client can call batch-scoped endpoints without a + # separate GET /api/batches round-trip (see /api/parse-837 for the + # parallel change). + return StreamingResponse( + _ndjson_stream_835(result, batch_id=rec.id), + media_type="application/x-ndjson", + ) + + +@router.post("/api/parse-999") +async def parse_999_endpoint( + request: Request, + file: UploadFile = File(...), +) -> Any: + """Parse a 999 ACK file, persist a row, and return JSON. + + Behavior mirrors ``/api/parse-835``: + - 400 on empty / undecodable / malformed EDI (never 500). + - 200 on success with ``{"ack": {id, accepted_count, rejected_count, + received_count, ack_code, raw_999_text}, "parsed": }``. + + The persisted ``acks.source_batch_id`` is a synthetic id + (``999-``) because a received 999 has no inbound 837 batch + to FK to. The dashboard's `/acks` list surfaces these; operators + can still see which interchange each row came from. + """ + 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_999_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 999") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + received, accepted, rejected, ack_code = _ack_count_summary(result) + icn = result.envelope.control_number + synthetic_id = _ack_synthetic_source_batch_id(icn) + + # Build the raw 999 text from the parsed result (round-trip). + raw_999_text = serialize_999(result, interchange_control_number=icn or "000000001") + + # SP6 T4: move claims whose 999 set was rejected into ClaimState.REJECTED. + # The 999's set_control_number (AK202) is the source 837's ST02; in + # practice we look it up against patient_control_number because that's + # the field 999 ACKs cross-reference in this product. + with db.SessionLocal()() as session: + def _lookup(pcn: str): + return session.query(Claim).filter_by(patient_control_number=pcn).first() + _rejection_result = apply_999_rejections( + session, result, claim_lookup=_lookup, + ) + if _rejection_result.matched: + bus = request.app.state.event_bus + for cid in _rejection_result.matched: + await bus.publish("claim.rejected", {"claim_id": cid}) + if _rejection_result.orphans: + log.warning( + "999 had %d orphan set refs: %s", + len(_rejection_result.orphans), + _rejection_result.orphans[:5], + ) + + row = store.add_ack( + source_batch_id=synthetic_id, + accepted_count=accepted, + rejected_count=rejected, + received_count=received, + ack_code=ack_code, + raw_json=json.loads(result.model_dump_json()), + event_bus=request.app.state.event_bus, + ) + + # SP11: append one audit row per rejected claim. Each row chains + # to the previous one — see cyclone.audit_log. + if _rejection_result.matched: + with db.SessionLocal()() as audit_s: + for cid in _rejection_result.matched: + append_event(audit_s, AuditEvent( + event_type="claim.rejected", + entity_type="claim", + entity_id=cid, + payload={"source_batch_id": synthetic_id, "ack_id": row.id}, + actor="999-parser", + user_id=_actor_user_id(request), + )) + audit_s.commit() + + # SP28: auto-link the 999 AK2 set-responses to claims via the + # D10 two-pass join (ST02 via batch envelope index primary, + # Claim.patient_control_number fallback). Each created ClaimAck + # row publishes claim_ack_written so the live-tail subscribers + # on the claim and ack side see the link immediately. + claim_ack_links_count = 0 + with db.SessionLocal()() as link_s: + batch_index = store.batch_envelope_index() + + def _pcn_lookup(pcn: str): + return ( + link_s.query(Claim) + .filter(Claim.patient_control_number == pcn) + .first() + ) + + link_result = _apply_999_acceptances( + link_s, result, ack_id=row.id, + batch_envelope_index=batch_index, + pc_claim_lookup=_pcn_lookup, + ) + for link_row in link_result.linked: + store.add_claim_ack( + claim_id=link_row.claim_id, + batch_id=link_row.batch_id, + ack_id=row.id, + ack_kind="999", + ak2_index=link_row.ak2_index, + set_control_number=link_row.set_control_number, + set_accept_reject_code=link_row.set_accept_reject_code, + linked_by="auto", + event_bus=request.app.state.event_bus, + ) + claim_ack_links_count += 1 + + return JSONResponse(content={ + "ack": { + "id": row.id, + "accepted_count": accepted, + "rejected_count": rejected, + "received_count": received, + "ack_code": ack_code, + "source_batch_id": synthetic_id, + "raw_999_text": raw_999_text, + "claim_ack_links_count": claim_ack_links_count, + }, + "parsed": json.loads(result.model_dump_json()), + }) + + +@router.post("/api/parse-ta1") +async def parse_ta1_endpoint( + request: Request, + file: UploadFile = File(...), +) -> Any: + """Parse a TA1 (Interchange Acknowledgment) file, persist a row, return JSON. + + Mirrors ``/api/parse-999`` but for the lower-level envelope ack: + - 400 on empty / undecodable / malformed EDI (never 500). + - 200 on success with ``{"ta1": {id, control_number, ack_code, + note_code, interchange_date, interchange_time, sender_id, + receiver_id, source_batch_id, raw_ta1_text}, "parsed": + }``. + + The persisted ``ta1_acks.source_batch_id`` is a synthetic id + (``TA1-``) because a received TA1 has no inbound batch to + FK to. The dashboard's ``/ta1-acks`` list surfaces these. + + SP25: threads ``event_bus=request.app.state.event_bus`` into + ``store.add_ta1_ack`` so the live-tail ``ta1_ack_received`` + stream fires on manual uploads (not just on the SFTP poller). + """ + 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_ta1_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 TA1") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + # Build the raw TA1 text from the parsed result (round-trip). + raw_ta1_text = _serialize_ta1(result) + + row = store.add_ta1_ack( + source_batch_id=result.source_batch_id, + control_number=result.ta1.control_number, + interchange_date=result.ta1.interchange_date, + interchange_time=result.ta1.interchange_time, + ack_code=result.ta1.ack_code, + note_code=result.ta1.note_code, + ack_generated_date=result.ta1.ack_generated_date, + sender_id=result.envelope.sender_id, + receiver_id=result.envelope.receiver_id, + raw_json=json.loads(result.model_dump_json()), + event_bus=request.app.state.event_bus, + ) + + # SP28: TA1 envelope-level link to the originating Batch. The + # closure here matches the most-recent Batch whose envelope + # sender_id/receiver_id matches the TA1 — see spec §D4. + claim_ack_links_count = 0 + with db.SessionLocal()() as link_s: + def _batch_lookup(sender_id, receiver_id): + rows = ( + link_s.query(Batch) + .filter( + Batch.kind == "837p", + Batch.raw_result_json.isnot(None), + ) + .order_by(Batch.parsed_at.desc()) + .all() + ) + for row in rows: + env = (row.raw_result_json or {}).get("envelope") or {} + if ( + env.get("sender_id") == sender_id + and env.get("receiver_id") == receiver_id + ): + return row + return None + + link_result = _apply_ta1_envelope_link( + link_s, result, ack_id=row.id, + batch_lookup=_batch_lookup, + ) + for link_row in link_result.linked: + store.add_claim_ack( + claim_id=link_row.claim_id, + batch_id=link_row.batch_id, + ack_id=row.id, + ack_kind="ta1", + ak2_index=link_row.ak2_index, + set_control_number=link_row.set_control_number, + set_accept_reject_code=link_row.set_accept_reject_code, + linked_by="auto", + event_bus=request.app.state.event_bus, + ) + claim_ack_links_count += 1 + + return JSONResponse(content={ + "ta1": { + "id": row.id, + "control_number": result.ta1.control_number, + "ack_code": result.ta1.ack_code, + "note_code": result.ta1.note_code, + "interchange_date": result.ta1.interchange_date.isoformat() + if result.ta1.interchange_date else None, + "interchange_time": result.ta1.interchange_time, + "sender_id": result.envelope.sender_id, + "receiver_id": result.envelope.receiver_id, + "source_batch_id": result.source_batch_id, + "raw_ta1_text": raw_ta1_text, + "claim_ack_links_count": claim_ack_links_count, + }, + "parsed": json.loads(result.model_dump_json()), + }) + + +@router.post("/api/parse-277ca") +async def parse_277ca_endpoint( + request: Request, + file: UploadFile = File(...), +) -> Any: + """Parse a 277CA Claim Acknowledgment file, persist a row, and stamp rejections. + + Behavior mirrors ``/api/parse-999``: + - 400 on empty / undecodable / malformed EDI (never 500). + - 200 on success with ``{"ack": {id, control_number, accepted_count, + rejected_count, payer_claim_control_numbers, raw_277ca_text}, + "parsed": }``. + + After parse, runs :func:`apply_277ca_rejections` to stamp the + payer-rejected fields on each matching claim row. The Inbox + Payer-Rejected lane lights up as a side-effect of this call. + """ + 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_277ca_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 277CA") + return JSONResponse( + status_code=500, + content={"error": "Internal server error", "detail": str(exc)}, + ) + + icn = result.envelope.control_number + synthetic_id = _277ca_synthetic_source_batch_id(icn) + + accepted = sum(1 for s in result.claim_statuses if s.classification == "accepted") + paid = sum(1 for s in result.claim_statuses if s.classification == "paid") + rejected = sum(1 for s in result.claim_statuses if s.classification == "rejected") + pended = sum(1 for s in result.claim_statuses if s.classification == "pended") + + # Persist the 277CA row first so we have an id to attach to claims. + # SP25: thread the event bus so ``two77ca_ack_received`` fires. + row = store.add_277ca_ack( + source_batch_id=synthetic_id, + control_number=icn, + accepted_count=accepted, + rejected_count=rejected, + paid_count=paid, + pended_count=pended, + raw_json=json.loads(result.model_dump_json()), + event_bus=request.app.state.event_bus, + ) + + # Stamp payer-rejection fields on matching claims. The 277CA's + # REF*1K carries the patient's claim control number we sent in + # CLM01 — same convention the 999 ACK uses, so the lookup hits + # Claim.patient_control_number (mirrors apply_999_rejections). + with db.SessionLocal()() as session: + def _lookup(pcn: str): + return ( + session.query(Claim) + .filter(Claim.patient_control_number == pcn) + .first() + ) + apply_result = apply_277ca_rejections( + session, result, claim_lookup=_lookup, two77ca_id=row.id, + ) + if apply_result.matched: + bus = request.app.state.event_bus + for cid in apply_result.matched: + await bus.publish("claim.payer_rejected", {"claim_id": cid}) + # SP11: audit trail for each payer-rejected claim. + with db.SessionLocal()() as audit_s: + for cid in apply_result.matched: + append_event(audit_s, AuditEvent( + event_type="claim.payer_rejected", + entity_type="claim", + entity_id=cid, + payload={ + "source_batch_id": synthetic_id, + "277ca_id": row.id, + }, + actor="277ca-parser", + user_id=_actor_user_id(request), + )) + audit_s.commit() + if apply_result.orphans: + log.warning( + "277CA had %d orphan status entries (no matching claim): %s", + len(apply_result.orphans), + apply_result.orphans[:5], + ) + + # SP28: auto-link the 277CA ClaimStatus entries to claims via + # the D10 two-pass join. Each ClaimAck row publishes + # claim_ack_written so the live-tail subscribers on the claim + # and ack side see the link immediately. + claim_ack_links_count = 0 + with db.SessionLocal()() as link_s: + batch_index = store.batch_envelope_index() + + def _pcn_lookup(pcn: str): + return ( + link_s.query(Claim) + .filter(Claim.patient_control_number == pcn) + .first() + ) + + link_result = _apply_277ca_acks( + link_s, result, ack_id=row.id, + batch_envelope_index=batch_index, + pc_claim_lookup=_pcn_lookup, + ) + for link_row in link_result.linked: + store.add_claim_ack( + claim_id=link_row.claim_id, + batch_id=link_row.batch_id, + ack_id=row.id, + ack_kind="277ca", + ak2_index=link_row.ak2_index, + set_control_number=link_row.set_control_number, + set_accept_reject_code=link_row.set_accept_reject_code, + linked_by="auto", + event_bus=request.app.state.event_bus, + ) + claim_ack_links_count += 1 + + return JSONResponse(content={ + "ack": { + "id": row.id, + "control_number": icn, + "accepted_count": accepted, + "rejected_count": rejected, + "paid_count": paid, + "pended_count": pended, + "source_batch_id": synthetic_id, + "matched_claim_ids": apply_result.matched, + "orphan_status_codes": apply_result.orphans, + "claim_ack_links_count": claim_ack_links_count, + }, + "parsed": json.loads(result.model_dump_json()), + }) \ No newline at end of file diff --git a/backend/src/cyclone/handlers/handle_835.py b/backend/src/cyclone/handlers/handle_835.py index 75db89f..c7535a5 100644 --- a/backend/src/cyclone/handlers/handle_835.py +++ b/backend/src/cyclone/handlers/handle_835.py @@ -66,10 +66,12 @@ def handle( here was dead code anyway (the 835 event name is ``remittance_written``, not ``ack_received``). """ - # TODO(sp27-pre-t5): move PAYER_FACTORIES_835 out of api.py into - # ``cyclone.payers`` to remove the lazy cyclic import below. The - # import works today because api.py also imports scheduler lazily. - from cyclone.api import PAYER_FACTORIES_835 + # SP36 Task 16: PAYER_FACTORIES_835 moved from ``cyclone.api`` + # to ``cyclone.api_routers._shared`` (cross-router helper). Import + # from the new home. The lazy import is still needed to avoid a + # circular import at registry load time (handle_835 is imported + # by the scheduler during the api lifespan). + from cyclone.api_routers._shared import PAYER_FACTORIES_835 config = PAYER_FACTORIES_835["co_medicaid_835"]() try: