diff --git a/docs/superpowers/plans/2026-07-02-cyclone-ack-live-tail.md b/docs/superpowers/plans/2026-07-02-cyclone-ack-live-tail.md new file mode 100644 index 0000000..dbceef4 --- /dev/null +++ b/docs/superpowers/plans/2026-07-02-cyclone-ack-live-tail.md @@ -0,0 +1,1856 @@ +# SP25 — Ack Live-Tail Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make the Acks page reflect newly-downloaded 999 / TA1 acks within one heartbeat, by moving the `ack_received` pubsub publish from the handlers into the store (mirroring the existing `claim_written` / `remittance_written` / `activity_recorded` pattern), adding two NDJSON stream endpoints, and wiring the Acks page into the live-tail hook triplet. + +**Architecture:** The store's `add_999_ack` / `add_ta1_ack` / `add_277ca_ack` gain an `event_bus` parameter and call `_sync_publish` after the row commits, publishing `ack_received` / `ta1_ack_received` / `two77ca_ack_received` with the full `to_ui_*` row shape. Two new stream endpoints (`/api/acks/stream`, `/api/ta1-acks/stream`) subscribe to those events and emit NDJSON snapshots + live items. The Acks page mounts `useTailStream("acks")` and `useMergedTail("acks", data?.items ?? [])` for the 999 register, and the same triplet for `"ta1_acks"` in the TA1 section. The `` is dropped into the hero next to the existing "Envelope accepted" badge. + +**Tech Stack:** Python 3.11+, FastAPI, SQLAlchemy, asyncio, Pydantic; React 18, TypeScript, TanStack Query, Zustand. Tests: pytest (backend), vitest + happy-dom + @testing-library/react (frontend). + +**Spec:** [`docs/superpowers/specs/2026-07-02-cyclone-ack-live-tail-design.md`](../specs/2026-07-02-cyclone-ack-live-tail-design.md) + +--- + +## File structure + +``` +backend/src/cyclone/ + store/ + ui.py [MODIFY] add to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack + acks.py [MODIFY] add_999_ack / add_ta1_ack / add_277ca_ack take event_bus, _sync_publish + __init__.py [MODIFY] re-export to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack + api_routers/ + acks.py [MODIFY] import to_ui_ack from store; drop _ack_to_ui + ta1_acks.py [MODIFY] import to_ui_ta1_ack from store; drop _ta1_to_ui + handlers/ + handle_999.py [MODIFY] drop event_bus kwarg + publish block + TODO docstring + handle_ta1.py [MODIFY] drop event_bus kwarg + publish block + TODO docstring + handle_277ca.py [MODIFY] drop event_bus kwarg + publish block + TODO docstring + handle_835.py [MODIFY] drop event_bus kwarg + publish block + TODO docstring + api.py [MODIFY] /api/acks/stream + /api/ta1-acks/stream; thread event_bus into /api/parse-999, /api/parse-ta1, /api/parse-277ca; move 277CA serializer to store/ui.py + +backend/tests/ + test_store_acks_publish.py [NEW] publish-from-store tests + test_api_ack_stream.py [NEW] /api/acks/stream + /api/ta1-acks/stream integration + +src/ + lib/tail-stream.ts [MODIFY] TailResource += "acks" | "ta1_acks" + store/tail-store.ts [MODIFY] add acks/ta1Acks slices + addAck/addTa1Ack/reset cases + hooks/useTailStream.ts [MODIFY] dispatch += acks/ta1_acks cases + hooks/useMergedTail.ts [MODIFY] slice selectors for the new resources + pages/Acks.tsx [MODIFY] useTailStream + useMergedTail + + +src/store/tail-store.test.ts [NEW] slice tests +src/hooks/useTailStream.test.ts [NEW or EXTEND] dispatch tests +src/hooks/useMergedTail.test.ts [NEW or EXTEND] slice-selector tests +src/pages/Acks.test.tsx [EXTEND] live-tail test cases +``` + +--- + +## Task 0: Worktree + clean baseline + +**Files:** +- Create: `.worktrees/sp25-ack-live-tail/` (worktree) +- Create: branch `sp25-ack-live-tail` off `main` + +- [ ] **Step 1: Create the worktree** + +```bash +cd /home/tyler/dev/cyclone +git worktree add .worktrees/sp25-ack-live-tail -b sp25-ack-live-tail +``` + +- [ ] **Step 2: Symlink the existing venv + node_modules into the worktree** + +The worktree only checks out tracked files. The untracked `.venv/` and `node_modules/` would otherwise need a full reinstall. Symlink the existing copies so the test runs use the same packages: + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +ln -s ../../backend/.venv backend/.venv +ln -s ../../node_modules node_modules +``` + +- [ ] **Step 3: Run the backend baseline test suite** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail/backend +.venv/bin/pytest -q 2>&1 | tail -10 +``` + +Expected: all tests pass (the merge gate; should be ~89 files). If anything fails, stop and report — don't proceed with a broken baseline. + +- [ ] **Step 4: Run the frontend baseline test suite** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm test 2>&1 | tail -20 +``` + +Expected: all vitest tests pass. If anything fails, stop and report. + +- [ ] **Step 5: Run the typecheck + lint** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm run typecheck 2>&1 | tail -5 +npm run lint 2>&1 | tail -5 +``` + +Expected: clean. + +--- + +## Task 1: Move `_ack_to_ui` to `store/ui.py` as `to_ui_ack` + +**Files:** +- Modify: `backend/src/cyclone/store/ui.py` +- Modify: `backend/src/cyclone/store/__init__.py` +- Modify: `backend/src/cyclone/api_routers/acks.py` +- Test: `backend/tests/test_store_ui_acks.py` (new) + +- [ ] **Step 1: Write the failing test** + +Create `backend/tests/test_store_ui_acks.py`: + +```python +"""Tests for the moved-to-store UI serializers for 999/TA1/277CA acks.""" +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest +from sqlalchemy.orm import Session + +from cyclone import db +from cyclone.store.ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack + + +@pytest.fixture +def ack_row() -> db.Ack: + """Insert a 999 ack row and return it.""" + with db.SessionLocal()() as s: + row = db.Ack( + source_batch_id="999-TEST-1", + accepted_count=3, + rejected_count=1, + received_count=4, + ack_code="P", + parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc), + raw_json={ + "envelope": {"control_number": "000000001"}, + "set_responses": [ + {"set_control_number": "PCN-42"}, + ], + }, + ) + s.add(row) + s.commit() + s.refresh(row) + return row + + +def test_to_ui_ack_round_trips_999_fields(ack_row): + body = to_ui_ack(ack_row) + assert body["id"] == ack_row.id + assert body["source_batch_id"] == "999-TEST-1" + assert body["accepted_count"] == 3 + assert body["rejected_count"] == 1 + assert body["received_count"] == 4 + assert body["ack_code"] == "P" + assert body["parsed_at"] == "2026-07-02T12:00:00Z" + # patient_control_number comes from raw_json.set_responses[0] + assert body["patient_control_number"] == "PCN-42" +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_ui_acks.py -v 2>&1 | tail -15 +``` + +Expected: `ImportError: cannot import name 'to_ui_ack' from 'cyclone.store.ui'`. + +- [ ] **Step 3: Implement `to_ui_ack` in `store/ui.py`** + +Append to `backend/src/cyclone/store/ui.py`: + +```python +def to_ui_ack(row: db.Ack) -> dict: + """Map a 999 ``Ack`` ORM row to the wire shape used by ``/api/acks``. + + Field names match the rest of the Cyclone API (snake_case). The + frontend ``useAcks`` hook re-shapes this to the camelCase ``Ack`` + interface in ``src/types/index.ts``. + + Adds ``patient_control_number`` pulled from ``raw_json`` so the + operator can correlate each 999 back to the original claim + batch. Mirrors the previous in-router ``_ack_to_ui`` shape + byte-for-byte so the live-tail event payload matches the list + endpoint. + """ + body = { + "id": row.id, + "source_batch_id": row.source_batch_id, + "accepted_count": row.accepted_count, + "rejected_count": row.rejected_count, + "received_count": row.received_count, + "ack_code": row.ack_code, + "parsed_at": _iso_z(row.parsed_at), + "patient_control_number": None, + } + raw = row.raw_json or {} + try: + set_responses = raw.get("set_responses") or [] + if set_responses: + body["patient_control_number"] = set_responses[0].get("set_control_number") + except (AttributeError, TypeError): + pass + return body + + +def to_ui_ta1_ack(row: db.Ta1Ack) -> dict: + """Map a TA1 ``Ta1Ack`` ORM row to the wire shape used by ``/api/ta1-acks``. + + Mirrors the previous in-router ``_ta1_to_ui`` shape byte-for-byte + so the live-tail event payload matches the list endpoint. + """ + return { + "id": row.id, + "control_number": row.control_number, + "ack_code": row.ack_code, + "note_code": row.note_code, + "interchange_date": row.interchange_date.isoformat() + if row.interchange_date else None, + "interchange_time": row.interchange_time, + "sender_id": row.sender_id, + "receiver_id": row.receiver_id, + "source_batch_id": row.source_batch_id, + "parsed_at": _iso_z(row.parsed_at), + } + + +def to_ui_two77ca_ack(row: db.Two77caAck) -> dict: + """Map a 277CA ``Two77caAck`` ORM row to the wire shape used by + ``/api/277ca-acks``. + + Mirrors the previous in-``api.py`` serializer byte-for-byte so the + live-tail event payload matches the list endpoint. + """ + return { + "id": row.id, + "source_batch_id": row.source_batch_id, + "control_number": row.control_number, + "accepted_count": row.accepted_count, + "rejected_count": row.rejected_count, + "paid_count": row.paid_count, + "pended_count": row.pended_count, + "parsed_at": _iso_z(row.parsed_at), + } +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_ui_acks.py -v 2>&1 | tail -10 +``` + +Expected: PASS. + +- [ ] **Step 5: Re-export from `store/__init__.py`** + +Add to the `from .ui import (...)` block in `backend/src/cyclone/store/__init__.py`: + +```python +from .ui import ( + to_ui_ack, + to_ui_claim_detail, + to_ui_claim_from_orm, + to_ui_provider, + to_ui_remittance_from_orm, + to_ui_remittance_with_adjustments, + to_ui_ta1_ack, + to_ui_two77ca_ack, +) +``` + +- [ ] **Step 6: Update `api_routers/acks.py` to import from the store** + +Replace the local `_ack_to_ui` definition with an import. In `backend/src/cyclone/api_routers/acks.py`: + +```python +from cyclone.store import store, to_ui_ack +``` + +Then update the two call sites (`list_acks_endpoint` and `get_ack_endpoint`) from `_ack_to_ui(r)` to `to_ui_ack(r)`, and remove the `def _ack_to_ui(row)` block entirely. Also remove the now-unused `from cyclone.parsers.serialize_999 import serialize_999` import is still needed for the detail endpoint, so leave it. Remove the unused `from cyclone.parsers.models_999 import ParseResult999` IF nothing else in the file uses it (it does, for the regenerated text — keep). + +Verify by grepping: `grep -n "_ack_to_ui" backend/src/cyclone/api_routers/acks.py` should return zero matches after the edit. + +- [ ] **Step 7: Update `api_routers/ta1_acks.py` to import from the store** + +Same pattern. Replace `def _ta1_to_ui(row)` and `def _serialize_ta1_from_row(row)` usages with the store imports: + +```python +from cyclone.store import store, to_ui_ta1_ack +``` + +Replace `_ta1_to_ui(row)` calls with `to_ui_ta1_ack(row)`. Keep the `_serialize_ta1_from_row` local helper (it reconstructs a TA1 segment, which isn't a UI row shape — different concern, stays here). + +- [ ] **Step 8: Find and move the 277CA serializer from `api.py`** + +In `backend/src/cyclone/api.py`, find the 277CA list endpoint (around line 1033) and its local serializer function. Move the local serializer body to `store/ui.py` as `to_ui_two77ca_ack` (already done in Step 3), and replace the local function with an import. + +Search: +```bash +grep -n "def _two77ca_to_ui\|def _277ca_to_ui\|/api/277ca-acks" backend/src/cyclone/api.py +``` + +Replace any `def _xxx_to_ui(row)` body in `api.py` for the 277CA list endpoint with: + +```python +items = [to_ui_two77ca_ack(r) for r in rows[:limit]] +``` + +And add the import at the top of `api.py`: + +```python +from cyclone.store import store, to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack +``` + +(merge with the existing `from cyclone.store import store` line — single import is fine). + +- [ ] **Step 9: Run the existing acks + ta1-acks tests** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_acks_persists.py backend/tests/test_api_acks.py backend/tests/test_api_ta1_acks.py -v 2>&1 | tail -15 +``` + +(Adjust filenames to match what's on disk — find them with `ls backend/tests/test_api_acks*` and `ls backend/tests/test_api_ta1*`.) + +Expected: all pass (the move was byte-for-byte). + +- [ ] **Step 10: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/store/ui.py \ + backend/src/cyclone/store/__init__.py \ + backend/src/cyclone/api_routers/acks.py \ + backend/src/cyclone/api_routers/ta1_acks.py \ + backend/src/cyclone/api.py \ + backend/tests/test_store_ui_acks.py +git commit -m "feat(sp25): move ack UI serializers to store/ui.py" +``` + +--- + +## Task 2: `add_999_ack` publishes `ack_received` + +**Files:** +- Modify: `backend/src/cyclone/store/acks.py` +- Modify: `backend/src/cyclone/store/__init__.py` (re-export `publish` is not needed; the `add_ack` facade already exists) +- Test: `backend/tests/test_store_acks_publish.py` (new) + +- [ ] **Step 1: Write the failing test** + +Create `backend/tests/test_store_acks_publish.py`: + +```python +"""Tests that the store's ACK add methods publish on the event bus. + +Per SP25, the store is the single write surface; the handlers and +parse endpoints no longer need to publish. The publish is best-effort +— a failed subscriber MUST NOT roll back the persisted row. +""" +from __future__ import annotations + +import asyncio +import json +from datetime import date, datetime, timezone + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.api import app +from cyclone.pubsub import EventBus +from cyclone.store import store +from cyclone.store.ui import to_ui_ack + + +@pytest.fixture +def bus() -> EventBus: + """Fresh EventBus on the app for the duration of one test.""" + new_bus = EventBus(max_queue_size=64) + app.state.event_bus = new_bus + return new_bus + + +def _drain(bus: EventBus, kind: str) -> list[dict]: + """Pull every queued event for ``kind`` off the bus and return them.""" + events: list[dict] = [] + for queue in bus._subscribers.get(kind, []): + while not queue.empty(): + events.append(queue.get_nowait()) + return events + + +def test_add_999_ack_publishes_ack_received(bus: EventBus): + with db.SessionLocal()() as s: + row = store.add_999_ack( + source_batch_id="999-SP25-1", + accepted_count=2, + rejected_count=0, + received_count=2, + ack_code="A", + raw_json={ + "envelope": {"control_number": "000000001"}, + "set_responses": [], + }, + event_bus=bus, + ) + + events = _drain(bus, "ack_received") + assert len(events) == 1 + payload = events[0] + # The event payload MUST match the list-endpoint shape so a + # live-tail consumer sees the same row shape as a JSON fetch. + assert payload["_kind"] == "ack_received" + assert payload["id"] == row.id + assert payload["source_batch_id"] == "999-SP25-1" + assert payload["accepted_count"] == 2 + assert payload["rejected_count"] == 0 + assert payload["received_count"] == 2 + assert payload["ack_code"] == "A" + + +def test_add_999_ack_publish_failure_does_not_rollback(bus: EventBus): + """Best-effort publish: a throwing subscriber MUST NOT roll back the row.""" + + class _Boom: + def __init__(self) -> None: + self.published: list[tuple[str, dict]] = [] + + def publish(self, kind: str, payload: dict) -> None: + # Mimic a misbehaving subscriber that raises. The store + # catches the error and continues. + raise RuntimeError("subscriber exploded") + + boom = _Boom() + with db.SessionLocal()() as s: + row = store.add_999_ack( + source_batch_id="999-SP25-BOOM", + accepted_count=1, + rejected_count=0, + received_count=1, + ack_code="A", + raw_json={}, + event_bus=boom, + ) + + # Row still landed despite the publish failure. + assert store.get_ack(row.id) is not None +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py -v 2>&1 | tail -15 +``` + +Expected: both tests fail (the `event_bus=` kwarg doesn't exist yet on `add_999_ack`). + +- [ ] **Step 3: Modify `add_999_ack` to accept `event_bus` and publish** + +In `backend/src/cyclone/store/acks.py`, update the signature and body: + +```python +def add_999_ack( + *, + source_batch_id: str, + accepted_count: int, + rejected_count: int, + received_count: int, + ack_code: str, + raw_json: dict, + event_bus: "EventBus | None" = None, +) -> db.Ack: + """Persist a 999 ACK row and return it. + + SP25: when ``event_bus`` is provided, publishes one + ``ack_received`` event with the ``to_ui_ack`` row shape after the + row commits. Best-effort — a publish failure never rolls back the + persisted row. Mirrors the existing ``claim_written`` publish in + ``store.write.add_record``. + """ + with db.SessionLocal()() as s: + row = Ack( + source_batch_id=source_batch_id, + accepted_count=accepted_count, + rejected_count=rejected_count, + received_count=received_count, + ack_code=ack_code, + parsed_at=utcnow(), + raw_json=raw_json, + ) + s.add(row) + s.commit() + s.refresh(row) + row_id = row.id + + if event_bus is not None: + try: + # Re-fetch via a fresh session so the payload reflects the + # committed row shape, not the in-session row. + with db.SessionLocal()() as s: + payload = to_ui_ack(s.get(Ack, row_id)) + _safe_publish(event_bus, "ack_received", payload) + except Exception: # noqa: BLE001 + import logging + logging.getLogger(__name__).exception( + "add_999_ack: event publish failed for ack %s", row_id, + ) + + return row +``` + +Add the helper at the top of `acks.py`: + +```python +def _safe_publish(event_bus, kind: str, payload: dict) -> None: + """Best-effort sync publish. Tolerates async-only event buses. + + Real ``EventBus`` instances expose both an async ``publish`` and a + private ``_enqueue_or_drop_oldest`` we can call without awaiting. + Tests sometimes pass a stub object with only a sync ``publish`` — + handle both, and never let a publish failure roll back the row. + """ + # Async real bus: schedule the coroutine without awaiting. The + # event loop is on the same thread as the FastAPI request, so the + # coroutine runs to completion on the next loop tick. + publish = getattr(event_bus, "publish", None) + if publish is not None and callable(publish) and not _is_coroutine_func(publish): + try: + publish(kind, payload) + return + except Exception: + raise + # Fallback: directly enqueue. Mirrors ``EventBus._enqueue_or_drop_oldest``. + queue_getter = getattr(event_bus, "_subscribers", None) + if isinstance(queue_getter, dict): + for queue in list(queue_getter.get(kind, ())): + try: + queue.put_nowait({**payload, "_kind": kind}) + except Exception: # noqa: BLE001 + pass + + +def _is_coroutine_func(fn) -> bool: + import asyncio + return asyncio.iscoroutinefunction(fn) +``` + +Wait — the existing `EventBus.publish` is async. We need a sync wrapper. The store's existing `_sync_publish` (in `store/write.py:235`) does exactly this — it reaches into the private `_subscribers` and `_enqueue_or_drop_oldest`. Reuse it. + +Replace the helper above with an import: + +```python +from .write import _sync_publish +``` + +And call `_sync_publish(event_bus, "ack_received", payload)` directly. If the bus is a stub, `_sync_publish` will silently no-op (no `_subscribers` attr). Update the `_safe_publish` to be a thin wrapper that catches and logs: + +```python +def _safe_publish(event_bus, kind: str, payload: dict) -> None: + """Best-effort publish. Never raises. Mirrors write._publish_events_sync.""" + try: + _sync_publish(event_bus, kind, payload) + except Exception: # noqa: BLE001 + import logging + logging.getLogger(__name__).exception( + "%s: event publish failed", kind, + ) +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py::test_add_999_ack_publishes_ack_received -v 2>&1 | tail -10 +``` + +Expected: PASS. The second test (boom) will still fail at this point because we haven't added the try/except in `_safe_publish` correctly — re-check after running. + +- [ ] **Step 5: Run both publish tests** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py -v 2>&1 | tail -10 +``` + +Expected: both pass. + +- [ ] **Step 6: Update the `CycloneStore.add_ack` facade** + +In `backend/src/cyclone/store/__init__.py`, the `add_ack` facade currently is `def add_ack(self, **kwargs): return add_999_ack(**kwargs)`. The new signature is forward-compatible — `**kwargs` carries `event_bus` through. No change required; verify: + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_acks_persists.py -v 2>&1 | tail -10 +``` + +Expected: existing tests still pass. + +- [ ] **Step 7: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/store/acks.py \ + backend/tests/test_store_acks_publish.py +git commit -m "feat(sp25): add_999_ack publishes ack_received" +``` + +--- + +## Task 3: `add_ta1_ack` publishes `ta1_ack_received` + +**Files:** +- Modify: `backend/src/cyclone/store/acks.py` +- Test: extend `backend/tests/test_store_acks_publish.py` + +- [ ] **Step 1: Write the failing test** + +Add to `backend/tests/test_store_acks_publish.py`: + +```python +def test_add_ta1_ack_publishes_ta1_ack_received(bus: EventBus): + with db.SessionLocal()() as s: + row = store.add_ta1_ack( + source_batch_id="TA1-SP25-1", + control_number="000000001", + interchange_date=date(2026, 7, 2), + interchange_time="1200", + ack_code="A", + note_code="000", + ack_generated_date=None, + sender_id="SENDER", + receiver_id="RECEIVER", + raw_json={"envelope": {"control_number": "000000001"}}, + event_bus=bus, + ) + + events = _drain(bus, "ta1_ack_received") + assert len(events) == 1 + payload = events[0] + assert payload["_kind"] == "ta1_ack_received" + assert payload["id"] == row.id + assert payload["control_number"] == "000000001" + assert payload["ack_code"] == "A" + assert payload["sender_id"] == "SENDER" + assert payload["receiver_id"] == "RECEIVER" + assert payload["interchange_date"] == "2026-07-02" +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py::test_add_ta1_ack_publishes_ta1_ack_received -v 2>&1 | tail -10 +``` + +Expected: fail (event_bus kwarg missing on add_ta1_ack). + +- [ ] **Step 3: Modify `add_ta1_ack` to accept `event_bus` and publish** + +In `backend/src/cyclone/store/acks.py`, update: + +```python +def add_ta1_ack( + *, + source_batch_id: str, + control_number: str, + interchange_date: date | None, + interchange_time: str | None, + ack_code: str, + note_code: str | None, + ack_generated_date: date | None, + sender_id: str, + receiver_id: str, + raw_json: dict, + event_bus: "EventBus | None" = None, +) -> db.Ta1Ack: + """Persist a TA1 row and return it. SP25: publishes ``ta1_ack_received``.""" + with db.SessionLocal()() as s: + row = db.Ta1Ack( + source_batch_id=source_batch_id, + control_number=control_number, + interchange_date=interchange_date, + interchange_time=interchange_time, + ack_code=ack_code, + note_code=note_code, + ack_generated_date=ack_generated_date, + sender_id=sender_id, + receiver_id=receiver_id, + parsed_at=utcnow(), + raw_json=raw_json, + ) + s.add(row) + s.commit() + s.refresh(row) + row_id = row.id + + if event_bus is not None: + try: + with db.SessionLocal()() as s: + payload = to_ui_ta1_ack(s.get(db.Ta1Ack, row_id)) + _safe_publish(event_bus, "ta1_ack_received", payload) + except Exception: # noqa: BLE001 + import logging + logging.getLogger(__name__).exception( + "add_ta1_ack: event publish failed for ta1 %s", row_id, + ) + + return row +``` + +Add the import at the top of `acks.py` (next to the existing `from . import utcnow`): + +```python +from .ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py -v 2>&1 | tail -10 +``` + +Expected: all three pass (the 999 test, the boom test, and the new TA1 test). + +- [ ] **Step 5: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/store/acks.py \ + backend/tests/test_store_acks_publish.py +git commit -m "feat(sp25): add_ta1_ack publishes ta1_ack_received" +``` + +--- + +## Task 4: `add_277ca_ack` publishes `two77ca_ack_received` + +**Files:** +- Modify: `backend/src/cyclone/store/acks.py` +- Test: extend `backend/tests/test_store_acks_publish.py` + +- [ ] **Step 1: Write the failing test** + +```python +def test_add_277ca_ack_publishes_two77ca_ack_received(bus: EventBus): + with db.SessionLocal()() as s: + row = store.add_277ca_ack( + source_batch_id="277CA-SP25-1", + control_number="000000001", + accepted_count=2, + rejected_count=1, + paid_count=0, + pended_count=0, + raw_json={"envelope": {"control_number": "000000001"}, "claim_statuses": []}, + event_bus=bus, + ) + + events = _drain(bus, "two77ca_ack_received") + assert len(events) == 1 + payload = events[0] + assert payload["_kind"] == "two77ca_ack_received" + assert payload["id"] == row.id + assert payload["accepted_count"] == 2 + assert payload["rejected_count"] == 1 +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py::test_add_277ca_ack_publishes_two77ca_ack_received -v 2>&1 | tail -10 +``` + +Expected: fail. + +- [ ] **Step 3: Modify `add_277ca_ack` to accept `event_bus` and publish** + +In `backend/src/cyclone/store/acks.py`, update: + +```python +def add_277ca_ack( + *, + source_batch_id: str, + control_number: str, + accepted_count: int, + rejected_count: int, + paid_count: int, + pended_count: int, + raw_json: dict, + event_bus: "EventBus | None" = None, +) -> db.Two77caAck: + """Persist a 277CA row. SP25: publishes ``two77ca_ack_received``.""" + with db.SessionLocal()() as s: + row = db.Two77caAck( + source_batch_id=source_batch_id, + control_number=control_number, + accepted_count=accepted_count, + rejected_count=rejected_count, + paid_count=paid_count, + pended_count=pended_count, + parsed_at=utcnow(), + raw_json=raw_json, + ) + s.add(row) + s.commit() + s.refresh(row) + row_id = row.id + + if event_bus is not None: + try: + with db.SessionLocal()() as s: + payload = to_ui_two77ca_ack(s.get(db.Two77caAck, row_id)) + _safe_publish(event_bus, "two77ca_ack_received", payload) + except Exception: # noqa: BLE001 + import logging + logging.getLogger(__name__).exception( + "add_277ca_ack: event publish failed for 277ca %s", row_id, + ) + + return row +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py -v 2>&1 | tail -10 +``` + +Expected: all 5 tests pass. + +- [ ] **Step 5: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/store/acks.py \ + backend/tests/test_store_acks_publish.py +git commit -m "feat(sp25): add_277ca_ack publishes two77ca_ack_received" +``` + +--- + +## Task 5: Drop `event_bus` kwarg from the handlers + +**Files:** +- Modify: `backend/src/cyclone/handlers/handle_999.py` +- Modify: `backend/src/cyclone/handlers/handle_ta1.py` +- Modify: `backend/src/cyclone/handlers/handle_277ca.py` +- Modify: `backend/src/cyclone/handlers/handle_835.py` + +The store now owns the publish. The handlers stop accepting `event_bus`. Remove the `event_bus: Optional[object] = None` parameter, the `if event_bus is not None: ...` block, and the `TODO(sp27-task-6)` docstring text from all four handlers. + +- [ ] **Step 1: Update `handle_999.py`** + +Remove: +- The `event_bus: Optional[object] = None` parameter from the signature. +- The `if event_bus is not None:` publish block. +- The `TODO(sp27-task-6): bridge async EventBus.publish → sync caller` comment from the docstring. +- The "Reserved for the FastAPI-endpoint migration in Task 6" sentence. +- The `Optional` import if it's now unused. + +Update the docstring to say: "Parse a 999, apply rejections, persist ack row. The store publishes the ``ack_received`` event (SP25); the handler no longer accepts an event_bus." + +- [ ] **Step 2: Update `handle_ta1.py`** + +Remove the `event_bus: Optional[object] = None` parameter, the `if event_bus is not None: ...` block, the `TODO(sp27-task-6)` comment, the "Reserved for the FastAPI-endpoint migration in Task 6" sentence, and the `Optional` import if unused after the edit. + +The docstring gains this final sentence: "The store publishes the ``ta1_ack_received`` event (SP25); the handler no longer accepts an event_bus." + +- [ ] **Step 3: Update `handle_277ca.py`** + +Same set of removals as Step 1 (the `event_bus` parameter, the publish block, the `TODO(sp27-task-6)` comment, the migration sentence). Replace the docstring's final sentence with: "The store publishes the ``two77ca_ack_received`` event (SP25); the handler no longer accepts an event_bus." + +- [ ] **Step 4: Update `handle_835.py`** + +Same set of removals as Step 1. The 835 handler publishes a `remittance_written` event (via the store's 835 write path), NOT a 277CA-style ack event — but its `event_bus` kwarg was still being accepted and never used, so remove it for consistency with the other three handlers. The docstring's final sentence becomes: "The 835 write path publishes the ``remittance_written`` event through the store (SP21); this handler no longer accepts an event_bus." + +- [ ] **Step 5: Run the handler tests + scheduler tests** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/ -k "handle or scheduler" -v 2>&1 | tail -20 +``` + +Expected: all pass. The `HANDLERS` typing signature (`Callable[[str, str], tuple[str, int]]`) doesn't change because the handlers still take 2 args. + +- [ ] **Step 6: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/handlers/handle_999.py \ + backend/src/cyclone/handlers/handle_ta1.py \ + backend/src/cyclone/handlers/handle_277ca.py \ + backend/src/cyclone/handlers/handle_835.py +git commit -m "feat(sp25): drop event_bus kwarg from handlers (store owns publish)" +``` + +--- + +## Task 6: Thread `event_bus` from the parse endpoints to the store + +**Files:** +- Modify: `backend/src/cyclone/api.py` (3 endpoints: `/api/parse-999`, `/api/parse-ta1`, `/api/parse-277ca`) + +- [ ] **Step 1: Update `/api/parse-999`** + +In `backend/src/cyclone/api.py`, the existing endpoint at line ~697 calls `store.add_ack(...)` without `event_bus=`. Update the call to: + +```python +row = store.add_ack( + source_batch_id=synthetic_id, + accepted_count=accepted, + rejected_count=rejected, + received_count=received, + ack_code=ack_code, + raw_json=json.loads(result.model_dump_json()), + event_bus=request.app.state.event_bus, +) +``` + +- [ ] **Step 2: Update `/api/parse-ta1`** + +Same pattern: add `event_bus=request.app.state.event_bus` to the `store.add_ta1_ack(...)` call. The TA1 endpoint also needs a `request: Request` parameter if it doesn't have one (it does for `/api/parse-999`; check `/api/parse-ta1` at line 824 — it does NOT currently have `request`. Add it as the first parameter.) + +- [ ] **Step 3: Update `/api/parse-277ca`** + +Same pattern: add `request: Request` parameter if missing, thread `event_bus=request.app.state.event_bus` to the `store.add_277ca_ack(...)` call. + +- [ ] **Step 4: Add a test asserting the parse endpoint fires the event** + +Add to `backend/tests/test_store_acks_publish.py`: + +```python +def test_parse_999_endpoint_fires_ack_received_event(bus: EventBus): + client = TestClient(app) + text = (Path(__file__).parent / "fixtures" / "minimal_999.txt").read_text() + resp = client.post( + "/api/parse-999", + files={"file": ("minimal_999.txt", text, "text/plain")}, + ) + assert resp.status_code == 200, resp.text + + events = _drain(bus, "ack_received") + assert len(events) == 1 + assert events[0]["_kind"] == "ack_received" +``` + +(Requires `minimal_999.txt` to exist in `backend/tests/fixtures/`. If it doesn't, copy one from the prodfiles archive first per the `cyclone-tests` skill — drop-in rule.) + +- [ ] **Step 5: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_store_acks_publish.py -v 2>&1 | tail -10 +``` + +Expected: all pass. + +- [ ] **Step 6: Run the full existing acks test suite** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_999.py backend/tests/test_api_parse_persists.py -v 2>&1 | tail -15 +``` + +Expected: all pass (regression check). + +- [ ] **Step 7: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/api.py \ + backend/tests/test_store_acks_publish.py +git commit -m "feat(sp25): thread event_bus from /api/parse-999, parse-ta1, parse-277ca" +``` + +--- + +## Task 7: `/api/acks/stream` endpoint + +**Files:** +- Modify: `backend/src/cyclone/api.py` +- Test: `backend/tests/test_api_ack_stream.py` (new) + +- [ ] **Step 1: Write the failing test** + +Create `backend/tests/test_api_ack_stream.py`: + +```python +"""Integration tests for /api/acks/stream and /api/ta1-acks/stream.""" +from __future__ import annotations + +import asyncio +import json +from datetime import date, datetime, timezone + +import pytest +from fastapi.testclient import TestClient + +from cyclone import db +from cyclone.api import app +from cyclone.pubsub import EventBus +from cyclone.store import store + + +@pytest.fixture +def bus() -> EventBus: + new_bus = EventBus(max_queue_size=64) + app.state.event_bus = new_bus + return new_bus + + +def _read_ndjson(client: TestClient, path: str, n_lines: int) -> list[dict]: + """Read ``n_lines`` NDJSON lines from a streaming endpoint.""" + with client.stream("GET", path, headers={"Accept": "application/x-ndjson"}) as r: + assert r.status_code == 200, r.text + assert r.headers["content-type"] == "application/x-ndjson" + lines: list[dict] = [] + for raw in r.iter_lines(): + if not raw: + continue + lines.append(json.loads(raw)) + if len(lines) >= n_lines: + break + return lines + + +def test_acks_stream_snapshots_existing_rows(bus: EventBus): + # Seed one row. + with db.SessionLocal()() as s: + store.add_999_ack( + source_batch_id="999-SEED-1", + accepted_count=1, rejected_count=0, received_count=1, + ack_code="A", raw_json={}, event_bus=bus, + ) + + client = TestClient(app) + lines = _read_ndjson(client, "/api/acks/stream", 2) + # First: the seeded row. Second: snapshot_end. + assert lines[0]["type"] == "item" + assert lines[0]["data"]["source_batch_id"] == "999-SEED-1" + assert lines[1] == {"type": "snapshot_end", "data": {"count": 1}} + + +def test_ta1_acks_stream_snapshots_existing_rows(bus: EventBus): + with db.SessionLocal()() as s: + store.add_ta1_ack( + source_batch_id="TA1-SEED-1", + control_number="000000001", + interchange_date=date(2026, 7, 2), + interchange_time="1200", + ack_code="A", note_code="000", + ack_generated_date=None, + sender_id="S", receiver_id="R", + raw_json={}, event_bus=bus, + ) + + client = TestClient(app) + lines = _read_ndjson(client, "/api/ta1-acks/stream", 2) + assert lines[0]["type"] == "item" + assert lines[0]["data"]["control_number"] == "000000001" + assert lines[1] == {"type": "snapshot_end", "data": {"count": 1}} +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_ack_stream.py -v 2>&1 | tail -10 +``` + +Expected: 404 (the routes don't exist yet). + +- [ ] **Step 3: Implement the stream endpoints** + +In `backend/src/cyclone/api.py`, add the two new endpoints right next to the existing `claims_stream` endpoint (line ~1741) so the route order stays correct. Mirror the `claims_stream` shape — register BEFORE the `/{ack_id}` detail route: + +```python +@app.get("/api/acks/stream", dependencies=[Depends(matrix_gate)]) +async def acks_stream( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream 999 ACKs as NDJSON: snapshot first, then live events. + + Subscribes to ``ack_received``. Default sort is newest-first + (auto-increment id desc), matching the list endpoint. + + Wire format matches the other three live-tail endpoints + (``/api/claims/stream`` / ``/api/remittances/stream`` / + ``/api/activity/stream``): one ``item`` per row, ``snapshot_end`` + marker, then live items, with a 15s heartbeat cadence. + """ + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.list_acks()[:limit] + for row in rows: + yield _ndjson_line({"type": "item", "data": to_ui_ack(row)}) + yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + async for chunk in _tail_events(request, bus, ["ack_received"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") + + +@app.get("/api/ta1-acks/stream", dependencies=[Depends(matrix_gate)]) +async def ta1_acks_stream( + request: Request, + limit: int = Query(100, ge=1, le=1000), +) -> StreamingResponse: + """Stream TA1 envelope acks as NDJSON. Subscribes to ``ta1_ack_received``.""" + bus: EventBus = request.app.state.event_bus + + async def gen() -> AsyncIterator[bytes]: + rows = store.list_ta1_acks()[:limit] + for row in rows: + yield _ndjson_line({"type": "item", "data": to_ui_ta1_ack(row)}) + yield _ndjson_line({"type": "snapshot_end", "data": {"count": len(rows)}}) + async for chunk in _tail_events(request, bus, ["ta1_ack_received"]): + yield chunk + + return StreamingResponse(gen(), media_type="application/x-ndjson") +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_ack_stream.py -v 2>&1 | tail -10 +``` + +Expected: both pass. + +- [ ] **Step 5: Add a live-event test (snapshot then live event)** + +Add to `backend/tests/test_api_ack_stream.py`: + +```python +def test_acks_stream_emits_live_event(bus: EventBus): + client = TestClient(app) + with client.stream("GET", "/api/acks/stream", headers={"Accept": "application/x-ndjson"}) as r: + assert r.status_code == 200 + # Read snapshot_end first. + first = json.loads(next(r.iter_lines())) + assert first["type"] == "snapshot_end" + + # Publish a live event by calling the store. + with db.SessionLocal()() as s: + store.add_999_ack( + source_batch_id="999-LIVE-1", + accepted_count=1, rejected_count=0, received_count=1, + ack_code="A", raw_json={}, event_bus=bus, + ) + + # The next line is the live event item. + second = json.loads(next(r.iter_lines())) + assert second["type"] == "item" + assert second["data"]["source_batch_id"] == "999-LIVE-1" +``` + +- [ ] **Step 6: Run the live-event test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest backend/tests/test_api_ack_stream.py::test_acks_stream_emits_live_event -v 2>&1 | tail -10 +``` + +Expected: PASS. + +- [ ] **Step 7: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add backend/src/cyclone/api.py \ + backend/tests/test_api_ack_stream.py +git commit -m "feat(sp25): /api/acks/stream + /api/ta1-acks/stream NDJSON endpoints" +``` + +--- + +## Task 8: Frontend — extend `TailResource` type + +**Files:** +- Modify: `src/lib/tail-stream.ts` + +- [ ] **Step 1: Extend the `TailResource` union** + +In `src/lib/tail-stream.ts:29`, change: + +```ts +export type TailResource = "claims" | "remittances" | "activity" | "acks" | "ta1_acks"; +``` + +- [ ] **Step 2: Run the typecheck** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm run typecheck 2>&1 | tail -10 +``` + +Expected: TS errors in `useTailStream.ts` and `useMergedTail.ts` (the dispatch and slice selectors don't cover the new cases yet). That's fine — Tasks 9-10 fix them. + +- [ ] **Step 3: Commit (as part of the TailResource extension — separate commit for clarity)** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add src/lib/tail-stream.ts +git commit -m "feat(sp25): TailResource += acks, ta1_acks" +``` + +--- + +## Task 9: Frontend — extend `useTailStore` with `acks` and `ta1_acks` slices + +**Files:** +- Modify: `src/store/tail-store.ts` +- Test: `src/store/tail-store.test.ts` (new) + +- [ ] **Step 1: Write the failing test** + +Create `src/store/tail-store.test.ts`: + +```ts +// @vitest-environment happy-dom +import { describe, expect, it, beforeEach } from "vitest"; +import { useTailStore } from "./tail-store"; +import type { Ack, Ta1Ack } from "@/types"; + +const ack: Ack = { + id: 1, + sourceBatchId: "999-1", + acceptedCount: 1, + rejectedCount: 0, + receivedCount: 1, + ackCode: "A", + parsedAt: "2026-07-02T12:00:00Z", +}; + +const ta1: Ta1Ack = { + id: 2, + controlNumber: "000000001", + ackCode: "A", + noteCode: null, + interchangeDate: "2026-07-02", + interchangeTime: "1200", + senderId: "S", + receiverId: "R", + sourceBatchId: "TA1-1", + parsedAt: "2026-07-02T12:00:00Z", +}; + +describe("useTailStore — acks slice", () => { + beforeEach(() => { + useTailStore.getState().reset("acks"); + useTailStore.getState().reset("ta1_acks"); + }); + + it("addAck appends in arrival order", () => { + useTailStore.getState().addAck(ack); + expect(useTailStore.getState().acks[1]).toEqual(ack); + expect(useTailStore.getState().ackOrder).toEqual(["1"]); + }); + + it("addAck dedups by id (first write wins)", () => { + useTailStore.getState().addAck(ack); + useTailStore.getState().addAck({ ...ack, ackCode: "R" }); + expect(useTailStore.getState().acks[1]?.ackCode).toBe("A"); + expect(useTailStore.getState().ackOrder).toEqual(["1"]); + }); + + it("addTa1Ack appends in arrival order", () => { + useTailStore.getState().addTa1Ack(ta1); + expect(useTailStore.getState().ta1Acks[2]).toEqual(ta1); + expect(useTailStore.getState().ta1AckOrder).toEqual(["2"]); + }); + + it("reset(acks) clears the slice", () => { + useTailStore.getState().addAck(ack); + useTailStore.getState().reset("acks"); + expect(useTailStore.getState().acks).toEqual({}); + expect(useTailStore.getState().ackOrder).toEqual([]); + }); +}); +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/store/tail-store.test.ts 2>&1 | tail -10 +``` + +Expected: TS / runtime errors on the missing slice. + +- [ ] **Step 3: Implement the slices** + +In `src/store/tail-store.ts`: + +1. Import `Ack` and `Ta1Ack` from `@/types` (replace the existing `Activity, Claim, Remittance` import line): + +```ts +import type { Activity, Ack, Claim, Remittance, Ta1Ack } from "@/types"; +``` + +2. Add to the `TailStore` interface: + +```ts + // --- ACKs (999) ------------------------------------------------------- + acks: Record; + ackOrder: string[]; + // --- TA1 ACKs -------------------------------------------------------- + ta1Acks: Record; + ta1AckOrder: string[]; + + addAck: (a: Ack) => void; + addTa1Ack: (t: Ta1Ack) => void; +``` + +3. Add to the initial state (after `remitOrder: []`): + +```ts + acks: {}, + ackOrder: [], + ta1Acks: {}, + ta1AckOrder: [], +``` + +4. Add `addAck` / `addTa1Ack` setters and the matching `reset` cases. Mirror the existing `addClaim` / `addRemittance` / `reset` patterns exactly. + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/store/tail-store.test.ts 2>&1 | tail -10 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add src/store/tail-store.ts \ + src/store/tail-store.test.ts +git commit -m "feat(sp25): useTailStore adds acks + ta1_acks slices" +``` + +--- + +## Task 10: Frontend — extend `useTailStream` dispatch + +**Files:** +- Modify: `src/hooks/useTailStream.ts` +- Test: `src/hooks/useTailStream.test.ts` (new) + +- [ ] **Step 1: Write the failing test** + +Create `src/hooks/useTailStream.test.ts`: + +```ts +// @vitest-environment happy-dom +import { afterEach, describe, expect, it, vi, beforeEach } from "vitest"; + +vi.mock("@/lib/tail-stream", async () => { + const actual = await vi.importActual( + "@/lib/tail-stream", + ); + return { + ...actual, + streamTail: vi.fn(), + }; +}); + +import { useTailStore } from "@/store/tail-store"; +import { useTailStream } from "./useTailStream"; +import type { Ack, Ta1Ack } from "@/types"; + +const ack: Ack = { + id: "1", + sourceBatchId: "999-1", + acceptedCount: 1, + rejectedCount: 0, + receivedCount: 1, + ackCode: "A", + parsedAt: "2026-07-02T12:00:00Z", +}; + +const ta1: Ta1Ack = { + id: "2", + controlNumber: "000000001", + ackCode: "A", + noteCode: null, + interchangeDate: "2026-07-02", + interchangeTime: "1200", + senderId: "S", + receiverId: "R", + sourceBatchId: "TA1-1", + parsedAt: "2026-07-02T12:00:00Z", +}; + +describe("useTailStream dispatch", () => { + beforeEach(() => { + useTailStore.getState().reset("acks"); + useTailStore.getState().reset("ta1_acks"); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("dispatches acks items into the acks slice", () => { + // Simulate by directly calling the store's addAck — the dispatch + // logic is what we're testing. The hook just wires streamTail to + // the right store setter. + useTailStore.getState().addAck(ack); + expect(useTailStore.getState().acks["1"]).toEqual(ack); + }); + + it("dispatches ta1_acks items into the ta1Acks slice", () => { + useTailStore.getState().addTa1Ack(ta1); + expect(useTailStore.getState().ta1Acks["2"]).toEqual(ta1); + }); +}); +``` + +This is a "smoke" test that exercises the new slice setters. The full hook coverage lives in the integration test on the Acks page (Task 12). + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/hooks/useTailStream.test.ts 2>&1 | tail -10 +``` + +Expected: PASS (because the store slices from Task 9 are already there). The point of this task is to wire `useTailStream` to call them. The real test of that wiring is the Acks page integration in Task 12. + +- [ ] **Step 3: Update `useTailStream` dispatch** + +In `src/hooks/useTailStream.ts:69-79`, extend the `dispatch` switch: + +```ts +function dispatch(resource: TailResource, data: unknown): void { + const store = useTailStore.getState(); + switch (resource) { + case "claims": + store.addClaim(data as Claim); + break; + case "remittances": + store.addRemittance(data as Remittance); + break; + case "activity": + store.addActivity(data as Activity); + break; + case "acks": + store.addAck(data as Ack); + break; + case "ta1_acks": + store.addTa1Ack(data as Ta1Ack); + break; + } +} +``` + +Add `Ack` and `Ta1Ack` to the `@/types` import. + +- [ ] **Step 4: Run the typecheck** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm run typecheck 2>&1 | tail -10 +``` + +Expected: clean. + +- [ ] **Step 5: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add src/hooks/useTailStream.ts \ + src/hooks/useTailStream.test.ts +git commit -m "feat(sp25): useTailStream dispatch for acks + ta1_acks" +``` + +--- + +## Task 11: Frontend — extend `useMergedTail` slice selectors + +**Files:** +- Modify: `src/hooks/useMergedTail.ts` +- Test: `src/hooks/useMergedTail.test.ts` (new) + +- [ ] **Step 1: Write the failing test** + +Create `src/hooks/useMergedTail.test.ts`: + +```ts +// @vitest-environment happy-dom +import { renderHook } from "@testing-library/react"; +import { describe, expect, it, beforeEach } from "vitest"; +import { useTailStore } from "@/store/tail-store"; +import { useMergedTail } from "./useMergedTail"; +import type { Ack, Ta1Ack } from "@/types"; + +const base: Ack[] = [ + { + id: "1", + sourceBatchId: "999-1", + acceptedCount: 1, + rejectedCount: 0, + receivedCount: 1, + ackCode: "A", + parsedAt: "2026-07-02T12:00:00Z", + }, +]; + +const live: Ack = { + id: "2", + sourceBatchId: "999-2", + acceptedCount: 2, + rejectedCount: 0, + receivedCount: 2, + ackCode: "A", + parsedAt: "2026-07-02T12:01:00Z", +}; + +describe("useMergedTail — acks slice", () => { + beforeEach(() => { + useTailStore.getState().reset("acks"); + useTailStore.getState().reset("ta1_acks"); + }); + + it("merges base + tail with dedup", () => { + useTailStore.getState().addAck(live); + const { result } = renderHook(() => useMergedTail("acks", base)); + expect(result.current).toEqual([...base, live]); + }); + + it("dedups tail items already in base", () => { + useTailStore.getState().addAck(base[0]!); + const { result } = renderHook(() => useMergedTail("acks", base)); + expect(result.current).toEqual(base); + }); +}); +``` + +- [ ] **Step 2: Run the test, see it fail** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/hooks/useMergedTail.test.ts 2>&1 | tail -10 +``` + +Expected: TS error on the missing `case "acks"` in the slice switch. + +- [ ] **Step 3: Extend the slice switch in `useMergedTail`** + +In `src/hooks/useMergedTail.ts:30-60`, add two cases: + +```ts + case "acks": { + const out: unknown[] = []; + for (const id of s.ackOrder) { + const v = s.acks[id]; + if (v !== undefined) out.push(v); + } + return out; + } + case "ta1_acks": { + const out: unknown[] = []; + for (const id of s.ta1AckOrder) { + const v = s.ta1Acks[id]; + if (v !== undefined) out.push(v); + } + return out; + } +``` + +- [ ] **Step 4: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/hooks/useMergedTail.test.ts 2>&1 | tail -10 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add src/hooks/useMergedTail.ts \ + src/hooks/useMergedTail.test.ts +git commit -m "feat(sp25): useMergedTail slice selectors for acks + ta1_acks" +``` + +--- + +## Task 12: Frontend — wire `useTailStream` + `useMergedTail` + `` into the Acks page + +**Files:** +- Modify: `src/pages/Acks.tsx` + +- [ ] **Step 1: Add the imports** + +In `src/pages/Acks.tsx`, add at the top: + +```ts +import { useTailStream } from "@/hooks/useTailStream"; +import { useMergedTail } from "@/hooks/useMergedTail"; +import { TailStatusPill } from "@/components/TailStatusPill"; +``` + +- [ ] **Step 2: Wire the 999 stream into the page body** + +In the `Acks` function, after the `useAcks(...)` call: + +```ts + const { status: acksStatus, lastEventAt: acksLastEventAt, forceReconnect: acksReconnect } = + useTailStream("acks"); + const items = useMergedTail("acks", data?.items ?? []); +``` + +Replace the existing `const items = data?.items ?? [];` line with the `useMergedTail` call (the new items array now includes the live tail). + +- [ ] **Step 3: Drop a `` into the hero next to the "Envelope accepted" badge** + +Find the existing `
}` block in the hero (around line 188) and add the pill next to it, inside the same flex container: + +```tsx +
+ +
+ {/* existing badge content */} +
+
+``` + +The pill follows the existing TailStatusPill contract: status ∈ {connecting, live, reconnecting, stalled, error, closed}; lastEventAt is a Date | null; onReconnect is a no-arg callback. + +- [ ] **Step 4: Wire the TA1 stream into `Ta1AcksSection`** + +Find the `Ta1AcksSection` function and add (next to its existing `useTa1Acks` call): + +```ts + const { status: ta1Status, lastEventAt: ta1LastEventAt, forceReconnect: ta1Reconnect } = + useTailStream("ta1_acks"); + const items = useMergedTail("ta1_acks", data?.items ?? []); +``` + +Replace `const items = data?.items ?? [];` with the new `useMergedTail` call. + +Drop a `` into the section header (next to the "Envelope acks" eyebrow), inside the same flex container: + +```tsx +
+ +
+ + Envelope acks +
+
+``` + +- [ ] **Step 5: Run the existing `Acks.test.tsx`** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/pages/Acks.test.tsx 2>&1 | tail -20 +``` + +Expected: all 7 existing tests still pass. The mock for `useTailStream` isn't required for these tests because they don't assert on the pill — but the new code calls the hook, which will try to open a stream. The tests will need a mock for `@/hooks/useTailStream` to prevent the hook from making a real fetch. + +Add at the top of `Acks.test.tsx`: + +```ts +vi.mock("@/hooks/useTailStream", () => ({ + useTailStream: () => ({ + status: "live" as const, + lastEventAt: new Date(), + error: null, + forceReconnect: () => {}, + }), +})); +``` + +This stubs the hook to a stable "live" state for the existing tests, so they pass without trying to open a real stream. + +- [ ] **Step 6: Run the test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/pages/Acks.test.tsx 2>&1 | tail -10 +``` + +Expected: all 7 pass. + +- [ ] **Step 7: Add a live-tail test case to `Acks.test.tsx`** + +Add a new test that mocks `useTailStore` directly to simulate a live event landing after the page renders: + +```ts +it("test_live_ack_event_appears_in_register", async () => { + (api.listAcks as unknown as ReturnType).mockResolvedValue({ + items: [], + total: 0, + returned: 0, + has_more: false, + }); + + const { unmount } = renderIntoContainer(React.createElement(Acks)); + await waitForText("Awaiting first 837 with ack=true"); + + // Simulate a live event landing in the tail store. + const { useTailStore } = await import("@/store/tail-store"); + act(() => { + useTailStore.getState().addAck({ + id: 99, + sourceBatchId: "b-live-1", + acceptedCount: 1, + rejectedCount: 0, + receivedCount: 1, + ackCode: "A", + parsedAt: "2026-07-02T12:00:00Z", + }); + }); + + await waitForText("b-live-1"); + expect(document.body.textContent).toContain("99"); + + unmount(); +}); +``` + +- [ ] **Step 8: Run the new test, see it pass** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npx vitest run src/pages/Acks.test.tsx -v 2>&1 | tail -20 +``` + +Expected: all 8 tests pass. + +- [ ] **Step 9: Commit** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git add src/pages/Acks.tsx \ + src/pages/Acks.test.tsx +git commit -m "feat(sp25): Acks page live-tail subscription (999 + TA1)" +``` + +--- + +## Task 13: Full verification + +- [ ] **Step 1: Run the full backend test suite** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +.venv/bin/pytest -q 2>&1 | tail -10 +``` + +Expected: all tests pass (the merge gate). + +- [ ] **Step 2: Run the full frontend test suite** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm test 2>&1 | tail -10 +``` + +Expected: all tests pass. + +- [ ] **Step 3: Run typecheck + lint + build** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +npm run typecheck 2>&1 | tail -5 +npm run lint 2>&1 | tail -5 +npm run build 2>&1 | tail -10 +``` + +Expected: clean across the board. + +- [ ] **Step 4: Sanity check the wire format manually (optional but recommended)** + +Boot the backend and hit the new stream endpoint with `curl`: + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +CYCLONE_AUTH_DISABLED=1 .venv/bin/python -m cyclone serve & +sleep 2 +curl -N -H "Accept: application/x-ndjson" http://127.0.0.1:8000/api/acks/stream | head -3 +kill %1 +``` + +Expected: NDJSON lines with `{"type":"item",...}` and `{"type":"snapshot_end",...}`. + +--- + +## Task 14: Final review and merge prep + +- [ ] **Step 1: Self-review the diff** + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git log --oneline main..HEAD +git diff main...HEAD --stat +``` + +Expected: a list of `feat(sp25): ...` commits and a clean diff stat. + +- [ ] **Step 2: Run the `review` skill (or `pr-reviewer` subagent) against the branch** + +Per the project workflow, dispatch a code-review subagent on the branch: + +```bash +cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail +git push origin sp25-ack-live-tail # if remote exists +``` + +If a remote exists, open a PR titled `SP25 Ack live-tail` and run the `review` skill on the PR. If no remote, dispatch a `general-purpose` subagent with the prompt: "Review the git diff from main to HEAD on this branch for the SP25 ack live-tail increment. Look for: (a) any place that broke the existing `claim_written` / `remittance_written` / `activity_recorded` publish path; (b) any unsynced event payload vs list-endpoint shape; (c) any auth-gate regression on the new endpoints; (d) any TODO comments that survived." + +- [ ] **Step 3: Address review findings** + +If the reviewer returns issues, fix them as separate `fix(sp25): ...` commits on the same branch. Re-run the affected test suites. + +- [ ] **Step 4: Final merge to main (per the SP-N flow)** + +```bash +cd /home/tyler/dev/cyclone +git checkout main +git merge --no-ff sp25-ack-live-tail -m "merge: SP25 ack live-tail into main" +``` + +No squash, no rebase. The SP-N merge commit IS the audit trail.