feat(sp27): extract handle_277ca from scheduler.py into handlers/

backend/src/cyclone/handlers/handle_277ca.py

@@ -0,0 +1,122 @@
+"""Handle a 277CA Claim Acknowledgment file (SP27 Task 4).
+
+Lifted verbatim from ``scheduler.py:_handle_277ca``. The handler owns
+its own DB session, dispatches to ``parse_277ca_text``, persists the
+277CA ack row, applies 277CA rejections to matched claims via
+``inbox_state.apply_277ca_rejections``, and emits
+``claim.payer_rejected`` audit events for each newly-stamped claim.
+
+The actor tag (``"277ca-parser-scheduler"``) is preserved so the
+audit log keeps tracing back to the same source after extraction.
+Both the FastAPI endpoint and the scheduler path (re)use this module
+— the API migration drops the inline copy in Task 6.
+
+``claim.rejected_after_remit`` audit emission (when a 277CA rejects
+a claim that already has ``matched_remittance_id`` set) is deferred
+to SP27 Task 13. Today the handler only emits ``claim.payer_rejected``.
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Optional
+
+from cyclone import db
+from cyclone.audit_log import AuditEvent, append_event
+from cyclone.handlers._ack_id import two77ca_synthetic_source_batch_id
+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.store import store as cycl_store
+
+log = logging.getLogger(__name__)
+
+
+def handle(
+    text: str,
+    source_file: str,
+    *,
+    event_bus: Optional[object] = None,
+) -> tuple[str, int]:
+    """Parse a 277CA, persist ack + stamp payer-rejected claims.
+
+    Args:
+        text: Raw 277CA document bytes (decoded).
+        source_file: Filename the 277CA came from.
+        event_bus: Optional pubsub handle. Reserved for the
+            FastAPI-endpoint migration in Task 6 — the scheduler
+            doesn't pass one.
+
+    Returns:
+        ``(parser_used, claim_count)`` tuple where ``claim_count`` is
+        the number of STC statuses in the file (one per claim).
+
+    Raises:
+        ValueError: on parser-level failure (wraps CycloneParseError).
+    """
+    try:
+        result = parse_277ca_text(text, input_file=source_file)
+    except CycloneParseError as exc:
+        raise ValueError(f"277CA parse error: {exc}") from exc
+
+    icn = result.envelope.control_number
+    synthetic_id = two77ca_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"
+    )
+
+    with db.SessionLocal()() as session:
+        row = cycl_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()),
+        )
+        def _lookup(pcn: str):
+            return (
+                session.query(db.Claim)
+                .filter(db.Claim.patient_control_number == pcn)
+                .first()
+            )
+        apply_result = apply_277ca_rejections(
+            session, result, claim_lookup=_lookup, two77ca_id=row.id,
+        )
+        if apply_result.matched:
+            for cid in apply_result.matched:
+                append_event(session, 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-scheduler",
+                ))
+        session.commit()
+
+    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller
+    # (same gap as handle_999 + handle_ta1; Task 6 will fix when the
+    # FastAPI endpoints migrate to call these handlers).
+    if event_bus is not None:
+        publish = getattr(event_bus, "publish", None)
+        if callable(publish):
+            try:
+                publish("ack_received", {
+                    "source_batch_id": synthetic_id,
+                    "ack_code": "277CA",
+                    "kind": "277CA",
+                })
+            except Exception as exc:  # noqa: BLE001
+                log.warning("event_bus publish failed: %s", exc)
+
+    return ("parse_277ca", len(result.claim_statuses))

backend/src/cyclone/scheduler.py

@@ -55,11 +55,9 @@ from cyclone.audit_log import AuditEvent, append_event
 from cyclone.clearhouse import InboundFile, SftpClient
 from cyclone.db import ProcessedInboundFile
 from cyclone.edi.filenames import parse_inbound_filename
+from cyclone.handlers import handle_277ca as _handle_277ca
 from cyclone.handlers import handle_999 as _handle_999
 from cyclone.handlers import handle_ta1 as _handle_ta1
-from cyclone.handlers._ack_id import (
-    two77ca_synthetic_source_batch_id as _277ca_synthetic_source_batch_id,
-)
 from cyclone.inbox_state_277ca import apply_277ca_rejections
 from cyclone.providers import SftpBlock
 
@@ -146,10 +144,10 @@ class SchedulerStatus:
 # persists its own DB rows. The scheduler records the outcome.
 # ---------------------------------------------------------------------------
 #
-# 999 (SP27 Task 2) and TA1 (Task 3) now live in ``cyclone.handlers``;
+# 999 (Task 2), TA1 (Task 3), and 277CA (Task 4) now live in
-# the 277CA and 835 handlers stay inline and are lifted in Tasks 4-5.
+# ``cyclone.handlers``; only the 835 handler stays inline and is
-# The HANDLERS dict literal below references the alias imports so the
+# lifted in Task 5. The HANDLERS dict literal below references the
-# wiring stays identical.
+# alias imports so the wiring stays identical.
 
 
 def _handle_835(text: str, source_file: str) -> tuple[str, int]:
@@ -196,61 +194,11 @@ def _handle_835(text: str, source_file: str) -> tuple[str, int]:
     return "parse_835", len(result.claims)
 
 
-def _handle_277ca(text: str, source_file: str) -> tuple[str, int]:
+# The 277CA handler has moved to ``cyclone.handlers.handle_277ca``
-    """Parse a 277CA, persist ack + stamp payer-rejected claims."""
+# (SP27 Task 4); the inline def was deleted. The HANDLERS dict literal
-    from cyclone.parsers.parse_277ca import parse_277ca_text
+# below still references ``_handle_277ca`` because the import-alias
-    from cyclone.parsers.exceptions import CycloneParseError
+# line at the top of this module binds that name to the new function.
-
+# Only the 835 handler stays inline (Task 5).
-    try:
-        result = parse_277ca_text(text, input_file=source_file)
-    except CycloneParseError as exc:
-        raise ValueError(f"277CA parse error: {exc}") from 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"
-    )
-
-    with db.SessionLocal()() as session:
-        row = cycl_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()),
-        )
-        def _lookup(pcn: str):
-            return (
-                session.query(db.Claim)
-                .filter(db.Claim.patient_control_number == pcn)
-                .first()
-            )
-        apply_result = apply_277ca_rejections(
-            session, result, claim_lookup=_lookup, two77ca_id=row.id,
-        )
-        if apply_result.matched:
-            for cid in apply_result.matched:
-                append_event(session, 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-scheduler",
-                ))
-        session.commit()
-    return "parse_277ca", len(result.claim_statuses)
 
 
 # Map file_type → handler. Mirrors ROUTED_FILE_TYPES.
@@ -265,23 +213,15 @@ HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = {
 
 # ---------------------------------------------------------------------------
 # Light copies of helpers the API endpoints use, so the scheduler can
-# run without depending on the FastAPI module.
+# Run without depending on the FastAPI module.
-# ---------------------------------------------------------------------------
+#
-
+# Note: ``_277ca_synthetic_source_batch_id`` lived here as a
-
+# scheduler-local duplicate of
-def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
+# ``cyclone.handlers._ack_id.two77ca_synthetic_source_batch_id`` until
-    """Synthetic batches.id for a received 277CA with no source batch.
+# SP27 Task 4 landed; the inline def has been deleted because
-
+# ``handle_277ca`` now imports the canonical copy. The historical
-    Kept here as a scheduler-local copy because ``_handle_277ca`` is
+# helper was the only remaining inline def in this section — the
-    still inline (SP27 Task 4 will move it next to ``handle_999`` in
+# surviving inline handler is ``_handle_835`` (lifts in Task 5).
-    ``cyclone/handlers/``). For the 999 side, ``_handle_999`` is
-    already imported from ``cyclone.handlers.handle_999`` (Task 2).
-
-    TODO(sp27-task-4): delete this duplicate when ``_handle_277ca``
-    moves. The canonical copy now lives in
-    ``cyclone.handlers._ack_id.two77ca_synthetic_source_batch_id``.
-    """
-    return f"277CA-{(interchange_control_number or '').strip() or '000000001'}"
 
 
 # ---------------------------------------------------------------------------

backend/tests/test_handlers_277ca.py

@@ -0,0 +1,141 @@
+"""Direct tests for the ``handle_277ca`` handler (SP27 Task 4).
+
+Locks the handler's contract independent of the scheduler lifecycle
+so a regression in the scheduler wiring doesn't hide a regression
+in the handler.
+
+Note: ``claim.rejected_after_remit`` audit emission (when a 277CA
+rejects a claim already matched to a remit) is deferred to SP27
+Task 13. For now we lock the existing claim.payer_rejected behavior
+only.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from cyclone import db
+from cyclone.audit_log import AuditEvent
+from cyclone.handlers.handle_277ca import handle
+from cyclone.parsers.exceptions import CycloneParseError
+
+ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_277ca.txt"
+REJECTED = Path(__file__).parent / "fixtures" / "minimal_277ca_rejected_only.txt"
+
+
+def _seed_claim(claim_id: str, pcn: str) -> None:
+    with db.SessionLocal()() as s:
+        from cyclone.db import Claim
+        s.add(Claim(
+            id=claim_id, batch_id="BATCH-1",
+            patient_control_number=pcn, charge_amount=100.00,
+        ))
+        s.commit()
+
+
+def test_handle_277ca_persists_ack_with_classification_counts():
+    """Happy path: persist a 277CA ack row with the right counts.
+
+    The minimal_277ca.txt fixture has 3 claim statuses:
+    - CLAIM001 → A3:19 → accepted
+    - CLAIM002 → A6:19 → rejected
+    - CLAIM003 → A8:19 → pended
+
+    So accepted=1, rejected=1, pended=1, paid=0.
+    """
+    text = ACCEPTED.read_text()
+    parser_used, claim_count = handle(text, source_file=ACCEPTED.name)
+    assert parser_used == "parse_277ca"
+    assert claim_count == 3
+
+    # Lock persistence: a row was added to two77ca_acks.
+    with db.SessionLocal()() as session:
+        from cyclone.db import Two77caAck
+        rows = (
+            session.query(Two77caAck)
+            .filter_by(source_batch_id="277CA-000000123")
+            .all()
+        )
+        assert len(rows) == 1
+        row = rows[0]
+        assert row.control_number == "000000123"
+        assert row.accepted_count == 1
+        assert row.rejected_count == 1
+        assert row.pended_count == 1
+        assert row.paid_count == 0
+
+
+def test_handle_277ca_accepted_only_no_inbox_change():
+    """A 277CA with only accepted statuses (no rejections) leaves the
+    inbox unchanged — no claim.payer_rejected audit row, no
+    payer_rejected_at stamp on any claim.
+    """
+    text = ACCEPTED.read_text()
+    # Seed all 3 claims so the 277CA can match them by PCN.
+    _seed_claim("c1", "CLAIM001")
+    _seed_claim("c2", "CLAIM002")
+    _seed_claim("c3", "CLAIM003")
+
+    handle(text, source_file=ACCEPTED.name)
+
+    with db.SessionLocal()() as session:
+        from cyclone.db import Claim
+        # CLAIM001 (accepted A3) — no payer_rejected_at
+        c1 = session.get(Claim, "c1")
+        assert c1.payer_rejected_at is None
+        # CLAIM002 (rejected A6) — stamped
+        c2 = session.get(Claim, "c2")
+        assert c2.payer_rejected_at is not None
+        assert c2.payer_rejected_status_code == "A6"
+        # CLAIM003 (pended A8) — no payer_rejected_at (A8 is pended, not rejected)
+        c3 = session.get(Claim, "c3")
+        assert c3.payer_rejected_at is None
+        # Activity-event audit row exists for the rejected one (c2)
+        from sqlalchemy import select
+        events = session.execute(
+            select(db.AuditLog.__table__.c.event_type).where(
+                db.AuditLog.__table__.c.entity_id == "c2"
+            )
+        ).all()
+        event_types = [e[0] for e in events]
+        assert "claim.payer_rejected" in event_types
+
+
+def test_handle_277ca_rejects_only_one_claim():
+    """The rejected_only fixture has one A7:19 for CLAIM099.
+    Seed a matching claim and verify only it gets stamped.
+    """
+    _seed_claim("c_rejected", "CLAIM099")
+    _seed_claim("c_unrelated", "OTHERPCN")
+
+    text = REJECTED.read_text()
+    parser_used, claim_count = handle(text, source_file=REJECTED.name)
+    assert parser_used == "parse_277ca"
+    assert claim_count == 1  # one claim_status in the fixture
+
+    with db.SessionLocal()() as session:
+        from cyclone.db import Claim, Two77caAck
+        # The matching claim is stamped as rejected.
+        c_rej = session.get(Claim, "c_rejected")
+        assert c_rej.payer_rejected_at is not None
+        assert c_rej.payer_rejected_status_code == "A7"
+        # The unrelated claim is untouched.
+        c_unr = session.get(Claim, "c_unrelated")
+        assert c_unr.payer_rejected_at is None
+        # The two77ca_ack row was persisted with the rejected count.
+        rows = (
+            session.query(Two77caAck)
+            .filter_by(source_batch_id="277CA-000000789")
+            .all()
+        )
+        assert len(rows) == 1
+        assert rows[0].rejected_count == 1
+        assert rows[0].accepted_count == 0
+
+
+def test_handle_277ca_raises_value_error_on_bad_x12():
+    """Garbage that tokenize may accept but parse_277ca will reject."""
+    bad = "ISA*00*bad~ST*277CA*0001~SE*2*0001~IEA*0*0~"
+    with pytest.raises((CycloneParseError, ValueError)):
+        handle(bad, source_file="bad.277ca")