--- name: cyclone-tail description: "Cyclone live-tail streaming wire format and the useTailStream / useMergedTail hook triplet. Use when: adding a new streaming list page, changing the wire format, debugging stalled/reconnecting state, or modifying the StatusPill behavior." --- # cyclone-tail Cyclone keeps the Claims, Remittances, and Activity pages live without polling: every store write publishes an internal EventBus event, the page opens a `GET /api//stream` HTTP/1.1 chunked-NDJSON connection, and new rows land in the table the moment they hit the database. This skill codifies the wire format, the hook triplet, and the backoff/stall machinery so additions stay consistent with the three streaming pages already shipped (`Claims`, `Remittances`, `ActivityLog`). ## When to use - **Adding a new streaming page.** Mounting `useTailStream(resource)` on a page — you need the hook triplet shape (initial fetch + `useTailStream` + `useMergedTail`), the `` wiring, and the dedup rules in `useMergedTail`. - **Changing the wire format.** Adding a new event type — update `TailEvent` in `src/lib/tail-stream.ts:22-44`, the dispatch switch in `src/hooks/useTailStream.ts:173-195`, the emitter in `backend/src/cyclone/api_helpers.py:tail_events()`, and `references/wire-format.md`. - **Debugging stalled/reconnecting state.** Confirm whether the backend is heartbeating (`CYCLONE_TAIL_HEARTBEAT_S`, default `15s`) or the stall timer fired (`STALL_TIMEOUT_MS = 30_000` at `src/hooks/useTailStream.ts:53`). - **Tuning heartbeat/stall timing.** Changing the 30s stall threshold or the 15s heartbeat interval — README's "Status pill" + "Knobs" tables need to stay in sync (`README.md:109-129`). ## Conventions 1. **Wire format.** Newline-delimited JSON. Every line is `{"type": ..., "data": ...}`. Known `type` values: `item` (per-row envelope), `snapshot_end` (`{"count": N}` marker after the snapshot), `heartbeat` (`{"ts": ""}` keep-alive), `item_dropped` (`{"id": "..."}` queue-overflow notice), `error` (`{"message": "..."}` promoted to a thrown error by the hook). Defined at `src/lib/tail-stream.ts:22-44`; emitted by `backend/src/cyclone/api.py:1357-2006`. See `references/wire-format.md`. 2. **Hook triplet.** Streaming pages compose three pieces: `useTailStream(resource)` (`src/hooks/useTailStream.ts:80` — opens the stream, drives the backoff/stall state machine, dispatches `item` events into `useTailStore`), `useMergedTail(resource, baseItems, filterFn?)` (`src/hooks/useMergedTail.ts:25` — returns `baseItems + tailSlice` dedup'd against `baseItems`), and a per-resource initial-fetch hook (`useClaims`, `useRemittances`, `useActivity`). The page wires all three — see `src/pages/Claims.tsx:87-89`. 3. **Stall threshold.** 30 seconds of total silence — heartbeats included — flips status to `stalled` and surfaces the `↻ Reconnect` button on ``. Constant: `STALL_TIMEOUT_MS = 30_000` at `src/hooks/useTailStream.ts:53`. Re-armed on every event including `heartbeat` and `item_dropped` (`useTailStream.ts:124-140`). Don't change without updating `README.md:109-123`. 4. **Snapshot first.** Every stream emits the snapshot before any live `item` events, then closes with exactly one `snapshot_end` carrying `{"count": N}`. The hook uses `snapshot_end` to flip `` from `connecting` to `live` and to reset the reconnect backoff counter (`useTailStream.ts:174-180`). 5. **Content-Type.** Stream endpoints respond with `media_type="application/x-ndjson"` — see `backend/src/cyclone/api.py:1401,1894,2005`. Frontend sets `Accept: application/x-ndjson` at `src/lib/tail-stream.ts:67-70`. Never `application/json` for a stream endpoint. 6. **Backoff.** Transient errors retry with `1s → 2s → 4s → 8s → 16s → 30s` capped — `BACKOFF_STEPS_MS` at `src/hooks/useTailStream.ts:48-50`. Counter resets on every `snapshot_end`. 7. **Heartbeat knob.** Idle heartbeat interval is configurable via `CYCLONE_TAIL_HEARTBEAT_S` (default `15`, parsed at call time in `backend/src/cyclone/api_helpers.py:185-198`). Tests override to a small value to keep runtime bounded. ## Patterns ### Page-hook skeleton — `Claims.tsx` Pattern from `src/pages/Claims.tsx:85-90`. Three hooks in order: `useClaims(params)` (initial TanStack Query fetch), `useTailStream("claims")` (opens the stream, owns status state), and `useMergedTail("claims", data?.items ?? [], tailFilterFn)` (combines initial snapshot with live tail, dedup'd by id, filtered by the page's predicate applied AFTER dedup at `useMergedTail.ts:72-74`). ```ts import { useClaims } from "@/hooks/useClaims"; import { useTailStream } from "@/hooks/useTailStream"; import { useMergedTail } from "@/hooks/useMergedTail"; export function ClaimsPage() { const { data } = useClaims({ status: "submitted" }); const { status, lastEventAt, forceReconnect } = useTailStream("claims"); const tailFilterFn = (c: Claim) => c.status === "submitted"; const items = useMergedTail("claims", data?.items ?? [], tailFilterFn); // + } ``` ### Backend `/api/foo/stream` endpoint Pattern from `backend/src/cyclone/api.py:1357-1401`. Register BEFORE `/api/foo/{foo_id}` so the literal `stream` segment doesn't match as an id. Two phases — eager snapshot, then live subscription — wrapped in `StreamingResponse` with `media_type="application/x-ndjson"`. ```python @app.get("/api/foo/stream") async def foo_stream( request: Request, status: str | None = Query(None), limit: int = Query(100, ge=1, le=1000), ) -> StreamingResponse: bus: EventBus = request.app.state.event_bus async def gen() -> AsyncIterator[bytes]: # 1. Snapshot. rows = store.iter_foos(status=status, limit=limit) for row in rows: yield _ndjson_line({"type": "item", "data": row}) yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) # 2. Live subscription + heartbeats. async for chunk in _tail_events(request, bus, ["foo_written"]): yield chunk return StreamingResponse(gen(), media_type="application/x-ndjson") ``` `_ndjson_line` and `_tail_events` live in `backend/src/cyclone/api_helpers.py`; the latter forwards EventBus events as `item` lines and emits `heartbeat` lines on the cadence from `heartbeat_seconds()`. ### `` wiring Pattern from `src/components/TailStatusPill.tsx:22-83`. The pill takes `status` + `lastEventAt` from `useTailStream` plus `forceReconnect` so the `↻ Reconnect` button wires to the hook's `reconnectNonce` bump (aborts and reopens the stream — `useTailStream.ts:99-101`). The button only renders when `status === "stalled" || status === "error"` (`TailStatusPill.tsx:52`). ## Anti-patterns - **Don't hand-roll `fetch` + `ReadableStream` parsing in a page.** All stream consumers go through `streamTail(resource, opts?)` at `src/lib/tail-stream.ts:58`. The parser handles `TextDecoderStream`, newline splitting, malformed-line tolerance (`console.warn` + skip), the `KNOWN_TYPES` allowlist, and abort-on-signal semantics (`tail-stream.ts:75,98,107`). Duplicating it loses all of those. - **Don't change the wire format on one endpoint without updating the others.** The parser (`src/lib/tail-stream.ts`) and the hook dispatch switch (`useTailStream.ts:173-195`) are shared across all three live streams. Adding a new event type means updating `TailEvent`, `KNOWN_TYPES`, the dispatch `case`, the `armStall` re-arm list, and the backend emitter — in that order. - **Don't emit `item` events before `snapshot_end`.** The hook uses `snapshot_end` as the marker to flip `` from `connecting` to `live` and to reset the reconnect backoff counter. Emitting `item`s first lands rows in `useTailStore` but the UI still reads "Connecting" — operators see a flash of stale state. Emit snapshot, then `snapshot_end`, then live (`api.py:1386-1401`). - **Don't change the 30s stall threshold without updating the README "Status pill" table.** The constant (`STALL_TIMEOUT_MS = 30_000` at `useTailStream.ts:53`) is the contract the pill is documented against — a bump needs the matching edit at `README.md:118`. - **Don't add `useTailStream` calls from `useFoo.ts` hooks.** Streaming connections belong on the page (`src/pages/Claims.tsx`, `Remittances.tsx`, `ActivityLog.tsx`). Hoisting into `useFoo` couples the lifecycle to whoever mounts it and breaks one-resource-one-page ownership. ## Related skills - **`cyclone-frontend-page`** — page components live in `src/pages/`; load when adding or refactoring a streaming page to confirm the route + `PageHeader` + table conventions. - **`cyclone-api-router`** — endpoint conventions (`api_routers/` for resource-group routers, `api.py` for the live-tail endpoints); load when adding or changing an HTTP endpoint that surfaces streamed or paginated data. - **`cyclone-store`** — write-path conventions in `store.py` and the pubsub event contract (`claim_written`, `remittance_written`, `activity_recorded`); load when adding a new entity whose writes should fan out to a stream endpoint. - **`cyclone-tests`** — frontend `*.test.tsx` siblings cover `useTailStream`, `useMergedTail`, `TailStatusPill`; backend `test_api_stream_live.py` covers the three live-tail endpoints; load when the increment changes wire-format behavior or adds a streaming hook.