diff --git a/opc/plugins/office_ui/chat_store.py b/opc/plugins/office_ui/chat_store.py
index 84de3be..1dfea8a 100644
--- a/opc/plugins/office_ui/chat_store.py
+++ b/opc/plugins/office_ui/chat_store.py
@@ -299,6 +299,55 @@ class ChatStore:
             return None
         return await self._message_scope(str(message_id).strip())
 
+    async def _merge_into_same_scope_row(
+        self,
+        message_id: str,
+        *,
+        channel_id: str,
+        project_id: str,
+        candidate: dict[str, Any],
+    ) -> dict[str, Any] | None:
+        """Merge ``candidate`` into an already-persisted row with the same id/scope.
+
+        Returns the merged row when the update happened (or nothing changed), or
+        None when the row could not be loaded. Backfill and the live insert path
+        can race on the same message id; the duplicate must merge in place, never
+        be re-inserted under a scoped alias id in the same channel.
+        """
+        cursor = await self._db.execute(
+            "SELECT message_id, channel_id, sender, sender_name, content, "
+            "timestamp, reply_to_id, mentions, metadata "
+            "FROM messages WHERE message_id = ? AND channel_id = ? AND project_id = ?",
+            (message_id, channel_id, project_id),
+        )
+        row = await cursor.fetchone()
+        if row is None:
+            return None
+        existing = self._row_to_message_dict(row)
+        merged = self._merge_duplicate_messages(existing, candidate)
+        if self._message_persisted_equal(existing, merged):
+            return merged
+        merged_timestamp = self._message_timestamp(merged) or time.time()
+        await self._db.execute(
+            "UPDATE messages SET sender = ?, sender_name = ?, content = ?, timestamp = ?, "
+            "reply_to_id = ?, mentions = ?, metadata = ? WHERE message_id = ? AND channel_id = ? AND project_id = ?",
+            (
+                merged["sender"],
+                merged["sender_name"],
+                merged["content"],
+                merged_timestamp,
+                merged.get("reply_to_id"),
+                json.dumps(merged.get("mentions", [])),
+                json.dumps(merged.get("metadata", {})),
+                message_id,
+                channel_id,
+                project_id,
+            ),
+        )
+        merged["timestamp"] = merged_timestamp
+        merged["created_at"] = merged_timestamp
+        return merged
+
     async def _allocate_scoped_message_id(
         self,
         message_id: str,
@@ -1070,6 +1119,20 @@ class ChatStore:
                 continue
 
             existing_scope = await self._message_scope(mid)
+            if existing_scope == (channel_id, project_id):
+                # The row appeared after our initial snapshot (a live insert
+                # raced this backfill). Merge in place — never re-insert the
+                # same message under a scoped alias id in its own channel.
+                merged = await self._merge_into_same_scope_row(
+                    mid,
+                    channel_id=channel_id,
+                    project_id=project_id,
+                    candidate=normalized_message,
+                )
+                if merged is not None:
+                    existing_ids.add(mid)
+                    existing_messages.append(merged)
+                    continue
             if existing_scope and existing_scope != (channel_id, project_id):
                 metadata = dict(normalized_message.get("metadata", {}) or {})
                 metadata.setdefault("ui_message_id", mid)
@@ -1114,6 +1177,16 @@ class ChatStore:
                     ),
                 )
             except sqlite3.IntegrityError:
+                merged = await self._merge_into_same_scope_row(
+                    normalized_message["message_id"],
+                    channel_id=channel_id,
+                    project_id=project_id,
+                    candidate=normalized_message,
+                )
+                if merged is not None:
+                    existing_ids.add(normalized_message["message_id"])
+                    existing_messages.append(merged)
+                    continue
                 metadata = dict(normalized_message.get("metadata", {}) or {})
                 metadata.setdefault("ui_message_id", normalized_message["message_id"])
                 normalized_message["metadata"] = metadata
diff --git a/opc/plugins/office_ui/ws_handler.py b/opc/plugins/office_ui/ws_handler.py
index c083b8f..391983b 100644
--- a/opc/plugins/office_ui/ws_handler.py
+++ b/opc/plugins/office_ui/ws_handler.py
@@ -3068,6 +3068,91 @@ class WSHandler:
             },
         })
 
+    async def _ensure_reply_projected(
+        self,
+        *,
+        channel_id: str,
+        project_id: str,
+        session_id: str | None,
+        engine: Any | None = None,
+    ) -> None:
+        """Last-resort invariant: the session's newest persisted top-level reply
+        must exist in the UI channel once the turn has unwound.
+
+        The transcript sync is the normal projection path; when it is starved,
+        cancelled, or misses the row (project 000, 2026-07-07 19:21/20:27), the
+        engine has replied but the user sees an empty conversation forever.
+        Detection is by the transcript message id, so an already-projected reply
+        (any channel) is never duplicated.
+        """
+        if not session_id:
+            return
+        runtime_engine = engine or self.engine
+        store = getattr(runtime_engine, "store", None)
+        if not self._store_is_ready(store):
+            return
+        lister = getattr(store, "list_session_messages", None)
+        parts_loader = getattr(store, "list_session_parts", None)
+        if not callable(lister) or not callable(parts_loader):
+            return
+        try:
+            records = await lister(session_id)
+        except Exception:
+            logger.opt(exception=True).debug("reply projection: failed to list session messages")
+            return
+        latest = None
+        for record in reversed(records or []):
+            if str(getattr(record, "role", "") or "").strip().lower() != "assistant":
+                continue
+            metadata = dict(getattr(record, "metadata", {}) or {})
+            if str(metadata.get("kind", "") or "").strip() != "top_level_reply":
+                continue
+            latest = record
+            break
+        if latest is None:
+            return
+        message_id = str(getattr(latest, "message_id", "") or "").strip()
+        if not message_id:
+            return
+        try:
+            if await self.chat_store.message_scope(message_id) is not None:
+                return
+        except Exception:
+            return
+        try:
+            parts = await parts_loader(session_id, message_id)
+        except Exception:
+            logger.opt(exception=True).debug("reply projection: failed to load reply parts")
+            return
+        text = "\n".join(
+            chunk
+            for part in parts or []
+            if str(getattr(part, "part_type", "") or "") == "text"
+            for chunk in [str(dict(getattr(part, "payload", {}) or {}).get("text", "") or "")]
+            if chunk
+        ).strip()
+        if not text:
+            return
+        logger.warning(
+            f"Top-level reply {message_id} missing from UI channel {channel_id} after "
+            "transcript sync; projecting it directly"
+        )
+        reply_metadata = dict(getattr(latest, "metadata", {}) or {})
+        reply_metadata.setdefault("kind", "top_level_reply")
+        reply_metadata.setdefault("source", "engine")
+        reply_metadata.setdefault("ui_message_id", message_id)
+        reply_metadata["reply_projection_fallback"] = True
+        msg = await self.chat_store.insert_message(
+            channel_id=channel_id,
+            sender="assistant",
+            sender_name="OPC",
+            content=text,
+            project_id=project_id,
+            metadata=reply_metadata,
+            message_id=message_id,
+        )
+        await self.broadcast({"type": "session_message", "payload": msg})
+
     async def _sync_task_transcript_messages(
         self,
         task_id: str,
@@ -8202,6 +8287,12 @@ class WSHandler:
                     engine=engine,
                     latest_assistant_metadata=checkpoint_meta if checkpoint_meta else None,
                 )
+                await self._ensure_reply_projected(
+                    channel_id=channel_id,
+                    project_id=pid,
+                    session_id=session_id or (str(getattr(task, "session_id", "") or "").strip() if task else None),
+                    engine=engine,
+                )
 
                 # ── Status: idle only while the engine left the task active ──
                 store = engine.store
diff --git a/tests/test_reply_projection_invariant.py b/tests/test_reply_projection_invariant.py
new file mode 100644
index 0000000..70a3427
--- /dev/null
+++ b/tests/test_reply_projection_invariant.py
@@ -0,0 +1,157 @@
+"""Invariant: a completed turn's top-level reply must be visible in the UI channel.
+
+Project 000 forensics (2026-07-07 19:21 / 20:27): the engine recorded the
+assistant reply in session_messages, but the reply never appeared in the
+ui_state messages table, so the user stared at an empty conversation. The UI
+projection of engine replies runs only as a post-turn transcript sync inside
+_process_session_message; if that step is starved, cancelled, or misses the
+row, nothing reconciles the channel afterwards.
+
+These tests drive WSHandler._process_session_message against a real OPCStore
+and a real ChatStore with the engine's process_message mocked to behave like
+the incident turn (records the transcript row, returns the reply text).
+"""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+import uuid
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import aiosqlite
+
+from opc.core.models import SessionMessageRecord, SessionPartRecord, Task
+from opc.database.store import OPCStore
+from opc.plugins.office_ui.chat_store import ChatStore
+from opc.plugins.office_ui.event_adapter import EventAdapter
+from opc.plugins.office_ui.ws_handler import WSHandler
+
+PROJECT_ID = "proj-reply"
+REPLY_TEXT = (
+    "A legacy company runtime run was found for this session. "
+    "Legacy runs are read-only and cannot be resumed under the work-item runtime."
+)
+
+
+class ReplyProjectionInvariantTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        root = Path(self._tmp.name)
+
+        self.store = OPCStore(root / "tasks.db")
+        await self.store.initialize()
+        self.addAsyncCleanup(self.store.close)
+
+        db = await aiosqlite.connect(root / "ui_state.db")
+        self.chat_store = ChatStore(db)
+        await self.chat_store.initialize()
+        self.addAsyncCleanup(db.close)
+
+        self.session_id = str(uuid.uuid4())
+        self.task_id = str(uuid.uuid4())
+        await self.store.save_task(Task(
+            id=self.task_id,
+            title="Competitive analysis",
+            project_id=PROJECT_ID,
+            session_id=self.session_id,
+        ))
+
+        self.engine = MagicMock()
+        self.engine.store = self.store
+        self.engine.project_id = PROJECT_ID
+        self.engine.memory = None
+        self.engine.get_latest_pending_checkpoint_for_session = AsyncMock(return_value=None)
+        self.engine.get_pending_checkpoints_for_session = AsyncMock(return_value=[])
+        self.engine.process_message = AsyncMock(side_effect=self._engine_turn)
+
+        self.handler = WSHandler(self.engine, MagicMock(), self.chat_store, EventAdapter())
+        self.broadcasts: list[dict] = []
+        self.handler.broadcast = AsyncMock(side_effect=lambda msg: self.broadcasts.append(msg))
+        self.handler._send_ack = AsyncMock()
+
+    async def _engine_turn(self, content: str, **_kwargs: Any) -> str:
+        """Mimic the incident turn: persist user + assistant transcript rows,
+        return the reply text (engine's process_message contract)."""
+        for role, text, kind in (
+            ("user", content, "top_level_user_turn"),
+            ("assistant", REPLY_TEXT, "top_level_reply"),
+        ):
+            record = SessionMessageRecord(
+                session_id=self.session_id,
+                role=role,
+                metadata={
+                    "project_id": PROJECT_ID,
+                    "session_id": self.session_id,
+                    "interface": "office_ui",
+                    "kind": kind,
+                },
+            )
+            await self.store.save_session_message(record)
+            await self.store.save_session_part(SessionPartRecord(
+                message_id=record.message_id,
+                session_id=self.session_id,
+                part_type="text",
+                payload={"text": text},
+            ))
+        return REPLY_TEXT
+
+    async def _channel_contents(self) -> list[tuple[str, str]]:
+        channel_id = f"session:{self.task_id}"
+        cursor = await self.chat_store._db.execute(
+            "SELECT sender, content FROM messages WHERE channel_id = ? AND project_id = ? "
+            "ORDER BY timestamp",
+            (channel_id, PROJECT_ID),
+        )
+        return [(str(row[0]), str(row[1])) for row in await cursor.fetchall()]
+
+    async def test_reply_reaches_ui_channel_after_turn(self) -> None:
+        await self.handler._process_session_message(
+            self.task_id,
+            "你的交付文件在哪里？",
+            session_id=self.session_id,
+        )
+
+        rows = await self._channel_contents()
+        assistant_rows = [content for sender, content in rows if sender != "user"]
+        self.assertTrue(
+            any(REPLY_TEXT.split(".")[0] in content for content in assistant_rows),
+            f"assistant reply missing from UI channel; channel rows: {rows!r}",
+        )
+
+    async def test_reply_projected_even_if_transcript_sync_misses(self) -> None:
+        """The last-resort projection must cover sync failures (starvation,
+        cancellation, mapping defects) — the incident's exact shape."""
+        self.handler._sync_task_transcript_messages = AsyncMock(return_value=0)
+
+        await self.handler._process_session_message(
+            self.task_id,
+            "你的交付文件在哪里？",
+            session_id=self.session_id,
+        )
+
+        rows = await self._channel_contents()
+        assistant_rows = [content for sender, content in rows if sender != "user"]
+        self.assertTrue(
+            any(REPLY_TEXT.split(".")[0] in content for content in assistant_rows),
+            f"assistant reply missing from UI channel; channel rows: {rows!r}",
+        )
+        # And it must not double-insert when the sync did work: run a normal
+        # turn in the same channel and count copies of its reply.
+        self.handler._sync_task_transcript_messages = WSHandler._sync_task_transcript_messages.__get__(self.handler)
+        await self.handler._process_session_message(
+            self.task_id,
+            "再问一次",
+            session_id=self.session_id,
+        )
+        rows = await self._channel_contents()
+        copies = [content for sender, content in rows if sender != "user" and REPLY_TEXT[:40] in content]
+        self.assertLessEqual(len(copies), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_session_integration.py b/tests/test_session_integration.py
index a75baf4..410bdb9 100644
--- a/tests/test_session_integration.py
+++ b/tests/test_session_integration.py
@@ -6476,6 +6476,50 @@ class TestSnapshotBuilderSessionData(unittest.IsolatedAsyncioTestCase):
         finally:
             await chat_store._db.close()
 
+    async def test_backfill_merges_same_scope_row_that_raced_the_snapshot(self) -> None:
+        """A live insert landing after the backfill snapshot must merge in place,
+        never persist a second copy under a `::`-scoped alias id (project 000:
+        the same reply was stored twice in one channel)."""
+        chat_store = await _make_chat_store()
+        try:
+            await chat_store.create_session_channel("race-task", "Race", project_id="p1")
+            channel_id = "session:race-task"
+
+            original_scope = chat_store._message_scope
+
+            async def racing_scope(message_id: str):
+                # Simulate the live insert path landing the same row after the
+                # backfill snapshot was taken but before its INSERT runs.
+                if message_id == "engine-race-1" and await original_scope(message_id) is None:
+                    await chat_store.insert_message(
+                        channel_id=channel_id,
+                        sender="assistant",
+                        sender_name="OPC",
+                        content="Reply text",
+                        message_id="engine-race-1",
+                        project_id="p1",
+                        metadata={"note": "live-copy"},
+                    )
+                return await original_scope(message_id)
+
+            chat_store._message_scope = racing_scope
+
+            await chat_store.backfill_messages(channel_id, [{
+                "message_id": "engine-race-1",
+                "sender": "assistant",
+                "sender_name": "OPC",
+                "content": "Reply text",
+                "timestamp": time.time(),
+                "metadata": {"source": "engine", "role": "assistant"},
+            }], project_id="p1")
+
+            rows = await chat_store.get_channel_messages(channel_id, limit=20, project_id="p1")
+            self.assertEqual(len(rows), 1)
+            self.assertEqual(rows[0]["message_id"], "engine-race-1")
+            self.assertNotIn("::", rows[0]["message_id"])
+        finally:
+            await chat_store._db.close()
+
     async def test_build_collab_sync_marks_primary_as_company_runtime_from_children(self) -> None:
         """Legacy primary sessions should still be marked as company runtimes when child work items exist."""
         from opc.plugins.office_ui.snapshot_builder import build_collab_sync