Files

T

cyclone d8c03fde3f docs(plan): plan for SP25 ack live-tail

2026-07-02 08:34:54 -06:00

59 KiB

Raw Permalink Blame History

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 <TailStatusPill> 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

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 + <TailStatusPill>

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

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:

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

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

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

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:

"""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

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:

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

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:

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:

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:

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:

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:

items = [to_ui_two77ca_ack(r) for r in rows[:limit]]

And add the import at the top of api.py:

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

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

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:

"""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

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:

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:

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:

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:

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

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

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:

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

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:

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

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:

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):

from .ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack

Step 4: Run the test, see it pass

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

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

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

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:

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

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

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

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

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:

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:

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

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

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

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:

"""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

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:

@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

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:

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

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

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:

export type TailResource = "claims" | "remittances" | "activity" | "acks" | "ta1_acks";

Step 2: Run the typecheck

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)

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:

// @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

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:

Import Ack and Ta1Ack from @/types (replace the existing Activity, Claim, Remittance import line):

import type { Activity, Ack, Claim, Remittance, Ta1Ack } from "@/types";

Add to the TailStore interface:

  // --- ACKs (999) -------------------------------------------------------
  acks: Record<string, Ack>;
  ackOrder: string[];
  // --- TA1 ACKs --------------------------------------------------------
  ta1Acks: Record<string, Ta1Ack>;
  ta1AckOrder: string[];

  addAck: (a: Ack) => void;
  addTa1Ack: (t: Ta1Ack) => void;

Add to the initial state (after remitOrder: []):

  acks: {},
  ackOrder: [],
  ta1Acks: {},
  ta1AckOrder: [],

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

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

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:

// @vitest-environment happy-dom
import { afterEach, describe, expect, it, vi, beforeEach } from "vitest";

vi.mock("@/lib/tail-stream", async () => {
  const actual = await vi.importActual<typeof import("@/lib/tail-stream")>(
    "@/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

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:

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

cd /home/tyler/dev/cyclone/.worktrees/sp25-ack-live-tail
npm run typecheck 2>&1 | tail -10

Expected: clean.

Step 5: Commit

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:

// @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

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:

      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

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

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` + `<TailStatusPill>` into the Acks page

Files:

Modify: src/pages/Acks.tsx
Step 1: Add the imports

In src/pages/Acks.tsx, add at the top:

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:

  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 <TailStatusPill> into the hero next to the "Envelope accepted" badge

Find the existing <div className={cn("inline-flex items-center gap-2 rounded-full ...">} block in the hero (around line 188) and add the pill next to it, inside the same flex container:

            <div className="flex items-center gap-2">
              <TailStatusPill
                status={acksStatus}
                lastEventAt={acksLastEventAt}
                onReconnect={acksReconnect}
              />
              <div className={cn(/* existing badge className */)}>
                {/* existing badge content */}
              </div>
            </div>

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):

  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 <TailStatusPill> into the section header (next to the "Envelope acks" eyebrow), inside the same flex container:

              <div className="flex items-center gap-2">
                <TailStatusPill
                  status={ta1Status}
                  lastEventAt={ta1LastEventAt}
                  onReconnect={ta1Reconnect}
                />
                <div className="eyebrow flex items-center gap-2 mb-2">
                  <span className="inline-block h-px w-6 bg-foreground/20" />
                  Envelope acks
                </div>
              </div>

Step 5: Run the existing Acks.test.tsx

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:

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

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:

it("test_live_ack_event_appears_in_register", async () => {
  (api.listAcks as unknown as ReturnType<typeof vi.fn>).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

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

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

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

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

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:

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

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:

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)

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.

59 KiB Raw Permalink Blame History

SP25 — Ack Live-Tail Implementation Plan

File structure

Task 0: Worktree + clean baseline

Task 1: Move _ack_to_ui to store/ui.py as to_ui_ack

Task 2: add_999_ack publishes ack_received

Task 3: add_ta1_ack publishes ta1_ack_received

Task 4: add_277ca_ack publishes two77ca_ack_received

Task 5: Drop event_bus kwarg from the handlers

Task 6: Thread event_bus from the parse endpoints to the store

Task 7: /api/acks/stream endpoint

Task 8: Frontend — extend TailResource type

Task 9: Frontend — extend useTailStore with acks and ta1_acks slices

Task 10: Frontend — extend useTailStream dispatch

Task 11: Frontend — extend useMergedTail slice selectors

Task 12: Frontend — wire useTailStream + useMergedTail + <TailStatusPill> into the Acks page

Task 13: Full verification

Task 14: Final review and merge prep

59 KiB

Raw Permalink Blame History

Task 1: Move `_ack_to_ui` to `store/ui.py` as `to_ui_ack`

Task 2: `add_999_ack` publishes `ack_received`

Task 3: `add_ta1_ack` publishes `ta1_ack_received`

Task 4: `add_277ca_ack` publishes `two77ca_ack_received`

Task 5: Drop `event_bus` kwarg from the handlers

Task 6: Thread `event_bus` from the parse endpoints to the store

Task 7: `/api/acks/stream` endpoint

Task 8: Frontend — extend `TailResource` type

Task 9: Frontend — extend `useTailStore` with `acks` and `ta1_acks` slices

Task 10: Frontend — extend `useTailStream` dispatch

Task 11: Frontend — extend `useMergedTail` slice selectors

Task 12: Frontend — wire `useTailStream` + `useMergedTail` + `<TailStatusPill>` into the Acks page