Merge remote-tracking branch 'origin/main' into hermes/hermes-6b48295e

tests/gateway/test_matrix_approval_reaction_fail_closed.py

@@ -28,13 +28,38 @@ def _stub_mautrix():
         sys.modules.setdefault(sub, types.ModuleType(sub))
     sys.modules.setdefault("mautrix", stub)
     m = sys.modules["mautrix.types"]
-    for attr in (
-        "ContentURI", "EventID", "EventType", "PaginationDirection",
-        "PresenceState", "RoomCreatePreset", "RoomID", "SyncToken",
-        "TrustState", "UserID",
-    ):
-        if not hasattr(m, attr):
-            setattr(m, attr, str)
+
+    class EventType:
+        ROOM_MESSAGE = "m.room.message"
+        REACTION = "m.reaction"
+        ROOM_ENCRYPTED = "m.room.encrypted"
+        ROOM_NAME = "m.room.name"
+
+    class PaginationDirection:
+        BACKWARD = "b"
+        FORWARD = "f"
+
+    class PresenceState:
+        ONLINE = "online"
+        OFFLINE = "offline"
+        UNAVAILABLE = "unavailable"
+
+    class RoomCreatePreset:
+        PRIVATE = "private_chat"
+        PUBLIC = "public_chat"
+        TRUSTED_PRIVATE = "trusted_private_chat"
+
+    class TrustState:
+        UNVERIFIED = 0
+        VERIFIED = 1
+
+    for attr in ("ContentURI", "EventID", "RoomID", "SyncToken", "UserID"):
+        setattr(m, attr, str)
+    m.EventType = EventType
+    m.PaginationDirection = PaginationDirection
+    m.PresenceState = PresenceState
+    m.RoomCreatePreset = RoomCreatePreset
+    m.TrustState = TrustState
 
 
 _stub_mautrix()

tests/gateway/test_matrix_exec_approval.py

@@ -27,9 +27,9 @@ class TestMatrixExecApprovalReactions:
         assert result.success is True
         assert adapter._approval_prompt_by_session["sess-1"] == "$evt1"
         assert adapter._approval_prompts_by_event["$evt1"].session_key == "sess-1"
-        assert adapter._send_reaction.await_count == 2
+        assert adapter._send_reaction.await_count == 3
         emojis = [call.args[2] for call in adapter._send_reaction.await_args_list]
-        assert emojis == ["✅", "❎"]
+        assert emojis == ["✅", "♾️", "❌"]
 
     @pytest.mark.asyncio
     async def test_reaction_resolves_pending_approval(self, monkeypatch):

tests/gateway/test_matrix_project_context_isolation.py

@@ -0,0 +1,510 @@
+"""Matrix Project A / Project B context-isolation regressions."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from gateway.config import GatewayConfig, Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent
+from gateway.session import (
+    SessionContext,
+    SessionEntry,
+    SessionSource,
+    build_session_context_prompt,
+    build_session_key,
+)
+
+PROJECT_A_ROOM_ID = "!projectA:example.org"
+PROJECT_B_ROOM_ID = "!projectB:example.org"
+PROJECT_A_NAME = "Project - Project A"
+PROJECT_B_NAME = "Project - Project B"
+PROJECT_A_TOPIC = "Architecture and deploy plan for Project A"
+PROJECT_B_TOPIC = "Migration and branch plan for Project B"
+PROJECT_A_ALIAS = "#project-a:example.org"
+PROJECT_B_ALIAS = "#project-b:example.org"
+SENDER = "@alice:example.org"
+
+
+def _make_adapter():
+    from gateway.platforms.matrix import MatrixAdapter
+
+    adapter = MatrixAdapter(
+        PlatformConfig(
+            enabled=True,
+            token="test-token",
+            extra={"homeserver": "https://matrix.example.org", "user_id": "@bot:example.org"},
+        )
+    )
+    adapter._user_id = "@bot:example.org"
+    adapter._require_mention = False
+    adapter._auto_thread = False
+    adapter._matrix_session_scope = "room"
+    adapter._text_batch_delay_seconds = 0
+    adapter._background_read_receipt = MagicMock()
+    adapter._get_display_name = AsyncMock(return_value="Alice")
+    adapter._client = _FakeMatrixClient()
+    return adapter
+
+
+class _FakeMatrixClient:
+    def __init__(self):
+        self.state_store = MagicMock()
+        self.state_store.get_members = AsyncMock(return_value=["@bot:example.org", SENDER])
+
+    async def get_state_event(self, room_id, event_type):
+        rid = str(room_id)
+        state = {
+            PROJECT_A_ROOM_ID: {
+                "m.room.name": {"content": {"name": PROJECT_A_NAME}},
+                "m.room.topic": {"content": {"topic": PROJECT_A_TOPIC}},
+                "m.room.canonical_alias": {"content": {"alias": PROJECT_A_ALIAS}},
+            },
+            PROJECT_B_ROOM_ID: {
+                "m.room.name": {"content": {"name": PROJECT_B_NAME}},
+                "m.room.topic": {"content": {"topic": PROJECT_B_TOPIC}},
+                "m.room.canonical_alias": {"content": {"alias": PROJECT_B_ALIAS}},
+            },
+        }
+        value = state.get(rid, {}).get(str(event_type))
+        if value is None:
+            raise KeyError((rid, event_type))
+        return value
+
+
+async def _source_for(adapter, room_id: str, event_id: str = "$event"):
+    ctx = await adapter._resolve_message_context(
+        room_id=room_id,
+        sender=SENDER,
+        event_id=event_id,
+        body="What is next?",
+        source_content={"body": "What is next?"},
+        relates_to={},
+    )
+    assert ctx is not None
+    return ctx[-1]
+
+
+def _matrix_event(room_id: str, event_id: str, body: str = "What is next?"):
+    event = MagicMock()
+    event.room_id = room_id
+    event.sender = SENDER
+    event.event_id = event_id
+    event.timestamp = int(time.time() * 1000)
+    event.server_timestamp = event.timestamp
+    event.content = {"msgtype": "m.text", "body": body}
+    return event
+
+
+def _context_for(source: SessionSource) -> SessionContext:
+    return SessionContext(
+        source=source,
+        connected_platforms=[Platform.MATRIX],
+        home_channels={},
+        session_key=build_session_key(source),
+        session_id="session-test",
+    )
+
+
+@pytest.mark.asyncio
+async def test_matrix_source_includes_room_name_topic_and_message_id():
+    adapter = _make_adapter()
+    source = await _source_for(adapter, PROJECT_B_ROOM_ID, "$project-b-msg")
+
+    assert source.chat_id == PROJECT_B_ROOM_ID
+    assert source.chat_name == PROJECT_B_NAME
+    assert source.chat_topic == PROJECT_B_TOPIC
+    assert source.guild_id == "example.org"
+    assert source.message_id == "$project-b-msg"
+    assert source.parent_chat_id is None
+
+
+@pytest.mark.asyncio
+async def test_matrix_project_a_and_project_b_have_distinct_session_keys():
+    adapter = _make_adapter()
+    source_a = await _source_for(adapter, PROJECT_A_ROOM_ID, "$a")
+    source_b = await _source_for(adapter, PROJECT_B_ROOM_ID, "$b")
+
+    assert source_a.chat_id != source_b.chat_id
+    assert source_a.chat_name == PROJECT_A_NAME
+    assert source_b.chat_name == PROJECT_B_NAME
+    assert build_session_key(source_a) != build_session_key(source_b)
+
+
+@pytest.mark.asyncio
+async def test_matrix_project_b_prompt_contains_project_b_not_project_a():
+    adapter = _make_adapter()
+    source_b = await _source_for(adapter, PROJECT_B_ROOM_ID, "$b")
+
+    prompt = build_session_context_prompt(_context_for(source_b))
+
+    assert PROJECT_B_NAME in prompt
+    assert PROJECT_B_TOPIC in prompt
+    assert PROJECT_B_ROOM_ID in prompt
+    assert "Matrix room boundary" in prompt
+    assert PROJECT_A_NAME not in prompt
+    assert PROJECT_A_TOPIC not in prompt
+
+
+@pytest.mark.asyncio
+async def test_matrix_project_context_survives_sequential_messages():
+    adapter = _make_adapter()
+    adapter._matrix_session_scope = "room"
+    first = await _source_for(adapter, PROJECT_B_ROOM_ID, "$b1")
+    second = await _source_for(adapter, PROJECT_B_ROOM_ID, "$b2")
+
+    assert first.thread_id is None
+    assert second.thread_id is None
+    assert first.chat_name == PROJECT_B_NAME
+    assert second.chat_name == PROJECT_B_NAME
+    assert build_session_key(first) == build_session_key(second)
+
+
+@pytest.mark.asyncio
+async def test_matrix_session_scope_auto_and_thread_preserve_synthetic_threads():
+    adapter = _make_adapter()
+    adapter._auto_thread = True
+    adapter._matrix_session_scope = "auto"
+    auto_source = await _source_for(adapter, PROJECT_B_ROOM_ID, "$auto")
+    assert auto_source.thread_id == "$auto"
+
+    adapter._matrix_session_scope = "thread"
+    thread_source = await _source_for(adapter, PROJECT_B_ROOM_ID, "$thread")
+    assert thread_source.thread_id == "$thread"
+
+    real_thread = await adapter._resolve_message_context(
+        room_id=PROJECT_B_ROOM_ID,
+        sender=SENDER,
+        event_id="$reply",
+        body="thread reply",
+        source_content={"body": "thread reply"},
+        relates_to={"rel_type": "m.thread", "event_id": "$root"},
+    )
+    assert real_thread is not None
+    assert real_thread[-1].thread_id == "$root"
+
+
+@pytest.mark.asyncio
+async def test_matrix_project_context_survives_concurrent_messages():
+    from gateway.run import GatewayRunner
+    from gateway.session_context import get_session_env
+
+    async def observe(room_id: str):
+        adapter = _make_adapter()
+        source = await _source_for(adapter, room_id, f"${room_id}")
+        context = _context_for(source)
+        runner = object.__new__(GatewayRunner)
+        tokens = runner._set_session_env(context)
+        try:
+            await asyncio.sleep(0)
+            return SimpleNamespace(
+                chat_id=get_session_env("HERMES_SESSION_CHAT_ID"),
+                chat_name=get_session_env("HERMES_SESSION_CHAT_NAME"),
+                session_key=get_session_env("HERMES_SESSION_KEY"),
+            )
+        finally:
+            runner._clear_session_env(tokens)
+
+    observed_a, observed_b = await asyncio.gather(
+        observe(PROJECT_A_ROOM_ID),
+        observe(PROJECT_B_ROOM_ID),
+    )
+
+    assert observed_a.chat_id == PROJECT_A_ROOM_ID
+    assert observed_b.chat_id == PROJECT_B_ROOM_ID
+    assert observed_a.chat_name == PROJECT_A_NAME
+    assert observed_b.chat_name == PROJECT_B_NAME
+    assert observed_a.session_key != observed_b.session_key
+
+
+@pytest.mark.asyncio
+async def test_matrix_inbound_handler_emits_project_b_metadata_not_project_a():
+    adapter = _make_adapter()
+    captured = []
+
+    async def capture(event):
+        captured.append(event)
+
+    adapter.handle_message = capture
+
+    await adapter._on_room_message(_matrix_event(PROJECT_B_ROOM_ID, "$project-b"))
+
+    assert len(captured) == 1
+    source = captured[0].source
+    assert source.chat_id == PROJECT_B_ROOM_ID
+    assert source.chat_name == PROJECT_B_NAME
+    assert source.chat_topic == PROJECT_B_TOPIC
+    assert source.message_id == "$project-b"
+    assert PROJECT_A_NAME not in repr(source.to_dict())
+
+
+@pytest.mark.asyncio
+async def test_matrix_inbound_handler_keeps_project_a_and_b_distinct():
+    adapter = _make_adapter()
+    captured = []
+
+    async def capture(event):
+        captured.append(event)
+
+    adapter.handle_message = capture
+
+    await adapter._on_room_message(_matrix_event(PROJECT_A_ROOM_ID, "$project-a", "A"))
+    await adapter._on_room_message(_matrix_event(PROJECT_B_ROOM_ID, "$project-b", "B"))
+
+    assert [event.source.chat_id for event in captured] == [
+        PROJECT_A_ROOM_ID,
+        PROJECT_B_ROOM_ID,
+    ]
+    assert [event.source.chat_name for event in captured] == [
+        PROJECT_A_NAME,
+        PROJECT_B_NAME,
+    ]
+    assert build_session_key(captured[0].source) != build_session_key(captured[1].source)
+
+
+def test_matrix_room_scope_group_sessions_per_user_true_separates_users():
+    alice = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    bob = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    bob.user_id = "@bob:example.org"
+    alice.thread_id = None
+    bob.thread_id = None
+
+    assert build_session_key(alice, group_sessions_per_user=True) != build_session_key(
+        bob,
+        group_sessions_per_user=True,
+    )
+
+
+def test_matrix_room_scope_group_sessions_per_user_false_shares_room():
+    alice = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    bob = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    bob.user_id = "@bob:example.org"
+    alice.thread_id = None
+    bob.thread_id = None
+
+    assert build_session_key(alice, group_sessions_per_user=False) == build_session_key(
+        bob,
+        group_sessions_per_user=False,
+    )
+
+
+def _make_matrix_source(room_id: str, room_name: str, topic: str) -> SessionSource:
+    return SessionSource(
+        platform=Platform.MATRIX,
+        chat_id=room_id,
+        chat_name=room_name,
+        chat_type="group",
+        user_id=SENDER,
+        user_name="Alice",
+        chat_topic=topic,
+    )
+
+
+def _entry(source: SessionSource, session_id: str, title: str | None = None) -> SessionEntry:
+    return SessionEntry(
+        session_key=build_session_key(source),
+        session_id=session_id,
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        origin=source,
+        display_name=title or source.chat_name,
+        platform=Platform.MATRIX,
+        chat_type="group",
+    )
+
+
+def _make_runner(current_source: SessionSource, entries: list[SessionEntry]):
+    from gateway.run import GatewayRunner
+
+    runner = object.__new__(GatewayRunner)
+    runner.config = GatewayConfig(platforms={Platform.MATRIX: PlatformConfig(enabled=True)})
+    adapter = MagicMock()
+    adapter._matrix_session_scope = "room"
+    runner.adapters = {Platform.MATRIX: adapter}
+    runner.session_store = MagicMock()
+    runner.session_store._entries = {entry.session_key: entry for entry in entries}
+    current = next((e for e in entries if e.origin and e.origin.chat_id == current_source.chat_id), entries[0])
+    runner.session_store.get_or_create_session.return_value = current
+    runner.session_store.switch_session.return_value = current
+    runner.session_store.load_transcript.return_value = [{"role": "user", "content": "hello"}]
+    runner._running_agents = {}
+    runner._session_run_generation = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._release_running_agent_state = MagicMock()
+    runner._clear_session_boundary_security_state = MagicMock()
+    runner._evict_cached_agent = MagicMock()
+    runner._queue_depth = MagicMock(return_value=0)
+    runner._session_db = MagicMock()
+    runner._session_db.list_sessions_rich.return_value = [
+        {"id": entry.session_id, "title": entry.display_name, "preview": ""}
+        for entry in entries
+    ]
+    runner._session_db.resolve_resume_session_id.side_effect = lambda sid: sid
+    runner._session_db.get_session_title.side_effect = lambda sid: {
+        entry.session_id: entry.display_name for entry in entries
+    }.get(sid)
+    runner._session_db.get_session.return_value = None
+    return runner
+
+
+def _event(text: str, source: SessionSource) -> MessageEvent:
+    return MessageEvent(text=text, source=source, message_id="$cmd")
+
+
+@pytest.mark.asyncio
+async def test_matrix_status_reports_current_matrix_room_scope():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_b = _entry(source_b, "session-b", "Project B Plan")
+    runner = _make_runner(source_b, [_entry(source_a, "session-a", "Project A Plan"), entry_b])
+
+    result = await runner._handle_status_command(_event("/status", source_b))
+
+    assert "Matrix scope:" in result
+    assert PROJECT_B_NAME in result
+    assert PROJECT_B_ROOM_ID in result
+    assert "session_scope: room" in result
+    session_key = build_session_key(source_b)
+    assert session_key not in result
+    assert session_key[:8] not in result
+    assert "session_key: sha256:" in result
+    assert PROJECT_A_NAME not in result
+    assert PROJECT_A_ROOM_ID not in result
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_does_not_cross_rooms_by_default():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_a = _entry(source_a, "session-a", "Project A Plan")
+    entry_b = _entry(source_b, "session-b", "Project B Plan")
+    runner = _make_runner(source_b, [entry_a, entry_b])
+    runner._session_db.resolve_session_by_title.return_value = "session-a"
+
+    result = await runner._handle_resume_command(_event("/resume Project A Plan", source_b))
+
+    assert "blocked" in result
+    assert PROJECT_A_NAME in result
+    runner.session_store.switch_session.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_allows_same_room_session():
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_b = _entry(source_b, "session-b-old", "Project B Plan")
+    runner = _make_runner(source_b, [entry_b])
+    runner.session_store.get_or_create_session.return_value = _entry(
+        source_b, "session-b-current", "Current Project B"
+    )
+    runner.session_store.switch_session.return_value = entry_b
+    runner._session_db.resolve_session_by_title.return_value = "session-b-old"
+
+    result = await runner._handle_resume_command(_event("/resume Project B Plan", source_b))
+
+    assert "Resumed session" in result
+    runner.session_store.switch_session.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_quoted_title_same_room():
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_b = _entry(source_b, "session-b-old", "Project B Plan")
+    runner = _make_runner(source_b, [entry_b])
+    runner.session_store.get_or_create_session.return_value = _entry(
+        source_b, "session-b-current", "Current Project B"
+    )
+    runner.session_store.switch_session.return_value = entry_b
+    runner._session_db.resolve_session_by_title.return_value = "session-b-old"
+
+    result = await runner._handle_resume_command(
+        _event('/resume "Project B Plan"', source_b)
+    )
+
+    assert "Resumed session" in result
+    runner._session_db.resolve_session_by_title.assert_called_once_with("Project B Plan")
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_quoted_title_cross_room_blocked():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_a = _entry(source_a, "session-a", "Project A Plan")
+    entry_b = _entry(source_b, "session-b", "Project B Plan")
+    runner = _make_runner(source_b, [entry_a, entry_b])
+    runner._session_db.resolve_session_by_title.return_value = "session-a"
+
+    result = await runner._handle_resume_command(
+        _event('/resume "Project A Plan"', source_b)
+    )
+
+    assert "blocked" in result
+    runner.session_store.switch_session.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_malformed_quote_returns_helpful_error():
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    runner = _make_runner(source_b, [_entry(source_b, "session-b", "Project B Plan")])
+
+    result = await runner._handle_resume_command(
+        _event('/resume "Project B Plan', source_b)
+    )
+
+    assert "Could not parse" in result
+    assert "quotes" in result
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_cross_room_requires_explicit_flag_and_warns():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    entry_a = _entry(source_a, "session-a", "Project A Plan")
+    entry_b = _entry(source_b, "session-b", "Project B Plan")
+    runner = _make_runner(source_b, [entry_a, entry_b])
+    runner.session_store.switch_session.return_value = entry_a
+    runner._session_db.resolve_session_by_title.return_value = "session-a"
+
+    result = await runner._handle_resume_command(
+        _event("/resume --cross-room Project A Plan", source_b)
+    )
+
+    assert "Cross-room resume" in result
+    assert PROJECT_B_NAME in result
+    runner.session_store.switch_session.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_lists_only_current_room_by_default():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    runner = _make_runner(
+        source_b,
+        [_entry(source_a, "session-a", "Project A Plan"), _entry(source_b, "session-b", "Project B Plan")],
+    )
+
+    result = await runner._handle_resume_command(_event("/resume", source_b))
+
+    assert "Project B Plan" in result
+    assert "Project A Plan" not in result
+
+
+@pytest.mark.asyncio
+async def test_matrix_resume_all_lists_room_names():
+    source_a = _make_matrix_source(PROJECT_A_ROOM_ID, PROJECT_A_NAME, PROJECT_A_TOPIC)
+    source_b = _make_matrix_source(PROJECT_B_ROOM_ID, PROJECT_B_NAME, PROJECT_B_TOPIC)
+    runner = _make_runner(
+        source_b,
+        [_entry(source_a, "session-a", "Project A Plan"), _entry(source_b, "session-b", "Project B Plan")],
+    )
+
+    result = await runner._handle_resume_command(_event("/resume --all", source_b))
+
+    assert "Project A Plan" in result
+    assert PROJECT_A_NAME in result
+    assert "Project B Plan" in result

tests/gateway/test_restart_drain.py

@@ -197,8 +197,10 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
     runner, _adapter = make_restart_runner()
     popen_calls = []
 
+    monkeypatch.setattr(gateway_run.sys, "platform", "linux")
     monkeypatch.setattr(gateway_run, "_resolve_hermes_bin", lambda: ["/usr/bin/hermes"])
     monkeypatch.setattr(gateway_run.os, "getpid", lambda: 321)
+    monkeypatch.setenv("_HERMES_GATEWAY", "1")
     monkeypatch.setattr(shutil, "which", lambda cmd: "/usr/bin/setsid" if cmd == "setsid" else None)
 
     def fake_popen(cmd, **kwargs):
@@ -217,6 +219,72 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
     assert kwargs["start_new_session"] is True
     assert kwargs["stdout"] is subprocess.DEVNULL
     assert kwargs["stderr"] is subprocess.DEVNULL
+    # The watcher must NOT inherit the gateway marker, or the CLI's
+    # self-restart loop guard refuses to run `hermes gateway restart`.
+    assert kwargs["env"].get("_HERMES_GATEWAY") is None
+
+
+def test_windows_gateway_venv_imports_add_site_packages(monkeypatch, tmp_path):
+    venv_dir = tmp_path / "venv"
+    site_packages = venv_dir / "Lib" / "site-packages"
+    pth_extra = tmp_path / "pywin32_system32"
+    site_packages.mkdir(parents=True)
+    pth_extra.mkdir()
+    (site_packages / "pywin32.pth").write_text(str(pth_extra), encoding="utf-8")
+    project_root = str(gateway_run.Path(gateway_run.__file__).resolve().parent.parent)
+
+    monkeypatch.setattr(gateway_run.sys, "platform", "win32")
+    monkeypatch.setattr(gateway_run.sys, "path", ["existing"])
+    monkeypatch.setenv("VIRTUAL_ENV", str(venv_dir))
+    monkeypatch.setenv("PYTHONPATH", "already-there")
+
+    gateway_run._ensure_windows_gateway_venv_imports()
+
+    assert gateway_run.sys.path[:2] == [project_root, str(site_packages)]
+    assert str(pth_extra) in gateway_run.sys.path
+    assert gateway_run.os.environ["VIRTUAL_ENV"] == str(venv_dir.resolve())
+    pythonpath = gateway_run.os.environ["PYTHONPATH"].split(gateway_run.os.pathsep)
+    assert pythonpath[:3] == [project_root, str(site_packages), "already-there"]
+
+
+@pytest.mark.asyncio
+async def test_windows_detached_restart_scrubs_gateway_marker(monkeypatch, tmp_path):
+    runner, _adapter = make_restart_runner()
+    popen_calls = []
+    venv_dir = tmp_path / "venv"
+    site_packages = venv_dir / "Lib" / "site-packages"
+    site_packages.mkdir(parents=True)
+
+    monkeypatch.setattr(gateway_run.sys, "platform", "win32")
+    monkeypatch.setattr(gateway_run, "_resolve_hermes_bin", lambda: ["hermes"])
+    monkeypatch.setattr(gateway_run.os, "getpid", lambda: 321)
+    monkeypatch.setenv("_HERMES_GATEWAY", "1")
+    monkeypatch.setenv("VIRTUAL_ENV", str(venv_dir))
+
+    import hermes_cli._subprocess_compat as subprocess_compat
+
+    monkeypatch.setattr(
+        subprocess_compat,
+        "windows_detach_popen_kwargs",
+        lambda: {},
+    )
+
+    def fake_popen(cmd, **kwargs):
+        popen_calls.append((cmd, kwargs))
+        return MagicMock()
+
+    monkeypatch.setattr(subprocess, "Popen", fake_popen)
+
+    await runner._launch_detached_restart_command()
+
+    assert len(popen_calls) == 1
+    cmd, kwargs = popen_calls[0]
+    assert cmd[-3:] == ["hermes", "gateway", "restart"]
+    assert kwargs["env"].get("_HERMES_GATEWAY") is None
+    assert kwargs["env"]["VIRTUAL_ENV"] == str(venv_dir)
+    assert str(site_packages) in kwargs["env"]["PYTHONPATH"].split(gateway_run.os.pathsep)
+    assert kwargs["stdout"] is subprocess.DEVNULL
+    assert kwargs["stderr"] is subprocess.DEVNULL
 
 
 # ── Shutdown notification tests ──────────────────────────────────────

tests/gateway/test_run_progress_topics.py

@@ -1488,3 +1488,72 @@ async def test_terminal_progress_no_bash_block_in_verbose_mode(monkeypatch, tmp_
     all_content = " ".join(call["content"] for call in adapter.sent)
     all_content += " ".join(call["content"] for call in adapter.edits)
     assert "```bash" not in all_content
+
+class MultiTerminalCommandAgent:
+    """Emits several consecutive terminal tool.started events, then a
+    different tool, then terminal again — to exercise header collapsing."""
+
+    def __init__(self, **kwargs):
+        self.tool_progress_callback = kwargs.get("tool_progress_callback")
+        self.tools = []
+
+    def run_conversation(self, message, conversation_history=None, task_id=None):
+        cb = self.tool_progress_callback
+        cb("tool.started", "terminal", "echo one", {"command": "echo one"})
+        cb("tool.started", "terminal", "echo two", {"command": "echo two"})
+        cb("tool.started", "terminal", "echo three", {"command": "echo three"})
+        cb("tool.started", "web_search", "query stuff", {"query": "query stuff"})
+        cb("tool.started", "terminal", "echo four", {"command": "echo four"})
+        time.sleep(0.35)
+        return {"final_response": "done", "messages": [], "api_calls": 1}
+
+
+@pytest.mark.asyncio
+async def test_consecutive_terminal_progress_collapses_headers(monkeypatch, tmp_path):
+    """Back-to-back terminal calls render ONE "terminal" header followed by
+    adjacent code blocks; a different tool in between resets the header so the
+    next terminal call gets a fresh one."""
+    monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all")
+
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    fake_run_agent = types.ModuleType("run_agent")
+    fake_run_agent.AIAgent = MultiTerminalCommandAgent
+    monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
+    import tools.terminal_tool  # noqa: F401 - register terminal emoji
+
+    adapter = CodeBlockProgressAdapter(platform=Platform.TELEGRAM)
+    runner = _make_runner(adapter)
+    gateway_run = importlib.import_module("gateway.run")
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})
+
+    source = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="12345",
+        chat_type="dm",
+        thread_id=None,
+    )
+
+    result = await runner._run_agent(
+        message="hello",
+        context_prompt="",
+        history=[],
+        source=source,
+        session_id="sess-terminal-consecutive",
+        session_key="agent:main:telegram:dm:12345",
+    )
+
+    assert result["final_response"] == "done"
+    contents = [call["content"] for call in adapter.sent] + [
+        call["content"] for call in adapter.edits
+    ]
+    final = max(contents, key=len) if contents else ""
+    # All four commands present as code blocks.
+    for cmd in ("echo one", "echo two", "echo three", "echo four"):
+        assert cmd in final
+    # Exactly TWO terminal headers: one for the first run of three calls,
+    # one for the terminal call after web_search broke the streak.
+    assert final.count("terminal\n```") == 2

tests/gateway/test_session.py

@@ -611,6 +611,30 @@ class TestSessionStoreSwitchSession:
         db.close()
 
 
+class TestSessionStoreLookupBySessionId:
+    @pytest.fixture()
+    def store(self, tmp_path):
+        config = GatewayConfig()
+        with patch("gateway.session.SessionStore._ensure_loaded"):
+            s = SessionStore(sessions_dir=tmp_path, config=config)
+        s._db = None
+        s._loaded = True
+        return s
+
+    def test_returns_active_entry_for_persisted_session_id(self, store):
+        source = SessionSource(
+            platform=Platform.MATRIX,
+            chat_id="!room:example.org",
+            chat_type="group",
+            user_id="@alice:example.org",
+        )
+        entry = store.get_or_create_session(source)
+
+        assert store.lookup_by_session_id(entry.session_id) is entry
+        assert store.lookup_by_session_id("missing") is None
+        assert store.lookup_by_session_id("") is None
+
+
 class TestWhatsAppSessionKeyConsistency:
     """Regression: WhatsApp session keys must collapse JID/LID aliases to a
     single stable identity for both DM chat_ids and group participant_ids."""

tests/gateway/test_stream_consumer.py

@@ -794,9 +794,11 @@ class TestSegmentBreakOnToolBoundary:
         )
 
     @pytest.mark.asyncio
-    async def test_fallback_final_deletes_partial_after_chunks_succeed(self):
-        """After fallback chunks land, the frozen partial must be deleted so
-        the user sees only the complete response (#16668)."""
+    async def test_fallback_final_deletes_partial_after_full_resend(self):
+        """After fallback re-sends the COMPLETE response, the frozen partial
+        must be deleted so the user sees only the complete response (#16668).
+        Full resend happens when the visible prefix doesn't match the final
+        text (e.g. post-segment-break content, #10807)."""
         adapter = MagicMock()
         adapter.send = AsyncMock(
             return_value=SimpleNamespace(success=True, message_id="msg_new"),
@@ -810,14 +812,49 @@ class TestSegmentBreakOnToolBoundary:
         config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
         consumer = GatewayStreamConsumer(adapter, "chat_123", config)
 
-        # Seed the consumer as if it already edited a partial message that
-        # later got stuck (flood control etc.) — _message_id is the stale id.
+        # The stale partial shows pre-tool text that is NOT a prefix of the
+        # final response — fallback re-sends the complete final text.
+        consumer._message_id = "msg_partial"
+        consumer._last_sent_text = "Let me check that for you…"
+
+        await consumer._send_fallback_final("Working on it. Done!")
+
+        adapter.delete_message.assert_awaited_once_with("chat_123", "msg_partial")
+        assert consumer._final_response_sent is True
+
+    @pytest.mark.asyncio
+    async def test_fallback_final_keeps_partial_after_tail_only_send(self):
+        """When the fallback sends only the missing TAIL (visible prefix
+        matches the final text), the partial message IS the head of the
+        answer — deleting it would leave the user with only the last part
+        of the response (the 'model sent only the second half' bug)."""
+        adapter = MagicMock()
+        adapter.send = AsyncMock(
+            return_value=SimpleNamespace(success=True, message_id="msg_new"),
+        )
+        adapter.edit_message = AsyncMock(
+            return_value=SimpleNamespace(success=True),
+        )
+        adapter.delete_message = AsyncMock(return_value=None)
+        adapter.MAX_MESSAGE_LENGTH = 4096
+
+        config = StreamConsumerConfig(edit_interval=0.01, buffer_threshold=5)
+        consumer = GatewayStreamConsumer(adapter, "chat_123", config)
+
+        # Visible partial is a true prefix of the final response — the
+        # fallback dedup sends only the tail.
         consumer._message_id = "msg_partial"
         consumer._last_sent_text = "Working on i"
 
         await consumer._send_fallback_final("Working on it. Done!")
 
-        adapter.delete_message.assert_awaited_once_with("chat_123", "msg_partial")
+        # Tail was sent...
+        sent_contents = [
+            c.kwargs.get("content", "") for c in adapter.send.call_args_list
+        ]
+        assert any("Done!" in s and "Working on i" not in s for s in sent_contents)
+        # ...and the head-bearing partial was NOT deleted.
+        adapter.delete_message.assert_not_awaited()
         assert consumer._final_response_sent is True
 
     @pytest.mark.asyncio

tests/gateway/test_stream_consumer_fresh_final.py

@@ -347,6 +347,200 @@ class TestSegmentBreakDoesNotMarkFinalSent:
         assert any("answer is 42" in t for t in self._delivered_texts(adapter))
 
 
+class TestCancelledBestEffortDeliveryFinalizes:
+    """Cancel-path best-effort delivery must go through the finalize path.
+
+    The gateway cancels the consumer shortly after finish(). The
+    CancelledError handler re-delivers the accumulated text; previously it
+    did so with finalize=False, so REQUIRES_EDIT_FINALIZE platforms
+    (Telegram) kept the plain streaming preview — the whole final reply
+    rendered with raw markdown markers — while the success flags still
+    suppressed the gateway's formatted re-send.
+    """
+
+    @pytest.mark.asyncio
+    async def test_cancel_best_effort_edit_is_finalized(self):
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        consumer = GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
+            ),
+        )
+        consumer.on_delta("Reply with **bold** and `code` markers.")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)  # preview lands; message_id set
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+        finalize_edits = [
+            c for c in adapter.edit_message.call_args_list
+            if c.kwargs.get("finalize")
+        ]
+        assert finalize_edits, (
+            "cancel best-effort delivery must use finalize=True so "
+            "REQUIRES_EDIT_FINALIZE platforms apply final formatting"
+        )
+        assert consumer.final_response_sent is True
+        assert consumer.final_content_delivered is True
+
+    @pytest.mark.asyncio
+    async def test_cancel_best_effort_failure_keeps_gateway_resend_possible(self):
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        consumer = GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
+            ),
+        )
+        consumer.on_delta("Reply with **bold** and `code` markers.")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)
+        # Best-effort delivery at cancel time fails.
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
+            success=False, error="boom",
+        ))
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+        assert consumer.final_response_sent is False
+        assert consumer.final_content_delivered is False
+
+    @pytest.mark.asyncio
+    async def test_cancel_without_preview_makes_no_delivery_attempt(self):
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        consumer = GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
+            ),
+        )
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.02)
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+        adapter.edit_message.assert_not_called()
+        assert consumer.final_response_sent is False
+        assert consumer.final_content_delivered is False
+
+    @pytest.mark.asyncio
+    async def test_cancel_with_fresh_final_enabled_delivers_and_flags_via_handler(self):
+        """With fresh_final_after_seconds enabled and an aged preview, the
+        finalized cancel-path delivery is eligible for fresh-final
+        (delete + fresh send). is_turn_final=False keeps _try_fresh_final
+        from setting the flags itself; the cancel handler sets them after
+        the successful delivery."""
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        adapter.send.side_effect = [
+            SimpleNamespace(success=True, message_id="initial_preview"),
+            SimpleNamespace(success=True, message_id="fresh_final"),
+        ]
+        consumer = GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=0.01, buffer_threshold=5, cursor=" ▉",
+                fresh_final_after_seconds=0.001,
+            ),
+        )
+        consumer.on_delta("Reply with **bold** and `code` markers.")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)
+        consumer._message_created_ts = 0.0  # force the preview stale
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+        # Fresh-final engaged: a second send replaced the stale preview.
+        assert adapter.send.call_count == 2
+        adapter.delete_message.assert_awaited_once_with("chat", "initial_preview")
+        # Flags were set by the cancel handler after successful delivery.
+        assert consumer.final_response_sent is True
+        assert consumer.final_content_delivered is True
+
+
+class TestGotDoneOverflowSplitNotRefinalized:
+    """A got_done finalize edit that split-and-delivered across continuation
+    messages must not be followed by the redundant requires-finalize edit.
+
+    After a split, the consumer adopts the last continuation as the live
+    message and the redundant finalize edit re-submits the FULL accumulated
+    text against it; the adapter pre-flights that into another overflow
+    split, editing chunk 1 over the continuation and re-sending the rest,
+    so the user sees duplicated chunks. The finalize signal was already
+    carried by the split edit itself.
+    """
+
+    def _consumer(self, adapter):
+        # High interval/threshold so the only edit is the got_done finalize.
+        return GatewayStreamConsumer(
+            adapter=adapter,
+            chat_id="chat",
+            config=StreamConsumerConfig(
+                edit_interval=10.0, buffer_threshold=10_000, cursor=" ▉",
+            ),
+        )
+
+    @pytest.mark.asyncio
+    async def test_split_finalize_edit_is_not_refinalized(self):
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
+            success=True,
+            message_id="cont_2",
+            continuation_message_ids=("cont_2",),
+        ))
+        consumer = self._consumer(adapter)
+        consumer.on_delta("oversize **markdown** final reply")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)  # preview send lands; no interval edits
+        consumer.finish()
+        await task
+
+        finalize_edits = [
+            c for c in adapter.edit_message.call_args_list
+            if c.kwargs.get("finalize")
+        ]
+        assert len(finalize_edits) == 1, (
+            "split finalize edit must not be re-finalized; the redundant "
+            "edit re-splits the full text into the adopted continuation "
+            "and duplicates chunks on screen"
+        )
+        assert consumer.final_response_sent is True
+        assert consumer.final_content_delivered is True
+
+    @pytest.mark.asyncio
+    async def test_non_split_finalize_edit_still_gets_explicit_refinalize(self):
+        """The narrow fix must not regress the requires-finalize contract:
+        a normal (non-split) got_done edit is still followed by the
+        explicit finalize edit (#25010 semantics unchanged)."""
+        adapter = _make_adapter()
+        adapter.REQUIRES_EDIT_FINALIZE = True
+        adapter.edit_message = AsyncMock(return_value=SimpleNamespace(
+            success=True, message_id="initial_preview",
+        ))
+        consumer = self._consumer(adapter)
+        consumer.on_delta("short final reply")
+        task = asyncio.create_task(consumer.run())
+        await asyncio.sleep(0.05)
+        consumer.finish()
+        await task
+
+        finalize_edits = [
+            c for c in adapter.edit_message.call_args_list
+            if c.kwargs.get("finalize")
+        ]
+        assert len(finalize_edits) == 2
+        assert consumer.final_response_sent is True
+
+
 class TestStreamConsumerConfigFreshFinalField:
     """The dataclass field must exist and default to 0 (disabled)."""
 

tests/gateway/test_telegram_overflow_partial.py

@@ -0,0 +1,140 @@
+"""Regression coverage for partial Telegram overflow delivery."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import PlatformConfig
+from gateway.platforms.base import SendResult
+from gateway.platforms.telegram import TelegramAdapter
+from gateway.stream_consumer import GatewayStreamConsumer
+
+
+def _message(message_id: int | str) -> SimpleNamespace:
+    return SimpleNamespace(message_id=message_id)
+
+
+@pytest.fixture
+def telegram_adapter() -> TelegramAdapter:
+    adapter = TelegramAdapter(PlatformConfig(enabled=True, token="fake-token"))
+    adapter._bot = MagicMock()
+    object.__setattr__(adapter, "MAX_MESSAGE_LENGTH", 160)
+    return adapter
+
+
+@pytest.mark.asyncio
+async def test_edit_overflow_split_reports_success_when_all_continuations_land(telegram_adapter):
+    """Complete overflow delivery keeps the existing successful contract."""
+    content = "word " * 120
+    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
+    telegram_adapter._bot.send_message = AsyncMock(
+        side_effect=[_message(202), _message(203), _message(204), _message(205)]
+    )
+
+    result = await telegram_adapter._edit_overflow_split(
+        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
+    )
+
+    assert result.success is True
+    assert result.message_id == result.continuation_message_ids[-1]
+    assert result.raw_response is None
+    assert telegram_adapter._bot.edit_message_text.await_count == 1
+    assert telegram_adapter._bot.send_message.await_count == len(result.continuation_message_ids)
+    for call in telegram_adapter._bot.send_message.await_args_list:
+        assert call.kwargs["message_thread_id"] == 77
+
+
+@pytest.mark.asyncio
+async def test_edit_overflow_split_reports_later_partial_failure_after_some_continuations_land(telegram_adapter):
+    """Partial metadata tracks the last delivered continuation before failure."""
+    content = "word " * 120
+    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
+    telegram_adapter._bot.send_message = AsyncMock(
+        side_effect=[
+            _message(202),
+            RuntimeError("telegram send failed"),
+            RuntimeError("telegram send failed"),
+        ]
+    )
+
+    result = await telegram_adapter._edit_overflow_split(
+        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
+    )
+
+    assert result.success is False
+    assert result.message_id == "202"
+    assert result.raw_response["partial_overflow"] is True
+    assert result.raw_response["delivered_chunks"] == 2
+    assert result.raw_response["last_message_id"] == "202"
+    assert result.continuation_message_ids == ("202",)
+
+
+@pytest.mark.asyncio
+async def test_edit_overflow_split_reports_partial_failure_when_continuation_fails(telegram_adapter):
+    """A failed continuation must not be reported as final delivery."""
+    content = "word " * 120
+    telegram_adapter._bot.edit_message_text = AsyncMock(return_value=True)
+    telegram_adapter._bot.send_message = AsyncMock(
+        side_effect=[RuntimeError("telegram send failed"), RuntimeError("telegram send failed")]
+    )
+
+    result = await telegram_adapter._edit_overflow_split(
+        "12345", "201", content, finalize=False, metadata={"thread_id": "77"}
+    )
+
+    assert result.success is False
+    assert result.retryable is True
+    assert result.error == "overflow_continuation_failed"
+    assert result.message_id == "201"
+    assert result.raw_response["partial_overflow"] is True
+    assert result.raw_response["delivered_chunks"] == 1
+    assert result.raw_response["total_chunks"] > 1
+    assert result.raw_response["last_message_id"] == "201"
+    assert result.raw_response["delivered_prefix"]
+    assert result.continuation_message_ids == ()
+
+
+@pytest.mark.asyncio
+async def test_stream_consumer_fallback_sends_tail_after_partial_overflow():
+    """A partial overflow edit enters fallback instead of marking final delivered."""
+    adapter = MagicMock()
+    adapter.MAX_MESSAGE_LENGTH = 4096
+    adapter.edit_message = AsyncMock(
+        return_value=SendResult(
+            success=False,
+            message_id="preview-1",
+            error="overflow_continuation_failed",
+            retryable=True,
+            raw_response={
+                "partial_overflow": True,
+                "delivered_chunks": 1,
+                "total_chunks": 2,
+                "last_message_id": "preview-1",
+                "delivered_prefix": "hello ",
+            },
+        )
+    )
+    adapter.send = AsyncMock(return_value=SendResult(success=True, message_id="tail-1"))
+    adapter.delete_message = AsyncMock(return_value=True)
+
+    consumer = GatewayStreamConsumer(adapter, "chat-1", metadata={"thread_id": "77"})
+    consumer._message_id = "preview-1"
+    consumer._last_sent_text = "hello "
+
+    ok = await consumer._send_or_edit("hello world", finalize=True)
+
+    assert ok is False
+    assert consumer.final_response_sent is False
+    assert consumer.final_content_delivered is False
+    assert consumer._fallback_final_send is True
+    assert consumer._fallback_prefix == "hello "
+
+    await consumer._send_fallback_final("hello world")
+
+    adapter.send.assert_awaited_once()
+    assert adapter.send.await_args.kwargs["content"] == "world"
+    assert adapter.send.await_args.kwargs["metadata"] == {"thread_id": "77"}
+    adapter.delete_message.assert_not_awaited()
+    assert consumer.final_response_sent is True
+    assert consumer.final_content_delivered is True

tests/gateway/test_telegram_voice_v0_regressions.py

@@ -0,0 +1,71 @@
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from gateway.config import Platform
+from gateway.platforms.telegram import TelegramAdapter
+from gateway.run import GatewayRunner
+from gateway.session import SessionSource
+
+
+def _source():
+    return SessionSource(platform=Platform.TELEGRAM, chat_id="12345", chat_type="dm")
+
+
+def _runner(adapter=None):
+    runner = object.__new__(GatewayRunner)
+    runner.config = SimpleNamespace(
+        stt_enabled=True,
+        group_sessions_per_user=True,
+        thread_sessions_per_user=False,
+    )
+    runner.adapters = {Platform.TELEGRAM: adapter} if adapter else {}
+    runner._consume_pending_native_image_paths = lambda _key: []
+    runner._session_key_for_source = lambda _source: "telegram:dm:12345"
+    runner._thread_metadata_for_source = lambda *_args, **_kwargs: {}
+    runner._reply_anchor_for_event = lambda _event: None
+    return runner
+
+
+def test_telegram_audio_size_gate_rejects_oversized_media_before_download():
+    adapter = object.__new__(TelegramAdapter)
+    adapter._max_doc_bytes = 1024
+
+    allowed, note = adapter._telegram_media_size_allowed(
+        SimpleNamespace(file_size=2048),
+        "voice message",
+    )
+
+    assert allowed is False
+    assert "exceeds" in note
+    assert "voice message" in note
+
+
+@pytest.mark.asyncio
+async def test_voice_tts_is_explicit_audio_reply_opt_in():
+    adapter = SimpleNamespace(
+        _auto_tts_disabled_chats=set(),
+        _auto_tts_enabled_chats=set(),
+    )
+    runner = _runner(adapter)
+    runner._voice_mode = {}
+    runner._voice_provider_mode = {}
+    runner._save_voice_modes = lambda: None
+    runner._save_voice_provider_modes = lambda: None
+
+    event = SimpleNamespace(
+        source=_source(),
+        get_command_args=lambda: "tts",
+    )
+    result = await GatewayRunner._handle_voice_command(runner, event)
+
+    assert runner._voice_mode["telegram:12345"] == "all"
+    assert "12345" in adapter._auto_tts_enabled_chats
+    assert result

tests/gateway/test_whatsapp_stale_bridge.py

@@ -0,0 +1,341 @@
+"""Tests for the WhatsApp stale-bridge staleness handshake.
+
+Regression tests for the stale-bridge trap: ``connect()`` reused any
+already-running bridge with ``status: connected`` unconditionally, and
+``disconnect()`` only kills bridges the adapter spawned itself.  A
+long-lived bridge process therefore survived gateway restarts AND
+``hermes update``, serving pre-update bridge.js behavior forever (e.g.
+no inbound media download → images/voice notes arrive as placeholders).
+
+The fix: bridge.js reports a hash of its own source in ``/health``
+(``scriptHash``); the adapter compares it against the bridge.js on disk
+and restarts the bridge on mismatch.  Bridges that predate the handshake
+report no hash and are treated as stale by definition.
+
+Also covers the npm dependency-refresh stamp: deps are reinstalled when
+package.json changes, not only when node_modules is missing.
+"""
+
+import asyncio
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from gateway.config import Platform
+
+
+class _AsyncCM:
+    """Minimal async context manager returning a fixed value."""
+
+    def __init__(self, value):
+        self.value = value
+
+    async def __aenter__(self):
+        return self.value
+
+    async def __aexit__(self, *exc):
+        return False
+
+
+def _make_adapter(bridge_script: str = "/tmp/test-bridge.js",
+                  session_path: Path = Path("/tmp/test-wa-session")):
+    """Create a WhatsAppAdapter with test attributes (bypass __init__)."""
+    from gateway.platforms.whatsapp import WhatsAppAdapter
+
+    adapter = WhatsAppAdapter.__new__(WhatsAppAdapter)
+    adapter.platform = Platform.WHATSAPP
+    adapter.config = MagicMock()
+    adapter._bridge_port = 19876
+    adapter._bridge_script = bridge_script
+    adapter._session_path = session_path
+    adapter._bridge_log_fh = None
+    adapter._bridge_log = None
+    adapter._bridge_process = None
+    adapter._reply_prefix = None
+    adapter._running = False
+    adapter._message_handler = None
+    adapter._fatal_error_code = None
+    adapter._fatal_error_message = None
+    adapter._fatal_error_retryable = True
+    adapter._fatal_error_handler = None
+    adapter._active_sessions = {}
+    adapter._pending_messages = {}
+    adapter._background_tasks = set()
+    adapter._auto_tts_disabled_chats = set()
+    adapter._message_queue = asyncio.Queue()
+    adapter._http_session = None
+    return adapter
+
+
+def _mock_health(json_data):
+    """Mock aiohttp.ClientSession whose GET returns 200 + *json_data*."""
+    mock_resp = MagicMock()
+    mock_resp.status = 200
+    mock_resp.json = AsyncMock(return_value=json_data)
+    mock_session = MagicMock()
+    mock_session.get = MagicMock(return_value=_AsyncCM(mock_resp))
+    mock_session.close = AsyncMock()
+    return MagicMock(return_value=_AsyncCM(mock_session))
+
+
+def _setup_bridge_dir(tmp_path: Path) -> Path:
+    """Create a real bridge dir with bridge.js + package.json + creds."""
+    bridge_dir = tmp_path / "whatsapp-bridge"
+    bridge_dir.mkdir()
+    (bridge_dir / "bridge.js").write_text("// current bridge code\n")
+    (bridge_dir / "package.json").write_text('{"name": "bridge"}\n')
+    session_path = tmp_path / "session"
+    session_path.mkdir()
+    (session_path / "creds.json").write_text("{}")
+    return bridge_dir
+
+
+def _fresh_node_modules(bridge_dir: Path) -> None:
+    """Create node_modules with a stamp matching the current package.json."""
+    from gateway.platforms.whatsapp import _file_content_hash
+
+    nm = bridge_dir / "node_modules"
+    nm.mkdir()
+    (nm / ".hermes-pkg-hash").write_text(
+        _file_content_hash(bridge_dir / "package.json")
+    )
+
+
+class TestFileContentHash:
+    def test_hashes_file(self, tmp_path):
+        from gateway.platforms.whatsapp import _file_content_hash
+
+        f = tmp_path / "x.js"
+        f.write_text("abc")
+        h = _file_content_hash(f)
+        assert len(h) == 16
+        assert h == _file_content_hash(f)  # deterministic
+
+    def test_changes_with_content(self, tmp_path):
+        from gateway.platforms.whatsapp import _file_content_hash
+
+        f = tmp_path / "x.js"
+        f.write_text("abc")
+        h1 = _file_content_hash(f)
+        f.write_text("def")
+        assert _file_content_hash(f) != h1
+
+    def test_missing_file_returns_empty(self, tmp_path):
+        from gateway.platforms.whatsapp import _file_content_hash
+
+        assert _file_content_hash(tmp_path / "nope.js") == ""
+
+    def test_matches_bridge_js_self_hash_algorithm(self, tmp_path):
+        """Python and Node must compute the same hash for the same bytes."""
+        import hashlib
+
+        from gateway.platforms.whatsapp import _file_content_hash
+
+        f = tmp_path / "bridge.js"
+        f.write_bytes(b"const x = 1;\n")
+        # Node side: createHash('sha256').update(bytes).digest('hex').slice(0, 16)
+        expected = hashlib.sha256(b"const x = 1;\n").hexdigest()[:16]
+        assert _file_content_hash(f) == expected
+
+
+class TestStaleBridgeHandshake:
+    @pytest.mark.asyncio
+    async def test_reuses_bridge_when_hash_matches(self, tmp_path):
+        from gateway.platforms.whatsapp import _file_content_hash
+
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        disk_hash = _file_content_hash(bridge_dir / "bridge.js")
+        mock_client = _mock_health({"status": "connected", "scriptHash": disk_hash})
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", mock_client), \
+             patch("gateway.platforms.whatsapp.asyncio.create_task") as mock_task, \
+             patch("subprocess.Popen") as mock_popen, \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True), \
+             patch.object(adapter, "_mark_connected", create=True):
+            result = await adapter.connect()
+
+        assert result is True
+        mock_popen.assert_not_called()  # reused, never spawned
+        mock_task.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_restarts_bridge_on_hash_mismatch(self, tmp_path):
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        mock_client = _mock_health(
+            {"status": "connected", "scriptHash": "deadbeefdeadbeef"}
+        )
+        # Spawned bridge dies immediately → connect() returns False, but the
+        # assertion that matters is that the stale bridge was NOT reused and
+        # a new process spawn was attempted.
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", mock_client), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process") as mock_kill_port, \
+             patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            result = await adapter.connect()
+
+        assert result is False  # mock proc died; not the point of the test
+        mock_popen.assert_called_once()  # stale bridge replaced, not reused
+        mock_kill_port.assert_called_once_with(adapter._bridge_port)
+
+    @pytest.mark.asyncio
+    async def test_restarts_unversioned_bridge(self, tmp_path):
+        """Bridges predating the handshake report no scriptHash → stale."""
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        # Old bridge /health payload: no scriptHash key at all
+        mock_client = _mock_health({"status": "connected"})
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", mock_client), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process"), \
+             patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            await adapter.connect()
+
+        mock_popen.assert_called_once()
+
+
+class TestDepRefreshStamp:
+    @pytest.mark.asyncio
+    async def test_skips_install_when_stamp_fresh(self, tmp_path):
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process"), \
+             patch("subprocess.run") as mock_run, \
+             patch("subprocess.Popen", return_value=mock_proc), \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            await adapter.connect()
+
+        mock_run.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_reinstalls_when_package_json_changed(self, tmp_path):
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        # Simulate `hermes update` bumping the Baileys pin
+        (bridge_dir / "package.json").write_text('{"name": "bridge", "v": 2}\n')
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process"), \
+             patch("subprocess.run", return_value=MagicMock(returncode=0)) as mock_run, \
+             patch("subprocess.Popen", return_value=mock_proc), \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            await adapter.connect()
+
+        mock_run.assert_called_once()
+        assert "install" in mock_run.call_args[0][0]
+        # Stamp updated to the new package.json hash
+        from gateway.platforms.whatsapp import _file_content_hash
+        stamp = (bridge_dir / "node_modules" / ".hermes-pkg-hash").read_text().strip()
+        assert stamp == _file_content_hash(bridge_dir / "package.json")
+
+    @pytest.mark.asyncio
+    async def test_installs_when_node_modules_missing(self, tmp_path):
+        bridge_dir = _setup_bridge_dir(tmp_path)  # no node_modules
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        def _npm_install(*args, **kwargs):
+            # npm creates node_modules as a side effect
+            (bridge_dir / "node_modules").mkdir(exist_ok=True)
+            return MagicMock(returncode=0)
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process"), \
+             patch("subprocess.run", side_effect=_npm_install) as mock_run, \
+             patch("subprocess.Popen", return_value=mock_proc), \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            await adapter.connect()
+
+        mock_run.assert_called_once()
+
+
+class TestCacheDirEnvPassthrough:
+    @pytest.mark.asyncio
+    async def test_bridge_spawn_env_has_cache_dirs(self, tmp_path):
+        bridge_dir = _setup_bridge_dir(tmp_path)
+        _fresh_node_modules(bridge_dir)
+        adapter = _make_adapter(
+            bridge_script=str(bridge_dir / "bridge.js"),
+            session_path=tmp_path / "session",
+        )
+        mock_proc = MagicMock()
+        mock_proc.poll.return_value = 1
+        mock_proc.returncode = 1
+
+        with patch("gateway.platforms.whatsapp.check_whatsapp_requirements", return_value=True), \
+             patch("aiohttp.ClientSession", _mock_health({"status": "disconnected"})), \
+             patch("gateway.platforms.whatsapp.asyncio.sleep", new_callable=AsyncMock), \
+             patch("gateway.platforms.whatsapp._kill_stale_bridge_by_pidfile"), \
+             patch("gateway.platforms.whatsapp._kill_port_process"), \
+             patch("subprocess.Popen", return_value=mock_proc) as mock_popen, \
+             patch.object(adapter, "_acquire_platform_lock", return_value=True, create=True):
+            await adapter.connect()
+
+        env = mock_popen.call_args.kwargs["env"]
+        from gateway.platforms.base import (
+            get_audio_cache_dir,
+            get_document_cache_dir,
+            get_image_cache_dir,
+        )
+        assert env["HERMES_IMAGE_CACHE_DIR"] == str(get_image_cache_dir())
+        assert env["HERMES_AUDIO_CACHE_DIR"] == str(get_audio_cache_dir())
+        assert env["HERMES_DOCUMENT_CACHE_DIR"] == str(get_document_cache_dir())