feat(sp25): drop event_bus kwarg from handlers; thread event_bus into parse endpoints

backend/src/cyclone/api.py

@@ -775,6 +775,7 @@ async def parse_999_endpoint(
         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
@@ -824,6 +825,7 @@ def _ta1_synthetic_source_batch_id(interchange_control_number: str) -> str:
 
 @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.
@@ -838,6 +840,10 @@ async def parse_ta1_endpoint(
     The persisted ``ta1_acks.source_batch_id`` is a synthetic id
     (``TA1-<ISA13>``) 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:
@@ -881,6 +887,7 @@ async def parse_ta1_endpoint(
         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,
     )
 
     return JSONResponse(content={
@@ -965,6 +972,7 @@ async def parse_277ca_endpoint(
     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,
@@ -973,6 +981,7 @@ async def parse_277ca_endpoint(
         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

backend/src/cyclone/handlers/handle_277ca.py

@@ -19,7 +19,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.audit_log import AuditEvent, append_event
@@ -35,17 +34,12 @@ 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
@@ -53,6 +47,10 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_277ca_ack``) owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg.
     """
     try:
         result = parse_277ca_text(text, input_file=source_file)
@@ -104,19 +102,4 @@ def handle(
                 ))
         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/handlers/handle_835.py

@@ -26,11 +26,9 @@ sync/async gap as handle_999 / handle_ta1 / handle_277ca.
 """
 from __future__ import annotations
 
-import json
 import logging
 import uuid
 from datetime import datetime, timezone
-from typing import Optional
 
 from cyclone.parsers.exceptions import CycloneParseError
 from cyclone.parsers.parse_835 import parse as parse_835
@@ -44,17 +42,12 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse an 835, run validation, persist batch + remittances.
 
     Args:
         text: Raw 835 document bytes (decoded).
         source_file: Filename the 835 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. ``claim_count`` is the
@@ -66,6 +59,12 @@ def handle(
         ValueError: on parser-level failure (wraps CycloneParseError).
             Validation failures don't raise — they're stamped into
             ``summary.passed = 0``.
+
+    SP25: the 835 write path already publishes ``remittance_written``
+    via ``CycloneStore.add`` (SP21 split). The handler no longer
+    accepts an ``event_bus`` kwarg — the ``ack_received`` publish
+    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
@@ -105,17 +104,4 @@ def handle(
     )
     cycl_store.add(rec)
 
-    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller.
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            try:
-                publish("ack_received", {
-                    "source_batch_id": rec.id,
-                    "ack_code": "835",
-                    "kind": "835",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_835", len(result.claims))

backend/src/cyclone/handlers/handle_999.py

@@ -20,7 +20,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.audit_log import AuditEvent, append_event
@@ -39,8 +38,6 @@ 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.
 
@@ -48,10 +45,6 @@ def handle(
         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
@@ -60,6 +53,12 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_999_ack``) now owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg. Both the scheduler path (no bus) and the
+    FastAPI endpoint path (passes the bus to the store directly) see
+    the same row, the same event, and the same payload.
     """
     try:
         result = parse_999_text(text, input_file=source_file)
@@ -105,25 +104,4 @@ def handle(
         )
         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/handlers/handle_ta1.py

@@ -16,7 +16,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.parsers.exceptions import CycloneParseError
@@ -29,8 +28,6 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse a TA1, persist the interchange ack row.
 
@@ -40,9 +37,6 @@ def handle(
             attribution; the ``batches.id`` for TA1 rows is derived
             internally from the parsed envelope's control number
             (``TA1-{ICN}``).
-        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. TA1 always returns
@@ -50,6 +44,10 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_ta1_ack``) owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg.
     """
     try:
         result = parse_ta1_text(text, input_file=source_file)
@@ -71,19 +69,4 @@ def handle(
         )
         session.commit()
 
-    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller
-    # (same gap as handle_999; both fixed when the FastAPI endpoints
-    # migrate to call these handlers instead of inline code).
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            try:
-                publish("ack_received", {
-                    "source_batch_id": result.source_batch_id,
-                    "ack_code": result.ta1.ack_code,
-                    "kind": "TA1",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_ta1", 1)

backend/tests/test_handlers_999.py

@@ -96,28 +96,65 @@ def test_handle_999_distinct_filenames_get_distinct_synthetic_ids():
         assert len(rows_b) == 1
 
 
-def test_handle_999_event_bus_parameter_is_optional():
+def test_handle_999_no_longer_accepts_event_bus_kwarg():
-    """Scheduler passes event_bus=None; api.py passes app.state.event_bus.
+    """SP25: the handler returns ``(parser_used, claim_count)`` only.
-    Both must work without crashing on publish."""
+
+    Pre-SP25 the handler accepted a keyword-only ``event_bus=`` that
+    it tried to publish to directly — a sync caller invoking the
+    real async ``EventBus.publish`` produced an unawaited coroutine
+    that was silently swallowed by the bare ``except``. The store
+    now owns publish-from-store; the handler's surface is reduced
+    to (text, source_file) so the sync/async gap is closed.
+    """
     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.
+    assert claim_count >= 1
-    class _Stub:
-        def __init__(self) -> None:
-            self.calls: list[tuple[str, dict]] = []
 
-        def publish(self, kind: str, payload: dict) -> None:
+    # The handler MUST reject the legacy ``event_bus=`` kwarg so any
-            self.calls.append((kind, payload))
+    # stale caller (e.g. the inline copy in api.py that Task 6 was
+    # supposed to migrate) fails loudly during refactors rather than
+    # silently dropping events on the floor.
+    with pytest.raises(TypeError, match="event_bus"):
+        handle(text, source_file=ACCEPTED.name, event_bus=None)
 
-    spy = _Stub()
+
-    parser_used, claim_count = handle(
+def test_handle_999_publishes_via_store():
-        text, source_file=ACCEPTED.name, event_bus=spy,
+    """SP25: the row → event chain goes through the store, not the handler.
-    )
+
-    assert parser_used == "parse_999"
+    This is the regression test for the original bug: the handler
-    # The stub pattern is sync. Real EventBus is async and the handler
+    tried to publish via ``event_bus.publish(...)`` synchronously,
-    # is sync — Task 6 bridges that. For now, just verify the sync
+    which silently dropped events because the real bus is async.
-    # stub fires (or doesn't) without crashing.
+    With the store owning publish, every write surfaces a real
-    # We don't assert the call count here because the async-vs-sync
+    ``ack_received`` event.
-    # gap is owned by Task 6 — see TODO(sp27-task-6) in handle_999.
+    """
+    from cyclone.pubsub import EventBus
+    from cyclone.store import store
+
+    bus = EventBus(max_queue_size=64)
+    bus.subscribe_raw(["ack_received"])
+
+    text = ACCEPTED.read_text()
+    # The handler itself takes NO bus — the operator (scheduler or
+    # endpoint) threads the bus to the store. We can't plumb a bus
+    # through handle(), but we can verify that *any* path that uses
+    # the store's add_ack fires the event. This is enough to lock
+    # the regression: pre-SP25 the handler would have published via
+    # its own path and the store's publish would be a no-op. Post-SP25
+    # only the store path exists.
+    with db.SessionLocal()() as _s:
+        row = store.add_ack(
+            source_batch_id="999-HANDLE-PATH",
+            accepted_count=1,
+            rejected_count=0,
+            received_count=1,
+            ack_code="A",
+            raw_json={"envelope": {"control_number": "000000001"}},
+            event_bus=bus,
+        )
+
+    queue = bus._subscribers["ack_received"][0]
+    assert not queue.empty()
+    event = queue.get_nowait()
+    assert event["_kind"] == "ack_received"
+    assert event["id"] == row.id