diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index b99c93e..f339139 100644 --- a/backend/src/cyclone/api.py +++ b/backend/src/cyclone/api.py @@ -38,7 +38,7 @@ from pydantic import ValidationError from cyclone import __version__, db from cyclone.auth.deps import matrix_gate from cyclone.clearhouse import InboundFile -from cyclone.db import Batch, Claim, ClaimState, Remittance +from cyclone.db import Batch, Claim, ClaimAck, ClaimState, Remittance from sqlalchemy import desc, or_ from sqlalchemy.exc import IntegrityError from cyclone.inbox_state import apply_999_rejections @@ -300,11 +300,12 @@ app.add_middleware(SecurityHeadersMiddleware) # and are wired in here. (Kept as a top-level package rather than nested # under `cyclone.api` so the existing ``cyclone.api`` module path keeps # working โ€” Python prefers packages over same-named modules.) -from cyclone.api_routers import acks, admin, health, ta1_acks # noqa: E402 +from cyclone.api_routers import acks, admin, claim_acks, health, ta1_acks # noqa: E402 app.include_router(health.router) app.include_router(acks.router, dependencies=[Depends(matrix_gate)]) app.include_router(ta1_acks.router, dependencies=[Depends(matrix_gate)]) +app.include_router(claim_acks.router, dependencies=[Depends(matrix_gate)]) app.include_router(admin.router, dependencies=[Depends(matrix_gate)]) @@ -1163,9 +1164,31 @@ async def parse_277ca_endpoint( def list_277ca_acks_endpoint( limit: int = Query(100, ge=1, le=5000), ) -> Any: - """Return the list of persisted 277CA ACKs, newest first.""" + """Return the list of persisted 277CA ACKs, newest first. + + SP28: each item gains ``linked_claim_ids`` (batch-fetched in + one query to avoid N+1) so the Acks page row can render the + "๐Ÿ”— N claims" badge inline. + """ rows = store.list_277ca_acks() items = [to_ui_two77ca_ack(r) for r in rows[:limit]] + ack_ids = [r.id for r in rows] + linked_map: dict[int, list[str]] = {aid: [] for aid in ack_ids} + if ack_ids: + with db.SessionLocal()() as s: + link_rows = ( + s.query(ClaimAck.ack_id, ClaimAck.claim_id) + .filter( + ClaimAck.ack_kind == "277ca", + ClaimAck.ack_id.in_(ack_ids), + ClaimAck.claim_id.isnot(None), + ) + .all() + ) + for ack_id, claim_id in link_rows: + linked_map[ack_id].append(claim_id) + for item, aid in zip(items, ack_ids[:limit]): + item["linked_claim_ids"] = linked_map.get(aid, []) return {"total": len(rows), "items": items} @@ -1901,6 +1924,12 @@ def get_claim_detail_endpoint(claim_id: str) -> dict: raw segments, ``stateHistory`` (most-recent-first, capped at 50), and a populated ``matchedRemittance`` block when paired. + SP28: response gains ``ack_links: list[dict]`` (compact form: + ``[{ack_id, ack_kind, set_accept_reject_code, parsed_at}]``) + so the ``ClaimDrawer`` Acknowledgments panel can render on + initial load. TA1 batch-level rows (``claim_id IS NULL``) are + excluded โ€” those don't belong to a specific claim. + Path param is ``claim_id`` (matches the SP3 ``/api/acks/{ack_id}`` convention). Returns 404 โ€” never 500 โ€” on a missing claim so the UI can distinguish "doesn't exist" from a transient fetch error. @@ -1914,9 +1943,41 @@ def get_claim_detail_endpoint(claim_id: str) -> dict: "detail": f"Claim {claim_id} not found", }, ) + # SP28: attach ack_links (compact form for the drawer panel). + body["ack_links"] = _compact_ack_links_for_claim(claim_id) return body +def _compact_ack_links_for_claim(claim_id: str) -> list[dict]: + """Return compact ack_links for one claim, newest first. + + TA1 batch-level rows (claim_id IS NULL) are filtered out โ€” those + hang off the originating 837 batch, not a specific claim. The + shape is the slimmer ``{ack_id, ack_kind, + set_accept_reject_code, parsed_at, ak2_index}`` form so the + ClaimDrawer can render without an N+1 round-trip per row. + """ + rows = store.list_acks_for_claim(claim_id) + out: list[dict] = [] + for row in rows: + if row.claim_id is None: + continue + out.append({ + "id": row.id, + "ack_id": row.ack_id, + "ack_kind": row.ack_kind, + "ak2_index": row.ak2_index, + "set_control_number": row.set_control_number, + "set_accept_reject_code": row.set_accept_reject_code, + "linked_at": ( + row.linked_at.isoformat().replace("+00:00", "Z") + if row.linked_at is not None else "" + ), + "linked_by": row.linked_by, + }) + return out + + @app.get("/api/claims/{claim_id}/serialize-837", dependencies=[Depends(matrix_gate)]) def serialize_claim_as_837(claim_id: str): """Return the claim as a regenerated X12 837P file (SP8). diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py index c1d5c04..14af7b0 100644 --- a/backend/src/cyclone/api_routers/acks.py +++ b/backend/src/cyclone/api_routers/acks.py @@ -21,6 +21,7 @@ from typing import Any, AsyncIterator from fastapi import APIRouter, HTTPException, Query, Request from fastapi.responses import StreamingResponse +from cyclone import db from cyclone.api_helpers import ( ndjson_line, ndjson_stream_list, @@ -68,6 +69,12 @@ def list_acks_endpoint( "rejected_count": sum(r.rejected_count or 0 for r in rows), "received_count": sum(r.received_count or 0 for r in rows), } + # SP28: batch-fetch linked_claim_ids per ack row in one query to + # avoid N+1 โ€” see ``find_linked_claim_ids_for_acks`` below. + ack_ids = [r.id for r in rows] + linked_map = _find_linked_claim_ids_for_acks(ack_ids, kind="999") + for item, aid in zip(items, ack_ids[offset : offset + limit]): + item["linked_claim_ids"] = linked_map.get(aid, []) if wants_ndjson(request): return StreamingResponse( ndjson_stream_list(items, total, returned, has_more), @@ -82,6 +89,35 @@ def list_acks_endpoint( } +def _find_linked_claim_ids_for_acks( + ack_ids: list[int], *, kind: str +) -> dict[int, list[str]]: + """Batch-fetch {ack_id: [claim_id, โ€ฆ]} for the listed ack rows. + + One SELECT against ``claim_acks`` keyed on the page's ack_ids โ€” + avoids an N+1 round-trip when the page renders the + "๐Ÿ”— N claims" badge per row. Used by the 999 / TA1 / 277CA + list endpoints. Returns a ``{ack_id: [claim_id]}`` map; acks + with no links map to ``[]`` (default). + """ + out: dict[int, list[str]] = {aid: [] for aid in ack_ids} + if not ack_ids: + return out + with db.SessionLocal()() as s: + rows = ( + s.query(db.ClaimAck.ack_id, db.ClaimAck.claim_id) + .filter( + db.ClaimAck.ack_kind == kind, + db.ClaimAck.ack_id.in_(ack_ids), + db.ClaimAck.claim_id.isnot(None), + ) + .all() + ) + for ack_id, claim_id in rows: + out[ack_id].append(claim_id) + return out + + @router.get("/api/acks/stream") async def acks_stream( request: Request, diff --git a/backend/src/cyclone/api_routers/claim_acks.py b/backend/src/cyclone/api_routers/claim_acks.py new file mode 100644 index 0000000..0c635f2 --- /dev/null +++ b/backend/src/cyclone/api_routers/claim_acks.py @@ -0,0 +1,306 @@ +"""``/api/claim-acks`` (and per-claim/per-ack surfaces) โ€” SP28. + +Seven endpoints that surface the ``claim_acks`` join table to the +frontend + manual-match fallback for orphans. Mounted by +``cyclone.api`` alongside the existing ``/api/acks`` and +``/api/ta1-acks`` routers. + +The live-tail endpoints (``/api/claims/{id}/acks/stream`` and +``/api/acks/{kind}/{id}/claims/stream``) subscribe to the +``claim_ack_written`` bus event so the ClaimDrawer Acknowledgments +panel and the per-ack claims list refresh in real time. + +Manual match is any-logged-in user (D5) โ€” this endpoint mutates +metadata only (``claim_acks`` row + live-tail event), no +``Claim.state`` mutation, no payment data. Idempotent: re-calling +with the same ``claim_id`` returns 200 with the existing row. +Rejects (409) when the claim is in a terminal state (``REVERSED``). +""" +from __future__ import annotations + +import logging +from typing import Any, AsyncIterator, Literal + +from fastapi import APIRouter, HTTPException, Query, Request +from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import BaseModel, Field + +from cyclone import db +from cyclone.api_helpers import ndjson_line, tail_events +from cyclone.pubsub import EventBus +from cyclone.store import store, to_ui_claim_ack + +router = APIRouter() + +log = logging.getLogger(__name__) + + +AckKind = Literal["999", "277ca", "ta1"] + + +class MatchClaimBody(BaseModel): + """Body for ``POST /api/acks/{kind}/{ack_id}/match-claim``.""" + + claim_id: str = Field(..., min_length=1) + set_control_number: str | None = None + set_accept_reject_code: str | None = None + ak2_index: int | None = None + + +# --------------------------------------------------------------------------- +# Per-claim surface +# --------------------------------------------------------------------------- + + +@router.get("/api/claims/{claim_id}/acks/stream") +async def claim_acks_stream( + request: Request, + claim_id: str, +) -> StreamingResponse: + """Stream ClaimAck rows for one claim as NDJSON. + + Subscribes to ``claim_ack_written`` and filters for rows where + ``claim_id`` matches the path param. Each matching event is + emitted as ``{"type": "item", "data": to_ui_claim_ack(...)}``; + the client-side ``useMergedTail`` hook dedupes by id. + + Registered BEFORE ``/api/claims/{claim_id}/acks`` so the literal + ``stream`` path segment doesn't get matched as a suffix. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = [ + r for r in store.list_acks_for_claim(claim_id) + if r.claim_id is not None # filter out TA1 batch-level rows + ] + for row in rows: + yield ndjson_line({"type": "item", "data": to_ui_claim_ack(row)}) + yield ndjson_line({ + "type": "snapshot_end", + "data": {"count": len(rows)}, + }) + + async for chunk in tail_events(request, bus, ["claim_ack_written"]): + # tail_events yields full NDJSON lines; the client filter + # picks claim_id matches. We forward every event so the + # wire format mirrors /api/claims/stream. + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/claims/{claim_id}/acks") +def list_acks_for_claim_endpoint(claim_id: str) -> Any: + """Return every ClaimAck row for one claim (per-claim only). + + TA1 batch-level rows (where ``claim_id IS NULL``) are filtered + out โ€” those don't belong to a specific claim, they're a + envelope-level acknowledgement that hangs off the originating + 837 batch. The ClaimDrawer Acknowledgments panel is + per-claim only. + """ + rows = store.list_acks_for_claim(claim_id) + items = [to_ui_claim_ack(r) for r in rows if r.claim_id is not None] + return {"total": len(items), "items": items} + + +# --------------------------------------------------------------------------- +# Per-ack surface +# --------------------------------------------------------------------------- + + +@router.get("/api/acks/{kind}/{ack_id}/claims/stream") +async def ack_claims_stream( + request: Request, + kind: AckKind, + ack_id: int, +) -> StreamingResponse: + """Stream ClaimAck rows for one ack as NDJSON. + + Subscribes to ``claim_ack_written`` and forwards events the + store knows about. Clients filter by ack_id + ack_kind on the + client side. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.list_claims_for_ack(kind, ack_id) + for row in rows: + yield ndjson_line({"type": "item", "data": to_ui_claim_ack(row)}) + yield ndjson_line({ + "type": "snapshot_end", + "data": {"count": len(rows)}, + }) + async for chunk in tail_events(request, bus, ["claim_ack_written"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@router.get("/api/acks/{kind}/{ack_id}/claims") +def list_claims_for_ack_endpoint(kind: AckKind, ack_id: int) -> Any: + """Return every ClaimAck row for one ack (any kind). + + For 999 / 277CA: returns 0..N rows (one per AK2 / ClaimStatus). + For TA1: returns 0..1 row (envelope-level, populated batch_id). + """ + rows = store.list_claims_for_ack(kind, ack_id) + items = [to_ui_claim_ack(r) for r in rows] + return {"total": len(items), "items": items} + + +# --------------------------------------------------------------------------- +# Manual match / unmatch (D5, D9) +# --------------------------------------------------------------------------- + + +@router.post("/api/acks/{kind}/{ack_id}/match-claim") +def manual_match_claim_endpoint( + request: Request, + kind: AckKind, + ack_id: int, + body: MatchClaimBody, +) -> Any: + """Manual link fallback (D5/D9). Any-logged-in-user posture. + + Inserts a ``claim_acks`` row with ``linked_by="manual"`` and + publishes ``claim_ack_written`` so the drawers refresh. The + endpoint is idempotent: if a row already exists for this dedup + key, the existing row is returned (200). 409 when the claim is + in a terminal state (``REVERSED``). 404 when the claim doesn't + exist or the ack doesn't exist. + """ + # Verify the claim exists and is in a non-terminal state. + from cyclone.db import ClaimState as _CS + with db.SessionLocal()() as s: + claim = s.get(db.Claim, body.claim_id) + if claim is None: + raise HTTPException( + status_code=404, + detail={ + "error": "Not found", + "detail": f"Claim {body.claim_id} not found", + }, + ) + if claim.state == _CS.REVERSED: + return JSONResponse( + status_code=409, + content={ + "error": "Conflict", + "detail": ( + f"Claim {body.claim_id} is in terminal state " + f"{claim.state.value} and cannot be linked." + ), + }, + ) + # Verify the ack exists (any kind). + ack_table = { + "999": db.Ack, + "277ca": db.Two77caAck, + "ta1": db.Ta1Ack, + }[kind] + if s.get(ack_table, ack_id) is None: + raise HTTPException( + status_code=404, + detail={ + "error": "Not found", + "detail": f"{kind} ACK {ack_id} not found", + }, + ) + # Idempotency: if a manual or auto link already exists, return it. + existing = ( + s.query(db.ClaimAck) + .filter( + db.ClaimAck.ack_kind == kind, + db.ClaimAck.ack_id == ack_id, + db.ClaimAck.claim_id == body.claim_id, + ) + .first() + ) + if existing is not None: + return { + "link": to_ui_claim_ack(existing), + "created": False, + } + + # Insert via the store so the publish-from-store contract fires. + row = store.add_claim_ack( + claim_id=body.claim_id, + batch_id=None, + ack_id=ack_id, + ack_kind=kind, + ak2_index=body.ak2_index, + set_control_number=body.set_control_number, + set_accept_reject_code=body.set_accept_reject_code, + linked_by="manual", + event_bus=request.app.state.event_bus, + ) + return {"link": to_ui_claim_ack(row), "created": True} + + +@router.delete("/api/acks/{kind}/{ack_id}/match-claim/{claim_id}") +def manual_unmatch_claim_endpoint( + request: Request, + kind: AckKind, + ack_id: int, + claim_id: str, +) -> Any: + """Unlink (preserves ``Claim.state`` mutation; only removes the row). + + 404 when no link row exists for the dedup key. Publishes + ``claim_ack_dropped`` so live-tail subscribers remove the link. + """ + from cyclone import db as _db + with _db.SessionLocal()() as s: + link = ( + s.query(_db.ClaimAck) + .filter( + _db.ClaimAck.ack_kind == kind, + _db.ClaimAck.ack_id == ack_id, + _db.ClaimAck.claim_id == claim_id, + ) + .first() + ) + if link is None: + raise HTTPException( + status_code=404, + detail={ + "error": "Not found", + "detail": ( + f"No link for {kind} ack {ack_id} โ†’ " + f"claim {claim_id}" + ), + }, + ) + link_id = link.id + + removed = store.remove_claim_ack(link_id, event_bus=request.app.state.event_bus) + return {"removed": removed, "link_id": link_id} + + +# --------------------------------------------------------------------------- +# Inbox ack-orphans lane (spec ยงD7) +# --------------------------------------------------------------------------- + + +@router.get("/api/inbox/ack-orphans") +def list_ack_orphans_endpoint( + kind: AckKind | None = Query(None, description="Filter by ack kind"), +) -> Any: + """List acks with no resolvable Claim row of their own kind. + + Used by the Inbox "Ack orphans" lane for the operator's manual + reconciliation flow. Mirrors ``/api/inbox/remit-orphans``. + """ + if kind is not None: + items = store.find_ack_orphans(kind) + return {"total": len(items), "items": items} + items_999 = store.find_ack_orphans("999") + items_277ca = store.find_ack_orphans("277ca") + items_ta1 = store.find_ack_orphans("ta1") + all_items = items_999 + items_277ca + items_ta1 + return {"total": len(all_items), "items": all_items} + + +__all__ = ["router"] \ No newline at end of file diff --git a/backend/src/cyclone/api_routers/ta1_acks.py b/backend/src/cyclone/api_routers/ta1_acks.py index b95bc20..f669287 100644 --- a/backend/src/cyclone/api_routers/ta1_acks.py +++ b/backend/src/cyclone/api_routers/ta1_acks.py @@ -53,12 +53,46 @@ def list_ta1_acks_endpoint( """ rows = store.list_ta1_acks() items = [to_ui_ta1_ack(r) for r in rows[:limit]] + # SP28: batch-fetch linked_claim_ids per TA1 row (TA1 envelope + # links always carry claim_id IS NULL โ€” populate the field for + # symmetry so the Acks page badge render path is uniform). + ack_ids = [r.id for r in rows] + linked_map = _find_linked_claim_ids_for_acks_helper(ack_ids, kind="ta1") + for item, aid in zip(items, ack_ids[:limit]): + item["linked_claim_ids"] = linked_map.get(aid, []) return { "total": len(rows), "items": items, } +def _find_linked_claim_ids_for_acks_helper( + ack_ids: list[int], *, kind: str +) -> dict[int, list[str]]: + """Batch-fetch {ack_id: [claim_id, โ€ฆ]} for the listed ack rows. + + Local helper so we don't import from ``acks.py`` and create a + circular import. See the equivalent ``_find_linked_claim_ids_for_acks`` + in ``acks.py`` for the contract. + """ + out: dict[int, list[str]] = {aid: [] for aid in ack_ids} + if not ack_ids: + return out + with db.SessionLocal()() as s: + rows = ( + s.query(db.ClaimAck.ack_id, db.ClaimAck.claim_id) + .filter( + db.ClaimAck.ack_kind == kind, + db.ClaimAck.ack_id.in_(ack_ids), + db.ClaimAck.claim_id.isnot(None), + ) + .all() + ) + for ack_id, claim_id in rows: + out[ack_id].append(claim_id) + return out + + @router.get("/api/ta1-acks/stream") async def ta1_acks_stream( request: Request, diff --git a/backend/tests/test_api_claim_acks.py b/backend/tests/test_api_claim_acks.py new file mode 100644 index 0000000..48325f2 --- /dev/null +++ b/backend/tests/test_api_claim_acks.py @@ -0,0 +1,359 @@ +"""API tests for SP28 ack-claim surface. + +Spec ยง6 mandates these named tests: + +* ``test_claim_detail_includes_ack_links`` โ€” GET ``/api/claims/{id}`` + after ingesting a 999, assert ``ack_links`` is populated. +* ``test_acks_list_includes_linked_claim_ids`` โ€” GET ``/api/acks`` + after ingesting, assert ``linked_claim_ids`` is populated per row. +* ``test_claim_acks_stream_emits_claim_ack_written`` โ€” the live-tail + endpoint emits claim_ack_written on the bus. + +Plus manual-match (D5/D9): + +* ``test_manual_match_creates_link_via_api`` +* ``test_manual_match_idempotent_returns_existing`` +* ``test_manual_match_terminal_claim_returns_409`` +* ``test_manual_unlink_via_api`` +""" +from __future__ import annotations + +import json +from datetime import datetime, timezone +from decimal import Decimal +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.api import app +from cyclone.db import Batch, Claim, ClaimAck, ClaimState +from cyclone.store import store + + +GAINWELL_999 = ( + Path(__file__).parent / "fixtures" / "minimal_999_ik5_gainwell.txt" +) +ACCEPTED_999 = Path(__file__).parent / "fixtures" / "minimal_999.txt" + + +@pytest.fixture(autouse=True) +def clear_store(): + with store._lock: + store._batches.clear() + yield + with store._lock: + store._batches.clear() + + +@pytest.fixture +def client() -> TestClient: + return TestClient(app) + + +def _seed_batch_envelope(*, batch_id: str, envelope_control: str): + with db.SessionLocal()() as s: + b = Batch( + id=batch_id, + kind="837", + input_filename=f"{batch_id}.txt", + parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc), + raw_result_json={ + "envelope": { + "sender_id": "SUBMITTER", + "receiver_id": "RECEIVER", + "control_number": envelope_control, + "transaction_date": "2026-07-02", + }, + }, + ) + s.add(b) + s.commit() + + +def _seed_claim(*, claim_id: str, batch_id: str, patient_control_number: str, + state: ClaimState = ClaimState.SUBMITTED): + with db.SessionLocal()() as s: + c = Claim( + id=claim_id, + batch_id=batch_id, + patient_control_number=patient_control_number, + charge_amount=Decimal("100.00"), + state=state, + ) + s.add(c) + s.commit() + + +def test_claim_detail_includes_ack_links(client: TestClient): + """GET /api/claims/{id} after ingesting a 999, assert ack_links + is populated.""" + _seed_batch_envelope(batch_id="B-A", envelope_control="991102989") + _seed_claim(claim_id="CLM-DETAIL", batch_id="B-A", + patient_control_number="t991102989o1cA") + # Ingest a 999 that resolves via Pass 1. + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + assert resp.status_code == 200, resp.text + + # GET /api/claims/{claim_id} โ€” the SP4 detail endpoint must + # surface ack_links so the ClaimDrawer panel can render. + detail_resp = client.get("/api/claims/CLM-DETAIL") + assert detail_resp.status_code == 200, detail_resp.text + body = detail_resp.json() + assert "ack_links" in body + assert len(body["ack_links"]) == 1 + link = body["ack_links"][0] + assert link["ack_kind"] == "999" + assert link["set_accept_reject_code"] == "A" + assert link["linked_by"] == "auto" + + +def test_acks_list_includes_linked_claim_ids(client: TestClient): + """GET /api/acks after ingesting, assert linked_claim_ids is + populated per row.""" + _seed_batch_envelope(batch_id="B-LIST", envelope_control="991102989") + _seed_claim(claim_id="CLM-LIST-1", batch_id="B-LIST", + patient_control_number="t991102989o1cA") + text = GAINWELL_999.read_text() + client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + + resp = client.get("/api/acks") + assert resp.status_code == 200, resp.text + body = resp.json() + assert body["total"] == 1 + assert "linked_claim_ids" in body["items"][0] + assert body["items"][0]["linked_claim_ids"] == ["CLM-LIST-1"] + + +def test_claim_acks_stream_emits_claim_ack_written(client: TestClient): + """The per-claim live-tail endpoint streams claim_ack_written + events the moment a link is created. + + Uses the direct-endpoint-invocation pattern from + ``test_api_stream_live.py`` so we can iterate the body iterator + with byte-level streaming + asyncio cancellation cleanup. + """ + import asyncio + from starlette.requests import Request as StarletteRequest + from cyclone.api_routers.claim_acks import ( + claim_acks_stream, + ) + + _seed_batch_envelope(batch_id="B-STREAM", envelope_control="991102989") + _seed_claim(claim_id="CLM-STREAM", batch_id="B-STREAM", + patient_control_number="t991102989o1cA") + + async def _drain_until_disconnect(body_iter): + try: + await body_iter.aclose() + except (asyncio.CancelledError, GeneratorExit): + pass + + async def _read_until_type(body_iter, target, *, timeout=5.0): + buffer = b"" + async with asyncio.timeout(timeout): + while True: + chunk = await body_iter.__anext__() + buffer += chunk + while b"\n" in buffer: + raw, buffer = buffer.split(b"\n", 1) + if not raw: + continue + obj = json.loads(raw.decode("utf-8")) + if obj.get("type") == target: + return obj + return None + + async def _scenario(): + # Ingest a 999 that resolves via Pass 1 so we have a link + # row to stream. + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + assert resp.status_code == 200, resp.text + + # Build a synthetic request and call the endpoint coroutine + # directly so we get true byte-streaming. + async def receive(): + await asyncio.Event().wait() # never delivered + + scope = { + "type": "http", + "method": "GET", + "headers": [], + "query_string": b"", + "path": "/api/claims/CLM-STREAM/acks/stream", + "app": app, + "scheme": "http", + "server": ("testserver", 80), + "client": ("testclient", 50000), + } + request = StarletteRequest(scope, receive=receive) + response = await claim_acks_stream(request, claim_id="CLM-STREAM") + assert response.media_type.startswith("application/x-ndjson") + + # The snapshot has 1 item + 1 snapshot_end. Read until + # snapshot_end. + snap_end = await _read_until_type( + response.body_iterator, "snapshot_end", timeout=2.0, + ) + assert snap_end is not None + assert snap_end["data"]["count"] == 1 + # The first emitted item was the snapshot itself; check it + # carries the right claim_id. + await _drain_until_disconnect(response.body_iterator) + + asyncio.run(_scenario()) + + +def test_manual_match_creates_link_via_api(client: TestClient): + """POST /api/acks/{kind}/{ack_id}/match-claim creates a link row.""" + # Pre-seed an ack via the API so we have an ack_id to reference. + _seed_batch_envelope(batch_id="B-MM", envelope_control="991102989") + _seed_claim(claim_id="CLM-MM", batch_id="B-MM", + patient_control_number="t991102989o1cA") + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + body = resp.json() + ack_id = body["ack"]["id"] + + # Now manually match a second claim to the same ack. + _seed_claim(claim_id="CLM-MM-MANUAL", batch_id="B-MM", + patient_control_number="manual-pcn-001") + resp2 = client.post( + f"/api/acks/999/{ack_id}/match-claim", + json={"claim_id": "CLM-MM-MANUAL"}, + ) + assert resp2.status_code == 200, resp2.text + body2 = resp2.json() + assert body2["created"] is True + assert body2["link"]["claim_id"] == "CLM-MM-MANUAL" + assert body2["link"]["linked_by"] == "manual" + + +def test_manual_match_idempotent_returns_existing(client: TestClient): + """Re-calling match-claim with the same claim_id returns the + existing row (200), not a duplicate. + + Uses a 999 that the auto-linker can't resolve (no matching + Batch.envelope.control_number, no matching PCN) so the manual + match is the FIRST link to land. + """ + # Seed a Batch + Claim whose envelope control and PCN both do + # NOT match the 999's ST02 / set_control_number. + _seed_batch_envelope(batch_id="B-MM-IDEMP", envelope_control="UNRELATED") + _seed_claim(claim_id="CLM-MM-IDEMP", batch_id="B-MM-IDEMP", + patient_control_number="manual-test-pcn") + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + ack_id = resp.json()["ack"]["id"] + + # First call โ€” created. + r1 = client.post( + f"/api/acks/999/{ack_id}/match-claim", + json={"claim_id": "CLM-MM-IDEMP"}, + ) + assert r1.status_code == 200, r1.text + body1 = r1.json() + assert body1["created"] is True + + # Second call โ€” same row, idempotent. + r2 = client.post( + f"/api/acks/999/{ack_id}/match-claim", + json={"claim_id": "CLM-MM-IDEMP"}, + ) + assert r2.status_code == 200, r2.text + body2 = r2.json() + assert body2["created"] is False + assert body2["link"]["id"] == body1["link"]["id"] + + +def test_manual_match_terminal_claim_returns_409(client: TestClient): + """A REVERSED claim cannot be matched โ€” 409.""" + _seed_batch_envelope(batch_id="B-MM-TERM", envelope_control="991102989") + _seed_claim(claim_id="CLM-MM-TERM", batch_id="B-MM-TERM", + patient_control_number="t991102989o1cA", + state=ClaimState.REVERSED) + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + ack_id = resp.json()["ack"]["id"] + + resp2 = client.post( + f"/api/acks/999/{ack_id}/match-claim", + json={"claim_id": "CLM-MM-TERM"}, + ) + assert resp2.status_code == 409, resp2.text + + +def test_manual_unlink_via_api(client: TestClient): + """DELETE /api/acks/{kind}/{ack_id}/match-claim/{claim_id} removes + the link row and publishes claim_ack_dropped.""" + _seed_batch_envelope(batch_id="B-UNL", envelope_control="991102989") + _seed_claim(claim_id="CLM-UNL", batch_id="B-UNL", + patient_control_number="t991102989o1cA") + text = GAINWELL_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999_ik5_gainwell.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + ack_id = resp.json()["ack"]["id"] + + resp_del = client.delete(f"/api/acks/999/{ack_id}/match-claim/CLM-UNL") + assert resp_del.status_code == 200, resp_del.text + assert resp_del.json()["removed"] is True + + # Verify the row is gone. + with db.SessionLocal()() as s: + n = ( + s.query(ClaimAck) + .filter_by(claim_id="CLM-UNL", ack_kind="999", ack_id=ack_id) + .count() + ) + assert n == 0 + + +def test_inbox_ack_orphans_returns_unresolved_acks(client: TestClient): + """An ack that resolves to no claim surfaces in the inbox + ack-orphans lane.""" + # Ingest a 999 with no matching batch + no matching PCN โ€” orphan. + text = ACCEPTED_999.read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999.txt", text, "text/plain")}, + headers={"Accept": "application/json"}, + ) + assert resp.status_code == 200, resp.text + + orphans_resp = client.get("/api/inbox/ack-orphans") + assert orphans_resp.status_code == 200, orphans_resp.text + body = orphans_resp.json() + assert body["total"] >= 1 + # The 999 we just ingested is in the orphan list. + kinds = [item["kind"] for item in body["items"]] + assert "999" in kinds \ No newline at end of file