feat(sp25): /api/acks/stream + /api/ta1-acks/stream NDJSON endpoints

Adds two streaming endpoints that match the live-tail wire format
established by /api/claims/stream, /api/remittances/stream, and
/api/activity/stream:

  * /api/acks/stream      — subscribes to ack_received
  * /api/ta1-acks/stream  — subscribes to ta1_ack_received

Both yield a snapshot of existing rows (newest first, capped by the
`limit` query param), then a `snapshot_end` line, then forward
live events from the bus. They are registered before the
/{ack_id} path-param endpoints so the literal `stream` segment
isn't matched as an id.

Tests use the same direct-coroutine pattern as test_api_stream_live.py
because httpx.ASGITransport buffers the response body and never
delivers a disconnect message — iterating body_iterator directly
with body_iterator.aclose() simulates a client disconnect.

backend/src/cyclone/api_routers/acks.py

@@ -1,4 +1,4 @@
-"""``/api/acks`` — list & detail endpoints for the 999 ACK inbox.
+"""``/api/acks`` — list, detail, and live-tail stream for 999 ACKs.
 
 These are the persisted acknowledgment rows produced by
 ``POST /api/parse-999``. The frontend ``useAcks`` hook re-shapes the
@@ -7,18 +7,29 @@ list payload to its ``Ack`` interface in ``src/types/index.ts``.
 The detail endpoint returns the full ``raw_json`` payload plus the
 regenerated ``raw_999_text`` so the UI can show "view source" without a
 second round-trip.
+
+SP25: ``/api/acks/stream`` joins the live-tail triplet — the Acks
+page mounts ``useTailStream("acks")`` and ``useMergedTail("acks", …)``
+to see new 999 acks the moment they land (whether from the SFTP
+poller or a manual upload).
 """
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import Any, AsyncIterator
 
 from fastapi import APIRouter, HTTPException, Query, Request
 from fastapi.responses import StreamingResponse
 
-from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
+from cyclone.api_helpers import (
+    ndjson_line,
+    ndjson_stream_list,
+    tail_events,
+    wants_ndjson,
+)
 from cyclone.parsers.models_999 import ParseResult999
 from cyclone.parsers.serialize_999 import serialize_999
+from cyclone.pubsub import EventBus
 from cyclone.store import store, to_ui_ack
 
 router = APIRouter()
@@ -71,6 +82,35 @@ def list_acks_endpoint(
     }
 
 
+@router.get("/api/acks/stream")
+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.
+
+    SP25: this endpoint joins the live-tail triplet — subscribes to
+    ``ack_received`` and emits one ``item`` per snapshot row plus a
+    single ``snapshot_end`` line, then forwards live events from
+    the bus. Matches the wire format used by ``/api/claims/stream``,
+    ``/api/remittances/stream``, and ``/api/activity/stream``.
+
+    NOTE: registered BEFORE ``/api/acks/{ack_id}`` so the literal
+    ``stream`` path segment doesn't get matched as an ack id.
+    """
+    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")
+
+
 @router.get("/api/acks/{ack_id}")
 def get_ack_endpoint(ack_id: int) -> dict:
     """Return one persisted ACK row with its parsed detail.

backend/src/cyclone/api_routers/ta1_acks.py

@@ -1,4 +1,4 @@
-"""``/api/ta1-acks`` — list & detail endpoints for persisted TA1 envelopes.
+"""``/api/ta1-acks`` — list, detail, and live-tail stream for TA1 envelopes.
 
 TA1 is the interchange-control ACK (ISA/IEA acknowledgement). It's a
 single segment, no functional group, no transaction set. Cyclone
@@ -7,14 +7,21 @@ row can sit alongside the 999 / 277CA ack rows without special-casing.
 
 The detail endpoint also reconstructs the TA1 segment string
 (``TA1*...~``) so the operator can copy it into a downstream tool.
+
+SP25: ``/api/ta1-acks/stream`` joins the live-tail triplet — the
+Acks page mounts ``useTailStream("ta1_acks")`` so the TA1 envelope
+ack section sees new rows the moment they land.
 """
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, AsyncIterator
 
-from fastapi import APIRouter, HTTPException, Query
+from fastapi import APIRouter, HTTPException, Query, Request
+from fastapi.responses import StreamingResponse
 
+from cyclone.api_helpers import ndjson_line, tail_events
 from cyclone import db
+from cyclone.pubsub import EventBus
 from cyclone.store import store, to_ui_ta1_ack
 
 router = APIRouter()
@@ -52,6 +59,30 @@ def list_ta1_acks_endpoint(
     }
 
 
+@router.get("/api/ta1-acks/stream")
+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`` and emits the same wire format
+    as the other live-tail endpoints. Registered BEFORE
+    ``/api/ta1-acks/{ack_id}`` so ``stream`` isn't matched as an id.
+    """
+    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")
+
+
 @router.get("/api/ta1-acks/{ack_id}")
 def get_ta1_ack_endpoint(ack_id: int) -> dict:
     """Return one persisted TA1 ACK row with its parsed detail."""

backend/tests/test_api_ack_stream.py

@@ -0,0 +1,303 @@
+"""Integration tests for /api/acks/stream and /api/ta1-acks/stream.
+
+SP25: both endpoints follow the live-tail wire format used by
+/api/claims/stream, /api/remittances/stream, /api/activity/stream:
+
+  * Snapshot: one ``item`` line per existing row (newest first).
+  * Closing: one ``snapshot_end`` line.
+  * Live: forwarded ``item`` lines for each ``ack_received`` /
+    ``ta1_ack_received`` event published on the bus.
+
+These tests follow the same direct-coroutine pattern as
+``test_api_stream_live.py`` because ``httpx.ASGITransport`` buffers
+the response and never delivers a disconnect message — calling the
+endpoint coroutine with a synthetic ``Request`` lets us iterate the
+body iterator byte-by-byte and use ``body_iterator.aclose()`` to
+simulate a client disconnect.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+from datetime import date
+
+import pytest
+from starlette.requests import Request
+
+from cyclone import db
+from cyclone.api import app
+from cyclone.api_routers.acks import acks_stream
+from cyclone.api_routers.ta1_acks import ta1_acks_stream
+from cyclone.pubsub import EventBus
+from cyclone.store import store
+
+
+# ---------------------------------------------------------------------------
+# Shared helpers — mirror the patterns in test_api_stream_live.py
+# ---------------------------------------------------------------------------
+
+
+def _make_request(path: str = "/api/acks/stream") -> Request:
+    """Synthetic Starlette ``Request`` with a no-op receive channel."""
+
+    async def receive():
+        # Block forever; ``is_disconnected``'s cancel scope will cancel
+        # this await and we'll return None (treated as "no message").
+        await asyncio.Event().wait()
+
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "headers": [],
+        "query_string": b"",
+        "path": path,
+        "app": app,
+        "scheme": "http",
+        "server": ("testserver", 80),
+        "client": ("testclient", 50000),
+    }
+    return Request(scope, receive=receive)
+
+
+def _resolve_query_defaults(endpoint) -> dict:
+    """Resolve FastAPI Query(...) defaults to their inner ``.default``."""
+    import inspect
+    sig = inspect.signature(endpoint)
+    resolved: dict = {}
+    for name, param in sig.parameters.items():
+        default = param.default
+        if hasattr(default, "default"):
+            resolved[name] = default.default
+        else:
+            resolved[name] = default
+    return resolved
+
+
+async def _call_endpoint(endpoint, path: str):
+    request = _make_request(path)
+    defaults = _resolve_query_defaults(endpoint)
+    defaults.pop("request", None)
+    return await endpoint(request, **defaults)
+
+
+async def _read_lines(body_iter, n: int, timeout: float = 5.0) -> list[str]:
+    buffer = b""
+    lines: list[str] = []
+    async with asyncio.timeout(timeout):
+        while len(lines) < n:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                break
+            buffer += chunk
+            while b"\n" in buffer and len(lines) < n:
+                raw, buffer = buffer.split(b"\n", 1)
+                if raw:
+                    lines.append(raw.decode("utf-8"))
+    return lines
+
+
+async def _read_one_line(body_iter, timeout: float = 5.0) -> str | None:
+    buffer = b""
+    async with asyncio.timeout(timeout):
+        while True:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                return None
+            buffer += chunk
+            if b"\n" in buffer:
+                raw, _ = buffer.split(b"\n", 1)
+                if raw:
+                    return raw.decode("utf-8")
+
+
+async def _read_until_type(
+    body_iter, target: str, *, timeout: float = 5.0, max_lines: int = 50
+) -> dict | None:
+    buffer = b""
+    seen = 0
+    async with asyncio.timeout(timeout):
+        while seen < max_lines:
+            try:
+                chunk = await body_iter.__anext__()
+            except StopAsyncIteration:
+                return None
+            buffer += chunk
+            while b"\n" in buffer:
+                raw, buffer = buffer.split(b"\n", 1)
+                if not raw:
+                    continue
+                seen += 1
+                obj = json.loads(raw.decode("utf-8"))
+                if obj.get("type") == target:
+                    return obj
+    return None
+
+
+async def _drain_until_disconnect(body_iter, max_chunks: int = 20) -> None:
+    try:
+        await body_iter.aclose()
+    except (asyncio.CancelledError, GeneratorExit):
+        pass
+
+
+# ---------------------------------------------------------------------------
+# Per-test heartbeat patch — 0.2s so idle generators exit promptly.
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _short_heartbeat(monkeypatch):
+    monkeypatch.setenv("CYCLONE_TAIL_HEARTBEAT_S", "0.2")
+
+
+@pytest.fixture
+def bus() -> EventBus:
+    """Fresh EventBus attached to app.state. Tests publish via ``bus.publish``.
+
+    ``add_ack`` / ``add_ta1_ack`` already publish to whichever bus is
+    passed in, but the live-event tests want to publish synchronously
+    against ``app.state.event_bus`` so the existing-tail subscription
+    picks them up.
+    """
+    new_bus = EventBus(max_queue_size=64)
+    app.state.event_bus = new_bus
+    return new_bus
+
+
+# ---------------------------------------------------------------------------
+# /api/acks/stream
+# ---------------------------------------------------------------------------
+
+
+async def test_acks_stream_snapshots_existing_rows(bus: EventBus):
+    """An existing 999 row surfaces as the first ``item`` line."""
+    with db.SessionLocal()() as s:
+        store.add_ack(
+            source_batch_id="999-SEED-1",
+            accepted_count=1,
+            rejected_count=0,
+            received_count=1,
+            ack_code="A",
+            raw_json={
+                "envelope": {"control_number": "000000001"},
+                "set_responses": [{"set_control_number": "PCN-1"}],
+            },
+            event_bus=bus,
+        )
+
+    response = await _call_endpoint(acks_stream, "/api/acks/stream")
+    assert response.media_type.startswith("application/x-ndjson")
+
+    lines = await _read_lines(response.body_iterator, n=2)
+    items = [json.loads(line) for line in lines]
+    assert items[0]["type"] == "item"
+    assert items[0]["data"]["source_batch_id"] == "999-SEED-1"
+    assert items[0]["data"]["patient_control_number"] == "PCN-1"
+    assert items[1] == {"type": "snapshot_end", "data": {"count": 1}}
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_acks_stream_emits_live_event(bus: EventBus):
+    """A 999 written while the stream is open shows up as a live item."""
+    response = await _call_endpoint(acks_stream, "/api/acks/stream")
+
+    # Drain snapshot (1 snapshot_end line, no prior rows).
+    snap = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap == {"type": "snapshot_end", "data": {"count": 0}}
+
+    # Advance one step so ``subscribe_raw`` registers. Next line may be
+    # a heartbeat — that's fine, we just need the subscription live.
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    with db.SessionLocal()() as s:
+        store.add_ack(
+            source_batch_id="999-LIVE-1",
+            accepted_count=1,
+            rejected_count=0,
+            received_count=1,
+            ack_code="A",
+            raw_json={
+                "envelope": {"control_number": "000000002"},
+                "set_responses": [],
+            },
+            event_bus=bus,
+        )
+
+    obj = await _read_until_type(
+        response.body_iterator, "item", timeout=2.0,
+    )
+    assert obj is not None, "no item line arrived after publish"
+    assert obj["data"]["source_batch_id"] == "999-LIVE-1"
+    await _drain_until_disconnect(response.body_iterator)
+
+
+# ---------------------------------------------------------------------------
+# /api/ta1-acks/stream
+# ---------------------------------------------------------------------------
+
+
+async def test_ta1_acks_stream_snapshots_existing_rows(bus: EventBus):
+    """An existing TA1 row surfaces as the first ``item`` line."""
+    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={"envelope": {"control_number": "000000001"}},
+            event_bus=bus,
+        )
+
+    response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream")
+    assert response.media_type.startswith("application/x-ndjson")
+
+    lines = await _read_lines(response.body_iterator, n=2)
+    items = [json.loads(line) for line in lines]
+    assert items[0]["type"] == "item"
+    assert items[0]["data"]["control_number"] == "000000001"
+    assert items[0]["data"]["interchange_date"] == "2026-07-02"
+    assert items[1] == {"type": "snapshot_end", "data": {"count": 1}}
+    await _drain_until_disconnect(response.body_iterator)
+
+
+async def test_ta1_acks_stream_emits_live_event(bus: EventBus):
+    """A TA1 written while the stream is open shows up as a live item."""
+    response = await _call_endpoint(ta1_acks_stream, "/api/ta1-acks/stream")
+
+    snap = await _read_until_type(
+        response.body_iterator, "snapshot_end", timeout=2.0,
+    )
+    assert snap == {"type": "snapshot_end", "data": {"count": 0}}
+    _ = await _read_one_line(response.body_iterator, timeout=2.0)
+
+    with db.SessionLocal()() as s:
+        store.add_ta1_ack(
+            source_batch_id="TA1-LIVE-1",
+            control_number="000000099",
+            interchange_date=date(2026, 7, 2),
+            interchange_time="1330",
+            ack_code="A",
+            note_code="000",
+            ack_generated_date=None,
+            sender_id="SX",
+            receiver_id="RX",
+            raw_json={"envelope": {"control_number": "000000099"}},
+            event_bus=bus,
+        )
+
+    obj = await _read_until_type(
+        response.body_iterator, "item", timeout=2.0,
+    )
+    assert obj is not None, "no item line arrived after publish"
+    assert obj["data"]["control_number"] == "000000099"
+    assert obj["data"]["sender_id"] == "SX"
+    await _drain_until_disconnect(response.body_iterator)