feat(sp-skill-catalog): add cyclone-tail skill (live-tail wire format)

2026-06-21 12:19:16 -06:00
parent 4bdca6168c
commit 963e608b10
2 changed files with 276 additions and 0 deletions
@@ -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/<resource>/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 `<TailStatusPill>` 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": "<iso-8601>"}` 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 `<TailStatusPill>`. 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
+   `<TailStatusPill>` 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);
+  // <TailStatusPill status={status} lastEventAt={lastEventAt}
+  //   onReconnect={forceReconnect} /> + <Table items={items} />
+}
+```
+
+### 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()`.
+
+### `<TailStatusPill>` 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 `<TailStatusPill>` 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.
@@ -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 `<TailStatusPill>` from `connecting` to `live`, resets the reconnect backoff counter to 0 (`useTailStream.ts:174-180`). |
+| `heartbeat`    | `{"ts": "<iso-8601 UTC>"}`                | 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": "<string>"}`                      | 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": "<string>"}`                 | 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`.