diff --git a/backend/src/cyclone/api.py b/backend/src/cyclone/api.py
index db7bfac..f7163ec 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,
 )
 
@@ -1036,7 +1037,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 +1047,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..19a6636 100644
--- a/backend/src/cyclone/api_routers/acks.py
+++ b/backend/src/cyclone/api_routers/acks.py
@@ -19,48 +19,16 @@ from fastapi.responses import StreamingResponse
 from cyclone.api_helpers import ndjson_stream_list, wants_ndjson
 from cyclone.parsers.models_999 import ParseResult999
 from cyclone.parsers.serialize_999 import serialize_999
-from cyclone.store import store
+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 +48,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
@@ -117,7 +85,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..fb81b53 100644
--- a/backend/src/cyclone/api_routers/ta1_acks.py
+++ b/backend/src/cyclone/api_routers/ta1_acks.py
@@ -15,26 +15,14 @@ from typing import Any
 from fastapi import APIRouter, HTTPException, Query
 
 from cyclone import db
-from cyclone.store import store
+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,7 +45,7 @@ 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,
@@ -70,7 +58,7 @@ def get_ta1_ack_endpoint(ack_id: int) -> dict:
     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/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/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_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"