# 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`](../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**

```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<string, Ack>;
  ackOrder: string[];
  // --- TA1 ACKs --------------------------------------------------------
  ta1Acks: Record<string, Ta1Ack>;
  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<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**

```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` + `<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:

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

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

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

```tsx
              <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`**

```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<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**

```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.