cyclone/docs/superpowers/plans/2026-06-20-cyclone-live-tail.md

# Sub-project 5 — Live Tail (NDJSON Pub/Sub): Implementation Plan

**Date:** 2026-06-20
**Status:** Approved
**Branch:** `live-tail`
**Spec:** `docs/superpowers/specs/2026-06-20-cyclone-live-tail-design.md`

26 tasks across 5 phases. Follows the same task pattern as the SP3/SP4 plans (`- [ ] Step 1 → Step N → commit`). TDD throughout.

Adds real-time "tail -f" updates to `Claims`, `Remittances`, and `ActivityLog`: when a new claim/remittance/activity event lands in the DB, it appears on the relevant page without a manual refresh. Mechanism is **long-lived NDJSON GET endpoints** backed by an **in-process asyncio pub/sub bus** — no SSE, no polling, no new infra deps.

## Task 0: Add `pytest-asyncio` to dev deps

**Files:**
- Modify: `backend/pyproject.toml` (append `"pytest-asyncio>=0.23,<1"` to `dev` list)
- Modify: `backend/pyproject.toml` (under `[tool.pytest.ini_options]`, set `asyncio_mode = "auto"`)

```bash
.venv/bin/pip install -e '.[dev]'
.venv/bin/pytest -q   # smoke: full suite still green
```

- [ ] **Step 1: Add the dep + config**
- [ ] **Step 2: Install + run full suite, confirm green**
- [ ] **Step 3: Commit**

```bash
git add backend/pyproject.toml backend/uv.lock
git commit -m "build(deps): pytest-asyncio for SP5 stream tests"
```

## File inventory

**Backend (2 new, 3 modified):**
- `backend/src/cyclone/pubsub.py` — `EventBus` (NEW)
- `backend/src/cyclone/api.py` — +3 streaming endpoints; lifespan handler init `app.state.event_bus` (MODIFIED)
- `backend/src/cyclone/store.py` — `upsert_claim/upsert_remittance/record_activity` publish via injected bus (MODIFIED)
- `backend/src/cyclone/__main__.py` — drop the `db.init_db()` call (now lives in lifespan) (MODIFIED)
- `backend/tests/test_pubsub.py` — ~6 tests (NEW)
- `backend/tests/test_api_stream_live.py` — ~8 tests (NEW)

**Frontend (5 new, 4 modified):**
- `src/lib/tail-stream.ts` — NDJSON parser over `ReadableStream` (NEW)
- `src/store/tail-store.ts` — Zustand append-only, FIFO-capped slices (NEW)
- `src/hooks/useTailStream.ts` — connection lifecycle hook (NEW)
- `src/hooks/useMergedTail.ts` — base + tail merge with filter (NEW)
- `src/components/TailStatusPill.tsx` — status badge + reconnect button (NEW)
- `src/lib/tail-stream.test.ts` — ~5 tests (NEW)
- `src/store/tail-store.test.ts` — ~5 tests (NEW)
- `src/hooks/useTailStream.test.ts` — ~6 tests (NEW)
- `src/hooks/useMergedTail.test.ts` — ~4 tests (NEW)
- `src/pages/Claims.tsx` — swap `data.items` → `useMergedTail` + toolbar pill (MODIFIED)
- `src/pages/Remittances.tsx` — same (MODIFIED)
- `src/pages/ActivityLog.tsx` — same (MODIFIED)
- `src/pages/Claims.test.tsx` — +2 live-tail integration tests (MODIFIED)

Total: **9 new files, 6 modified, ~36 new tests.**

---

## Phase 1 — Backend: `EventBus`

### Task 1: `cyclone.pubsub.EventBus` skeleton + `publish`

**Files:**
- Create: `backend/src/cyclone/pubsub.py`
- Create: `backend/tests/test_pubsub.py`

API per spec §3.1:
```python
class EventBus:
    def __init__(self, max_queue_size: int = 256) -> None: ...
    async def publish(self, kind: str, payload: dict) -> None: ...
    def subscribe(self, kinds: list[str]) -> "AsyncIterator[dict]": ...
```

Internals:
- `self._subscribers: dict[str, list[asyncio.Queue[dict]]]`
- `publish`: iterate `_subscribers[kind]`, call `queue.put_nowait(event)`. On `QueueFull`, drop oldest via `get_nowait() + task_done`, then enqueue. Never blocks.
- `subscribe`: returns an async generator; each yielded value is `{**payload, "_kind": kind}`.
- The subscriber queue size equals `max_queue_size` (default 256).

Tests (`backend/tests/test_pubsub.py`, ~6):
- `test_publish_no_subscribers_is_noop` — `publish("x", {"a": 1})` on a fresh bus doesn't raise.
- `test_publish_to_n_subscribers_delivers_to_all`
- `test_subscriber_only_receives_matching_kinds` — subscribe to `["a"]`, publish `b` + `a`, only `a` arrives.
- `test_slow_subscriber_does_not_block_publish` — one subscriber queue full, publish from another task completes within 10ms.
- `test_queue_overflow_drops_oldest` — fill a queue to capacity, publish another, the oldest is gone.
- `test_multiple_concurrent_subscribers_each_get_all_events`

- [ ] **Step 1: Add failing tests** in `backend/tests/test_pubsub.py` (`@pytest.mark.asyncio` on each)
- [ ] **Step 2: Run tests, confirm red**
- [ ] **Step 3: Implement `EventBus`** in `backend/src/cyclone/pubsub.py`
- [ ] **Step 4: Run tests, confirm green**
- [ ] **Step 5: Commit**

```bash
git add backend/src/cyclone/pubsub.py backend/tests/test_pubsub.py
git commit -m "feat(pubsub): EventBus with drop-oldest overflow, per-kind fan-out"
```

### Task 2: Module-level `get_event_bus()` accessor

**Files:**
- Modify: `backend/src/cyclone/pubsub.py`
- Modify: `backend/tests/test_pubsub.py`

Add (per spec §3.3):
```python
def get_event_bus() -> EventBus:
    """Module-level accessor; tests stub via monkeypatch."""
    from cyclone.api import app  # late import to avoid circular
    return app.state.event_bus
```

Test:
- `test_get_event_bus_returns_attribute_error_when_app_not_initialized` — calling before lifespan raises; verifies the late-import contract.

- [ ] **Step 1: Add failing test**
- [ ] **Step 2: Implement the accessor**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add backend/src/cyclone/pubsub.py backend/tests/test_pubsub.py
git commit -m "feat(pubsub): get_event_bus() late-import accessor"
```

### Task 3: Phase 1 commit + full suite

```bash
.venv/bin/pytest -q
```

Expected: ~417 + 1 skip + ~7 new = ~424 pass. All green.

- [ ] **Step 1: Run full backend suite, confirm green**
- [ ] **Step 2: Tag phase complete (no commit unless changes)**

---

## Phase 2 — Backend: lifespan + store publishing

### Task 4: Add FastAPI lifespan handler that initializes `event_bus`

**Files:**
- Modify: `backend/src/cyclone/api.py`

Add at module top:
```python
from contextlib import asynccontextmanager
from cyclone.pubsub import EventBus

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Initialize DB schema/migrations inside the lifespan so the bus has
    # access to a fully-populated store before any request lands.
    from cyclone import db
    db.init_db()
    app.state.event_bus = EventBus()
    yield
    # No teardown needed — single-process local-only.

app = FastAPI(..., lifespan=lifespan)
```

Important: keep the existing `@app.on_event("startup")` if present, OR remove it and rely solely on the lifespan. (Verify which is currently in use; if `@app.on_event` is absent, this is a pure add.)

- [ ] **Step 1: Add the lifespan handler + wire it to `app`**
- [ ] **Step 2: Smoke: import `cyclone.api` succeeds; `app.router.lifespan_context` is set**
- [ ] **Step 3: Commit**

```bash
git add backend/src/cyclone/api.py
git commit -m "feat(api): lifespan handler initializes EventBus + db"
```

### Task 5: Drop `db.init_db()` from `__main__.py`

**Files:**
- Modify: `backend/src/cyclone/__main__.py`

Remove the explicit `db.init_db()` call before `uvicorn.main()` — lifespan handles it now.

- [ ] **Step 1: Remove the line**
- [ ] **Step 2: Smoke: `python -m cyclone serve &; curl /api/health; kill %1`**
- [ ] **Step 3: Commit**

```bash
git add backend/src/cyclone/__main__.py
git commit -m "refactor(main): rely on lifespan for db init (SP5)"
```

### Task 6: `store.upsert_claim` publishes `claim_written`

**Files:**
- Modify: `backend/src/cyclone/store.py`
- Modify: `backend/tests/test_store.py`

Change signature to accept an optional `event_bus: EventBus | None = None`. After the `s.commit()` for an upsert:
```python
if event_bus is not None:
    await event_bus.publish("claim_written", <ui-shaped claim dict>)
```

Test:
- `test_upsert_claim_publishes_event_with_ui_shape` — instantiate an `EventBus`, subscribe to `["claim_written"]`, call `upsert_claim(bus=bus, ...)`, drain the subscriber, assert event payload has same keys as `iter_claims` returns.

- [ ] **Step 1: Add failing test**
- [ ] **Step 2: Implement the publish call**
- [ ] **Step 3: Run test, confirm green**
- [ ] **Step 4: Commit**

```bash
git add backend/src/cyclone/store.py backend/tests/test_store.py
git commit -m "feat(store): upsert_claim publishes claim_written to EventBus"
```

### Task 7: `store.upsert_remittance` + `record_activity` publish

**Files:**
- Modify: `backend/src/cyclone/store.py`
- Modify: `backend/tests/test_store.py`

Same pattern as Task 6 for the other two write paths:
- `upsert_remittance` → `"remittance_written"` (payload = `to_ui_remittance` shape)
- `record_activity` → `"activity_recorded"` (payload = `to_activity_event` shape)

Tests:
- `test_upsert_remittance_publishes_remittance_written`
- `test_record_activity_publishes_activity_recorded`

- [ ] **Step 1: Add 2 failing tests**
- [ ] **Step 2: Implement both publish calls**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add backend/src/cyclone/store.py backend/tests/test_store.py
git commit -m "feat(store): upsert_remittance + record_activity publish events"
```

### Task 8: Wire `event_bus` into parse-837 / parse-835 endpoints

**Files:**
- Modify: `backend/src/cyclone/api.py`

Update the two parse endpoints to pass `event_bus=app.state.event_bus` into the store calls. Test impact: existing parse-837/parse-835 tests should continue to pass — the bus is wired but no subscribers are attached in those tests, so `publish` is a no-op.

- [ ] **Step 1: Wire the bus into both endpoints**
- [ ] **Step 2: Run full backend suite, confirm green**
- [ ] **Step 3: Commit**

```bash
git add backend/src/cyclone/api.py
git commit -m "feat(api): parse endpoints pass EventBus into store writes"
```

### Task 9: Phase 2 commit + full suite

```bash
.venv/bin/pytest -q
```

Expected: ~424 + ~3 new = ~427 pass. All green.

- [ ] **Step 1: Run full suite, confirm green**
- [ ] **Step 2: Tag phase complete**

---

## Phase 3 — Backend: streaming endpoints

### Task 10: `_stream_ndjson` helper

**Files:**
- Modify: `backend/src/cyclone/api.py`

Helper that wraps a generator of dict events into NDJSON lines:
```python
def _ndjson_line(event: dict) -> bytes:
    return (json.dumps(event, separators=(",", ":")) + "\n").encode("utf-8")
```

- [ ] **Step 1: Add the helper**
- [ ] **Step 2: Commit**

```bash
git add backend/src/cyclone/api.py
git commit -m "refactor(api): _ndjson_line helper for streaming responses"
```

### Task 11: `GET /api/claims/stream`

**Files:**
- Modify: `backend/src/cyclone/api.py`
- Create: `backend/tests/test_api_stream_live.py`

Endpoint signature mirrors `list_claims` query params (forwarded):
```python
@app.get("/api/claims/stream")
async def claims_stream(
    request: Request,
    status: str | None = None,
    provider_npi: str | None = None,
    payer: str | None = None,
    date_from: str | None = None,
    date_to: str | None = None,
    sort: str | None = None,
    order: str = "desc",
    limit: int = 100,
):
    bus: EventBus = request.app.state.event_bus
    async def gen():
        # 1. Snapshot
        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
        async for event in bus.subscribe(["claim_written"]):
            if await request.is_disconnected():
                return
            yield _ndjson_line({"type": "item", "data": event})
    return StreamingResponse(gen(), media_type="application/x-ndjson")
```

Tests (with `httpx.AsyncClient` + `pytest-asyncio`):
- `test_claims_stream_yields_snapshot_then_snapshot_end` — pre-populate 3 claims, hit stream, assert first 3 lines are `{"type":"item"}`, 4th is `{"type":"snapshot_end","data":{"count":3}}`.
- `test_claims_stream_emits_new_item_after_publish` — open stream in one task, call `bus.publish("claim_written", payload)` in another, assert the new item appears within 100ms.
- `test_claims_stream_heartbeat_after_15s_idle` — override heartbeat to 1s via env var or test fixture; assert heartbeat line arrives within 1.5s.

- [ ] **Step 1: Add 3 failing tests**
- [ ] **Step 2: Implement the endpoint (without heartbeat first)**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Add the 15s heartbeat via `asyncio.wait_for` loop** (overridable for tests)
- [ ] **Step 5: Run tests, confirm green**
- [ ] **Step 6: Commit**

```bash
git add backend/src/cyclone/api.py backend/tests/test_api_stream_live.py
git commit -m "feat(api): GET /api/claims/stream — snapshot + subscribe + heartbeat"
```

### Task 12: `GET /api/remittances/stream`

**Files:**
- Modify: `backend/src/cyclone/api.py`
- Modify: `backend/tests/test_api_stream_live.py`

Same shape; subscribes to `["remittance_written"]`. Default sort `-received_date`.

Tests:
- `test_remittances_stream_yields_snapshot_then_subscribes`
- `test_remittances_stream_emits_after_upsert_remittance`

- [ ] **Step 1: Add 2 failing tests**
- [ ] **Step 2: Implement the endpoint**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add backend/src/cyclone/api.py backend/tests/test_api_stream_live.py
git commit -m "feat(api): GET /api/remittances/stream"
```

### Task 13: `GET /api/activity/stream`

**Files:**
- Modify: `backend/src/cyclone/api.py`
- Modify: `backend/tests/test_api_stream_live.py`

Same shape; subscribes to `["activity_recorded"]`. Default sort `-timestamp`, default `limit=50` (smaller — activity is high-volume).

Tests:
- `test_activity_stream_yields_snapshot_then_subscribes`
- `test_activity_stream_emits_after_record_activity`

- [ ] **Step 1: Add 2 failing tests**
- [ ] **Step 2: Implement the endpoint**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add backend/src/cyclone/api.py backend/tests/test_api_stream_live.py
git commit -m "feat(api): GET /api/activity/stream (limit=50)"
```

### Task 14: Client disconnect cancels subscription

**Files:**
- Modify: `backend/tests/test_api_stream_live.py`

Test:
- `test_client_disconnect_cancels_subscription` — open stream, read `snapshot_end`, close the response, then assert `bus._subscribers["claim_written"]` is empty (use a fresh bus per test).

- [ ] **Step 1: Add failing test**
- [ ] **Step 2: If not already implemented (the `request.is_disconnected()` check in Task 11 covers it), confirm**
- [ ] **Step 3: Run test, confirm green**
- [ ] **Step 4: Commit if any change**

### Task 15: Phase 3 commit + full suite

```bash
.venv/bin/pytest -q
```

Expected: ~427 + ~10 new = ~437 pass. All green.

- [ ] **Step 1: Run full suite, confirm green**
- [ ] **Step 2: Tag phase complete**

---

## Phase 4 — Frontend: NDJSON parser + tail store

### Task 16: `src/lib/tail-stream.ts` — NDJSON parser

**Files:**
- Create: `src/lib/tail-stream.ts`
- Create: `src/lib/tail-stream.test.ts`

API per spec §3.4:
```ts
export type TailEvent =
  | { type: "item"; data: unknown }
  | { type: "snapshot_end"; data: { count: number } }
  | { type: "heartbeat"; data: { ts: string } }
  | { type: "item_dropped"; data: { id: string } }
  | { type: "error"; data: { message: string } };

export type TailResource = "claims" | "remittances" | "activity";

export async function* streamTail(
  resource: TailResource,
  opts?: { signal?: AbortSignal; baseUrl?: string },
): AsyncIterableIterator<TailEvent>;
```

Implementation: `fetch(${baseUrl}/api/${resource}/stream, { signal })` with `Accept: application/x-ndjson` header; pipe `response.body!` through `TextDecoderStream` then an NDJSON-line splitter that yields `JSON.parse`d typed events. Malformed lines are skipped with `console.warn`.

Tests (`src/lib/tail-stream.test.ts`, ~5):
- `test_parses_well_formed_ndjson_into_typed_events` — feed a `ReadableStream` of well-formed lines (use a `TransformStream` helper in tests).
- `test_yields_typed_error_on_error_line` — `{"type":"error","data":{"message":"x"}}` yields as `{ type: "error", data: { message: "x" } }`.
- `test_handles_heartbeat_silently` — heartbeat line yields correctly typed.
- `test_abort_signal_cancels_mid_stream` — pass an `AbortController`, abort after first event, generator throws/returns without further yield.
- `test_malformed_line_skipped_next_valid_still_emitted` — bad JSON in middle, surrounding good lines still parse.

- [ ] **Step 1: Add failing tests**
- [ ] **Step 2: Implement `streamTail`**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add src/lib/tail-stream.ts src/lib/tail-stream.test.ts
git commit -m "feat(frontend): NDJSON streamTail parser for /api/{resource}/stream"
```

### Task 17: `src/store/tail-store.ts` — Zustand append-only with FIFO cap

**Files:**
- Create: `src/store/tail-store.ts`
- Create: `src/store/tail-store.test.ts`

API per spec §3.4:
```ts
export const TAIL_CAP = 10_000;

interface TailStore {
  claims: Record<string, ClaimListItem>;
  remittances: Record<string, RemitListItem>;
  activity: ActivityEvent[];
  addClaim: (c: ClaimListItem) => void;
  addRemittance: (r: RemitListItem) => void;
  addActivity: (a: ActivityEvent) => void;
  reset: (resource: TailResource) => void;
}

export const useTailStore = create<TailStore>((set) => ({ ... }));
```

Dedup: `addClaim(c)` is a no-op when `c.id` already present. FIFO cap: when `Object.keys(claims).length > TAIL_CAP`, drop the oldest by id (sort keys by insertion order via a small parallel `OrderMap`, or just evict the first 100 oldest when the cap is exceeded).

Tests (~5):
- `test_add_claim_adds_new_claim_keyed_by_id`
- `test_add_claim_with_duplicate_id_is_noop`
- `test_reset_claims_clears_only_claims_slice`
- `test_add_activity_appends` (no stable id)
- `test_fifo_cap_evicts_oldest_when_over_10000`

- [ ] **Step 1: Add failing tests**
- [ ] **Step 2: Implement the store**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add src/store/tail-store.ts src/store/tail-store.test.ts
git commit -m "feat(frontend): tail-store Zustand with FIFO cap 10k per slice"
```

### Task 18: Phase 4 commit + frontend check

```bash
npm run typecheck
npm test -- --run
```

Expected: 124 + ~10 new = ~134 frontend pass.

- [ ] **Step 1: Run typecheck + tests, confirm green**
- [ ] **Step 2: Commit if any change**

---

## Phase 5 — Frontend: hooks + page integration

### Task 19: `src/hooks/useTailStream.ts` — connection lifecycle

**Files:**
- Create: `src/hooks/useTailStream.ts`
- Create: `src/hooks/useTailStream.test.ts`

API per spec §3.4:
```ts
export type TailStatus = "connecting" | "live" | "reconnecting" | "closed" | "stalled" | "error";

export function useTailStream(resource: TailResource): {
  status: TailStatus;
  lastEventAt: Date | null;
  error: Error | null;
  forceReconnect: () => void;
};
```

Behavior (per spec):
- On mount: opens stream; status = `connecting`.
- On `snapshot_end`: status = `live`.
- On `item`: dispatches to `useTailStore`; updates `lastEventAt`.
- On error: status = `reconnecting`; backoff 1s → 2s → 4s → 8s → 16s → 30s cap.
- If no event (incl. heartbeat) for 30s: status = `stalled`.
- On unmount: abort; status = `closed`.
- `forceReconnect()`: cancels and re-opens immediately.

Tests (~6) — mock `streamTail` with a vitest module mock:
- `test_on_mount_status_connecting_then_live_after_snapshot_end`
- `test_on_stream_error_status_error`
- `test_on_abort_status_closed_no_reconnect`
- `test_on_event_dispatches_to_tail_store` — assert `useTailStore.getState().addClaim` is called.
- `test_reconnect_status_cycles_reconnecting_to_connecting_to_live`
- `test_reconnect_dedup_duplicate_items_appear_once_in_store`

- [ ] **Step 1: Add failing tests**
- [ ] **Step 2: Implement the hook** (use `useEffect` + `useRef` for backoff state; `AbortController` per attempt)
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add src/hooks/useTailStream.ts src/hooks/useTailStream.test.ts
git commit -m "feat(frontend): useTailStream lifecycle hook with backoff + stall detection"
```

### Task 20: `src/hooks/useMergedTail.ts` — merge base + tail with filter

**Files:**
- Create: `src/hooks/useMergedTail.ts`
- Create: `src/hooks/useMergedTail.test.ts`

API per spec §3.4:
```ts
export function useMergedTail<T extends { id: string }>(
  resource: TailResource,
  baseItems: T[],
  filterFn?: (item: T) => boolean,
): T[];
```

Behavior:
- Reads the matching slice from `useTailStore`.
- Applies `filterFn` to tail items if provided.
- Dedups by `id` against `baseItems`.
- Returns `baseItems` followed by tail items in arrival order.

Tests (~4):
- `test_merges_base_and_tail_by_id_base_first`
- `test_filter_predicate_drops_tail_items_that_dont_match`
- `test_empty_tail_store_returns_base_unchanged`
- `test_new_tail_item_appears_at_top`

- [ ] **Step 1: Add failing tests**
- [ ] **Step 2: Implement the hook**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add src/hooks/useMergedTail.ts src/hooks/useMergedTail.test.ts
git commit -m "feat(frontend): useMergedTail merges base + tail with filter"
```

### Task 21: `src/components/TailStatusPill.tsx` — status badge + reconnect button

**Files:**
- Create: `src/components/TailStatusPill.tsx`

Small component (no test needed beyond a smoke render check inside Claims.test.tsx):
- Props: `{ status: TailStatus; lastEventAt: Date | null; onReconnect: () => void }`
- Renders the existing `<Badge>` with variant per status (live=success, connecting/reconnecting=warning, closed/stalled/error=error/destructive), plus "Last event: 12s ago" subtitle (computed via `useEffect` ticking every second), plus a "↻ Reconnect" button shown only when `status === "stalled" || status === "error"`.

- [ ] **Step 1: Implement the component**
- [ ] **Step 2: Commit**

```bash
git add src/components/TailStatusPill.tsx
git commit -m "feat(frontend): TailStatusPill with reconnect button"
```

### Task 22: `Claims.tsx` — swap to `useMergedTail` + toolbar pill

**Files:**
- Modify: `src/pages/Claims.tsx`
- Modify: `src/pages/Claims.test.tsx`

Wire-up:
- Replace `data?.items ?? []` with `useMergedTail("claims", data?.items ?? [], filterFn)`.
- Call `useTailStream("claims")` for status.
- Render `<TailStatusPill>` in the toolbar with status + lastEventAt + onReconnect = `forceReconnect`.
- Existing tests must still pass; add 2 new tests:
  - `test_live_tail_arrival_triggers_row_in_table` — mock `streamTail` to emit an `item` for a new claim, assert row count + row text appears.
  - `test_status_pill_shows_live_after_stream_connects` — mock `streamTail` to emit `snapshot_end`, assert pill text contains `live`.

- [ ] **Step 1: Add 2 failing tests**
- [ ] **Step 2: Wire up the page**
- [ ] **Step 3: Run tests, confirm green**
- [ ] **Step 4: Commit**

```bash
git add src/pages/Claims.tsx src/pages/Claims.test.tsx
git commit -m "feat(frontend): Claims page wires live tail + TailStatusPill"
```

### Task 23: `Remittances.tsx` + `ActivityLog.tsx` — same pattern

**Files:**
- Modify: `src/pages/Remittances.tsx`
- Modify: `src/pages/ActivityLog.tsx`

Mirror Task 22: swap to `useMergedTail("remittances", ...)` / `useMergedTail("activity", ...)`, render `<TailStatusPill>` in the toolbar. Use the same `filterFn` the page already passes to its list query.

- [ ] **Step 1: Wire both pages**
- [ ] **Step 2: Run typecheck + tests**
- [ ] **Step 3: Commit**

```bash
git add src/pages/Remittances.tsx src/pages/ActivityLog.tsx
git commit -m "feat(frontend): Remittances + ActivityLog wire live tail"
```

### Task 24: Phase 5 commit + full suite

```bash
npm run typecheck
npm test -- --run
.venv/bin/pytest -q
```

Expected: ~134 + ~10 new = ~144 frontend; ~437 backend. All green.

- [ ] **Step 1: Run all checks, confirm green**
- [ ] **Step 2: Commit if any change**

---

## Phase 6 — Docs + smoke + merge

### Task 25: End-to-end smoke

Per spec §8 smoke:

1. Start backend (`backend/.venv/bin/python -m cyclone serve &`).
2. Start frontend (`npm run dev &`).
3. Open `http://localhost:5173/claims` in browser.
4. In another tab (or via curl), POST `co_medicaid_837p.txt` to `/api/parse-837` (`curl -F file=@...`).
5. Verify the first `/claims` tab shows the new rows appearing without manual refresh.
6. Verify the status pill reads `live`.
7. Verify the backend logs show no errors.
8. Kill both servers.

- [ ] **Step 1: Run the smoke**
- [ ] **Step 2: Commit empty**

```bash
git commit --allow-empty -m "smoke: end-to-end SP5 (live tail) flow passes"
```

### Task 26: README + merge to main

**Files:**
- Modify: `README.md` — under "Roadmap" or new "Live updates" section, document the tail behavior + status pill + reconnect UX.

- [ ] **Step 1: Update README**
- [ ] **Step 2: Commit**

```bash
git add README.md
git commit -m "docs: document live-tail behavior in README"
```

### Task 27: Cleanup + merge to main

1. `git status` — clean.
2. `git merge --ff-only live-tail`.
3. From `main`: `cd backend && .venv/bin/pytest -q` → must show ~437 pass + 1 skip; `cd .. && npm test -- --run` → ~144 pass.
4. `git worktree remove .worktrees/live-tail` (if applicable).
5. `git branch -d live-tail`.
6. Update SP4 plan checkboxes (if any drift) — not applicable here; this is the SP5 plan itself. Tick all `[ ]` to `[x]` in this file before committing.

- [ ] **Step 1: Verify clean**
- [ ] **Step 2: Discard stale uncommitted changes (if any)**
- [ ] **Step 3: Fast-forward merge to main**
- [ ] **Step 4: Remove worktree + delete branch**
- [ ] **Step 5: Tick all `[ ]` to `[x]` in this file; commit as `docs: tick SP5 plan checkboxes`**

---

## Test count targets

| Phase | New backend | New frontend | Running total |
|-------|-------------|--------------|---------------|
| Start (post-SP4) | 417 + 1 skip | 124 | 417 / 124 |
| 0 (deps) | 0 | 0 | 417 / 124 |
| P1 end | +7 | 0 | 424 / 124 |
| P2 end | +3 | 0 | 427 / 124 |
| P3 end | +10 | 0 | 437 / 124 |
| P4 end | 0 | +10 | 437 / 134 |
| P5 end | 0 | +10 (incl. 2 page tests) | 437 / 144 |
| **Final target** | **~437 + 1 skip** | **~144** | — |

## Risk reminders

- **`pytest-asyncio` config** — set `asyncio_mode = "auto"` so `@pytest.mark.asyncio` isn't required per test. Verify no other test file gets accidentally parallelized.
- **Lifespan replaces on_event** — if any existing startup hook is wired via `@app.on_event("startup")`, either fold it into the lifespan or leave both (idempotent). Check before deleting the existing handler in `__main__`.
- **Tail-store growth** — cap at 10k per slice (FIFO). Operators who never refresh the page won't OOM the tab.
- **Heartbeat timing in tests** — make the 15s heartbeat overridable via env var (`CYCLONE_HEARTBEAT_INTERVAL_S`) so tests can use 0.1s.
- **Stall threshold vs reconnect** — 30s stall threshold; the hook should NOT auto-reconnect on stall (user confirms). Distinguish from `reconnecting` (auto-backoff in progress).
- **NDJSON partial lines** — the splitter must buffer trailing partial chunks across stream chunks; verify with a test that streams a line across two `ReadableStream` pulls.
- **Filter drift on the client** — tail-store contains items that may not match the current page filter; `useMergedTail` filters them out. Verify with a test that flips the filter mid-stream.
- **ActivityLog already streams from parse-837** — the new tail stream is **additive**, not a replacement. The page already has its own `parse837` NDJSON consumer; the tail stream survives parse completion.

## Worktree note

Use `superpowers:using-git-worktrees` to create `.worktrees/live-tail` from `main` before Task 1. All commits land on the `live-tail` branch until the final ff-merge to `main`.