diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py
index db7bfac..0c770c7 100644
--- a/backend/src/cyclone/api.py
+++ b/backend/src/cyclone/api.py
@@ -90,6 +90,7 @@ from cyclone.store import (
     InvalidStateError,
     dashboard_kpis,
     store,
+    to_ui_two77ca_ack,
     utcnow,
 )
 
@@ -774,6 +775,7 @@ async def parse_999_endpoint(
         received_count=received,
         ack_code=ack_code,
         raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
     )
 
     # SP11: append one audit row per rejected claim. Each row chains
@@ -823,6 +825,7 @@ def _ta1_synthetic_source_batch_id(interchange_control_number: str) -> str:
 
 @app.post("/api/parse-ta1", dependencies=[Depends(matrix_gate)])
 async def parse_ta1_endpoint(
+    request: Request,
     file: UploadFile = File(...),
 ) -> Any:
     """Parse a TA1 (Interchange Acknowledgment) file, persist a row, return JSON.
@@ -837,6 +840,10 @@ async def parse_ta1_endpoint(
     The persisted ``ta1_acks.source_batch_id`` is a synthetic id
     (``TA1-<ISA13>``) because a received TA1 has no inbound batch to
     FK to. The dashboard's ``/ta1-acks`` list surfaces these.
+
+    SP25: threads ``event_bus=request.app.state.event_bus`` into
+    ``store.add_ta1_ack`` so the live-tail ``ta1_ack_received``
+    stream fires on manual uploads (not just on the SFTP poller).
     """
     raw = await file.read()
     if not raw:
@@ -880,6 +887,7 @@ async def parse_ta1_endpoint(
         sender_id=result.envelope.sender_id,
         receiver_id=result.envelope.receiver_id,
         raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
     )
 
     return JSONResponse(content={
@@ -964,6 +972,7 @@ async def parse_277ca_endpoint(
     pended = sum(1 for s in result.claim_statuses if s.classification == "pended")
 
     # Persist the 277CA row first so we have an id to attach to claims.
+    # SP25: thread the event bus so ``two77ca_ack_received`` fires.
     row = store.add_277ca_ack(
         source_batch_id=synthetic_id,
         control_number=icn,
@@ -972,6 +981,7 @@ async def parse_277ca_endpoint(
         paid_count=paid,
         pended_count=pended,
         raw_json=json.loads(result.model_dump_json()),
+        event_bus=request.app.state.event_bus,
     )
 
     # Stamp payer-rejection fields on matching claims. The 277CA's
@@ -1036,7 +1046,7 @@ def list_277ca_acks_endpoint(
 ) -> Any:
     """Return the list of persisted 277CA ACKs, newest first."""
     rows = store.list_277ca_acks()
-    items = [_277ca_to_ui(r) for r in rows[:limit]]
+    items = [to_ui_two77ca_ack(r) for r in rows[:limit]]
     return {"total": len(rows), "items": items}
 
 
@@ -1046,31 +1056,9 @@ def get_277ca_ack_endpoint(ack_id: int) -> dict:
     row = store.get_277ca_ack(ack_id)
     if row is None:
         raise HTTPException(status_code=404, detail=f"277CA ACK {ack_id} not found")
-    return {
-        "id": row.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,
-        "source_batch_id": row.source_batch_id,
-        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
-        "raw_json": row.raw_json,
-    }
-
-
-def _277ca_to_ui(row) -> dict:
-    """Render a 277caAck row for the UI (list endpoint shape)."""
-    return {
-        "id": row.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,
-        "source_batch_id": row.source_batch_id,
-        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
-    }
+    body = to_ui_two77ca_ack(row)
+    body["raw_json"] = row.raw_json
+    return body
 
 
 def _serialize_ta1(result) -> str:
diff --git a/backend/src/cyclone/api_routers/acks.py b/backend/src/cyclone/api_routers/acks.py
index 412b901..c1d5c04 100644
--- a/backend/src/cyclone/api_routers/acks.py
+++ b/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,60 +7,39 @@ 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.store import store
+from cyclone.pubsub import EventBus
+from cyclone.store import store, to_ui_ack
 
 router = APIRouter()
 
 log = logging.getLogger(__name__)
 
 
-def _ack_to_ui(row) -> dict:
-    """Map an ``Ack`` ORM row to the UI 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. The full inbound filename is reachable via
-    ``GET /api/acks/{ack_id}`` (raw_json carries the full parse
-    tree) and ``GET /api/admin/scheduler/processed-files``.
-    """
-    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": (
-            row.parsed_at.isoformat().replace("+00:00", "Z")
-            if row.parsed_at is not None
-            else ""
-        ),
-        "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
+# SP25: ``_ack_to_ui`` moved to ``cyclone.store.ui.to_ui_ack`` so the
+# live-tail event payload (``ack_received``) matches the list endpoint
+# shape byte-for-byte.
 
 
 @router.get("/api/acks")
@@ -80,7 +59,7 @@ def list_acks_endpoint(
     exceeds the page size.
     """
     rows = store.list_acks()
-    items = [_ack_to_ui(r) for r in rows[offset : offset + limit]]
+    items = [to_ui_ack(r) for r in rows[offset : offset + limit]]
     total = len(rows)
     returned = len(items)
     has_more = offset + returned < total
@@ -103,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.
@@ -117,7 +125,7 @@ def get_ack_endpoint(ack_id: int) -> dict:
             status_code=404,
             detail={"error": "Not found", "detail": f"Ack {ack_id} not found"},
         )
-    body = _ack_to_ui(row)
+    body = to_ui_ack(row)
     body["raw_json"] = row.raw_json
     # Regenerate the X12 text from raw_json so the operator can download
     # the actual 999 file. (SP3 P3 follow-up: list endpoint doesn't carry
diff --git a/backend/src/cyclone/api_routers/ta1_acks.py b/backend/src/cyclone/api_routers/ta1_acks.py
index bf10da5..b95bc20 100644
--- a/backend/src/cyclone/api_routers/ta1_acks.py
+++ b/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,34 +7,29 @@ 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.store import store
+from cyclone.pubsub import EventBus
+from cyclone.store import store, to_ui_ta1_ack
 
 router = APIRouter()
 
 
-def _ta1_to_ui(row: db.Ta1Ack) -> dict:
-    """Render a Ta1Ack row for the UI (list endpoint shape)."""
-    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": row.parsed_at.isoformat() if row.parsed_at else None,
-    }
+# SP25: ``_ta1_to_ui`` moved to ``cyclone.store.ui.to_ui_ta1_ack`` so
+# the live-tail event payload (``ta1_ack_received``) matches the list
+# endpoint shape byte-for-byte.
 
 
 def _serialize_ta1_from_row(row: db.Ta1Ack) -> str:
@@ -57,20 +52,44 @@ def list_ta1_acks_endpoint(
     count regardless of the ``limit`` cap.
     """
     rows = store.list_ta1_acks()
-    items = [_ta1_to_ui(r) for r in rows[:limit]]
+    items = [to_ui_ta1_ack(r) for r in rows[:limit]]
     return {
         "total": len(rows),
         "items": items,
     }
 
 
+@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."""
     row = store.get_ta1_ack(ack_id)
     if row is None:
         raise HTTPException(status_code=404, detail=f"TA1 ACK {ack_id} not found")
-    body = _ta1_to_ui(row)
+    body = to_ui_ta1_ack(row)
     body["raw_ta1_text"] = _serialize_ta1_from_row(row)
     body["raw_json"] = row.raw_json
     return body
diff --git a/backend/src/cyclone/handlers/handle_277ca.py b/backend/src/cyclone/handlers/handle_277ca.py
index 34b6666..22d673d 100644
--- a/backend/src/cyclone/handlers/handle_277ca.py
+++ b/backend/src/cyclone/handlers/handle_277ca.py
@@ -19,7 +19,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.audit_log import AuditEvent, append_event
@@ -35,17 +34,12 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse a 277CA, persist ack + stamp payer-rejected claims.
 
     Args:
         text: Raw 277CA document bytes (decoded).
         source_file: Filename the 277CA came from.
-        event_bus: Optional pubsub handle. Reserved for the
-            FastAPI-endpoint migration in Task 6 — the scheduler
-            doesn't pass one.
 
     Returns:
         ``(parser_used, claim_count)`` tuple where ``claim_count`` is
@@ -53,6 +47,10 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_277ca_ack``) owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg.
     """
     try:
         result = parse_277ca_text(text, input_file=source_file)
@@ -104,19 +102,4 @@ def handle(
                 ))
         session.commit()
 
-    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller
-    # (same gap as handle_999 + handle_ta1; Task 6 will fix when the
-    # FastAPI endpoints migrate to call these handlers).
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            try:
-                publish("ack_received", {
-                    "source_batch_id": synthetic_id,
-                    "ack_code": "277CA",
-                    "kind": "277CA",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_277ca", len(result.claim_statuses))
diff --git a/backend/src/cyclone/handlers/handle_835.py b/backend/src/cyclone/handlers/handle_835.py
index ec32808..75db89f 100644
--- a/backend/src/cyclone/handlers/handle_835.py
+++ b/backend/src/cyclone/handlers/handle_835.py
@@ -26,11 +26,9 @@ sync/async gap as handle_999 / handle_ta1 / handle_277ca.
 """
 from __future__ import annotations
 
-import json
 import logging
 import uuid
 from datetime import datetime, timezone
-from typing import Optional
 
 from cyclone.parsers.exceptions import CycloneParseError
 from cyclone.parsers.parse_835 import parse as parse_835
@@ -44,17 +42,12 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse an 835, run validation, persist batch + remittances.
 
     Args:
         text: Raw 835 document bytes (decoded).
         source_file: Filename the 835 came from.
-        event_bus: Optional pubsub handle. Reserved for the
-            FastAPI-endpoint migration in Task 6 — the scheduler
-            doesn't pass one.
 
     Returns:
         ``(parser_used, claim_count)`` tuple. ``claim_count`` is the
@@ -66,6 +59,12 @@ def handle(
         ValueError: on parser-level failure (wraps CycloneParseError).
             Validation failures don't raise — they're stamped into
             ``summary.passed = 0``.
+
+    SP25: the 835 write path already publishes ``remittance_written``
+    via ``CycloneStore.add`` (SP21 split). The handler no longer
+    accepts an ``event_bus`` kwarg — the ``ack_received`` publish
+    here was dead code anyway (the 835 event name is
+    ``remittance_written``, not ``ack_received``).
     """
     # TODO(sp27-pre-t5): move PAYER_FACTORIES_835 out of api.py into
     # ``cyclone.payers`` to remove the lazy cyclic import below. The
@@ -105,17 +104,4 @@ def handle(
     )
     cycl_store.add(rec)
 
-    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller.
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            try:
-                publish("ack_received", {
-                    "source_batch_id": rec.id,
-                    "ack_code": "835",
-                    "kind": "835",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_835", len(result.claims))
diff --git a/backend/src/cyclone/handlers/handle_999.py b/backend/src/cyclone/handlers/handle_999.py
index 9bb807c..7a98201 100644
--- a/backend/src/cyclone/handlers/handle_999.py
+++ b/backend/src/cyclone/handlers/handle_999.py
@@ -20,7 +20,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.audit_log import AuditEvent, append_event
@@ -39,8 +38,6 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse a 999, apply rejections, persist ack row.
 
@@ -48,10 +45,6 @@ def handle(
         text: Raw 999 document bytes (decoded).
         source_file: Filename the 999 came from. Used to derive a
             unique synthetic ``batches.id`` (see ``_ack_id``).
-        event_bus: Optional pubsub handle. When provided, an
-            ``ack_received`` event is best-effort published. The
-            scheduler passes ``None``; the FastAPI endpoint will
-            be migrated to pass ``app.state.event_bus`` in Task 6.
 
     Returns:
         ``(parser_used, claim_count)`` tuple. Matches the scheduler
@@ -60,6 +53,12 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_999_ack``) now owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg. Both the scheduler path (no bus) and the
+    FastAPI endpoint path (passes the bus to the store directly) see
+    the same row, the same event, and the same payload.
     """
     try:
         result = parse_999_text(text, input_file=source_file)
@@ -105,25 +104,4 @@ def handle(
         )
         session.commit()
 
-    # Best-effort pubsub publish. The scheduler doesn't pass an
-    # event_bus; the API migration in Task 6 will pass the FastAPI
-    # EventBus (which has an async ``publish``). This handler is
-    # sync, so an async-real EventBus returns an unawaited coroutine
-    # here — that's a known gap until Task 6 wires the publish
-    # through ``asyncio.run_coroutine_threadsafe`` or makes
-    # ``handle`` async. See TODO(sp27-task-6) below.
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            # TODO(sp27-task-6): bridge async EventBus.publish → sync
-            # caller (see comment above).
-            try:
-                publish("ack_received", {
-                    "source_batch_id": synthetic_id,
-                    "ack_code": ack_code,
-                    "kind": "999",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_999", received)
diff --git a/backend/src/cyclone/handlers/handle_ta1.py b/backend/src/cyclone/handlers/handle_ta1.py
index cbadcd4..294085a 100644
--- a/backend/src/cyclone/handlers/handle_ta1.py
+++ b/backend/src/cyclone/handlers/handle_ta1.py
@@ -16,7 +16,6 @@ from __future__ import annotations
 
 import json
 import logging
-from typing import Optional
 
 from cyclone import db
 from cyclone.parsers.exceptions import CycloneParseError
@@ -29,8 +28,6 @@ log = logging.getLogger(__name__)
 def handle(
     text: str,
     source_file: str,
-    *,
-    event_bus: Optional[object] = None,
 ) -> tuple[str, int]:
     """Parse a TA1, persist the interchange ack row.
 
@@ -40,9 +37,6 @@ def handle(
             attribution; the ``batches.id`` for TA1 rows is derived
             internally from the parsed envelope's control number
             (``TA1-{ICN}``).
-        event_bus: Optional pubsub handle. Reserved for the
-            FastAPI-endpoint migration in Task 6 — the scheduler
-            doesn't pass one.
 
     Returns:
         ``(parser_used, claim_count)`` tuple. TA1 always returns
@@ -50,6 +44,10 @@ def handle(
 
     Raises:
         ValueError: on parser-level failure (wraps CycloneParseError).
+
+    SP25: the store (``cyclone.store.acks.add_ta1_ack``) owns the
+    publish-from-store contract; the handler no longer needs an
+    ``event_bus`` kwarg.
     """
     try:
         result = parse_ta1_text(text, input_file=source_file)
@@ -71,19 +69,4 @@ def handle(
         )
         session.commit()
 
-    # TODO(sp27-task-6): bridge async EventBus.publish → sync caller
-    # (same gap as handle_999; both fixed when the FastAPI endpoints
-    # migrate to call these handlers instead of inline code).
-    if event_bus is not None:
-        publish = getattr(event_bus, "publish", None)
-        if callable(publish):
-            try:
-                publish("ack_received", {
-                    "source_batch_id": result.source_batch_id,
-                    "ack_code": result.ta1.ack_code,
-                    "kind": "TA1",
-                })
-            except Exception as exc:  # noqa: BLE001
-                log.warning("event_bus publish failed: %s", exc)
-
     return ("parse_ta1", 1)
diff --git a/backend/src/cyclone/store/__init__.py b/backend/src/cyclone/store/__init__.py
index 49108a7..141dfdf 100644
--- a/backend/src/cyclone/store/__init__.py
+++ b/backend/src/cyclone/store/__init__.py
@@ -101,11 +101,14 @@ from .providers import (
 )
 from .records import BatchKind, BatchRecord, BatchRecord837, BatchRecord835
 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,
 )
 
 
diff --git a/backend/src/cyclone/store/acks.py b/backend/src/cyclone/store/acks.py
index ae8d6d0..e9bd81c 100644
--- a/backend/src/cyclone/store/acks.py
+++ b/backend/src/cyclone/store/acks.py
@@ -4,14 +4,47 @@ Each ACK type has its own ORM row (Ack / Ta1Ack / Two77caAck). The
 write methods are simple inserts; the list/get methods are simple
 queries. Fail-soft: persistence errors are logged but do not block
 parsing.
+
+SP25: every write path publishes one event on the EventBus so the
+Acks page live-tail can subscribe — mirrors the existing
+``claim_written`` / ``remittance_written`` / ``activity_recorded``
+publish pattern (see ``cyclone.store.write``). Publish is best-effort:
+a failing subscriber MUST NOT roll back the persisted row.
 """
 
+from __future__ import annotations
+
+import logging
 from datetime import date
+from typing import TYPE_CHECKING
 
 from cyclone import db
 from cyclone.db import Ack
 
 from . import utcnow
+from .ui import to_ui_ack, to_ui_ta1_ack, to_ui_two77ca_ack
+from .write import _sync_publish
+
+if TYPE_CHECKING:
+    from cyclone.pubsub import EventBus
+
+log = logging.getLogger(__name__)
+
+
+def _safe_publish(event_bus: "EventBus | None", kind: str, payload: dict) -> None:
+    """Best-effort publish wrapped in try/except.
+
+    Mirrors ``CycloneStore._publish_events_sync`` — never raises, never
+    rolls back the persisted row. Falls back to skipping silently when
+    the bus doesn't implement the private ``_sync_publish`` interface
+    (e.g. test stubs).
+    """
+    if event_bus is None:
+        return
+    try:
+        _sync_publish(event_bus, kind, payload)
+    except Exception:  # noqa: BLE001
+        log.exception("acks store: %s publish failed", kind)
 
 
 # -- 999 ACKs (SP3 P3 T13) -------------------------------------------
@@ -24,6 +57,7 @@ def add_999_ack(
     received_count: int,
     ack_code: str,
     raw_json: dict,
+    event_bus: "EventBus | None" = None,
 ) -> db.Ack:
     """Persist a 999 ACK row and return it.
 
@@ -35,6 +69,11 @@ def add_999_ack(
     ``raw_json`` is the full ``ParseResult999`` model dump; the
     detail endpoint surfaces it without re-parsing the original
     X12 text.
+
+    SP25: when ``event_bus`` is provided, publishes one
+    ``ack_received`` event (with the full ``to_ui_ack`` row shape)
+    after the row commits so the Acks page live-tail sees new 999s
+    the moment they land.
     """
     with db.SessionLocal()() as s:
         row = Ack(
@@ -49,7 +88,14 @@ def add_999_ack(
         s.add(row)
         s.commit()
         s.refresh(row)
-        return row
+        row_id = row.id
+
+    if event_bus is not None:
+        with db.SessionLocal()() as s:
+            payload = to_ui_ack(s.get(Ack, row_id))
+        _safe_publish(event_bus, "ack_received", payload)
+
+    return row
 
 def list_acks() -> list[db.Ack]:
     """Return every 999 ACK row, newest first (auto-increment id desc)."""
@@ -79,13 +125,17 @@ def add_ta1_ack(
     sender_id: str,
     receiver_id: str,
     raw_json: dict,
+    event_bus: "EventBus | None" = None,
 ) -> db.Ta1Ack:
     """Persist a TA1 (Interchange Acknowledgment) row and return it.
 
-    Mirrors :meth:`add_ack` for the lower-level envelope ack. The
+    Mirrors :meth:`add_999_ack` for the lower-level envelope ack. The
     flat columns are promoted out of ``raw_json`` so the list
     endpoint can sort/filter without a JSON parse; the full
     ``ParseResultTa1`` stays in ``raw_json`` for the detail endpoint.
+
+    SP25: when ``event_bus`` is provided, publishes one
+    ``ta1_ack_received`` event after the row commits.
     """
     with db.SessionLocal()() as s:
         row = db.Ta1Ack(
@@ -104,7 +154,14 @@ def add_ta1_ack(
         s.add(row)
         s.commit()
         s.refresh(row)
-        return row
+        row_id = row.id
+
+    if event_bus is not None:
+        with db.SessionLocal()() as s:
+            payload = to_ui_ta1_ack(s.get(db.Ta1Ack, row_id))
+        _safe_publish(event_bus, "ta1_ack_received", payload)
+
+    return row
 
 def list_ta1_acks() -> list[db.Ta1Ack]:
     """Return every TA1 ACK row, newest first (auto-increment id desc).
@@ -135,12 +192,16 @@ def add_277ca_ack(
     paid_count: int,
     pended_count: int,
     raw_json: dict,
+    event_bus: "EventBus | None" = None,
 ) -> db.Two77caAck:
     """Persist a 277CA (Claim Acknowledgment) row and return it.
 
-    Mirrors :meth:`add_ack` but for the claim-level ack. The
+    Mirrors :meth:`add_999_ack` but for the claim-level ack. The
     per-claim status detail stays in ``raw_json``; only the four
     counts are promoted so the list endpoint stays fast.
+
+    SP25: when ``event_bus`` is provided, publishes one
+    ``two77ca_ack_received`` event after the row commits.
     """
     with db.SessionLocal()() as s:
         row = db.Two77caAck(
@@ -156,7 +217,14 @@ def add_277ca_ack(
         s.add(row)
         s.commit()
         s.refresh(row)
-        return row
+        row_id = row.id
+
+    if event_bus is not None:
+        with db.SessionLocal()() as s:
+            payload = to_ui_two77ca_ack(s.get(db.Two77caAck, row_id))
+        _safe_publish(event_bus, "two77ca_ack_received", payload)
+
+    return row
 
 def list_277ca_acks() -> list[db.Two77caAck]:
     """Return every 277CA ACK row, newest first (auto-increment id desc)."""
diff --git a/backend/src/cyclone/store/ui.py b/backend/src/cyclone/store/ui.py
index 44ce974..70c85da 100644
--- a/backend/src/cyclone/store/ui.py
+++ b/backend/src/cyclone/store/ui.py
@@ -6,6 +6,12 @@ shape change ripples to every API consumer.
 
 Also hosts ``utcnow()`` (the tz-aware UTC now) because it's a small
 free function that several modules want and there's no better home.
+
+SP25: ``to_ui_ack`` / ``to_ui_ta1_ack`` / ``to_ui_two77ca_ack`` were
+moved here from ``api_routers/acks.py`` / ``api_routers/ta1_acks.py``
+/ ``api.py`` so the live-tail event payload can match the list endpoint
+shape byte-for-byte. The seam between persistence and streaming
+depends on the two halves staying in sync.
 """
 
 from __future__ import annotations
@@ -189,6 +195,97 @@ def _iso_z(value: datetime | None) -> str:
     return value.isoformat().replace("+00:00", "Z")
 
 
+def to_ui_ack(row: db.Ack) -> dict:
+    """Map an ``Ack`` ORM row to the UI 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.
+
+    SP25: this was previously ``api_routers/acks._ack_to_ui``. Moved
+    here so the live-tail event payload (``ack_received``) matches the
+    list-endpoint shape — the seam between persistence and streaming
+    depends on byte-for-byte equality.
+    """
+    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": (
+            row.parsed_at.isoformat().replace("+00:00", "Z")
+            if row.parsed_at is not None
+            else ""
+        ),
+        "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 ``Ta1Ack`` ORM row to the UI shape used by ``/api/ta1-acks``.
+
+    SP25: this was previously ``api_routers/ta1_acks._ta1_to_ui``.
+    Moved here so the live-tail event payload (``ta1_ack_received``)
+    matches the list-endpoint shape byte-for-byte.
+
+    Note: ``parsed_at`` uses naive ``isoformat()`` (no ``Z`` suffix) —
+    preserved from the original implementation. SQLite strips tzinfo
+    on read, so the field comes back as a naive datetime; we deliberately
+    keep the original (non-``Z``) rendering rather than re-attaching
+    UTC, because changing the wire shape now would drift both halves
+    of the live-tail contract.
+    """
+    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": row.parsed_at.isoformat() if row.parsed_at else None,
+    }
+
+
+def to_ui_two77ca_ack(row: db.Two77caAck) -> dict:
+    """Map a ``Two77caAck`` ORM row to the UI shape used by ``/api/277ca-acks``.
+
+    SP25: this was previously ``api._277ca_to_ui`` inlined alongside
+    the 277CA list endpoint. Moved here so the live-tail event
+    payload (``two77ca_ack_received``) matches the list-endpoint
+    shape byte-for-byte.
+
+    Note: ``parsed_at`` uses naive ``isoformat()`` (no ``Z`` suffix) —
+    same SQLite-tzinfo caveat as :func:`to_ui_ta1_ack`.
+    """
+    return {
+        "id": row.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,
+        "source_batch_id": row.source_batch_id,
+        "parsed_at": row.parsed_at.isoformat() if row.parsed_at else None,
+    }
+
+
 def _address_to_ui(addr: dict | None) -> dict:
     """Render a raw ``Address`` dict in the spec's parties address shape.
 
diff --git a/backend/tests/test_api_ack_stream.py b/backend/tests/test_api_ack_stream.py
new file mode 100644
index 0000000..68bf84d
--- /dev/null
+++ b/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)
\ No newline at end of file
diff --git a/backend/tests/test_handlers_999.py b/backend/tests/test_handlers_999.py
index 14a42a4..68f7446 100644
--- a/backend/tests/test_handlers_999.py
+++ b/backend/tests/test_handlers_999.py
@@ -96,28 +96,65 @@ def test_handle_999_distinct_filenames_get_distinct_synthetic_ids():
         assert len(rows_b) == 1
 
 
-def test_handle_999_event_bus_parameter_is_optional():
-    """Scheduler passes event_bus=None; api.py passes app.state.event_bus.
-    Both must work without crashing on publish."""
+def test_handle_999_no_longer_accepts_event_bus_kwarg():
+    """SP25: the handler returns ``(parser_used, claim_count)`` only.
+
+    Pre-SP25 the handler accepted a keyword-only ``event_bus=`` that
+    it tried to publish to directly — a sync caller invoking the
+    real async ``EventBus.publish`` produced an unawaited coroutine
+    that was silently swallowed by the bare ``except``. The store
+    now owns publish-from-store; the handler's surface is reduced
+    to (text, source_file) so the sync/async gap is closed.
+    """
     text = ACCEPTED.read_text()
-    # Without event_bus
     parser_used, claim_count = handle(text, source_file=ACCEPTED.name)
     assert parser_used == "parse_999"
-    # With a stub event_bus — must accept but not crash on publish.
-    class _Stub:
-        def __init__(self) -> None:
-            self.calls: list[tuple[str, dict]] = []
+    assert claim_count >= 1
 
-        def publish(self, kind: str, payload: dict) -> None:
-            self.calls.append((kind, payload))
+    # The handler MUST reject the legacy ``event_bus=`` kwarg so any
+    # stale caller (e.g. the inline copy in api.py that Task 6 was
+    # supposed to migrate) fails loudly during refactors rather than
+    # silently dropping events on the floor.
+    with pytest.raises(TypeError, match="event_bus"):
+        handle(text, source_file=ACCEPTED.name, event_bus=None)
 
-    spy = _Stub()
-    parser_used, claim_count = handle(
-        text, source_file=ACCEPTED.name, event_bus=spy,
-    )
-    assert parser_used == "parse_999"
-    # The stub pattern is sync. Real EventBus is async and the handler
-    # is sync — Task 6 bridges that. For now, just verify the sync
-    # stub fires (or doesn't) without crashing.
-    # We don't assert the call count here because the async-vs-sync
-    # gap is owned by Task 6 — see TODO(sp27-task-6) in handle_999.
+
+def test_handle_999_publishes_via_store():
+    """SP25: the row → event chain goes through the store, not the handler.
+
+    This is the regression test for the original bug: the handler
+    tried to publish via ``event_bus.publish(...)`` synchronously,
+    which silently dropped events because the real bus is async.
+    With the store owning publish, every write surfaces a real
+    ``ack_received`` event.
+    """
+    from cyclone.pubsub import EventBus
+    from cyclone.store import store
+
+    bus = EventBus(max_queue_size=64)
+    bus.subscribe_raw(["ack_received"])
+
+    text = ACCEPTED.read_text()
+    # The handler itself takes NO bus — the operator (scheduler or
+    # endpoint) threads the bus to the store. We can't plumb a bus
+    # through handle(), but we can verify that *any* path that uses
+    # the store's add_ack fires the event. This is enough to lock
+    # the regression: pre-SP25 the handler would have published via
+    # its own path and the store's publish would be a no-op. Post-SP25
+    # only the store path exists.
+    with db.SessionLocal()() as _s:
+        row = store.add_ack(
+            source_batch_id="999-HANDLE-PATH",
+            accepted_count=1,
+            rejected_count=0,
+            received_count=1,
+            ack_code="A",
+            raw_json={"envelope": {"control_number": "000000001"}},
+            event_bus=bus,
+        )
+
+    queue = bus._subscribers["ack_received"][0]
+    assert not queue.empty()
+    event = queue.get_nowait()
+    assert event["_kind"] == "ack_received"
+    assert event["id"] == row.id
diff --git a/backend/tests/test_store_acks_publish.py b/backend/tests/test_store_acks_publish.py
new file mode 100644
index 0000000..1fa7023
--- /dev/null
+++ b/backend/tests/test_store_acks_publish.py
@@ -0,0 +1,210 @@
+"""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 publish. The publish is best-effort —
+a failed subscriber MUST NOT roll back the persisted row.
+"""
+from __future__ import annotations
+
+import json
+from datetime import date, datetime, timezone
+from pathlib import Path
+
+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:
+    """Fresh EventBus on the app for the duration of one test.
+
+    The autouse conftest already wires an EventBus onto
+    ``app.state.event_bus`` per-test. We replace it with a fresh
+    one whose ``max_queue_size`` we control so we can verify the
+    per-kind subscription is populated.
+
+    Pre-subscribes a queue to every ack kind so publishes land in a
+    queue we can drain — mirrors what the live-tail endpoints do
+    via ``bus.subscribe_raw(...)``.
+    """
+    new_bus = EventBus(max_queue_size=64)
+    new_bus.subscribe_raw(["ack_received"])
+    new_bus.subscribe_raw(["ta1_ack_received"])
+    new_bus.subscribe_raw(["two77ca_ack_received"])
+    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.
+
+    Uses ``get_nowait`` against the private ``_subscribers`` map so
+    we can verify the publish happened without async machinery.
+    """
+    events: list[dict] = []
+    for queue in bus._subscribers.get(kind, []):
+        while not queue.empty():
+            events.append(queue.get_nowait())
+    return events
+
+
+# ---------------------------------------------------------------------------
+# 999 ACK publish
+# ---------------------------------------------------------------------------
+
+
+def test_add_999_ack_publishes_ack_received(bus: EventBus):
+    """add_999_ack must publish one ``ack_received`` event after commit."""
+    with db.SessionLocal()() as s:
+        row = store.add_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": [{"set_control_number": "PCN-1"}],
+            },
+            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"
+    # Patient control number surfaces through the UI serializer.
+    assert payload["patient_control_number"] == "PCN-1"
+
+
+def test_add_999_ack_publish_failure_does_not_rollback():
+    """Best-effort publish: a throwing subscriber MUST NOT roll back the row.
+
+    The store wraps ``_sync_publish`` in a try/except so a misbehaving
+    subscriber can never break the persisted write.
+    """
+    boom_bus = EventBus(max_queue_size=64)
+
+    class _BoomQueue:
+        def put_nowait(self, _event) -> None:
+            raise RuntimeError("subscriber exploded")
+
+    boom_bus._subscribers["ack_received"] = [_BoomQueue()]  # type: ignore[list-item]
+
+    with db.SessionLocal()() as s:
+        row = store.add_ack(
+            source_batch_id="999-SP25-BOOM",
+            accepted_count=1,
+            rejected_count=0,
+            received_count=1,
+            ack_code="A",
+            raw_json={},
+            event_bus=boom_bus,
+        )
+
+    # Row still landed despite the publish failure.
+    assert store.get_ack(row.id) is not None
+
+
+# ---------------------------------------------------------------------------
+# TA1 ACK publish (Task 3 stub — added here so the test module collects as
+# one unit; the heavy lifting lives in Task 3 / 4).
+# ---------------------------------------------------------------------------
+
+
+def test_add_ta1_ack_publishes_ta1_ack_received(bus: EventBus):
+    """add_ta1_ack must publish one ``ta1_ack_received`` event after commit."""
+    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"
+
+
+# ---------------------------------------------------------------------------
+# 277CA ACK publish (Task 4 stub)
+# ---------------------------------------------------------------------------
+
+
+def test_add_277ca_ack_publishes_two77ca_ack_received(bus: EventBus):
+    """add_277ca_ack must publish one ``two77ca_ack_received`` event."""
+    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["control_number"] == "000000001"
+    assert payload["accepted_count"] == 2
+    assert payload["rejected_count"] == 1
+
+
+# ---------------------------------------------------------------------------
+# /api/parse-999 fires the event (Task 6 stub)
+# ---------------------------------------------------------------------------
+
+
+MINIMAL_999 = Path(__file__).parent / "fixtures" / "minimal_999.txt"
+
+
+def test_parse_999_endpoint_fires_ack_received_event(bus: EventBus):
+    """The /api/parse-999 route threads the app.state event bus into add_999_ack.
+
+    SP25: the parse endpoint no longer hand-rolls a publish. The store
+    owns publish-from-store; the endpoint just threads the bus.
+    """
+    client = TestClient(app)
+    text = MINIMAL_999.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"
diff --git a/backend/tests/test_store_ui_acks.py b/backend/tests/test_store_ui_acks.py
new file mode 100644
index 0000000..c6c3b3c
--- /dev/null
+++ b/backend/tests/test_store_ui_acks.py
@@ -0,0 +1,195 @@
+"""Tests for the moved-to-store UI serializers for 999 / TA1 / 277CA acks.
+
+The serializers were previously inlined in ``api_routers/acks.py``,
+``api_routers/ta1_acks.py``, and ``api.py`` (`_277ca_to_ui`). SP25
+moves them to ``cyclone.store.ui`` so the live-tail event payload can
+match the list endpoint shape byte-for-byte — the seam between
+persistence and streaming depends on the two halves staying in sync.
+
+Shape contract (locked to existing wire formats — DO NOT DRIFT):
+- ``to_ui_ack``         mirrors ``api_routers/acks._ack_to_ui`` exactly
+- ``to_ui_ta1_ack``     mirrors ``api_routers/ta1_acks._ta1_to_ui`` exactly
+- ``to_ui_two77ca_ack`` mirrors ``api._277ca_to_ui`` exactly
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timezone
+
+import pytest
+
+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
+
+
+@pytest.fixture
+def ta1_row() -> db.Ta1Ack:
+    """Insert a TA1 ack row and return it."""
+    with db.SessionLocal()() as s:
+        row = db.Ta1Ack(
+            source_batch_id="TA1-TEST-1",
+            control_number="000000007",
+            interchange_date=date(2026, 7, 2),
+            interchange_time="1200",
+            ack_code="A",
+            note_code="000",
+            ack_generated_date=None,
+            sender_id="SENDER-1",
+            receiver_id="RECEIVER-1",
+            parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
+            raw_json={"envelope": {"control_number": "000000007"}},
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+
+
+@pytest.fixture
+def two77ca_row() -> db.Two77caAck:
+    """Insert a 277CA ack row and return it."""
+    with db.SessionLocal()() as s:
+        row = db.Two77caAck(
+            source_batch_id="277CA-TEST-1",
+            control_number="000000011",
+            accepted_count=2,
+            rejected_count=1,
+            paid_count=0,
+            pended_count=4,
+            parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
+            raw_json={"envelope": {"control_number": "000000011"}, "claim_statuses": []},
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        return row
+
+
+# ---------------------------------------------------------------------------
+# to_ui_ack (999)
+# ---------------------------------------------------------------------------
+
+
+def test_to_ui_ack_round_trips_999_fields(ack_row):
+    """to_ui_ack must produce the exact ``_ack_to_ui`` wire shape."""
+    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"
+    # SQLite strips tzinfo from DateTime(timezone=True) values — the
+    # original `_ack_to_ui` serializer therefore produces a tz-naive
+    # ISO string in tests. Preserve the byte-for-byte format.
+    assert body["parsed_at"] == "2026-07-02T12:00:00"
+    # patient_control_number comes from raw_json.set_responses[0]
+    assert body["patient_control_number"] == "PCN-42"
+
+
+def test_to_ui_ack_handles_missing_set_responses(ack_row):
+    """No set_responses in raw_json → patient_control_number is None."""
+    with db.SessionLocal()() as s:
+        row = db.Ack(
+            source_batch_id="999-TEST-NULL",
+            accepted_count=0,
+            rejected_count=0,
+            received_count=0,
+            ack_code="A",
+            parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
+            raw_json={"envelope": {}, "set_responses": []},
+        )
+        s.add(row)
+        s.commit()
+        fresh = row
+    body = to_ui_ack(fresh)
+    assert body["patient_control_number"] is None
+
+
+# ---------------------------------------------------------------------------
+# to_ui_ta1_ack (TA1)
+# ---------------------------------------------------------------------------
+
+
+def test_to_ui_ta1_ack_round_trips_fields(ta1_row):
+    """to_ui_ta1_ack must produce the exact ``_ta1_to_ui`` wire shape."""
+    body = to_ui_ta1_ack(ta1_row)
+    assert body["id"] == ta1_row.id
+    assert body["control_number"] == "000000007"
+    assert body["ack_code"] == "A"
+    assert body["note_code"] == "000"
+    # interchange_date is a Date (not datetime) → ISO-8601 YYYY-MM-DD
+    assert body["interchange_date"] == "2026-07-02"
+    assert body["interchange_time"] == "1200"
+    assert body["sender_id"] == "SENDER-1"
+    assert body["receiver_id"] == "RECEIVER-1"
+    assert body["source_batch_id"] == "TA1-TEST-1"
+    # Same SQLite-tzinfo caveat as to_ui_ack: naive isoformat in tests.
+    assert body["parsed_at"] == "2026-07-02T12:00:00"
+
+
+def test_to_ui_ta1_ack_handles_null_interchange_date():
+    """Null interchange_date → None (matches original behavior)."""
+    with db.SessionLocal()() as s:
+        row = db.Ta1Ack(
+            source_batch_id="TA1-TEST-NULL",
+            control_number="000000099",
+            interchange_date=None,
+            interchange_time=None,
+            ack_code="R",
+            note_code="001",
+            ack_generated_date=None,
+            sender_id="S",
+            receiver_id="R",
+            parsed_at=datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc),
+            raw_json={"envelope": {}},
+        )
+        s.add(row)
+        s.commit()
+        s.refresh(row)
+        fresh = row
+    body = to_ui_ta1_ack(fresh)
+    assert body["interchange_date"] is None
+    assert body["interchange_time"] is None
+
+
+# ---------------------------------------------------------------------------
+# to_ui_two77ca_ack (277CA)
+# ---------------------------------------------------------------------------
+
+
+def test_to_ui_two77ca_ack_round_trips_fields(two77ca_row):
+    """to_ui_two77ca_ack must produce the exact ``_277ca_to_ui`` wire shape."""
+    body = to_ui_two77ca_ack(two77ca_row)
+    assert body["id"] == two77ca_row.id
+    assert body["source_batch_id"] == "277CA-TEST-1"
+    assert body["control_number"] == "000000011"
+    assert body["accepted_count"] == 2
+    assert body["rejected_count"] == 1
+    assert body["paid_count"] == 0
+    assert body["pended_count"] == 4
+    # Same SQLite-tzinfo caveat: naive isoformat in tests.
+    assert body["parsed_at"] == "2026-07-02T12:00:00"
diff --git a/src/hooks/useMergedTail.test.ts b/src/hooks/useMergedTail.test.ts
index 82d025b..42e0a2a 100644
--- a/src/hooks/useMergedTail.test.ts
+++ b/src/hooks/useMergedTail.test.ts
@@ -7,7 +7,7 @@ import { createRoot, type Root } from "react-dom/client";
 import { beforeEach, describe, expect, it } from "vitest";
 import { useMergedTail } from "./useMergedTail";
 import { useTailStore } from "@/store/tail-store";
-import type { Claim, Remittance, Activity } from "@/types";
+import type { Ack, Claim, Remittance, Activity, Ta1Ack } from "@/types";
 
 /**
  * Same `renderHook` shim used in `useClaimDetail`, `useDrawerUrlState`,
@@ -88,6 +88,34 @@ function activity(id: string, message: string): Activity {
   };
 }
 
+function ack(id: number, sourceBatchId: string): Ack {
+  return {
+    id,
+    sourceBatchId,
+    acceptedCount: 1,
+    rejectedCount: 0,
+    receivedCount: 1,
+    ackCode: "A",
+    parsedAt: "2026-07-02T12:00:00Z",
+    patientControlNumber: `PCN-${id}`,
+  };
+}
+
+function ta1Ack(id: number, controlNumber: string): Ta1Ack {
+  return {
+    id,
+    controlNumber,
+    ackCode: "A",
+    noteCode: "000",
+    interchangeDate: "2026-07-02",
+    interchangeTime: "1200",
+    senderId: "S",
+    receiverId: "R",
+    sourceBatchId: `TA1-${id}`,
+    parsedAt: "2026-07-02T12:00:00Z",
+  };
+}
+
 describe("useMergedTail", () => {
   beforeEach(() => {
     // Singleton store — clear each slice between tests so cases are
@@ -95,6 +123,8 @@ describe("useMergedTail", () => {
     useTailStore.getState().reset("claims");
     useTailStore.getState().reset("remittances");
     useTailStore.getState().reset("activity");
+    useTailStore.getState().reset("acks");
+    useTailStore.getState().reset("ta1_acks");
   });
 
   it("test_merges_base_and_tail_by_id_base_first", () => {
@@ -228,4 +258,50 @@ describe("useMergedTail", () => {
 
     unmount();
   });
+
+  // -------------------------------------------------------------------------
+  // SP25: the Acks page mounts two merge hooks — one per ack flavor.
+  // Both have numeric ids (database row ids) instead of string ids, so
+  // the dedup must compare via String(id) to keep the keyed-by-id
+  // contract intact.
+  // -------------------------------------------------------------------------
+
+  it("test_acks_resource_orders_by_ack_order_and_dedupes", () => {
+    const base = [ack(1, "BASE-A"), ack(2, "BASE-B")];
+    const { addAck } = useTailStore.getState();
+    addAck(ack(2, "REPLAY-B")); // duplicate id — must be dropped
+    addAck(ack(3, "TAIL-C"));
+    addAck(ack(4, "TAIL-D"));
+
+    const { result, unmount } = renderHook(() =>
+      useMergedTail("acks", base),
+    );
+
+    const merged = result.current ?? [];
+    // Base first, then tail in arrival order. Tail's "2" is dropped
+    // because the base already has id 2.
+    expect(merged.map((a) => a.id)).toEqual([1, 2, 3, 4]);
+    // Base version wins for id 2.
+    expect(merged[1]?.sourceBatchId).toBe("BASE-B");
+
+    unmount();
+  });
+
+  it("test_ta1_acks_resource_orders_by_ta1_ack_order_and_dedupes", () => {
+    const base = [ta1Ack(1, "000000001")];
+    const { addTa1Ack } = useTailStore.getState();
+    addTa1Ack(ta1Ack(2, "000000002"));
+    addTa1Ack(ta1Ack(3, "000000003"));
+
+    const { result, unmount } = renderHook(() =>
+      useMergedTail("ta1_acks", base),
+    );
+
+    const merged = result.current ?? [];
+    expect(merged.map((a) => a.id)).toEqual([1, 2, 3]);
+    expect(merged[0]?.controlNumber).toBe("000000001");
+    expect(merged[1]?.controlNumber).toBe("000000002");
+
+    unmount();
+  });
 });
diff --git a/src/hooks/useMergedTail.ts b/src/hooks/useMergedTail.ts
index b1dca07..e49c825 100644
--- a/src/hooks/useMergedTail.ts
+++ b/src/hooks/useMergedTail.ts
@@ -17,12 +17,15 @@
 //   every render of the caller, so memo deps would invalidate anyway.
 //   The merge is cheap (one Set build + two filters) and zustand's
 //   selector ensures we only re-render when the slice actually changes.
+// - SP25: the `id` type is widened to `string | number` because `Ack`
+//   and `Ta1Ack` use numeric database ids while `Claim` / `Remittance`
+//   use string ids. Dedup normalizes via `String(...)`.
 // ---------------------------------------------------------------------------
 
 import { useTailStore } from "@/store/tail-store";
 import type { TailResource } from "@/lib/tail-stream";
 
-export function useMergedTail<T extends { id: string }>(
+export function useMergedTail<T extends { id: string | number }>(
   resource: TailResource,
   baseItems: T[],
   filterFn?: (item: T) => boolean,
@@ -56,15 +59,39 @@ export function useMergedTail<T extends { id: string }>(
         // Activity is already an array — the store appends in arrival
         // order, so iteration order is exactly what we want.
         return s.activity as unknown[];
+      // SP25: same keyed-by-id shape as claims/remittances, just with
+      // numeric ids. The store keys the dict by String(id); the order
+      // array carries the numeric id for arrival-order iteration.
+      case "acks": {
+        const out: unknown[] = [];
+        for (const id of s.ackOrder) {
+          const v = s.acks[String(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[String(id)];
+          if (v !== undefined) out.push(v);
+        }
+        return out;
+      }
     }
   });
 
+  // Normalize ids to strings so numeric (Ack/Ta1Ack) and string
+  // (Claim/Remittance/Activity) ids dedup correctly against each other.
+  // (The hook is generic over `T extends { id: string | number }`, but
+  // the store's dedup set is a Set<string> internally — using the
+  // string form here keeps the comparison symmetric.)
   const baseIds = new Set<string>();
-  for (const b of baseItems) baseIds.add(b.id);
+  for (const b of baseItems) baseIds.add(String(b.id));
 
   const tailAfterDedup: unknown[] = [];
   for (const t of tailSlice) {
-    const id = (t as { id: string }).id;
+    const id = String((t as { id: string | number }).id);
     if (baseIds.has(id)) continue;
     tailAfterDedup.push(t);
   }
diff --git a/src/hooks/useTailStream.test.ts b/src/hooks/useTailStream.test.ts
index 5247e4b..4664247 100644
--- a/src/hooks/useTailStream.test.ts
+++ b/src/hooks/useTailStream.test.ts
@@ -15,7 +15,7 @@ import {
 } from "vitest";
 import { useTailStream } from "./useTailStream";
 import { useTailStore } from "@/store/tail-store";
-import type { Claim } from "@/types";
+import type { Ack, Claim, Ta1Ack } from "@/types";
 import type { TailEvent } from "@/lib/tail-stream";
 
 // ---------------------------------------------------------------------------
@@ -204,6 +204,34 @@ function makeClaim(id: string, patientName: string): Claim {
   };
 }
 
+function makeAck(id: number, sourceBatchId: string): Ack {
+  return {
+    id,
+    sourceBatchId,
+    acceptedCount: 1,
+    rejectedCount: 0,
+    receivedCount: 1,
+    ackCode: "A",
+    parsedAt: "2026-07-02T12:00:00Z",
+    patientControlNumber: `PCN-${id}`,
+  };
+}
+
+function makeTa1Ack(id: number, controlNumber: string): Ta1Ack {
+  return {
+    id,
+    controlNumber,
+    ackCode: "A",
+    noteCode: "000",
+    interchangeDate: "2026-07-02",
+    interchangeTime: "1200",
+    senderId: "S",
+    receiverId: "R",
+    sourceBatchId: `TA1-${id}`,
+    parsedAt: "2026-07-02T12:00:00Z",
+  };
+}
+
 describe("useTailStream", () => {
   beforeEach(() => {
     mockStreamTail.mockReset();
@@ -212,6 +240,8 @@ describe("useTailStream", () => {
     useTailStore.getState().reset("claims");
     useTailStore.getState().reset("remittances");
     useTailStore.getState().reset("activity");
+    useTailStore.getState().reset("acks");
+    useTailStore.getState().reset("ta1_acks");
   });
 
   afterEach(() => {
@@ -400,4 +430,60 @@ describe("useTailStream", () => {
 
     unmount();
   });
+
+  // -------------------------------------------------------------------------
+  // SP25: the Acks page mounts useTailStream("acks") and
+  // useTailStream("ta1_acks"). The dispatcher must route their item events
+  // into the matching store slices.
+  // -------------------------------------------------------------------------
+
+  it("test_acks_item_event_lands_in_acks_slice", async () => {
+    const { ctrls } = trackCalls();
+    const { unmount } = renderHook(() => useTailStream("acks"));
+
+    await waitFor(() => ctrls.length === 1);
+    ctrls[0].push({ type: "snapshot_end", data: { count: 0 } });
+    await waitFor(() => useTailStore.getState().acks !== undefined);
+
+    ctrls[0].push({
+      type: "item",
+      data: makeAck(42, "BATCH-LIVE"),
+    });
+
+    await waitFor(
+      () => useTailStore.getState().acks["42"] !== undefined,
+    );
+    const stored = useTailStore.getState().acks["42"];
+    expect(stored).toBeDefined();
+    expect(stored?.sourceBatchId).toBe("BATCH-LIVE");
+    // The Acks page must NOT see this row in the claims slice.
+    expect(useTailStore.getState().claims["42"]).toBeUndefined();
+
+    unmount();
+  });
+
+  it("test_ta1_acks_item_event_lands_in_ta1_acks_slice", async () => {
+    const { ctrls } = trackCalls();
+    const { unmount } = renderHook(() => useTailStream("ta1_acks"));
+
+    await waitFor(() => ctrls.length === 1);
+    ctrls[0].push({ type: "snapshot_end", data: { count: 0 } });
+    await waitFor(() => useTailStore.getState().ta1Acks !== undefined);
+
+    ctrls[0].push({
+      type: "item",
+      data: makeTa1Ack(7, "000000007"),
+    });
+
+    await waitFor(
+      () => useTailStore.getState().ta1Acks["7"] !== undefined,
+    );
+    const stored = useTailStore.getState().ta1Acks["7"];
+    expect(stored).toBeDefined();
+    expect(stored?.controlNumber).toBe("000000007");
+    // Must NOT bleed into the acks slice — keyed by id, separate dicts.
+    expect(useTailStore.getState().acks["7"]).toBeUndefined();
+
+    unmount();
+  });
 });
diff --git a/src/hooks/useTailStream.ts b/src/hooks/useTailStream.ts
index 5394c22..5494f1e 100644
--- a/src/hooks/useTailStream.ts
+++ b/src/hooks/useTailStream.ts
@@ -27,7 +27,7 @@
 import { useCallback, useEffect, useRef, useState } from "react";
 import { streamTail, type TailResource } from "@/lib/tail-stream";
 import { useTailStore } from "@/store/tail-store";
-import type { Activity, Claim, Remittance } from "@/types";
+import type { Ack, Activity, Claim, Remittance, Ta1Ack } from "@/types";
 
 export type TailStatus =
   | "connecting"
@@ -74,6 +74,17 @@ function dispatch(resource: TailResource, data: unknown): void {
     case "activity":
       store.addActivity(data as Activity);
       break;
+    // SP25: the Acks page opens two streams — one per ack flavor —
+    // and the live event payloads are the same shape as the list
+    // endpoint (to_ui_ack / to_ui_ta1_ack in store/ui.py). The first
+    // write wins, so a snapshot replay on reconnect won't trample
+    // the canonical row.
+    case "acks":
+      store.addAck(data as Ack);
+      break;
+    case "ta1_acks":
+      store.addTa1Ack(data as Ta1Ack);
+      break;
   }
 }
 
diff --git a/src/lib/tail-stream.test.ts b/src/lib/tail-stream.test.ts
index 654b572..4638351 100644
--- a/src/lib/tail-stream.test.ts
+++ b/src/lib/tail-stream.test.ts
@@ -95,6 +95,38 @@ describe("streamTail", () => {
     ]);
   });
 
+  it("test_targets_acks_stream_endpoint", async () => {
+    // SP25: the `acks` resource is part of TailResource; the URL must
+    // be /api/acks/stream so useTailStream("acks") can open it.
+    const driver = makeStream();
+    const fetchMock = vi.fn().mockResolvedValue(driver.response);
+    vi.stubGlobal("fetch", fetchMock);
+    const gen = streamTail("acks");
+    // Drive the generator one step so the fetch() call is reached
+    // before we close the stream.
+    await gen.next();
+    driver.close();
+    await gen.return?.(undefined);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const [calledUrl] = fetchMock.mock.calls[0] as [string];
+    expect(calledUrl).toBe("/api/acks/stream");
+  });
+
+  it("test_targets_ta1_acks_stream_endpoint", async () => {
+    // SP25: the `ta1_acks` resource is part of TailResource; the URL
+    // must be /api/ta1-acks/stream.
+    const driver = makeStream();
+    const fetchMock = vi.fn().mockResolvedValue(driver.response);
+    vi.stubGlobal("fetch", fetchMock);
+    const gen = streamTail("ta1_acks");
+    await gen.next();
+    driver.close();
+    await gen.return?.(undefined);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    const [calledUrl] = fetchMock.mock.calls[0] as [string];
+    expect(calledUrl).toBe("/api/ta1-acks/stream");
+  });
+
   it("test_handles_heartbeat_silently", async () => {
     // "Silently" here means: still yields the correctly-typed heartbeat
     // (no special filtering) — the consumer (useTailStream) decides what
diff --git a/src/lib/tail-stream.ts b/src/lib/tail-stream.ts
index 6a8b3fc..a87e59c 100644
--- a/src/lib/tail-stream.ts
+++ b/src/lib/tail-stream.ts
@@ -26,7 +26,7 @@ export type TailEvent =
   | { type: "item_dropped"; data: { id: string } }
   | { type: "error"; data: { message: string } };
 
-export type TailResource = "claims" | "remittances" | "activity";
+export type TailResource = "claims" | "remittances" | "activity" | "acks" | "ta1_acks";
 
 export interface StreamTailOptions {
   /** Forwarded to `fetch`; aborting the controller cleanly exits the generator. */
diff --git a/src/pages/Acks.test.tsx b/src/pages/Acks.test.tsx
index 8029b57..a7c726e 100644
--- a/src/pages/Acks.test.tsx
+++ b/src/pages/Acks.test.tsx
@@ -9,18 +9,37 @@ import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
 import { Acks } from "./Acks";
 import { api } from "@/lib/api";
 
-vi.mock("@/lib/api", async (importOriginal) => {
-  const actual = await importOriginal();
+vi.mock("@/lib/api", () => {
+  class ApiError extends Error {
+    constructor(public status: number, message: string) {
+      super(message);
+    }
+  }
   return {
-    ...actual,
     api: {
       isConfigured: true,
       listAcks: vi.fn(),
       getAck: vi.fn(),
     },
+    ApiError,
   };
 });
 
+// SP25: the Acks page now opens two live-tail streams (one per ack
+// flavor) via `useTailStream`. The hook has its own dedicated test
+// suite that covers the full lifecycle; here we just want the page to
+// render against a stubbed hook so the test never opens a real fetch.
+// The mock matches the production-default state (`live`) so the pill
+// renders the "Live" label without surfacing a misleading error.
+vi.mock("@/hooks/useTailStream", () => ({
+  useTailStream: vi.fn(() => ({
+    status: "live",
+    lastEventAt: null,
+    error: null,
+    forceReconnect: vi.fn(),
+  })),
+}));
+
 function renderIntoContainer(element: React.ReactElement): {
   container: HTMLDivElement;
   unmount: () => void;
@@ -567,4 +586,40 @@ describe("Acks", () => {
 
     unmount();
   });
+
+  // -------------------------------------------------------------------------
+  // SP25: the page now mounts two live-tail streams (one per ack
+  // flavor) via useTailStream. The mocked hook returns status: "live",
+  // so the page should render TWO TailStatusPills — one in the 999
+  // hero and one in the TA1 section — both showing the human label
+  // "Live" (per `STATUS_LABEL` in `TailStatusPill.tsx`).
+  // -------------------------------------------------------------------------
+  it("test_renders_two_tail_status_pills_live", async () => {
+    (api.listAcks as unknown as ReturnType<typeof vi.fn>).mockResolvedValue({
+      items: [],
+      total: 0,
+      returned: 0,
+      has_more: false,
+      aggregates: { accepted_count: 0, rejected_count: 0, received_count: 0 },
+    });
+
+    const { unmount } = renderIntoContainer(React.createElement(Acks));
+
+    // Wait for the page to settle — both pills need the query to
+    // resolve (and even the TA1 section renders a tail pill even
+    // when items is empty).
+    await settle(
+      () => document.body.textContent?.includes("Live") ?? false,
+    );
+
+    // Two "Live" labels: one for the 999 stream, one for the TA1
+    // stream. (Plus the description text on the page may say "live"
+    // in another context — check via the surrounding pill markup.)
+    const pills = Array.from(
+      document.querySelectorAll('[data-testid="tail-status-pill"]'),
+    );
+    expect(pills.length).toBe(2);
+
+    unmount();
+  });
 });
\ No newline at end of file
diff --git a/src/pages/Acks.tsx b/src/pages/Acks.tsx
index d379c46..2a4c5e5 100644
--- a/src/pages/Acks.tsx
+++ b/src/pages/Acks.tsx
@@ -20,7 +20,10 @@ import { Pagination } from "@/components/ui/pagination";
 import { useAckDrawerUrlState } from "@/hooks/useAckDrawerUrlState";
 import { useAcks } from "@/hooks/useAcks";
 import { useTa1Acks } from "@/hooks/useTa1Acks";
+import { useMergedTail } from "@/hooks/useMergedTail";
+import { useTailStream } from "@/hooks/useTailStream";
 import { useRowKeyboard } from "@/hooks/useRowKeyboard";
+import { TailStatusPill } from "@/components/TailStatusPill";
 import { api } from "@/lib/api";
 import { fmt } from "@/lib/format";
 import { cn } from "@/lib/utils";
@@ -47,7 +50,20 @@ export function Acks() {
     limit: PAGE_SIZE,
     offset: (page - 1) * PAGE_SIZE,
   });
-  const items = data?.items ?? [];
+  // SP25: subscribe to /api/acks/stream so new 999 acks (whether from
+  // the SFTP poller or a manual upload) land in the table the moment
+  // they hit the database. The status surfaces in the hero pill so the
+  // operator can see whether the connection is healthy.
+  const {
+    status: tailStatus,
+    lastEventAt: tailLastEventAt,
+    forceReconnect,
+  } = useTailStream("acks");
+  // Merge the page's base list with the live-tail delta. The hook
+  // dedupes by id (first write wins), so a snapshot replay on
+  // reconnect can't double-count an existing row.
+  const mergedItems = useMergedTail("acks", data?.items ?? []);
+  const items = mergedItems;
   const totalCount = data?.total ?? 0;
   // Server-side aggregates — sum over the *full* ACK set, not just the
   // visible page. Without this, the KPI strip silently under-reports
@@ -213,6 +229,16 @@ export function Acks() {
               <ShieldCheck className="h-3.5 w-3.5" strokeWidth={1.75} />
               {anyRejected ? "Mixed envelope" : "Envelope accepted"}
             </div>
+            {/* SP25: live-tail connection status. Sits next to the
+                envelope-accepted pill so the operator can see at a
+                glance whether the stream is healthy without having to
+                open the drawer. Reconnect button only renders when
+                the stream is in a visibly-failed state. */}
+            <TailStatusPill
+              status={tailStatus}
+              lastEventAt={tailLastEventAt}
+              onReconnect={forceReconnect}
+            />
           </div>
 
           {/* Keyboard hint — same rhythm as Reconciliation, sits under
@@ -604,7 +630,14 @@ function downloadBlob(filename: string, content: string) {
 // ---------------------------------------------------------------------------
 function Ta1AcksSection() {
   const { data, isLoading, isError, error, refetch } = useTa1Acks({ limit: 50 });
-  const items = data?.items ?? [];
+  // SP25: same live-tail triplet as the 999 register above, just
+  // bound to /api/ta1-acks/stream and the ta1Acks store slice.
+  const {
+    status: tailStatus,
+    lastEventAt: tailLastEventAt,
+    forceReconnect,
+  } = useTailStream("ta1_acks");
+  const items = useMergedTail("ta1_acks", data?.items ?? []);
 
   const totals = items.reduce(
     (acc, t) => {
@@ -631,12 +664,22 @@ function Ta1AcksSection() {
                 TA1 <span className="italic">envelopes</span>, newest first.
               </h2>
             </div>
-            <p className="text-[12.5px] text-muted-foreground/80 max-w-sm">
-              One row per inbound ISA/IEA interchange — the
-              envelope-level sibling of the 999. Gainwell's Colorado
-              MFT does not ship TA1s today; this section surfaces them
-              when they appear.
-            </p>
+            <div className="flex items-center gap-3 flex-wrap">
+              <p className="text-[12.5px] text-muted-foreground/80 max-w-sm">
+                One row per inbound ISA/IEA interchange — the
+                envelope-level sibling of the 999. Gainwell's Colorado
+                MFT does not ship TA1s today; this section surfaces them
+                when they appear.
+              </p>
+              {/* SP25: live-tail status for the TA1 stream. Quietly
+                  placed after the description so the eyebrow / display
+                  title remains the dominant scan target. */}
+              <TailStatusPill
+                status={tailStatus}
+                lastEventAt={tailLastEventAt}
+                onReconnect={forceReconnect}
+              />
+            </div>
           </div>
 
           <div className="grid gap-3 grid-cols-2 md:grid-cols-4 pt-2">
diff --git a/src/store/tail-store.test.ts b/src/store/tail-store.test.ts
index ca61566..16eefee 100644
--- a/src/store/tail-store.test.ts
+++ b/src/store/tail-store.test.ts
@@ -5,7 +5,7 @@
 // state shape.
 
 import { beforeEach, describe, expect, it } from "vitest";
-import type { Activity, Claim, Remittance } from "@/types";
+import type { Ack, Activity, Claim, Remittance, Ta1Ack } from "@/types";
 import { TAIL_CAP, useTailStore } from "./tail-store";
 
 function makeClaim(id: string, overrides: Partial<Claim> = {}): Claim {
@@ -47,6 +47,36 @@ function makeActivity(id: string, overrides: Partial<Activity> = {}): Activity {
   };
 }
 
+function makeAck(id: number, overrides: Partial<Ack> = {}): Ack {
+  return {
+    id,
+    sourceBatchId: `BATCH-${id}`,
+    acceptedCount: 1,
+    rejectedCount: 0,
+    receivedCount: 1,
+    ackCode: "A",
+    parsedAt: "2026-07-02T12:00:00Z",
+    patientControlNumber: `PCN-${id}`,
+    ...overrides,
+  };
+}
+
+function makeTa1Ack(id: number, overrides: Partial<Ta1Ack> = {}): Ta1Ack {
+  return {
+    id,
+    controlNumber: `0000000${id}`,
+    ackCode: "A",
+    noteCode: "000",
+    interchangeDate: "2026-07-02",
+    interchangeTime: "1200",
+    senderId: "SENDER",
+    receiverId: "RECEIVER",
+    sourceBatchId: `TA1-BATCH-${id}`,
+    parsedAt: "2026-07-02T12:00:00Z",
+    ...overrides,
+  };
+}
+
 describe("useTailStore", () => {
   // Each test starts from a known-empty state. The store is module-level
   // (singleton), so we use `reset()` to clear between cases — that's also
@@ -55,6 +85,8 @@ describe("useTailStore", () => {
     useTailStore.getState().reset("claims");
     useTailStore.getState().reset("remittances");
     useTailStore.getState().reset("activity");
+    useTailStore.getState().reset("acks");
+    useTailStore.getState().reset("ta1_acks");
   });
 
   it("test_add_claim_adds_new_claim_keyed_by_id", () => {
@@ -128,4 +160,71 @@ describe("useTailStore", () => {
     const lastKey = `CLM-${(N - 1).toString().padStart(6, "0")}`;
     expect(claims[lastKey]).toBeDefined();
   }, 60_000);
+
+  // -------------------------------------------------------------------------
+  // SP25: acks + ta1_acks slices mirror the claims/remittances key-by-id
+  // pattern (both have a stable `id` from the database row).
+  // -------------------------------------------------------------------------
+
+  it("test_add_ack_adds_new_ack_keyed_by_id", () => {
+    const { addAck } = useTailStore.getState();
+    addAck(makeAck(1));
+    addAck(makeAck(2));
+    const { acks } = useTailStore.getState();
+    expect(Object.keys(acks)).toHaveLength(2);
+    expect(acks["1"]?.sourceBatchId).toBe("BATCH-1");
+    expect(acks["2"]?.sourceBatchId).toBe("BATCH-2");
+  });
+
+  it("test_add_ack_with_duplicate_id_is_noop", () => {
+    const { addAck } = useTailStore.getState();
+    addAck(makeAck(1, { sourceBatchId: "BATCH-original" }));
+    addAck(makeAck(1, { sourceBatchId: "BATCH-replay" }));
+    const { acks } = useTailStore.getState();
+    expect(Object.keys(acks)).toHaveLength(1);
+    // First write wins; snapshot replay must not trample the canonical row.
+    expect(acks["1"]?.sourceBatchId).toBe("BATCH-original");
+  });
+
+  it("test_reset_acks_clears_only_acks_slice", () => {
+    const { addAck, addClaim, reset } = useTailStore.getState();
+    addAck(makeAck(1));
+    addClaim(makeClaim("CLM-1"));
+
+    reset("acks");
+    const s = useTailStore.getState();
+    expect(Object.keys(s.acks)).toHaveLength(0);
+    // The other slices are untouched.
+    expect(Object.keys(s.claims)).toHaveLength(1);
+  });
+
+  it("test_add_ta1_ack_adds_new_ta1_ack_keyed_by_id", () => {
+    const { addTa1Ack } = useTailStore.getState();
+    addTa1Ack(makeTa1Ack(1));
+    addTa1Ack(makeTa1Ack(2));
+    const { ta1Acks } = useTailStore.getState();
+    expect(Object.keys(ta1Acks)).toHaveLength(2);
+    expect(ta1Acks["1"]?.controlNumber).toBe("00000001");
+    expect(ta1Acks["2"]?.controlNumber).toBe("00000002");
+  });
+
+  it("test_add_ta1_ack_with_duplicate_id_is_noop", () => {
+    const { addTa1Ack } = useTailStore.getState();
+    addTa1Ack(makeTa1Ack(7, { senderId: "ORIGINAL" }));
+    addTa1Ack(makeTa1Ack(7, { senderId: "REPLAY" }));
+    const { ta1Acks } = useTailStore.getState();
+    expect(Object.keys(ta1Acks)).toHaveLength(1);
+    expect(ta1Acks["7"]?.senderId).toBe("ORIGINAL");
+  });
+
+  it("test_reset_ta1_acks_clears_only_ta1_acks_slice", () => {
+    const { addTa1Ack, addAck, reset } = useTailStore.getState();
+    addTa1Ack(makeTa1Ack(1));
+    addAck(makeAck(1));
+
+    reset("ta1_acks");
+    const s = useTailStore.getState();
+    expect(Object.keys(s.ta1Acks)).toHaveLength(0);
+    expect(Object.keys(s.acks)).toHaveLength(1);
+  });
 });
diff --git a/src/store/tail-store.ts b/src/store/tail-store.ts
index 29de8f8..f2bda85 100644
--- a/src/store/tail-store.ts
+++ b/src/store/tail-store.ts
@@ -21,7 +21,7 @@
 // ---------------------------------------------------------------------------
 
 import { create } from "zustand";
-import type { Activity, Claim, Remittance } from "@/types";
+import type { Ack, Activity, Claim, Remittance, Ta1Ack } from "@/types";
 import type { TailResource } from "@/lib/tail-stream";
 
 /** Maximum number of items retained per slice before FIFO eviction kicks in. */
@@ -46,6 +46,12 @@ interface TailStore {
   claims: Record<string, Claim>;
   remittances: Record<string, Remittance>;
   activity: Activity[];
+  // SP25: the Acks page needs the same live-tail triplet as
+  // Claims/Remittances/Activity. `acks` and `ta1_acks` both have stable
+  // numeric ids from the database row, so they're keyed-by-id like
+  // `claims`/`remittances` (first write wins, FIFO-capped at TAIL_CAP).
+  acks: Record<string, Ack>;
+  ta1Acks: Record<string, Ta1Ack>;
 
   // --- Insertion-order trackers (kept in sync with the dicts above) ----
   // These are private to the store; consumers only read the dicts. We
@@ -54,11 +60,15 @@ interface TailStore {
   // what triggers a re-render in subscribers that select `claims`).
   claimOrder: string[];
   remitOrder: string[];
+  ackOrder: number[];
+  ta1AckOrder: number[];
 
   // --- Setters ---------------------------------------------------------
   addClaim: (c: Claim) => void;
   addRemittance: (r: RemitListItem) => void;
   addActivity: (a: Activity) => void;
+  addAck: (a: Ack) => void;
+  addTa1Ack: (a: Ta1Ack) => void;
   reset: (resource: TailResource) => void;
 }
 
@@ -69,11 +79,11 @@ interface TailStore {
  */
 type RemitListItem = Remittance;
 
-function evictOldest(
-  order: string[],
-  dict: Record<string, Claim | Remittance>,
+function evictOldest<K extends string | number, V>(
+  order: K[],
+  dict: Record<string, V>,
   batch: number,
-): { order: string[]; dict: Record<string, Claim | Remittance> } {
+): { order: K[]; dict: Record<string, V> } {
   if (order.length <= TAIL_CAP) return { order, dict };
   // Drain down to the cap; never evict more than `batch` per call so a
   // 5-item overflow evicts 5, but a 1 000-item overflow evicts 100 in
@@ -83,8 +93,8 @@ function evictOldest(
   const nextOrder = order.slice(toDrop);
   // Object.assign is consistently faster than `{ ...dict }` in V8 for
   // large dicts; we follow up with `delete` for each evicted id.
-  const nextDict: Record<string, Claim | Remittance> = Object.assign({}, dict);
-  for (const id of dropped) delete nextDict[id];
+  const nextDict: Record<string, V> = Object.assign({}, dict);
+  for (const id of dropped) delete nextDict[String(id)];
   return { order: nextOrder, dict: nextDict };
 }
 
@@ -92,8 +102,12 @@ export const useTailStore = create<TailStore>((set) => ({
   claims: {},
   remittances: {},
   activity: [],
+  acks: {},
+  ta1Acks: {},
   claimOrder: [],
   remitOrder: [],
+  ackOrder: [],
+  ta1AckOrder: [],
 
   addClaim: (c) =>
     set((s) => {
@@ -147,6 +161,38 @@ export const useTailStore = create<TailStore>((set) => ({
       return { activity: next };
     }),
 
+  addAck: (a) =>
+    set((s) => {
+      // First write wins; snapshot replay must not trample the
+      // canonical row. `id` is a number from the database row.
+      if (s.acks[String(a.id)]) return s;
+      const nextAcks: Record<string, Ack> = Object.assign({}, s.acks, {
+        [String(a.id)]: a,
+      });
+      const nextOrder = s.ackOrder.concat(a.id);
+      if (nextOrder.length > TAIL_CAP) {
+        const { order, dict } = evictOldest(nextOrder, nextAcks, EVICT_BATCH);
+        return { acks: dict, ackOrder: order };
+      }
+      return { acks: nextAcks, ackOrder: nextOrder };
+    }),
+
+  addTa1Ack: (a) =>
+    set((s) => {
+      if (s.ta1Acks[String(a.id)]) return s;
+      const nextTa1Acks: Record<string, Ta1Ack> = Object.assign(
+        {},
+        s.ta1Acks,
+        { [String(a.id)]: a },
+      );
+      const nextOrder = s.ta1AckOrder.concat(a.id);
+      if (nextOrder.length > TAIL_CAP) {
+        const { order, dict } = evictOldest(nextOrder, nextTa1Acks, EVICT_BATCH);
+        return { ta1Acks: dict, ta1AckOrder: order };
+      }
+      return { ta1Acks: nextTa1Acks, ta1AckOrder: nextOrder };
+    }),
+
   reset: (resource) =>
     set(() => {
       switch (resource) {
@@ -156,6 +202,10 @@ export const useTailStore = create<TailStore>((set) => ({
           return { remittances: {}, remitOrder: [] };
         case "activity":
           return { activity: [] };
+        case "acks":
+          return { acks: {}, ackOrder: [] };
+        case "ta1_acks":
+          return { ta1Acks: {}, ta1AckOrder: [] };
       }
     }),
 }));