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

tests/hermes_cli/test_apply_profile_override.py

@@ -0,0 +1,141 @@
+"""Regression tests for _apply_profile_override HERMES_HOME guard (issue #22502).
+
+When HERMES_HOME is set to the hermes root (e.g. systemd hardcodes
+HERMES_HOME=/root/.hermes), _apply_profile_override must still read
+active_profile and update HERMES_HOME to the profile directory.
+
+When HERMES_HOME is already a profile directory (.../profiles/<name>),
+_apply_profile_override must trust it and return without re-reading
+active_profile (child-process inheritance contract).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+
+def _run_apply_profile_override(
+    tmp_path, monkeypatch, *, hermes_home: str | None, active_profile: str | None,
+    argv: list[str] | None = None,
+):
+    """Run _apply_profile_override in isolation.
+
+    Returns the value of os.environ["HERMES_HOME"] after the call,
+    or None if unset.
+    """
+    hermes_root = tmp_path / ".hermes"
+    hermes_root.mkdir(parents=True, exist_ok=True)
+
+    if active_profile is not None:
+        (hermes_root / "active_profile").write_text(active_profile)
+
+    if active_profile and active_profile != "default":
+        (hermes_root / "profiles" / active_profile).mkdir(parents=True, exist_ok=True)
+
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    if hermes_home is not None:
+        monkeypatch.setenv("HERMES_HOME", hermes_home)
+    else:
+        monkeypatch.delenv("HERMES_HOME", raising=False)
+
+    monkeypatch.setattr(sys, "argv", argv or ["hermes", "gateway", "start"])
+
+    from hermes_cli.main import _apply_profile_override
+    _apply_profile_override()
+
+    return os.environ.get("HERMES_HOME")
+
+
+class TestApplyProfileOverrideHermesHomeGuard:
+    """Regression guard for issue #22502.
+
+    Verifies that HERMES_HOME pointing to the hermes root does NOT suppress
+    the active_profile check, while HERMES_HOME already pointing to a
+    profile directory IS trusted as-is.
+    """
+
+    def test_hermes_home_at_root_with_active_profile_is_redirected(
+        self, tmp_path, monkeypatch
+    ):
+        """HERMES_HOME=/root/.hermes + active_profile=coder must redirect
+        HERMES_HOME to .../profiles/coder.
+
+        Bug scenario from #22502: systemd sets HERMES_HOME to the hermes root
+        and the user switches to a profile via `hermes profile use`.
+        Before the fix, the guard returned early and active_profile was ignored.
+        """
+        hermes_root = tmp_path / ".hermes"
+        hermes_root.mkdir(parents=True, exist_ok=True)
+
+        result = _run_apply_profile_override(
+            tmp_path,
+            monkeypatch,
+            hermes_home=str(hermes_root),
+            active_profile="coder",
+        )
+
+        assert result is not None, "HERMES_HOME must be set after profile redirect"
+        assert "profiles" in result, (
+            f"Expected HERMES_HOME to point into profiles/ dir, got: {result!r}"
+        )
+        assert result.endswith("coder"), (
+            f"Expected HERMES_HOME to end with 'coder', got: {result!r}"
+        )
+
+    def test_hermes_home_already_profile_dir_is_trusted(self, tmp_path, monkeypatch):
+        """HERMES_HOME=.../profiles/coder must not be overridden even when
+        active_profile says something different.
+
+        Preserves the child-process inheritance contract: a subprocess spawned
+        with HERMES_HOME already set to a specific profile must stay in that
+        profile.
+        """
+        hermes_root = tmp_path / ".hermes"
+        profile_dir = hermes_root / "profiles" / "coder"
+        profile_dir.mkdir(parents=True, exist_ok=True)
+
+        (hermes_root / "active_profile").write_text("other")
+
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+        monkeypatch.setenv("HERMES_HOME", str(profile_dir))
+        monkeypatch.setattr(sys, "argv", ["hermes", "gateway", "start"])
+
+        from hermes_cli.main import _apply_profile_override
+        _apply_profile_override()
+
+        assert os.environ.get("HERMES_HOME") == str(profile_dir), (
+            "HERMES_HOME must remain unchanged when already pointing to a profile dir"
+        )
+
+    def test_hermes_home_unset_reads_active_profile(self, tmp_path, monkeypatch):
+        """Classic case: HERMES_HOME unset + active_profile=coder must set
+        HERMES_HOME to the profile directory (existing behaviour must not regress).
+        """
+        result = _run_apply_profile_override(
+            tmp_path,
+            monkeypatch,
+            hermes_home=None,
+            active_profile="coder",
+        )
+
+        assert result is not None
+        assert "coder" in result
+
+    def test_hermes_home_unset_default_profile_no_redirect(self, tmp_path, monkeypatch):
+        """active_profile=default must not redirect HERMES_HOME."""
+        hermes_root = tmp_path / ".hermes"
+        hermes_root.mkdir(parents=True, exist_ok=True)
+
+        monkeypatch.setattr(Path, "home", lambda: tmp_path)
+        monkeypatch.delenv("HERMES_HOME", raising=False)
+        monkeypatch.setattr(sys, "argv", ["hermes", "gateway", "start"])
+        (hermes_root / "active_profile").write_text("default")
+
+        from hermes_cli.main import _apply_profile_override
+        _apply_profile_override()
+
+        assert os.environ.get("HERMES_HOME") is None

tests/hermes_cli/test_cmd_update.py

@@ -112,12 +112,14 @@ class TestCmdUpdateBranchFallback:
     def test_update_refreshes_repo_and_tui_node_dependencies(
         self, mock_run, mock_which, _mock_web_ui_build_needed, mock_args
     ):
+        from hermes_cli import main as hm
+
         mock_which.side_effect = {"uv": "/usr/bin/uv", "npm": "/usr/bin/npm"}.get
         mock_run.side_effect = _make_run_side_effect(
             branch="main", verify_ok=True, commit_count="1"
         )
-
-        cmd_update(mock_args)
+        with patch.object(hm, "_is_termux_env", return_value=False):
+            cmd_update(mock_args)
 
         npm_calls = [
             (call.args[0], call.kwargs.get("cwd"))
@@ -146,9 +148,11 @@ class TestCmdUpdateBranchFallback:
             "--no-audit",
             "--progress=false",
         ]
-        assert npm_calls == [
+        assert npm_calls[:2] == [
             (full_flags, PROJECT_ROOT),
             (app_flags, PROJECT_ROOT / "ui-tui"),
+        ]
+        assert npm_calls[2:] == [
             (["/usr/bin/npm", "ci", "--silent"], PROJECT_ROOT / "apps" / "dashboard"),
             (["/usr/bin/npm", "run", "build"], PROJECT_ROOT / "apps" / "dashboard"),
         ]
@@ -268,3 +272,26 @@ def test_is_termux_env_false_for_non_termux_prefix():
     from hermes_cli import main as hm
 
     assert hm._is_termux_env({"PREFIX": "/usr/local"}) is False
+
+
+def test_load_installable_optional_extras_supports_termux_group(tmp_path, monkeypatch):
+    from hermes_cli import main as hm
+
+    pyproject = tmp_path / "pyproject.toml"
+    pyproject.write_text(
+        """
+[project]
+name = "x"
+version = "0.0.0"
+
+[project.optional-dependencies]
+all = ["x[mcp]"]
+termux-all = ["x[termux]", "x[mcp]"]
+mcp = ["mcp>=1"]
+termux = ["rich>=14"]
+""".strip()
+    )
+    monkeypatch.setattr(hm, "PROJECT_ROOT", tmp_path)
+
+    assert hm._load_installable_optional_extras(group="all") == ["mcp"]
+    assert hm._load_installable_optional_extras(group="termux-all") == ["termux", "mcp"]

tests/hermes_cli/test_codex_cli_model_picker.py

@@ -75,6 +75,37 @@ def test_normal_path_still_works(hermes_auth_only_env):
     assert "openai-codex" in slugs
 
 
+def test_codex_picker_uses_live_codex_catalog(hermes_auth_only_env, tmp_path, monkeypatch):
+    """The gateway /model picker should surface Codex CLI-only listed models."""
+    from hermes_cli.model_switch import list_authenticated_providers
+
+    codex_home = tmp_path / "codex-home"
+    codex_home.mkdir()
+    (codex_home / "models_cache.json").write_text(json.dumps({
+        "models": [
+            {"slug": "gpt-5.5", "priority": 0, "supported_in_api": True},
+            {"slug": "gpt-5.3-codex-spark", "priority": 7, "supported_in_api": False},
+        ]
+    }))
+    monkeypatch.setenv("CODEX_HOME", str(codex_home))
+    # Force the cache fallback path — without this the test issues a real
+    # 10s HTTP probe to chatgpt.com/backend-api/codex/models which is both
+    # slow and non-deterministic in CI/sandboxed environments.
+    monkeypatch.setattr(
+        "hermes_cli.codex_models._fetch_models_from_api",
+        lambda access_token: [],
+    )
+
+    providers = list_authenticated_providers(
+        current_provider="openai-codex",
+        max_models=10,
+    )
+
+    codex = next(p for p in providers if p["slug"] == "openai-codex")
+    assert "gpt-5.3-codex-spark" in codex["models"]
+    assert codex["total_models"] == len(codex["models"])
+
+
 @pytest.fixture()
 def claude_code_only_env(tmp_path, monkeypatch):
     """Set up an environment where Anthropic credentials only exist in

tests/hermes_cli/test_codex_models.py

@@ -1,10 +1,6 @@
 import json
-import os
-import sys
 from unittest.mock import patch
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
 from hermes_cli.codex_models import DEFAULT_CODEX_MODELS, get_codex_model_ids
 
 
@@ -17,6 +13,7 @@ def test_get_codex_model_ids_prioritizes_default_and_cache(tmp_path, monkeypatch
             {
                 "models": [
                     {"slug": "gpt-5.3-codex", "priority": 20, "supported_in_api": True},
+                    {"slug": "gpt-5.3-codex-spark", "priority": 6, "supported_in_api": False},
                     {"slug": "gpt-5.1-codex", "priority": 5, "supported_in_api": True},
                     {"slug": "gpt-5.4", "priority": 1, "supported_in_api": True},
                     {"slug": "gpt-5-hidden-codex", "priority": 2, "visibility": "hidden"},
@@ -31,6 +28,9 @@ def test_get_codex_model_ids_prioritizes_default_and_cache(tmp_path, monkeypatch
     assert models[0] == "gpt-5.2-codex"
     assert "gpt-5.1-codex" in models
     assert "gpt-5.3-codex" in models
+    # Codex CLI marks Spark unsupported in the public API, but the Codex
+    # backend still accepts it via the OAuth-backed CLI/Hermes route.
+    assert "gpt-5.3-codex-spark" in models
     # Non-codex-suffixed models are included when the cache says they're available
     assert "gpt-5.4" in models
     assert "gpt-5.4-mini" in models
@@ -54,7 +54,7 @@ def test_get_codex_model_ids_falls_back_to_curated_defaults(tmp_path, monkeypatc
 
     assert models[: len(DEFAULT_CODEX_MODELS)] == DEFAULT_CODEX_MODELS
     assert "gpt-5.4" in models
-    assert "gpt-5.3-codex-spark" not in models
+    assert "gpt-5.3-codex-spark" in models
 
 
 def test_get_codex_model_ids_adds_forward_compat_models_from_templates(monkeypatch):
@@ -65,7 +65,49 @@ def test_get_codex_model_ids_adds_forward_compat_models_from_templates(monkeypat
 
     models = get_codex_model_ids(access_token="codex-access-token")
 
-    assert models == ["gpt-5.2-codex", "gpt-5.4-mini", "gpt-5.4", "gpt-5.3-codex"]
+    assert models == [
+        "gpt-5.2-codex",
+        "gpt-5.4-mini",
+        "gpt-5.4",
+        "gpt-5.3-codex",
+        "gpt-5.3-codex-spark",
+    ]
+
+
+def test_fetch_from_api_keeps_supported_in_api_false_models(monkeypatch):
+    """Regression: gpt-5.3-codex-spark is returned by the live Codex backend
+    with ``supported_in_api: false`` because it isn't in the public OpenAI
+    API. The Codex CLI / OAuth route still serves it for ChatGPT Pro
+    accounts, so we must not drop it on that flag. visibility=hidden is
+    the separate signal that *should* still filter entries out.
+    """
+    import sys
+    from hermes_cli import codex_models
+
+    class _FakeResp:
+        status_code = 200
+
+        def json(self):
+            return {
+                "models": [
+                    {"slug": "gpt-5.5", "priority": 0, "supported_in_api": True},
+                    {"slug": "gpt-5.3-codex-spark", "priority": 7, "supported_in_api": False},
+                    {"slug": "gpt-5-internal", "priority": 99, "visibility": "hidden"},
+                ]
+            }
+
+    class _FakeHttpx:
+        @staticmethod
+        def get(url, headers=None, timeout=None):
+            return _FakeResp()
+
+    monkeypatch.setitem(sys.modules, "httpx", _FakeHttpx)
+
+    models = codex_models._fetch_models_from_api(access_token="tok")
+
+    assert "gpt-5.5" in models
+    assert "gpt-5.3-codex-spark" in models
+    assert "gpt-5-internal" not in models
 
 
 def test_model_command_uses_runtime_access_token_for_codex_list(monkeypatch):

tests/hermes_cli/test_curator_recent_run_notice.py

@@ -0,0 +1,162 @@
+"""Tests for `_print_curator_recent_run_notice`.
+
+The notice prints the most recent curator run summary on `hermes update`,
+exactly once per run. Show-once is enforced by stamping
+`last_run_summary_shown_at` in curator state after printing.
+
+Why this matters: the curator runs in the background (gateway tick + CLI
+session start) so users normally never see the rename map. `hermes update`
+is the high-attention surface where consolidations should land.
+"""
+
+from __future__ import annotations
+
+import importlib
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def curator_env(tmp_path, monkeypatch, capsys):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    (home / "skills").mkdir()
+    (home / "logs").mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+    import hermes_constants
+    importlib.reload(hermes_constants)
+    from agent import curator
+    importlib.reload(curator)
+    from hermes_cli import main as hermes_main
+    importlib.reload(hermes_main)
+
+    yield {
+        "curator": curator,
+        "main": hermes_main,
+        "capsys": capsys,
+    }
+
+
+def _set_state(curator_mod, **fields):
+    state = curator_mod.load_state()
+    state.update(fields)
+    curator_mod.save_state(state)
+
+
+def test_silent_when_no_curator_run_yet(curator_env):
+    """First-run notice handles this case; recent-run notice stays silent."""
+    curator_env["main"]._print_curator_recent_run_notice()
+    out = curator_env["capsys"].readouterr().out
+    assert "Skill curator — last run" not in out
+
+
+def test_silent_when_summary_is_single_line(curator_env):
+    """No archives = no rename map = nothing to surface. But still stamps shown."""
+    now = datetime.now(timezone.utc).isoformat()
+    _set_state(
+        curator_env["curator"],
+        last_run_at=now,
+        last_run_summary="auto: no changes; llm: no change",
+    )
+    curator_env["main"]._print_curator_recent_run_notice()
+    out = curator_env["capsys"].readouterr().out
+    assert "Skill curator — last run" not in out
+    # Should still mark shown so we don't reconsider on every update.
+    state = curator_env["curator"].load_state()
+    assert state["last_run_summary_shown_at"] == now
+
+
+def test_prints_multiline_summary_with_rename_map(curator_env):
+    """Multi-line summary (rename map appended) prints with timestamp + footer."""
+    now = datetime.now(timezone.utc).isoformat()
+    summary = (
+        "auto: 1 marked stale; llm: consolidated 2 into 1\n"
+        "archived 2 skill(s):\n"
+        "  • pdf-extraction → document-tools\n"
+        "  • docx-extraction → document-tools\n"
+        "full report: hermes curator status"
+    )
+    _set_state(
+        curator_env["curator"],
+        last_run_at=now,
+        last_run_summary=summary,
+    )
+    curator_env["main"]._print_curator_recent_run_notice()
+    out = curator_env["capsys"].readouterr().out
+    assert "Skill curator — last run" in out
+    assert "pdf-extraction → document-tools" in out
+    assert "docx-extraction → document-tools" in out
+    assert "shows once per curator run" in out
+
+
+def test_show_once_semantics(curator_env):
+    """Calling twice prints once; second call is silent until a new run lands."""
+    now = datetime.now(timezone.utc).isoformat()
+    summary = (
+        "auto: no changes; llm: consolidated 1 into 1\n"
+        "archived 1 skill(s):\n"
+        "  • old → new\n"
+        "full report: hermes curator status"
+    )
+    _set_state(
+        curator_env["curator"],
+        last_run_at=now,
+        last_run_summary=summary,
+    )
+
+    curator_env["main"]._print_curator_recent_run_notice()
+    first = curator_env["capsys"].readouterr().out
+    assert "old → new" in first
+
+    curator_env["main"]._print_curator_recent_run_notice()
+    second = curator_env["capsys"].readouterr().out
+    assert second == "", "second call must be silent (already shown)"
+
+
+def test_new_run_resets_show_once(curator_env):
+    """A newer curator run with rename data prints again, even though one was already shown."""
+    older = (datetime.now(timezone.utc) - timedelta(hours=8)).isoformat()
+    _set_state(
+        curator_env["curator"],
+        last_run_at=older,
+        last_run_summary=(
+            "auto: no changes; llm: consolidated 1 into 1\n"
+            "archived 1 skill(s):\n"
+            "  • thing-a → umbrella\n"
+            "full report: hermes curator status"
+        ),
+    )
+    curator_env["main"]._print_curator_recent_run_notice()
+    curator_env["capsys"].readouterr()  # drain
+
+    # New run lands.
+    newer = datetime.now(timezone.utc).isoformat()
+    _set_state(
+        curator_env["curator"],
+        last_run_at=newer,
+        last_run_summary=(
+            "auto: no changes; llm: consolidated 1 into 1\n"
+            "archived 1 skill(s):\n"
+            "  • thing-b → umbrella\n"
+            "full report: hermes curator status"
+        ),
+    )
+    curator_env["main"]._print_curator_recent_run_notice()
+    out = curator_env["capsys"].readouterr().out
+    assert "thing-b → umbrella" in out
+    assert "thing-a" not in out  # only the newer run shows
+
+
+def test_format_time_ago_buckets(curator_env):
+    """Smoke test the time formatter — drives the `last run Xh ago` line."""
+    fmt = curator_env["main"]._format_time_ago
+    now = datetime.now(timezone.utc)
+    assert fmt((now - timedelta(seconds=10)).isoformat()) == "just now"
+    assert fmt((now - timedelta(minutes=5)).isoformat()) == "5m ago"
+    assert fmt((now - timedelta(hours=3)).isoformat()) == "3h ago"
+    assert fmt((now - timedelta(days=2)).isoformat()) == "2d ago"
+    assert fmt("not-a-real-iso-string") == "recently"

tests/hermes_cli/test_destructive_slash_confirm_gate.py

@@ -0,0 +1,86 @@
+"""Tests for the approvals.destructive_slash_confirm config gate.
+
+Destructive session slash commands (/clear, /new, /reset, /undo) discard
+conversation state.  This config key (default True) gates a three-option
+confirmation prompt — "Always Approve" flips the key to False so future
+destructive commands run silently.
+
+See gateway/run.py::_maybe_confirm_destructive_slash and
+cli.py::_confirm_destructive_slash for the runtime gate.
+"""
+
+from __future__ import annotations
+
+from hermes_cli.config import DEFAULT_CONFIG
+
+
+class TestDestructiveSlashConfirmDefault:
+    def test_default_config_has_the_key(self):
+        approvals = DEFAULT_CONFIG.get("approvals")
+        assert isinstance(approvals, dict)
+        assert "destructive_slash_confirm" in approvals
+
+    def test_default_is_true(self):
+        # New installs confirm by default — destructive commands must not
+        # silently wipe history without an explicit user "yes".
+        assert DEFAULT_CONFIG["approvals"]["destructive_slash_confirm"] is True
+
+    def test_shape_matches_other_approval_keys(self):
+        approvals = DEFAULT_CONFIG["approvals"]
+        assert isinstance(approvals.get("destructive_slash_confirm"), bool)
+        # Sibling key shape sanity — same flat dict level as mcp_reload_confirm.
+        assert isinstance(approvals.get("mcp_reload_confirm"), bool)
+
+
+class TestUserConfigMerge:
+    """If a user has a pre-existing config without this key, load_config
+    should fill it in from DEFAULT_CONFIG (deep merge preserves keys the
+    user didn't override)."""
+
+    def test_existing_user_config_without_key_gets_default(self, tmp_path, monkeypatch):
+        import yaml
+
+        home = tmp_path / ".hermes"
+        home.mkdir()
+        cfg_path = home / "config.yaml"
+        legacy = {
+            "approvals": {"mode": "manual", "timeout": 60, "cron_mode": "deny"},
+        }
+        cfg_path.write_text(yaml.safe_dump(legacy))
+
+        monkeypatch.setenv("HERMES_HOME", str(home))
+        import importlib
+        import hermes_cli.config as cfg_mod
+        importlib.reload(cfg_mod)
+
+        cfg = cfg_mod.load_config()
+        assert cfg["approvals"]["destructive_slash_confirm"] is True
+
+    def test_existing_user_config_with_false_key_survives_merge(
+        self, tmp_path, monkeypatch,
+    ):
+        """A user who clicked "Always Approve" (key=false) must keep that
+        setting — the default-true value must not win on later loads.
+        """
+        import yaml
+
+        home = tmp_path / ".hermes"
+        home.mkdir()
+        cfg_path = home / "config.yaml"
+        user_cfg = {
+            "approvals": {
+                "mode": "manual",
+                "timeout": 60,
+                "cron_mode": "deny",
+                "destructive_slash_confirm": False,
+            },
+        }
+        cfg_path.write_text(yaml.safe_dump(user_cfg))
+
+        monkeypatch.setenv("HERMES_HOME", str(home))
+        import importlib
+        import hermes_cli.config as cfg_mod
+        importlib.reload(cfg_mod)
+
+        cfg = cfg_mod.load_config()
+        assert cfg["approvals"]["destructive_slash_confirm"] is False

tests/hermes_cli/test_doctor_dedicated_provider_skip.py

@@ -0,0 +1,50 @@
+"""Regression: hermes doctor must not run a generic Bearer-auth health
+check for providers that already have a dedicated check (Anthropic,
+OpenRouter, Bedrock).
+
+Anthropic's native API requires `x-api-key` + `anthropic-version` headers;
+the generic loop sends `Authorization: Bearer ...` which Anthropic answers
+with HTTP 404. The dedicated check at hermes_cli/doctor.py already covers
+Anthropic with the right headers, so the pluggable profile must be
+skipped by `_build_apikey_providers_list()`.
+
+See: NousResearch/hermes-agent#22346
+"""
+
+from __future__ import annotations
+
+
+def test_build_apikey_providers_list_skips_dedicated_check_providers():
+    from hermes_cli import doctor
+
+    # Force a rebuild — the module caches the list on first call.
+    doctor._APIKEY_PROVIDERS_CACHE = None
+    entries = doctor._build_apikey_providers_list()
+
+    # Tuple shape: (display_name, env_vars, default_url, base_env, supports_health_check)
+    names = {entry[0].lower() for entry in entries}
+    assert not any("anthropic" in name for name in names), (
+        f"Anthropic provider profile leaked into generic Bearer-auth health "
+        f"check loop. Dedicated check above already covers it with "
+        f"x-api-key headers. Got entries: {sorted(names)}"
+    )
+    assert not any("openrouter" in name for name in names), (
+        f"OpenRouter has a dedicated check; generic loop must skip it. "
+        f"Got: {sorted(names)}"
+    )
+    assert not any("bedrock" in name for name in names), (
+        f"Bedrock uses AWS SDK creds, not Bearer auth; generic loop must skip. "
+        f"Got: {sorted(names)}"
+    )
+
+
+def test_build_apikey_providers_list_includes_non_dedicated_providers():
+    """Sanity guard: the skip-set must not strip every provider."""
+    from hermes_cli import doctor
+
+    doctor._APIKEY_PROVIDERS_CACHE = None
+    entries = doctor._build_apikey_providers_list()
+
+    names = {entry[0] for entry in entries}
+    assert "DeepSeek" in names
+    assert "Z.AI / GLM" in names

tests/hermes_cli/test_gateway.py

@@ -13,6 +13,21 @@ def _install_fake_gateway_run(monkeypatch, start_gateway):
     module = ModuleType("gateway.run")
     module.start_gateway = start_gateway
     monkeypatch.setitem(sys.modules, "gateway.run", module)
+    # ``run_gateway()`` calls ``refresh_systemd_unit_if_needed()`` on every
+    # invocation so that restart settings stay current after exit-code-75
+    # respawns. That helper writes to ``Path.home() / ".config/systemd/user
+    # /hermes-gateway.service"`` and runs ``systemctl --user daemon-reload``
+    # — both target the *real* user environment because the conftest only
+    # sandboxes ``HERMES_HOME``, not ``HOME``. Tests that drive
+    # ``run_gateway()`` end-to-end with a fake ``start_gateway`` MUST stub
+    # the refresh call too, or every run rewrites the developer's installed
+    # unit (baking in the test's pytest-tmp ``HERMES_HOME`` value, which
+    # systemd then uses on the next boot — silently breaking the gateway
+    # for the developer).
+    monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
+    monkeypatch.setattr(
+        gateway, "refresh_systemd_unit_if_needed", lambda system=False: False
+    )
 
 
 def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys):
@@ -90,6 +105,66 @@ def test_run_gateway_root_guard_has_escape_hatch(monkeypatch):
     assert calls == [(True, 2)]
 
 
+def test_run_gateway_windows_foreground_keeps_ctrl_c_enabled(monkeypatch):
+    calls = []
+
+    def fake_start_gateway(*, replace, verbosity):
+        calls.append((replace, verbosity))
+        return object()
+
+    class _TTY:
+        def isatty(self):
+            return True
+
+    signal_calls = []
+
+    def fake_signal(sig, handler):
+        signal_calls.append((sig, handler))
+
+    _install_fake_gateway_run(monkeypatch, fake_start_gateway)
+    monkeypatch.setattr(gateway, "is_windows", lambda: True)
+    monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
+    monkeypatch.setattr(gateway.sys, "stdin", _TTY())
+    monkeypatch.delenv("HERMES_GATEWAY_DETACHED", raising=False)
+    monkeypatch.setattr(gateway.signal, "signal", fake_signal)
+    monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)
+
+    gateway.run_gateway()
+
+    assert calls == [(False, 0)]
+    assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) not in signal_calls
+
+
+def test_run_gateway_windows_detached_absorbs_console_controls(monkeypatch):
+    calls = []
+
+    def fake_start_gateway(*, replace, verbosity):
+        calls.append((replace, verbosity))
+        return object()
+
+    class _TTY:
+        def isatty(self):
+            return True
+
+    signal_calls = []
+
+    def fake_signal(sig, handler):
+        signal_calls.append((sig, handler))
+
+    _install_fake_gateway_run(monkeypatch, fake_start_gateway)
+    monkeypatch.setattr(gateway, "is_windows", lambda: True)
+    monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
+    monkeypatch.setattr(gateway.sys, "stdin", _TTY())
+    monkeypatch.setenv("HERMES_GATEWAY_DETACHED", "1")
+    monkeypatch.setattr(gateway.signal, "signal", fake_signal)
+    monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)
+
+    gateway.run_gateway()
+
+    assert calls == [(False, 0)]
+    assert (gateway.signal.SIGINT, gateway.signal.SIG_IGN) in signal_calls
+
+
 class TestSystemdLingerStatus:
     def test_reports_enabled(self, monkeypatch):
         monkeypatch.setattr(gateway, "is_linux", lambda: True)
@@ -344,6 +419,15 @@ def test_find_gateway_pids_falls_back_to_pid_file_when_process_scan_fails(monkey
     monkeypatch.setattr(gateway, "is_windows", lambda: False)
     monkeypatch.setattr("gateway.status.get_running_pid", lambda: 321)
 
+    # /proc walk is the first path tried (#22693). Force os.listdir on /proc
+    # to raise so the function falls back to ps, where fake_run takes over.
+    _real_listdir = gateway.os.listdir
+    def _no_proc_listdir(path):
+        if path == "/proc":
+            raise OSError("test stub: /proc unavailable")
+        return _real_listdir(path)
+    monkeypatch.setattr(gateway.os, "listdir", _no_proc_listdir)
+
     def fake_run(cmd, **kwargs):
         if cmd[:4] == ["ps", "-A", "eww", "-o"]:
             return SimpleNamespace(returncode=1, stdout="", stderr="ps failed")

tests/hermes_cli/test_gateway_proc_fallback.py

@@ -0,0 +1,138 @@
+"""Tests for /proc-based gateway PID detection in Docker environments.
+
+Verifies that _scan_gateway_pids() uses /proc/*/cmdline when available
+(Docker without procps) and falls back to ps only when /proc is absent.
+
+See: NousResearch/hermes-agent#7622
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+import hermes_cli.gateway as gateway_mod
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_GATEWAY_CMD = "python -m hermes_cli.main gateway run"
+_OTHER_CMD = "python -m some_other_thing"
+
+
+def _fake_proc_dir(entries: dict):
+    """Return side_effects that simulate /proc: isdir → True, listdir → pids,
+    open(cmdline) → null-delimited command bytes."""
+    def _isdir(path):
+        return str(path) == "/proc"
+
+    def _listdir(path):
+        if str(path) == "/proc":
+            return [str(pid) for pid in entries] + ["self", "version"]
+        raise FileNotFoundError(path)
+
+    def _open(path, mode="r", **kwargs):
+        path_str = str(path)
+        if "/cmdline" in path_str:
+            pid = int(path_str.split("/proc/")[1].split("/")[0])
+            raw = entries.get(pid, "").encode("utf-8").replace(b" ", b"\x00")
+            m = MagicMock()
+            m.read.return_value = raw
+            m.__enter__ = lambda s: s
+            m.__exit__ = MagicMock(return_value=False)
+            return m
+        raise FileNotFoundError(path)
+
+    return _isdir, _listdir, _open
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+
+class TestProcFallback:
+    """_scan_gateway_pids reads /proc when available, skips ps."""
+
+    def test_detects_gateway_pid_via_proc(self):
+        my_pid = os.getpid()
+        entries = {
+            my_pid: "python -m hermes_cli.main",   # own process — excluded
+            12345: _GATEWAY_CMD,
+            99999: _OTHER_CMD,
+        }
+        _isdir, _listdir, _open = _fake_proc_dir(entries)
+
+        with (
+            patch("hermes_cli.gateway.is_windows", return_value=False),
+            patch("os.path.isdir", side_effect=_isdir),
+            patch("os.listdir", side_effect=_listdir),
+            patch("builtins.open", side_effect=_open),
+            patch("hermes_cli.gateway._get_ancestor_pids", return_value=set()),
+            patch("subprocess.run") as mock_ps,
+        ):
+            pids = gateway_mod._scan_gateway_pids(set(), all_profiles=True)
+
+        assert 12345 in pids
+        assert 99999 not in pids
+        mock_ps.assert_not_called()  # ps must NOT be called when /proc worked
+
+    def test_excludes_own_pid_from_proc_scan(self):
+        my_pid = os.getpid()
+        entries = {my_pid: _GATEWAY_CMD}
+        _isdir, _listdir, _open = _fake_proc_dir(entries)
+
+        with (
+            patch("hermes_cli.gateway.is_windows", return_value=False),
+            patch("os.path.isdir", side_effect=_isdir),
+            patch("os.listdir", side_effect=_listdir),
+            patch("builtins.open", side_effect=_open),
+            patch("hermes_cli.gateway._get_ancestor_pids", return_value=set()),
+            patch("subprocess.run"),
+        ):
+            pids = gateway_mod._scan_gateway_pids(set(), all_profiles=True)
+
+        assert my_pid not in pids
+
+    def test_falls_back_to_ps_when_proc_absent(self):
+        ps_output = f"12345 {_GATEWAY_CMD}\n99999 {_OTHER_CMD}\n"
+        mock_result = MagicMock()
+        mock_result.returncode = 0
+        mock_result.stdout = ps_output
+
+        with (
+            patch("hermes_cli.gateway.is_windows", return_value=False),
+            patch("os.path.isdir", return_value=False),
+            patch("hermes_cli.gateway._get_ancestor_pids", return_value=set()),
+            patch("subprocess.run", return_value=mock_result) as mock_ps,
+        ):
+            pids = gateway_mod._scan_gateway_pids(set(), all_profiles=True)
+
+        mock_ps.assert_called_once()
+        assert 12345 in pids
+
+    def test_proc_permission_error_skips_pid(self):
+        def _isdir(path):
+            return str(path) == "/proc"
+
+        def _listdir(path):
+            if str(path) == "/proc":
+                return ["12345", "self"]
+            raise FileNotFoundError
+
+        def _open(path, mode="r", **kwargs):
+            raise PermissionError("no access")
+
+        with (
+            patch("hermes_cli.gateway.is_windows", return_value=False),
+            patch("os.path.isdir", side_effect=_isdir),
+            patch("os.listdir", side_effect=_listdir),
+            patch("builtins.open", side_effect=_open),
+            patch("hermes_cli.gateway._get_ancestor_pids", return_value=set()),
+            patch("subprocess.run") as mock_ps,
+        ):
+            pids = gateway_mod._scan_gateway_pids(set(), all_profiles=True)
+
+        # PermissionError swallowed — empty result, no crash
+        assert 12345 not in pids
+        mock_ps.assert_not_called()  # /proc dir existed, so ps not called

tests/hermes_cli/test_gateway_service.py

@@ -1,13 +1,14 @@
 """Tests for gateway service management helpers."""
 
 import os
-import pwd
 import subprocess
 from pathlib import Path
 from types import SimpleNamespace
 
 import pytest
 
+pwd = pytest.importorskip("pwd")
+
 import hermes_cli.gateway as gateway_cli
 from gateway import status
 from gateway.restart import (
@@ -233,6 +234,60 @@ class TestSystemdServiceRefresh:
         assert unit_path.read_text(encoding="utf-8") == "new unit\n"
         assert ["systemctl", "--user", "daemon-reload"] in calls
 
+    def test_refresh_refuses_to_bake_pytest_tmpdir_into_real_user_unit(
+        self, tmp_path, monkeypatch
+    ):
+        """Defense in depth: ``refresh_systemd_unit_if_needed()`` runs every
+        time ``run_gateway()`` starts. The user-scope unit path resolves
+        under ``Path.home()`` (NOT sandboxed by conftest), and
+        ``generate_systemd_unit()`` bakes ``HERMES_HOME`` into the unit's
+        ``Environment=`` line. Without this guard, any test that drives
+        ``run_gateway()`` end-to-end on a real Linux dev box silently
+        rewrites the developer's installed gateway unit with a
+        ``/tmp/pytest-of-.../hermes_test`` HERMES_HOME — silently breaking
+        their gateway on the next boot. The guard sniffs the generated
+        unit body for tmpdir markers and refuses the write. Tests that
+        legitimately exercise the refresh flow patch
+        ``generate_systemd_unit`` to return synthetic content that doesn't
+        carry those markers.
+        """
+        unit_path = tmp_path / "hermes-gateway.service"
+        unit_path.write_text("old unit\n", encoding="utf-8")
+
+        monkeypatch.setattr(
+            gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
+        )
+        # Realistic generated unit referencing a pytest tmpdir HERMES_HOME
+        polluted_unit = (
+            "[Service]\n"
+            'Environment="HERMES_HOME=/tmp/pytest-of-alice/pytest-42/'
+            'popen-gw0/test_x/hermes_test"\n'
+        )
+        monkeypatch.setattr(
+            gateway_cli,
+            "generate_systemd_unit",
+            lambda system=False, run_as_user=None: polluted_unit,
+        )
+
+        # If the guard fails, daemon-reload would be called — record it.
+        ran = []
+
+        def fake_run(cmd, check=True, **kwargs):
+            ran.append(cmd)
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+        monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
+
+        result = gateway_cli.refresh_systemd_unit_if_needed(system=False)
+
+        assert result is False, "refresh should refuse to write a polluted unit"
+        assert (
+            unit_path.read_text(encoding="utf-8") == "old unit\n"
+        ), "installed unit must be left untouched"
+        assert not any(
+            "daemon-reload" in str(c) for c in ran
+        ), "daemon-reload must not run when write was refused"
+
 
 class TestRequireServiceInstalled:
     def test_exits_with_install_hint_when_unit_missing(self, tmp_path, monkeypatch, capsys):
@@ -1284,20 +1339,17 @@ class TestSystemServiceIdentityRootHandling:
 
     def test_auto_detected_root_is_rejected(self, monkeypatch):
         """When root is auto-detected (not explicitly requested), raise."""
-        import pwd
         import grp
 
         monkeypatch.delenv("SUDO_USER", raising=False)
         monkeypatch.setenv("USER", "root")
         monkeypatch.setenv("LOGNAME", "root")
 
-        import pytest
         with pytest.raises(ValueError, match="pass --run-as-user root to override"):
             gateway_cli._system_service_identity(run_as_user=None)
 
     def test_explicit_root_is_allowed(self, monkeypatch):
         """When root is explicitly passed via --run-as-user root, allow it."""
-        import pwd
         import grp
 
         root_info = pwd.getpwnam("root")
@@ -1309,7 +1361,6 @@ class TestSystemServiceIdentityRootHandling:
 
     def test_non_root_user_passes_through(self, monkeypatch):
         """Normal non-root user works as before."""
-        import pwd
         import grp
 
         monkeypatch.delenv("SUDO_USER", raising=False)

tests/hermes_cli/test_kanban_cli.py

@@ -331,13 +331,64 @@ def test_run_slash_specify_end_to_end(kanban_home, monkeypatch):
 
 
 def test_run_slash_specify_help_is_reachable(kanban_home):
-    """`--help` on a subcommand is handled by argparse itself — it prints
-    to the process stdout and raises SystemExit before run_slash's output
-    redirection is installed, so the returned string is the usage-error
-    sentinel. All we're asserting here is that the subcommand is
-    registered (no "unknown action" error) — the shape of the help text
-    is covered by the direct argparse tests in test_kanban_specify.py."""
+    """`-h`/`--help` on a subcommand returns the actual help text — see
+    issue #21794. argparse writes help to stdout and exits 0; run_slash
+    must capture both streams and treat exit 0 as success, not error."""
     out = kc.run_slash("specify --help")
-    # Either the usage-error sentinel (stdout swallowed by argparse) or
-    # a real help rendering — both mean the subcommand exists.
-    assert "usage error" in out.lower() or "specify" in out.lower()
+    assert "specify" in out.lower()
+    # Help dump should NOT come back wrapped as a usage error.
+    assert not out.startswith("⚠")
+
+
+# ---------------------------------------------------------------------------
+# /kanban help / no-args / unknown-action UX (issue #21794)
+# ---------------------------------------------------------------------------
+
+def test_run_slash_bare_returns_curated_help(kanban_home):
+    """Bare `/kanban` returns the curated short-help block — not a 5KB
+    argparse usage dump."""
+    out = kc.run_slash("")
+    assert "/kanban" in out
+    assert "list" in out
+    assert "show" in out
+    # Sanity: should be a chat-friendly size, not the raw usage tree.
+    assert len(out) < 2000
+    # Shouldn't surface argparse's usage-error sentinel.
+    assert "usage error" not in out.lower()
+
+
+@pytest.mark.parametrize("alias", ["help", "--help", "-h", "?"])
+def test_run_slash_help_aliases_match_bare(kanban_home, alias):
+    """Every documented help alias produces the same curated output."""
+    bare = kc.run_slash("")
+    out = kc.run_slash(alias)
+    assert out == bare
+
+
+def test_run_slash_subcommand_help_returns_help_text(kanban_home):
+    """`/kanban show -h` returns the actual subcommand help, not a
+    fake `(usage error: 0)` sentinel."""
+    out = kc.run_slash("show -h")
+    assert "task_id" in out
+    assert "/kanban show" in out
+    assert not out.startswith("⚠")
+
+
+def test_run_slash_unknown_action_friendly_error(kanban_home):
+    """Unknown subcommand surfaces a single-line usage error prefixed
+    with our marker — no `(usage error: 2)` wrapping, no doubled
+    `kanban kanban` prog string."""
+    out = kc.run_slash("frobnicate")
+    assert "/kanban" in out
+    assert "frobnicate" in out
+    assert "/kanban-wrap" not in out
+    assert "/kanban kanban" not in out
+    assert "(usage error: " not in out
+
+
+def test_run_slash_missing_required_arg_friendly_error(kanban_home):
+    """Missing positional argument shows the subcommand-scoped usage
+    line, not the top-level kanban tree."""
+    out = kc.run_slash("show")
+    assert "/kanban show" in out
+    assert "task_id" in out

tests/hermes_cli/test_kanban_core_functionality.py

@@ -2507,6 +2507,27 @@ def test_build_worker_context_caps_prior_attempts(kanban_home):
         conn.close()
 
 
+def test_build_worker_context_renders_author_with_safe_framing(kanban_home):
+    """Author rendering wraps the operator-controlled author in code fences
+    + "comment from worker" prefix so a misleading HERMES_PROFILE name
+    (e.g. "hermes-system", "operator") can't be misread as a system
+    directive above the comment body. Defense-in-depth — see #22452."""
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="t", assignee="worker")
+        kb.add_comment(conn, tid, author="hermes-system", body="some note")
+        ctx = kb.build_worker_context(conn, tid)
+
+        # No bold-author rendering anywhere in the context.
+        assert "**hermes-system**" not in ctx
+        # Explicit provenance prefix is present.
+        assert "comment from worker `hermes-system` at " in ctx
+        # The body still renders.
+        assert "some note" in ctx
+    finally:
+        conn.close()
+
+
 def test_build_worker_context_caps_comments(kanban_home):
     """Same cap for comments — comment-storm tasks stay bounded."""
     conn = kb.connect()
@@ -2516,10 +2537,15 @@ def test_build_worker_context_caps_comments(kanban_home):
             kb.add_comment(conn, tid, author=f"u{i % 3}", body=f"comment {i}")
         ctx = kb.build_worker_context(conn, tid)
         # Only _CTX_MAX_COMMENTS most-recent shown in full
-        comment_count = ctx.count("**u")
-        # 3 distinct authors u0/u1/u2 so the count is trickier; use the
-        # "comment N" body text to count.
-        body_count = sum(1 for line in ctx.splitlines() if line.startswith("comment "))
+        # Count by body text since author rendering uses code-fenced
+        # "comment from worker `<author>` at <ts>:" framing (#22452).
+        # Comment bodies are "comment 0".."comment 99" so we need to
+        # match the body specifically (digit suffix), not the author
+        # provenance line (which also starts with "comment ").
+        import re
+        body_count = sum(
+            1 for line in ctx.splitlines() if re.fullmatch(r"comment \d+", line)
+        )
         assert body_count == kb._CTX_MAX_COMMENTS, (
             f"expected {kb._CTX_MAX_COMMENTS} comments shown, got {body_count}"
         )

tests/hermes_cli/test_kanban_db.py

@@ -298,6 +298,122 @@ def test_block_then_unblock(kanban_home):
         assert kb.get_task(conn, t).status == "ready"
 
 
+# ---------------------------------------------------------------------------
+# Parent-completion invariant at the claim gate (RCA t_a6acd07d)
+# ---------------------------------------------------------------------------
+
+def test_claim_rejects_when_parents_not_done(kanban_home):
+    """claim_task must refuse ready->running if any parent isn't 'done'.
+
+    Simulates the create-then-link race: a task gets status='ready' via a
+    racy writer while it still has undone parents. The claim gate must
+    detect the violation, demote the child back to 'todo', append a
+    'claim_rejected' event, and return None. Covers Fix 1 of the RCA.
+    """
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="a")
+        child = kb.create_task(
+            conn, title="child", assignee="a", parents=[parent],
+        )
+        # Child correctly starts 'todo' because parent is not 'done'.
+        assert kb.get_task(conn, child).status == "todo"
+        # Simulate the race: a racy writer force-promotes the child to
+        # 'ready' while parent is still pending.
+        conn.execute(
+            "UPDATE tasks SET status='ready' WHERE id=?", (child,),
+        )
+        conn.commit()
+        assert kb.get_task(conn, child).status == "ready"
+
+        result = kb.claim_task(conn, child, claimer="host:1")
+
+    assert result is None
+    with kb.connect() as conn:
+        assert kb.get_task(conn, child).status == "todo"
+        events = conn.execute(
+            "SELECT kind, payload FROM task_events "
+            "WHERE task_id = ? ORDER BY id",
+            (child,),
+        ).fetchall()
+    kinds = [e["kind"] for e in events]
+    assert "claim_rejected" in kinds
+    # No 'claimed' event was emitted for the blocked attempt.
+    assert "claimed" not in kinds
+
+
+def test_claim_succeeds_once_parents_done(kanban_home):
+    """After parents complete, recompute_ready -> claim_task must succeed."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="a")
+        child = kb.create_task(
+            conn, title="child", assignee="a", parents=[parent],
+        )
+        kb.claim_task(conn, parent)
+        assert kb.complete_task(conn, parent, result="ok")
+        kb.recompute_ready(conn)
+        assert kb.get_task(conn, child).status == "ready"
+        claimed = kb.claim_task(conn, child, claimer="host:1")
+    assert claimed is not None
+    assert claimed.status == "running"
+
+
+def test_create_with_parents_stays_todo_until_parents_done(kanban_home):
+    """kanban_create(parents=[...]) must land in 'todo' and only promote on parent done."""
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="a")
+        child = kb.create_task(
+            conn, title="child", assignee="a", parents=[parent],
+        )
+        assert kb.get_task(conn, child).status == "todo"
+        # Dispatcher tick between create and some later event must NOT
+        # produce a winner for this child.
+        promoted = kb.recompute_ready(conn)
+        assert promoted == 0
+        assert kb.get_task(conn, child).status == "todo"
+        # Complete parent; complete_task internally runs recompute_ready,
+        # which promotes the child to 'ready'.
+        kb.claim_task(conn, parent)
+        kb.complete_task(conn, parent, result="ok")
+        assert kb.get_task(conn, child).status == "ready"
+
+
+def test_unblock_with_pending_parents_goes_to_todo(kanban_home):
+    """unblock_task must re-gate on parent completion (Fix 3).
+
+    A task blocked while parents are still in progress must return to
+    'todo' (not 'ready') on unblock. Otherwise the dispatcher will claim
+    it immediately, repeating Bug 2 from the RCA.
+    """
+    with kb.connect() as conn:
+        parent = kb.create_task(conn, title="parent", assignee="a")
+        child = kb.create_task(
+            conn, title="child", assignee="a", parents=[parent],
+        )
+        # Force child into 'blocked' regardless of parent progress
+        # (simulates a worker that self-blocked, or an operator block).
+        conn.execute(
+            "UPDATE tasks SET status='blocked' WHERE id=?", (child,),
+        )
+        conn.commit()
+        assert kb.unblock_task(conn, child)
+        assert kb.get_task(conn, child).status == "todo"
+        # After parent completes + recompute, the child is ready.
+        kb.claim_task(conn, parent)
+        kb.complete_task(conn, parent, result="ok")
+        kb.recompute_ready(conn)
+        assert kb.get_task(conn, child).status == "ready"
+
+
+def test_unblock_without_parents_goes_to_ready(kanban_home):
+    """Parent-free unblock still produces 'ready' (behavior preserved)."""
+    with kb.connect() as conn:
+        t = kb.create_task(conn, title="lone", assignee="a")
+        kb.claim_task(conn, t)
+        assert kb.block_task(conn, t, reason="need input")
+        assert kb.unblock_task(conn, t)
+        assert kb.get_task(conn, t).status == "ready"
+
+
 def test_assign_refuses_while_running(kanban_home):
     with kb.connect() as conn:
         t = kb.create_task(conn, title="x", assignee="a")
@@ -966,3 +1082,120 @@ def test_connect_falls_back_to_delete_on_locking_protocol(kanban_home, caplog):
     tasks = kb.list_tasks(conn)
     assert any(row.id == t for row in tasks)
     conn.close()
+
+
+def test_unlink_tasks_triggers_recompute_ready(kanban_home):
+    """Regression test for issue #22459.
+
+    Removing a dependency via unlink_tasks must immediately promote the child
+    to ready when all remaining parents are done — same contract as
+    complete_task and unblock_task.
+
+    Before the fix, child stayed 'todo' indefinitely after unlink; only the
+    next dispatcher tick or a manual 'hermes kanban recompute' would promote it.
+    """
+    with kb.connect() as conn:
+        # A is done.
+        a = kb.create_task(conn, title="parent-done")
+        kb.complete_task(conn, a)
+
+        # C is running (not done) — blocks child B.
+        c = kb.create_task(conn, title="parent-running")
+        kb.claim_task(conn, c, claimer="worker:1")
+
+        # B depends on both A (done) and C (running) → stays todo.
+        b = kb.create_task(conn, title="child", parents=[a, c])
+        assert kb.get_task(conn, b).status == "todo"
+
+        # Remove the blocking dependency C → B.
+        removed = kb.unlink_tasks(conn, c, b)
+        assert removed is True
+
+        # B's only remaining parent is A (done) → must be ready immediately.
+        assert kb.get_task(conn, b).status == "ready", (
+            "child should promote to ready immediately after unlink_tasks "
+            "removes its last blocking dependency"
+        )
+# ---------------------------------------------------------------------------
+# _add_column_if_missing / _migrate_add_optional_columns idempotency (#21708)
+# ---------------------------------------------------------------------------
+
+def test_add_column_if_missing_is_idempotent_on_race(kanban_home):
+    """``_add_column_if_missing`` must swallow 'duplicate column name' errors.
+
+    Regression for #21708: the kanban dispatcher opens the DB twice per tick
+    (once via _tick_once_for_board, once via init_db's discard-and-reconnect
+    path).  A second concurrent connection runs _migrate_add_optional_columns
+    before the first one commits, so ALTER TABLE raises OperationalError with
+    'duplicate column name: consecutive_failures'.  Without the idempotency
+    guard that crashes the dispatcher on the first tick after every restart.
+    """
+    import sqlite3
+
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        "CREATE TABLE tasks (id INTEGER PRIMARY KEY, title TEXT NOT NULL)"
+    )
+
+    # First call adds the column — returns True.
+    added = kb._add_column_if_missing(conn, "tasks", "extra_col", "extra_col TEXT")
+    assert added is True
+    cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")}
+    assert "extra_col" in cols
+
+    # Second call on same connection — column already exists — must return
+    # False without raising, simulating the race the dispatcher hits.
+    added_again = kb._add_column_if_missing(
+        conn, "tasks", "extra_col", "extra_col TEXT"
+    )
+    assert added_again is False
+
+    conn.close()
+
+
+def test_migrate_add_optional_columns_tolerates_concurrent_migration(kanban_home):
+    """Full _migrate_add_optional_columns must not raise when columns already
+    exist (issue #21708 race window — two connections migrate concurrently)."""
+    import sqlite3
+
+    # Schema already in fully-migrated state (all optional columns present).
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    conn.execute(
+        """
+        CREATE TABLE tasks (
+            id INTEGER PRIMARY KEY,
+            title TEXT NOT NULL,
+            tenant TEXT,
+            result TEXT,
+            idempotency_key TEXT,
+            consecutive_failures INTEGER NOT NULL DEFAULT 0,
+            worker_pid INTEGER,
+            last_failure_error TEXT,
+            max_runtime_seconds INTEGER,
+            last_heartbeat_at INTEGER,
+            current_run_id INTEGER,
+            workflow_template_id TEXT,
+            current_step_key TEXT,
+            skills TEXT,
+            max_retries INTEGER
+        )
+        """
+    )
+    conn.execute(
+        """
+        CREATE TABLE task_events (
+            id         INTEGER PRIMARY KEY AUTOINCREMENT,
+            task_id    TEXT NOT NULL DEFAULT '',
+            run_id     INTEGER,
+            kind       TEXT NOT NULL DEFAULT '',
+            payload    TEXT,
+            created_at INTEGER NOT NULL DEFAULT 0
+        )
+        """
+    )
+
+    # Running migration on an already-migrated schema must not raise.
+    kb._migrate_add_optional_columns(conn)
+    conn.close()

tests/hermes_cli/test_kanban_notify.py

@@ -0,0 +1,303 @@
+import asyncio
+import pytest
+
+from pathlib import Path
+from hermes_cli import kanban_db as kb
+from unittest.mock import AsyncMock, MagicMock, patch
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def kanban_home(tmp_path, monkeypatch):
+    home = tmp_path / ".hermes"
+    home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    kb.init_db()
+    return home
+
+
+@pytest.mark.asyncio
+async def test_notifier_unsubs_after_completed_event(kanban_home):
+    """
+    Subscription should be remove after completed event
+    """
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="test task", assignee="worker1")
+        kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1")
+        kb.complete_task(conn, tid, result="completed by agent")
+    finally:
+        conn.close()
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    fake_adapter = MagicMock()
+
+    async def _send_and_stop(chat_id, msg, metadata=None):
+        runner._running = False
+
+    fake_adapter.send = AsyncMock(side_effect=_send_and_stop)
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+
+    async def _fast_sleep(_):
+        await _orig_sleep(0)
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    fake_adapter.send.assert_called_once()
+    call_msg = fake_adapter.send.call_args[0][1]
+    assert "completed" in call_msg
+
+    conn = kb.connect()
+    try:
+        subs = kb.list_notify_subs(conn, tid)
+    finally:
+        conn.close()
+    assert subs == [], "Subscription should be unsub after completed event"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('kind', ["gave_up", "crashed", "timed_out"])
+async def test_notifier_unsubs_after_abnormal_events(kind, kanban_home):
+    """
+    Event kind of gave_up, crashed, time_out would be cover, and remove subscription
+    """
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+
+    conn = kb.connect()
+
+    try:
+        tid = kb.create_task(conn, title=f"test {kind} task", assignee="worker1")
+        kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1")
+        kb._append_event(conn, tid, kind=kind)
+    finally:
+        conn.close()
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    fake_adapter = MagicMock()
+
+    async def _send_and_stop(chat_id, msg, metadata=None):
+        runner._running = False
+
+    fake_adapter.send = AsyncMock(side_effect=_send_and_stop)
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+
+    async def _fast_sleep(_):
+        await _orig_sleep(0)
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    fake_adapter.send.assert_called_once()
+    assert kind.replace('_', ' ') in fake_adapter.send.call_args[0][1]
+
+    conn = kb.connect()
+    try:
+        subs = kb.list_notify_subs(conn, tid)
+    finally:
+        conn.close()
+    assert subs == [], "Subscription should be unsub after abnormal crash"
+
+
+@pytest.mark.asyncio
+async def test_notifier_second_blocked_delivers(kanban_home):
+    """
+    After the first blocked, should receive second blocked notification.
+    """
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    delivered_msgs: list[str] = []
+
+    async def _capture_send(chat_id, msg, metadata=None):
+        delivered_msgs.append(msg)
+
+    fake_adapter = MagicMock()
+    fake_adapter.send = AsyncMock(side_effect=_capture_send)
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+    tick_count = 0
+
+    async def _fast_sleep(_):
+        nonlocal tick_count
+        await _orig_sleep(0)
+        tick_count += 1
+        if tick_count >= 6:
+            runner._running = False
+
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="test task", assignee="worker1")
+        kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1")
+
+        # Cycle 1: blocked
+        kb.block_task(conn, tid, reason="first block")
+    finally:
+        conn.close()
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    # Cycle 2: unblock → block run again
+    runner._running = True
+    tick_count = 0
+
+    conn = kb.connect()
+    try:
+        kb.unblock_task(conn, tid)
+        kb.block_task(conn, tid, reason="second block")
+    finally:
+        conn.close()
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    blocked_deliveries = [m for m in delivered_msgs if "blocked" in m]
+    assert "second block" not in blocked_deliveries[0]
+    assert "second block" in blocked_deliveries[1]
+    assert len(blocked_deliveries) == 2, (
+        f"Should receive 2 blocked notification, but only get {len(blocked_deliveries)} count\n"
+        f"Message {delivered_msgs}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Regression: gateway watchers must not double-init the kanban DB.
+#
+# Both the notifier watcher (`_kanban_notifier_watcher`) and the dispatcher
+# tick (`_tick_once_for_board`) used to call `_kb.connect(board=slug)`
+# immediately followed by `_kb.init_db(board=slug)`. Since `connect()`
+# already runs the schema + idempotent migration on first open per process,
+# the explicit `init_db()` was redundant — and worse, `init_db()`
+# deliberately busts the per-process cache and re-runs the migration on a
+# *second* connection, which races the first.  On legacy DBs this surfaced
+# as `duplicate column name: <col>` (now tolerated by
+# `_add_column_if_missing`) and intermittent `database is locked` errors
+# (issue #21378).
+#
+# The fix removes the `init_db()` calls in both watchers; this regression
+# test pins that behaviour so we don't reintroduce them.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_notifier_does_not_call_init_db(kanban_home):
+    """Notifier watcher path must not invoke `_kb.init_db` (issue #21378)."""
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    fake_adapter = MagicMock()
+    fake_adapter.send = AsyncMock()
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+    tick_count = 0
+
+    async def _fast_sleep(_):
+        nonlocal tick_count
+        await _orig_sleep(0)
+        tick_count += 1
+        if tick_count >= 3:
+            runner._running = False
+
+    init_db_calls: list[object] = []
+    real_init_db = kb.init_db
+
+    def _spy_init_db(*args, **kwargs):
+        init_db_calls.append((args, kwargs))
+        return real_init_db(*args, **kwargs)
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep), \
+         patch("hermes_cli.kanban_db.init_db", side_effect=_spy_init_db):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    assert init_db_calls == [], (
+        "_kanban_notifier_watcher must not call init_db on every tick — "
+        "connect() handles first-run schema init. "
+        "Reintroducing init_db revives issue #21378. "
+        f"Got {len(init_db_calls)} call(s): {init_db_calls}"
+    )
+
+
+def test_dispatcher_tick_does_not_call_init_db(kanban_home, monkeypatch):
+    """`_tick_once_for_board` must not invoke `_kb.init_db` (issue #21378).
+
+    `connect()` already runs the schema + idempotent migration on first open
+    per process. The explicit `init_db()` call was redundant and triggered a
+    second migration on a second connection that raced the first.
+    """
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from unittest.mock import patch
+
+    runner = object.__new__(GatewayRunner)
+
+    init_db_calls: list[object] = []
+    real_init_db = kb.init_db
+
+    def _spy_init_db(*args, **kwargs):
+        init_db_calls.append((args, kwargs))
+        return real_init_db(*args, **kwargs)
+
+    # The dispatcher watcher's tick lives as a local closure inside
+    # `_kanban_dispatcher_watcher`. Read the source and assert the
+    # specific patterns that would reintroduce the bug are absent.
+    import inspect
+    src = inspect.getsource(GatewayRunner._kanban_dispatcher_watcher)
+    assert "_kb.init_db(board=slug)" not in src, (
+        "_kanban_dispatcher_watcher must not call _kb.init_db(board=slug) — "
+        "see issue #21378. Use connect() alone; it runs migrations on first "
+        "open per process."
+    )
+
+    notifier_src = inspect.getsource(GatewayRunner._kanban_notifier_watcher)
+    assert "_kb.init_db(board=slug)" not in notifier_src, (
+        "_kanban_notifier_watcher must not call _kb.init_db(board=slug) — "
+        "see issue #21378."
+    )

tests/hermes_cli/test_openai_codex_model_validation_fallback.py

@@ -1,9 +1,18 @@
 """Regression tests for OpenAI Codex model validation when the listing lags behind
 actually usable backend model IDs.
 
-The bug: `/model` and `switch_model()` reject `gpt-5.3-codex-spark` because the
-OpenAI Codex listing omits it, even though direct runtime calls with
-`--provider openai-codex -m gpt-5.3-codex-spark` succeed.
+The bug originally reported in #16172: `/model` and `switch_model()` rejected
+`gpt-5.3-codex-spark` because the curated listing omitted it, even though direct
+runtime calls succeeded. PR #19729 fixed this by soft-accepting unknown-but-
+plausible Codex slugs with a warning, and this test pins the soft-accept
+behavior so it doesn't regress.
+
+Note: gpt-5.3-codex-spark itself is now in the curated catalog (PR #22991),
+so the real-world Spark request takes the `recognized=True` fast path. This
+test still uses Spark as the example slug but explicitly mocks
+``provider_model_ids`` to omit it, exercising the soft-accept path generically
+for any future entitlement-gated Codex slug that ships before Hermes catalogs
+it.
 """
 
 from unittest.mock import patch

tests/hermes_cli/test_plugins.py

@@ -1232,3 +1232,77 @@ class TestPluginDispatchTool:
             result = ctx.dispatch_tool("fake", {})
 
         assert '"error"' in result
+
+
+class TestPluginDebugLogging:
+    """HERMES_PLUGINS_DEBUG opt-in stderr handler for plugin developers."""
+
+    def test_debug_handler_not_installed_when_env_var_absent(self, monkeypatch):
+        """Without the env var, no stderr handler is attached."""
+        monkeypatch.delenv("HERMES_PLUGINS_DEBUG", raising=False)
+        from hermes_cli import plugins as plugins_mod
+
+        # Snapshot, then force a re-evaluation.
+        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
+        original_debug = plugins_mod._PLUGINS_DEBUG
+        original_handlers = list(plugins_mod.logger.handlers)
+        try:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = False
+            plugins_mod._install_plugin_debug_handler(force=True)
+            assert plugins_mod._PLUGINS_DEBUG is False
+            assert plugins_mod._DEBUG_HANDLER_INSTALLED is False
+            # No new stderr handler was attached.
+            assert plugins_mod.logger.handlers == original_handlers
+        finally:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
+            plugins_mod._PLUGINS_DEBUG = original_debug
+            plugins_mod.logger.handlers = original_handlers
+
+    def test_debug_handler_installed_when_env_var_set(self, monkeypatch):
+        """With HERMES_PLUGINS_DEBUG=1, a DEBUG-level stderr handler is attached."""
+        monkeypatch.setenv("HERMES_PLUGINS_DEBUG", "1")
+        from hermes_cli import plugins as plugins_mod
+
+        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
+        original_debug = plugins_mod._PLUGINS_DEBUG
+        original_level = plugins_mod.logger.level
+        original_handlers = list(plugins_mod.logger.handlers)
+        try:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = False
+            plugins_mod._install_plugin_debug_handler(force=True)
+            assert plugins_mod._PLUGINS_DEBUG is True
+            assert plugins_mod._DEBUG_HANDLER_INSTALLED is True
+            assert plugins_mod.logger.level == logging.DEBUG
+            new_handlers = [
+                h for h in plugins_mod.logger.handlers if h not in original_handlers
+            ]
+            assert len(new_handlers) == 1
+            assert isinstance(new_handlers[0], logging.StreamHandler)
+            assert new_handlers[0].level == logging.DEBUG
+        finally:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
+            plugins_mod._PLUGINS_DEBUG = original_debug
+            plugins_mod.logger.setLevel(original_level)
+            plugins_mod.logger.handlers = original_handlers
+
+    def test_debug_handler_idempotent(self, monkeypatch):
+        """Calling install twice (without force) does not double-attach."""
+        monkeypatch.setenv("HERMES_PLUGINS_DEBUG", "1")
+        from hermes_cli import plugins as plugins_mod
+
+        original_installed = plugins_mod._DEBUG_HANDLER_INSTALLED
+        original_debug = plugins_mod._PLUGINS_DEBUG
+        original_level = plugins_mod.logger.level
+        original_handlers = list(plugins_mod.logger.handlers)
+        try:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = False
+            plugins_mod._install_plugin_debug_handler(force=True)
+            count_after_first = len(plugins_mod.logger.handlers)
+            plugins_mod._install_plugin_debug_handler()  # no force
+            count_after_second = len(plugins_mod.logger.handlers)
+            assert count_after_first == count_after_second
+        finally:
+            plugins_mod._DEBUG_HANDLER_INSTALLED = original_installed
+            plugins_mod._PLUGINS_DEBUG = original_debug
+            plugins_mod.logger.setLevel(original_level)
+            plugins_mod.logger.handlers = original_handlers

tests/hermes_cli/test_plugins_cmd.py

@@ -12,9 +12,11 @@ import pytest
 import yaml
 
 from hermes_cli.plugins_cmd import (
+    PluginOperationError,
     _copy_example_files,
     _read_manifest,
     _repo_name_from_url,
+    _resolve_git_executable,
     _resolve_git_url,
     _sanitize_plugin_name,
     plugins_command,
@@ -99,6 +101,69 @@ class TestResolveGitUrl:
             _resolve_git_url("a/b/c")
 
 
+# ── _resolve_git_executable ─────────────────────────────────────────────────
+
+
+class TestResolveGitExecutable:
+    """Fallback resolution when bare ``git`` is not discoverable via ``PATH``."""
+
+    def teardown_method(self):
+        _resolve_git_executable.cache_clear()
+
+    def test_prefers_shutil_which(self):
+        import hermes_cli.plugins_cmd as pc
+
+        _resolve_git_executable.cache_clear()
+        with patch.object(pc.shutil, "which", return_value="/usr/local/bin/git"):
+            assert pc._resolve_git_executable() == "/usr/local/bin/git"
+
+    def test_fallback_posix_first_matching_path(self):
+        import hermes_cli.plugins_cmd as pc
+
+        _resolve_git_executable.cache_clear()
+
+        def _isfile(p: str) -> bool:
+            return p == "/usr/local/bin/git"
+
+        with patch.object(pc.shutil, "which", return_value=None):
+            with patch.object(pc.os, "name", "posix"):
+                with patch.object(pc.os.path, "isfile", side_effect=_isfile):
+                    assert pc._resolve_git_executable() == "/usr/local/bin/git"
+
+    def test_returns_none_when_unavailable(self):
+        import hermes_cli.plugins_cmd as pc
+
+        _resolve_git_executable.cache_clear()
+        with patch.object(pc.shutil, "which", return_value=None):
+            with patch.object(pc.os, "name", "posix"):
+                with patch.object(pc.os.path, "isfile", return_value=False):
+                    assert pc._resolve_git_executable() is None
+
+    def test_git_pull_uses_resolved_executable(self, tmp_path):
+        import hermes_cli.plugins_cmd as pc
+
+        _resolve_git_executable.cache_clear()
+        with patch.object(
+            pc,
+            "_resolve_git_executable",
+            return_value="/resolved/git",
+        ):
+            with patch.object(pc.subprocess, "run") as run:
+                run.return_value = MagicMock(returncode=0, stdout="Already up to date\n", stderr="")
+                ok, msg = pc._git_pull_plugin_dir(tmp_path)
+        assert ok is True
+        run.assert_called_once()
+        assert run.call_args[0][0][0] == "/resolved/git"
+
+    def test_install_core_raises_when_git_unresolved(self):
+        import hermes_cli.plugins_cmd as pc
+
+        _resolve_git_executable.cache_clear()
+        with patch.object(pc, "_resolve_git_executable", return_value=None):
+            with pytest.raises(PluginOperationError, match="git is not installed"):
+                pc._install_plugin_core("owner/repo", force=True)
+
+
 # ── _repo_name_from_url ──────────────────────────────────────────────────
 
 

tests/hermes_cli/test_post_setup_gating.py

@@ -0,0 +1,71 @@
+"""Tests for the post_setup install-state gate in `_toolset_needs_configuration_prompt`.
+
+Regression coverage for the cua-driver silent-no-op bug (issue #22737).
+
+When a no-key provider's only install side-effect is a `post_setup` hook
+(cua-driver, etc.), the gate function used to fall through to the
+`_toolset_has_keys` catch-all, which returned True for any provider with
+empty `env_vars` — causing `hermes tools` to write the toolset to config
+and exit `✓ Saved` without ever invoking the post_setup install. These
+tests pin the new predicate-aware behaviour so the regression doesn't
+sneak back in.
+"""
+
+from __future__ import annotations
+
+
+class TestPostSetupGate:
+    def test_cua_driver_missing_forces_setup(self, monkeypatch, tmp_path):
+        """When cua-driver isn't on PATH, the gate must return True so the
+        provider-setup flow runs and triggers `_run_post_setup`."""
+        from hermes_cli import tools_config
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setattr(tools_config.shutil, "which", lambda name: None)
+
+        assert tools_config._toolset_needs_configuration_prompt(
+            "computer_use", {}
+        ) is True
+
+    def test_cua_driver_installed_skips_setup(self, monkeypatch, tmp_path):
+        """When cua-driver is already on PATH, the gate must return False
+        so a re-save through `hermes tools` doesn't re-prompt the user."""
+        from hermes_cli import tools_config
+
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+        monkeypatch.setattr(
+            tools_config.shutil,
+            "which",
+            lambda name: "/usr/local/bin/cua-driver" if name == "cua-driver" else None,
+        )
+
+        assert tools_config._toolset_needs_configuration_prompt(
+            "computer_use", {}
+        ) is False
+
+    def test_post_setup_predicate_exception_does_not_block(self, monkeypatch):
+        """A predicate that raises must be treated as 'satisfied' so a
+        broken check can't strand the user in an infinite setup loop."""
+        from hermes_cli import tools_config
+
+        def _boom():
+            raise RuntimeError("predicate broken")
+
+        monkeypatch.setitem(tools_config._POST_SETUP_INSTALLED, "cua_driver", _boom)
+        assert tools_config._post_setup_already_installed("cua_driver") is True
+
+    def test_unregistered_post_setup_treated_as_satisfied(self):
+        """post_setup keys without a registered predicate must default to
+        'satisfied' so we don't change behaviour for hooks we haven't
+        explicitly opted in (kittentts, piper, agent_browser, etc.)."""
+        from hermes_cli import tools_config
+
+        assert tools_config._post_setup_already_installed("does_not_exist") is True
+
+    def test_cua_driver_predicate_registered(self):
+        """Keep an explicit pin on the cua_driver entry so accidental
+        deletion of the registry row would fail this test rather than
+        silently restore the original silent-no-op bug."""
+        from hermes_cli import tools_config
+
+        assert "cua_driver" in tools_config._POST_SETUP_INSTALLED

tests/hermes_cli/test_tencent_tokenhub_provider.py

@@ -304,12 +304,20 @@ class TestTencentTokenhubURLMapping:
 
 
 class TestTencentTokenhubContextLength:
-    """hy3-preview context length is registered."""
+    """hy3-preview has a context-length entry registered.
 
-    def test_hy3_preview_context_length(self):
+    Asserting the relationship (registered + ≥ 4096) instead of a
+    specific value, per AGENTS.md "Don't write change-detector tests".
+    The previous version of this class pinned an exact integer that
+    broke whenever Tencent / OpenRouter bumped the published context
+    window (#22268).
+    """
+
+    def test_hy3_preview_has_registered_context_length(self):
         from agent.model_metadata import get_model_context_length
         ctx = get_model_context_length("hy3-preview")
-        assert ctx == 256000
+        assert isinstance(ctx, int)
+        assert ctx >= 4096, f"hy3-preview context length looks unset/wrong: {ctx}"
 
 
 # =============================================================================

tests/hermes_cli/test_tools_config.py

@@ -119,6 +119,64 @@ def test_get_platform_tools_homeassistant_toolset_off_for_cron_when_hass_token_m
     assert "homeassistant" not in cron_enabled
 
 
+def test_get_platform_tools_expands_composite_when_mixed_with_configurable():
+    """``[hermes-cli, spotify]`` (composite + configurable) must keep the full
+    ``hermes-cli`` toolset alongside the explicit Spotify opt-in. The
+    has_explicit_config branch used to drop ``hermes-cli`` on the floor,
+    leaving sessions with only ``{spotify, kanban}``."""
+    config = {"platform_toolsets": {"cli": ["hermes-cli", "spotify"]}}
+
+    enabled = _get_platform_tools(config, "cli", include_default_mcp_servers=False)
+
+    # Native tools must reappear.
+    for ts in ("terminal", "file", "web", "browser", "memory", "delegation",
+               "code_execution", "todo", "session_search", "skills"):
+        assert ts in enabled, f"{ts} should be enabled when hermes-cli is listed"
+    # User explicitly opted into Spotify — must survive _DEFAULT_OFF_TOOLSETS subtraction.
+    assert "spotify" in enabled
+
+
+def test_get_platform_tools_composite_only_unchanged():
+    """Composite-only config (no configurable in list) must still take the
+    else-branch path and produce the full toolset — guards against the new
+    code accidentally hijacking the composite-only case."""
+    composite_only = _get_platform_tools(
+        {"platform_toolsets": {"cli": ["hermes-cli"]}},
+        "cli",
+        include_default_mcp_servers=False,
+    )
+    default = _get_platform_tools({}, "cli", include_default_mcp_servers=False)
+
+    assert composite_only == default
+
+
+def test_get_platform_tools_configurable_only_no_expansion():
+    """Configurable-only list (no composite) must not pull in unrelated
+    toolsets — guards against the expansion firing when ``composite_tools``
+    is empty."""
+    config = {"platform_toolsets": {"cli": ["terminal", "file"]}}
+
+    enabled = _get_platform_tools(config, "cli", include_default_mcp_servers=False)
+
+    assert "terminal" in enabled
+    assert "file" in enabled
+    # Web shouldn't sneak in via the new expansion path.
+    assert "web" not in enabled
+
+
+def test_get_platform_tools_mixed_does_not_resurrect_default_off():
+    """Expansion must subtract _DEFAULT_OFF_TOOLSETS from the implicit
+    pull-in. Without this, ``hermes-cli`` expansion would re-enable
+    ``moa`` / ``rl`` / ``homeassistant`` for users who never opted in."""
+    config = {"platform_toolsets": {"cli": ["hermes-cli", "terminal"]}}
+
+    enabled = _get_platform_tools(config, "cli", include_default_mcp_servers=False)
+
+    assert "terminal" in enabled
+    assert "moa" not in enabled
+    assert "rl" not in enabled
+
+
 def test_get_platform_tools_preserves_explicit_empty_selection():
     config = {"platform_toolsets": {"cli": []}}
 

tests/hermes_cli/test_tui_resume_flow.py

@@ -419,6 +419,72 @@ def test_oneshot_distinguishes_disabled_mcp_from_unknown(monkeypatch, capsys):
     assert "mcp-off" in err
 
 
+def test_oneshot_wires_session_db_for_recall(monkeypatch):
+    """hermes -z bypasses HermesCLI, but recall still needs SessionDB."""
+    from hermes_cli.oneshot import _run_agent
+
+    captured = {}
+    sentinel_db = object()
+
+    class FakeAgent:
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+            self.suppress_status_output = False
+            self.stream_delta_callback = object()
+            self.tool_gen_callback = object()
+
+        def chat(self, prompt):
+            captured["prompt"] = prompt
+            return "ok"
+
+    class FakeSessionDB:
+        def __new__(cls):
+            return sentinel_db
+
+    def mod(name, **attrs):
+        module = types.ModuleType(name)
+        for key, value in attrs.items():
+            setattr(module, key, value)
+        return module
+
+    monkeypatch.setitem(sys.modules, "run_agent", mod("run_agent", AIAgent=FakeAgent))
+    monkeypatch.setitem(sys.modules, "hermes_state", mod("hermes_state", SessionDB=FakeSessionDB))
+    monkeypatch.setitem(
+        sys.modules,
+        "hermes_cli.config",
+        mod("hermes_cli.config", load_config=lambda: {"model": {"default": "m"}}),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "hermes_cli.models",
+        mod("hermes_cli.models", detect_provider_for_model=lambda *_args, **_kwargs: None),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "hermes_cli.runtime_provider",
+        mod(
+            "hermes_cli.runtime_provider",
+            resolve_runtime_provider=lambda **_kwargs: {
+                "api_key": "k",
+                "base_url": "u",
+                "provider": "p",
+                "api_mode": "chat_completions",
+                "credential_pool": None,
+            },
+        ),
+    )
+    monkeypatch.setitem(
+        sys.modules,
+        "hermes_cli.tools_config",
+        mod("hermes_cli.tools_config", _get_platform_tools=lambda *_args, **_kwargs: {"session_search"}),
+    )
+
+    assert _run_agent("recall this") == "ok"
+    assert captured["session_db"] is sentinel_db
+    assert captured["enabled_toolsets"] == ["session_search"]
+    assert captured["prompt"] == "recall this"
+
+
 def test_launch_tui_exports_model_provider_and_toolsets(monkeypatch, main_mod):
     captured = {}
     active_path_during_call = None

tests/hermes_cli/test_update_autostash.py

@@ -311,7 +311,8 @@ def test_cmd_update_retries_optional_extras_individually_when_all_fails(monkeypa
     """When .[all] fails, update should keep base deps and retry extras individually."""
     _setup_update_mocks(monkeypatch, tmp_path)
     monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/uv" if name == "uv" else None)
-    monkeypatch.setattr(hermes_main, "_load_installable_optional_extras", lambda: ["matrix", "mcp"])
+    monkeypatch.setattr(hermes_main, "_is_termux_env", lambda env=None: False)
+    monkeypatch.setattr(hermes_main, "_load_installable_optional_extras", lambda group="all": ["matrix", "mcp"])
 
     recorded = []
 
@@ -360,6 +361,7 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
     """When .[all] succeeds, no fallback should be attempted."""
     _setup_update_mocks(monkeypatch, tmp_path)
     monkeypatch.setattr("shutil.which", lambda name: "/usr/bin/uv" if name == "uv" else None)
+    monkeypatch.setattr(hermes_main, "_is_termux_env", lambda env=None: False)
 
     recorded = []
 
@@ -384,6 +386,36 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
     assert ".[all]" in install_cmds[0]
 
 
+def test_install_with_optional_fallback_honors_custom_group(monkeypatch):
+    """Termux update path should target .[termux-all] when requested."""
+    calls = []
+    monkeypatch.setattr(
+        hermes_main,
+        "_load_installable_optional_extras",
+        lambda group="all": ["termux", "mcp"] if group == "termux-all" else [],
+    )
+
+    def fake_run_with_heartbeat(cmd, **kwargs):
+        calls.append(cmd)
+        if cmd[-1] == ".[termux-all]":
+            raise CalledProcessError(returncode=1, cmd=cmd)
+        return None
+
+    monkeypatch.setattr(hermes_main, "_run_install_with_heartbeat", fake_run_with_heartbeat)
+
+    hermes_main._install_python_dependencies_with_optional_fallback(
+        ["/usr/bin/uv", "pip"],
+        group="termux-all",
+    )
+
+    assert calls == [
+        ["/usr/bin/uv", "pip", "install", "-e", ".[termux-all]"],
+        ["/usr/bin/uv", "pip", "install", "-e", "."],
+        ["/usr/bin/uv", "pip", "install", "-e", ".[termux]"],
+        ["/usr/bin/uv", "pip", "install", "-e", ".[mcp]"],
+    ]
+
+
 def test_install_heartbeat_prints_when_dependency_install_is_silent(monkeypatch, capsys):
     """Long quiet installs should emit periodic heartbeat lines."""