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

Cyclone Live Tail — Design Spec

Date: 2026-06-20 Status: Approved Branch: live-tail

1. Scope

Add real-time "tail -f" updates to the Claims, Remittances, and Activity pages. When a new claim/remittance/activity event lands in the DB (via parse-837, parse-835, reconciliation match, etc.), it appears in the relevant page without a manual refresh.

The mechanism is long-lived NDJSON GET endpoints backed by an in-process pub/sub bus, not SSE and not polling. This fits Cyclone's local-only single-process posture, reuses the existing Accept: application/x-ndjson content-negotiation pattern, and avoids adding new dependencies.

Out of scope (explicitly):

• Batches / Providers / Acks pages (per "claims + remittances + activity" choice).
• Filter-aware server push (server pushes all; client filters).
• Cross-process pub/sub (single FastAPI process only).
• SSE.
• Cursor-based replay on reconnect (full snapshot replay + client-side dedup).
• Multi-tab sync (multiple tabs each open their own stream; acceptable for v1).
• Bi-directional updates (e.g. drawer mutation → other tabs see it).

2. Architecture

┌─────────────────────────────────────────────────────────────────┐
│ Frontend (browser, React 19 + TanStack Query v5)                │
│                                                                 │
│  Pages: Claims.tsx / Remittances.tsx / ActivityLog.tsx          │
│     │                                                           │
│     └─ existing useClaims / useRemittances / useActivity        │
│        (JSON GET, filter/sort/pagination — unchanged)           │
│                                                                 │
│  NEW: useTailStream(resource) ── fetch ReadableStream reader    │
│     │                                                           │
│     └─ tail-store (Zustand) — append-only items array           │
│                                                                 │
│  NEW: useMergedTail(resource, baseItems, filterFn)              │
│     merges filtered JSON results with tail-store items          │
└────────────────────────────┬────────────────────────────────────┘
                             │ GET /api/{res}/stream
                             │ Accept: application/x-ndjson
                             ▼
┌─────────────────────────────────────────────────────────────────┐
│ Backend (FastAPI)                                               │
│                                                                 │
│  NEW: pubsub.py — EventBus (per-process asyncio pub/sub)        │
│     • publish(kind, payload)                                    │
│     • subscribe(kinds) -> AsyncIterator[Event]                  │
│                                                                 │
│  NEW: 3 streaming GET endpoints (claims, remittances, activity)  │
│     • yield initial snapshot as NDJSON item lines               │
│     • yield {"type":"snapshot_end"} once caught up              │
│     • subscribe to EventBus; yield {"type":"item", ...} on push │
│     • heartbeat {"type":"heartbeat"} every 15s                  │
│                                                                 │
│  MODIFIED: parse_837/parse_835/reconcile_match/store.*          │
│     • call event_bus.publish(kind, payload) after write          │
└─────────────────────────────────────────────────────────────────┘

Key decisions:

• One stream per resource (not multiplexed) — simpler URL semantics, easier per-page lifecycle.
• Existing JSON GETs unchanged — pagination, filters, sorting still work; the tail is an additive append layer.
• No SSE — stays in the existing NDJSON surface.
• One Zustand store, three slices — useTailStore.claims/remittances/activity; pages subscribe to their slice via selectors.

3. Locked decisions

3.1 EventBus: in-process, asyncio-backed, drop-on-overflow

A new module backend/src/cyclone/pubsub.py with a single EventBus class:

class EventBus:
    def __init__(self, max_queue_size: int = 256) -> None: ...
    async def publish(self, kind: str, payload: dict) -> None: ...
    async def subscribe(self, kinds: list[str]) -> AsyncIterator[dict]: ...

Internals:

• self._subscribers: dict[str, list[asyncio.Queue[dict]]] — keyed by kind, list of per-subscriber queues.
• publish() iterates the matching kind subscribers, calls queue.put_nowait(event) on each. If a queue is full (asyncio.QueueFull), the oldest event is dropped via queue.get_nowait() + task_done, then the new one is enqueued. Never blocks.
• subscribe() yields an async generator; each yielded value is {**payload, "_kind": kind}.
• Held on app.state.event_bus (FastAPI app-state pattern); initialized in cyclone.api:app.on_event("startup") (or lifespan handler).
• One instance per process. Tests instantiate a fresh EventBus() per test.

Why drop-on-overflow (not block): Cyclone is local-only. A subscriber that falls behind (e.g. browser tab backgrounded for 10 minutes) should not slow the publisher (which is the live parse path the operator is actively watching). Dropping the oldest event keeps the queue bounded; the client detects staleness via lastEventAt and forces a reconnect.

3.2 Three new NDJSON streaming endpoints

GET /api/claims/stream
GET /api/remittances/stream
GET /api/activity/stream

Each is a StreamingResponse(media_type="application/x-ndjson") driven by an async generator. Default filter is "show me what I would see on page 1 of the list" — i.e. same default limit=100, sort=-submission_date (or equivalent) as the corresponding JSON GET. Query params that match the JSON GET (filter, sort, order) are forwarded so clients can scope the stream to e.g. "submitted-only claims".

Each generator:

1. Iterates the matching store.iter_*() with the same defaults as the JSON GET, yielding:

   {"type":"item","data":<existing_item>}
   {"type":"item","data":<existing_item>}
   ...
   {"type":"snapshot_end","data":{"count":N}}
   

2. Then awaits events from event_bus.subscribe([<kinds>]):

   {"type":"item","data":<new_item>}      # pushed from the bus
   {"type":"item","data":<new_item>}
   ...
   

3. Every 15 seconds of idle (no items, no events), yields:

   {"type":"heartbeat","data":{"ts":"2026-06-20T..."}}
   

4. On client_disconnected (Starlette detects request abort), cancels the inner subscribe() task and returns. No error response — the connection simply ends.

The kinds each endpoint subscribes to:

• /api/claims/stream → ["claim_written"]
• /api/remittances/stream → ["remittance_written"]
• /api/activity/stream → ["activity_recorded"]

3.3 Store write paths publish events

backend/src/cyclone/store.py already owns the write paths for claims, remittances, and activity. Each successful write is followed by:

await app.state.event_bus.publish(kind, payload)

payload is the serialized version of the row as it would appear in the JSON GET response — same shape, same camelCase keys — so the client doesn't need to re-fetch.

The event_bus is injected via a thin module-level helper to avoid threading app.state through every store method:

# backend/src/cyclone/pubsub.py
def get_event_bus() -> EventBus:
    """Module-level accessor; tests stub this."""
    from cyclone.api import app  # late import to avoid circular
    return app.state.event_bus

Write-paths that publish:

• upsert_claim(...) → publishes "claim_written" with the claim dict.
• upsert_remittance(...) → publishes "remittance_written" with the remittance dict.
• record_activity(...) → publishes "activity_recorded" with the activity event dict.

3.4 Frontend: NDJSON parser + tail store + merge hook

src/lib/tail-stream.ts — pure parser over a ReadableStream:

type TailEvent =
  | { type: "item"; data: unknown }
  | { type: "snapshot_end"; data: { count: number } }
  | { type: "heartbeat"; data: { ts: string } }
  | { type: "error"; data: { message: string } };

async function* streamTail(
  resource: "claims" | "remittances" | "activity",
  opts?: { signal?: AbortSignal; baseUrl?: string }
): AsyncIterableIterator<TailEvent>;

Implementation: fetch(/api/${resource}/stream, { headers: { Accept: "application/x-ndjson" }, signal }), then response.body!.pipeThrough(new TextDecoderStream()).pipeThrough(ndjsonLineSplit()), then JSON.parse each line, validate type is one of the known values, yield typed TailEvent.

src/store/tail-store.ts — Zustand store, three slices:

type TailStore = {
  claims: Record<string, ClaimListItem>;        // dedup by id
  remittances: Record<string, RemitListItem>;   // dedup by id
  activity: ActivityEvent[];                    // append-only (composite id)

  addClaim: (c: ClaimListItem) => void;
  addRemittance: (r: RemitListItem) => void;
  addActivity: (a: ActivityEvent) => void;
  reset: (resource: "claims" | "remittances" | "activity") => void;
};

Dedup: addClaim(c) does set((s) => s.claims[c.id] ? s : { claims: { ...s.claims, [c.id]: c } }) — no-op on duplicate.

src/hooks/useTailStream.ts — connection lifecycle:

type TailStatus = "connecting" | "live" | "reconnecting" | "closed" | "stalled" | "error";

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

Behavior:

• On mount: opens stream; status = connecting.
• On snapshot_end: status = live.
• On item event: dispatches to tail-store; updates lastEventAt.
• On disconnect/parse error: status = reconnecting; schedules reconnect with exponential backoff (1s, 2s, 4s, 8s, 16s, 30s cap).
• If no event (including heartbeat) for 30s: status = stalled. Toolbar shows "↻ reconnect" button. No silent auto-reconnect — user confirms.
• If tab visibility returns and lastEventAt is >30s old: forces reconnect.
• On unmount: aborts controller; status = closed.

src/hooks/useMergedTail.ts — merge base + tail with filter:

function useMergedTail<T>(
  resource: TailResource,
  baseItems: T[],
  filterFn?: (item: T) => boolean
): T[];

Behavior: reads the matching slice from tail-store; for each item applies filterFn (if provided); dedups by id against baseItems; returns baseItems followed by tail items in arrival order.

The page passes the SAME filter function used by the JSON results so tail items that don't match the current filter don't appear.

3.5 Page integration

Each of Claims.tsx, Remittances.tsx, ActivityLog.tsx:

- const items = data?.items ?? [];
+ const items = useMergedTail("claims", data?.items ?? [], filterFn);
+ const { status, lastEventAt } = useTailStream("claims");

Toolbar additions (each page):

• Small status pill: 🟢 live / 🟡 reconnecting... / 🔴 offline (using the existing Badge primitive).
• "Last event: 12s ago" subtitle.
• Counter: "N new since you opened this page" (resets on filter change or manual refresh).

For ActivityLog specifically: the page already shows streaming progress from parse837's NDJSON stream. The tail stream is an additional, always-on subscription that survives parse837 completion. They coexist; the activity event kind filter on the page hides parse events if "filter to matches only" is active.

4. Data flow

Initial page load

1. User opens /claims.
2. <Claims> mounts.
3. useClaims() fires JSON GET /api/claims?limit=100 → returns { items: [...100 rows], total, has_more }.
4. <Claims> renders the table from data.items.
5. useTailStream('claims') fires GET /api/claims/stream.
6. Stream yields up to 100 existing items as {"type":"item"} lines → tail-store.claims populated.
7. Stream yields {"type":"snapshot_end"} → status flips to live.
8. useMergedTail('claims', data.items, filterFn) returns data.items (no new items yet, merge is a no-op).
9. Page is indistinguishable from the old behavior so far.

A new claim arrives (Upload page submits)

1. POST /api/parse-837 (Upload page).
2. Backend persists claim in store.upsert_claim(...).
3. Store calls event_bus.publish("claim_written", payload).
4. EventBus puts payload on every subscriber's queue.
5. /api/claims/stream subscriber wakes; yields {"type":"item","data":<claim>}.
6. useTailStream receives it; tail-store.addClaim(payload).
7. useMergedTail selector re-fires (Zustand subscription); new row appears at the top of the table.
8. Row-flash CSS animation triggers (~800ms fade).
9. Toolbar indicator ticks to 🟢 live · 1 new.

Network blip

1. TCP connection drops mid-stream.
2. useTailStream reader errors.
3. Status flips to reconnecting; backoff timer starts (1s, 2s, 4s, ...).
4. New fetch fires; stream replays the snapshot from the start.
5. snapshot_end fires; status flips back to live.
6. Existing tail-store items NOT cleared across reconnect. Merge keeps them; dedup drops the dups.

Filter change

1. User clicks "submitted only" filter on Claims.
2. useClaims() refires JSON GET with new filter param.
3. useMergedTail('claims', data.items, filterFn) re-fires.
4. tail-store may contain items that DON'T match the new filter (e.g. a "paid" claim arrived while filter was "submitted").
5. useMergedTail applies the SAME filterFn to tail-store items before merging. Only matching tail items appear.

Page navigates away

1. <Claims> unmounts.
2. useTailStream cleanup aborts the controller.
3. Stream endpoint sees client disconnect; cancels subscription.
4. tail-store.claims is NOT cleared (acceptable — only Claims reads it).
5. Next visit to /claims re-opens the stream; tail-store updates.

5. Error handling

Failure	User-visible behavior
Stream endpoint 5xx (e.g. db not initialized)	useTailStream catches; status = error; toolbar shows red ⚠ tail offline with retry button. JSON GET still works → page is degraded but functional.
Stream endpoint 4xx	Same as 5xx.
TCP connection drops mid-stream	status = reconnecting; exponential backoff (1s, 2s, 4s, 8s, 16s, 30s cap); reconnect replays snapshot, dedup handles duplicates.
Malformed NDJSON line	Parser skips the line, logs console.warn, increments parseErrors counter exposed on hook status.
{"type":"error"} line from server	Hook flips to error status; toolbar shows the server's error message.
EventBus queue overflow	Drop oldest queued event, log warn, emit {"type":"item_dropped","id":...} so client can decide to refetch. Drop-not-block.
Tab backgrounded	Browser throttles fetch; on focus return, hook detects stale lastEventAt (>30s while hidden), forces reconnect.
Backend process restart	All subscribers drop; clients reconnect with backoff; on reconnect they get a fresh snapshot.

6. API contract

GET /api/claims/stream

Query params (forwarded from /api/claims): status, provider_npi, payer, date_from, date_to, sort, order, limit.

Response: Content-Type: application/x-ndjson. Stream of lines:

{"type":"item","data":{"id":"...","patientName":"...","billedAmount":85.40,...}}
{"type":"item","data":{"id":"...","patientName":"...","billedAmount":155.76,...}}
...
{"type":"snapshot_end","data":{"count":100}}
{"type":"item","data":{"id":"<new>","patientName":"...","billedAmount":42.00,...}}
{"type":"heartbeat","data":{"ts":"2026-06-20T12:34:56.789Z"}}

Errors: 4xx/5xx before the stream begins returns a JSON {detail: "..."}. After the stream begins, errors are emitted as {"type":"error","data":{"message":"..."}}.

GET /api/remittances/stream

Same shape as claims. Default sort: -received_date. Subscribes to remittance_written.

GET /api/activity/stream

Same shape. Default sort: -timestamp, limit=50 (smaller — activity is high-volume). Subscribes to activity_recorded.

Event payload shape

Each NDJSON item.data line is the same JSON shape as the corresponding list endpoint's items[i]. No new fields, no renamed fields — the client uses the exact same row component for both JSON-fetched and tail-fetched items.

7. File inventory

Backend (3 modified, 2 new):

• backend/src/cyclone/pubsub.py — EventBus (NEW)
• backend/src/cyclone/api.py — +3 streaming endpoints, register event_bus on app state (MODIFIED)
• backend/src/cyclone/store.py — publish events from upsert_claim, upsert_remittance, record_activity (MODIFIED)
• backend/tests/test_pubsub.py — ~6 tests (NEW)
• backend/tests/test_api_stream_live.py — ~8 tests (NEW)

Frontend (2 modified, 5 new):

• src/lib/tail-stream.ts — NDJSON parser over fetch stream (NEW)
• src/store/tail-store.ts — Zustand append-only store (NEW)
• src/hooks/useTailStream.ts — connection lifecycle (NEW)
• src/hooks/useMergedTail.ts — base + tail merge with filter (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 indicator (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 files, ~36 new tests.

8. Testing strategy

Backend

test_pubsub.py (~6 tests):

• publish to 0 subscribers is a no-op
• publish to N subscribers delivers to all
• subscriber only receives matching kinds
• slow subscriber doesn't block publish
• queue overflow drops oldest, not newest
• multiple concurrent subscribers each get all events

test_api_stream_live.py (~8 tests):

• /api/claims/stream yields existing items then snapshot_end
• emits a new item within 100ms when upsert_claim is called from another coroutine
• /api/remittances/stream mirrors claims behavior
• /api/activity/stream mirrors claims behavior
• heartbeat fires within 15-16s on idle (use 1s test override)
• client disconnect cancels subscription (verify via event_bus._subscribers count)
• malformed event payload doesn't crash the stream
• 3 concurrent subscribers all receive the same new event

Frontend

tail-stream.test.ts (~5 tests):

• parses well-formed NDJSON into typed events
• yields typed error event on {"type":"error"}
• handles heartbeat lines silently
• AbortSignal cancels mid-stream cleanly
• malformed line is skipped, next valid line still emitted

tail-store.test.ts (~5 tests):

• addClaim adds new claim keyed by id
• addClaim with duplicate id is a no-op
• reset('claims') clears only claims slice
• addActivity appends (no stable id)

useTailStream.test.ts (~6 tests):

• on mount: opens stream, status flips to live after snapshot_end
• on stream error: status = error
• on abort: status = closed, no reconnect attempt
• on event arrival: dispatches to tail-store
• reconnect: status cycles reconnecting → connecting → live
• reconnect dedup: items arriving twice only appear once in store

useMergedTail.test.ts (~4 tests):

• merges baseItems + tail-store items by id, baseItems first
• filter predicate drops tail-store items that don't match
• empty tail-store returns baseItems unchanged
• new tail item appears at top

Integration test in Claims.test.tsx (+2 tests):

• live-tail arrival triggers a row in the rendered table
• status badge shows live indicator after stream connects

Total: ~36 new tests.

Smoke

End-to-end after implementation:

1. Start backend (python -m cyclone serve &).
2. Start frontend dev server.
3. Open /claims in one tab.
4. Open Upload page in another tab; submit co_medicaid_837p.txt.
5. Verify the claims appear in the first tab without manual refresh.
6. Verify status pill shows 🟢 live.
7. Disconnect WiFi for 10s; reconnect; verify snapshot replays without duplicates.

9. Risk reminders

• Subscriber slowdown — EventBus drops oldest queued event under backpressure, never blocks publisher. Verified by test_pubsub slow-subscriber case.
• Snapshot replay on reconnect — produces duplicate NDJSON lines, dedup by id in tail-store. Verified by useTailStream.test.ts reconnect-dedup case.
• Tail-store grows unbounded — cap each slice at 10,000 items, evict FIFO. New cap test in tail-store.test.ts.
• HTTP proxy timeouts — heartbeat every 15s keeps idle connections alive; documented in endpoint comment.
• TestClient + async streams — httpx.AsyncClient is required (sync TestClient doesn't support long-lived streaming responses). Backend tests use pytest-asyncio mode.
• Multi-tab duplication — two open /claims tabs both render the same new row when it arrives. Acceptable for v1.
• Filter drift — server always pushes all new items; client filters. If filter is "submitted only" and a "paid" item arrives, it lives in tail-store but isn't rendered until the filter changes. Then it appears.

10. Success criteria

• Open /claims; row count grows visibly when a new batch is uploaded in another tab.
• Disconnect WiFi for 10s; reconnect; table catches up via snapshot replay.
• No console errors during normal operation; reconnect errors are quiet (logged, not surfaced).
• All ~36 new tests pass; full backend + frontend suite still green.
• Smoke: backend live, upload 837, verify row appears in claims table without manual refresh.
• Toolbar shows correct status pill (live / reconnecting / stalled / error) in each scenario.
• Filter changes correctly hide/show tail-arrived items per the active filter.