diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py index f7163ec..0c770c7 100644 --- a/backend/src/cyclone/api.py +++ b/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-``) 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 diff --git a/backend/src/cyclone/handlers/handle_277ca.py b/backend/src/cyclone/handlers/handle_277ca.py index 34b6666..22d673d 100644 --- a/backend/src/cyclone/handlers/handle_277ca.py +++ b/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)) diff --git a/backend/src/cyclone/handlers/handle_835.py b/backend/src/cyclone/handlers/handle_835.py index ec32808..75db89f 100644 --- a/backend/src/cyclone/handlers/handle_835.py +++ b/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)) diff --git a/backend/src/cyclone/handlers/handle_999.py b/backend/src/cyclone/handlers/handle_999.py index 9bb807c..7a98201 100644 --- a/backend/src/cyclone/handlers/handle_999.py +++ b/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) diff --git a/backend/src/cyclone/handlers/handle_ta1.py b/backend/src/cyclone/handlers/handle_ta1.py index cbadcd4..294085a 100644 --- a/backend/src/cyclone/handlers/handle_ta1.py +++ b/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) diff --git a/backend/tests/test_handlers_999.py b/backend/tests/test_handlers_999.py index 14a42a4..68f7446 100644 --- a/backend/tests/test_handlers_999.py +++ b/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(): - """Scheduler passes event_bus=None; api.py passes app.state.event_bus. - Both must work without crashing on publish.""" +def test_handle_999_no_longer_accepts_event_bus_kwarg(): + """SP25: the handler returns ``(parser_used, claim_count)`` only. + + 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. - class _Stub: - def __init__(self) -> None: - self.calls: list[tuple[str, dict]] = [] + assert claim_count >= 1 - def publish(self, kind: str, payload: dict) -> None: - self.calls.append((kind, payload)) + # The handler MUST reject the legacy ``event_bus=`` kwarg so any + # 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( - 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. + +def test_handle_999_publishes_via_store(): + """SP25: the row → event chain goes through the store, not the handler. + + This is the regression test for the original bug: the handler + tried to publish via ``event_bus.publish(...)`` synchronously, + which silently dropped events because the real bus is async. + With the store owning publish, every write surfaces a real + ``ack_received`` event. + """ + 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