diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py
index d3592e5..65fbb58 100644
--- a/backend/src/cyclone/api.py
+++ b/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)
 # --------------------------------------------------------------------------- #
diff --git a/backend/src/cyclone/pubsub.py b/backend/src/cyclone/pubsub.py
index 795d228..9b0e7b3 100644
--- a/backend/src/cyclone/pubsub.py
+++ b/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()
diff --git a/backend/tests/test_api_stream_live.py b/backend/tests/test_api_stream_live.py
new file mode 100644
index 0000000..e825324
--- /dev/null
+++ b/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}"
+    )
\ No newline at end of file
diff --git a/backend/tests/test_pubsub.py b/backend/tests/test_pubsub.py
index 83c16e2..d1a4906 100644
--- a/backend/tests/test_pubsub.py
+++ b/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