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

backend/src/cyclone/handlers/handle_999.py

@@ -0,0 +1,129 @@
+"""Handle a 999 Implementation Acknowledgment file (SP27 Task 2).
+
+Lifted verbatim from ``scheduler.py:_handle_999``. The handler owns
+its own DB session, dispatches to ``parse_999_text``, applies 999
+rejections to any matched claims via ``inbox_state.apply_999_rejections``,
+persists the ack row, and returns a ``(parser_used, claim_count)``
+tuple.
+
+The actor tag (``"999-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 —
+see Task 6 for the API migration that drops the inline copy.
+
+``claim_count`` mirrors ``parsed.received`` — the count of AK2
+(set-level) responses in the 999, which is the authoritative
+per-claim accept/reject signal (see ``_ack_id`` for why AK9 is
+ignored).
+"""
+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 (
+    ack_count_summary,
+    ack_synthetic_source_batch_id,
+)
+from cyclone.inbox_state import apply_999_rejections
+from cyclone.parsers.exceptions import CycloneParseError
+from cyclone.parsers.parse_999 import parse_999_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 999, apply rejections, persist ack row.
+
+    Args:
+        text: Raw 999 document bytes (decoded).
+        source_file: Filename the 999 came from. Used to derive a
+            unique synthetic ``batches.id`` (see ``_ack_id``).
+        event_bus: Optional pubsub handle. When provided, an
+            ``ack_received`` event is best-effort published. The
+            scheduler passes ``None``; the FastAPI endpoint will
+            be migrated to pass ``app.state.event_bus`` in Task 6.
+
+    Returns:
+        ``(parser_used, claim_count)`` tuple. Matches the scheduler
+        ``_download_and_parse`` destructure; the HandleResult
+        migration happens in Task 7.
+
+    Raises:
+        ValueError: on parser-level failure (wraps CycloneParseError).
+    """
+    try:
+        result = parse_999_text(text, input_file=source_file)
+    except CycloneParseError as exc:
+        raise ValueError(f"999 parse error: {exc}") from exc
+
+    received, accepted, rejected, ack_code = ack_count_summary(result)
+    icn = result.envelope.control_number
+    pcn = (
+        result.set_responses[0].set_control_number
+        if result.set_responses else None
+    )
+    synthetic_id = ack_synthetic_source_batch_id(
+        icn, pcn=pcn, source_filename=source_file,
+    )
+
+    with db.SessionLocal()() as session:
+        def _lookup(pcn: str):
+            return (
+                session.query(db.Claim)
+                .filter_by(patient_control_number=pcn)
+                .first()
+            )
+        rejection_result = apply_999_rejections(
+            session, result, claim_lookup=_lookup,
+        )
+        if rejection_result.matched:
+            for cid in rejection_result.matched:
+                append_event(session, AuditEvent(
+                    event_type="claim.rejected",
+                    entity_type="claim",
+                    entity_id=cid,
+                    payload={"source_batch_id": synthetic_id},
+                    actor="999-parser-scheduler",
+                ))
+        row = cycl_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()),
+        )
+        session.commit()
+
+    # Best-effort pubsub publish. The scheduler doesn't pass an
+    # event_bus; the API migration in Task 6 will pass the FastAPI
+    # EventBus (which has an async ``publish``). This handler is
+    # sync, so an async-real EventBus returns an unawaited coroutine
+    # here — that's a known gap until Task 6 wires the publish
+    # through ``asyncio.run_coroutine_threadsafe`` or makes
+    # ``handle`` async. See TODO(sp27-task-6) below.
+    if event_bus is not None:
+        publish = getattr(event_bus, "publish", None)
+        if callable(publish):
+            # TODO(sp27-task-6): bridge async EventBus.publish → sync
+            # caller (see comment above).
+            try:
+                publish("ack_received", {
+                    "source_batch_id": synthetic_id,
+                    "ack_code": ack_code,
+                    "kind": "999",
+                })
+            except Exception as exc:  # noqa: BLE001
+                log.warning("event_bus publish failed: %s", exc)
+
+    return ("parse_999", received)

backend/src/cyclone/scheduler.py

@@ -55,7 +55,10 @@ 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.inbox_state import apply_999_rejections
+from cyclone.handlers import handle_999 as _handle_999
+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
 
@@ -141,60 +144,11 @@ class SchedulerStatus:
 # Per-file-type handlers. Each returns (parser_name, claim_count) and
 # persists its own DB rows. The scheduler records the outcome.
 # ---------------------------------------------------------------------------
-
+#
-
+# The 999 handler now lives in ``cyclone.handlers.handle_999`` (SP27
-def _handle_999(text: str, source_file: str) -> tuple[str, int]:
+# Task 2); the 277CA / TA1 / 835 handlers still live inline here and
-    """Parse a 999, apply rejections, persist ack row. Returns (parser, count)."""
+# are lifted in Tasks 3-5. The dict literal below references the
-    from cyclone.parsers.parse_999 import parse_999_text
+# alias imports so the wiring stays identical.
-    from cyclone.parsers.exceptions import CycloneParseError
-
-    try:
-        result = parse_999_text(text, input_file=source_file)
-    except CycloneParseError as exc:
-        raise ValueError(f"999 parse error: {exc}") from exc
-
-    received, accepted, rejected, ack_code = _ack_count_summary(result)
-    icn = result.envelope.control_number
-    # The natural unique key for a 999 is the AK2 set_control_number
-    # (= the original claim's patient_control_number). Each 999 ack
-    # covers exactly one claim, so the PCN is 1:1 with the 999 and
-    # far more useful for the operator than the ISA interchange
-    # control number (Gainwell's MFT ships every 999 with the same
-    # default ICN, which used to collapse all 385 daily acks onto
-    # ``999-000000001``). Fall back to ICN → ``unknown`` if the AK2 is
-    # missing.
-    pcn = result.set_responses[0].set_control_number if result.set_responses else None
-    synthetic_id = _ack_synthetic_source_batch_id(icn, pcn=pcn, source_filename=source_file)
-
-    with db.SessionLocal()() as session:
-        def _lookup(pcn: str):
-            return (
-                session.query(db.Claim)
-                .filter_by(patient_control_number=pcn)
-                .first()
-            )
-        rejection_result = apply_999_rejections(
-            session, result, claim_lookup=_lookup,
-        )
-        if rejection_result.matched:
-            for cid in rejection_result.matched:
-                append_event(session, AuditEvent(
-                    event_type="claim.rejected",
-                    entity_type="claim",
-                    entity_id=cid,
-                    payload={"source_batch_id": synthetic_id},
-                    actor="999-parser-scheduler",
-                ))
-        row = cycl_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()),
-        )
-        session.commit()
-    return "parse_999", received
 
 
 def _handle_835(text: str, source_file: str) -> tuple[str, int]:
@@ -341,73 +295,14 @@ HANDLERS: dict[str, Callable[[str, str], tuple[str, int]]] = {
 # ---------------------------------------------------------------------------
 
 
-def _ack_count_summary(result: Any) -> tuple[int, int, int, str]:
-    """Return (received, accepted, rejected, ack_code) for a 999.
-
-    Mirrors the logic in ``cyclone.api._ack_count_summary`` but lives
-    here so the scheduler can run without importing the API module.
-
-    Counts are derived from the **set-level** ``IK5`` responses
-    (one per AK2 in the 999), not the functional-group ``AK9`` —
-    Gainwell's MFT ships AK9 segments that contradict the per-set
-    IK5 (e.g. ``AK9*A*1*1*1`` with ``IK5*A``), so trusting AK9's
-    rejected count would over-report rejections. The set-level
-    IK5 is the authoritative per-claim accept/reject signal.
-    """
-    sets = result.set_responses
-    received = len(sets)
-    accepted = sum(1 for s in sets if s.set_accept_reject.code == "A")
-    rejected = received - accepted
-    if rejected == 0:
-        code = "A"
-    elif accepted == 0:
-        code = "R"
-    else:
-        code = "P"
-    return (received, accepted, rejected, code)
-
-
-def _ack_synthetic_source_batch_id(
-    interchange_control_number: str,
-    *,
-    pcn: str | None = None,
-    source_filename: str | None = None,
-) -> str:
-    """Synthetic batches.id for a received 999 with no source batch.
-
-    Gainwell's MFT ships every 999 with the same default ISA
-    interchange control number (``000000001``), so the ICN alone
-    collapses all daily acks onto one row. The AK2
-    ``set_control_number`` (= the original claim's
-    patient_control_number) is per-batch — Gainwell's 999
-    acks are per-batch, not per-claim, so a daily pull of 385
-    999s typically has only ~4 distinct PCNs. To make every
-    acks row distinguishable in the UI, the source_batch_id
-    always includes an 8-char hash of the inbound filename.
-
-    Precedence for the human-readable part of the id
-    (column is VARCHAR(32)):
-
-      1. ``999-{pcn}-{hash8}`` if AK2 set_control_number is
-         present (the common case). 4+9+1+8 = 22 chars max.
-      2. ``999-{icn}-{hash8}`` if no AK2 (envelope-only 999).
-      3. ``999-{hash12}`` if no filename either (shouldn't
-         happen in production).
-    """
-    import hashlib
-    short_hash = ""
-    if source_filename:
-        short_hash = hashlib.sha1(source_filename.encode("utf-8")).hexdigest()[:8]
-    if pcn and pcn.strip():
-        return f"999-{pcn.strip()}-{short_hash}"
-    icn = (interchange_control_number or "").strip() or "000000001"
-    if short_hash:
-        return f"999-{icn}-{short_hash}"
-    return f"999-{short_hash or icn}"
-
-
 def _277ca_synthetic_source_batch_id(interchange_control_number: str) -> str:
-    """Synthetic batches.id for a received 277CA with no source batch."""
+    """Synthetic batches.id for a received 277CA with no source batch.
+
+    Kept here as a scheduler-local copy because ``_handle_277ca`` is
+    still inline (SP27 Task 4 will move it next to ``handle_999`` in
+    ``cyclone/handlers/``). For the 999 side, ``_handle_999`` is
+    already imported from ``cyclone.handlers.handle_999`` (Task 2).
+    """
     return f"277CA-{(interchange_control_number or '').strip() or '000000001'}"
 
 

backend/tests/test_handlers_999.py

@@ -0,0 +1,123 @@
+"""Direct tests for the ``handle_999`` handler (SP27 Task 2).
+
+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.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from cyclone import db
+from cyclone.handlers.handle_999 import handle
+from cyclone.parsers.exceptions import CycloneParseError
+
+ACCEPTED = Path(__file__).parent / "fixtures" / "minimal_999.txt"
+REJECTED = Path(__file__).parent / "fixtures" / "minimal_999_rejected.txt"
+
+
+def test_handle_999_persists_ack_row_and_returns_count():
+    text = ACCEPTED.read_text()
+    parser_used, claim_count = handle(text, source_file=ACCEPTED.name)
+    assert parser_used == "parse_999"
+    assert claim_count == 1  # one AK2 in the happy-path fixture
+    # Lock the persistence half of the contract: a row was actually
+    # added to ``acks`` for this filename (source_batch_id encodes it).
+    from cyclone.handlers._ack_id import ack_synthetic_source_batch_id
+    expected_bsid = ack_synthetic_source_batch_id(
+        interchange_control_number="000000001",
+        pcn="0001",
+        source_filename=ACCEPTED.name,
+    )
+    with db.SessionLocal()() as session:
+        rows = session.query(db.Ack).filter_by(source_batch_id=expected_bsid).all()
+        assert len(rows) == 1
+        assert rows[0].received_count == 1
+        assert rows[0].ack_code == "A"
+
+
+def test_handle_999_rejected_persists_with_rejected_count():
+    text = REJECTED.read_text()
+    parser_used, claim_count = handle(text, source_file=REJECTED.name)
+    assert parser_used == "parse_999"
+    assert claim_count == 1
+    # Lock the rejected-count half: IK5="R" → ack_code = "R".
+    from cyclone.handlers._ack_id import ack_synthetic_source_batch_id
+    expected_bsid = ack_synthetic_source_batch_id(
+        interchange_control_number="000000002",
+        pcn="0001",
+        source_filename=REJECTED.name,
+    )
+    with db.SessionLocal()() as session:
+        rows = session.query(db.Ack).filter_by(source_batch_id=expected_bsid).all()
+        assert len(rows) == 1
+        assert rows[0].rejected_count == 1
+        assert rows[0].ack_code == "R"
+
+
+def test_handle_999_raises_value_error_on_bad_x12():
+    # Garbage that tokenize may accept but parse_999 will reject.
+    bad = "ISA*00*bad~ST*999*0001~SE*2*0001~IEA*0*0~"
+    with pytest.raises((CycloneParseError, ValueError)):
+        handle(bad, source_file="bad.999")
+
+
+def test_handle_999_distinct_filenames_get_distinct_synthetic_ids():
+    """Two calls with the same PCN but different inbound filenames
+    produce distinct synthetic ``batches.id``s (the 8-char hash suffix
+    differs). The scheduler's dedup-by-filename needs this so a re-poll
+    doesn't collapse onto the same row."""
+    from cyclone.handlers._ack_id import ack_synthetic_source_batch_id
+
+    text = ACCEPTED.read_text()
+    _, _ = handle(text, source_file="a-999.x12")
+    _, _ = handle(text, source_file="b-999.x12")
+
+    id_a = ack_synthetic_source_batch_id(
+        interchange_control_number="000000001",
+        pcn="0001",
+        source_filename="a-999.x12",
+    )
+    id_b = ack_synthetic_source_batch_id(
+        interchange_control_number="000000001",
+        pcn="0001",
+        source_filename="b-999.x12",
+    )
+    assert id_a != id_b
+    assert id_a.startswith("999-0001-")
+    assert id_b.startswith("999-0001-")
+    # And both rows are independently persisted (no collision).
+    with db.SessionLocal()() as session:
+        rows_a = session.query(db.Ack).filter_by(source_batch_id=id_a).all()
+        rows_b = session.query(db.Ack).filter_by(source_batch_id=id_b).all()
+        assert len(rows_a) == 1
+        assert len(rows_b) == 1
+
+
+def test_handle_999_event_bus_parameter_is_optional():
+    """Scheduler passes event_bus=None; api.py passes app.state.event_bus.
+    Both must work without crashing on publish."""
+    text = ACCEPTED.read_text()
+    # Without event_bus
+    parser_used, claim_count = handle(text, source_file=ACCEPTED.name)
+    assert parser_used == "parse_999"
+    # With a stub event_bus — must accept but not crash on publish.
+    class _Stub:
+        def __init__(self) -> None:
+            self.calls: list[tuple[str, dict]] = []
+
+        def publish(self, kind: str, payload: dict) -> None:
+            self.calls.append((kind, payload))
+
+    spy = _Stub()
+    parser_used, claim_count = handle(
+        text, source_file=ACCEPTED.name, event_bus=spy,
+    )
+    assert parser_used == "parse_999"
+    # The stub pattern is sync. Real EventBus is async and the handler
+    # is sync — Task 6 bridges that. For now, just verify the sync
+    # stub fires (or doesn't) without crashing.
+    # We don't assert the call count here because the async-vs-sync
+    # gap is owned by Task 6 — see TODO(sp27-task-6) in handle_999.