diff --git a/.superpowers/skills/cyclone-tail/SKILL.md b/.superpowers/skills/cyclone-tail/SKILL.md new file mode 100644 index 0000000..f57554e --- /dev/null +++ b/.superpowers/skills/cyclone-tail/SKILL.md @@ -0,0 +1,200 @@ +--- +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. \ No newline at end of file diff --git a/.superpowers/skills/cyclone-tail/references/wire-format.md b/.superpowers/skills/cyclone-tail/references/wire-format.md new file mode 100644 index 0000000..517244f --- /dev/null +++ b/.superpowers/skills/cyclone-tail/references/wire-format.md @@ -0,0 +1,76 @@ +# Live-tail wire format — field reference + +The Cyclone live-tail NDJSON contract is owned by the frontend parser +(`src/lib/tail-stream.ts:22-44`) and the backend emitter +(`backend/src/cyclone/api_helpers.py:tail_events()` + the three +endpoints in `backend/src/cyclone/api.py:1357-2006`). This file is the +quick reference; the canonical prose lives in the README and the +source-of-truth type definitions. + +## Source + +Wire format excerpt copied verbatim from `README.md:76-94` (the +"Live updates → Wire format" section). The README is the user-facing +exposition; this reference adds the per-line field semantics and the +parser-tolerance rules from the code. + +> Each stream endpoint emits newline-delimited JSON. The first batch +> is the **snapshot** of currently-known rows; after that comes +> **`snapshot_end`** with the count, then the **live** events. +> +> ```json +> {"type":"item","data":{"id":"CLM-1", "...":"..."}} +> {"type":"item","data":{"id":"CLM-2", "...":"..."}} +> {"type":"snapshot_end","data":{"count":2}} +> {"type":"item","data":{"id":"CLM-3", "...":"..."}} ← live +> {"type":"heartbeat","data":{"ts":"2026-06-20T23:17:09Z"}} ← idle keep-alive +> ``` +> +> Lines are `{"type": ..., "data": ...}`; known types are `item`, +> `snapshot_end`, `heartbeat`, and (rare) `item_dropped` / +> `error`. Heartbeats keep the connection alive when nothing is +> happening — clients flip to `stalled` after 30s of total silence +> (heartbeat or otherwise) and surface a **↻ Reconnect** button. + +## Per-line field reference + +| `type` | `data` shape | Required? | Emitted by | Parser behavior | +| -------------- | ----------------------------------------- | --------- | ------------------------------------------- | -------------------------------------------- | +| `item` | resource-specific row (`Claim` / `Remittance` / `Activity`) | yes, in `data` (the per-row envelope) | snapshot loop + `_tail_events` forwarding `claim_written` / `remittance_written` / `activity_recorded` | `dispatch(resource, ev.data)` → `useTailStore.addClaim` / `addRemittance` / `addActivity` (first-write-wins dedup on the id-keyed slices — `tail-store.ts:104,120`). | +| `snapshot_end` | `{"count": N}` (integer ≥ 0) | yes, on every stream | `api.py:1395,1889,1999` (one per stream, after the snapshot loop) | Flips `` from `connecting` to `live`, resets the reconnect backoff counter to 0 (`useTailStream.ts:174-180`). | +| `heartbeat` | `{"ts": ""}` | yes, but only when idle | `_tail_events` in `api_helpers.py:241-245` (cadence from `heartbeat_seconds()`, default 15s) | Re-arms the stall timer (`useTailStream.ts:124-140`); no state change. | +| `item_dropped` | `{"id": ""}` | optional (rare; queue overflow) | EventBus drop-oldest path (per the spec at `docs/superpowers/specs/2026-06-20-cyclone-live-tail-design.md:299`) | Re-arms the stall timer; no state change. The `id` field is informational — the hook does not refetch on drop, the page does (see Spec §3.6). | +| `error` | `{"message": ""}` | optional (server-side failure) | Reserved for future server-emitted errors (currently only thrown client-side) | Hook promotes to a thrown `Error(message)` so the catch block runs the reconnect machinery (`useTailStream.ts:190-194`). | + +## Parser tolerance + +The shared parser at `src/lib/tail-stream.ts` is intentionally +forgiving so a single bad frame doesn't kill the stream: + +- **Trailing `\r`** is stripped per line (`tail-stream.ts:116`) so a + CRLF-terminated stream still parses. +- **Malformed JSON** (`JSON.parse` throws) → `console.warn` + skip the + line; the iterator continues (`tail-stream.ts:122-131`). +- **Unknown `type`** (not in `KNOWN_TYPES`) → `console.warn` + skip + (`tail-stream.ts:141-148`). Adding a new event type is + forward-compatible: old clients see warn lines, new clients see the + typed event. +- **Empty lines** (consecutive `\n`s) → silently skipped + (`tail-stream.ts:118`). +- **No trailing newline** → flushed as a final partial line on + stream close (`tail-stream.ts:154-178`). +- **Abort signal** → iterator exits cleanly without throwing + (`tail-stream.ts:75,98,107`). + +## Endpoint inventory + +| Method | Path | Subscribes to | Default sort | Defined at | +| ------ | ------------------------- | -------------------- | ------------------- | ----------------------------------- | +| GET | `/api/claims/stream` | `claim_written` | `-submission_date` | `backend/src/cyclone/api.py:1357` | +| GET | `/api/remittances/stream` | `remittance_written` | `-received_date` | `backend/src/cyclone/api.py:1858` | +| GET | `/api/activity/stream` | `activity_recorded` | `-timestamp` (limit 50) | `backend/src/cyclone/api.py:1971` | + +All three accept the same query params as their non-streaming +counterparts (`status`, `payer`, `date_from`, …) so a frontend can +swap a one-shot fetch for a tail with no URL surgery. Responses are +`Content-Type: application/x-ndjson`. \ No newline at end of file