fix(mcp): avoid replaying historical events on startup (#13414)

EventBridge initialized each session's last_seen timestamp to 0.0, so
the first poll after 'hermes mcp serve' starts treated every saved
user/assistant message in state.db as a fresh events_poll event.

The fix establishes a per-session timestamp baseline on startup via
_establish_baseline(), recording the latest existing message timestamp
without emitting events. Only messages written after the baseline are
delivered on subsequent polls.

Also hoists _ts_float to module-level (needed by _establish_baseline)
and adds ImportError fallbacks for hermes_constants imports so the
bridge works in environments where the module isn't on the path.

Salvage of #13414 by @afurm, re-applied by @HeLLGURD in #41239.

Co-authored-by: afurm <afurm@users.noreply.github.com>

mcp_serve.py

@@ -62,9 +62,11 @@ except ImportError:
 
 def _get_sessions_dir() -> Path:
     """Return the sessions directory using HERMES_HOME."""
-    from hermes_constants import get_hermes_home
-
-    return get_hermes_home() / "sessions"
+    try:
+        from hermes_constants import get_hermes_home
+        return get_hermes_home() / "sessions"
+    except ImportError:
+        return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "sessions"
 
 
 def _get_session_db():
@@ -192,9 +194,13 @@ def _load_sessions_index_from_json() -> dict:
 
 def _load_channel_directory() -> dict:
     """Load the cached channel directory for available targets."""
-    from hermes_constants import get_hermes_home
-
-    directory_file = get_hermes_home() / "channel_directory.json"
+    try:
+        from hermes_constants import get_hermes_home
+        directory_file = get_hermes_home() / "channel_directory.json"
+    except ImportError:
+        directory_file = Path(
+            os.environ.get("HERMES_HOME", Path.home() / ".hermes")
+        ) / "channel_directory.json"
 
     if not directory_file.exists():
         return {}
@@ -292,6 +298,21 @@ class QueueEvent:
     data: dict = field(default_factory=dict)
 
 
+def _ts_float(ts) -> float:
+    """Normalize a message timestamp (epoch int/float or ISO string) to float."""
+    if isinstance(ts, (int, float)):
+        return float(ts)
+    if isinstance(ts, str) and ts:
+        try:
+            return float(ts)
+        except ValueError:
+            try:
+                return datetime.fromisoformat(ts).timestamp()
+            except Exception:
+                return 0.0
+    return 0.0
+
+
 class EventBridge:
     """Background poller that watches SessionDB for new messages and
     maintains an in-memory event queue with waiter support.
@@ -318,6 +339,13 @@ class EventBridge:
         """Start the background polling thread."""
         if self._running:
             return
+        # Snapshot existing history BEFORE the poll loop starts so pre-existing
+        # messages are not replayed as new events on startup (#13414). Sessions
+        # that first appear afterwards are absent from the baseline and default
+        # to last_seen=0.0 in _poll_once, so new-conversation delivery is
+        # preserved. Unit tests that drive _poll_once directly bypass start()
+        # and still observe first-poll delivery.
+        self._establish_baseline()
         self._running = True
         self._thread = threading.Thread(target=self._poll_loop, daemon=True)
         self._thread.start()
@@ -419,6 +447,46 @@ class EventBridge:
                 self._queue.pop(0)
         self._new_event.set()
 
+    def _establish_baseline(self) -> None:
+        """Record the latest per-session message timestamp and the current
+        state.db mtime WITHOUT emitting events, so startup does not replay
+        history (#13414).
+
+        Only sessions that already exist at startup are baselined; a session
+        that first appears afterwards is absent here and defaults to
+        last_seen=0.0 in _poll_once, so a brand-new conversation's first
+        message is still delivered on its state.db-change tick.
+        """
+        db = _get_session_db()
+        if not db:
+            return
+        try:
+            from hermes_constants import get_hermes_home
+            db_file = get_hermes_home() / "state.db"
+        except ImportError:
+            db_file = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "state.db"
+        try:
+            self._state_db_mtime = db_file.stat().st_mtime if db_file.exists() else 0.0
+        except OSError:
+            self._state_db_mtime = 0.0
+        try:
+            self._cached_sessions_index = _load_sessions_index()
+        except Exception:
+            self._cached_sessions_index = {}
+        for session_key, entry in self._cached_sessions_index.items():
+            session_id = entry.get("session_id", "")
+            if not session_id:
+                continue
+            try:
+                messages = db.get_messages(session_id)
+            except Exception:
+                continue
+            all_ts = [_ts_float(m.get("timestamp", 0)) for m in (messages or ())]
+            if all_ts:
+                latest = max(all_ts)
+                if latest > 0.0:
+                    self._last_poll_timestamps[session_key] = latest
+
     def _poll_loop(self):
         """Background loop: poll SessionDB for new messages."""
         db = _get_session_db()
@@ -444,9 +512,11 @@ class EventBridge:
         eliminating the old dual-file (sessions.json + state.db) race that
         could drop brand-new conversations (#8925).
         """
-        from hermes_constants import get_hermes_home
-
-        db_file = get_hermes_home() / "state.db"
+        try:
+            from hermes_constants import get_hermes_home
+            db_file = get_hermes_home() / "state.db"
+        except ImportError:
+            db_file = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "state.db"
 
         try:
             db_mtime = db_file.stat().st_mtime if db_file.exists() else 0.0
@@ -478,23 +548,8 @@ class EventBridge:
             if not messages:
                 continue
 
-            # Normalize timestamps to float for comparison
-            def _ts_float(ts) -> float:
-                if isinstance(ts, (int, float)):
-                    return float(ts)
-                if isinstance(ts, str) and ts:
-                    try:
-                        return float(ts)
-                    except ValueError:
-                        # ISO string — parse to epoch
-                        try:
-                            from datetime import datetime
-                            return datetime.fromisoformat(ts).timestamp()
-                        except Exception:
-                            return 0.0
-                return 0.0
-
-            # Find messages newer than our last seen timestamp
+            # Find messages newer than our last seen timestamp (see the
+            # module-level _ts_float helper for timestamp normalization).
             new_messages = []
             for msg in messages:
                 ts = _ts_float(msg.get("timestamp", 0))

tests/test_mcp_serve.py

@@ -259,9 +259,23 @@ def fake_mcp_server(populated_sessions_dir, mock_session_db, monkeypatch):
 # 1. UNIT TESTS — helpers, extraction, attachments
 # ---------------------------------------------------------------------------
 
+class TestImports:
+    def test_import_module(self):
+        import mcp_serve
+        assert hasattr(mcp_serve, "create_mcp_server")
+        assert hasattr(mcp_serve, "run_mcp_server")
+        assert hasattr(mcp_serve, "EventBridge")
+
+    def test_mcp_available_flag(self):
+        import mcp_serve
+        assert isinstance(mcp_serve._MCP_SERVER_AVAILABLE, bool)
 
 
 class TestHelpers:
+    def test_get_sessions_dir(self, tmp_path):
+        from mcp_serve import _get_sessions_dir
+        result = _get_sessions_dir()
+        assert result == tmp_path / "sessions"
 
     def test_coerce_int_handles_invalid_and_out_of_range_values(self):
         from mcp_serve import _coerce_int
@@ -272,7 +286,16 @@ class TestHelpers:
         assert _coerce_int(999, default=50, minimum=1, maximum=200) == 200
         assert _coerce_int(-5, default=50, minimum=1, maximum=200) == 1
 
+    def test_load_sessions_index_empty(self, sessions_dir, monkeypatch):
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_get_sessions_dir", lambda: sessions_dir)
+        assert mcp_serve._load_sessions_index() == {}
 
+    def test_load_sessions_index_with_data(self, populated_sessions_dir, monkeypatch):
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_get_sessions_dir", lambda: populated_sessions_dir)
+        result = mcp_serve._load_sessions_index()
+        assert len(result) == 3
 
     def test_load_sessions_index_corrupt(self, sessions_dir, monkeypatch):
         (sessions_dir / "sessions.json").write_text("not json!")
@@ -295,6 +318,11 @@ class TestContentExtraction:
         ]}
         assert _extract_message_content(msg) == "A\nB"
 
+    def test_empty(self):
+        from mcp_serve import _extract_message_content
+        assert _extract_message_content({"content": ""}) == ""
+        assert _extract_message_content({}) == ""
+        assert _extract_message_content({"content": None}) == ""
 
 
 class TestAttachmentExtraction:
@@ -307,8 +335,21 @@ class TestAttachmentExtraction:
         assert len(att) == 1
         assert att[0] == {"type": "image", "url": "http://x.com/pic.jpg"}
 
+    def test_media_tag_in_text(self):
+        from mcp_serve import _extract_attachments
+        msg = {"content": "Here MEDIA: /tmp/out.png done"}
+        att = _extract_attachments(msg)
+        assert len(att) == 1
+        assert att[0] == {"type": "media", "path": "/tmp/out.png"}
 
+    def test_multiple_media_tags(self):
+        from mcp_serve import _extract_attachments
+        msg = {"content": "MEDIA: /a.png and MEDIA: /b.mp3"}
+        assert len(_extract_attachments(msg)) == 2
 
+    def test_no_attachments(self):
+        from mcp_serve import _extract_attachments
+        assert _extract_attachments({"content": "plain text"}) == []
 
     def test_image_content_block(self):
         from mcp_serve import _extract_attachments
@@ -322,6 +363,11 @@ class TestAttachmentExtraction:
 # ---------------------------------------------------------------------------
 
 class TestEventBridge:
+    def test_create(self):
+        from mcp_serve import EventBridge
+        b = EventBridge()
+        assert b._cursor == 0
+        assert b._queue == []
 
     def test_enqueue_and_poll(self):
         from mcp_serve import EventBridge, QueueEvent
@@ -333,8 +379,29 @@ class TestEventBridge:
         assert r["events"][0]["type"] == "message"
         assert r["next_cursor"] == 1
 
+    def test_cursor_filter(self):
+        from mcp_serve import EventBridge, QueueEvent
+        b = EventBridge()
+        for i in range(5):
+            b._enqueue(QueueEvent(cursor=0, type="message", session_key=f"s{i}"))
+        r = b.poll_events(after_cursor=3)
+        assert len(r["events"]) == 2
+        assert r["events"][0]["session_key"] == "s3"
 
+    def test_session_filter(self):
+        from mcp_serve import EventBridge, QueueEvent
+        b = EventBridge()
+        b._enqueue(QueueEvent(cursor=0, type="message", session_key="a"))
+        b._enqueue(QueueEvent(cursor=0, type="message", session_key="b"))
+        b._enqueue(QueueEvent(cursor=0, type="message", session_key="a"))
+        r = b.poll_events(after_cursor=0, session_key="a")
+        assert len(r["events"]) == 2
 
+    def test_poll_empty(self):
+        from mcp_serve import EventBridge
+        r = EventBridge().poll_events(after_cursor=0)
+        assert r["events"] == []
+        assert r["next_cursor"] == 0
 
     def test_poll_limit(self):
         from mcp_serve import EventBridge, QueueEvent
@@ -344,8 +411,37 @@ class TestEventBridge:
         r = b.poll_events(after_cursor=0, limit=3)
         assert len(r["events"]) == 3
 
+    def test_wait_immediate(self):
+        from mcp_serve import EventBridge, QueueEvent
+        b = EventBridge()
+        b._enqueue(QueueEvent(cursor=0, type="message", session_key="t",
+                              data={"content": "hi"}))
+        event = b.wait_for_event(after_cursor=0, timeout_ms=100)
+        assert event is not None
+        assert event["type"] == "message"
 
+    def test_wait_timeout(self):
+        from mcp_serve import EventBridge
+        start = time.monotonic()
+        event = EventBridge().wait_for_event(after_cursor=0, timeout_ms=150)
+        assert event is None
+        assert time.monotonic() - start >= 0.1
 
+    def test_wait_wakes_on_enqueue(self):
+        from mcp_serve import EventBridge, QueueEvent
+        b = EventBridge()
+        result = [None]
+
+        def waiter():
+            result[0] = b.wait_for_event(after_cursor=0, timeout_ms=5000)
+
+        t = threading.Thread(target=waiter)
+        t.start()
+        time.sleep(0.05)
+        b._enqueue(QueueEvent(cursor=0, type="message", session_key="wake"))
+        t.join(timeout=2)
+        assert result[0] is not None
+        assert result[0]["session_key"] == "wake"
 
     def test_queue_limit(self):
         from mcp_serve import EventBridge, QueueEvent, QUEUE_LIMIT
@@ -389,6 +485,10 @@ class TestEventBridge:
         assert result["resolved"] is True
         assert len(b.list_pending_approvals()) == 0
 
+    def test_respond_nonexistent(self):
+        from mcp_serve import EventBridge
+        r = EventBridge().respond_to_approval("nope", "deny")
+        assert "error" in r
 
 
 # ---------------------------------------------------------------------------
@@ -434,6 +534,14 @@ class TestE2EConversationsList:
         platforms = {c["platform"] for c in result["conversations"]}
         assert platforms == {"telegram", "discord", "slack"}
 
+    def test_list_sorted_by_updated(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "conversations_list")
+        keys = [c["session_key"] for c in result["conversations"]]
+        # Telegram (14:30) > Discord (13:00) > Slack (11:00)
+        assert keys[0] == "agent:main:telegram:dm:123456"
+        assert keys[1] == "agent:main:discord:group:789:456"
+        assert keys[2] == "agent:main:slack:group:C1234:U5678"
 
     def test_filter_by_platform(self, mcp_server_e2e, _event_loop):
         server, _ = mcp_server_e2e
@@ -441,8 +549,21 @@ class TestE2EConversationsList:
         assert result["count"] == 1
         assert result["conversations"][0]["platform"] == "discord"
 
+    def test_filter_by_platform_case_insensitive(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "conversations_list", {"platform": "TELEGRAM"})
+        assert result["count"] == 1
 
+    def test_search_by_name(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "conversations_list", {"search": "Alice"})
+        assert result["count"] == 1
+        assert result["conversations"][0]["display_name"] == "Alice"
 
+    def test_search_no_match(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "conversations_list", {"search": "nobody"})
+        assert result["count"] == 0
 
     def test_limit(self, mcp_server_e2e, _event_loop):
         server, _ = mcp_server_e2e
@@ -479,7 +600,21 @@ class TestE2EMessagesRead:
         assert "user" in roles
         assert "assistant" in roles
 
+    def test_read_messages_content(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "messages_read",
+                          {"session_key": "agent:main:telegram:dm:123456"})
+        contents = [m["content"] for m in result["messages"]]
+        assert "Hello Alice!" in contents
+        assert "Hi! How can I help?" in contents
 
+    def test_read_messages_have_ids(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "messages_read",
+                          {"session_key": "agent:main:telegram:dm:123456"})
+        for msg in result["messages"]:
+            assert "id" in msg
+            assert msg["id"]  # non-empty
 
     def test_read_with_limit(self, mcp_server_e2e, _event_loop):
         server, _ = mcp_server_e2e
@@ -517,10 +652,29 @@ class TestE2EAttachmentsFetch:
         assert result["attachments"][0]["type"] == "media"
         assert result["attachments"][0]["path"] == "/tmp/screenshot.png"
 
+    def test_fetch_from_nonexistent_message(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "attachments_fetch", {
+            "session_key": "agent:main:telegram:dm:123456",
+            "message_id": "99999",
+        })
+        assert "error" in result
 
+    def test_fetch_from_nonexistent_session(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "attachments_fetch", {
+            "session_key": "nonexistent:key",
+            "message_id": "1",
+        })
+        assert "error" in result
 
 
 class TestE2EEventsPoll:
+    def test_poll_empty(self, mcp_server_e2e, _event_loop):
+        server, bridge = mcp_server_e2e
+        result = _run_tool(server, "events_poll")
+        assert result["events"] == []
+        assert result["next_cursor"] == 0
 
     def test_poll_with_events(self, mcp_server_e2e, _event_loop):
         from mcp_serve import QueueEvent
@@ -573,7 +727,24 @@ class TestE2EEventsWait:
         assert result["event"] is None
         assert result["reason"] == "timeout"
 
+    def test_wait_with_existing_event(self, mcp_server_e2e, _event_loop):
+        from mcp_serve import QueueEvent
+        server, bridge = mcp_server_e2e
+        bridge._enqueue(QueueEvent(cursor=0, type="message",
+                                   session_key="test",
+                                   data={"content": "waiting for this"}))
+        result = _run_tool(server, "events_wait", {"timeout_ms": 100})
+        assert result["event"] is not None
+        assert result["event"]["content"] == "waiting for this"
 
+    def test_wait_caps_timeout(self, mcp_server_e2e, _event_loop):
+        """Timeout should be capped at 300000ms (5 min)."""
+        from mcp_serve import QueueEvent
+        server, bridge = mcp_server_e2e
+        bridge._enqueue(QueueEvent(cursor=0, type="message", session_key="t"))
+        # Even with huge timeout, should return immediately since event exists
+        result = _run_tool(server, "events_wait", {"timeout_ms": 999999})
+        assert result["event"] is not None
 
 class TestMCPToolParameterCoercion:
     def test_conversations_list_coerces_string_limit(self, fake_mcp_server, _event_loop):
@@ -581,6 +752,14 @@ class TestMCPToolParameterCoercion:
         result = _run_tool(server, "conversations_list", {"limit": "2"})
         assert result["count"] == 2
 
+    def test_messages_read_coerces_string_limit(self, fake_mcp_server, _event_loop):
+        server, _ = fake_mcp_server
+        result = _run_tool(
+            server,
+            "messages_read",
+            {"session_key": "agent:main:telegram:dm:123456", "limit": "2"},
+        )
+        assert result["count"] == 2
 
     def test_events_poll_coerces_string_cursor_and_limit(self, fake_mcp_server, _event_loop):
         from mcp_serve import QueueEvent
@@ -647,10 +826,54 @@ class TestE2EChannelsList:
         assert result["count"] == 1
         assert result["channels"][0]["target"] == "slack:C1234"
 
+    def test_channels_with_directory(self, mcp_server_e2e, _event_loop, monkeypatch):
+        """Populated channel_directory.json should be unwrapped via the 'platforms' key.
 
+        Regression test for issue #21474: the writer wraps platforms under
+        {"updated_at": ..., "platforms": {...}} but the reader was iterating
+        directory.items() directly, so channels_list always returned 0.
+        """
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_load_channel_directory", lambda: {
+            "updated_at": "2026-05-07T12:00:00",
+            "platforms": {
+                "telegram": [
+                    {"id": "123456", "name": "Alice", "type": "dm"},
+                    {"id": "-100999", "name": "Dev Group", "type": "group"},
+                ],
+                "discord": [
+                    {"id": "789", "name": "general", "type": "text"},
+                ],
+            },
+        })
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "channels_list")
+        assert result["count"] == 3
+        targets = {c["target"] for c in result["channels"]}
+        assert targets == {"telegram:123456", "telegram:-100999", "discord:789"}
+
+    def test_channels_with_directory_platform_filter(self, mcp_server_e2e, _event_loop, monkeypatch):
+        """Platform filter should work against the wrapped 'platforms' payload."""
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_load_channel_directory", lambda: {
+            "updated_at": "2026-05-07T12:00:00",
+            "platforms": {
+                "telegram": [{"id": "123456", "name": "Alice", "type": "dm"}],
+                "discord": [{"id": "789", "name": "general", "type": "text"}],
+            },
+        })
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "channels_list", {"platform": "discord"})
+        assert result["count"] == 1
+        assert result["channels"][0]["target"] == "discord:789"
 
 
 class TestE2EPermissions:
+    def test_list_empty(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "permissions_list_open")
+        assert result["count"] == 0
+        assert result["approvals"] == []
 
     def test_list_with_approvals(self, mcp_server_e2e, _event_loop):
         server, bridge = mcp_server_e2e
@@ -675,6 +898,12 @@ class TestE2EPermissions:
         check = _run_tool(server, "permissions_list_open")
         assert check["count"] == 0
 
+    def test_respond_deny(self, mcp_server_e2e, _event_loop):
+        server, bridge = mcp_server_e2e
+        bridge._pending_approvals["a2"] = {"id": "a2", "kind": "plugin"}
+        result = _run_tool(server, "permissions_respond",
+                          {"id": "a2", "decision": "deny"})
+        assert result["resolved"] is True
 
     def test_respond_invalid_decision(self, mcp_server_e2e, _event_loop):
         server, bridge = mcp_server_e2e
@@ -683,6 +912,11 @@ class TestE2EPermissions:
                           {"id": "a3", "decision": "maybe"})
         assert "error" in result
 
+    def test_respond_nonexistent(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        result = _run_tool(server, "permissions_respond",
+                          {"id": "nope", "decision": "deny"})
+        assert "error" in result
 
 
 # ---------------------------------------------------------------------------
@@ -703,6 +937,10 @@ class TestToolRegistration:
         }
         assert expected == tool_names, f"Missing: {expected - tool_names}, Extra: {tool_names - expected}"
 
+    def test_tools_have_descriptions(self, mcp_server_e2e, _event_loop):
+        server, _ = mcp_server_e2e
+        for tool in server._tool_manager.list_tools():
+            assert tool.description, f"Tool {tool.name} has no description"
 
 
 # ---------------------------------------------------------------------------
@@ -710,6 +948,11 @@ class TestToolRegistration:
 # ---------------------------------------------------------------------------
 
 class TestServerCreation:
+    def test_create_server(self, populated_sessions_dir, monkeypatch):
+        pytest.importorskip("mcp", reason="MCP SDK not installed")
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_get_sessions_dir", lambda: populated_sessions_dir)
+        assert mcp_serve.create_mcp_server() is not None
 
     def test_create_with_bridge(self, populated_sessions_dir, monkeypatch):
         pytest.importorskip("mcp", reason="MCP SDK not installed")
@@ -777,6 +1020,11 @@ class TestCliIntegration:
 # ---------------------------------------------------------------------------
 
 class TestEdgeCases:
+    def test_empty_sessions_json(self, sessions_dir, monkeypatch):
+        (sessions_dir / "sessions.json").write_text("{}")
+        import mcp_serve
+        monkeypatch.setattr(mcp_serve, "_get_sessions_dir", lambda: sessions_dir)
+        assert mcp_serve._load_sessions_index() == {}
 
     def test_sessions_without_origin(self, sessions_dir, monkeypatch):
         data = {"agent:main:telegram:dm:111": {
@@ -920,6 +1168,69 @@ class TestEventBridgePollE2E:
         assert db.call_count == first_calls, \
             "Second poll should skip DB queries when files unchanged"
 
+    def test_poll_detects_new_message_after_db_write(self, tmp_path, monkeypatch):
+        """Write a new message to the DB after first poll, verify it's detected."""
+        import mcp_serve
+        sessions_dir = tmp_path / "sessions"
+        sessions_dir.mkdir()
+        monkeypatch.setattr(mcp_serve, "_get_sessions_dir", lambda: sessions_dir)
+
+        session_id = "20260329_150000_new_msg"
+        db_path = tmp_path / "state.db"
+
+        sessions_data = {
+            "agent:main:telegram:dm:new": {
+                "session_key": "agent:main:telegram:dm:new",
+                "session_id": session_id,
+                "platform": "telegram",
+                "updated_at": "2026-03-29T15:00:05",
+                "origin": {"platform": "telegram", "chat_id": "new"},
+            }
+        }
+        (sessions_dir / "sessions.json").write_text(json.dumps(sessions_data))
+        _create_test_db(db_path, session_id, [
+            {"role": "user", "content": "First", "timestamp": "2026-03-29T15:00:01"},
+        ])
+
+        class TestDB:
+            def get_messages(self, sid):
+                conn = sqlite3.connect(str(db_path))
+                conn.row_factory = sqlite3.Row
+                rows = conn.execute(
+                    "SELECT * FROM messages WHERE session_id = ? ORDER BY id",
+                    (sid,),
+                ).fetchall()
+                conn.close()
+                return [dict(r) for r in rows]
+
+        db = TestDB()
+        bridge = mcp_serve.EventBridge()
+
+        # First poll
+        bridge._poll_once(db)
+        r1 = bridge.poll_events(after_cursor=0)
+        assert len(r1["events"]) == 1
+
+        # Add a new message to the DB
+        conn = sqlite3.connect(str(db_path))
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp) VALUES (?, ?, ?, ?)",
+            (session_id, "assistant", "New reply!", "2026-03-29T15:00:10"),
+        )
+        conn.commit()
+        conn.close()
+        # Touch the DB file to update mtime (WAL mode may not update mtime on small writes)
+        os.utime(db_path, None)
+
+        # Update sessions.json updated_at to trigger re-check
+        sessions_data["agent:main:telegram:dm:new"]["updated_at"] = "2026-03-29T15:00:10"
+        (sessions_dir / "sessions.json").write_text(json.dumps(sessions_data))
+
+        # Second poll — should detect the new message
+        bridge._poll_once(db)
+        r2 = bridge.poll_events(after_cursor=r1["next_cursor"])
+        assert len(r2["events"]) == 1
+        assert r2["events"][0]["content"] == "New reply!"
 
     def test_poll_picks_up_new_conversation_on_db_change(
         self, tmp_path, monkeypatch
@@ -984,3 +1295,93 @@ class TestEventBridgePollE2E:
         assert result["events"][0]["session_key"] == "agent:main:telegram:dm:late"
         assert result["events"][0]["content"].startswith("Hello from a freshly")
 
+    def test_startup_baseline_suppresses_historical_replay(self, tmp_path, monkeypatch):
+        """start()'s baseline records existing history without emitting it, so a
+        fresh EventBridge does not replay stored messages on startup; only
+        messages written after the baseline are delivered."""
+        import mcp_serve
+
+        db_path = tmp_path / "state.db"
+        db_path.write_text("placeholder")
+        session_id = "20260329_150000_history"
+        monkeypatch.setattr(
+            mcp_serve, "_load_sessions_index",
+            lambda: {
+                "agent:main:telegram:dm:hist": {
+                    "session_id": session_id,
+                    "platform": "telegram",
+                    "origin": {"platform": "telegram", "chat_id": "hist"},
+                }
+            },
+        )
+        store = [{
+            "id": 1, "role": "user", "content": "pre-existing history",
+            "timestamp": "2026-03-29T15:00:00",
+        }]
+
+        class DB:
+            def get_messages(self, sid):
+                return list(store)
+
+        monkeypatch.setattr(mcp_serve, "_get_session_db", lambda: DB())
+
+        bridge = mcp_serve.EventBridge()
+        bridge._establish_baseline()
+        # Messages that existed before start() are not replayed.
+        assert bridge.poll_events(after_cursor=0)["events"] == []
+
+        # A message written after the baseline IS delivered on the next tick.
+        store.append({
+            "id": 2, "role": "assistant", "content": "arrived after start",
+            "timestamp": "2026-03-29T15:05:00",
+        })
+        os.utime(db_path, None)  # bump mtime so the poll gate opens
+        bridge._poll_once(DB())
+        events = bridge.poll_events(after_cursor=0)["events"]
+        assert len(events) == 1
+        assert events[0]["content"] == "arrived after start"
+
+    def test_new_conversation_after_baseline_is_delivered(self, tmp_path, monkeypatch):
+        """A conversation that first appears AFTER the startup baseline is still
+        delivered on its state.db-change tick — sessions absent from the
+        baseline default to last_seen=0.0."""
+        import mcp_serve
+
+        db_path = tmp_path / "state.db"
+        db_path.write_text("placeholder")
+        index: dict = {}
+        messages: dict = {}
+        monkeypatch.setattr(mcp_serve, "_load_sessions_index", lambda: dict(index))
+
+        class DB:
+            def get_messages(self, sid):
+                return list(messages.get(sid, []))
+
+        monkeypatch.setattr(mcp_serve, "_get_session_db", lambda: DB())
+
+        bridge = mcp_serve.EventBridge()
+        bridge._establish_baseline()  # no conversations exist yet
+
+        # The gateway registers a brand-new conversation + its first message.
+        sid = "20260329_150000_fresh"
+        index["agent:main:telegram:dm:fresh"] = {
+            "session_id": sid,
+            "platform": "telegram",
+            "origin": {"platform": "telegram", "chat_id": "fresh"},
+        }
+        messages[sid] = [{
+            "id": 1, "role": "user", "content": "hello after baseline",
+            "timestamp": "2026-03-29T15:10:00",
+        }]
+        os.utime(db_path, None)
+        bridge._poll_once(DB())
+
+        events = bridge.poll_events(after_cursor=0)["events"]
+        assert len(events) == 1
+        assert events[0]["session_key"] == "agent:main:telegram:dm:fresh"
+        assert events[0]["content"] == "hello after baseline"
+
+    def test_poll_interval_is_200ms(self):
+        """Verify the poll interval constant."""
+        from mcp_serve import POLL_INTERVAL
+        assert POLL_INTERVAL == 0.2