From d109138ef5bdad2231d6be311c1cb6ec117b2631 Mon Sep 17 00:00:00 2001 From: HeLLGURD <129007007+HeLLGURD@users.noreply.github.com> Date: Sat, 1 Aug 2026 11:17:03 +0530 Subject: [PATCH] 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 --- mcp_serve.py | 107 ++++++++--- tests/test_mcp_serve.py | 401 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 482 insertions(+), 26 deletions(-) diff --git a/mcp_serve.py b/mcp_serve.py index bb8e950309..23f248164b 100644 --- a/mcp_serve.py +++ b/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)) diff --git a/tests/test_mcp_serve.py b/tests/test_mcp_serve.py index 515d812614..6c58796577 100644 --- a/tests/test_mcp_serve.py +++ b/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