Merge branch 'main' into fix/nemo-relay-adaptive-config-shape

tests/gateway/test_42039_duplicate_user_message.py

@@ -0,0 +1,241 @@
+"""Tests for #42039 — user messages stored twice in state.db.
+
+When the agent has its own SessionDB reference (``_session_db is not None``),
+``_flush_messages_to_session_db()`` persists messages to SQLite during the
+agent run.  The gateway's ``append_to_transcript()`` must then use
+``skip_db=True`` on all fallback paths to prevent writing a second copy
+to the same SQLite file.
+
+This test covers the two fallback paths that previously lacked
+``skip_db=agent_persisted``:
+
+1. ``agent_failed_early`` path — transient 429/timeout failures
+2. ``not new_messages`` path — edge case where ``history_offset`` exceeds
+   the actual message count
+"""
+
+import sys
+import types
+from datetime import datetime
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import gateway.run as gateway_run
+from gateway.config import GatewayConfig, Platform
+from gateway.platforms.base import MessageEvent
+from gateway.session import SessionEntry, SessionSource
+
+
+def _bootstrap(monkeypatch, tmp_path):
+    """Minimal GatewayRunner setup shared by all tests in this module."""
+    fake_dotenv = types.ModuleType("dotenv")
+    fake_dotenv.load_dotenv = lambda *args, **kwargs: None
+    monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
+
+    config = GatewayConfig()
+    runner = gateway_run.GatewayRunner(config)
+    runner.adapters = {}
+    runner._running_agents = {}
+    runner._running_agents_ts = {}
+    runner._pending_messages = {}
+    runner._pending_approvals = {}
+    runner._is_user_authorized = lambda _source: True
+    runner._set_session_env = lambda _context: None
+    runner._handle_active_session_busy_message = AsyncMock(return_value=False)
+    runner._session_db = MagicMock()
+    runner._recover_telegram_topic_thread_id = lambda _source: None
+    runner._cache_session_source = lambda _key, _source: None
+    runner._is_session_run_current = lambda _key, _gen: True
+    runner._begin_session_run_generation = lambda _key: 1
+    runner._reply_anchor_for_event = lambda _event: None
+    runner._get_guild_id = lambda _event: None
+    runner._should_send_voice_reply = lambda *_a, **_kw: False
+    runner.hooks = MagicMock()
+    runner.hooks.emit = AsyncMock()
+
+    runner.session_store = MagicMock()
+    runner.session_store.get_or_create_session.return_value = SessionEntry(
+        session_key="agent:main:telegram:group:-1001:12345",
+        session_id="sess-dedup",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="group",
+    )
+    runner.session_store.load_transcript.return_value = []
+    runner.session_store.append_to_transcript = MagicMock()
+    runner.session_store.update_session = MagicMock()
+
+    monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
+    monkeypatch.setattr(
+        gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "fake"}
+    )
+    monkeypatch.setattr(
+        "agent.model_metadata.get_model_context_length",
+        lambda *_args, **_kwargs: 100_000,
+    )
+    return runner
+
+
+def _event():
+    return MessageEvent(
+        text="hello world",
+        source=SessionSource(
+            platform=Platform.TELEGRAM,
+            chat_id="-1001",
+            chat_type="group",
+            user_id="12345",
+        ),
+        message_id="msg-42",
+    )
+
+
+def _source():
+    return SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="-1001",
+        chat_type="group",
+        user_id="12345",
+    )
+
+
+def _assert_user_call_has_skip_db(calls, expected_skip_db: bool):
+    """Find append_to_transcript calls with role='user' and check skip_db."""
+    user_calls = []
+    for call in calls:
+        args = call.args
+        if len(args) >= 2 and isinstance(args[1], dict):
+            if args[1].get("role") == "user":
+                user_calls.append(call)
+    assert len(user_calls) >= 1, (
+        f"Expected at least one user-role append_to_transcript call, "
+        f"got calls: {[c.args for c in calls if len(c.args)>=2]}"
+    )
+    for call in user_calls:
+        actual = call.kwargs.get("skip_db", False)
+        assert actual == expected_skip_db, (
+            f"Expected skip_db={expected_skip_db} for user-role call, "
+            f"got skip_db={actual}. kwargs={call.kwargs}"
+        )
+
+
+# ── Test 1: agent_failed_early path uses skip_db=True ─────────────────
+
+
+@pytest.mark.asyncio
+async def test_agent_failed_early_skip_db_when_agent_has_session_db(
+    monkeypatch, tmp_path
+):
+    runner = _bootstrap(monkeypatch, tmp_path)
+
+    # Agent fails with transient 429
+    runner._run_agent = AsyncMock(
+        return_value={
+            "failed": True,
+            "final_response": None,
+            "error": "429 Too Many Requests — rate limit exceeded",
+            "messages": [],
+            "history_offset": 0,
+            "last_prompt_tokens": 0,
+        }
+    )
+
+    await runner._handle_message_with_agent(
+        _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
+    )
+
+    _assert_user_call_has_skip_db(
+        runner.session_store.append_to_transcript.call_args_list, True
+    )
+
+
+# ── Test 2: agent_failed_early with no _session_db → skip_db not True ─
+
+
+@pytest.mark.asyncio
+async def test_agent_failed_early_no_skip_db_when_no_session_db(
+    monkeypatch, tmp_path
+):
+    runner = _bootstrap(monkeypatch, tmp_path)
+    runner._session_db = None  # No agent DB → agent_persisted=False
+
+    runner._run_agent = AsyncMock(
+        return_value={
+            "failed": True,
+            "final_response": None,
+            "error": "ReadTimeout: timed out",
+            "messages": [],
+            "history_offset": 0,
+            "last_prompt_tokens": 0,
+        }
+    )
+
+    await runner._handle_message_with_agent(
+        _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
+    )
+
+    _assert_user_call_has_skip_db(
+        runner.session_store.append_to_transcript.call_args_list, False
+    )
+
+
+# ── Test 3: not-new-messages path uses skip_db=True ───────────────────
+
+
+@pytest.mark.asyncio
+async def test_not_new_messages_skip_db_when_agent_has_session_db(
+    monkeypatch, tmp_path
+):
+    runner = _bootstrap(monkeypatch, tmp_path)
+
+    # Agent succeeds but history_offset equals messages length → no new messages
+    runner._run_agent = AsyncMock(
+        return_value={
+            "final_response": "Hello!",
+            "messages": [{"role": "user", "content": "hi"}],
+            "tools": [],
+            "history_offset": 1,  # equals len(messages) → new_messages=[]
+            "last_prompt_tokens": 0,
+        }
+    )
+
+    await runner._handle_message_with_agent(
+        _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
+    )
+
+    _assert_user_call_has_skip_db(
+        runner.session_store.append_to_transcript.call_args_list, True
+    )
+
+
+# ── Test 4: normal path (new_messages found) uses skip_db=True ────────
+
+
+@pytest.mark.asyncio
+async def test_normal_path_skip_db_when_agent_has_session_db(
+    monkeypatch, tmp_path
+):
+    runner = _bootstrap(monkeypatch, tmp_path)
+
+    # Agent succeeds with new messages
+    runner._run_agent = AsyncMock(
+        return_value={
+            "final_response": "Hello!",
+            "messages": [
+                {"role": "user", "content": "hi"},
+                {"role": "assistant", "content": "Hello!"},
+            ],
+            "tools": [],
+            "history_offset": 0,
+            "last_prompt_tokens": 0,
+        }
+    )
+
+    await runner._handle_message_with_agent(
+        _event(), _source(), "agent:main:telegram:group:-1001:12345", 1
+    )
+
+    _assert_user_call_has_skip_db(
+        runner.session_store.append_to_transcript.call_args_list, True
+    )

tests/hermes_cli/test_auth_codex_provider.py

@@ -301,19 +301,23 @@ def test_save_codex_tokens_syncs_credential_pool(tmp_path, monkeypatch):
 
 
 def test_save_codex_tokens_syncs_manual_device_code_entries(tmp_path, monkeypatch):
-    """Re-auth must also refresh ``manual:device_code`` pool entries.
+    """Re-auth must refresh ``manual:device_code`` entries that are true
+    aliases of the singleton, while leaving INDEPENDENT entries alone.
 
-    Regression for #33538: a user who hit #33000 before the #33164 fix landed
-    would have run ``hermes auth add openai-codex`` as a workaround, leaving
-    a pool entry with ``source="manual:device_code"``.  On every subsequent
-    re-auth via setup/model picker, the singleton-seeded ``device_code`` entry
-    got refreshed but the ``manual:device_code`` entry stayed stale, recreating
-    the same 401 token_invalidated symptom that #33164 was supposed to fix.
+    Original regression for #33538: a user who hit #33000 before the #33164
+    fix landed would have run ``hermes auth add openai-codex`` as a
+    workaround, leaving a pool entry with ``source="manual:device_code"``.
+    On every subsequent re-auth via setup/model picker, the singleton-seeded
+    ``device_code`` entry got refreshed but the ``manual:device_code`` entry
+    stayed stale, recreating the same 401 token_invalidated symptom that
+    #33164 was supposed to fix.
 
-    An interactive Codex device-code re-auth proves the user owns the ChatGPT
-    account, so it is safe to refresh every device-code-backed entry in the
-    pool — but NOT independent ``manual:api_key`` entries (separate accounts /
-    explicit API keys).
+    Narrowed for #39236: the original fix treated every ``manual:device_code``
+    entry as a singleton-alias and refreshed them all, which silently
+    clobbered independent accounts added via ``hermes auth add openai-codex``.
+    The current behavior refreshes only entries whose access_token matches
+    the *previous* singleton access_token (true legacy aliases), and leaves
+    distinct-token entries alone (independent accounts).
     """
     hermes_home = tmp_path / "hermes"
     hermes_home.mkdir(parents=True, exist_ok=True)
@@ -335,16 +339,30 @@ def test_save_codex_tokens_syncs_manual_device_code_entries(tmp_path, monkeypatc
                     "access_token": "old-at",
                     "refresh_token": "old-rt",
                 },
+                # Legacy alias from the #33000 workaround era — its tokens
+                # match the singleton, so it is a true alias and SHOULD be
+                # refreshed (preserves #33538 behavior).
                 {
-                    "id": "auth-add",
+                    "id": "legacy-alias",
                     "source": "manual:device_code",
                     "auth_type": "oauth",
-                    "access_token": "stale-manual-at",
-                    "refresh_token": "stale-manual-rt",
+                    "access_token": "old-at",
+                    "refresh_token": "old-rt",
                     "last_status": "exhausted",
                     "last_error_code": 401,
                     "last_error_reason": "token_invalidated",
                 },
+                # Independent account from `hermes auth add openai-codex` —
+                # its tokens are distinct from the singleton.  Must NOT be
+                # overwritten by a re-auth that targeted a different account
+                # (#39236).
+                {
+                    "id": "independent",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "independent-at",
+                    "refresh_token": "independent-rt",
+                },
                 {
                     "id": "api-key",
                     "source": "manual:api_key",
@@ -363,18 +381,23 @@ def test_save_codex_tokens_syncs_manual_device_code_entries(tmp_path, monkeypatc
     pool = auth["credential_pool"]["openai-codex"]
 
     # Singleton-seeded device_code entry: refreshed and error markers cleared.
-    seeded = next(e for e in pool if e["source"] == "device_code")
+    seeded = next(e for e in pool if e["id"] == "seeded")
     assert seeded["access_token"] == "fresh-at"
     assert seeded["refresh_token"] == "fresh-rt"
 
-    # manual:device_code entry: ALSO refreshed (the new behavior).
-    manual_dc = next(e for e in pool if e["source"] == "manual:device_code")
-    assert manual_dc["access_token"] == "fresh-at"
-    assert manual_dc["refresh_token"] == "fresh-rt"
-    assert manual_dc["last_refresh"] == "2026-05-28T00:00:00Z"
-    assert manual_dc["last_status"] is None
-    assert manual_dc["last_error_code"] is None
-    assert manual_dc["last_error_reason"] is None
+    # Legacy alias (tokens matched previous singleton): ALSO refreshed.
+    legacy = next(e for e in pool if e["id"] == "legacy-alias")
+    assert legacy["access_token"] == "fresh-at"
+    assert legacy["refresh_token"] == "fresh-rt"
+    assert legacy["last_refresh"] == "2026-05-28T00:00:00Z"
+    assert legacy["last_status"] is None
+    assert legacy["last_error_code"] is None
+    assert legacy["last_error_reason"] is None
+
+    # Independent manual:device_code entry: NOT overwritten (#39236).
+    independent = next(e for e in pool if e["id"] == "independent")
+    assert independent["access_token"] == "independent-at"
+    assert independent["refresh_token"] == "independent-rt"
 
     # manual:api_key entry: untouched — independent credential.
     api_key = next(e for e in pool if e["source"] == "manual:api_key")
@@ -382,6 +405,333 @@ def test_save_codex_tokens_syncs_manual_device_code_entries(tmp_path, monkeypatc
     assert "refresh_token" not in api_key or api_key.get("refresh_token") is None
 
 
+def test_save_codex_tokens_does_not_overwrite_independent_manual_entries(tmp_path, monkeypatch):
+    """Re-auth must NOT overwrite ``manual:device_code`` entries that hold
+    independent token material (different OpenAI/ChatGPT accounts).
+
+    Regression for #39236: ``hermes auth add openai-codex`` for accounts B and C
+    routes through ``_save_codex_tokens`` because the singleton path is the
+    only Codex OAuth save flow.  The #33538 fix refreshed every
+    ``manual:device_code`` entry on every re-auth, which works fine for the
+    one-account/legacy-workaround case but silently overwrote distinct
+    independent accounts with the latest-authenticated tokens (labels
+    preserved, token material clobbered, status/quota readings then lie).
+
+    The safe invariant: an entry is a singleton-alias only when its current
+    access_token matches the *previous* singleton access_token.  Manual
+    entries whose tokens never matched the singleton are independent accounts
+    and must be left alone.
+    """
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir(parents=True, exist_ok=True)
+    (hermes_home / "auth.json").write_text(json.dumps({
+        "version": 1,
+        "providers": {
+            "openai-codex": {
+                # Old singleton tokens — represent "account A" which the user
+                # logged in with via setup originally.
+                "tokens": {"access_token": "acctA-at", "refresh_token": "acctA-rt"},
+                "last_refresh": "2026-01-01T00:00:00Z",
+                "auth_mode": "chatgpt",
+                "label": "account-A",
+            },
+        },
+        "credential_pool": {
+            "openai-codex": [
+                # The seeded singleton mirror of account A.
+                {
+                    "id": "seeded",
+                    "label": "account-A",
+                    "source": "device_code",
+                    "auth_type": "oauth",
+                    "access_token": "acctA-at",
+                    "refresh_token": "acctA-rt",
+                },
+                # Two INDEPENDENT manual entries added later via
+                # ``hermes auth add openai-codex`` (account B and account C).
+                # Each has its OWN distinct token material, unrelated to the
+                # singleton.
+                {
+                    "id": "acctB",
+                    "label": "account-B",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "acctB-at",
+                    "refresh_token": "acctB-rt",
+                },
+                {
+                    "id": "acctC",
+                    "label": "account-C",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "acctC-at",
+                    "refresh_token": "acctC-rt",
+                },
+            ],
+        },
+    }))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    # User re-authenticates account A — fresh device-code login produces new
+    # tokens.  The legitimate update is the seeded singleton mirror; the
+    # independent acctB/acctC entries must be untouched.
+    _save_codex_tokens(
+        {"access_token": "acctA-new-at", "refresh_token": "acctA-new-rt"},
+        last_refresh="2026-06-05T00:00:00Z",
+    )
+
+    auth = json.loads((hermes_home / "auth.json").read_text())
+    pool = auth["credential_pool"]["openai-codex"]
+
+    # Singleton-seeded entry: refreshed (legitimate sync).
+    seeded = next(e for e in pool if e["source"] == "device_code")
+    assert seeded["access_token"] == "acctA-new-at"
+    assert seeded["refresh_token"] == "acctA-new-rt"
+    assert seeded["last_refresh"] == "2026-06-05T00:00:00Z"
+
+    # acctB: INDEPENDENT entry — must NOT be overwritten.
+    acctB = next(e for e in pool if e["id"] == "acctB")
+    assert acctB["access_token"] == "acctB-at", (
+        "acctB was clobbered by acctA re-auth (#39236 regression)"
+    )
+    assert acctB["refresh_token"] == "acctB-rt"
+
+    # acctC: INDEPENDENT entry — must NOT be overwritten.
+    acctC = next(e for e in pool if e["id"] == "acctC")
+    assert acctC["access_token"] == "acctC-at", (
+        "acctC was clobbered by acctA re-auth (#39236 regression)"
+    )
+    assert acctC["refresh_token"] == "acctC-rt"
+
+
+def test_save_codex_tokens_still_refreshes_legacy_manual_alias(tmp_path, monkeypatch):
+    """The #33538 legacy use case must keep working.
+
+    A user who hit #33000 before the #33164 fix landed might have run
+    ``hermes auth add openai-codex`` as a workaround when there was no
+    singleton entry — that created a ``manual:device_code`` pool entry that
+    holds the SAME token material as the (later) singleton.  This entry is a
+    true alias of the singleton and SHOULD still be refreshed on subsequent
+    re-auths, otherwise it goes stale and recreates the #33538 symptom.
+
+    The distinguishing signal: a legacy alias has access_token == previous
+    singleton access_token; an independent account does not.
+    """
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir(parents=True, exist_ok=True)
+    (hermes_home / "auth.json").write_text(json.dumps({
+        "version": 1,
+        "providers": {
+            "openai-codex": {
+                "tokens": {"access_token": "shared-at", "refresh_token": "shared-rt"},
+                "last_refresh": "2026-01-01T00:00:00Z",
+                "auth_mode": "chatgpt",
+            },
+        },
+        "credential_pool": {
+            "openai-codex": [
+                {
+                    "id": "seeded",
+                    "source": "device_code",
+                    "auth_type": "oauth",
+                    "access_token": "shared-at",
+                    "refresh_token": "shared-rt",
+                },
+                {
+                    "id": "legacy",
+                    "label": "legacy-alias",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    # Token material matches the singleton — this is a true
+                    # alias from the #33000 workaround era.
+                    "access_token": "shared-at",
+                    "refresh_token": "shared-rt",
+                    "last_status": "exhausted",
+                    "last_error_code": 401,
+                    "last_error_reason": "token_invalidated",
+                },
+            ],
+        },
+    }))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    _save_codex_tokens(
+        {"access_token": "fresh-at", "refresh_token": "fresh-rt"},
+        last_refresh="2026-06-05T00:00:00Z",
+    )
+
+    auth = json.loads((hermes_home / "auth.json").read_text())
+    pool = auth["credential_pool"]["openai-codex"]
+
+    # Singleton: refreshed.
+    seeded = next(e for e in pool if e["source"] == "device_code")
+    assert seeded["access_token"] == "fresh-at"
+
+    # Legacy alias: still refreshed (preserves #33538 fix).
+    legacy = next(e for e in pool if e["id"] == "legacy")
+    assert legacy["access_token"] == "fresh-at"
+    assert legacy["refresh_token"] == "fresh-rt"
+    assert legacy["last_refresh"] == "2026-06-05T00:00:00Z"
+    # Error markers cleared on the refreshed entry.
+    assert legacy["last_status"] is None
+    assert legacy["last_error_code"] is None
+    assert legacy["last_error_reason"] is None
+
+
+def test_save_codex_tokens_handles_missing_previous_singleton_tokens(tmp_path, monkeypatch):
+    """First-ever Codex save (no prior singleton tokens) must not crash.
+
+    Edge case: a user has only pool entries (e.g. via direct auth.json edit
+    or a partial state from a corrupted upgrade), no `providers.openai-codex.tokens`
+    block at all.  The previous-singleton-tokens guard must handle missing
+    state gracefully — fall back to "no previous tokens", which means no
+    pool entry can be a true alias and only the singleton-seeded entry gets
+    written.
+    """
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir(parents=True, exist_ok=True)
+    (hermes_home / "auth.json").write_text(json.dumps({
+        "version": 1,
+        "providers": {},
+        "credential_pool": {
+            "openai-codex": [
+                {
+                    "id": "preexisting",
+                    "label": "pre-existing-manual",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "preexisting-at",
+                    "refresh_token": "preexisting-rt",
+                },
+            ],
+        },
+    }))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    _save_codex_tokens(
+        {"access_token": "first-at", "refresh_token": "first-rt"},
+        last_refresh="2026-06-05T00:00:00Z",
+    )
+
+    auth = json.loads((hermes_home / "auth.json").read_text())
+    pool = auth["credential_pool"]["openai-codex"]
+    # Pre-existing independent entry with no relationship to a (now-new)
+    # singleton MUST be preserved.
+    pre = next(e for e in pool if e["id"] == "preexisting")
+    assert pre["access_token"] == "preexisting-at"
+    assert pre["refresh_token"] == "preexisting-rt"
+
+
+def test_save_codex_tokens_alias_match_uses_access_token_only(tmp_path, monkeypatch):
+    """A manual entry counts as an alias if its access_token matches the
+    previous singleton access_token, regardless of refresh_token presence.
+
+    Some legacy entries (older auth.json schemas, pre-refresh-token versions)
+    have access_token but no refresh_token.  These should still be treated as
+    aliases when the access_token matches.
+    """
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir(parents=True, exist_ok=True)
+    (hermes_home / "auth.json").write_text(json.dumps({
+        "version": 1,
+        "providers": {
+            "openai-codex": {
+                "tokens": {"access_token": "shared-at", "refresh_token": "shared-rt"},
+                "auth_mode": "chatgpt",
+            },
+        },
+        "credential_pool": {
+            "openai-codex": [
+                {
+                    "id": "alias-no-refresh",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "shared-at",
+                    # No refresh_token at all — legacy schema.
+                },
+            ],
+        },
+    }))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    _save_codex_tokens(
+        {"access_token": "new-at", "refresh_token": "new-rt"},
+        last_refresh="2026-06-05T00:00:00Z",
+    )
+
+    auth = json.loads((hermes_home / "auth.json").read_text())
+    pool = auth["credential_pool"]["openai-codex"]
+    alias = next(e for e in pool if e["id"] == "alias-no-refresh")
+    # Treated as alias → refreshed with new tokens.
+    assert alias["access_token"] == "new-at"
+    assert alias["refresh_token"] == "new-rt"
+
+
+def test_save_codex_tokens_clears_error_markers_only_on_refreshed_entries(tmp_path, monkeypatch):
+    """Error markers must be cleared only on entries that were actually
+    refreshed by this re-auth.  Independent ``manual:device_code`` entries
+    with their own stale-error markers must be left alone (their stale state
+    is not the current re-auth's business).
+    """
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir(parents=True, exist_ok=True)
+    (hermes_home / "auth.json").write_text(json.dumps({
+        "version": 1,
+        "providers": {
+            "openai-codex": {
+                "tokens": {"access_token": "acctA-at", "refresh_token": "acctA-rt"},
+                "auth_mode": "chatgpt",
+            },
+        },
+        "credential_pool": {
+            "openai-codex": [
+                {
+                    "id": "seeded",
+                    "source": "device_code",
+                    "auth_type": "oauth",
+                    "access_token": "acctA-at",
+                    "refresh_token": "acctA-rt",
+                    "last_status": "exhausted",
+                    "last_error_code": 401,
+                },
+                {
+                    "id": "acctB",
+                    "source": "manual:device_code",
+                    "auth_type": "oauth",
+                    "access_token": "acctB-at",
+                    "refresh_token": "acctB-rt",
+                    "last_status": "exhausted",
+                    "last_error_code": 429,
+                    "last_error_reason": "quota_exhausted",
+                },
+            ],
+        },
+    }))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    _save_codex_tokens(
+        {"access_token": "fresh-at", "refresh_token": "fresh-rt"},
+        last_refresh="2026-06-05T00:00:00Z",
+    )
+
+    auth = json.loads((hermes_home / "auth.json").read_text())
+    pool = auth["credential_pool"]["openai-codex"]
+
+    # Singleton: refreshed AND error markers cleared.
+    seeded = next(e for e in pool if e["id"] == "seeded")
+    assert seeded["access_token"] == "fresh-at"
+    assert seeded["last_status"] is None
+    assert seeded["last_error_code"] is None
+
+    # Independent acctB: NOT refreshed AND error markers NOT cleared.
+    # (Its 429 quota state belongs to acctB's own account, not acctA's re-auth.)
+    acctB = next(e for e in pool if e["id"] == "acctB")
+    assert acctB["access_token"] == "acctB-at"  # not overwritten
+    assert acctB["last_status"] == "exhausted"  # not cleared
+    assert acctB["last_error_code"] == 429
+    assert acctB["last_error_reason"] == "quota_exhausted"
+
+
 def test_import_codex_cli_tokens(tmp_path, monkeypatch):
     codex_home = tmp_path / "codex-cli"
     codex_home.mkdir(parents=True, exist_ok=True)

tests/hermes_cli/test_auth_commands.py

@@ -397,15 +397,92 @@ def test_auth_add_codex_oauth_persists_pool_entry(tmp_path, monkeypatch):
 
     payload = json.loads((tmp_path / "hermes" / "auth.json").read_text())
     entries = payload["credential_pool"]["openai-codex"]
-    entry = next(item for item in entries if item["source"] == "device_code")
+    # The add path now creates a distinct, self-contained ``manual:device_code``
+    # pool entry per account instead of routing through the singleton save path
+    # (which collapsed multiple accounts into the latest login — #39236).
+    entry = next(item for item in entries if item["source"] == "manual:device_code")
     assert payload["active_provider"] == "openai-codex"
-    assert payload["providers"]["openai-codex"]["tokens"]["access_token"] == token
+    # No singleton ``providers.openai-codex`` block is written by the add path.
+    assert "openai-codex" not in payload.get("providers", {})
     assert entry["label"] == "codex@example.com"
-    assert entry["source"] == "device_code"
+    assert entry["source"] == "manual:device_code"
+    assert entry["access_token"] == token
     assert entry["refresh_token"] == "refresh-token"
     assert entry["base_url"] == "https://chatgpt.com/backend-api/codex"
 
 
+def test_auth_add_codex_oauth_keeps_distinct_pool_accounts(tmp_path, monkeypatch):
+    """Two ``hermes auth add openai-codex`` runs for different ChatGPT
+    accounts must produce two independent pool entries with distinct tokens.
+
+    Regression for #39236: the add path used to route through the singleton
+    ``_save_codex_tokens`` save, so the second login overwrote the first
+    account's singleton-mirrored ``device_code`` entry instead of adding a
+    second independent one. ``hermes auth list`` showed two labels sharing
+    one token pair, and rotation silently always used the latest account.
+    """
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
+    _write_auth_store(tmp_path, {"version": 1, "providers": {}})
+    first_token = _jwt_with_email("first-codex@example.com")
+    second_token = _jwt_with_email("second-codex@example.com")
+    logins = iter(
+        [
+            {
+                "tokens": {
+                    "access_token": first_token,
+                    "refresh_token": "first-refresh-token",
+                },
+                "base_url": "https://chatgpt.com/backend-api/codex",
+                "last_refresh": "2026-03-23T10:00:00Z",
+            },
+            {
+                "tokens": {
+                    "access_token": second_token,
+                    "refresh_token": "second-refresh-token",
+                },
+                "base_url": "https://chatgpt.com/backend-api/codex",
+                "last_refresh": "2026-03-23T10:05:00Z",
+            },
+        ]
+    )
+    monkeypatch.setattr("hermes_cli.auth._codex_device_code_login", lambda: next(logins))
+
+    from hermes_cli.auth_commands import auth_add_command
+    from agent.credential_pool import load_pool
+
+    class _Args:
+        provider = "openai-codex"
+        auth_type = "oauth"
+        api_key = None
+        label = None
+
+    auth_add_command(_Args())
+    auth_add_command(_Args())
+
+    pool = load_pool("openai-codex")
+    entries = pool.entries()
+
+    assert [entry.source for entry in entries] == [
+        "manual:device_code",
+        "manual:device_code",
+    ]
+    assert [entry.label for entry in entries] == [
+        "first-codex@example.com",
+        "second-codex@example.com",
+    ]
+    assert [entry.access_token for entry in entries] == [first_token, second_token]
+    assert [entry.refresh_token for entry in entries] == [
+        "first-refresh-token",
+        "second-refresh-token",
+    ]
+
+    payload = json.loads((tmp_path / "hermes" / "auth.json").read_text())
+    # No singleton block — the add path is now pool-only.
+    assert "openai-codex" not in payload.get("providers", {})
+    # First add activated the provider; second add left it as-is.
+    assert payload["active_provider"] == "openai-codex"
+
+
 def test_auth_add_xai_oauth_sets_active_provider(tmp_path, monkeypatch):
     """hermes auth add xai-oauth must write providers singleton and set active_provider.
 
@@ -1313,9 +1390,9 @@ def test_auth_add_codex_clears_suppression_marker(tmp_path, monkeypatch):
     payload = json.loads((hermes_home / "auth.json").read_text())
     # Suppression marker must be cleared
     assert "openai-codex" not in payload.get("suppressed_sources", {})
-    # New pool entry must be present
+    # New pool entry must be present (distinct manual:device_code entry — #39236)
     entries = payload["credential_pool"]["openai-codex"]
-    assert any(e["source"] == "device_code" for e in entries)
+    assert any(e["source"] == "manual:device_code" for e in entries)
     assert payload["active_provider"] == "openai-codex"
 
 

tests/hermes_cli/test_gui_command.py

@@ -519,3 +519,78 @@ def test_gui_does_not_retry_when_purge_finds_nothing(tmp_path, monkeypatch, caps
     mock_purge.assert_called_once()
     assert mock_run.call_count == 1
     assert "Desktop GUI build failed" in capsys.readouterr().out
+
+
+class _FakeProc:
+    """Minimal psutil.Process stand-in for the lock-breaker tests."""
+
+    def __init__(self, pid: int, exe: str | None):
+        self.pid = pid
+        self.info = {"pid": pid, "exe": exe}
+        self.terminated = False
+        self.killed = False
+
+    def terminate(self):
+        self.terminated = True
+
+    def kill(self):
+        self.killed = True
+
+
+def test_stop_desktop_build_lock_noop_off_windows(tmp_path, monkeypatch):
+    """POSIX can unlink a running binary, so the helper is a no-op there."""
+    desktop_dir = tmp_path / "apps" / "desktop"
+    exe = desktop_dir / "release" / "linux-unpacked" / "hermes"
+    exe.parent.mkdir(parents=True)
+    exe.write_text("", encoding="utf-8")
+    monkeypatch.setattr(cli_main.sys, "platform", "linux")
+
+    proc = _FakeProc(4321, str(exe))
+    with patch("psutil.process_iter", return_value=[proc]) as it:
+        assert cli_main._stop_desktop_processes_locking_build(desktop_dir) == []
+    it.assert_not_called()
+    assert proc.terminated is False
+
+
+def test_stop_desktop_build_lock_terminates_only_release_procs(tmp_path, monkeypatch):
+    desktop_dir = tmp_path / "apps" / "desktop"
+    release = desktop_dir / "release" / "win-unpacked"
+    release.mkdir(parents=True)
+    locker_exe = release / "Hermes.exe"
+    locker_exe.write_text("", encoding="utf-8")
+    other_exe = tmp_path / "elsewhere" / "Hermes.exe"
+    other_exe.parent.mkdir(parents=True)
+    other_exe.write_text("", encoding="utf-8")
+
+    monkeypatch.setattr(cli_main.sys, "platform", "win32")
+    monkeypatch.setattr(cli_main.os, "getpid", lambda: 999)
+
+    locker = _FakeProc(101, str(locker_exe))
+    unrelated = _FakeProc(102, str(other_exe))
+    selfish = _FakeProc(999, str(locker_exe))  # our own PID — never killed
+    no_exe = _FakeProc(103, None)
+
+    captured = {}
+
+    def _wait(procs, timeout=None):
+        captured["waited"] = list(procs)
+        return procs, []
+
+    with patch("psutil.process_iter", return_value=[locker, unrelated, selfish, no_exe]), \
+         patch("psutil.wait_procs", side_effect=_wait):
+        stopped = cli_main._stop_desktop_processes_locking_build(desktop_dir)
+
+    assert stopped == [101]
+    assert locker.terminated is True
+    assert unrelated.terminated is False
+    assert selfish.terminated is False
+    assert captured["waited"] == [locker]
+
+
+def test_stop_desktop_build_lock_no_release_dir(tmp_path, monkeypatch):
+    desktop_dir = tmp_path / "apps" / "desktop"
+    desktop_dir.mkdir(parents=True)
+    monkeypatch.setattr(cli_main.sys, "platform", "win32")
+    with patch("psutil.process_iter") as it:
+        assert cli_main._stop_desktop_processes_locking_build(desktop_dir) == []
+    it.assert_not_called()

tests/hermes_cli/test_update_autostash.py

@@ -350,7 +350,7 @@ def test_cmd_update_retries_optional_extras_individually_when_all_fails(monkeypa
 
     def fake_run(cmd, **kwargs):
         recorded.append(cmd)
-        if cmd == ["git", "fetch", "origin"]:
+        if cmd == ["git", "fetch", "origin", "main"]:
             return SimpleNamespace(stdout="", stderr="", returncode=0)
         if cmd == ["git", "rev-parse", "--abbrev-ref", "HEAD"]:
             return SimpleNamespace(stdout="main\n", stderr="", returncode=0)
@@ -399,7 +399,7 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
 
     def fake_run(cmd, **kwargs):
         recorded.append(cmd)
-        if cmd == ["git", "fetch", "origin"]:
+        if cmd == ["git", "fetch", "origin", "main"]:
             return SimpleNamespace(stdout="", stderr="", returncode=0)
         if cmd == ["git", "rev-parse", "--abbrev-ref", "HEAD"]:
             return SimpleNamespace(stdout="main\n", stderr="", returncode=0)
@@ -630,6 +630,23 @@ def test_cmd_update_no_checkout_when_already_on_main(monkeypatch, tmp_path):
     assert len(checkout_calls) == 0
 
 
+def test_cmd_update_fetch_is_scoped_to_target_branch(monkeypatch, tmp_path):
+    """The update fetch must name the target branch. A bare `git fetch origin`
+    pulls every ref, and this repo has thousands of auto-generated branches, so
+    an unscoped fetch can stall for minutes on a non-single-branch checkout."""
+    _setup_update_mocks(monkeypatch, tmp_path)
+    monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/uv" if name == "uv" else None)
+
+    side_effect, recorded = _make_update_side_effect()
+    monkeypatch.setattr(hermes_main.subprocess, "run", side_effect)
+
+    hermes_main.cmd_update(SimpleNamespace())
+
+    fetch_calls = [c for c in recorded if "fetch" in c]
+    assert fetch_calls == [["git", "fetch", "origin", "main"]]
+    assert ["git", "fetch", "origin"] not in recorded
+
+
 # ---------------------------------------------------------------------------
 # Fetch failure — friendly error messages
 # ---------------------------------------------------------------------------

tests/hermes_cli/test_web_server_pty_import.py

@@ -0,0 +1,83 @@
+"""Test the platform-branched PTY bridge import in hermes_cli.web_server.
+
+The /api/pty WebSocket handler in web_server.py picks its bridge at import
+time via ``sys.platform.startswith("win")`` — Windows gets the ConPTY
+backend, POSIX gets the fcntl/termios one.  Both branches must:
+
+  1. Expose ``PtyBridge`` as the bridge class (or None) and
+     ``PtyUnavailableError`` as an exception class.
+  2. Set ``_PTY_BRIDGE_AVAILABLE`` correctly.
+  3. Never raise at import time when the platform-native dependency is
+     missing — the dashboard's non-chat tabs must keep loading.
+
+This test asserts the live state on whichever platform CI runs on, plus a
+source-text check confirming the branch shape is preserved so a future
+refactor can't accidentally collapse it back to a POSIX-only import.
+"""
+
+from __future__ import annotations
+
+import sys
+
+import pytest
+
+from hermes_cli import web_server
+
+
+def test_web_server_exposes_pty_bridge_symbols():
+    """The two symbols /api/pty consumes must always exist."""
+    assert hasattr(web_server, "PtyBridge")
+    assert hasattr(web_server, "PtyUnavailableError")
+    assert hasattr(web_server, "_PTY_BRIDGE_AVAILABLE")
+    # PtyUnavailableError is always an exception class — either the real
+    # one from the platform bridge, or the local fallback class.
+    assert isinstance(web_server.PtyUnavailableError, type)
+    assert issubclass(web_server.PtyUnavailableError, BaseException)
+
+
+@pytest.mark.skipif(not sys.platform.startswith("win"), reason="Windows-only")
+def test_web_server_uses_win_pty_bridge_on_windows():
+    """On native Windows, web_server.PtyBridge must be the ConPTY backend."""
+    from hermes_cli.win_pty_bridge import WinPtyBridge
+
+    assert web_server.PtyBridge is WinPtyBridge
+    assert web_server._PTY_BRIDGE_AVAILABLE is True
+    # And the error class must be the one from the same module so isinstance
+    # checks in /api/pty's spawn fallback path actually work.
+    from hermes_cli.win_pty_bridge import PtyUnavailableError as WinErr
+
+    assert web_server.PtyUnavailableError is WinErr
+
+
+@pytest.mark.skipif(sys.platform.startswith("win"), reason="POSIX-only")
+def test_web_server_uses_posix_pty_bridge_on_posix():
+    """On POSIX, the bridge must be the fcntl/termios PtyBridge."""
+    from hermes_cli.pty_bridge import PtyBridge as PosixBridge
+    from hermes_cli.pty_bridge import PtyUnavailableError as PosixErr
+
+    assert web_server.PtyBridge is PosixBridge
+    assert web_server._PTY_BRIDGE_AVAILABLE is True
+    assert web_server.PtyUnavailableError is PosixErr
+
+
+def test_pty_bridge_import_block_is_platform_branched():
+    """Source-level guard: a future refactor must not collapse the branch
+    back to a single POSIX import.  Reads web_server.py directly so this
+    fails the same way on every OS — the runtime symbol checks above can
+    pass even when the branch shape is wrong on the current platform."""
+    src = pytest.importorskip("inspect").getsource(web_server)
+    # The shape we expect (from PR #39913):
+    #
+    #   if sys.platform.startswith("win"):
+    #       try:
+    #           from hermes_cli.win_pty_bridge import WinPtyBridge as PtyBridge, ...
+    #       except ImportError:
+    #           PtyBridge = None
+    #           ...
+    #   else:
+    #       try:
+    #           from hermes_cli.pty_bridge import PtyBridge, PtyUnavailableError
+    #       ...
+    assert 'sys.platform.startswith("win")' in src or "sys.platform.startswith('win')" in src
+    assert "from hermes_cli.win_pty_bridge import" in src
+    assert "from hermes_cli.pty_bridge import" in src

tests/hermes_cli/test_win_pty_bridge.py

@@ -0,0 +1,315 @@
+"""Unit tests for hermes_cli.win_pty_bridge — ConPTY spawning + byte forwarding.
+
+Windows-only counterpart to tests/hermes_cli/test_pty_bridge.py.  Drives
+``WinPtyBridge`` with minimal Windows processes (``cmd.exe``, ``python -c …``)
+to verify it behaves like a PTY you can read/write/resize/close, then a small
+set of platform-fallback assertions (``is_available``, ``PtyUnavailableError``)
+that run on every OS so the import surface stays exercised in CI.
+
+The bridge is the ConPTY backend behind the dashboard ``/chat`` tab — see
+``hermes_cli/web_server.py`` ``/api/pty`` handler — so these tests are the
+unit-level half of the integration check that the dashboard chat pane is
+actually live on native Windows.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import time
+
+import pytest
+
+# WinPtyBridge can be imported on every platform — ``is_available`` just
+# returns False when pywinpty isn't usable.  Importing the module itself
+# must never raise, otherwise the web_server import branch becomes a trap.
+from hermes_cli.win_pty_bridge import PtyUnavailableError, WinPtyBridge
+
+windows_only = pytest.mark.skipif(
+    not sys.platform.startswith("win"),
+    reason="ConPTY bridge is Windows-only",
+)
+
+
+def _read_until(bridge: WinPtyBridge, needle: bytes, timeout: float = 10.0) -> bytes:
+    """Accumulate PTY output until we see ``needle`` or time out.
+
+    Mirrors the helper in test_pty_bridge.py so failures look familiar.
+    """
+    deadline = time.monotonic() + timeout
+    buf = bytearray()
+    while time.monotonic() < deadline:
+        chunk = bridge.read(timeout=0.2)
+        if chunk is None:
+            break
+        buf.extend(chunk)
+        if needle in buf:
+            return bytes(buf)
+    return bytes(buf)
+
+
+# ---------------------------------------------------------------------------
+# Cross-platform fallback semantics
+# ---------------------------------------------------------------------------
+
+
+class TestWinPtyBridgeUnavailable:
+    """Module-level surface that must stay importable on every OS so the
+    web_server platform branch doesn't blow up at import time when pywinpty
+    is missing or the host isn't Windows."""
+
+    def test_error_is_importable_and_carries_message(self):
+        err = PtyUnavailableError("conpty missing")
+        assert "conpty" in str(err)
+
+    def test_bridge_class_is_importable(self):
+        # The platform-branched import in web_server.py relies on this:
+        #     from hermes_cli.win_pty_bridge import WinPtyBridge, PtyUnavailableError
+        # Both symbols must always exist; ``is_available()`` is the gate.
+        assert WinPtyBridge is not None
+        assert callable(WinPtyBridge.is_available)
+
+    @pytest.mark.skipif(sys.platform.startswith("win"), reason="non-Windows only")
+    def test_spawn_raises_unavailable_off_windows(self):
+        with pytest.raises(PtyUnavailableError):
+            WinPtyBridge.spawn(["true"])
+
+
+# ---------------------------------------------------------------------------
+# Windows-only end-to-end behaviour
+# ---------------------------------------------------------------------------
+
+
+@windows_only
+class TestWinPtyBridgeSpawn:
+    def test_is_available_on_windows(self):
+        assert WinPtyBridge.is_available() is True
+
+    def test_spawn_returns_bridge_with_pid(self):
+        bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
+        try:
+            assert bridge.pid > 0
+        finally:
+            bridge.close()
+
+    def test_spawn_raises_on_missing_argv0(self, tmp_path):
+        # pywinpty wraps CreateProcessW failures; surface as OSError / RuntimeError.
+        bogus = str(tmp_path / "definitely-not-a-real-binary.exe")
+        with pytest.raises((FileNotFoundError, OSError, RuntimeError, PtyUnavailableError)):
+            WinPtyBridge.spawn([bogus])
+
+
+@windows_only
+class TestWinPtyBridgeIO:
+    def test_reads_child_stdout(self):
+        bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo hermes-ok"])
+        try:
+            output = _read_until(bridge, b"hermes-ok")
+            assert b"hermes-ok" in output
+        finally:
+            bridge.close()
+
+    def test_write_sends_to_child_stdin(self):
+        # python -c reads stdin, echoes a marker, exits.  More reliable than
+        # ``cat`` (not on Windows) and doesn't depend on a particular shell.
+        script = (
+            "import sys; "
+            "line = sys.stdin.readline().strip(); "
+            "sys.stdout.write('GOT:' + line + '\\n'); "
+            "sys.stdout.flush()"
+        )
+        bridge = WinPtyBridge.spawn([sys.executable, "-c", script])
+        try:
+            bridge.write(b"hello-pty\r\n")
+            output = _read_until(bridge, b"GOT:hello-pty")
+            assert b"GOT:hello-pty" in output
+        finally:
+            bridge.close()
+
+    def test_write_after_close_is_silent(self):
+        bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
+        bridge.close()
+        # Must not raise — the dashboard WebSocket reader sometimes writes
+        # a final keystroke after the user has already closed the tab.
+        bridge.write(b"ignored")
+
+    def test_read_returns_none_after_child_exits(self):
+        bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "echo done"])
+        try:
+            _read_until(bridge, b"done")
+            # Give the child a beat to exit, then drain until EOF.
+            deadline = time.monotonic() + 5.0
+            while bridge.is_alive() and time.monotonic() < deadline:
+                bridge.read(timeout=0.1)
+            got_none = False
+            for _ in range(20):
+                if bridge.read(timeout=0.1) is None:
+                    got_none = True
+                    break
+            assert got_none, "WinPtyBridge.read did not return None after child EOF"
+        finally:
+            bridge.close()
+
+
+@windows_only
+class TestWinPtyBridgeResize:
+    def test_resize_does_not_raise_on_live_child(self):
+        # ConPTY exposes no ioctl-equivalent for reading the child's current
+        # winsize from Python land, so we can't verify the new dimensions
+        # the way the POSIX test does (which reads TIOCGWINSZ).  What we
+        # CAN guarantee is what the dashboard depends on: ``resize`` never
+        # raises, the bridge stays alive, and subsequent I/O still works.
+        bridge = WinPtyBridge.spawn(
+            [sys.executable, "-c", "import time; time.sleep(1.0)"],
+            cols=80,
+            rows=24,
+        )
+        try:
+            bridge.resize(cols=123, rows=45)
+            assert bridge.is_alive()
+        finally:
+            bridge.close()
+
+    def test_resize_clamps_garbage_dimensions(self):
+        # Mirror the POSIX clamp test: a broken winsize probe must never
+        # propagate to the ConPTY API.  131072 > unsigned short max — the
+        # bridge has to coerce it down without raising.
+        bridge = WinPtyBridge.spawn(
+            [sys.executable, "-c", "import time; time.sleep(1.0)"],
+            cols=80,
+            rows=24,
+        )
+        try:
+            bridge.resize(cols=131072, rows=1)  # must not raise
+            bridge.resize(cols=0, rows=-5)      # nor this
+            assert bridge.is_alive()
+        finally:
+            bridge.close()
+
+    def test_resize_after_close_is_silent(self):
+        bridge = WinPtyBridge.spawn(["cmd.exe", "/c", "exit 0"])
+        bridge.close()
+        # Must not raise — closed bridges still receive late resize escapes
+        # from xterm.js when the browser tab is closed mid-stream.
+        bridge.resize(cols=100, rows=40)
+
+
+@windows_only
+class TestClampDimension:
+    """The clamp helper is the load-bearing piece — the dashboard sends
+    untrusted winsize values straight from xterm.js, and pywinpty's
+    setwinsize will happily raise on out-of-range u16 values."""
+
+    def test_clamps_above_max(self):
+        from hermes_cli.win_pty_bridge import _MAX_COLS, _MAX_ROWS, _clamp
+
+        assert _clamp(131072, _MAX_COLS) == _MAX_COLS
+        assert _clamp(131072, _MAX_ROWS) == _MAX_ROWS
+
+    def test_floors_at_one(self):
+        from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
+
+        assert _clamp(0, _MAX_COLS) == 1
+        assert _clamp(-5, _MAX_COLS) == 1
+
+    def test_passes_through_sane_values(self):
+        from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
+
+        assert _clamp(80, _MAX_COLS) == 80
+        assert _clamp(2000, _MAX_COLS) == 2000
+
+    def test_non_numeric_falls_back_to_min(self):
+        from hermes_cli.win_pty_bridge import _MAX_COLS, _clamp
+
+        assert _clamp(None, _MAX_COLS) == 1  # type: ignore[arg-type]
+        assert _clamp("not-a-number", _MAX_COLS) == 1  # type: ignore[arg-type]
+        assert _clamp(float("nan"), _MAX_COLS) == 1  # type: ignore[arg-type]
+        assert _clamp(float("inf"), _MAX_COLS) == 1  # type: ignore[arg-type]
+
+
+@windows_only
+class TestWinPtyBridgeClose:
+    def test_close_is_idempotent(self):
+        bridge = WinPtyBridge.spawn(
+            [sys.executable, "-c", "import time; time.sleep(30)"]
+        )
+        bridge.close()
+        bridge.close()  # must not raise
+        assert not bridge.is_alive()
+
+    def test_close_terminates_long_running_child(self):
+        bridge = WinPtyBridge.spawn(
+            [sys.executable, "-c", "import time; time.sleep(30)"]
+        )
+        pid = bridge.pid
+        assert bridge.is_alive(), f"child pid {pid} not alive before close"
+        bridge.close()
+        # The bridge itself reports liveness via pywinpty.isalive(), which is
+        # the same probe the dashboard PTY reader uses to decide when to stop
+        # forwarding bytes — verifying that flips to False is the contract
+        # that matters for /api/pty.
+        deadline = time.monotonic() + 5.0
+        while bridge.is_alive() and time.monotonic() < deadline:
+            time.sleep(0.1)
+        assert not bridge.is_alive(), (
+            f"WinPtyBridge.is_alive() still True after close(); pid {pid}"
+        )
+
+
+@windows_only
+class TestWinPtyBridgeEnv:
+    def test_cwd_is_respected(self, tmp_path):
+        bridge = WinPtyBridge.spawn(
+            [sys.executable, "-c", "import os; print(os.getcwd())"],
+            cwd=str(tmp_path),
+        )
+        try:
+            # Path is case-insensitive on Windows; compare lowercased.
+            needle_resolved = str(tmp_path.resolve()).lower().encode()
+            deadline = time.monotonic() + 5.0
+            buf = bytearray()
+            while time.monotonic() < deadline:
+                chunk = bridge.read(timeout=0.2)
+                if chunk is None:
+                    break
+                buf.extend(chunk)
+                if needle_resolved in bytes(buf).lower():
+                    break
+            assert needle_resolved in bytes(buf).lower(), (
+                f"cwd {tmp_path!s} not echoed by child; got {bytes(buf)!r}"
+            )
+        finally:
+            bridge.close()
+
+    def test_env_is_forwarded(self):
+        bridge = WinPtyBridge.spawn(
+            [
+                sys.executable,
+                "-c",
+                "import os; print('HERMES_PTY_TEST=' + os.environ.get('HERMES_PTY_TEST',''))",
+            ],
+            env={**os.environ, "HERMES_PTY_TEST": "pty-env-works"},
+        )
+        try:
+            output = _read_until(bridge, b"pty-env-works")
+            assert b"pty-env-works" in output
+        finally:
+            bridge.close()
+
+    def test_spawn_defaults_term_when_not_set(self):
+        # The bridge should set TERM=xterm-256color when the caller's env
+        # doesn't already carry one — xterm.js expects ANSI/SGR sequences.
+        env = {k: v for k, v in os.environ.items() if k.upper() != "TERM"}
+        bridge = WinPtyBridge.spawn(
+            [
+                sys.executable,
+                "-c",
+                "import os; print('TERM=' + os.environ.get('TERM',''))",
+            ],
+            env=env,
+        )
+        try:
+            output = _read_until(bridge, b"TERM=")
+            assert b"TERM=xterm-256color" in output
+        finally:
+            bridge.close()

tests/plugins/platforms/photon/test_auth.py

@@ -0,0 +1,283 @@
+"""Tests for the Photon auth module (device login + project + user creation)."""
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import Any, Dict
+
+import pytest
+
+from plugins.platforms.photon import auth as photon_auth
+
+
+# ---------------------------------------------------------------------------
+# Fake httpx — we don't want to hit the real Photon API in unit tests.
+
+class _FakeResponse:
+    def __init__(
+        self,
+        *,
+        status: int = 200,
+        json_body: Any = None,
+        headers: Dict[str, str] | None = None,
+        text: str = "",
+    ) -> None:
+        self.status_code = status
+        self._json = json_body if json_body is not None else {}
+        self.headers = headers or {}
+        self.text = text
+
+    def json(self) -> Any:
+        return self._json
+
+    def raise_for_status(self) -> None:
+        if self.status_code >= 400:
+            raise RuntimeError(f"HTTP {self.status_code}")
+
+
+@pytest.fixture
+def tmp_hermes_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    home = tmp_path / "hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    # The auth module memoises by reading get_hermes_home at call time
+    # so the env var is what matters.
+    return home
+
+
+def test_store_and_load_photon_token(tmp_hermes_home: Path) -> None:
+    photon_auth.store_photon_token("abc123def456")
+    assert photon_auth.load_photon_token() == "abc123def456"
+
+    auth_json = json.loads((tmp_hermes_home / "auth.json").read_text())
+    assert "credential_pool" in auth_json
+    assert auth_json["credential_pool"]["photon"][0]["access_token"] == "abc123def456"
+
+
+def test_store_and_load_project_credentials(tmp_hermes_home: Path) -> None:
+    photon_auth.store_project_credentials(
+        "proj-uuid", "secret-key", name="Test Project",
+    )
+    pid, secret = photon_auth.load_project_credentials()
+    assert pid == "proj-uuid"
+    assert secret == "secret-key"
+
+
+def test_load_project_credentials_env_override(
+    tmp_hermes_home: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    photon_auth.store_project_credentials("from-file", "secret-file")
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "from-env")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "secret-env")
+    pid, secret = photon_auth.load_project_credentials()
+    assert pid == "from-env"
+    assert secret == "secret-env"
+
+
+def test_request_device_code(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: Dict[str, Any] = {}
+
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        captured["url"] = url
+        captured["body"] = json
+        return _FakeResponse(json_body={
+            "device_code": "dev-code-xyz",
+            "user_code": "ABCD-1234",
+            "verification_uri": "https://app.photon.codes/device",
+            "verification_uri_complete": "https://app.photon.codes/device?code=ABCD-1234",
+            "expires_in": 600,
+            "interval": 5,
+        })
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+
+    code = photon_auth.request_device_code()
+    assert code.device_code == "dev-code-xyz"
+    assert code.user_code == "ABCD-1234"
+    assert code.expires_in == 600
+    assert "/api/auth/device/code" in captured["url"]
+    assert captured["body"]["client_id"] == "hermes-agent"
+
+
+def test_poll_for_token_via_header(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Token from set-auth-token header is the documented mechanism."""
+
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        return _FakeResponse(
+            status=200,
+            json_body={"session": {}, "user": {}},
+            headers={"set-auth-token": "bearer-xyz"},
+        )
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+
+    code = photon_auth.DeviceCode(
+        device_code="d", user_code="u",
+        verification_uri="https://x", verification_uri_complete=None,
+        expires_in=10, interval=0,
+    )
+    token = photon_auth.poll_for_token(code, interval=0, timeout=2)
+    assert token == "bearer-xyz"
+
+
+def test_poll_for_token_via_body_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+    """If the header is absent we fall back to session.access_token."""
+
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        return _FakeResponse(
+            status=200,
+            json_body={"session": {"access_token": "from-body"}, "user": {}},
+        )
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    code = photon_auth.DeviceCode(
+        device_code="d", user_code="u",
+        verification_uri="https://x", verification_uri_complete=None,
+        expires_in=10, interval=0,
+    )
+    assert photon_auth.poll_for_token(code, interval=0, timeout=2) == "from-body"
+
+
+def test_poll_for_token_propagates_access_denied(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    def fake_post(url: str, *, json: Dict[str, Any], timeout: float) -> _FakeResponse:
+        return _FakeResponse(
+            status=400, json_body={"error": "access_denied"},
+        )
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    code = photon_auth.DeviceCode(
+        device_code="d", user_code="u",
+        verification_uri="https://x", verification_uri_complete=None,
+        expires_in=10, interval=0,
+    )
+    with pytest.raises(RuntimeError, match="access_denied"):
+        photon_auth.poll_for_token(code, interval=0, timeout=2)
+
+
+def test_create_user_rejects_invalid_phone() -> None:
+    with pytest.raises(ValueError, match="E.164"):
+        photon_auth.create_user(
+            "proj", "secret", phone_number="not-a-number",
+        )
+
+
+def test_create_user_posts_shared_type(monkeypatch: pytest.MonkeyPatch) -> None:
+    captured: Dict[str, Any] = {}
+
+    def fake_post(url: str, *, json: Dict[str, Any], auth: tuple, timeout: float) -> _FakeResponse:
+        captured["url"] = url
+        captured["body"] = json
+        captured["auth"] = auth
+        return _FakeResponse(json_body={
+            "succeed": True,
+            "data": {
+                "id": "user-uuid",
+                "phoneNumber": "+15551234567",
+                "assignedPhoneNumber": "+15559999999",
+            },
+        })
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    user = photon_auth.create_user(
+        "proj-id", "proj-secret",
+        phone_number="+15551234567",
+    )
+    assert user["assignedPhoneNumber"] == "+15559999999"
+    assert captured["auth"] == ("proj-id", "proj-secret")
+    assert captured["body"]["type"] == "shared"
+    assert captured["body"]["phoneNumber"] == "+15551234567"
+    assert "/projects/proj-id/users/" in captured["url"]
+
+
+def test_register_webhook_surfaces_secret(monkeypatch: pytest.MonkeyPatch) -> None:
+    def fake_post(url: str, *, json: Dict[str, Any], auth: tuple, timeout: float) -> _FakeResponse:
+        return _FakeResponse(json_body={
+            "succeed": True,
+            "data": {
+                "id": "wh-uuid",
+                "webhookUrl": json["webhookUrl"],
+                "signingSecret": "0" * 64,
+            },
+        })
+
+    monkeypatch.setattr(photon_auth.httpx, "post", fake_post)
+    data = photon_auth.register_webhook(
+        "proj", "secret", webhook_url="https://x.example.com/hook",
+    )
+    assert data["signingSecret"] == "0" * 64
+    assert data["webhookUrl"] == "https://x.example.com/hook"
+
+
+def test_persist_webhook_signing_secret_writes_env(
+    tmp_hermes_home: Path,
+) -> None:
+    """The helper hands the secret to save_env_value, never returns it."""
+    summary: list = []
+    response = {
+        "id": "wh-uuid",
+        "webhookUrl": "https://x.example.com/hook",
+        "signingSecret": "ABCDEF1234567890" * 4,
+    }
+    ok = photon_auth.persist_webhook_signing_secret(
+        response, on_summary=summary.append,
+    )
+
+    assert ok is True
+    env_path = tmp_hermes_home / ".env"
+    assert env_path.exists()
+    env_text = env_path.read_text()
+    assert "PHOTON_WEBHOOK_SECRET=ABCDEF1234567890" in env_text
+    # The on_summary callback gets the redacted response + a saved-to path;
+    # none of those strings should leak the raw secret.
+    joined = "\n".join(summary)
+    assert "<redacted>" in joined
+    assert "ABCDEF1234567890" not in joined
+
+
+def test_persist_webhook_signing_secret_no_secret_no_write(
+    tmp_hermes_home: Path,
+) -> None:
+    summary: list = []
+    ok = photon_auth.persist_webhook_signing_secret(
+        {"id": "wh-uuid", "webhookUrl": "https://x"},
+        on_summary=summary.append,
+    )
+    assert ok is False
+    # No env file written; summary callback still received the redacted
+    # response (without a signingSecret key, nothing to redact).
+    assert not (tmp_hermes_home / ".env").exists()
+
+
+def test_credential_summary_returns_only_display_strings(
+    tmp_hermes_home: Path,
+) -> None:
+    """credential_summary must not leak raw token/secret material."""
+    photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
+    photon_auth.store_project_credentials("proj-uuid", "secret-bbbbbbbbbbb")
+    summary = photon_auth.credential_summary()
+    blob = "\n".join(summary.values())
+    assert "token-aaaa" not in blob
+    assert "secret-bbbb" not in blob
+    assert summary["device_token"].startswith("✓")
+    assert summary["project_key"].startswith("✓")
+    assert summary["project_id"] == "proj-uuid"
+
+
+def test_print_credential_summary_emits_only_display_strings(
+    tmp_hermes_home: Path,
+) -> None:
+    """The emit callback must never receive raw credential bytes."""
+    photon_auth.store_photon_token("token-aaaaaaaaaaaaaaaa")
+    photon_auth.store_project_credentials("proj-uuid", "secret-bbbbbbbbbbb")
+    lines: list = []
+    photon_auth.print_credential_summary(lines.append)
+    blob = "\n".join(lines)
+    assert "token-aaaa" not in blob
+    assert "secret-bbbb" not in blob
+    assert "✓ stored" in blob   # device token line
+    assert "proj-uuid" in blob   # project id is intentionally surfaced
+    # Header is always emitted
+    assert any("Photon iMessage status" in line for line in lines)

tests/plugins/platforms/photon/test_inbound.py

@@ -0,0 +1,139 @@
+"""Inbound dispatch + dedup tests for PhotonAdapter.
+
+These tests bypass the aiohttp server — they call ``_dispatch_inbound``
+and ``_is_duplicate`` directly. That keeps them fast and means we can
+exercise the message-shape parsing logic without binding ports.
+"""
+from __future__ import annotations
+
+from typing import List
+
+import pytest
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.base import MessageEvent, MessageType
+from plugins.platforms.photon.adapter import PhotonAdapter
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter:
+    # Avoid touching real auth.json / env.
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
+    cfg = PlatformConfig(enabled=True, token="", extra={})
+    return PhotonAdapter(cfg)
+
+
+@pytest.mark.asyncio
+async def test_dispatch_text_dm(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured: List[MessageEvent] = []
+
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+
+    payload = {
+        "event": "messages",
+        "space": {"id": "any;-;+15551234567", "platform": "iMessage"},
+        "message": {
+            "id": "spc-msg-abc",
+            "platform": "iMessage",
+            "direction": "inbound",
+            "timestamp": "2026-05-14T19:06:32.000Z",
+            "sender": {"id": "+15551234567", "platform": "iMessage"},
+            "space": {"id": "any;-;+15551234567", "platform": "iMessage"},
+            "content": {"type": "text", "text": "hello world"},
+        },
+    }
+    await adapter._dispatch_inbound(payload)
+
+    assert len(captured) == 1
+    event = captured[0]
+    assert event.text == "hello world"
+    assert event.message_type == MessageType.TEXT
+    assert event.message_id == "spc-msg-abc"
+    src = event.source
+    assert src is not None
+    assert src.platform == Platform("photon")
+    assert src.chat_id == "any;-;+15551234567"
+    assert src.chat_type == "dm"
+    assert src.user_id == "+15551234567"
+
+
+@pytest.mark.asyncio
+async def test_dispatch_group_id_detected(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured: List[MessageEvent] = []
+
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+
+    payload = {
+        "event": "messages",
+        "space": {"id": "any;+;group-guid-xyz", "platform": "iMessage"},
+        "message": {
+            "id": "spc-msg-grp",
+            "timestamp": "2026-05-14T19:06:32.000Z",
+            "sender": {"id": "+15551234567"},
+            "space": {"id": "any;+;group-guid-xyz"},
+            "content": {"type": "text", "text": "hi group"},
+        },
+    }
+    await adapter._dispatch_inbound(payload)
+    assert captured[0].source.chat_type == "group"
+
+
+@pytest.mark.asyncio
+async def test_dispatch_attachment_surfaces_marker(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    adapter = _make_adapter(monkeypatch)
+    captured: List[MessageEvent] = []
+
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+
+    payload = {
+        "event": "messages",
+        "message": {
+            "id": "spc-msg-att",
+            "timestamp": "2026-05-14T19:06:32.000Z",
+            "sender": {"id": "+15551234567"},
+            "space": {"id": "any;-;+15551234567"},
+            "content": {
+                "type": "attachment",
+                "name": "IMG_4127.HEIC",
+                "mimeType": "image/heic",
+                "size": 12345,
+            },
+        },
+    }
+    await adapter._dispatch_inbound(payload)
+    assert len(captured) == 1
+    event = captured[0]
+    # Attachment carries metadata marker; mime → MessageType.PHOTO.
+    assert "Photon attachment received" in event.text
+    assert "IMG_4127.HEIC" in event.text
+    assert event.message_type == MessageType.PHOTO
+
+
+def test_is_duplicate_window(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    assert adapter._is_duplicate("id-1") is False
+    assert adapter._is_duplicate("id-1") is True
+    assert adapter._is_duplicate("id-2") is False
+    assert adapter._is_duplicate("id-1") is True  # still dup
+
+
+def test_check_requirements_without_node(monkeypatch: pytest.MonkeyPatch) -> None:
+    # If no node binary on PATH the adapter should refuse to start.
+    from plugins.platforms.photon import adapter as adapter_mod
+
+    monkeypatch.setattr(adapter_mod.shutil, "which", lambda _name: None)
+    assert adapter_mod.check_requirements() is False

tests/plugins/platforms/photon/test_mention_gating.py

@@ -0,0 +1,146 @@
+"""Group-chat mention-gating tests for PhotonAdapter.
+
+Parity with the BlueBubbles iMessage channel: when ``require_mention`` is
+enabled, group messages are dropped unless they hit a wake-word pattern,
+and the leading wake word is stripped from the ones that pass. DMs are
+never gated.
+
+These call ``_dispatch_inbound`` directly (no aiohttp / ports) and assert
+on what reaches ``handle_message``.
+"""
+from __future__ import annotations
+
+from typing import List
+
+import pytest
+
+from gateway.config import PlatformConfig
+from gateway.platforms.base import MessageEvent
+from plugins.platforms.photon.adapter import PhotonAdapter
+
+
+def _make_adapter(monkeypatch: pytest.MonkeyPatch, extra: dict | None = None) -> PhotonAdapter:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
+    monkeypatch.delenv("PHOTON_REQUIRE_MENTION", raising=False)
+    monkeypatch.delenv("PHOTON_MENTION_PATTERNS", raising=False)
+    cfg = PlatformConfig(enabled=True, token="", extra=extra or {})
+    return PhotonAdapter(cfg)
+
+
+def _group_payload(text: str) -> dict:
+    return {
+        "event": "messages",
+        "message": {
+            "id": f"grp-{abs(hash(text))}",
+            "timestamp": "2026-05-14T19:06:32.000Z",
+            "sender": {"id": "+15551234567"},
+            "space": {"id": "any;+;group-guid-xyz"},
+            "content": {"type": "text", "text": text},
+        },
+    }
+
+
+def _dm_payload(text: str) -> dict:
+    return {
+        "event": "messages",
+        "message": {
+            "id": f"dm-{abs(hash(text))}",
+            "timestamp": "2026-05-14T19:06:32.000Z",
+            "sender": {"id": "+15551234567"},
+            "space": {"id": "any;-;+15551234567"},
+            "content": {"type": "text", "text": text},
+        },
+    }
+
+
+def _capture(adapter: PhotonAdapter, monkeypatch: pytest.MonkeyPatch) -> List[MessageEvent]:
+    captured: List[MessageEvent] = []
+
+    async def fake_handle(event: MessageEvent) -> None:
+        captured.append(event)
+
+    monkeypatch.setattr(adapter, "handle_message", fake_handle)
+    return captured
+
+
+def test_require_mention_defaults_off(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)
+    assert adapter.require_mention is False
+    # Defaults compile to the two Hermes wake-word patterns.
+    assert len(adapter._mention_patterns) == 2
+
+
+@pytest.mark.asyncio
+async def test_group_message_dropped_without_mention(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch, extra={"require_mention": True})
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(_group_payload("just chatting, no wake word"))
+    assert captured == []
+
+
+@pytest.mark.asyncio
+async def test_group_message_passes_and_strips_wake_word(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch, extra={"require_mention": True})
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(_group_payload("Hermes what's the weather"))
+    assert len(captured) == 1
+    # Leading wake word stripped before dispatch.
+    assert captured[0].text == "what's the weather"
+
+
+@pytest.mark.asyncio
+async def test_dm_never_gated(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch, extra={"require_mention": True})
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(_dm_payload("no wake word here"))
+    assert len(captured) == 1
+    assert captured[0].text == "no wake word here"
+
+
+@pytest.mark.asyncio
+async def test_require_mention_off_passes_group_messages(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(monkeypatch)  # require_mention defaults off
+    captured = _capture(adapter, monkeypatch)
+
+    await adapter._dispatch_inbound(_group_payload("plain group chatter"))
+    assert len(captured) == 1
+    assert captured[0].text == "plain group chatter"
+
+
+def test_custom_mention_patterns_from_config(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(
+        monkeypatch,
+        extra={"require_mention": True, "mention_patterns": [r"(?<![\w@])@?amos\b[,:\-]?"]},
+    )
+    assert adapter.require_mention is True
+    assert len(adapter._mention_patterns) == 1
+    assert adapter._message_matches_mention_patterns("amos help me") is True
+    assert adapter._message_matches_mention_patterns("hermes help me") is False
+
+
+def test_mention_patterns_env_comma_separated(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id")
+    monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret")
+    monkeypatch.delenv("PHOTON_WEBHOOK_SECRET", raising=False)
+    monkeypatch.setenv("PHOTON_REQUIRE_MENTION", "true")
+    monkeypatch.setenv("PHOTON_MENTION_PATTERNS", r"bot\b, assistant\b")
+    cfg = PlatformConfig(enabled=True, token="", extra={})
+    adapter = PhotonAdapter(cfg)
+    assert adapter.require_mention is True
+    assert len(adapter._mention_patterns) == 2
+    assert adapter._message_matches_mention_patterns("hey bot") is True
+
+
+def test_invalid_pattern_skipped(monkeypatch: pytest.MonkeyPatch) -> None:
+    adapter = _make_adapter(
+        monkeypatch,
+        extra={"require_mention": True, "mention_patterns": ["(unclosed", r"good\b"]},
+    )
+    # Bad regex dropped, good one kept.
+    assert len(adapter._mention_patterns) == 1
+    assert adapter._message_matches_mention_patterns("a good thing") is True

tests/plugins/platforms/photon/test_signature.py

@@ -0,0 +1,95 @@
+"""Signature verification tests for the Photon webhook receiver."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import time
+
+import pytest
+
+from plugins.platforms.photon.adapter import verify_signature
+
+
+def _sign(secret: str, body: bytes, ts: int) -> str:
+    return "v0=" + hmac.new(
+        secret.encode(), f"v0:{ts}:".encode() + body, hashlib.sha256,
+    ).hexdigest()
+
+
+def test_accepts_valid_signature() -> None:
+    secret = "topsecret-32chars-or-whatever"
+    body = b'{"event":"messages"}'
+    ts = int(time.time())
+    sig = _sign(secret, body, ts)
+    assert verify_signature(
+        body=body, timestamp_header=str(ts), signature_header=sig,
+        signing_secret=secret,
+    )
+
+
+def test_rejects_tampered_body() -> None:
+    secret = "s"
+    body = b'{"event":"messages"}'
+    ts = int(time.time())
+    sig = _sign(secret, body, ts)
+    assert not verify_signature(
+        body=body + b" tamper", timestamp_header=str(ts),
+        signature_header=sig, signing_secret=secret,
+    )
+
+
+def test_rejects_wrong_secret() -> None:
+    body = b"x"
+    ts = int(time.time())
+    sig = _sign("right", body, ts)
+    assert not verify_signature(
+        body=body, timestamp_header=str(ts), signature_header=sig,
+        signing_secret="wrong",
+    )
+
+
+def test_rejects_drifted_timestamp() -> None:
+    secret = "s"
+    body = b"x"
+    ts = int(time.time()) - 3600  # 1h old; drift window is 5 min
+    sig = _sign(secret, body, ts)
+    assert not verify_signature(
+        body=body, timestamp_header=str(ts), signature_header=sig,
+        signing_secret=secret,
+    )
+
+
+def test_rejects_missing_v0_prefix() -> None:
+    secret = "s"
+    body = b"x"
+    ts = int(time.time())
+    raw_hex = hmac.new(
+        secret.encode(), f"v0:{ts}:".encode() + body, hashlib.sha256,
+    ).hexdigest()
+    # Strip the "v0=" prefix — verify_signature must reject.
+    assert not verify_signature(
+        body=body, timestamp_header=str(ts), signature_header=raw_hex,
+        signing_secret=secret,
+    )
+
+
+def test_rejects_empty_inputs() -> None:
+    assert not verify_signature(
+        body=b"x", timestamp_header="", signature_header="v0=abc",
+        signing_secret="s",
+    )
+    assert not verify_signature(
+        body=b"x", timestamp_header="123", signature_header="",
+        signing_secret="s",
+    )
+    assert not verify_signature(
+        body=b"x", timestamp_header="123", signature_header="v0=abc",
+        signing_secret="",
+    )
+
+
+def test_rejects_non_integer_timestamp() -> None:
+    assert not verify_signature(
+        body=b"x", timestamp_header="not-an-int",
+        signature_header="v0=abc", signing_secret="s",
+    )

tests/plugins/test_nemo_relay_plugin.py

@@ -2,10 +2,13 @@
 
 from __future__ import annotations
 
+import asyncio
 import builtins
+import gc
 import importlib
 import json
 import sys
+import warnings
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -37,7 +40,7 @@ class _FakeNemoRelay:
             call_end=self._tool_call_end,
             execute=self._tool_execute,
         )
-        self.plugin = SimpleNamespace(initialize=self._plugin_initialize)
+        self.plugin = SimpleNamespace(initialize=self._plugin_initialize, clear=self._plugin_clear)
         self.LLMRequest = _FakeLLMRequest
         self.AtofExporterConfig = _FakeAtofExporterConfig
         self.AtofExporterMode = SimpleNamespace(Append="append", Overwrite="overwrite")
@@ -93,6 +96,9 @@ class _FakeNemoRelay:
         self.events.append(("plugin.initialize", config))
         return {"diagnostics": []}
 
+    async def _plugin_clear(self):
+        self.events.append(("plugin.clear",))
+
 
 class _FakeLLMRequest:
     def __init__(self, headers, content):
@@ -115,6 +121,10 @@ class _FakeAtofExporter:
     def register(self, name):
         self.events.append(("atof.register", name, self.config.output_directory, self.config.filename))
 
+    def deregister(self, name):
+        self.events.append(("atof.deregister", name, self.config.output_directory, self.config.filename))
+        return True
+
 
 class _FakeAtifExporter:
     def __init__(self, events, session_id, agent_name, agent_version, kwargs):
@@ -445,6 +455,252 @@ output_directory = "{atif_dir}"
     assert atif_dir.is_dir()
 
 
+def test_nemo_relay_plugin_clears_plugins_toml_on_final_session_finalize_and_reinitializes(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        """
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+
+    plugin.on_session_start(session_id="s1")
+    plugin.on_session_finalize(session_id="s1", reason="shutdown")
+    plugin.on_session_start(session_id="s2")
+
+    event_names = [event[0] for event in fake.events]
+    assert event_names.count("plugin.initialize") == 2
+    assert event_names.count("plugin.clear") == 1
+
+
+def test_nemo_relay_plugin_keeps_plugins_toml_active_while_other_sessions_remain(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        """
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+
+    plugin.on_session_start(session_id="parent")
+    plugin.on_session_start(session_id="child")
+    plugin.on_session_finalize(session_id="child", reason="shutdown")
+    plugin.on_session_finalize(session_id="parent", reason="shutdown")
+
+    event_names = [event[0] for event in fake.events]
+    assert event_names.count("plugin.initialize") == 1
+    assert event_names.count("plugin.clear") == 1
+
+
+def test_nemo_relay_plugin_reinitializes_plugins_toml_inside_active_event_loop(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        """
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+
+    async def _drive() -> None:
+        plugin.on_session_start(session_id="s1")
+        plugin.on_session_finalize(session_id="s1", reason="shutdown")
+        plugin.on_session_start(session_id="s2")
+        await asyncio.sleep(0)
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        asyncio.run(_drive())
+        gc.collect()
+
+    assert not any("was never awaited" in str(w.message) for w in caught)
+    runtime = plugin._get_runtime()
+    assert runtime is not None
+    assert runtime._plugin_config_initialized is True
+    scope_push_names = [event[1] for event in fake.events if event[0] == "scope.push"]
+    assert "hermes-session-s2" in scope_push_names
+
+
+def test_nemo_relay_plugin_retries_plugins_toml_after_clear_failure(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+    initialize_calls = 0
+
+    async def _counting_initialize(config):
+        nonlocal initialize_calls
+        initialize_calls += 1
+        fake.events.append(("plugin.initialize.attempt", initialize_calls, config))
+        return {"diagnostics": []}
+
+    async def _failing_clear():
+        fake.events.append(("plugin.clear.failed",))
+        raise RuntimeError("boom")
+
+    fake.plugin.initialize = _counting_initialize
+    fake.plugin.clear = _failing_clear
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        """
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+
+    plugin.on_session_start(session_id="s1")
+    plugin.on_session_finalize(session_id="s1", reason="shutdown")
+    plugin.on_session_start(session_id="s2")
+
+    event_names = [event[0] for event in fake.events]
+    assert event_names.count("plugin.initialize.attempt") == 2
+    assert event_names.count("plugin.clear.failed") == 1
+    scope_push_names = [event[1] for event in fake.events if event[0] == "scope.push"]
+    assert "hermes-session-s2" in scope_push_names
+
+
+def test_nemo_relay_plugin_disables_direct_atif_when_plugins_toml_owns_atif(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        f"""
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+
+[components.config.atif]
+enabled = true
+output_directory = "{(tmp_path / "managed-atif").as_posix()}"
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1")
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif"))
+
+    plugin.on_session_start(session_id="s1")
+    plugin.on_session_finalize(session_id="s1", reason="shutdown")
+
+    event_names = [event[0] for event in fake.events]
+    assert "plugin.initialize" in event_names
+    assert "plugin.clear" in event_names
+    assert "atif.register" not in event_names
+    assert not (tmp_path / "direct-atif" / "hermes-atif-s1.json").exists()
+
+
+def test_nemo_relay_plugin_keeps_direct_atif_when_plugins_toml_init_fails(tmp_path, monkeypatch):
+    fake = _FakeNemoRelay()
+
+    async def _failing_initialize(config):
+        fake.events.append(("plugin.initialize.failed", config))
+        raise RuntimeError("boom")
+
+    fake.plugin.initialize = _failing_initialize
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        f"""
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+
+[components.config.atif]
+enabled = true
+output_directory = "{(tmp_path / "managed-atif").as_posix()}"
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_ENABLED", "1")
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATIF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atif"))
+
+    plugin.on_session_start(session_id="s1")
+    plugin.on_session_finalize(session_id="s1", reason="shutdown")
+
+    event_names = [event[0] for event in fake.events]
+    assert "plugin.initialize.failed" in event_names
+    assert "plugin.clear" not in event_names
+    assert "atif.register" in event_names
+    assert (tmp_path / "direct-atif" / "hermes-atif-s1.json").exists()
+
+
+def test_nemo_relay_plugin_retries_plugins_toml_after_fallback_only_session_and_clears_direct_atof(
+    tmp_path,
+    monkeypatch,
+):
+    fake = _FakeNemoRelay()
+    initialize_calls = 0
+
+    async def _flaky_initialize(config):
+        nonlocal initialize_calls
+        initialize_calls += 1
+        fake.events.append(("plugin.initialize.attempt", initialize_calls, config))
+        if initialize_calls == 1:
+            raise RuntimeError("boom")
+        return {"diagnostics": []}
+
+    fake.plugin.initialize = _flaky_initialize
+    plugin = _fresh_plugin(monkeypatch, fake)
+    plugins_toml = tmp_path / "plugins.toml"
+    plugins_toml.write_text(
+        f"""
+version = 1
+
+[[components]]
+kind = "observability"
+enabled = true
+
+[components.config.atof]
+enabled = true
+output_directory = "{(tmp_path / "managed-atof").as_posix()}"
+""",
+        encoding="utf-8",
+    )
+    monkeypatch.setenv("HERMES_NEMO_RELAY_PLUGINS_TOML", str(plugins_toml))
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_ENABLED", "1")
+    monkeypatch.setenv("HERMES_NEMO_RELAY_ATOF_OUTPUT_DIRECTORY", str(tmp_path / "direct-atof"))
+
+    plugin.on_session_start(session_id="s1")
+    plugin.on_session_finalize(session_id="s1", reason="shutdown")
+    plugin.on_session_start(session_id="s2")
+
+    runtime = plugin._get_runtime()
+    assert runtime is not None
+    assert runtime._plugin_config_initialized is True
+    event_names = [event[0] for event in fake.events]
+    assert event_names.count("plugin.initialize.attempt") == 2
+    assert event_names.count("atof.register") == 1
+    assert event_names.count("atof.deregister") == 1
+
+
 def test_nemo_relay_adaptive_llm_execution_middleware_preserves_raw_response(tmp_path, monkeypatch):
     fake = _FakeNemoRelay()
     plugin = _fresh_plugin(monkeypatch, fake)

tests/run_agent/test_partial_stream_finish_reason.py

@@ -136,6 +136,101 @@ class TestPartialStreamStubFinishReason:
         assert "write_file" in content
 
 
+# ── Clean stream-end mid-tool-call (no exception, no finish_reason) ─────────
+
+class TestCleanStreamEndMidToolCall:
+    """The upstream closes the SSE stream cleanly after delivering a tool
+    name + the opening '{' of its arguments — NO exception, NO finish_reason,
+    NO [DONE].  Observed live on NVIDIA Nemotron Ultra via the Nous dedicated
+    endpoint: it stalls/drops during large tool-arg generation.
+
+    The mock-builder must NOT stamp this as finish_reason='length' (which
+    routes it through the max_tokens-boost truncation path and finally
+    reports the misleading 'Response truncated due to output length limit').
+    It must route through the partial-stream-stub path so the loop reports
+    an honest mid-tool-call drop and asks the model to chunk its output.
+    """
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_no_finish_reason_partial_tool_args_routes_to_stub(
+        self, _mock_close, mock_create, monkeypatch,
+    ):
+        def _clean_ending_stream():
+            # Reasoning + tool name + the lone opening brace, then the
+            # generator simply RETURNS (StopIteration) — no raise, no
+            # finish_reason chunk, no [DONE].
+            yield _make_stream_chunk(content="\n")
+            yield _make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, tc_id="call_x", name="execute_code"),
+            ])
+            yield _make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, arguments="{"),
+            ])
+            # falls off the end — clean close, no terminator
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.side_effect = (
+            lambda *a, **kw: _clean_ending_stream()
+        )
+        mock_create.return_value = mock_client
+
+        agent = _make_agent()
+        agent._fire_stream_delta = lambda text: None
+
+        response = agent._interruptible_streaming_api_call({})
+
+        assert response.id == PARTIAL_STREAM_STUB_ID, (
+            "A clean stream-end mid tool-call (no finish_reason) must be "
+            "tagged as a partial-stream stub, not a 'stream-<uuid>' "
+            "truncation — otherwise the loop reports the false 'output "
+            "length limit' error."
+        )
+        assert response.choices[0].finish_reason == FINISH_REASON_LENGTH
+        assert response.choices[0].message.tool_calls is None, (
+            "Incomplete tool args must never auto-execute."
+        )
+        assert getattr(response, "_dropped_tool_names", None) == ["execute_code"]
+
+    @patch("run_agent.AIAgent._create_request_openai_client")
+    @patch("run_agent.AIAgent._close_request_openai_client")
+    def test_real_length_truncation_still_uses_uuid_id(
+        self, _mock_close, mock_create, monkeypatch,
+    ):
+        """Control: when the provider DOES send finish_reason='length' with
+        partial tool args, it is a genuine output cap — keep the existing
+        non-stub behaviour (boost max_tokens and retry)."""
+
+        def _capped_stream():
+            yield _make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, tc_id="call_y", name="execute_code"),
+            ])
+            yield _make_stream_chunk(tool_calls=[
+                _make_tool_call_delta(index=0, arguments="{"),
+            ])
+            # Provider explicitly reports the output cap.
+            yield _make_stream_chunk(finish_reason="length")
+
+        mock_client = MagicMock()
+        mock_client.chat.completions.create.side_effect = (
+            lambda *a, **kw: _capped_stream()
+        )
+        mock_create.return_value = mock_client
+
+        agent = _make_agent()
+        agent._fire_stream_delta = lambda text: None
+
+        response = agent._interruptible_streaming_api_call({})
+
+        assert response.id != PARTIAL_STREAM_STUB_ID, (
+            "A provider-reported finish_reason='length' is a real output cap "
+            "and must keep the existing truncation path, not the stream-drop "
+            "stub path."
+        )
+        assert response.id.startswith("stream-")
+        assert response.choices[0].finish_reason == FINISH_REASON_LENGTH
+
+
 # ── Length-continuation prompt branching ──────────────────────────────────
 
 class TestLengthContinuationPromptBranching:

tests/run_agent/test_run_agent.py

@@ -2402,15 +2402,20 @@ class TestConcurrentToolExecution:
 
     def test_concurrent_handles_tool_error(self, agent):
         """If one tool raises, others should still complete."""
-        tc1 = _mock_tool_call(name="web_search", arguments='{}', call_id="c1")
-        tc2 = _mock_tool_call(name="web_search", arguments='{}', call_id="c2")
+        # Distinguish the two calls by their arguments so the error is tied to
+        # a SPECIFIC tool call rather than invocation order. Concurrent
+        # execution gives no guarantee that c1's handler runs before c2's, so
+        # keying the raise on a call-order counter is racy: under thread-pool
+        # scheduling c2 could be invoked first, take the "first call raises"
+        # branch, and the error would land in messages[1] instead of
+        # messages[0]. Keying on args makes the assertion deterministic.
+        tc1 = _mock_tool_call(name="web_search", arguments='{"q": "boom"}', call_id="c1")
+        tc2 = _mock_tool_call(name="web_search", arguments='{"q": "ok"}', call_id="c2")
         mock_msg = _mock_assistant_msg(content="", tool_calls=[tc1, tc2])
         messages = []
 
-        call_count = [0]
         def fake_handle(name, args, task_id, **kwargs):
-            call_count[0] += 1
-            if call_count[0] == 1:
+            if args.get("q") == "boom":
                 raise RuntimeError("boom")
             return "success"
 
@@ -2418,9 +2423,11 @@ class TestConcurrentToolExecution:
             agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1")
 
         assert len(messages) == 2
-        # First tool should have error
+        # Results are ordered by tool_call_id; c1 raised, c2 succeeded.
+        assert messages[0]["tool_call_id"] == "c1"
         assert "Error" in messages[0]["content"] or "boom" in messages[0]["content"]
         # Second tool should succeed
+        assert messages[1]["tool_call_id"] == "c2"
         assert "success" in messages[1]["content"]
 
     def test_concurrent_interrupt_before_start(self, agent):
@@ -5788,7 +5795,15 @@ class TestStreamingApiCall:
         assert tc[0].function.name == "search"
         assert tc[1].function.name == "read"
 
-    def test_truncated_tool_call_args_upgrade_finish_reason_to_length(self, agent):
+    def test_truncated_tool_call_args_no_finish_reason_routes_to_stub(self, agent):
+        # Stream delivers a tool call with incomplete JSON args and then ENDS
+        # with no finish_reason (the SSE just stops — no terminator, no
+        # [DONE]).  This is an upstream mid-tool-call drop, NOT an output cap.
+        # The builder must route it through the partial-stream-stub path
+        # (id=PARTIAL_STREAM_STUB_ID, tool_calls=None so it can't execute,
+        # finish_reason=length so the loop's continuation machinery fires with
+        # chunking guidance) rather than stamping a normal 'length' truncation.
+        from hermes_constants import PARTIAL_STREAM_STUB_ID
         chunks = [
             _make_chunk(tool_calls=[_make_tc_delta(0, "call_1", "write_file", '{"path":"x.txt","content":"hel')]),
         ]
@@ -5796,6 +5811,24 @@ class TestStreamingApiCall:
 
         resp = agent._interruptible_streaming_api_call({"messages": []})
 
+        assert resp.id == PARTIAL_STREAM_STUB_ID
+        assert resp.choices[0].finish_reason == "length"
+        assert resp.choices[0].message.tool_calls is None
+        assert getattr(resp, "_dropped_tool_names", None) == ["write_file"]
+
+    def test_truncated_tool_call_args_with_length_finish_reason_upgrades(self, agent):
+        # Control: when the provider explicitly reports finish_reason='length'
+        # alongside incomplete tool args, it IS a genuine output cap.  Keep the
+        # existing behaviour — tool_calls preserved, finish_reason 'length' —
+        # so the max_tokens-boost truncation retry path still applies.
+        chunks = [
+            _make_chunk(tool_calls=[_make_tc_delta(0, "call_1", "write_file", '{"path":"x.txt","content":"hel')]),
+            _make_chunk(finish_reason="length"),
+        ]
+        agent.client.chat.completions.create.return_value = iter(chunks)
+
+        resp = agent._interruptible_streaming_api_call({"messages": []})
+
         tc = resp.choices[0].message.tool_calls
         assert len(tc) == 1
         assert tc[0].function.name == "write_file"

tests/test_project_metadata.py

@@ -88,6 +88,103 @@ def test_lazy_installable_extras_excluded_from_all():
         )
 
 
+def _exact_pins(specs):
+    pins = {}
+    for spec in specs:
+        requirement = spec.split(";", 1)[0].strip()
+        if "==" not in requirement:
+            continue
+        package, version = requirement.split("==", 1)
+        package = package.split("[", 1)[0].lower().replace("_", "-")
+        pins[package] = version
+    return pins
+
+
+def test_pyproject_aiohttp_pins_match_lazy_slack_pin():
+    """Avoid update/lazy-install churn from conflicting aiohttp pins.
+
+    pyproject extras (messaging/slack/homeassistant/sms) exact-pin aiohttp.
+    The Slack lazy-install deps (LAZY_DEPS['platform.slack']) also pin it.
+    If the two drift, `hermes update` resolves the pyproject pin and
+    downgrades aiohttp, reopening the CVEs the lazy pin fixed (#31817) —
+    only for Slack's lazy refresh to upgrade it again on next use.
+    """
+    from tools.lazy_deps import LAZY_DEPS
+
+    optional_dependencies = _load_optional_dependencies()
+    lazy_aiohttp = _exact_pins(LAZY_DEPS["platform.slack"])["aiohttp"]
+
+    pyproject_aiohttp_pins = {
+        extra: pins["aiohttp"]
+        for extra, specs in optional_dependencies.items()
+        if "aiohttp" in (pins := _exact_pins(specs))
+    }
+
+    assert pyproject_aiohttp_pins, "expected at least one pyproject extra to pin aiohttp"
+    mismatches = {
+        extra: pin
+        for extra, pin in pyproject_aiohttp_pins.items()
+        if pin != lazy_aiohttp
+    }
+    assert not mismatches, (
+        "pyproject.toml aiohttp pins must match "
+        "LAZY_DEPS['platform.slack'] to avoid hermes update downgrading "
+        "aiohttp before Slack's lazy refresh upgrades it again. "
+        f"lazy aiohttp=={lazy_aiohttp}; mismatched extras: {mismatches}"
+    )
+
+
+def test_pyproject_pins_match_lazy_deps_pins():
+    """Generalize #31817 to the whole pin surface, not just aiohttp.
+
+    Any package that is exact-pinned in BOTH a pyproject extra and a
+    `tools/lazy_deps.py` LAZY_DEPS entry must use the SAME version in both
+    places. When they drift, `hermes update` resolves the pyproject extra
+    pin and downgrades the package to the older version, reopening whatever
+    the lazy pin fixed (the aiohttp #31817 case, and the anthropic
+    CVE-2026-34450/34452 case found alongside it) — only for the lazy
+    refresh to re-upgrade it on next feature use. The lazy pin is the
+    security-current source of truth; extras must track it.
+    """
+    from tools.lazy_deps import LAZY_DEPS
+
+    optional_dependencies = _load_optional_dependencies()
+
+    # package -> version, as pinned across all pyproject extras. If an
+    # extra pins a package at a different version than another extra, that
+    # is itself a bug (caught below); here we just collect the set.
+    pyproject_pins: dict[str, set[str]] = {}
+    for specs in optional_dependencies.values():
+        for package, version in _exact_pins(specs).items():
+            pyproject_pins.setdefault(package, set()).add(version)
+
+    # package -> version, as pinned across all LAZY_DEPS entries.
+    lazy_pins: dict[str, set[str]] = {}
+    for specs in LAZY_DEPS.values():
+        if isinstance(specs, str):
+            specs = (specs,)
+        for package, version in _exact_pins(specs).items():
+            lazy_pins.setdefault(package, set()).add(version)
+
+    shared = sorted(set(pyproject_pins) & set(lazy_pins))
+    assert shared, "expected at least one package pinned in both pyproject and LAZY_DEPS"
+
+    drift = {
+        package: {
+            "pyproject": sorted(pyproject_pins[package]),
+            "lazy_deps": sorted(lazy_pins[package]),
+        }
+        for package in shared
+        if pyproject_pins[package] != lazy_pins[package]
+    }
+    assert not drift, (
+        "pyproject extras pins must match tools/lazy_deps.py LAZY_DEPS pins "
+        "for every shared package — otherwise `hermes update` downgrades the "
+        "package below the security-current lazy pin (see #31817). Drift: "
+        f"{drift}"
+    )
+
+
 def test_dev_extra_excluded_from_all():
     """End-user installs should not pull test/lint/debug tooling."""
     optional_dependencies = _load_optional_dependencies()

tests/tools/test_approval.py

@@ -9,6 +9,7 @@ from types import SimpleNamespace
 from unittest.mock import patch as mock_patch
 
 import tools.approval as approval_module
+from hermes_constants import get_hermes_home
 from tools.approval import (
     _get_approval_mode,
     _smart_approve,
@@ -424,6 +425,22 @@ class TestHermesConfigWriteProtection:
         dangerous, key, desc = detect_dangerous_command("sed --in-place 's/manual/off/' ~/.hermes/config.yaml")
         assert dangerous is True
 
+    def test_sed_in_place_absolute_hermes_home_config(self):
+        config_path = get_hermes_home() / "config.yaml"
+        dangerous, key, desc = detect_dangerous_command(
+            f"sed -i 's/manual/off/' {config_path}"
+        )
+        assert dangerous is True
+        assert "hermes config" in desc.lower() or "in-place" in desc.lower()
+
+    def test_sed_in_place_absolute_hermes_home_env(self):
+        env_path = get_hermes_home() / ".env"
+        dangerous, key, desc = detect_dangerous_command(
+            f"sed -i 's/API_KEY=.*/API_KEY=x/' {env_path}"
+        )
+        assert dangerous is True
+        assert "hermes config" in desc.lower() or "in-place" in desc.lower()
+
     def test_custom_hermes_home(self):
         dangerous, key, desc = detect_dangerous_command("echo x | tee $HERMES_HOME/config.yaml")
         assert dangerous is True
@@ -437,12 +454,33 @@ class TestHermesConfigWriteProtection:
         assert dangerous is True
         assert "in-place" in desc.lower() or "perl" in desc.lower()
 
+    def test_perl_in_place_absolute_hermes_home_config(self):
+        config_path = get_hermes_home() / "config.yaml"
+        dangerous, key, desc = detect_dangerous_command(
+            f"perl -i -pe 's/approvals.mode: on/approvals.mode: off/' {config_path}"
+        )
+        assert dangerous is True
+        assert "in-place" in desc.lower() or "perl" in desc.lower()
+
     def test_ruby_in_place_config(self):
         dangerous, key, desc = detect_dangerous_command(
             "ruby -i -pe 'gsub(/manual/, \"off\")' ~/.hermes/config.yaml"
         )
         assert dangerous is True
 
+    def test_ruby_in_place_absolute_hermes_home_env(self):
+        env_path = get_hermes_home() / ".env"
+        dangerous, key, desc = detect_dangerous_command(
+            f"ruby -i -pe 'gsub(/API_KEY=.*/, \"API_KEY=x\")' {env_path}"
+        )
+        assert dangerous is True
+
+    def test_regular_absolute_config_path_still_uses_project_rule(self):
+        dangerous, key, desc = detect_dangerous_command(
+            "sed -i 's/a/b/' /srv/app/config.yaml"
+        )
+        assert dangerous is False
+
     def test_perl_in_place_env(self):
         dangerous, key, desc = detect_dangerous_command(
             "perl -i -pe 's/SECRET=old/SECRET=new/' ~/.hermes/.env"