feat(store): add publishes claim_written + activity_recorded events

backend/src/cyclone/store.py

@@ -809,7 +809,12 @@ class CycloneStore:
 
     # -- write path -----------------------------------------------------
 
-    def add(self, record: BatchRecord) -> None:
+    def add(
+        self,
+        record: BatchRecord,
+        *,
+        event_bus: "EventBus | None" = None,
+    ) -> None:
         """Persist a parsed batch (837P or 835) to the DB.
 
         For 837P batches: inserts the Batch row, one Claim row per
@@ -830,10 +835,22 @@ class CycloneStore:
         has a fresh ``uuid4`` id from the API). O(n) per row, but
         acceptable for the small fixture sizes — production load is
         one batch at a time via the API, not bulk inserts.
+
+        When ``event_bus`` is provided, publishes one ``claim_written``
+        or ``remittance_written`` event per newly-inserted row plus an
+        ``activity_recorded`` event per activity row, after commit. The
+        publish calls are best-effort — failures are logged but do not
+        roll back the persisted batch.
         """
+        from cyclone.pubsub import EventBus
+
         import logging
         log = logging.getLogger(__name__)
 
+        # Track rows we actually inserted so we can publish events for them.
+        inserted_claim_ids: list[str] = []
+        inserted_remit_ids: list[str] = []
+
         with db.SessionLocal()() as s:
             batch_row = Batch(
                 id=record.id,
@@ -870,6 +887,7 @@ class CycloneStore:
                             "amount": float(claim.claim.total_charge or 0.0),
                         },
                     ))
+                    inserted_claim_ids.append(claim.claim_id)
             elif isinstance(record, BatchRecord835):
                 result835: ParseResult835 = record.result
                 payer_name = result835.payer.name
@@ -903,6 +921,7 @@ class CycloneStore:
                             "amount": float(cp.total_paid or 0.0),
                         },
                     ))
+                    inserted_remit_ids.append(cp.payer_claim_control_number)
             else:
                 raise TypeError(
                     f"Unsupported BatchRecord subclass: {type(record).__name__}"
@@ -922,6 +941,83 @@ class CycloneStore:
                     "reconcile.run failed for batch %s", record.id,
                 )
 
+        # Publish live-tail events synchronously. EventBus.publish is async
+        # but its body is purely synchronous ``put_nowait`` enqueues; we
+        # bypass the async wrapper and call the internal enqueue directly
+        # so callers (sync FastAPI endpoints, sync test harnesses) don't
+        # need to await.
+        if event_bus is not None and (inserted_claim_ids or inserted_remit_ids):
+            self._publish_events_sync(
+                event_bus, record, inserted_claim_ids, inserted_remit_ids,
+            )
+
+    def _publish_events_sync(
+        self,
+        event_bus: "EventBus",
+        record: BatchRecord,
+        claim_ids: list[str],
+        remit_ids: list[str],
+    ) -> None:
+        """Build UI-shaped payloads for newly-inserted rows and publish.
+
+        Runs after commit so subscribers can immediately re-fetch from
+        the API and see consistent data. Each ``claim_written`` /
+        ``remittance_written`` payload is identical to what the matching
+        list endpoint would return for that row.
+
+        This is sync because EventBus's enqueue path is sync; we don't
+        need a coroutine for ``put_nowait``.
+        """
+        import logging
+        log = logging.getLogger(__name__)
+
+        try:
+            with db.SessionLocal()() as s:
+                for cid in claim_ids:
+                    row = s.get(Claim, cid)
+                    if row is None:
+                        continue
+                    ui = to_ui_claim_from_orm(
+                        row, batch_id=row.batch_id or record.id,
+                        parsed_at=record.parsed_at,
+                    )
+                    self._sync_publish(event_bus, "claim_written", ui)
+                for rid in remit_ids:
+                    row = s.get(Remittance, rid)
+                    if row is None:
+                        continue
+                    ui = to_ui_remittance_from_orm(
+                        row, batch_id=row.batch_id or record.id,
+                        parsed_at=record.parsed_at,
+                    )
+                    self._sync_publish(event_bus, "remittance_written", ui)
+                # Activity events for this batch.
+                from sqlalchemy import select
+                activity_rows = s.execute(
+                    select(ActivityEvent).where(ActivityEvent.batch_id == record.id)
+                ).scalars().all()
+                for arow in activity_rows:
+                    ui = {
+                        "kind": arow.kind,
+                        "ts": arow.ts.isoformat().replace("+00:00", "Z"),
+                        "batchId": arow.batch_id,
+                        "claimId": arow.claim_id,
+                        "remittanceId": arow.remittance_id,
+                        "payload": arow.payload_json,
+                    }
+                    self._sync_publish(event_bus, "activity_recorded", ui)
+        except Exception:
+            log.exception("add: event publish failed for batch %s", record.id)
+
+    @staticmethod
+    def _sync_publish(event_bus: "EventBus", kind: str, payload: dict) -> None:
+        """Synchronous fan-out helper. Mirrors ``EventBus.publish`` but
+        bypasses the async wrapper so callers don't need an event loop.
+        """
+        event = {**payload, "_kind": kind}
+        for queue in list(event_bus._subscribers.get(kind, ())):
+            event_bus._enqueue_or_drop_oldest(queue, event)
+
     def _run_reconcile(self, batch_id: str) -> None:
         """T10 stub: invoke the reconcile orchestrator for a batch.
 

backend/tests/test_store.py

@@ -90,6 +90,39 @@ def test_module_singleton():
     assert isinstance(store, CycloneStore)
 
 
+@pytest.mark.asyncio
+async def test_add_837_publishes_claim_written_and_activity_recorded_events():
+    """store.add() with event_bus= should publish one claim_written event
+    per inserted claim and one activity_recorded event per activity row."""
+    from cyclone.pubsub import EventBus
+
+    bus = EventBus()
+    claim_q = bus.subscribe(["claim_written"])
+    activity_q = bus.subscribe(["activity_recorded"])
+
+    s = CycloneStore()
+    rec = _make_batch_record()
+    s.add(rec, event_bus=bus)
+
+    # Two claims + two claim_submitted activity events expected.
+    import asyncio
+    claims = []
+    for _ in range(2):
+        claims.append(await asyncio.wait_for(claim_q.__anext__(), timeout=0.5))
+    activities = []
+    for _ in range(2):
+        activities.append(await asyncio.wait_for(activity_q.__anext__(), timeout=0.5))
+
+    # Each event has _kind and a payload with the expected shape.
+    for ev in claims:
+        assert ev["_kind"] == "claim_written"
+        assert "id" in ev
+    for ev in activities:
+        assert ev["_kind"] == "activity_recorded"
+        assert ev["kind"] == "claim_submitted"
+        assert "claimId" in ev
+
+
 def test_add_837_persists_batch_and_claims():
     s = CycloneStore()
     rec = _make_batch_record()