feat(api): GET /api/{claims,remittances,activity}/stream + unsubscribe

Adds three live-tail streaming endpoints that emit an NDJSON snapshot
then forward new event-bus events as they arrive, with a 15s idle
heartbeat (overridable via CYCLONE_TAIL_HEARTBEAT_S for tests).

Each endpoint:
  1. yields a snapshot of existing rows as {"type":"item","data":<row>}
  2. terminates the snapshot with {"type":"snapshot_end","data":{"count":N}}
  3. subscribes to its event kind and forwards each new event as an
     {"type":"item","data":<event>} line
  4. emits a {"type":"heartbeat","data":{"ts":<iso>}} line every
     CYCLONE_TAIL_HEARTBEAT_S seconds when idle
  5. checks request.is_disconnected() before each yield and unsubscribes
     from the bus on cleanup so a closed stream releases its queue

The shared tail loop lives in api._tail_events, which polls
bus.subscribe_raw()'s queue directly instead of using the bus's
async-iterator wrapper — wait_for on an async generator cancels the
inner future on timeout, which poisons subsequent __anext__ calls with
StopAsyncIteration. Queue.get() is idempotent under cancellation, so
heartbeats don't break the subscription.

EventBus gains an unsubscribe(queue, kinds) method (idempotent) so
the tail loop can release its queue in a try/finally. The disconnect
test asserts the subscriber list is empty after the body iterator is
closed, validating no queue leak per open stream.

Tests in test_api_stream_live.py: 8 tests covering snapshot shape,
post-snapshot publish, heartbeat timing, multi-item snapshots, and
client disconnect cleanup. Plus 2 tests in test_pubsub.py for the
new unsubscribe method.

backend/src/cyclone/api.py

@@ -15,8 +15,10 @@ plus GET/POST with any header.
 
 from __future__ import annotations
 
+import asyncio
 import json
 import logging
+import os
 import uuid
 from contextlib import asynccontextmanager
 from typing import Any, AsyncIterator
@@ -738,6 +740,123 @@ def list_claims(
     }
 
 
+# --------------------------------------------------------------------------- #
+# Live-tail NDJSON streaming endpoints (Phase 3 — SP5)
+# --------------------------------------------------------------------------- #
+
+
+def _heartbeat_seconds() -> float:
+    """Return the configured tail heartbeat interval.
+
+    Read from ``CYCLONE_TAIL_HEARTBEAT_S`` at call time so tests can
+    monkeypatch the env var without reloading the module. Defaults to
+    15s (the production cadence); tests override to a small value (e.g.
+    0.2s) to keep their runtime bounded.
+    """
+    raw = os.environ.get("CYCLONE_TAIL_HEARTBEAT_S", "15")
+    try:
+        v = float(raw)
+    except ValueError:
+        return 15.0
+    return v if v > 0 else 15.0
+
+
+async def _tail_events(
+    request: Request, bus: EventBus, kinds: list[str]
+) -> AsyncIterator[bytes]:
+    """Forward subscribed events as ``item`` lines with periodic heartbeats.
+
+    Polls the underlying ``asyncio.Queue`` directly (via
+    :meth:`EventBus.subscribe_raw`) instead of awaiting the bus's
+    async-iterator wrapper. ``asyncio.wait_for`` cancels the inner
+    future on timeout, which would otherwise terminate the bus
+    iterator at its ``await`` point and break subsequent
+    ``__anext__`` calls with ``StopAsyncIteration``. Polling
+    ``queue.get()`` is idempotent under cancellation, so heartbeats
+    don't poison the subscription.
+
+    A ``try/finally`` unsubscribes the queue from the bus when the
+    caller disconnects or the generator is garbage collected —
+    otherwise the bus would leak one queue per open stream.
+    """
+    hb_s = _heartbeat_seconds()
+    queue, _sub = bus.subscribe_raw(kinds)
+    try:
+        while True:
+            if await request.is_disconnected():
+                return
+            get_task = asyncio.ensure_future(queue.get())
+            sleep_task = asyncio.ensure_future(asyncio.sleep(hb_s))
+            try:
+                done, pending = await asyncio.wait(
+                    {get_task, sleep_task},
+                    return_when=asyncio.FIRST_COMPLETED,
+                )
+            except BaseException:
+                get_task.cancel()
+                sleep_task.cancel()
+                raise
+            for t in pending:
+                t.cancel()
+            if get_task in done:
+                event = get_task.result()
+                yield _ndjson_line({"type": "item", "data": event})
+            else:
+                yield _ndjson_line({
+                    "type": "heartbeat",
+                    "data": {"ts": utcnow().isoformat().replace("+00:00", "Z")},
+                })
+    finally:
+        bus.unsubscribe(queue, kinds)
+
+
+@app.get("/api/claims/stream")
+async def claims_stream(
+    request: Request,
+    status: str | None = Query(None),
+    provider_npi: str | None = Query(None),
+    payer: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> StreamingResponse:
+    """Stream Claims as NDJSON: snapshot first, then live events.
+
+    Wire format:
+      * ``{"type":"item","data":<claim>}`` per snapshot row, then per
+        new ``claim_written`` event
+      * ``{"type":"snapshot_end","data":{"count":N}}`` after the snapshot
+      * ``{"type":"heartbeat","data":{"ts":<iso>}}`` every
+        ``CYCLONE_TAIL_HEARTBEAT_S`` seconds when idle
+
+    Query params mirror :func:`list_claims` so a frontend can swap a
+    one-shot fetch for a tail with no URL surgery.
+
+    NOTE: registered before ``/api/claims/{claim_id}`` so the literal
+    ``stream`` path segment doesn't get matched as a claim id.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        # 1. Snapshot (eager — iter_claims returns a list already).
+        rows = store.iter_claims(
+            status=status, provider_npi=provider_npi, payer=payer,
+            date_from=date_from, date_to=date_to,
+            sort=sort or "-submission_date", order=order, limit=limit,
+        )
+        for row in rows:
+            yield _ndjson_line({"type": "item", "data": row})
+        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
+
+        # 2. Subscribe + heartbeats.
+        async for chunk in _tail_events(request, bus, ["claim_written"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
 @app.get("/api/claims/{claim_id}")
 def get_claim_detail_endpoint(claim_id: str) -> dict:
     """Return one claim with full drawer context (SP4).
@@ -891,6 +1010,45 @@ def list_remittances(
     }
 
 
+@app.get("/api/remittances/stream")
+async def remittances_stream(
+    request: Request,
+    payer: str | None = Query(None),
+    claim_id: str | None = Query(None),
+    date_from: str | None = Query(None),
+    date_to: str | None = Query(None),
+    sort: str | None = Query(None),
+    order: str = Query("desc"),
+    limit: int = Query(100, ge=1, le=1000),
+) -> StreamingResponse:
+    """Stream Remittances as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``remittance_written``. Default sort is
+    ``-received_date`` (newest-first), matching the list endpoint's
+    most common sort.
+
+    NOTE: registered before ``/api/remittances/{remittance_id}`` so
+    the literal ``stream`` path segment doesn't get matched as a
+    remittance id.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        rows = store.iter_remittances(
+            payer=payer, claim_id=claim_id,
+            date_from=date_from, date_to=date_to,
+            sort=sort or "-received_date", order=order, limit=limit,
+        )
+        for row in rows:
+            yield _ndjson_line({"type": "item", "data": row})
+        yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}})
+
+        async for chunk in _tail_events(request, bus, ["remittance_written"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
 @app.get("/api/remittances/{remittance_id}")
 def get_remittance(remittance_id: str) -> dict:
     """Return one remittance with its labeled CAS ``adjustments`` array.
@@ -965,6 +1123,43 @@ def list_activity(
     }
 
 
+@app.get("/api/activity/stream")
+async def activity_stream(
+    request: Request,
+    kind: str | None = Query(None),
+    since: str | None = Query(None),
+    limit: int = Query(50, ge=1, le=500),
+) -> StreamingResponse:
+    """Stream Activity events as NDJSON: snapshot first, then live events.
+
+    Subscribes to ``activity_recorded``. Default ``limit`` is 50
+    (smaller than the list endpoint's 200) because activity is
+    high-volume — callers usually want the most recent handful, not a
+    full replay.
+    """
+    bus: EventBus = request.app.state.event_bus
+
+    async def gen() -> AsyncIterator[bytes]:
+        # Snapshot reuses the same in-memory filter as ``list_activity``
+        # so the two endpoints are interchangeable for the snapshot
+        # half.
+        events = store.recent_activity(limit=limit)
+        if kind is not None:
+            events = [e for e in events if e["kind"] == kind]
+        if since is not None:
+            events = [e for e in events if e["timestamp"] >= since]
+        for ev in events:
+            yield _ndjson_line({"type": "item", "data": ev})
+        yield _ndjson_line({
+            "type": "snapshot_end", "data": {"count": len(events)},
+        })
+
+        async for chunk in _tail_events(request, bus, ["activity_recorded"]):
+            yield chunk
+
+    return StreamingResponse(gen(), media_type="application/x-ndjson")
+
+
 # --------------------------------------------------------------------------- #
 # 999 ACKs (read views)
 # --------------------------------------------------------------------------- #

backend/src/cyclone/pubsub.py

@@ -52,6 +52,49 @@ class EventBus:
             self._subscribers.setdefault(kind, []).append(queue)
         return self._iterator(queue)
 
+    def subscribe_raw(
+        self, kinds: list[str]
+    ) -> tuple[asyncio.Queue[dict], AsyncIterator[dict]]:
+        """Like :meth:`subscribe` but also returns the underlying queue.
+
+        Most consumers should use :meth:`subscribe`. The live-tail
+        endpoints use ``subscribe_raw`` so they can race
+        ``queue.get()`` against a heartbeat sleep without poisoning
+        the subscription iterator: ``asyncio.wait_for`` cancels the
+        inner future on timeout, which terminates an async generator
+        at its ``await`` point and breaks all subsequent
+        ``__anext__`` calls. Polling ``queue.get()`` directly avoids
+        that because ``Queue.get`` is idempotent under cancellation.
+
+        Callers MUST pair this with :meth:`unsubscribe` in a
+        ``try/finally`` to release the queue when the consumer
+        disconnects — otherwise the bus leaks a queue per stream.
+        """
+        queue: asyncio.Queue[dict] = asyncio.Queue(maxsize=self._max_queue_size)
+        for kind in kinds:
+            self._subscribers.setdefault(kind, []).append(queue)
+        return queue, self._iterator(queue)
+
+    def unsubscribe(
+        self, queue: asyncio.Queue[dict], kinds: list[str]
+    ) -> None:
+        """Remove ``queue`` from each kind's subscriber list.
+
+        Idempotent: missing entries are silently ignored so callers
+        can run the cleanup in a ``finally`` without worrying about
+        double-unsubscribe on disconnect-then-GeneratorExit.
+        """
+        for kind in kinds:
+            subs = self._subscribers.get(kind)
+            if not subs:
+                continue
+            try:
+                subs.remove(queue)
+            except ValueError:
+                pass
+            if not subs:
+                self._subscribers.pop(kind, None)
+
     async def _iterator(self, queue: asyncio.Queue[dict]) -> AsyncIterator[dict]:
         while True:
             yield await queue.get()

backend/tests/test_api_stream_live.py

@@ -0,0 +1,607 @@
+"""Live-tail NDJSON streaming endpoint tests (Phase 3).
+
+Each ``/api/.../stream`` endpoint:
+  1. yields the current snapshot as ``{"type":"item","data":<row>}`` lines
+  2. terminates the snapshot with ``{"type":"snapshot_end","data":{"count":N}}``
+  3. subscribes to the relevant EventBus kind and forwards new events
+  4. emits a ``{"type":"heartbeat","data":{"ts":<iso>}}`` line every
+     ``CYCLONE_TAIL_HEARTBEAT_S`` seconds (default 15) when idle
+  5. returns cleanly when the client disconnects
+  6. responds with ``application/x-ndjson``
+
+Implementation note (deviation from plan)
+-----------------------------------------
+The plan suggested using ``httpx.AsyncClient(transport=ASGITransport(app))``
+and ``client.stream()``. That doesn't work for true long-lived streams:
+
+* ``httpx.ASGITransport``'s response is buffered (``ASGIResponseStream``
+  joins every body chunk at the end), so ``client.stream()`` blocks until
+  the app finishes — a live tail never finishes.
+* ``starlette.requests.Request.is_disconnected()`` relies on a
+  ``http.disconnect`` message in the ASGI receive channel, which
+  ``ASGITransport`` only delivers *after* the response body completes
+  (the receive coroutine awaits ``response_complete``). So even if the
+  caller closes the client side, the endpoint never observes a
+  disconnect.
+
+Instead, we call the endpoint coroutine directly, build a synthetic
+``Request`` and ``StreamingResponse``, and iterate ``body_iterator``
+ourselves. That gives us byte-by-byte streaming, a real disconnect
+signal via ``body_iterator.aclose()`` (which throws ``CancelledError``
+into the generator), and works for arbitrarily long-lived tails.
+
+The public surface (snapshot + subscribe + heartbeat + disconnect
+cleanup) is verified exactly as the plan requires; only the transport
+differs.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import date, datetime, timezone
+from decimal import Decimal
+
+import pytest
+from starlette.requests import Request
+
+from cyclone.api import app, claims_stream, remittances_stream, activity_stream
+from cyclone.parsers.models import (
+    BatchSummary,
+    BillingProvider,
+    ClaimHeader,
+    ClaimOutput,
+    Envelope,
+    Payer,
+    ParseResult,
+    Subscriber,
+    ValidationReport,
+)
+from cyclone.parsers.models_835 import (
+    ClaimAdjustment,
+    ClaimPayment,
+    Envelope as Envelope835,
+    FinancialInfo,
+    ParseResult835,
+    Payer835,
+    Payee835,
+    ReassociationTrace,
+    ServicePayment,
+    BatchSummary as BatchSummary835,
+)
+from cyclone.pubsub import EventBus
+from cyclone.store import (
+    BatchRecord,
+    BatchRecord835,
+    CycloneStore,
+)
+
+
+@pytest.fixture(autouse=True)
+def _short_heartbeat(monkeypatch):
+    """Set a 0.2s heartbeat for all stream tests in this module.
+
+    The production default is 15s. Tests use a tight value so that
+    generators sitting idle after the test reads its lines exit
+    promptly — ``aclose()`` on the body iterator cancels the
+    generator and the heartbeat timeout is the de-facto idle
+    signal in these tests.
+    """
+    monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")
+
+
+# --------------------------------------------------------------------------- #
+# Test helpers
+# --------------------------------------------------------------------------- #
+
+
+def _make_claim_837(claim_id: str = "CLM-1", charge: str = "124.00") -> ClaimOutput:
+    return ClaimOutput(
+        claim_id=claim_id,
+        control_number="0001",
+        transaction_date=date(2026, 6, 19),
+        billing_provider=BillingProvider(name="Test", npi="1234567890"),
+        subscriber=Subscriber(
+            first_name="Jane", last_name="Doe", member_id=f"M-{claim_id}",
+        ),
+        payer=Payer(name="Test Payer", id="P1"),
+        claim=ClaimHeader(
+            claim_id=claim_id, total_charge=Decimal(charge),
+            frequency_code="1", place_of_service="11",
+        ),
+        diagnoses=[],
+        service_lines=[],
+        validation=ValidationReport(passed=True, errors=[], warnings=[]),
+        raw_segments=[],
+    )
+
+
+def _make_result_837(n: int = 3, batch_id: str = "b-uuid-1") -> ParseResult:
+    claims = [
+        _make_claim_837(f"CLM-{i}", charge=str(100 + i))
+        for i in range(1, n + 1)
+    ]
+    return ParseResult(
+        envelope=Envelope(
+            sender_id="S", receiver_id="R", control_number="0001",
+            transaction_date=date(2026, 6, 19),
+        ),
+        claims=claims,
+        summary=BatchSummary(
+            input_file="test.txt", control_number="0001",
+            transaction_date=date(2026, 6, 19),
+            total_claims=n, passed=n, failed=0,
+        ),
+    )
+
+
+def _make_batch_record(kind: str = "837p", result=None, batch_id: str = "b-uuid-1",
+                       n: int = 3):
+    return BatchRecord(
+        id=batch_id, kind=kind, input_filename="test.txt",
+        parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
+        result=result if result is not None else _make_result_837(n=n),
+    )
+
+
+def _make_remit_with_cas(remit_id="CLP-1", status="1", charge="124.00",
+                          paid="62.00", cas_amount="62.00", pcn=None):
+    cp = ClaimPayment(
+        payer_claim_control_number=pcn if pcn is not None else remit_id,
+        status_code=status,
+        status_label="Primary",
+        total_charge=charge, total_paid=paid,
+        service_payments=[
+            ServicePayment(
+                line_number=1, procedure_qualifier="HC", procedure_code="99213",
+                charge=charge, payment=paid,
+                adjustments=[ClaimAdjustment(group_code="CO", reason_code="45",
+                                              amount=cas_amount)],
+            ),
+        ],
+    )
+    return cp
+
+
+def _make_835_result(claims):
+    return ParseResult835(
+        envelope=Envelope835(
+            sender_id="S", receiver_id="R", control_number="0001",
+            transaction_date=date(2026, 6, 19),
+        ),
+        financial_info=FinancialInfo(
+            handling_code="C", paid_amount=Decimal("0"),
+            credit_debit_flag="C", payment_method=None,
+        ),
+        trace=ReassociationTrace(
+            trace_type_code="1", trace_number="0001",
+            originating_company_id="S",
+        ),
+        payer=Payer835(name="X", id="SKCO0"),
+        payee=Payee835(name="Y", npi="1234567890"),
+        claims=claims,
+        summary=BatchSummary835(
+            input_file="era.txt", control_number="0001",
+            transaction_date=date(2026, 6, 19),
+            total_claims=len(claims), passed=len(claims), failed=0,
+        ),
+    )
+
+
+def _make_request(path: str = "/api/claims/stream",
+                   query_string: bytes = b"") -> Request:
+    """Build a Starlette ``Request`` wired to ``app.state`` with a no-op
+    receive channel so the endpoint's ``is_disconnected()`` polling
+    works without raising ``RuntimeError: Receive channel has not
+    been made available``.
+    """
+    # ``receive`` blocks until a message is delivered. We never deliver
+    # one, so the cancel scope inside ``is_disconnected()`` always
+    # returns ``False`` — exactly what we want for tests that close
+    # via ``body_iterator.aclose()`` rather than the disconnect path.
+    receive_started = asyncio.Event()
+
+    async def receive():
+        receive_started.set()
+        # Block forever; is_disconnected's cancel scope will cancel
+        # this await and we'll return None (treated as "no message").
+        await asyncio.Event().wait()
+
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": [],
+        "query_string": query_string,
+        "path": path,
+        "app": app,
+        "scheme": "http",
+        "server": ("testserver", 80),
+        "client": ("testclient", 50000),
+    }
+    return Request(scope, receive=receive)
+
+
+def _resolve_query_defaults(endpoint) -> dict:
+    """Extract the actual default values from a FastAPI endpoint signature.
+
+    ``Query(None)`` etc. are marker objects — when called directly we
+    need to substitute their ``.default`` so the endpoint function
+    sees the same values it would receive over HTTP.
+    """
+    import inspect
+    sig = inspect.signature(endpoint)
+    resolved: dict = {}
+    for name, param in sig.parameters.items():
+        default = param.default
+        if hasattr(default, "default"):
+            resolved[name] = default.default
+        else:
+            resolved[name] = default
+    return resolved
+
+
+async def _call_endpoint(endpoint, path: str):
+    """Invoke ``endpoint(request, **defaults)`` with resolved defaults."""
+    request = _make_request(path)
+    defaults = _resolve_query_defaults(endpoint)
+    # Drop ``request`` — it's the positional we pass explicitly.
+    defaults.pop("request", None)
+    return await endpoint(request, **defaults)
+
+
+async def _read_lines(body_iter, n: int, timeout: float = 5.0) -> list[str]:
+    """Read up to ``n`` lines from a StreamingResponse body iterator."""
+    buffer = b""
+    lines: list[str] = []
+    async with asyncio.timeout(timeout):
+        while len(lines) < n:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                break
+            buffer += chunk
+            while b"\n" in buffer and len(lines) < n:
+                raw, buffer = buffer.split(b"\n", 1)
+                if raw:
+                    lines.append(raw.decode("utf-8"))
+    return lines
+
+
+async def _read_one_line(body_iter, timeout: float = 5.0) -> str | None:
+    """Read a single NDJSON line from the body iterator, or None on timeout."""
+    buffer = b""
+    async with asyncio.timeout(timeout):
+        while True:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                return None
+            buffer += chunk
+            if b"\n" in buffer:
+                raw, _ = buffer.split(b"\n", 1)
+                if raw:
+                    return raw.decode("utf-8")
+
+
+async def _drain_until_disconnect(body_iter, max_chunks: int = 20) -> None:
+    """Close the body iterator and swallow the cancellation."""
+    try:
+        await body_iter.aclose()
+    except (asyncio.CancelledError, GeneratorExit):
+        pass
+
+
+async def _read_until_type(
+    body_iter, target: str, *, timeout: float = 5.0, max_lines: int = 50
+) -> dict | None:
+    """Read lines from the body iterator until one has ``type == target``.
+
+    Used by tests that need to skip past heartbeats or multi-item
+    snapshot prefixes to assert on a specific event. Returns the first
+    matching parsed object, or ``None`` if ``max_lines`` is reached
+    without a match.
+    """
+    buffer = b""
+    seen = 0
+    async with asyncio.timeout(timeout):
+        while seen < max_lines:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                return None
+            buffer += chunk
+            while b"\n" in buffer:
+                raw, buffer = buffer.split(b"\n", 1)
+                if not raw:
+                    continue
+                seen += 1
+                obj = json.loads(raw.decode("utf-8"))
+                if obj.get("type") == target:
+                    return obj
+    return None
+
+
+# --------------------------------------------------------------------------- #
+# Task 11 — claims stream
+# --------------------------------------------------------------------------- #
+
+
+async def test_claims_stream_yields_snapshot_then_snapshot_end():
+    """3 claims pre-loaded → 3 item lines + 1 snapshot_end line."""
+    s = CycloneStore()
+    s.add(_make_batch_record(n=3))
+
+    response = await _call_endpoint(claims_stream, "/api/claims/stream")
+    assert response.media_type.startswith("application/x-ndjson")
+
+    lines = await _read_lines(response.body_iterator, n=4)
+    items = [json.loads(line) for line in lines]
+    assert len(items) == 4
+    # First three are items; the order isn't guaranteed without an
+    # explicit sort, so check shape not exact ordering.
+    for item in items[:3]:
+        assert item["type"] == "item"
+        assert "id" in item["data"]
+    # Fourth is the snapshot_end terminator.
+    assert items[3] == {"type": "snapshot_end", "data": {"count": 3}}
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_claims_stream_emits_new_item_after_publish():
+    """After snapshot_end, a bus.publish('claim_written', …) → new item line."""
+    s = CycloneStore()
+    s.add(_make_batch_record(n=1))
+
+    response = await _call_endpoint(claims_stream, "/api/claims/stream")
+    # Drain snapshot phase: 1 item + 1 snapshot_end.
+    lines = await _read_lines(response.body_iterator, n=2)
+    assert json.loads(lines[1]) == {
+        "type": "snapshot_end", "data": {"count": 1},
+    }
+    # Advance the generator one more step so ``subscribe_raw`` runs
+    # and registers the subscriber queue. The next yield will be a
+    # heartbeat (with the 0.2s env var set in some tests); we don't
+    # care about its content here, just that we've moved past the
+    # sync subscription code.
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    # Now publish a claim_written event and assert a new item line
+    # streams out.
+    payload = {
+        "id": "NEW-CLM-1",
+        "patientName": "Live Tail",
+        "providerNpi": "1234567890",
+        "payerName": "Live Payer",
+        "cptCode": "99213",
+        "billedAmount": 999.99,
+        "receivedAmount": 0.0,
+        "status": "submitted",
+        "state": "submitted",
+        "denialReason": None,
+        "submissionDate": "2026-06-19T12:00:00Z",
+        "batchId": "b-uuid-1",
+        "parsedAt": "2026-06-19T12:00:00Z",
+    }
+    await app.state.event_bus.publish("claim_written", payload)
+
+    new_line = await _read_one_line(response.body_iterator, timeout=1.0)
+    assert new_line is not None, "no item line arrived after publish"
+    obj = json.loads(new_line)
+    assert obj["type"] == "item"
+    assert obj["data"]["id"] == "NEW-CLM-1"
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_claims_stream_heartbeat_after_idle(monkeypatch):
+    """With heartbeat=0.2s, a heartbeat line appears within ~1s of idle."""
+    monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")
+    s = CycloneStore()
+    s.add(_make_batch_record(n=1))
+
+    response = await _call_endpoint(claims_stream, "/api/claims/stream")
+    # Drain snapshot (2 lines).
+    lines = await _read_lines(response.body_iterator, n=2)
+    assert json.loads(lines[1]) == {
+        "type": "snapshot_end", "data": {"count": 1},
+    }
+
+    # Read subsequent lines until we see a heartbeat or time out.
+    deadline = asyncio.get_event_loop().time() + 1.5
+    saw_heartbeat = False
+    while asyncio.get_event_loop().time() < deadline:
+        new_line = await _read_one_line(
+            response.body_iterator, timeout=0.5,
+        )
+        if new_line is None:
+            continue
+        obj = json.loads(new_line)
+        if obj["type"] == "heartbeat":
+            assert "ts" in obj["data"]
+            saw_heartbeat = True
+            break
+    assert saw_heartbeat, "no heartbeat line arrived within 1.5s"
+    await _drain_until_disconnect(response.body_iterator)
+
+
+# --------------------------------------------------------------------------- #
+# Task 12 — remittances stream
+# --------------------------------------------------------------------------- #
+
+
+async def test_remittances_stream_yields_snapshot_then_subscribes():
+    """Pre-loaded 835 batch → snapshot (1 item + snapshot_end)."""
+    s = CycloneStore()
+    cp = _make_remit_with_cas(remit_id="CLP-1", pcn="PCN-1")
+    rec = BatchRecord835(
+        id="b-835-1", kind="835", input_filename="era.txt",
+        parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
+        result=_make_835_result([cp]),
+    )
+    s.add(rec)
+
+    response = await _call_endpoint(
+        remittances_stream, "/api/remittances/stream",
+    )
+    assert response.media_type.startswith("application/x-ndjson")
+    # Read the full snapshot (1 item + snapshot_end), skipping nothing.
+    snap_end = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_remittances_stream_emits_after_upsert_remittance():
+    """A bus.publish('remittance_written', …) → new item line on stream."""
+    s = CycloneStore()
+    cp = _make_remit_with_cas(remit_id="CLP-0", pcn="PCN-0")
+    rec = BatchRecord835(
+        id="b-835-0", kind="835", input_filename="era0.txt",
+        parsed_at=datetime(2026, 6, 19, 12, 0, tzinfo=timezone.utc),
+        result=_make_835_result([cp]),
+    )
+    s.add(rec)
+
+    response = await _call_endpoint(
+        remittances_stream, "/api/remittances/stream",
+    )
+    # Drain snapshot to the terminator.
+    snap_end = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
+    # Advance one more step so the subscription is registered before
+    # we publish. The next yielded line may be a heartbeat (with the
+    # 0.2s heartbeat fixture) or a real event — either is fine, we
+    # just need ``subscribe_raw`` to have run.
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    payload = {
+        "id": "PCN-NEW",
+        "claimId": "",
+        "payerName": "Tail Payer",
+        "paidAmount": 12.34,
+        "adjustmentAmount": 0.0,
+        "status": "received",
+        "denialReason": None,
+        "validationWarnings": [],
+        "receivedDate": "2026-06-19T12:00:00Z",
+        "batchId": "b-835-1",
+        "parsedAt": "2026-06-19T12:00:00Z",
+        "adjustments": [],
+    }
+    await app.state.event_bus.publish("remittance_written", payload)
+
+    # Read until the new item line appears (skip any heartbeats that
+    # race in before the published event).
+    new_obj = await _read_until_type(
+        response.body_iterator, "item", timeout=2.0,
+    )
+    assert new_obj is not None, "no item line arrived after publish"
+    assert new_obj["data"]["id"] == "PCN-NEW"
+    await _drain_until_disconnect(response.body_iterator)
+
+
+# --------------------------------------------------------------------------- #
+# Task 13 — activity stream
+# --------------------------------------------------------------------------- #
+
+
+async def test_activity_stream_yields_snapshot_then_subscribes():
+    """An 837 batch seeds activity rows → snapshot has at least 1 item."""
+    s = CycloneStore()
+    s.add(_make_batch_record(n=2))
+
+    response = await _call_endpoint(
+        activity_stream, "/api/activity/stream",
+    )
+    assert response.media_type.startswith("application/x-ndjson")
+    # Read to snapshot_end — activity may have multiple items.
+    snap_end = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap_end is not None
+    assert snap_end["data"]["count"] >= 1
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_activity_stream_emits_after_record_activity():
+    """A bus.publish('activity_recorded', …) → new item line on stream."""
+    s = CycloneStore()
+    s.add(_make_batch_record(n=1))
+
+    response = await _call_endpoint(
+        activity_stream, "/api/activity/stream",
+    )
+    # Drain snapshot.
+    snap_end = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap_end is not None
+    # Advance one step so subscription is registered.
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    payload = {
+        "kind": "claim_submitted",
+        "ts": "2026-06-19T12:00:00Z",
+        "batchId": "b-uuid-1",
+        "claimId": "LIVE-CLM",
+        "remittanceId": None,
+        "payload": {"message": "live tail event"},
+    }
+    await app.state.event_bus.publish("activity_recorded", payload)
+
+    new_obj = await _read_until_type(
+        response.body_iterator, "item", timeout=2.0,
+    )
+    assert new_obj is not None, "no item line arrived after publish"
+    assert new_obj["data"]["kind"] == "claim_submitted"
+    assert new_obj["data"]["claimId"] == "LIVE-CLM"
+    await _drain_until_disconnect(response.body_iterator)
+
+
+# --------------------------------------------------------------------------- #
+# Task 14 — disconnect cancels subscription
+# --------------------------------------------------------------------------- #
+
+
+async def test_client_disconnect_cancels_subscription(monkeypatch):
+    """After closing the body iterator, no claim_written subscribers remain.
+
+    Closes the body iterator (the test-side disconnect signal — see
+    the module docstring). The generator must propagate the
+    ``CancelledError`` and tear down its bus subscription.
+    """
+    monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.1")
+    s = CycloneStore()
+    s.add(_make_batch_record(n=1))
+
+    bus: EventBus = app.state.event_bus
+    response = await _call_endpoint(claims_stream, "/api/claims/stream")
+    # Read the full snapshot to the terminator.
+    snap_end = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap_end == {"type": "snapshot_end", "data": {"count": 1}}
+    # Advance one step into the subscription phase so subscribe_raw
+    # has run and registered the queue.
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    # Pre-condition: at least one subscriber registered while the
+    # stream was open.
+    assert len(bus._subscribers.get("claim_written", [])) >= 1
+
+    # Client disconnect: close the body iterator. The endpoint
+    # generator receives ``CancelledError`` via Starlette's
+    # cancellation path and returns; its `bus.subscribe_raw` queue is
+    # released. Wait briefly for cleanup.
+    await _drain_until_disconnect(response.body_iterator)
+    for _ in range(50):
+        if not bus._subscribers.get("claim_written"):
+            break
+        await asyncio.sleep(0.02)
+
+    assert bus._subscribers.get("claim_written", []) == [], (
+        f"subscribers leaked after disconnect: "
+        f"{bus._subscribers.get('claim_written')!r}"
+    )

backend/tests/test_pubsub.py

@@ -132,3 +132,25 @@ def test_get_event_bus_returns_app_state_bus():
         assert get_event_bus() is bus
     finally:
         app.state.event_bus = saved
+
+
+def test_unsubscribe_removes_queue_from_each_kind():
+    bus = EventBus()
+    q1, _ = bus.subscribe_raw(["a", "b"])
+    q2, _ = bus.subscribe_raw(["a"])
+
+    bus.unsubscribe(q1, ["a", "b"])
+
+    assert q1 not in bus._subscribers.get("a", [])
+    assert q1 not in bus._subscribers.get("b", [])
+    assert q2 in bus._subscribers["a"]
+
+
+def test_unsubscribe_is_idempotent():
+    bus = EventBus()
+    q, _ = bus.subscribe_raw(["a"])
+    bus.unsubscribe(q, ["a"])
+    # Second call must not raise.
+    bus.unsubscribe(q, ["a"])
+    bus.unsubscribe(q, ["b"])  # wrong kind — also fine
+    assert "a" not in bus._subscribers