Merge branch 'main' of github.com:NousResearch/hermes-agent into bb/gui

tests/cli/test_cli_goal_interrupt.py

@@ -0,0 +1,221 @@
+"""Tests for CLI goal-continuation interrupt handling.
+
+Covers:
+- Ctrl+C during a /goal turn auto-pauses the goal (no more continuations).
+- Empty/whitespace-only responses skip the judge (no phantom continuations).
+- Clean response without interrupt still drives the judge + enqueues.
+
+These tests exercise ``_maybe_continue_goal_after_turn`` directly on a
+minimal ``HermesCLI`` stub (pattern used elsewhere in tests/cli).
+"""
+
+from __future__ import annotations
+
+import queue
+import sys
+import uuid
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Fixtures
+# ──────────────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def hermes_home(tmp_path, monkeypatch):
+    """Isolated HERMES_HOME so SessionDB.state_meta writes stay hermetic."""
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    monkeypatch.setenv("HERMES_HOME", str(home))
+
+    # Bust the goal module's DB cache so it re-resolves HERMES_HOME each test.
+    from hermes_cli import goals
+    goals._DB_CACHE.clear()
+    yield home
+    goals._DB_CACHE.clear()
+
+
+def _make_cli_with_goal(session_id: str, goal_text: str = "build a thing"):
+    """Build a minimal HermesCLI stub with an active goal wired in."""
+    from cli import HermesCLI
+    from hermes_cli.goals import GoalManager
+
+    cli = HermesCLI.__new__(HermesCLI)
+    # State the hook + helpers touch directly.
+    cli._pending_input = queue.Queue()
+    cli._last_turn_interrupted = False
+    cli.conversation_history = []
+    # `_get_goal_manager()` reads `self.session_id` directly, not
+    # `self.agent.session_id`. Match the production lookup.
+    cli.session_id = session_id
+    cli.agent = MagicMock()
+    cli.agent.session_id = session_id
+
+    mgr = GoalManager(session_id=session_id, default_max_turns=5)
+    mgr.set(goal_text)
+    cli._goal_manager = mgr
+    return cli, mgr
+
+
+# ──────────────────────────────────────────────────────────────────────
+# Tests
+# ──────────────────────────────────────────────────────────────────────
+
+
+class TestInterruptAutoPause:
+    def test_interrupted_turn_pauses_goal_and_skips_continuation(self, hermes_home):
+        """Ctrl+C mid-turn must auto-pause the goal, not queue another round."""
+        sid = f"sid-interrupt-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        # Simulate an interrupted turn with a partial assistant reply.
+        cli._last_turn_interrupted = True
+        cli.conversation_history = [
+            {"role": "user", "content": "kickoff"},
+            {"role": "assistant", "content": "starting work..."},
+        ]
+
+        # Judge MUST NOT run on an interrupted turn. If it does, we've
+        # regressed — fail loudly instead of silently querying a mock.
+        with patch("hermes_cli.goals.judge_goal") as judge_mock:
+            judge_mock.side_effect = AssertionError(
+                "judge_goal called on an interrupted turn"
+            )
+            cli._maybe_continue_goal_after_turn()
+
+        # Pending input must NOT contain a continuation prompt.
+        assert cli._pending_input.empty(), (
+            "Interrupted turn should not enqueue a continuation prompt"
+        )
+
+        # Goal should be paused, not active.
+        state = mgr.state
+        assert state is not None
+        assert state.status == "paused"
+        assert "interrupt" in (state.paused_reason or "").lower()
+
+    def test_interrupted_turn_is_resumable(self, hermes_home):
+        """After auto-pause from Ctrl+C, /goal resume puts it back to active."""
+        sid = f"sid-resume-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        cli._last_turn_interrupted = True
+        cli.conversation_history = [
+            {"role": "assistant", "content": "partial"},
+        ]
+        with patch("hermes_cli.goals.judge_goal"):
+            cli._maybe_continue_goal_after_turn()
+        assert mgr.state.status == "paused"
+
+        mgr.resume()
+        assert mgr.state.status == "active"
+
+
+class TestEmptyResponseSkip:
+    def test_empty_response_does_not_invoke_judge(self, hermes_home):
+        """Whitespace-only replies skip judging (transient failure guard)."""
+        sid = f"sid-empty-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        cli._last_turn_interrupted = False
+        cli.conversation_history = [
+            {"role": "user", "content": "go"},
+            {"role": "assistant", "content": "   \n\n   "},
+        ]
+
+        with patch("hermes_cli.goals.judge_goal") as judge_mock:
+            judge_mock.side_effect = AssertionError(
+                "judge_goal called on an empty response"
+            )
+            cli._maybe_continue_goal_after_turn()
+
+        # No continuation queued; goal still active (neither paused nor done).
+        assert cli._pending_input.empty()
+        assert mgr.state.status == "active"
+
+    def test_no_assistant_message_skipped(self, hermes_home):
+        """Conversation with zero assistant replies must not trip the judge."""
+        sid = f"sid-noassistant-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        cli._last_turn_interrupted = False
+        cli.conversation_history = [
+            {"role": "user", "content": "go"},
+        ]
+
+        with patch("hermes_cli.goals.judge_goal") as judge_mock:
+            judge_mock.side_effect = AssertionError(
+                "judge_goal called without an assistant response"
+            )
+            cli._maybe_continue_goal_after_turn()
+
+        assert cli._pending_input.empty()
+        assert mgr.state.status == "active"
+
+
+class TestHealthyTurnStillRuns:
+    def test_clean_response_enqueues_continuation_when_judge_says_continue(
+        self, hermes_home,
+    ):
+        """Sanity check: the hook still works in the happy path."""
+        sid = f"sid-healthy-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        cli._last_turn_interrupted = False
+        cli.conversation_history = [
+            {"role": "user", "content": "go"},
+            {"role": "assistant", "content": "did some work, more to do"},
+        ]
+
+        # Force the judge to say "continue" without touching the network.
+        with patch(
+            "hermes_cli.goals.judge_goal",
+            return_value=("continue", "needs more steps", False),
+        ):
+            cli._maybe_continue_goal_after_turn()
+
+        # Continuation prompt must be queued.
+        assert not cli._pending_input.empty()
+        queued = cli._pending_input.get_nowait()
+        assert "Continuing toward your standing goal" in queued
+        assert mgr.state.status == "active"
+
+    def test_clean_response_marks_done_when_judge_says_done(self, hermes_home):
+        sid = f"sid-done-{uuid.uuid4().hex}"
+        cli, mgr = _make_cli_with_goal(sid)
+        cli._last_turn_interrupted = False
+        cli.conversation_history = [
+            {"role": "assistant", "content": "all finished, here's the result"},
+        ]
+
+        with patch(
+            "hermes_cli.goals.judge_goal",
+            return_value=("done", "goal satisfied", False),
+        ):
+            cli._maybe_continue_goal_after_turn()
+
+        assert cli._pending_input.empty()
+        assert mgr.state.status == "done"
+
+
+class TestInterruptFlagLifecycle:
+    def test_chat_resets_flag_at_entry(self, hermes_home):
+        """chat() must reset _last_turn_interrupted at the top of each turn.
+
+        This guards against stale flag state: if turn N was interrupted and
+        turn N+1 runs clean, the hook must not see True from N.
+        """
+        # We can't run chat() end-to-end here, but we can assert the reset
+        # is the first thing after the secret-capture registration by
+        # inspecting the source shape.
+        from cli import HermesCLI
+        import inspect
+
+        src = inspect.getsource(HermesCLI.chat)
+        # Look for an explicit reset near the top of chat().
+        head = src.split("if not self._ensure_runtime_credentials", 1)[0]
+        assert "self._last_turn_interrupted = False" in head, (
+            "chat() must reset _last_turn_interrupted before run_conversation "
+            "runs — otherwise a prior turn's interrupt state leaks into the "
+            "next turn's goal hook decision."
+        )

tests/cron/test_scheduler.py

@@ -351,6 +351,95 @@ class TestResolveDeliveryTarget:
         assert _resolve_delivery_targets({"deliver": []}) == []
 
 
+class TestRoutingIntents:
+    """``all`` routing intent expands at fire time."""
+
+    def test_all_expands_to_every_connected_home_channel(self, monkeypatch):
+        """deliver='all' fans out to every platform with a configured home channel."""
+        from cron.scheduler import _resolve_delivery_targets
+
+        monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-111")
+        monkeypatch.setenv("DISCORD_HOME_CHANNEL", "-222")
+        monkeypatch.setenv("SLACK_HOME_CHANNEL", "C333")
+        # Sanity: platforms without the env var must NOT appear in the expansion.
+        monkeypatch.delenv("SIGNAL_HOME_CHANNEL", raising=False)
+        monkeypatch.delenv("MATRIX_HOME_ROOM", raising=False)
+
+        targets = _resolve_delivery_targets({"deliver": "all", "origin": None})
+        platforms = sorted(t["platform"] for t in targets)
+
+        assert "telegram" in platforms
+        assert "discord" in platforms
+        assert "slack" in platforms
+        assert "signal" not in platforms
+        assert "matrix" not in platforms
+
+    def test_all_combines_with_explicit_target_and_dedups(self, monkeypatch):
+        """'telegram:-999,all' yields every home channel + the explicit target without dupes."""
+        from cron.scheduler import _resolve_delivery_targets
+
+        monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-111")
+        monkeypatch.setenv("DISCORD_HOME_CHANNEL", "-222")
+
+        # Explicit telegram target precedes 'all'. Expansion adds discord;
+        # the dedup pass collapses any (platform, chat_id, thread_id) repeats.
+        job = {"deliver": "telegram:-999,all", "origin": None}
+        targets = _resolve_delivery_targets(job)
+
+        platforms = sorted(t["platform"].lower() for t in targets)
+        assert "telegram" in platforms
+        assert "discord" in platforms
+        # Every target is unique on (platform, chat_id, thread_id).
+        keys = [(t["platform"].lower(), str(t["chat_id"]), t.get("thread_id")) for t in targets]
+        assert len(keys) == len(set(keys))
+
+    def test_all_with_no_connected_channels_returns_empty(self, monkeypatch):
+        """deliver='all' with nothing connected returns [] — delivery is recorded as failed upstream."""
+        from cron.scheduler import _resolve_delivery_targets
+
+        for var in ("TELEGRAM_HOME_CHANNEL", "DISCORD_HOME_CHANNEL", "SLACK_HOME_CHANNEL",
+                    "SIGNAL_HOME_CHANNEL", "MATRIX_HOME_ROOM", "MATTERMOST_HOME_CHANNEL",
+                    "SMS_HOME_CHANNEL", "EMAIL_HOME_ADDRESS", "DINGTALK_HOME_CHANNEL",
+                    "FEISHU_HOME_CHANNEL", "WECOM_HOME_CHANNEL", "WEIXIN_HOME_CHANNEL",
+                    "BLUEBUBBLES_HOME_CHANNEL", "QQBOT_HOME_CHANNEL", "QQ_HOME_CHANNEL"):
+            monkeypatch.delenv(var, raising=False)
+
+        assert _resolve_delivery_targets({"deliver": "all", "origin": None}) == []
+
+    def test_origin_comma_all_preserves_origin_first(self, monkeypatch):
+        """'origin,all' delivers to the origin platform plus every other home channel."""
+        from cron.scheduler import _resolve_delivery_targets
+
+        monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-111")
+        monkeypatch.setenv("DISCORD_HOME_CHANNEL", "-222")
+
+        job = {
+            "deliver": "origin,all",
+            "origin": {"platform": "discord", "chat_id": "888"},
+        }
+        targets = _resolve_delivery_targets(job)
+        platforms = sorted(t["platform"].lower() for t in targets)
+        assert "telegram" in platforms
+        assert "discord" in platforms
+
+        # The origin's explicit chat_id (888) wins the dedup race over the
+        # discord home channel (-222) because origin is resolved first.
+        discord = next(t for t in targets if t["platform"].lower() == "discord")
+        assert discord["chat_id"] == "888"
+
+    def test_all_token_case_insensitive(self, monkeypatch):
+        """'ALL' / 'All' / 'all' are all recognized."""
+        from cron.scheduler import _resolve_delivery_targets
+
+        monkeypatch.setenv("TELEGRAM_HOME_CHANNEL", "-111")
+        monkeypatch.setenv("DISCORD_HOME_CHANNEL", "-222")
+
+        for token in ("ALL", "All", "all"):
+            targets = _resolve_delivery_targets({"deliver": token, "origin": None})
+            platforms = sorted(t["platform"].lower() for t in targets)
+            assert platforms == ["discord", "telegram"], f"token={token!r} -> {platforms}"
+
+
 class TestDeliverResultWrapping:
     """Verify that cron deliveries are wrapped with header/footer and no longer mirrored."""
 

tests/gateway/test_api_server_runs.py

@@ -49,6 +49,7 @@ def _create_runs_app(adapter: APIServerAdapter) -> web.Application:
     app.router.add_post("/v1/runs", adapter._handle_runs)
     app.router.add_get("/v1/runs/{run_id}", adapter._handle_get_run)
     app.router.add_get("/v1/runs/{run_id}/events", adapter._handle_run_events)
+    app.router.add_post("/v1/runs/{run_id}/approval", adapter._handle_run_approval)
     app.router.add_post("/v1/runs/{run_id}/stop", adapter._handle_stop_run)
     return app
 
@@ -305,6 +306,98 @@ class TestRunEvents:
                 assert "run.completed" in body
                 assert "Hello!" in body
 
+
+    @pytest.mark.asyncio
+    async def test_approval_request_event_and_response_unblock_run(self, adapter):
+        """Dangerous-command approvals should surface on the run SSE stream."""
+        app = _create_runs_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            with patch.object(adapter, "_create_agent") as mock_create:
+                guard_result = {}
+
+                mock_agent = MagicMock()
+
+                def _run_with_approval(user_message=None, conversation_history=None, task_id=None):
+                    from tools.approval import check_all_command_guards
+
+                    result = check_all_command_guards("git reset --hard HEAD", "local")
+                    guard_result.update(result)
+                    return {"final_response": "approved" if result.get("approved") else "blocked"}
+
+                mock_agent.run_conversation.side_effect = _run_with_approval
+                mock_agent.session_prompt_tokens = 0
+                mock_agent.session_completion_tokens = 0
+                mock_agent.session_total_tokens = 0
+                mock_create.return_value = mock_agent
+
+                resp = await cli.post("/v1/runs", json={"input": "needs approval"})
+                assert resp.status == 202
+                data = await resp.json()
+                run_id = data["run_id"]
+
+                events_resp = await cli.get(f"/v1/runs/{run_id}/events")
+                assert events_resp.status == 200
+
+                approval_event = None
+                for _ in range(20):
+                    line = await asyncio.wait_for(events_resp.content.readline(), timeout=3.0)
+                    text = line.decode()
+                    if not text.startswith("data: "):
+                        continue
+                    event = json.loads(text[len("data: "):])
+                    if event.get("event") == "approval.request":
+                        approval_event = event
+                        break
+
+                assert approval_event is not None
+                assert approval_event["run_id"] == run_id
+                assert approval_event["command"] == "git reset --hard HEAD"
+                assert approval_event["pattern_key"]
+                assert "pattern_keys" in approval_event
+                assert approval_event["choices"] == ["once", "session", "always", "deny"]
+
+                approval_resp = await cli.post(
+                    f"/v1/runs/{run_id}/approval",
+                    json={"choice": "once"},
+                )
+                assert approval_resp.status == 200
+                approval_data = await approval_resp.json()
+                assert approval_data["resolved"] == 1
+                assert approval_data["choice"] == "once"
+
+                body = await events_resp.text()
+                assert "approval.responded" in body
+                assert "run.completed" in body
+
+                assert guard_result.get("approved") is True
+
+    @pytest.mark.asyncio
+    async def test_approval_response_without_pending_returns_409(self, adapter):
+        app = _create_runs_app(adapter)
+        async with TestClient(TestServer(app)) as cli:
+            with patch.object(adapter, "_create_agent") as mock_create:
+                mock_agent = MagicMock()
+                mock_agent.run_conversation.return_value = {"final_response": "done"}
+                mock_agent.session_prompt_tokens = 0
+                mock_agent.session_completion_tokens = 0
+                mock_agent.session_total_tokens = 0
+                mock_create.return_value = mock_agent
+
+                resp = await cli.post("/v1/runs", json={"input": "hello"})
+                data = await resp.json()
+                run_id = data["run_id"]
+
+                approval_resp = await cli.post(
+                    f"/v1/runs/{run_id}/approval",
+                    json={"choice": "once"},
+                )
+                assert approval_resp.status == 409
+                approval_data = await approval_resp.json()
+                assert approval_data["error"]["code"] in {
+                    "approval_not_active",
+                    "approval_not_pending",
+                }
+
     @pytest.mark.asyncio
     async def test_events_not_found_returns_404(self, adapter):
         app = _create_runs_app(adapter)

tests/hermes_cli/test_auth_nous_provider.py

@@ -1,7 +1,6 @@
 """Regression tests for Nous OAuth refresh + agent-key mint interactions."""
 
 import json
-import os
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -862,6 +861,46 @@ def test_refresh_token_reuse_detection_surfaces_actionable_message():
     assert exc_info.value.relogin_required is True
 
 
+def test_refresh_token_exchange_sends_refresh_token_header():
+    """Nous refresh tokens must be sent in a header so sandbox proxies can
+    substitute placeholder credentials without parsing form bodies.
+    """
+    from hermes_cli.auth import _refresh_access_token
+
+    class _FakeResponse:
+        status_code = 200
+
+        def json(self):
+            return {"access_token": "access-2", "refresh_token": "refresh-2"}
+
+    class _FakeClient:
+        def __init__(self):
+            self.kwargs = None
+
+        def post(self, *args, **kwargs):
+            del args
+            self.kwargs = kwargs
+            return _FakeResponse()
+
+    client = _FakeClient()
+
+    payload = _refresh_access_token(
+        client=client,
+        portal_base_url="https://portal.nousresearch.com",
+        client_id="hermes-cli",
+        refresh_token="refresh-1",
+    )
+
+    assert payload["access_token"] == "access-2"
+    assert payload["refresh_token"] == "refresh-2"
+    assert client.kwargs is not None
+    assert client.kwargs["headers"]["x-nous-refresh-token"] == "refresh-1"
+    assert client.kwargs["data"] == {
+        "grant_type": "refresh_token",
+        "client_id": "hermes-cli",
+    }
+
+
 def test_refresh_non_reuse_error_keeps_original_description():
     """Non-reuse invalid_grant errors must keep their original description untouched.
 

tests/hermes_cli/test_gmi_provider.py

@@ -284,6 +284,22 @@ class TestGmiAuxiliary:
         assert model == "google/gemini-3.1-flash-lite-preview"
         assert mock_openai.call_args.kwargs["api_key"] == "gmi-test-key"
         assert mock_openai.call_args.kwargs["base_url"] == "https://api.gmi-serving.com/v1"
+        # GMI profile declares default_headers with a HermesAgent User-Agent
+        # for traffic attribution. The generic profile-fallback branch in
+        # resolve_provider_client should carry it through to the OpenAI client.
+        headers = mock_openai.call_args.kwargs.get("default_headers", {})
+        assert headers.get("User-Agent", "").startswith("HermesAgent/")
+
+    def test_gmi_profile_declares_hermes_user_agent(self):
+        """The GMI plugin sets a HermesAgent/<ver> User-Agent on its profile."""
+        from providers import get_provider_profile
+
+        profile = get_provider_profile("gmi")
+        assert profile is not None
+        ua = profile.default_headers.get("User-Agent", "")
+        assert ua.startswith("HermesAgent/"), (
+            f"expected GMI profile User-Agent to start with 'HermesAgent/', got {ua!r}"
+        )
 
     def test_resolve_provider_client_accepts_gmi_alias(self, monkeypatch):
         monkeypatch.setenv("GMI_API_KEY", "gmi-test-key")

tests/hermes_cli/test_profile_distribution.py

@@ -0,0 +1,584 @@
+"""Tests for hermes_cli.profile_distribution — git-based profile installs.
+
+Covers manifest parsing, version requirement checks, install / update / describe
+on local-directory sources, and guards on what can and can't be installed.
+
+Transport-layer tests (git clone, URL handling) are exercised through live
+E2E runs, not unit tests — git itself is tested upstream, and subprocess-
+mocking git would just test the mock.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+from hermes_cli.profile_distribution import (
+    DEFAULT_DIST_OWNED,
+    DistributionError,
+    DistributionManifest,
+    EnvRequirement,
+    MANIFEST_FILENAME,
+    USER_OWNED_EXCLUDE,
+    _env_template_from_manifest,
+    _looks_like_git_url,
+    _parse_semver,
+    check_hermes_requires,
+    describe_distribution,
+    install_distribution,
+    plan_install,
+    read_manifest,
+    update_distribution,
+    write_manifest,
+)
+
+
+# ---------------------------------------------------------------------------
+# Isolated profile env (matches tests/hermes_cli/test_profiles.py)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def profile_env(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    default_home = tmp_path / ".hermes"
+    default_home.mkdir(exist_ok=True)
+    monkeypatch.setenv("HERMES_HOME", str(default_home))
+    return tmp_path
+
+
+def _make_staging_dir(root: Path, name: str = "src", *, manifest: DistributionManifest = None) -> Path:
+    """Build a local distribution staging directory (what a git clone would
+    contain after .git is removed).
+
+    Lays down a minimal but representative tree: SOUL.md, config.yaml,
+    mcp.json, one skill, one cron file, plus the distribution.yaml manifest.
+    """
+    staged = root / f"staging_{name}"
+    staged.mkdir(parents=True, exist_ok=True)
+    (staged / "SOUL.md").write_text("I am Source.\n")
+    (staged / "config.yaml").write_text("model:\n  model: gpt-4\n")
+    (staged / "mcp.json").write_text('{"servers": {}}\n')
+    (staged / "skills").mkdir(exist_ok=True)
+    (staged / "skills" / "demo").mkdir(exist_ok=True)
+    (staged / "skills" / "demo" / "SKILL.md").write_text(
+        "---\nname: demo\ndescription: test\n---\n# Demo skill\n"
+    )
+    (staged / "cron").mkdir(exist_ok=True)
+    (staged / "cron" / "daily.json").write_text('{"schedule": "0 9 * * *"}')
+
+    mf = manifest or DistributionManifest(name=name, version="0.1.0")
+    write_manifest(staged, mf)
+    return staged
+
+
+# ===========================================================================
+# Manifest parsing
+# ===========================================================================
+
+
+class TestManifestParsing:
+
+    def test_minimal_manifest(self, tmp_path):
+        (tmp_path / MANIFEST_FILENAME).write_text("name: minimal\n")
+        m = read_manifest(tmp_path)
+        assert m.name == "minimal"
+        assert m.version == "0.1.0"
+        assert m.env_requires == []
+        assert m.distribution_owned == []
+
+    def test_full_manifest(self, tmp_path):
+        (tmp_path / MANIFEST_FILENAME).write_text(
+            "name: telem\n"
+            "version: 1.2.3\n"
+            "description: Telem monitor\n"
+            "hermes_requires: '>=0.12.0'\n"
+            "author: Kyle\n"
+            "license: MIT\n"
+            "env_requires:\n"
+            "  - name: OPENAI_API_KEY\n"
+            "    description: OpenAI key\n"
+            "  - name: GRAPH_URL\n"
+            "    required: false\n"
+            "    default: http://127.0.0.1:8000\n"
+            "distribution_owned:\n"
+            "  - SOUL.md\n"
+            "  - skills/\n"
+        )
+        m = read_manifest(tmp_path)
+        assert m.name == "telem"
+        assert m.version == "1.2.3"
+        assert m.author == "Kyle"
+        assert m.license == "MIT"
+        assert len(m.env_requires) == 2
+        assert m.env_requires[0].name == "OPENAI_API_KEY"
+        assert m.env_requires[0].required is True
+        assert m.env_requires[1].required is False
+        assert m.env_requires[1].default == "http://127.0.0.1:8000"
+        assert m.distribution_owned == ["SOUL.md", "skills"]
+
+    def test_missing_name_rejected(self, tmp_path):
+        (tmp_path / MANIFEST_FILENAME).write_text("version: 1.0\n")
+        with pytest.raises(DistributionError, match="missing 'name'"):
+            read_manifest(tmp_path)
+
+    def test_env_requires_not_list_rejected(self, tmp_path):
+        (tmp_path / MANIFEST_FILENAME).write_text(
+            "name: bad\nenv_requires:\n  name: FOO\n"
+        )
+        with pytest.raises(DistributionError, match="env_requires must be a list"):
+            read_manifest(tmp_path)
+
+    def test_read_manifest_returns_none_when_absent(self, tmp_path):
+        assert read_manifest(tmp_path) is None
+
+    def test_owned_paths_default(self):
+        m = DistributionManifest(name="x")
+        assert m.owned_paths() == list(DEFAULT_DIST_OWNED)
+
+    def test_owned_paths_explicit(self):
+        m = DistributionManifest(name="x", distribution_owned=["SOUL.md", "skills"])
+        assert m.owned_paths() == ["SOUL.md", "skills"]
+
+    def test_roundtrip_write_read(self, tmp_path):
+        original = DistributionManifest(
+            name="rt",
+            version="1.0.0",
+            description="roundtrip",
+            env_requires=[EnvRequirement(name="FOO", description="foo")],
+        )
+        write_manifest(tmp_path, original)
+        parsed = read_manifest(tmp_path)
+        assert parsed.name == "rt"
+        assert parsed.env_requires[0].name == "FOO"
+
+
+# ===========================================================================
+# Version requirement checks
+# ===========================================================================
+
+
+class TestVersionRequires:
+
+    @pytest.mark.parametrize("spec,cur,ok", [
+        ("", "0.1.0", True),
+        (">=0.12.0", "0.12.0", True),
+        (">=0.12.0", "0.13.0", True),
+        (">=0.12.0", "0.11.9", False),
+        ("==0.12.0", "0.12.0", True),
+        ("==0.12.0", "0.13.0", False),
+        ("!=0.12.0", "0.13.0", True),
+        (">0.12.0", "0.12.1", True),
+        (">0.12.0", "0.12.0", False),
+        ("<0.13.0", "0.12.9", True),
+        ("<=0.12.0", "0.12.0", True),
+        ("0.12.0", "0.13.0", True),     # Bare = >=
+        ("0.12.0", "0.11.0", False),    # Bare = >=
+    ])
+    def test_check_matrix(self, spec, cur, ok):
+        if ok:
+            check_hermes_requires(spec, cur)
+        else:
+            with pytest.raises(DistributionError, match="requires Hermes"):
+                check_hermes_requires(spec, cur)
+
+    def test_parse_semver_handles_prerelease(self):
+        assert _parse_semver("0.12.0-rc1") == (0, 12, 0)
+        assert _parse_semver("v0.12.0+abc") == (0, 12, 0)
+
+    def test_parse_semver_pads(self):
+        assert _parse_semver("1") == (1, 0, 0)
+        assert _parse_semver("1.2") == (1, 2, 0)
+
+    def test_parse_semver_rejects_garbage(self):
+        with pytest.raises(DistributionError, match="Unparseable"):
+            _parse_semver("not-a-version")
+
+
+# ===========================================================================
+# Env template
+# ===========================================================================
+
+
+class TestEnvTemplate:
+
+    def test_required_is_uncommented(self):
+        m = DistributionManifest(
+            name="x",
+            env_requires=[EnvRequirement(name="FOO", description="foo key")],
+        )
+        out = _env_template_from_manifest(m)
+        assert "# foo key" in out
+        assert "# (required)" in out
+        assert "FOO=" in out
+        # No leading `# ` before FOO=
+        assert "\nFOO=" in out or out.startswith("FOO=") or "\nFOO=\n" in out or "FOO=\n" in out
+
+    def test_optional_is_commented(self):
+        m = DistributionManifest(
+            name="x",
+            env_requires=[EnvRequirement(name="BAR", required=False, default="http://x")],
+        )
+        out = _env_template_from_manifest(m)
+        assert "# (optional)" in out
+        assert "# BAR=http://x" in out
+
+    def test_empty_env_requires_is_header_only(self):
+        m = DistributionManifest(name="x")
+        out = _env_template_from_manifest(m)
+        assert "Hermes distribution" in out
+        assert "FOO" not in out
+
+
+# ===========================================================================
+# Source URL detection
+# ===========================================================================
+
+
+class TestLooksLikeGitUrl:
+
+    @pytest.mark.parametrize("src", [
+        "github.com/user/repo",
+        "https://github.com/user/repo",
+        "https://github.com/user/repo.git",
+        "http://example.com/repo",
+        "git@github.com:user/repo.git",
+        "ssh://git@example.com/repo.git",
+        "git://example.com/repo.git",
+    ])
+    def test_accepts_git_sources(self, src):
+        assert _looks_like_git_url(src)
+
+    @pytest.mark.parametrize("src", [
+        "/tmp/local/path",
+        "./relative/dir",
+        "~/profile",
+        "some-random-string",
+    ])
+    def test_rejects_non_git(self, src):
+        assert not _looks_like_git_url(src)
+
+
+# ===========================================================================
+# Install — fresh and force (from a local-directory source)
+# ===========================================================================
+
+
+class TestInstall:
+
+    def test_install_from_directory(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        plan = install_distribution(str(staged), name="installed")
+        assert plan.target_dir.is_dir()
+        assert (plan.target_dir / "SOUL.md").read_text() == "I am Source.\n"
+        assert (plan.target_dir / "skills" / "demo" / "SKILL.md").exists()
+        assert (plan.target_dir / "mcp.json").exists()
+        # Manifest on disk records canonical name + provenance
+        m = read_manifest(plan.target_dir)
+        assert m.name == "installed"
+        assert m.source == str(staged)
+
+    def test_install_uses_manifest_name_when_no_override(self, profile_env):
+        mf = DistributionManifest(name="telem", version="1.0.0")
+        staged = _make_staging_dir(profile_env, "telem", manifest=mf)
+        plan = install_distribution(str(staged))
+        assert plan.manifest.name == "telem"
+        assert plan.target_dir.name == "telem"
+
+    def test_install_rejects_existing_without_force(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        install_distribution(str(staged), name="existing")
+        with pytest.raises(DistributionError, match="already exists"):
+            install_distribution(str(staged), name="existing")
+
+    def test_install_with_force_overwrites(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        install_distribution(str(staged), name="target")
+        # Install again with --force succeeds
+        plan = install_distribution(str(staged), name="target", force=True)
+        assert plan.target_dir.is_dir()
+
+    def test_install_rejects_default_name(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        with pytest.raises(DistributionError, match="Cannot install"):
+            install_distribution(str(staged), name="default")
+
+    def test_install_rejects_non_distribution_directory(self, profile_env, tmp_path):
+        bogus = tmp_path / "bogus_dir"
+        bogus.mkdir()
+        (bogus / "some_file").write_text("hi")
+        with pytest.raises(DistributionError, match="No distribution.yaml"):
+            plan_install(str(bogus), tmp_path / "work", override_name="x")
+
+    def test_install_rejects_unknown_source(self, profile_env, tmp_path):
+        with pytest.raises(DistributionError, match="Cannot resolve"):
+            plan_install("definitely-not-a-thing", tmp_path / "work", override_name="x")
+
+    def test_install_emits_env_example_when_manifest_has_env(self, profile_env):
+        mf = DistributionManifest(
+            name="needs_env",
+            version="0.1.0",
+            env_requires=[EnvRequirement(name="OPENAI_API_KEY", description="key")],
+        )
+        staged = _make_staging_dir(profile_env, "needs_env", manifest=mf)
+        plan = install_distribution(str(staged), name="needs_env")
+        example = plan.target_dir / ".env.EXAMPLE"
+        assert example.is_file()
+        assert "OPENAI_API_KEY" in example.read_text()
+
+    def test_install_enforces_hermes_requires(self, profile_env, monkeypatch):
+        # Pin current Hermes version to something well below the requirement
+        import hermes_cli
+        monkeypatch.setattr(hermes_cli, "__version__", "0.1.0", raising=False)
+
+        mf = DistributionManifest(
+            name="future",
+            version="1.0.0",
+            hermes_requires=">=99.0.0",
+        )
+        staged = _make_staging_dir(profile_env, "future", manifest=mf)
+        with pytest.raises(DistributionError, match="requires Hermes"):
+            install_distribution(str(staged), name="future")
+
+
+# ===========================================================================
+# Update — preserves user data, preserves config by default
+# ===========================================================================
+
+
+class TestUpdate:
+
+    def test_update_preserves_user_data(self, profile_env):
+        # 1. Build staging dir, install
+        staged = _make_staging_dir(profile_env, "src")
+        plan = install_distribution(str(staged), name="telem")
+
+        # 2. Add user-owned data to the installed profile
+        (plan.target_dir / "memories").mkdir(exist_ok=True)
+        (plan.target_dir / "memories" / "MEMORY.md").write_text("# USER MEMORY\n")
+        (plan.target_dir / ".env").write_text("OPENAI_API_KEY=sk-user\n")
+        (plan.target_dir / "auth.json").write_text('{"user": "auth"}')
+        (plan.target_dir / "sessions").mkdir(exist_ok=True)
+        (plan.target_dir / "sessions" / "chat.json").write_text('{"s": 1}')
+
+        # 3. Bump source in the staging dir
+        (staged / "SOUL.md").write_text("I am Source v2.\n")
+
+        # 4. Update
+        update_distribution("telem", force_config=False)
+
+        # 5. Dist-owned changed
+        assert (plan.target_dir / "SOUL.md").read_text() == "I am Source v2.\n"
+        # 6. User-owned preserved
+        assert (plan.target_dir / "memories" / "MEMORY.md").read_text() == "# USER MEMORY\n"
+        assert (plan.target_dir / ".env").read_text() == "OPENAI_API_KEY=sk-user\n"
+        assert (plan.target_dir / "auth.json").read_text() == '{"user": "auth"}'
+        assert (plan.target_dir / "sessions" / "chat.json").read_text() == '{"s": 1}'
+
+    def test_update_preserves_config_by_default(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        plan = install_distribution(str(staged), name="t2")
+
+        # User edits config
+        (plan.target_dir / "config.yaml").write_text(
+            "model:\n  model: gpt-5\n# user override\n"
+        )
+
+        # Bump source config
+        (staged / "config.yaml").write_text("model:\n  model: claude\n")
+
+        update_distribution("t2", force_config=False)
+        assert "gpt-5" in (plan.target_dir / "config.yaml").read_text()
+        assert "user override" in (plan.target_dir / "config.yaml").read_text()
+
+    def test_update_force_config_overwrites(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        plan = install_distribution(str(staged), name="t3")
+
+        (plan.target_dir / "config.yaml").write_text("model:\n  model: gpt-5\n")
+
+        (staged / "config.yaml").write_text("model:\n  model: claude\n")
+
+        update_distribution("t3", force_config=True)
+        assert "claude" in (plan.target_dir / "config.yaml").read_text()
+        assert "gpt-5" not in (plan.target_dir / "config.yaml").read_text()
+
+    def test_update_missing_manifest_errors(self, profile_env):
+        # Make a profile without a manifest; update must refuse
+        from hermes_cli.profiles import create_profile
+        create_profile(name="plain", no_alias=True)
+        with pytest.raises(DistributionError, match="not a distribution"):
+            update_distribution("plain")
+
+
+# ===========================================================================
+# describe_distribution — info subcommand
+# ===========================================================================
+
+
+class TestDescribe:
+
+    def test_describe_existing_distribution(self, profile_env):
+        mf = DistributionManifest(
+            name="telem",
+            version="1.0.0",
+            description="compliance monitor",
+            env_requires=[EnvRequirement(name="API", description="api key")],
+        )
+        staged = _make_staging_dir(profile_env, "telem", manifest=mf)
+        install_distribution(str(staged), name="telem")
+        data = describe_distribution("telem")
+        assert data["name"] == "telem"
+        assert data["version"] == "1.0.0"
+        assert data["env_requires"][0]["name"] == "API"
+
+    def test_describe_non_distribution_returns_empty(self, profile_env):
+        from hermes_cli.profiles import create_profile
+        create_profile(name="plain", no_alias=True)
+        assert describe_distribution("plain") == {}
+
+    def test_describe_missing_profile_raises(self, profile_env):
+        with pytest.raises(DistributionError, match="does not exist"):
+            describe_distribution("nonexistent")
+
+
+# ===========================================================================
+# Security — USER_OWNED_EXCLUDE covers the right paths
+# ===========================================================================
+
+
+class TestSecurity:
+
+    def test_user_owned_exclude_covers_credentials(self):
+        assert "auth.json" in USER_OWNED_EXCLUDE
+        assert ".env" in USER_OWNED_EXCLUDE
+        assert "memories" in USER_OWNED_EXCLUDE
+        assert "sessions" in USER_OWNED_EXCLUDE
+        assert "local" in USER_OWNED_EXCLUDE
+
+    def test_install_does_not_import_credentials_from_staging(self, profile_env):
+        """If an author accidentally ships auth.json or .env in their
+        staging dir, the installer must NOT copy them to the target profile."""
+        staged = _make_staging_dir(profile_env, "src")
+        # Author leaks credentials into the staging tree (shouldn't happen, but...)
+        (staged / "auth.json").write_text('{"leaked": true}')
+        (staged / ".env").write_text("LEAKED=1")
+
+        plan = install_distribution(str(staged), name="clean")
+        assert not (plan.target_dir / "auth.json").exists(), "auth.json leaked"
+        # Fresh profile may have its own .env via the bootstrap; what we care
+        # about is that the leaked content didn't land in the target.
+        if (plan.target_dir / ".env").exists():
+            assert "LEAKED" not in (plan.target_dir / ".env").read_text()
+
+
+# ===========================================================================
+# Install-time metadata (installed_at stamp)
+# ===========================================================================
+
+
+class TestInstalledAtStamp:
+
+    def test_install_stamps_installed_at(self, profile_env):
+        staged = _make_staging_dir(profile_env, "src")
+        plan = install_distribution(str(staged), name="stamped")
+        mf = read_manifest(plan.target_dir)
+        assert mf.installed_at, "installed_at should be set after install"
+        # ISO-8601 UTC sanity: starts with 4-digit year, contains 'T', ends with '+00:00'.
+        assert mf.installed_at[:4].isdigit()
+        assert "T" in mf.installed_at
+        assert mf.installed_at.endswith("+00:00")
+
+    def test_update_refreshes_installed_at(self, profile_env, monkeypatch):
+        staged = _make_staging_dir(profile_env, "src")
+        install_distribution(str(staged), name="demo")
+        from hermes_cli.profiles import get_profile_dir
+        first = read_manifest(get_profile_dir("demo")).installed_at
+
+        # Freeze `datetime.now()` to a fixed future time so we can observe that
+        # update writes a NEW stamp (installs within the same second otherwise
+        # collide at iso-8601 seconds resolution).
+        import datetime as _dt
+        class _FakeDT(_dt.datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return _dt.datetime(2099, 1, 1, 0, 0, 0, tzinfo=tz or _dt.timezone.utc)
+        monkeypatch.setattr(
+            "hermes_cli.profile_distribution.datetime", _FakeDT, raising=True
+        )
+
+        from hermes_cli.profile_distribution import update_distribution
+        update_distribution("demo")
+        refreshed = read_manifest(get_profile_dir("demo")).installed_at
+        assert refreshed != first, "installed_at should change on update"
+        assert refreshed.startswith("2099-01-01"), refreshed
+
+
+# ===========================================================================
+# ProfileInfo exposes distribution metadata
+# ===========================================================================
+
+
+class TestProfileInfoDistribution:
+
+    def test_installed_distribution_shows_in_list(self, profile_env):
+        staged = _make_staging_dir(
+            profile_env, "src",
+            manifest=DistributionManifest(name="telem", version="1.2.3"),
+        )
+        install_distribution(str(staged), name="telem")
+
+        from hermes_cli.profiles import list_profiles
+        rows = {p.name: p for p in list_profiles()}
+        assert "telem" in rows
+        row = rows["telem"]
+        assert row.distribution_name == "telem"
+        assert row.distribution_version == "1.2.3"
+        assert row.distribution_source  # path populated, exact value depends on fixture
+
+    def test_plain_profile_has_no_distribution_fields(self, profile_env):
+        from hermes_cli.profiles import create_profile, list_profiles
+        create_profile(name="plain", no_alias=True)
+        rows = {p.name: p for p in list_profiles()}
+        assert rows["plain"].distribution_name is None
+        assert rows["plain"].distribution_version is None
+
+    def test_malformed_manifest_does_not_break_list(self, profile_env):
+        from hermes_cli.profiles import create_profile, list_profiles, get_profile_dir
+        create_profile(name="brokenmeta", no_alias=True)
+        # Write a distribution.yaml that isn't a valid mapping
+        (get_profile_dir("brokenmeta") / "distribution.yaml").write_text(
+            "not: [a, valid, mapping\n"  # broken YAML
+        )
+        # list_profiles must NOT raise; distribution_* stay None for this row.
+        rows = {p.name: p for p in list_profiles()}
+        assert rows["brokenmeta"].distribution_name is None
+
+
+# ===========================================================================
+# Error surfaces: validation failures should propagate as DistributionError
+# or ValueError (both caught and rendered cleanly by the CLI handler)
+# ===========================================================================
+
+
+class TestErrorSurfaces:
+
+    def test_bad_profile_name_raises_valueerror_not_traceback(self, profile_env, tmp_path):
+        """A manifest whose 'name' can't be used as a profile identifier
+        should raise ValueError from validate_profile_name — the CLI handler
+        catches both DistributionError and ValueError so users see a clean
+        'Error: ...' line instead of a Python traceback.
+        """
+        mf = DistributionManifest(name="Invalid Name With Spaces", version="0.1.0")
+        staged = _make_staging_dir(profile_env, "bad", manifest=mf)
+        with pytest.raises((ValueError, DistributionError)):
+            plan_install(str(staged), tmp_path / "work")
+
+    def test_path_traversal_name_rejected(self, profile_env, tmp_path):
+        mf = DistributionManifest(name="../../etc/passwd", version="0.1.0")
+        staged = _make_staging_dir(profile_env, "bad", manifest=mf)
+        with pytest.raises((ValueError, DistributionError)):
+            plan_install(str(staged), tmp_path / "work")
+

tests/hermes_cli/test_profiles.py

@@ -116,6 +116,14 @@ class TestValidateProfileName:
         with pytest.raises(ValueError):
             validate_profile_name("")
 
+    @pytest.mark.parametrize("name", ["hermes", "test", "tmp", "root", "sudo"])
+    def test_reserved_names_rejected(self, name):
+        """Reserved names collide with the Hermes install itself or with
+        common system binaries — reject them at validate time so
+        create/install/rename all share one gate."""
+        with pytest.raises(ValueError, match="reserved"):
+            validate_profile_name(name)
+
 
 # ===================================================================
 # TestGetProfileDir

tests/run_agent/test_provider_attribution_headers.py

@@ -65,6 +65,31 @@ def test_routermint_base_url_applies_user_agent_header(mock_openai):
     assert headers["User-Agent"].startswith("HermesAgent/")
 
 
+@patch("run_agent.OpenAI")
+def test_gmi_base_url_picks_up_profile_user_agent(mock_openai):
+    """GMI declares User-Agent on its ProviderProfile.default_headers.
+
+    The ``_apply_client_headers_for_base_url`` else-branch looks up the
+    provider profile and applies its default_headers, so no GMI-specific
+    branch is needed in run_agent.
+    """
+    mock_openai.return_value = MagicMock()
+    agent = AIAgent(
+        api_key="test-key",
+        base_url="https://api.gmi-serving.com/v1",
+        model="test/model",
+        provider="gmi",
+        quiet_mode=True,
+        skip_context_files=True,
+        skip_memory=True,
+    )
+
+    agent._apply_client_headers_for_base_url("https://api.gmi-serving.com/v1")
+
+    headers = agent._client_kwargs["default_headers"]
+    assert headers["User-Agent"].startswith("HermesAgent/")
+
+
 @patch("run_agent.OpenAI")
 def test_unknown_base_url_clears_default_headers(mock_openai):
     mock_openai.return_value = MagicMock()

tests/tools/test_cron_approval_mode.py

@@ -256,3 +256,77 @@ class TestCronModeInteractions:
 
         result = check_dangerous_command("rm -rf /tmp/stuff", "local")
         assert result["approved"]
+
+
+class TestCronWithGatewayOrigin:
+    """Cron jobs originating from a gateway platform must NOT be treated as gateway.
+
+    cron/scheduler.py binds HERMES_SESSION_PLATFORM via contextvars for
+    delivery routing (so cron output lands back in the origin chat). The
+    API-server approvals work (PR #20311) made check_dangerous_command treat
+    any contextvar-bound platform as a gateway session. That would route
+    cron-from-telegram/discord/etc. through submit_pending with no listener,
+    hanging the job instead of respecting approvals.cron_mode.
+    """
+
+    def test_cron_with_telegram_origin_uses_cron_mode_not_gateway(self, monkeypatch):
+        """Cron + contextvar platform=telegram + cron_mode=deny → BLOCKED, not pending."""
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="telegram", chat_id="123")
+        try:
+            from unittest.mock import patch as mock_patch
+            with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
+                result = check_dangerous_command("rm -rf /tmp/stuff", "local")
+                # Cron-mode path: BLOCKED message, NOT pending/approval_required.
+                assert not result["approved"]
+                assert "BLOCKED" in result["message"]
+                assert "cron_mode" in result["message"]
+                assert result.get("status") != "approval_required"
+        finally:
+            clear_session_vars(tokens)
+
+    def test_cron_with_telegram_origin_approve_mode_allows(self, monkeypatch):
+        """Cron + contextvar platform=telegram + cron_mode=approve → allowed via cron path."""
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="discord", chat_id="456")
+        try:
+            from unittest.mock import patch as mock_patch
+            with mock_patch("tools.approval._get_cron_approval_mode", return_value="approve"):
+                result = check_dangerous_command("rm -rf /tmp/stuff", "local")
+                assert result["approved"]
+                # Should NOT be a gateway-approval response.
+                assert result.get("status") != "approval_required"
+        finally:
+            clear_session_vars(tokens)
+
+    def test_cron_with_telegram_origin_combined_guard_uses_cron_mode(self, monkeypatch):
+        """check_all_command_guards must also honor cron_mode over gateway classification."""
+        monkeypatch.setenv("HERMES_CRON_SESSION", "1")
+        monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+        monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+        monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
+        monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+
+        from gateway.session_context import set_session_vars, clear_session_vars
+        tokens = set_session_vars(platform="telegram", chat_id="789")
+        try:
+            from unittest.mock import patch as mock_patch
+            with mock_patch("tools.approval._get_cron_approval_mode", return_value="deny"):
+                result = check_all_command_guards("rm -rf /tmp/stuff", "local")
+                assert not result["approved"]
+                assert "BLOCKED" in result["message"]
+                assert result.get("status") != "approval_required"
+        finally:
+            clear_session_vars(tokens)

tests/tools/test_microsoft_graph_auth.py

@@ -0,0 +1,179 @@
+"""Tests for tools/microsoft_graph_auth.py."""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+
+from tools.microsoft_graph_auth import (
+    CachedAccessToken,
+    DEFAULT_GRAPH_SCOPE,
+    GraphCredentials,
+    MicrosoftGraphConfigError,
+    MicrosoftGraphTokenError,
+    MicrosoftGraphTokenProvider,
+)
+
+
+class TestGraphCredentials:
+    def test_from_env_raises_for_missing_required_values(self):
+        with pytest.raises(MicrosoftGraphConfigError) as exc:
+            GraphCredentials.from_env({})
+        assert "MSGRAPH_TENANT_ID" in str(exc.value)
+        assert "MSGRAPH_CLIENT_ID" in str(exc.value)
+        assert "MSGRAPH_CLIENT_SECRET" in str(exc.value)
+
+    def test_from_env_optional_returns_none_when_not_configured(self):
+        assert GraphCredentials.from_env({}, required=False) is None
+
+    def test_from_env_builds_normalized_credentials(self):
+        creds = GraphCredentials.from_env(
+            {
+                "MSGRAPH_TENANT_ID": "tenant-123",
+                "MSGRAPH_CLIENT_ID": "client-456",
+                "MSGRAPH_CLIENT_SECRET": "secret-789",
+            }
+        )
+        assert creds is not None
+        assert creds.scope == DEFAULT_GRAPH_SCOPE
+        assert creds.token_url.endswith("/tenant-123/oauth2/v2.0/token")
+
+
+@pytest.mark.anyio
+class TestMicrosoftGraphTokenProvider:
+    async def test_reuses_cached_token_until_expiry(self):
+        calls: list[int] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            calls.append(1)
+            return httpx.Response(
+                200,
+                json={
+                    "access_token": f"token-{len(calls)}",
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                },
+            )
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+            transport=httpx.MockTransport(handler),
+        )
+
+        first = await provider.get_access_token()
+        second = await provider.get_access_token()
+
+        assert first == "token-1"
+        assert second == "token-1"
+        assert len(calls) == 1
+
+    async def test_concurrent_calls_share_one_token_fetch(self):
+        calls: list[int] = []
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+        )
+
+        async def _fake_fetch():
+            calls.append(1)
+            await asyncio.sleep(0)
+            return CachedAccessToken(
+                access_token="token-1",
+                token_type="Bearer",
+                expires_at=9_999_999_999,
+            )
+
+        provider._fetch_access_token = _fake_fetch  # type: ignore[method-assign]
+
+        first, second = await asyncio.gather(
+            provider.get_access_token(),
+            provider.get_access_token(),
+        )
+
+        assert first == "token-1"
+        assert second == "token-1"
+        assert len(calls) == 1
+
+    async def test_refreshes_when_cached_token_is_expired(self):
+        calls: list[int] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            calls.append(1)
+            expires_in = 0 if len(calls) == 1 else 3600
+            return httpx.Response(
+                200,
+                json={
+                    "access_token": f"token-{len(calls)}",
+                    "expires_in": expires_in,
+                    "token_type": "Bearer",
+                },
+            )
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+            transport=httpx.MockTransport(handler),
+            skew_seconds=0,
+        )
+
+        first = await provider.get_access_token()
+        second = await provider.get_access_token()
+
+        assert first == "token-1"
+        assert second == "token-2"
+        assert len(calls) == 2
+
+    async def test_force_refresh_bypasses_cache(self):
+        calls: list[int] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            calls.append(1)
+            return httpx.Response(
+                200,
+                json={
+                    "access_token": f"token-{len(calls)}",
+                    "expires_in": 3600,
+                },
+            )
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+            transport=httpx.MockTransport(handler),
+        )
+
+        first = await provider.get_access_token()
+        second = await provider.get_access_token(force_refresh=True)
+
+        assert first == "token-1"
+        assert second == "token-2"
+        assert len(calls) == 2
+
+    async def test_invalid_token_response_raises(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(200, json={"expires_in": 3600})
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+            transport=httpx.MockTransport(handler),
+        )
+
+        with pytest.raises(MicrosoftGraphTokenError) as exc:
+            await provider.get_access_token()
+        assert "access_token" in str(exc.value)
+
+    async def test_http_error_includes_server_message(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                401,
+                json={"error": "invalid_client", "error_description": "bad secret"},
+            )
+
+        provider = MicrosoftGraphTokenProvider(
+            GraphCredentials("tenant", "client", "secret"),
+            transport=httpx.MockTransport(handler),
+        )
+
+        with pytest.raises(MicrosoftGraphTokenError) as exc:
+            await provider.get_access_token()
+        assert "bad secret" in str(exc.value)

tests/tools/test_microsoft_graph_client.py

@@ -0,0 +1,257 @@
+"""Tests for tools/microsoft_graph_client.py."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import httpx
+import pytest
+
+from tools.microsoft_graph_auth import GraphCredentials, MicrosoftGraphTokenProvider
+from tools.microsoft_graph_client import (
+    MicrosoftGraphAPIError,
+    MicrosoftGraphClient,
+    MicrosoftGraphClientError,
+)
+
+
+def _make_provider() -> MicrosoftGraphTokenProvider:
+    provider = MicrosoftGraphTokenProvider(GraphCredentials("tenant", "client", "secret"))
+    provider._cached_token = type(  # type: ignore[attr-defined]
+        "Token",
+        (),
+        {
+            "access_token": "cached-token",
+            "is_expired": lambda self, skew_seconds=0: False,
+            "expires_in_seconds": 3600,
+        },
+    )()
+    return provider
+
+
+@pytest.mark.anyio
+class TestMicrosoftGraphClient:
+    async def test_attaches_bearer_token_header(self):
+        captured_auth: list[str] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            captured_auth.append(request.headers["Authorization"])
+            return httpx.Response(200, json={"ok": True})
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+        )
+        payload = await client.get_json("/me")
+        assert payload == {"ok": True}
+        assert captured_auth == ["Bearer cached-token"]
+
+    async def test_retries_on_rate_limit_and_uses_retry_after(self):
+        calls: list[int] = []
+        sleeps: list[float] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            calls.append(1)
+            if len(calls) == 1:
+                return httpx.Response(
+                    429,
+                    json={"error": {"code": "TooManyRequests", "message": "slow down"}},
+                    headers={"Retry-After": "3"},
+                )
+            return httpx.Response(200, json={"ok": True})
+
+        async def fake_sleep(delay: float) -> None:
+            sleeps.append(delay)
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+            sleep=fake_sleep,
+            max_retries=2,
+        )
+
+        payload = await client.get_json("/me")
+
+        assert payload == {"ok": True}
+        assert len(calls) == 2
+        assert sleeps == [3.0]
+
+    async def test_raises_api_error_after_retry_budget_exhausted(self):
+        sleeps: list[float] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(503, json={"error": {"message": "unavailable"}})
+
+        async def fake_sleep(delay: float) -> None:
+            sleeps.append(delay)
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+            sleep=fake_sleep,
+            max_retries=1,
+        )
+
+        with pytest.raises(MicrosoftGraphAPIError) as exc:
+            await client.get_json("/me")
+        assert exc.value.status_code == 503
+        assert sleeps == [0.5]
+
+    async def test_collect_paginated_flattens_value_arrays(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            if str(request.url).endswith("/items"):
+                return httpx.Response(
+                    200,
+                    json={
+                        "value": [{"id": "1"}],
+                        "@odata.nextLink": "https://graph.microsoft.com/v1.0/items?page=2",
+                    },
+                )
+            return httpx.Response(200, json={"value": [{"id": "2"}]})
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+        )
+        items = await client.collect_paginated("/items")
+        assert items == [{"id": "1"}, {"id": "2"}]
+
+    async def test_download_to_file_writes_binary_content(self, tmp_path: Path):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                200,
+                content=b"meeting-recording",
+                headers={"content-type": "video/mp4"},
+            )
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+        )
+        destination = tmp_path / "recording.mp4"
+        result = await client.download_to_file("/drive/item/content", destination)
+
+        assert destination.read_bytes() == b"meeting-recording"
+        assert result["content_type"] == "video/mp4"
+        assert result["size_bytes"] == len(b"meeting-recording")
+
+    async def test_download_to_file_streams_large_payload_in_chunks(
+        self, tmp_path: Path, monkeypatch
+    ):
+        """Recordings can be hundreds of MB; verify the body is streamed.
+
+        Uses a payload larger than the chunk size and counts how many
+        ``aiter_bytes`` iterations the download loop performs. If the
+        response were buffered in memory before the loop ran, only one
+        non-empty chunk would be yielded.
+        """
+        payload = b"x" * (512 * 1024)  # 512 KiB
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                200,
+                content=payload,
+                headers={"content-type": "video/mp4"},
+            )
+
+        chunk_calls: list[int] = []
+        original_aiter_bytes = httpx.Response.aiter_bytes
+
+        async def counting_aiter_bytes(self, chunk_size: int | None = None):
+            async for chunk in original_aiter_bytes(self, chunk_size):
+                chunk_calls.append(len(chunk))
+                yield chunk
+
+        monkeypatch.setattr(httpx.Response, "aiter_bytes", counting_aiter_bytes)
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+        )
+        destination = tmp_path / "big-recording.mp4"
+        result = await client.download_to_file(
+            "/drive/item/content", destination, chunk_size=65536
+        )
+
+        assert destination.read_bytes() == payload
+        assert result["size_bytes"] == len(payload)
+        assert len(chunk_calls) >= 2, (
+            "Expected multiple chunks; got a single chunk "
+            f"which suggests the body was buffered: {chunk_calls}"
+        )
+        assert not (tmp_path / "big-recording.mp4.part").exists()
+
+    async def test_download_to_file_retries_on_transient_server_error(
+        self, tmp_path: Path
+    ):
+        calls: list[int] = []
+        sleeps: list[float] = []
+
+        def handler(request: httpx.Request) -> httpx.Response:
+            calls.append(1)
+            if len(calls) == 1:
+                return httpx.Response(
+                    503, json={"error": {"message": "unavailable"}}
+                )
+            return httpx.Response(
+                200,
+                content=b"payload",
+                headers={"content-type": "application/octet-stream"},
+            )
+
+        async def fake_sleep(delay: float) -> None:
+            sleeps.append(delay)
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+            sleep=fake_sleep,
+            max_retries=2,
+        )
+        destination = tmp_path / "artifact.bin"
+        result = await client.download_to_file("/drive/item/content", destination)
+
+        assert destination.read_bytes() == b"payload"
+        assert result["size_bytes"] == len(b"payload")
+        assert len(calls) == 2
+        assert sleeps == [0.5]
+        assert not (tmp_path / "artifact.bin.part").exists()
+
+    async def test_download_to_file_cleans_partial_file_on_exhausted_retries(
+        self, tmp_path: Path
+    ):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(503, json={"error": {"message": "unavailable"}})
+
+        async def fake_sleep(delay: float) -> None:
+            return None
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+            sleep=fake_sleep,
+            max_retries=1,
+        )
+        destination = tmp_path / "artifact.bin"
+
+        with pytest.raises(MicrosoftGraphAPIError):
+            await client.download_to_file("/drive/item/content", destination)
+
+        assert not destination.exists()
+        assert not (tmp_path / "artifact.bin.part").exists()
+
+    async def test_invalid_json_response_raises_client_error(self):
+        def handler(request: httpx.Request) -> httpx.Response:
+            return httpx.Response(
+                200,
+                content=b"not-json",
+                headers={"content-type": "application/json"},
+            )
+
+        client = MicrosoftGraphClient(
+            _make_provider(),
+            transport=httpx.MockTransport(handler),
+        )
+
+        with pytest.raises(MicrosoftGraphClientError):
+            await client.get_json("/me")