1359b0753a
Brings the SP5 live-tail implementation into main:
Backend
* EventBus (cyclone.pubsub): per-kind fan-out with drop-oldest overflow
* FastAPI lifespan initializes the bus + db once per process
* store.add() publishes claim_written / remittance_written /
activity_recorded events on every batch write
* GET /api/{claims,remittances,activity}/stream: NDJSON snapshot +
live subscription + 15s idle heartbeat
* EventBus.unsubscribe() lets the tail loop release its queue on
client disconnect (no queue leak per open stream)
Frontend
* src/lib/tail-stream.ts: streamTail() async-generator over fetch
* src/store/tail-store.ts: zustand with FIFO cap 10k per slice
* src/hooks/useTailStream.ts: connecting/live/reconnecting/stalled/error/closed
state machine with 1→2→4→8→16→30s backoff
* src/hooks/useMergedTail.ts: base + tail merge with filter
* src/components/TailStatusPill.tsx: badge + Reconnect button
* Claims, Remittances, ActivityLog pages wired to the tail
Tests
* 437 backend tests pass (was 418 before SP5)
* 154 frontend tests pass (was 124)
* npm run typecheck clean
* end-to-end smoke: open /api/claims/stream, POST 837, see new claims
arrive in real time without refresh
# Conflicts:
# src/pages/ActivityLog.tsx
# src/pages/Remittances.test.tsx
# src/pages/Remittances.tsx
754 lines
27 KiB
Markdown
754 lines
27 KiB
Markdown
# 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`. |