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

# Conflicts:
#	tui_gateway/server.py
This commit is contained in:
Brooklyn Nicholson
2026-05-07 21:19:31 -04:00
288 changed files with 34632 additions and 1647 deletions
@@ -1179,3 +1179,87 @@ def test_shared_store_survives_across_profile_switch(
shared_after = auth_mod._read_shared_nous_state()
assert shared_after is not None
assert shared_after["refresh_token"] == "b-refresh-tok"
def test_runtime_refresh_uses_newer_shared_token_before_local_stale_token(
tmp_path, monkeypatch, shared_store_env,
):
"""A sibling profile may rotate the single-use Nous refresh token.
When this profile later wakes with an expired local token, runtime
resolution must adopt the shared token before refreshing. Otherwise it
can submit the stale local refresh token and trigger portal reuse
revocation for the whole shared session.
"""
from hermes_cli import auth as auth_mod
profile_b = tmp_path / "profile_b"
_setup_nous_auth(
profile_b,
access_token="local-expired-access",
refresh_token="local-stale-refresh",
)
monkeypatch.setenv("HERMES_HOME", str(profile_b))
shared_state = _full_state_fixture()
shared_state["access_token"] = "shared-fresh-access"
shared_state["refresh_token"] = "shared-fresh-refresh"
shared_state["expires_at"] = "2099-01-01T00:00:00+00:00"
auth_mod._write_shared_nous_state(shared_state)
def _refresh_should_not_happen(**_kwargs):
raise AssertionError("stale profile-local refresh token was used")
minted_with: list[str] = []
def _fake_mint_agent_key(*, client, portal_base_url, access_token, min_ttl_seconds):
minted_with.append(access_token)
return _mint_payload(api_key="agent-key-from-shared-token")
monkeypatch.setattr(auth_mod, "_refresh_access_token", _refresh_should_not_happen)
monkeypatch.setattr(auth_mod, "_mint_agent_key", _fake_mint_agent_key)
creds = auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
force_mint=True,
)
assert creds["api_key"] == "agent-key-from-shared-token"
assert minted_with == ["shared-fresh-access"]
profile_state = auth_mod.get_provider_auth_state("nous")
assert profile_state is not None
assert profile_state["refresh_token"] == "shared-fresh-refresh"
assert profile_state["access_token"] == "shared-fresh-access"
def test_managed_gateway_access_token_uses_newer_shared_token(
tmp_path, monkeypatch, shared_store_env,
):
"""Managed-tool token reads share the same stale-refresh-token hazard."""
from hermes_cli import auth as auth_mod
profile_b = tmp_path / "profile_b"
_setup_nous_auth(
profile_b,
access_token="local-expired-access",
refresh_token="local-stale-refresh",
)
monkeypatch.setenv("HERMES_HOME", str(profile_b))
shared_state = _full_state_fixture()
shared_state["access_token"] = "shared-fresh-access"
shared_state["refresh_token"] = "shared-fresh-refresh"
shared_state["expires_at"] = "2099-01-01T00:00:00+00:00"
auth_mod._write_shared_nous_state(shared_state)
def _refresh_should_not_happen(**_kwargs):
raise AssertionError("stale profile-local refresh token was used")
monkeypatch.setattr(auth_mod, "_refresh_access_token", _refresh_should_not_happen)
assert auth_mod.resolve_nous_access_token() == "shared-fresh-access"
profile_state = auth_mod.get_provider_auth_state("nous")
assert profile_state is not None
assert profile_state["refresh_token"] == "shared-fresh-refresh"
@@ -0,0 +1,360 @@
"""Tests for cross-profile auth fallback.
When ``HERMES_HOME`` points to a named profile, ``read_credential_pool()``
and ``get_provider_auth_state()`` fall back to the global-root
``auth.json`` per-provider when the profile has no entries for that
provider. Writes still target the profile only.
See the #18594 follow-up report: profile workers couldn't see providers
authenticated only at the global root.
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
def _make_auth_store(pool: dict | None = None, providers: dict | None = None) -> dict:
store: dict = {"version": 1}
if pool is not None:
store["credential_pool"] = pool
if providers is not None:
store["providers"] = providers
return store
@pytest.fixture()
def profile_env(tmp_path, monkeypatch):
"""Set up a global root + an active profile under Path.home()/.hermes/profiles/coder.
* Path.home() -> tmp_path
* Global root -> tmp_path/.hermes (has its own auth.json fixture)
* Profile -> tmp_path/.hermes/profiles/coder (active, HERMES_HOME points here)
This mirrors the real "named profile mounted under the default root"
layout that profile users actually have on disk.
"""
monkeypatch.setattr(Path, "home", lambda: tmp_path)
global_root = tmp_path / ".hermes"
global_root.mkdir()
profile_dir = global_root / "profiles" / "coder"
profile_dir.mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(profile_dir))
return {"global": global_root, "profile": profile_dir}
def _write(path: Path, payload: dict) -> None:
path.write_text(json.dumps(payload, indent=2))
# ---------------------------------------------------------------------------
# read_credential_pool — provider-slice reads
# ---------------------------------------------------------------------------
def test_profile_with_zero_entries_falls_back_to_global(profile_env):
"""Empty profile pool inherits the global-root entries for that provider."""
from hermes_cli.auth import read_credential_pool
_write(profile_env["global"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "glob-1",
"label": "global-key",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-global",
}],
}))
# Profile auth.json: exists but has no openrouter entries.
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={}))
entries = read_credential_pool("openrouter")
assert len(entries) == 1
assert entries[0]["id"] == "glob-1"
assert entries[0]["access_token"] == "sk-or-global"
def test_profile_with_entries_fully_shadows_global(profile_env):
"""Once the profile has any entries for a provider, global is ignored."""
from hermes_cli.auth import read_credential_pool
_write(profile_env["global"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "glob-1",
"label": "global-key",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-global",
}],
}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "prof-1",
"label": "profile-key",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-profile",
}],
}))
entries = read_credential_pool("openrouter")
assert len(entries) == 1
assert entries[0]["id"] == "prof-1"
assert entries[0]["access_token"] == "sk-or-profile"
def test_per_provider_shadowing_is_independent(profile_env):
"""Profile can override one provider while inheriting another from global."""
from hermes_cli.auth import read_credential_pool
_write(profile_env["global"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "glob-or",
"label": "global-or",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-global",
}],
"anthropic": [{
"id": "glob-ant",
"label": "global-ant",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-ant-global",
}],
}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={
# Profile has openrouter only — anthropic should still fall back.
"openrouter": [{
"id": "prof-or",
"label": "profile-or",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-profile",
}],
}))
or_entries = read_credential_pool("openrouter")
ant_entries = read_credential_pool("anthropic")
assert [e["id"] for e in or_entries] == ["prof-or"]
assert [e["id"] for e in ant_entries] == ["glob-ant"]
def test_missing_global_auth_file_is_safe(profile_env):
"""Profile processes that never had a global auth.json still work."""
from hermes_cli.auth import read_credential_pool
# No global auth.json written at all.
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "prof-1",
"label": "profile",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-profile",
}],
}))
assert read_credential_pool("openrouter")[0]["id"] == "prof-1"
assert read_credential_pool("anthropic") == []
def test_malformed_global_auth_file_does_not_break_profile_read(profile_env):
(profile_env["global"] / "auth.json").write_text("{not valid json")
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "prof-1",
"label": "profile",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-profile",
}],
}))
from hermes_cli.auth import read_credential_pool
# Profile reads still work; malformed global is silently ignored.
assert read_credential_pool("openrouter")[0]["id"] == "prof-1"
# And no fallback for anthropic since global is unreadable.
assert read_credential_pool("anthropic") == []
# ---------------------------------------------------------------------------
# read_credential_pool — whole-pool reads (provider_id=None)
# ---------------------------------------------------------------------------
def test_whole_pool_merges_global_providers_when_missing_locally(profile_env):
from hermes_cli.auth import read_credential_pool
_write(profile_env["global"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "glob-or",
"label": "global-or",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-global",
}],
"anthropic": [{
"id": "glob-ant",
"label": "global-ant",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-ant-global",
}],
}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "prof-or",
"label": "profile-or",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-or-profile",
}],
}))
pool = read_credential_pool(None)
# Profile wins for openrouter, global fills in anthropic.
assert [e["id"] for e in pool["openrouter"]] == ["prof-or"]
assert [e["id"] for e in pool["anthropic"]] == ["glob-ant"]
# ---------------------------------------------------------------------------
# get_provider_auth_state — singleton fallback
# ---------------------------------------------------------------------------
def test_provider_auth_state_falls_back_to_global_when_profile_has_none(profile_env):
from hermes_cli.auth import get_provider_auth_state
_write(profile_env["global"] / "auth.json", _make_auth_store(providers={
"nous": {"access_token": "nous-global", "refresh_token": "rt-global"},
}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(providers={}))
state = get_provider_auth_state("nous")
assert state is not None
assert state["access_token"] == "nous-global"
def test_provider_auth_state_profile_wins_when_present(profile_env):
from hermes_cli.auth import get_provider_auth_state
_write(profile_env["global"] / "auth.json", _make_auth_store(providers={
"nous": {"access_token": "nous-global"},
}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(providers={
"nous": {"access_token": "nous-profile"},
}))
state = get_provider_auth_state("nous")
assert state is not None
assert state["access_token"] == "nous-profile"
def test_provider_auth_state_returns_none_when_neither_has_it(profile_env):
from hermes_cli.auth import get_provider_auth_state
_write(profile_env["global"] / "auth.json", _make_auth_store(providers={}))
_write(profile_env["profile"] / "auth.json", _make_auth_store(providers={}))
assert get_provider_auth_state("nous") is None
# ---------------------------------------------------------------------------
# Classic mode — no fallback path should ever trigger
# ---------------------------------------------------------------------------
def test_classic_mode_does_not_double_read_same_file(tmp_path, monkeypatch):
"""In classic mode (HERMES_HOME == global root), no fallback path runs.
This guards against the merge accidentally duplicating entries when the
profile and global resolve to the same directory.
"""
# Put Path.home() under a subdir so the seat belt in _auth_file_path()
# sees tmp_path/home/.hermes as the "real home" — which is NOT equal
# to the HERMES_HOME we set (tmp_path/classic), so the guard passes.
fake_home = tmp_path / "home"
fake_home.mkdir()
monkeypatch.setattr(Path, "home", lambda: fake_home)
hermes_home = tmp_path / "classic"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
_write(hermes_home / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "only",
"label": "classic",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-classic",
}],
}))
from hermes_cli.auth import read_credential_pool, _global_auth_file_path
# Classic mode: HERMES_HOME is set to a custom path that is NOT under
# ~/.hermes/profiles/ — get_default_hermes_root() returns HERMES_HOME
# itself, so the profile root and global root are the same directory,
# and the helper correctly returns None (no fallback).
assert _global_auth_file_path() is None
# And the read should return exactly one entry (not two).
entries = read_credential_pool("openrouter")
assert len(entries) == 1
assert entries[0]["id"] == "only"
# ---------------------------------------------------------------------------
# Writes stay scoped to the profile
# ---------------------------------------------------------------------------
def test_write_credential_pool_targets_profile_not_global(profile_env):
from hermes_cli.auth import read_credential_pool, write_credential_pool
_write(profile_env["global"] / "auth.json", _make_auth_store(pool={
"openrouter": [{
"id": "glob-1",
"label": "global",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-global",
}],
}))
write_credential_pool("openrouter", [{
"id": "prof-new",
"label": "profile-new",
"auth_type": "api_key",
"priority": 0,
"source": "manual",
"access_token": "sk-profile-new",
}])
# Global auth.json unchanged.
global_data = json.loads((profile_env["global"] / "auth.json").read_text())
assert global_data["credential_pool"]["openrouter"][0]["id"] == "glob-1"
# Profile auth.json holds the new entry.
profile_data = json.loads((profile_env["profile"] / "auth.json").read_text())
assert profile_data["credential_pool"]["openrouter"][0]["id"] == "prof-new"
# Subsequent read returns profile (shadows global).
assert [e["id"] for e in read_credential_pool("openrouter")] == ["prof-new"]
@@ -0,0 +1,198 @@
"""Regression tests for TOCTOU-safe credential file writers in ``hermes_cli.auth``.
Background
==========
The three writers below used to create a temp file via ``Path.write_text`` /
``Path.open('w')`` and only ``chmod``'d it to ``0o600`` afterward. Between
create and chmod the file existed at the process umask (typically ``0o644``),
briefly exposing OAuth tokens to other local users on multi-user hosts. The
fix switches them to ``os.open(O_EXCL, mode=0o600)`` + ``os.fdopen`` +
``fsync`` so the file is atomic at ``0o600`` on creation. Mirrors the fixes
shipped for ``agent/google_oauth.py`` (#19673) and ``tools/mcp_oauth.py``
(#21148).
These tests stay green only while the token file and its parent directory
end up at ``0o600`` / ``0o700`` after every write. POSIX-only — the mode-bit
enforcement does not exist on Windows.
"""
from __future__ import annotations
import json
import os
import stat
import sys
from unittest.mock import patch
import pytest
pytestmark = pytest.mark.skipif(
sys.platform.startswith("win"),
reason="POSIX mode bits not enforced on Windows",
)
# ---------------------------------------------------------------------------
# _save_auth_store (~/.hermes/auth.json — every native OAuth provider)
# ---------------------------------------------------------------------------
def test_save_auth_store_writes_0o600_with_0o700_parent(tmp_path, monkeypatch):
"""``_save_auth_store`` must land ``auth.json`` at 0o600 and parent at 0o700."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
old_umask = os.umask(0o022) # make the race observable if it regresses
try:
from hermes_cli import auth as auth_mod
auth_store = {
"version": auth_mod.AUTH_STORE_VERSION,
"providers": {"openai-codex": {"tokens": {"access_token": "secret-x"}}},
"active_provider": "openai-codex",
}
auth_path = auth_mod._save_auth_store(auth_store)
finally:
os.umask(old_umask)
mode = stat.S_IMODE(auth_path.stat().st_mode)
parent_mode = stat.S_IMODE(auth_path.parent.stat().st_mode)
assert mode == 0o600, (
f"auth.json mode 0o{mode:o} != 0o600 — TOCTOU race regressed"
)
assert parent_mode == 0o700, (
f"auth.json parent dir mode 0o{parent_mode:o} != 0o700 — siblings can traverse"
)
# Content survived the rewrite
data = json.loads(auth_path.read_text())
assert data["providers"]["openai-codex"]["tokens"]["access_token"] == "secret-x"
# ---------------------------------------------------------------------------
# _save_qwen_cli_tokens (Qwen CLI OAuth tokens)
# ---------------------------------------------------------------------------
def test_save_qwen_cli_tokens_writes_0o600_with_0o700_parent(tmp_path, monkeypatch):
"""``_save_qwen_cli_tokens`` must land the token file at 0o600 and parent at 0o700."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
# The Qwen CLI auth path lives under $HOME/.qwen by default — isolate it.
monkeypatch.setenv("HOME", str(tmp_path))
old_umask = os.umask(0o022)
try:
from hermes_cli import auth as auth_mod
tokens = {
"access_token": "qwen-secret",
"refresh_token": "qwen-refresh",
"token_type": "Bearer",
"expiry_date": 123,
}
auth_path = auth_mod._save_qwen_cli_tokens(tokens)
finally:
os.umask(old_umask)
mode = stat.S_IMODE(auth_path.stat().st_mode)
parent_mode = stat.S_IMODE(auth_path.parent.stat().st_mode)
assert mode == 0o600, (
f"Qwen token file mode 0o{mode:o} != 0o600 — TOCTOU race regressed"
)
assert parent_mode == 0o700, (
f"Qwen token parent dir mode 0o{parent_mode:o} != 0o700"
)
data = json.loads(auth_path.read_text())
assert data["access_token"] == "qwen-secret"
# ---------------------------------------------------------------------------
# Nous shared-credential store write (inside _write_shared_nous_state)
# ---------------------------------------------------------------------------
def test_shared_nous_store_writes_0o600_with_0o700_parent(tmp_path, monkeypatch):
"""The Nous shared-credential store must land at 0o600 / parent 0o700."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
# _nous_shared_store_path() refuses to touch the real shared store during
# pytest runs; redirect it into tmp_path explicitly.
monkeypatch.setenv("HERMES_SHARED_AUTH_DIR", str(tmp_path / "shared"))
old_umask = os.umask(0o022)
try:
from hermes_cli import auth as auth_mod
state = {
"access_token": "nous-access-xxx",
"refresh_token": "nous-refresh-xxx",
"token_type": "Bearer",
"scope": "openid profile",
"client_id": "test-client",
"obtained_at": "2026-01-01T00:00:00Z",
"expires_at": "2026-01-01T01:00:00Z",
}
auth_mod._write_shared_nous_state(state)
path = auth_mod._nous_shared_store_path()
finally:
os.umask(old_umask)
assert path.exists(), "shared Nous store was not written"
mode = stat.S_IMODE(path.stat().st_mode)
parent_mode = stat.S_IMODE(path.parent.stat().st_mode)
assert mode == 0o600, (
f"Nous shared store mode 0o{mode:o} != 0o600 — TOCTOU race regressed"
)
assert parent_mode == 0o700, (
f"Nous shared store parent dir mode 0o{parent_mode:o} != 0o700"
)
data = json.loads(path.read_text())
assert data["refresh_token"] == "nous-refresh-xxx"
# ---------------------------------------------------------------------------
# Atomicity: verify ``os.open`` is called with an explicit 0o600 mode.
# ---------------------------------------------------------------------------
def test_save_auth_store_uses_os_open_with_0o600_mode(tmp_path, monkeypatch):
"""Regression: the writer must call ``os.open`` with an explicit restricted
mode so the file is created at 0o600 atomically — closing the TOCTOU
window the previous ``Path.open('w')`` left open (fd inherited process
umask and was briefly 0o644 before post-write chmod)."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
observed_opens: list[tuple[str, int, int]] = []
real_os_open = os.open
def spying_os_open(path, flags, mode=0o777, *args, **kwargs):
observed_opens.append((str(path), flags, mode))
return real_os_open(path, flags, mode, *args, **kwargs)
with patch.object(os, "open", spying_os_open):
from hermes_cli import auth as auth_mod
auth_mod._save_auth_store(
{"version": auth_mod.AUTH_STORE_VERSION, "providers": {}}
)
auth_tmp_opens = [
(p, fl, m) for (p, fl, m) in observed_opens if "auth.json.tmp" in p
]
assert auth_tmp_opens, (
f"os.open was never called for the auth.json temp file; "
f"observed={observed_opens!r}"
)
for path, flags, mode in auth_tmp_opens:
assert flags & os.O_CREAT, f"auth.json temp open missing O_CREAT: path={path}"
assert flags & os.O_EXCL, (
f"auth.json temp open missing O_EXCL — TOCTOU-safe pattern regressed: "
f"path={path}, flags={flags}"
)
# Must be exactly S_IRUSR | S_IWUSR (0o600) — no group/other bits.
expected = stat.S_IRUSR | stat.S_IWUSR
assert mode == expected, (
f"auth.json temp open mode 0o{mode:o} != 0o{expected:o}"
f"umask would apply and potentially expose tokens"
)
+13 -5
View File
@@ -153,14 +153,18 @@ class TestCmdUpdateBranchFallback:
(["/usr/bin/npm", "run", "build"], PROJECT_ROOT / "apps" / "dashboard"),
]
def test_update_non_interactive_skips_migration_prompt(self, mock_args, capsys):
"""When stdin/stdout aren't TTYs, config migration prompt is skipped."""
def test_update_non_interactive_runs_safe_config_migrations(self, mock_args, capsys):
"""Dashboard/web updates apply non-interactive migrations before restart."""
with patch("shutil.which", return_value=None), patch(
"subprocess.run"
) as mock_run, patch("builtins.input") as mock_input, patch(
"hermes_cli.config.get_missing_env_vars", return_value=["MISSING_KEY"]
), patch("hermes_cli.config.get_missing_config_fields", return_value=[]), patch(
"hermes_cli.config.check_config_version", return_value=(1, 2)
), patch(
"hermes_cli.config.get_missing_config_fields",
return_value=[{"key": "new.option", "default": True}],
), patch("hermes_cli.config.check_config_version", return_value=(1, 2)), patch(
"hermes_cli.config.migrate_config",
return_value={"env_added": [], "config_added": ["new.option"]},
), patch("hermes_cli.main.sys") as mock_sys:
mock_sys.stdin.isatty.return_value = False
mock_sys.stdout.isatty.return_value = False
@@ -171,8 +175,12 @@ class TestCmdUpdateBranchFallback:
cmd_update(mock_args)
mock_input.assert_not_called()
from hermes_cli.config import migrate_config
migrate_config.assert_called_once_with(interactive=False, quiet=False)
captured = capsys.readouterr()
assert "Non-interactive session" in captured.out
assert "applying safe config migrations" in captured.out
assert "API keys require manual entry" in captured.out
class TestCmdUpdateProfileSkillSync:
+87
View File
@@ -0,0 +1,87 @@
"""Tests for `hermes curator run` CLI behavior."""
from __future__ import annotations
from types import SimpleNamespace
def _args(**kwargs):
values = {
"dry_run": False,
"synchronous": False,
"background": False,
}
values.update(kwargs)
return SimpleNamespace(**values)
def test_run_defaults_to_synchronous(monkeypatch, capsys):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
calls = []
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(
curator_state,
"run_curator_review",
lambda **kwargs: calls.append(kwargs) or {"auto_transitions": {}},
)
assert curator_cli._cmd_run(_args()) == 0
assert calls[0]["synchronous"] is True
assert calls[0]["dry_run"] is False
assert "background" not in capsys.readouterr().out
def test_run_background_opts_into_async(monkeypatch, capsys):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
calls = []
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(
curator_state,
"run_curator_review",
lambda **kwargs: calls.append(kwargs) or {"auto_transitions": {}},
)
assert curator_cli._cmd_run(_args(background=True)) == 0
assert calls[0]["synchronous"] is False
assert "llm pass running in background" in capsys.readouterr().out
def test_run_sync_wins_over_background(monkeypatch):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
calls = []
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(
curator_state,
"run_curator_review",
lambda **kwargs: calls.append(kwargs) or {"auto_transitions": {}},
)
assert curator_cli._cmd_run(_args(synchronous=True, background=True)) == 0
assert calls[0]["synchronous"] is True
def test_dry_run_default_reports_synchronous_wording(monkeypatch, capsys):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(
curator_state,
"run_curator_review",
lambda **kwargs: {"auto_transitions": {}},
)
assert curator_cli._cmd_run(_args(dry_run=True)) == 0
out = capsys.readouterr().out
assert "When the report lands" not in out
assert "Read the report with `hermes curator status`" in out
+25
View File
@@ -175,3 +175,28 @@ def test_status_no_skills_produces_clean_empty_output(curator_status_env):
# None of the ranking sections render
assert "most active" not in out
assert "least active" not in out
def test_status_marks_missing_last_report_path(monkeypatch, capsys, tmp_path):
import agent.curator as curator_state
import hermes_cli.curator as curator_cli
import tools.skill_usage as skill_usage
missing_report = tmp_path / "stale-report"
monkeypatch.setattr(curator_state, "load_state", lambda: {
"paused": False,
"last_run_at": None,
"last_run_summary": "auto: no changes",
"run_count": 1,
"last_report_path": str(missing_report),
})
monkeypatch.setattr(curator_state, "is_enabled", lambda: True)
monkeypatch.setattr(curator_state, "get_interval_hours", lambda: 168)
monkeypatch.setattr(curator_state, "get_stale_after_days", lambda: 30)
monkeypatch.setattr(curator_state, "get_archive_after_days", lambda: 90)
monkeypatch.setattr(skill_usage, "agent_created_report", lambda: [])
assert curator_cli._cmd_status(SimpleNamespace()) == 0
out = capsys.readouterr().out
assert f"last report: {missing_report} (missing)" in out
+16 -9
View File
@@ -291,9 +291,11 @@ class TestCaptureLogSnapshotRedaction:
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
# Critical: ensure the user has NOT opted in to redaction. The whole
# point of this PR is that share-time redaction works for users who
# never set this env var.
# Baseline fixture: no explicit env-var opinion. With the post-#17691
# default of ON, the default-path tests below exercise the
# secure-default behaviour. The `force=True` regression test
# setenvs to "false" inline to prove force=True works even when
# the runtime flag is disabled.
monkeypatch.delenv("HERMES_REDACT_SECRETS", raising=False)
logs_dir = home / "logs"
@@ -324,21 +326,26 @@ class TestCaptureLogSnapshotRedaction:
assert _REDACT_FIXTURE_TOKEN in snap.tail_text
assert _REDACT_FIXTURE_TOKEN in (snap.full_text or "")
def test_force_true_overrides_unset_env_var(self, hermes_home_with_secret):
def test_force_true_works_when_redaction_disabled(
self, hermes_home_with_secret, monkeypatch
):
"""Regression test: redact_sensitive_text short-circuits without force=True.
If a future refactor drops `force=True` from `_redact_log_text`, this
test fails immediately. Without `force=True`, the redactor returns the
input unchanged when HERMES_REDACT_SECRETS is unset, and the feature
ships silently broken for its target audience.
input unchanged when HERMES_REDACT_SECRETS=false, and the share-time
redaction feature ships silently broken for users who opted out of
runtime redaction (e.g. developers working on the redactor itself).
"""
import os
# Force the runtime flag off so we're exercising the force=True path,
# not the default-on path.
monkeypatch.setenv("HERMES_REDACT_SECRETS", "false")
from hermes_cli.debug import _capture_log_snapshot
# Belt-and-suspenders: confirm the env var is genuinely unset for this
# test so we know we're exercising the force=True path.
assert os.environ.get("HERMES_REDACT_SECRETS", "") == ""
assert os.environ.get("HERMES_REDACT_SECRETS", "") == "false"
snap = _capture_log_snapshot("agent", tail_lines=10)
+59
View File
@@ -378,6 +378,11 @@ def test_run_doctor_termux_treats_docker_and_browser_warnings_as_expected(monkey
assert "1) pkg install nodejs" in out
assert "2) npm install -g agent-browser" in out
assert "3) agent-browser install" in out
assert "Termux compatibility fallbacks:" in out
assert "use .[termux-all] for broad compatibility" in out
assert "Matrix E2EE extra is excluded on Termux" in out
assert "Local faster-whisper extra is excluded on Termux" in out
assert "STT fallback: use Groq Whisper (set GROQ_API_KEY) or OpenAI Whisper (set VOICE_TOOLS_OPENAI_KEY)." in out
assert "docker not found (optional)" not in out
@@ -652,6 +657,60 @@ def test_run_doctor_kimi_cn_env_is_detected_and_probe_is_null_safe(monkeypatch,
assert any(url == "https://api.moonshot.cn/v1/models" for url, _, _ in calls)
def test_run_doctor_dashscope_retries_china_endpoint_after_intl_unauthorized(monkeypatch, tmp_path):
home = tmp_path / ".hermes"
home.mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
(home / ".env").write_text("DASHSCOPE_API_KEY=sk-test\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir(exist_ok=True)
monkeypatch.setattr(doctor_mod, "HERMES_HOME", home)
monkeypatch.setattr(doctor_mod, "PROJECT_ROOT", project)
monkeypatch.setattr(doctor_mod, "_DHH", str(home))
monkeypatch.setenv("DASHSCOPE_API_KEY", "sk-test")
monkeypatch.delenv("DASHSCOPE_BASE_URL", raising=False)
fake_model_tools = types.SimpleNamespace(
check_tool_availability=lambda *a, **kw: ([], []),
TOOLSET_REQUIREMENTS={},
)
monkeypatch.setitem(sys.modules, "model_tools", fake_model_tools)
try:
from hermes_cli import auth as _auth_mod
monkeypatch.setattr(_auth_mod, "get_nous_auth_status", lambda: {})
monkeypatch.setattr(_auth_mod, "get_codex_auth_status", lambda: {})
except ImportError:
pass
calls = []
def fake_get(url, headers=None, timeout=None):
calls.append((url, headers, timeout))
status = 200 if "dashscope.aliyuncs.com" in url else 401
return types.SimpleNamespace(status_code=status)
import httpx
monkeypatch.setattr(httpx, "get", fake_get)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
doctor_mod.run_doctor(Namespace(fix=False))
out = buf.getvalue()
assert "Alibaba/DashScope" in out
assert "invalid API key" not in out
assert any(
url == "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/models"
for url, _, _ in calls
)
assert any(
url == "https://dashscope.aliyuncs.com/compatible-mode/v1/models"
for url, _, _ in calls
)
@pytest.mark.parametrize("base_url", [None, "https://opencode.ai/zen/go/v1"])
def test_run_doctor_opencode_go_skips_invalid_models_probe(monkeypatch, tmp_path, base_url):
home = tmp_path / ".hermes"
+37
View File
@@ -53,6 +53,43 @@ def test_run_gateway_exits_nonzero_when_start_gateway_reports_failure(monkeypatc
assert calls == [(True, None)]
def test_run_gateway_refuses_root_in_official_docker(monkeypatch, tmp_path, capsys):
project_root = tmp_path / "opt" / "hermes"
(project_root / "docker").mkdir(parents=True)
(project_root / "docker" / "entrypoint.sh").write_text("#!/bin/sh\n")
monkeypatch.setattr(gateway, "PROJECT_ROOT", project_root)
monkeypatch.setattr(gateway.os, "geteuid", lambda: 0)
monkeypatch.delenv("HERMES_ALLOW_ROOT_GATEWAY", raising=False)
monkeypatch.setattr(gateway, "_is_official_docker_checkout", lambda: True)
with pytest.raises(SystemExit) as exc_info:
gateway.run_gateway()
assert exc_info.value.code == 1
out = capsys.readouterr().out
assert "Refusing to run the Hermes gateway as root" in out
assert "/opt/hermes/docker/entrypoint.sh" in out
def test_run_gateway_root_guard_has_escape_hatch(monkeypatch):
calls = []
def fake_start_gateway(*, replace, verbosity):
calls.append((replace, verbosity))
return object()
_install_fake_gateway_run(monkeypatch, fake_start_gateway)
monkeypatch.setattr(gateway.asyncio, "run", lambda coro: True)
monkeypatch.setattr(gateway.os, "geteuid", lambda: 0)
monkeypatch.setattr(gateway, "_is_official_docker_checkout", lambda: True)
monkeypatch.setenv("HERMES_ALLOW_ROOT_GATEWAY", "1")
gateway.run_gateway(verbose=2, replace=True)
assert calls == [(True, 2)]
class TestSystemdLingerStatus:
def test_reports_enabled(self, monkeypatch):
monkeypatch.setattr(gateway, "is_linux", lambda: True)
+294 -33
View File
@@ -2,6 +2,7 @@
import os
import pwd
import subprocess
from pathlib import Path
from types import SimpleNamespace
@@ -90,6 +91,13 @@ class TestSystemdServiceRefresh:
monkeypatch.setattr(gateway_cli, "generate_systemd_unit", lambda system=False, run_as_user=None: "new unit\n")
calls = []
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(gateway_cli, "_recover_pending_systemd_restart", lambda system=False, previous_pid=None: False)
monkeypatch.setattr(
gateway_cli,
"_wait_for_systemd_service_restart",
lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
)
def fake_run(cmd, check=True, **kwargs):
calls.append(cmd)
@@ -100,11 +108,12 @@ class TestSystemdServiceRefresh:
gateway_cli.systemd_restart()
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
assert calls[:4] == [
assert calls[:5] == [
["systemctl", "--user", "daemon-reload"],
["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus"],
["systemctl", "--user", "show", gateway_cli.get_service_name(), "--no-pager", "--property", "ActiveState,SubState,Result,ExecMainStatus,MainPID"],
["systemctl", "--user", "reset-failed", gateway_cli.get_service_name()],
["systemctl", "--user", "reload-or-restart", gateway_cli.get_service_name()],
["systemctl", "--user", "restart", gateway_cli.get_service_name()],
("wait", False, None),
]
def test_systemd_stop_marks_running_gateway_as_planned_stop(self, monkeypatch):
@@ -611,62 +620,141 @@ class TestGatewayServiceDetection:
assert gateway_cli._is_service_running() is False
class TestGatewaySystemServiceRouting:
def test_systemd_restart_self_requests_graceful_restart_and_waits(self, monkeypatch, capsys):
def test_systemd_restart_gracefully_restarts_running_service_and_waits(self, monkeypatch, capsys):
calls = []
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: calls.append(("refresh", system)))
monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 12.0)
monkeypatch.setattr(
"gateway.status.get_running_pid",
lambda: 654,
)
monkeypatch.setattr(
gateway_cli,
"_request_gateway_self_restart",
lambda pid: calls.append(("self", pid)) or True,
"_graceful_restart_via_sigusr1",
lambda pid, timeout: calls.append(("graceful", pid, timeout)) or True,
)
# Simulate: old process dies immediately, new process becomes active
kill_call_count = [0]
def fake_kill(pid, sig):
kill_call_count[0] += 1
if kill_call_count[0] >= 2: # first call checks, second = dead
raise ProcessLookupError()
monkeypatch.setattr(os, "kill", fake_kill)
# Simulate systemctl reset-failed/start followed by an active unit
new_pid = [None]
# Simulate systemctl reset-failed/restart followed by an active unit.
# A plain start does not break systemd's auto-restart timer once the
# old gateway has exited with the planned restart code.
def fake_subprocess_run(cmd, **kwargs):
if "reset-failed" in cmd:
calls.append(("reset-failed", cmd))
return SimpleNamespace(stdout="", returncode=0)
if "start" in cmd:
calls.append(("start", cmd))
if "restart" in cmd:
calls.append(("restart", cmd))
return SimpleNamespace(stdout="", returncode=0)
if "show" in cmd:
new_pid[0] = 999
return SimpleNamespace(
stdout="ActiveState=active\nSubState=running\nResult=success\nExecMainStatus=0\n",
returncode=0,
)
raise AssertionError(f"Unexpected systemctl call: {cmd}")
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_subprocess_run)
# get_running_pid returns new PID after restart
pid_calls = [0]
def fake_get_pid():
pid_calls[0] += 1
return 999 if pid_calls[0] > 1 else 654
monkeypatch.setattr("gateway.status.get_running_pid", fake_get_pid)
monkeypatch.setattr(
gateway_cli,
"_wait_for_systemd_service_restart",
lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
)
gateway_cli.systemd_restart()
assert ("self", 654) in calls
assert ("graceful", 654, 17.0) in calls
assert any(call[0] == "reset-failed" for call in calls)
assert any(call[0] == "start" for call in calls)
assert any(call[0] == "restart" for call in calls)
assert ("wait", False, 654) in calls
out = capsys.readouterr().out.lower()
assert "restarted" in out
assert "restarting gracefully" in out
def test_systemd_restart_uses_systemd_main_pid_when_pid_file_is_missing(self, monkeypatch, capsys):
calls = []
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
monkeypatch.setattr(gateway_cli, "_get_restart_drain_timeout", lambda: 10.0)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(
gateway_cli,
"_read_systemd_unit_properties",
lambda system=False: {
"ActiveState": "active",
"SubState": "running",
"Result": "success",
"ExecMainStatus": "0",
"MainPID": "777",
},
)
monkeypatch.setattr(
gateway_cli,
"_graceful_restart_via_sigusr1",
lambda pid, timeout: calls.append(("graceful", pid, timeout)) or True,
)
monkeypatch.setattr(gateway_cli, "_run_systemctl", lambda args, **kwargs: calls.append(args) or SimpleNamespace(stdout="", returncode=0))
monkeypatch.setattr(
gateway_cli,
"_wait_for_systemd_service_restart",
lambda system=False, previous_pid=None: calls.append(("wait", system, previous_pid)) or True,
)
gateway_cli.systemd_restart()
assert ("graceful", 777, 15.0) in calls
assert ("wait", False, 777) in calls
assert "restarting gracefully (pid 777)" in capsys.readouterr().out.lower()
def test_wait_for_systemd_restart_waits_for_runtime_running(self, monkeypatch, capsys):
monkeypatch.setattr(
gateway_cli,
"_read_systemd_unit_properties",
lambda system=False: {
"ActiveState": "active",
"SubState": "running",
"Result": "success",
"ExecMainStatus": "0",
"MainPID": "999",
},
)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(
gateway_cli,
"_gateway_runtime_status_for_pid",
lambda pid: {"pid": pid, "gateway_state": "running"},
)
assert gateway_cli._wait_for_systemd_service_restart(previous_pid=777, timeout=0.1) is True
assert "restarted (pid 999)" in capsys.readouterr().out.lower()
def test_systemd_restart_reports_start_limit_hit(self, monkeypatch, capsys):
calls = []
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
monkeypatch.setattr(gateway_cli, "_require_service_installed", lambda action, system=False: None)
monkeypatch.setattr(gateway_cli, "refresh_systemd_unit_if_needed", lambda system=False: None)
monkeypatch.setattr("gateway.status.get_running_pid", lambda: None)
monkeypatch.setattr(gateway_cli, "_recover_pending_systemd_restart", lambda system=False, previous_pid=None: False)
def fake_run_systemctl(args, **kwargs):
calls.append(args)
if args[0] == "show":
return SimpleNamespace(stdout="ActiveState=inactive\nSubState=dead\nResult=success\nExecMainStatus=0\nMainPID=0\n", stderr="", returncode=0)
if args[0] == "reset-failed":
return SimpleNamespace(stdout="", stderr="", returncode=0)
if args[0] == "restart":
raise subprocess.CalledProcessError(
1,
["systemctl", "--user", *args],
stderr="Job failed. See result 'start-limit-hit'.",
)
raise AssertionError(f"Unexpected args: {args}")
monkeypatch.setattr(gateway_cli, "_run_systemctl", fake_run_systemctl)
gateway_cli.systemd_restart()
assert ["restart", gateway_cli.get_service_name()] in calls
out = capsys.readouterr().out.lower()
assert "rate-limited by systemd" in out
assert "reset-failed" in out
def test_systemd_restart_recovers_failed_planned_restart(self, monkeypatch, capsys):
monkeypatch.setattr(gateway_cli, "_select_systemd_scope", lambda system=False: False)
@@ -711,6 +799,11 @@ class TestGatewaySystemServiceRouting:
"gateway.status.get_running_pid",
lambda: 999 if started["value"] else None,
)
monkeypatch.setattr(
gateway_cli,
"_gateway_runtime_status_for_pid",
lambda pid: {"pid": pid, "gateway_state": "running"},
)
gateway_cli.systemd_restart()
@@ -2177,3 +2270,171 @@ class TestSystemdInstallOffersLegacyRemoval:
assert prompt_called["count"] == 0
assert remove_called["invoked"] is False
class TestSystemScopeRequiresRootError:
"""Tests for the SystemScopeRequiresRootError replacement of sys.exit(1).
Before this change, ``_require_root_for_system_service`` called
``sys.exit(1)`` when non-root code tried a system-scope systemd
operation. The wizard's ``except Exception`` guards don't catch
``SystemExit`` (it's a ``BaseException`` subclass), so the user was
dumped at a bare shell prompt mid-setup. The fix raises a typed
exception instead, which the wizard intercepts and handles with
actionable remediation.
"""
def test_require_root_raises_when_non_root(self, monkeypatch):
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
with pytest.raises(gateway_cli.SystemScopeRequiresRootError) as excinfo:
gateway_cli._require_root_for_system_service("start")
assert excinfo.value.args[0] == "System gateway start requires root. Re-run with sudo."
assert excinfo.value.args[1] == "start"
# str(e) renders only the message, not the tuple repr, so that
# wizard format strings like f"Failed: {e}" print cleanly.
assert str(excinfo.value) == "System gateway start requires root. Re-run with sudo."
assert f"Failed: {excinfo.value}" == "Failed: System gateway start requires root. Re-run with sudo."
def test_require_root_noop_when_root(self, monkeypatch):
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 0)
# Should not raise, should not exit
gateway_cli._require_root_for_system_service("start")
def test_error_is_runtime_error_subclass(self):
"""Wizards use ``except Exception`` guards — the error must be a
``RuntimeError`` (catchable by ``Exception``), NOT a ``SystemExit``
(``BaseException``), so the wizard can recover from it.
"""
err = gateway_cli.SystemScopeRequiresRootError("msg", "start")
assert isinstance(err, RuntimeError)
assert isinstance(err, Exception)
assert not isinstance(err, SystemExit)
class TestSystemScopeWizardPreCheck:
"""Tests for _system_scope_wizard_would_need_root — the guard the
wizard uses to detect the dead-end BEFORE prompting the user to start
a service that will fail without sudo.
"""
@staticmethod
def _setup_units(tmp_path, monkeypatch, system_present: bool, user_present: bool):
sys_dir = tmp_path / "sys"
usr_dir = tmp_path / "usr"
sys_dir.mkdir()
usr_dir.mkdir()
if system_present:
(sys_dir / "hermes-gateway.service").write_text("[Unit]\n")
if user_present:
(usr_dir / "hermes-gateway.service").write_text("[Unit]\n")
monkeypatch.setattr(
gateway_cli,
"get_systemd_unit_path",
lambda system=False: (sys_dir if system else usr_dir) / "hermes-gateway.service",
)
def test_non_root_with_only_system_unit_returns_true(self, tmp_path, monkeypatch):
self._setup_units(tmp_path, monkeypatch, system_present=True, user_present=False)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
assert gateway_cli._system_scope_wizard_would_need_root() is True
def test_root_never_needs_root(self, tmp_path, monkeypatch):
self._setup_units(tmp_path, monkeypatch, system_present=True, user_present=False)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 0)
assert gateway_cli._system_scope_wizard_would_need_root() is False
def test_non_root_with_user_unit_present_returns_false(self, tmp_path, monkeypatch):
# User-scope unit present — user can start it themselves, no sudo needed.
self._setup_units(tmp_path, monkeypatch, system_present=True, user_present=True)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
assert gateway_cli._system_scope_wizard_would_need_root() is False
def test_non_root_with_no_units_returns_false(self, tmp_path, monkeypatch):
self._setup_units(tmp_path, monkeypatch, system_present=False, user_present=False)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
assert gateway_cli._system_scope_wizard_would_need_root() is False
def test_non_root_with_explicit_system_arg_returns_true(self, tmp_path, monkeypatch):
# Caller passed system=True explicitly (e.g. ``hermes gateway start --system``).
self._setup_units(tmp_path, monkeypatch, system_present=False, user_present=False)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
assert gateway_cli._system_scope_wizard_would_need_root(system=True) is True
class TestSystemScopeRemediationOutput:
"""Tests for _print_system_scope_remediation — the actionable guidance
shown when the wizard detects a system-scope-only setup as non-root.
"""
def test_start_remediation_mentions_sudo_systemctl_and_uninstall(self, capsys, monkeypatch):
monkeypatch.setattr(gateway_cli, "get_service_name", lambda: "hermes-gateway")
gateway_cli._print_system_scope_remediation("start")
out = capsys.readouterr().out
assert "system-wide service" in out
assert "start requires root" in out
assert "sudo systemctl start hermes-gateway" in out
assert "sudo hermes gateway uninstall --system" in out
assert "hermes gateway install" in out
def test_restart_remediation_uses_systemctl_restart(self, capsys, monkeypatch):
monkeypatch.setattr(gateway_cli, "get_service_name", lambda: "hermes-gateway")
gateway_cli._print_system_scope_remediation("restart")
out = capsys.readouterr().out
assert "restart requires root" in out
assert "sudo systemctl restart hermes-gateway" in out
def test_stop_remediation_uses_systemctl_stop(self, capsys, monkeypatch):
monkeypatch.setattr(gateway_cli, "get_service_name", lambda: "hermes-gateway")
gateway_cli._print_system_scope_remediation("stop")
out = capsys.readouterr().out
assert "stop requires root" in out
assert "sudo systemctl stop hermes-gateway" in out
class TestGatewayCommandCatchesSystemScopeError:
"""The direct CLI path (``hermes gateway start --system`` etc.) must
still exit 1 with a clean message when non-root. The top-level
``gateway_command`` catches ``SystemScopeRequiresRootError`` and
converts it back to ``sys.exit(1)``, preserving existing CLI behavior.
"""
def test_non_root_system_start_exits_one_with_clean_message(self, tmp_path, monkeypatch, capsys):
sys_dir = tmp_path / "sys"
usr_dir = tmp_path / "usr"
sys_dir.mkdir()
usr_dir.mkdir()
(sys_dir / "hermes-gateway.service").write_text("[Unit]\n")
monkeypatch.setattr(
gateway_cli,
"get_systemd_unit_path",
lambda system=False: (sys_dir if system else usr_dir) / "hermes-gateway.service",
)
monkeypatch.setattr(gateway_cli.os, "geteuid", lambda: 1000)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
monkeypatch.setattr(gateway_cli, "kill_gateway_processes", lambda **kw: 0)
args = SimpleNamespace(gateway_command="start", system=True, all=False)
with pytest.raises(SystemExit) as excinfo:
gateway_cli.gateway_command(args)
assert excinfo.value.code == 1
out = capsys.readouterr().out
# Renders the message, NOT the ``('msg', 'action')`` tuple repr
assert "System gateway start requires root. Re-run with sudo." in out
assert "('" not in out # no tuple repr leaking through
+175 -17
View File
@@ -40,14 +40,14 @@ class TestParseJudgeResponse:
def test_clean_json_done(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response('{"done": true, "reason": "all good"}')
done, reason, _ = _parse_judge_response('{"done": true, "reason": "all good"}')
assert done is True
assert reason == "all good"
def test_clean_json_continue(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response('{"done": false, "reason": "more work needed"}')
done, reason, _ = _parse_judge_response('{"done": false, "reason": "more work needed"}')
assert done is False
assert reason == "more work needed"
@@ -55,7 +55,7 @@ class TestParseJudgeResponse:
from hermes_cli.goals import _parse_judge_response
raw = '```json\n{"done": true, "reason": "done"}\n```'
done, reason = _parse_judge_response(raw)
done, reason, _ = _parse_judge_response(raw)
assert done is True
assert "done" in reason
@@ -64,7 +64,7 @@ class TestParseJudgeResponse:
from hermes_cli.goals import _parse_judge_response
raw = 'Looking at this... the agent says X. Verdict: {"done": false, "reason": "partial"}'
done, reason = _parse_judge_response(raw)
done, reason, _ = _parse_judge_response(raw)
assert done is False
assert reason == "partial"
@@ -72,24 +72,24 @@ class TestParseJudgeResponse:
from hermes_cli.goals import _parse_judge_response
for s in ("true", "yes", "done", "1"):
done, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
done, _, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
assert done is True
for s in ("false", "no", "not yet"):
done, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
done, _, _ = _parse_judge_response(f'{{"done": "{s}", "reason": "r"}}')
assert done is False
def test_malformed_json_fails_open(self):
"""Non-JSON → not done, with error-ish reason (so judge_goal can map to continue)."""
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response("this is not json at all")
done, reason, _ = _parse_judge_response("this is not json at all")
assert done is False
assert reason # non-empty
def test_empty_response(self):
from hermes_cli.goals import _parse_judge_response
done, reason = _parse_judge_response("")
done, reason, _ = _parse_judge_response("")
assert done is False
assert reason
@@ -103,13 +103,13 @@ class TestJudgeGoal:
def test_empty_goal_skipped(self):
from hermes_cli.goals import judge_goal
verdict, _ = judge_goal("", "some response")
verdict, _, _ = judge_goal("", "some response")
assert verdict == "skipped"
def test_empty_response_continues(self):
from hermes_cli.goals import judge_goal
verdict, _ = judge_goal("ship the thing", "")
verdict, _, _ = judge_goal("ship the thing", "")
assert verdict == "continue"
def test_no_aux_client_continues(self):
@@ -120,7 +120,7 @@ class TestJudgeGoal:
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(None, None),
):
verdict, _ = goals.judge_goal("my goal", "my response")
verdict, _, _ = goals.judge_goal("my goal", "my response")
assert verdict == "continue"
def test_api_error_continues(self):
@@ -133,7 +133,7 @@ class TestJudgeGoal:
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "response")
verdict, reason, _ = goals.judge_goal("goal", "response")
assert verdict == "continue"
assert "judge error" in reason.lower()
@@ -152,7 +152,7 @@ class TestJudgeGoal:
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "agent response")
verdict, reason, _ = goals.judge_goal("goal", "agent response")
assert verdict == "done"
assert reason == "achieved"
@@ -171,7 +171,7 @@ class TestJudgeGoal:
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, reason = goals.judge_goal("goal", "agent response")
verdict, reason, _ = goals.judge_goal("goal", "agent response")
assert verdict == "continue"
assert reason == "not yet"
@@ -260,7 +260,7 @@ class TestGoalManager:
mgr = GoalManager(session_id="eval-sid-1")
mgr.set("ship it")
with patch.object(goals, "judge_goal", return_value=("done", "shipped")):
with patch.object(goals, "judge_goal", return_value=("done", "shipped", False)):
decision = mgr.evaluate_after_turn("I shipped the feature.")
assert decision["verdict"] == "done"
@@ -276,7 +276,7 @@ class TestGoalManager:
mgr = GoalManager(session_id="eval-sid-2", default_max_turns=5)
mgr.set("a long goal")
with patch.object(goals, "judge_goal", return_value=("continue", "more work")):
with patch.object(goals, "judge_goal", return_value=("continue", "more work", False)):
decision = mgr.evaluate_after_turn("made some progress")
assert decision["verdict"] == "continue"
@@ -294,7 +294,7 @@ class TestGoalManager:
mgr = GoalManager(session_id="eval-sid-3", default_max_turns=2)
mgr.set("hard goal")
with patch.object(goals, "judge_goal", return_value=("continue", "not yet")):
with patch.object(goals, "judge_goal", return_value=("continue", "not yet", False)):
d1 = mgr.evaluate_after_turn("step 1")
assert d1["should_continue"] is True
assert mgr.state.turns_used == 1
@@ -356,3 +356,161 @@ def test_goal_command_dispatches_in_cli_registry_helpers():
assert "/goal" in COMMANDS
session_cmds = COMMANDS_BY_CATEGORY.get("Session", {})
assert "/goal" in session_cmds
# ──────────────────────────────────────────────────────────────────────
# Auto-pause on consecutive judge parse failures
# ──────────────────────────────────────────────────────────────────────
class TestJudgeParseFailureAutoPause:
"""Regression: weak judge models (e.g. deepseek-v4-flash) that return
empty strings or non-JSON prose must auto-pause the loop after N turns
instead of burning the whole turn budget."""
def test_parse_response_flags_empty_as_parse_failure(self):
from hermes_cli.goals import _parse_judge_response
done, reason, parse_failed = _parse_judge_response("")
assert done is False
assert parse_failed is True
assert "empty" in reason.lower()
def test_parse_response_flags_non_json_as_parse_failure(self):
from hermes_cli.goals import _parse_judge_response
done, reason, parse_failed = _parse_judge_response(
"Let me analyze whether the goal is fully satisfied based on the agent's response..."
)
assert done is False
assert parse_failed is True
assert "not json" in reason.lower()
def test_parse_response_clean_json_is_not_parse_failure(self):
from hermes_cli.goals import _parse_judge_response
done, _, parse_failed = _parse_judge_response(
'{"done": false, "reason": "more work"}'
)
assert done is False
assert parse_failed is False
def test_api_error_does_not_count_as_parse_failure(self):
"""Transient network/API errors must not trip the auto-pause guard."""
from hermes_cli import goals
fake_client = MagicMock()
fake_client.chat.completions.create.side_effect = RuntimeError("connection reset")
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, _, parse_failed = goals.judge_goal("goal", "response")
assert verdict == "continue"
assert parse_failed is False
def test_empty_judge_reply_flagged_as_parse_failure(self):
"""End-to-end: judge returns empty content → parse_failed=True."""
from hermes_cli import goals
fake_client = MagicMock()
fake_client.chat.completions.create.return_value = MagicMock(
choices=[MagicMock(message=MagicMock(content=""))]
)
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(fake_client, "judge-model"),
):
verdict, _, parse_failed = goals.judge_goal("goal", "response")
assert verdict == "continue"
assert parse_failed is True
def test_auto_pause_after_three_consecutive_parse_failures(self, hermes_home):
"""N=3 consecutive parse failures → auto-pause with config pointer."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager, DEFAULT_MAX_CONSECUTIVE_PARSE_FAILURES
assert DEFAULT_MAX_CONSECUTIVE_PARSE_FAILURES == 3
mgr = GoalManager(session_id="parse-fail-sid-1", default_max_turns=20)
mgr.set("do a thing")
with patch.object(
goals, "judge_goal", return_value=("continue", "judge returned empty response", True)
):
d1 = mgr.evaluate_after_turn("step 1")
assert d1["should_continue"] is True
assert mgr.state.consecutive_parse_failures == 1
d2 = mgr.evaluate_after_turn("step 2")
assert d2["should_continue"] is True
assert mgr.state.consecutive_parse_failures == 2
d3 = mgr.evaluate_after_turn("step 3")
assert d3["should_continue"] is False
assert d3["status"] == "paused"
assert mgr.state.consecutive_parse_failures == 3
# Message points at the config surface so the user can fix it.
assert "auxiliary" in d3["message"]
assert "goal_judge" in d3["message"]
assert "config.yaml" in d3["message"]
def test_parse_failure_counter_resets_on_good_reply(self, hermes_home):
"""A single good judge reply resets the counter — transient flakes don't pause."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="parse-fail-sid-2", default_max_turns=20)
mgr.set("another goal")
# Two parse failures…
with patch.object(
goals, "judge_goal", return_value=("continue", "not json", True)
):
mgr.evaluate_after_turn("step 1")
mgr.evaluate_after_turn("step 2")
assert mgr.state.consecutive_parse_failures == 2
# …then one clean reply resets the counter.
with patch.object(
goals, "judge_goal", return_value=("continue", "making progress", False)
):
d = mgr.evaluate_after_turn("step 3")
assert d["should_continue"] is True
assert mgr.state.consecutive_parse_failures == 0
def test_parse_failure_counter_not_incremented_by_api_errors(self, hermes_home):
"""API/transport errors must NOT count toward the auto-pause threshold."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager
mgr = GoalManager(session_id="parse-fail-sid-3", default_max_turns=20)
mgr.set("goal")
with patch.object(
goals, "judge_goal", return_value=("continue", "judge error: RuntimeError", False)
):
for _ in range(5):
d = mgr.evaluate_after_turn("still going")
assert d["should_continue"] is True
assert mgr.state.consecutive_parse_failures == 0
assert mgr.state.status == "active"
def test_consecutive_parse_failures_persists_across_goalmanager_reloads(
self, hermes_home
):
"""The counter must be durable so cross-session resumes see it."""
from hermes_cli import goals
from hermes_cli.goals import GoalManager, load_goal
mgr = GoalManager(session_id="parse-fail-sid-4", default_max_turns=20)
mgr.set("persistent goal")
with patch.object(
goals, "judge_goal", return_value=("continue", "empty", True)
):
mgr.evaluate_after_turn("r")
mgr.evaluate_after_turn("r")
reloaded = load_goal("parse-fail-sid-4")
assert reloaded is not None
assert reloaded.consecutive_parse_failures == 2
+55
View File
@@ -286,3 +286,58 @@ def test_run_slash_reassign_with_reclaim_flag(kanban_home):
assert "Reassigned" in out, out
out2 = kc.run_slash(f"show {tid}")
assert "newbie" in out2
# ---------------------------------------------------------------------------
# /kanban specify — slash surface (same entry point CLI + gateway use)
# ---------------------------------------------------------------------------
def test_run_slash_specify_end_to_end(kanban_home, monkeypatch):
"""The /kanban specify slash command routes through run_slash, which
both the interactive CLI and every gateway platform use. This test
covers both surfaces."""
from unittest.mock import MagicMock
# Create a triage task via the same slash surface.
create_out = kc.run_slash("create 'rough idea' --triage")
import re
m = re.search(r"(t_[a-f0-9]+)", create_out)
assert m, f"no task id in: {create_out!r}"
tid = m.group(1)
# Mock the auxiliary client so we don't hit a real provider.
resp = MagicMock()
resp.choices = [MagicMock()]
resp.choices[0].message.content = (
'{"title": "Spec: rough idea", "body": "**Goal**\\nShip it."}'
)
fake_client = MagicMock()
fake_client.chat.completions.create = MagicMock(return_value=resp)
monkeypatch.setattr(
"agent.auxiliary_client.get_text_auxiliary_client",
lambda *a, **kw: (fake_client, "test-model"),
)
# Specify via slash.
out = kc.run_slash(f"specify {tid}")
assert "Specified" in out
assert tid in out
# Task is promoted and retitled.
with kb.connect() as conn:
task = kb.get_task(conn, tid)
assert task.status in {"todo", "ready"}
assert task.title == "Spec: rough idea"
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."""
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()
@@ -90,22 +90,20 @@ def test_spawn_failure_auto_blocks_after_limit(kanban_home, all_assignees_spawna
conn = kb.connect()
try:
tid = kb.create_task(conn, title="x", assignee="worker")
# Three ticks below the default limit (5) → still ready, counter grows.
for i in range(3):
res = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5)
assert tid not in res.auto_blocked
assert kb.DEFAULT_FAILURE_LIMIT == 2
# One default-limit failure → still ready, counter grows.
res1 = kb.dispatch_once(conn, spawn_fn=_bad_spawn)
assert tid not in res1.auto_blocked
task = kb.get_task(conn, tid)
assert task.status == "ready"
assert task.consecutive_failures == 3
assert task.consecutive_failures == 1
# Two more ticks → fifth failure exceeds the limit.
res1 = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5)
assert tid not in res1.auto_blocked
res2 = kb.dispatch_once(conn, spawn_fn=_bad_spawn, failure_limit=5)
# Second default-limit failure trips the guard.
res2 = kb.dispatch_once(conn, spawn_fn=_bad_spawn)
assert tid in res2.auto_blocked
task = kb.get_task(conn, tid)
assert task.status == "blocked"
assert task.consecutive_failures >= 5
assert task.consecutive_failures >= 2
assert task.last_failure_error and "no PATH" in task.last_failure_error
finally:
conn.close()
@@ -170,6 +168,158 @@ def test_successful_completion_resets_failure_counter(kanban_home, all_assignees
conn.close()
def test_reassign_resets_failure_counter_for_new_profile(kanban_home, all_assignees_spawnable):
"""Retry streaks are scoped to a task/profile pair; reassigning is a
human recovery action and gives the new profile a fresh budget."""
conn = kb.connect()
try:
tid = kb.create_task(conn, title="x", assignee="worker")
with kb.write_txn(conn):
conn.execute(
"UPDATE tasks SET consecutive_failures = 1, "
"last_failure_error = 'timed out' WHERE id = ?",
(tid,),
)
assert kb.assign_task(conn, tid, "reviewer") is True
task = kb.get_task(conn, tid)
assert task.assignee == "reviewer"
assert task.consecutive_failures == 0
assert task.last_failure_error is None
finally:
conn.close()
def test_per_task_max_retries_overrides_dispatcher_limit(kanban_home, all_assignees_spawnable):
"""Per-task ``max_retries`` overrides both the caller-supplied
``failure_limit`` (gateway config) and the hardcoded default.
Three-tier resolution order:
1. ``task.max_retries`` (set via ``create_task(max_retries=N)`` /
``hermes kanban create --max-retries N``)
2. ``failure_limit`` kwarg passed by the caller (gateway threads
this from ``kanban.failure_limit`` config)
3. ``DEFAULT_FAILURE_LIMIT``
"""
conn = kb.connect()
try:
# max_retries=1 should trip on the FIRST failure, even though the
# caller is asking for failure_limit=10.
tid = kb.create_task(
conn, title="one-shot", assignee="worker", max_retries=1,
)
task = kb.get_task(conn, tid)
assert task.max_retries == 1, "per-task override must persist"
kb.claim_task(conn, tid)
tripped = kb._record_task_failure(
conn, tid,
error="first fail",
outcome="spawn_failed",
failure_limit=10, # far higher than per-task override
release_claim=True,
end_run=False,
)
assert tripped is True, "should auto-block on first failure"
task = kb.get_task(conn, tid)
assert task.status == "blocked"
assert task.consecutive_failures == 1
# gave_up event should record where the threshold came from
events = kb.list_events(conn, tid)
gave_up = [e for e in events if e.kind == "gave_up"]
assert gave_up, f"expected gave_up event, got {[e.kind for e in events]}"
assert gave_up[-1].payload.get("limit_source") == "task"
assert gave_up[-1].payload.get("effective_limit") == 1
finally:
conn.close()
def test_per_task_max_retries_allows_more_than_default(kanban_home, all_assignees_spawnable):
"""A task with ``max_retries=5`` does NOT auto-block at the default
limit of 2 it must reach the per-task override first."""
conn = kb.connect()
try:
tid = kb.create_task(
conn, title="flaky-retry", assignee="worker", max_retries=5,
)
# Four failures — still below the per-task threshold, should stay ready.
for i in range(1, 5):
kb.claim_task(conn, tid)
tripped = kb._record_task_failure(
conn, tid,
error=f"fail {i}",
outcome="spawn_failed",
# Caller passes the default so the dispatcher tier matches
# ``DEFAULT_FAILURE_LIMIT``; without the per-task override
# the breaker would have tripped at failure 2.
release_claim=True,
end_run=False,
)
assert tripped is False, f"shouldn't trip at failure {i} with max_retries=5"
task = kb.get_task(conn, tid)
assert task.status == "ready", f"at failure {i} status was {task.status}"
# Fifth failure trips the per-task limit.
kb.claim_task(conn, tid)
tripped = kb._record_task_failure(
conn, tid,
error="fail 5",
outcome="spawn_failed",
release_claim=True,
end_run=False,
)
assert tripped is True
task = kb.get_task(conn, tid)
assert task.status == "blocked"
assert task.consecutive_failures == 5
finally:
conn.close()
def test_max_retries_none_falls_through_to_dispatcher_limit(kanban_home, all_assignees_spawnable):
"""``max_retries=None`` (the default) falls through to the caller-
supplied ``failure_limit`` the gateway config tier."""
conn = kb.connect()
try:
tid = kb.create_task(conn, title="standard", assignee="worker")
task = kb.get_task(conn, tid)
assert task.max_retries is None
# Caller passes failure_limit=4 (simulates kanban.failure_limit=4).
# Should trip at 4, not at the DEFAULT_FAILURE_LIMIT of 2.
for i in range(1, 4):
kb.claim_task(conn, tid)
tripped = kb._record_task_failure(
conn, tid,
error=f"fail {i}",
outcome="spawn_failed",
failure_limit=4,
release_claim=True,
end_run=False,
)
assert tripped is False, f"premature trip at failure {i}"
kb.claim_task(conn, tid)
tripped = kb._record_task_failure(
conn, tid,
error="fail 4",
outcome="spawn_failed",
failure_limit=4,
release_claim=True,
end_run=False,
)
assert tripped is True
task = kb.get_task(conn, tid)
assert task.status == "blocked"
events = kb.list_events(conn, tid)
gave_up = [e for e in events if e.kind == "gave_up"]
assert gave_up[-1].payload.get("limit_source") == "dispatcher"
assert gave_up[-1].payload.get("effective_limit") == 4
finally:
conn.close()
def test_workspace_resolution_failure_also_counts(kanban_home, all_assignees_spawnable):
"""`dir:` workspace with no path should fail workspace resolution AND
count against the failure budget not just crash the tick."""
@@ -719,6 +869,48 @@ def test_max_runtime_terminates_overrun_worker(kanban_home):
_kb._pid_alive = original_alive
def test_repeated_timeouts_auto_block_at_default_limit(kanban_home):
"""Two timed_out outcomes on the same task/profile trip the retry guard."""
import hermes_cli.kanban_db as _kb
original_alive = _kb._pid_alive
_kb._pid_alive = lambda pid: False
def _age_active_run(conn, tid):
old_started = int(time.time()) - 30
with kb.write_txn(conn):
conn.execute(
"UPDATE task_runs SET started_at = ? "
"WHERE id = (SELECT current_run_id FROM tasks WHERE id = ?)",
(old_started, tid),
)
try:
conn = kb.connect()
try:
tid = kb.create_task(
conn, title="long job", assignee="worker",
max_runtime_seconds=1,
)
for expected_failures in (1, 2):
kb.claim_task(conn, tid)
kb._set_worker_pid(conn, tid, os.getpid())
_age_active_run(conn, tid)
timed_out = kb.enforce_max_runtime(conn, signal_fn=lambda pid, sig: None)
assert tid in timed_out
task = kb.get_task(conn, tid)
assert task.consecutive_failures == expected_failures
task = kb.get_task(conn, tid)
assert task.status == "blocked"
events = kb.list_events(conn, tid)
assert [e.kind for e in events].count("timed_out") == 2
gave_up = [e for e in events if e.kind == "gave_up"]
assert gave_up and gave_up[-1].payload["trigger_outcome"] == "timed_out"
finally:
conn.close()
finally:
_kb._pid_alive = original_alive
def test_max_runtime_none_means_no_cap(kanban_home):
"""A task with max_runtime_seconds=None is never timed out regardless
of how long it runs."""
@@ -3283,17 +3475,28 @@ def test_complete_prose_scan_ignores_existing_ids(kanban_home):
# Recovery helpers (reclaim + reassign)
# ---------------------------------------------------------------------------
def test_reclaim_task_resets_running_to_ready(kanban_home):
def test_reclaim_task_resets_running_to_ready(kanban_home, monkeypatch):
"""Manual reclaim releases the claim, resets status, and emits a
``reclaimed`` event even when claim_expires has not passed."""
import signal
import time
import secrets
import hermes_cli.kanban_db as _kb
conn = kb.connect()
try:
t = kb.create_task(conn, title="stuck", assignee="broken")
# Simulate a live claim (not expired).
lock = secrets.token_hex(8)
lock = f"{_kb._claimer_id().split(':', 1)[0]}:{secrets.token_hex(8)}"
future = int(time.time()) + 3600
killed: list[int] = []
state = {"alive": True}
def _signal(pid, sig):
killed.append(sig)
if sig == signal.SIGTERM:
state["alive"] = False
monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"])
conn.execute(
"UPDATE tasks SET status='running', claim_lock=?, claim_expires=?, "
"worker_pid=? WHERE id=?",
@@ -3312,7 +3515,7 @@ def test_reclaim_task_resets_running_to_ready(kanban_home):
assert kb.release_stale_claims(conn) == 0
# reclaim_task should work immediately.
assert kb.reclaim_task(conn, t, reason="test reason") is True
assert kb.reclaim_task(conn, t, reason="test reason", signal_fn=_signal) is True
row = conn.execute(
"SELECT status, claim_lock, worker_pid FROM tasks WHERE id=?",
@@ -3333,6 +3536,9 @@ def test_reclaim_task_resets_running_to_ready(kanban_home):
assert len(reclaim_evs) == 1
assert reclaim_evs[0].get("manual") is True
assert reclaim_evs[0].get("reason") == "test reason"
assert reclaim_evs[0].get("termination_attempted") is True
assert reclaim_evs[0].get("terminated") is True
assert killed == [signal.SIGTERM]
finally:
conn.close()
@@ -3561,6 +3767,100 @@ def test_detect_crashed_workers_increments_counter(kanban_home):
conn.close()
def test_detect_crashed_workers_protocol_violation_auto_blocks(kanban_home):
"""A worker that exited rc=0 while its task was still ``running``
is a protocol violation (agent answered conversationally without
calling kanban_complete / kanban_block). Retrying will just loop,
so auto-block immediately instead of waiting for the breaker to
trip at ``DEFAULT_FAILURE_LIMIT``.
Regression test for the respawn-loop-after-completion bug reported
against small local models (gemma4-e2b q4) where the model writes
the answer as plain text and the CLI exits rc=0 cleanly.
"""
import hermes_cli.kanban_db as _kb
conn = kb.connect()
try:
tid = kb.create_task(conn, title="quiet", assignee="worker")
host_prefix = _kb._claimer_id().split(":", 1)[0]
lock = f"{host_prefix}:mock"
kb.claim_task(conn, tid, claimer=lock)
fake_pid = 999998
kb._set_worker_pid(conn, tid, fake_pid)
# Simulate the reap loop having recorded a clean exit for this pid.
# os.W_EXITCODE(status=0, signal=0) == 0 on POSIX.
_kb._record_worker_exit(fake_pid, 0)
# Force liveness check to say "dead" for the fake pid.
original_alive = _kb._pid_alive
_kb._pid_alive = lambda p: False
try:
result_crashed = kb.detect_crashed_workers(conn)
finally:
_kb._pid_alive = original_alive
assert tid in result_crashed, "should be detected as crashed"
task = kb.get_task(conn, tid)
assert task.status == "blocked", (
f"protocol violation should auto-block on first occurrence, "
f"got status={task.status}"
)
assert "kanban_complete" in (task.last_failure_error or ""), (
f"expected protocol-violation message, got {task.last_failure_error!r}"
)
events = kb.list_events(conn, tid)
kinds = [e.kind for e in events]
assert "protocol_violation" in kinds, (
f"expected 'protocol_violation' event, got {kinds}"
)
# The ``crashed`` event would be misleading here — the worker
# didn't crash, it returned 0.
assert "crashed" not in kinds, (
f"should NOT emit 'crashed' event on clean exit, got {kinds}"
)
assert "gave_up" in kinds, (
f"breaker should trip, expected 'gave_up' event, got {kinds}"
)
finally:
conn.close()
def test_detect_crashed_workers_nonzero_exit_uses_default_limit(kanban_home):
"""A worker that exited non-zero (real error / crash) uses the
normal counter path one failure doesn't trip the breaker.
"""
import hermes_cli.kanban_db as _kb
conn = kb.connect()
try:
tid = kb.create_task(conn, title="crashy", assignee="worker")
host_prefix = _kb._claimer_id().split(":", 1)[0]
kb.claim_task(conn, tid, claimer=f"{host_prefix}:mock")
fake_pid = 999997
kb._set_worker_pid(conn, tid, fake_pid)
# W_EXITCODE(1, 0) == 256 — WIFEXITED True, WEXITSTATUS == 1.
_kb._record_worker_exit(fake_pid, 256)
original_alive = _kb._pid_alive
_kb._pid_alive = lambda p: False
try:
kb.detect_crashed_workers(conn)
finally:
_kb._pid_alive = original_alive
task = kb.get_task(conn, tid)
assert task.status == "ready", (
f"single non-zero crash shouldn't auto-block, got {task.status}"
)
assert task.consecutive_failures == 1
events = kb.list_events(conn, tid)
kinds = [e.kind for e in events]
assert "crashed" in kinds
assert "protocol_violation" not in kinds
finally:
conn.close()
def test_reclaim_task_clears_failure_counter(kanban_home):
"""Operator reclaim wipes the counter so the next retry gets a fresh
budget."""
+18 -3
View File
@@ -168,18 +168,33 @@ def test_claim_fails_on_non_ready(kanban_home):
assert kb.claim_task(conn, t) is None
def test_stale_claim_reclaimed(kanban_home):
def test_stale_claim_reclaimed(kanban_home, monkeypatch):
import signal
import hermes_cli.kanban_db as _kb
with kb.connect() as conn:
t = kb.create_task(conn, title="x", assignee="a")
kb.claim_task(conn, t)
host = _kb._claimer_id().split(":", 1)[0]
kb.claim_task(conn, t, claimer=f"{host}:worker")
killed: list[int] = []
state = {"alive": True}
def _signal(pid, sig):
killed.append(sig)
if sig == signal.SIGTERM:
state["alive"] = False
kb._set_worker_pid(conn, t, 12345)
# Rewind claim_expires so it looks stale.
conn.execute(
"UPDATE tasks SET claim_expires = ? WHERE id = ?",
(int(time.time()) - 3600, t),
)
reclaimed = kb.release_stale_claims(conn)
monkeypatch.setattr(_kb, "_pid_alive", lambda _pid: state["alive"])
reclaimed = kb.release_stale_claims(conn, signal_fn=_signal)
assert reclaimed == 1
assert kb.get_task(conn, t).status == "ready"
assert killed == [signal.SIGTERM]
def test_max_runtime_uses_current_run_start_after_retry(kanban_home):
+337
View File
@@ -0,0 +1,337 @@
"""Tests for the specifier module + `hermes kanban specify` CLI surface.
The auxiliary LLM client is mocked these tests don't hit any network or
real provider. They exercise the prompt plumbing, response parsing, DB
writes, and CLI flag surface.
"""
from __future__ import annotations
import argparse
import json as jsonlib
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from hermes_cli import kanban as kanban_cli
from hermes_cli import kanban_db as kb
from hermes_cli import kanban_specify as spec
@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
def _fake_aux_response(content: str):
"""Build a minimal object shaped like an OpenAI chat.completions result.
The specifier only reads ``resp.choices[0].message.content``, so we
avoid importing the openai SDK and build the tree with MagicMock.
"""
resp = MagicMock()
resp.choices = [MagicMock()]
resp.choices[0].message.content = content
return resp
def _mock_client_returning(content: str):
client = MagicMock()
client.chat.completions.create = MagicMock(return_value=_fake_aux_response(content))
return client
def _patch_aux_client(content: str, *, model: str = "test-model"):
"""Patch get_text_auxiliary_client at its source + at the module that
imported it lazily inside specify_task. Both patches are needed
because kanban_specify imports the function inside the function body.
"""
client = _mock_client_returning(content)
return patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(client, model),
), client
# ---------------------------------------------------------------------------
# JSON extraction helpers
# ---------------------------------------------------------------------------
def test_extract_json_blob_handles_plain_json():
raw = '{"title": "T", "body": "B"}'
assert spec._extract_json_blob(raw) == {"title": "T", "body": "B"}
def test_extract_json_blob_handles_fenced_json():
raw = '```json\n{"title": "T", "body": "B"}\n```'
assert spec._extract_json_blob(raw) == {"title": "T", "body": "B"}
def test_extract_json_blob_handles_prose_preamble():
raw = 'Sure! Here you go:\n{"title": "T", "body": "B"}\nThanks.'
assert spec._extract_json_blob(raw) == {"title": "T", "body": "B"}
def test_extract_json_blob_returns_none_for_unparseable():
assert spec._extract_json_blob("no json here") is None
assert spec._extract_json_blob("") is None
assert spec._extract_json_blob("{not: valid}") is None
# ---------------------------------------------------------------------------
# specify_task (module-level entry point)
# ---------------------------------------------------------------------------
def test_specify_task_happy_path(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
content = jsonlib.dumps({
"title": "Refined rough",
"body": "**Goal**\nA concrete goal.",
})
p, _ = _patch_aux_client(content)
with p:
outcome = spec.specify_task(tid, author="ace")
assert outcome.ok is True
assert outcome.task_id == tid
assert outcome.new_title == "Refined rough"
with kb.connect() as conn:
task = kb.get_task(conn, tid)
# Parent-free → recompute_ready promotes to ready.
assert task.status == "ready"
assert task.title == "Refined rough"
assert "**Goal**" in (task.body or "")
def test_specify_task_falls_back_to_body_only_on_bad_json(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="keep title", triage=True)
# Model returned plain markdown, no JSON object.
content = "Goal: Do a thing.\nApproach: Steps here."
p, _ = _patch_aux_client(content)
with p:
outcome = spec.specify_task(tid)
assert outcome.ok is True
with kb.connect() as conn:
t = kb.get_task(conn, tid)
# Title preserved (no JSON with a title key).
assert t.title == "keep title"
# Body replaced with the raw response.
assert "Goal:" in (t.body or "")
def test_specify_task_rejects_non_triage_task(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="ready task")
p, client = _patch_aux_client("unused")
with p:
outcome = spec.specify_task(tid)
assert outcome.ok is False
assert "not in triage" in outcome.reason
# LLM must not be invoked for a non-triage task — fail cheap.
assert client.chat.completions.create.call_count == 0
def test_specify_task_unknown_id(kanban_home):
p, client = _patch_aux_client("unused")
with p:
outcome = spec.specify_task("t_nope")
assert outcome.ok is False
assert "unknown task" in outcome.reason
assert client.chat.completions.create.call_count == 0
def test_specify_task_no_aux_client_configured(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(None, ""),
):
outcome = spec.specify_task(tid)
assert outcome.ok is False
assert "auxiliary client" in outcome.reason
# Task must stay in triage — we never touched it.
with kb.connect() as conn:
assert kb.get_task(conn, tid).status == "triage"
def test_specify_task_llm_api_error_keeps_task_in_triage(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
client = MagicMock()
client.chat.completions.create = MagicMock(side_effect=RuntimeError("429 rate limited"))
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(client, "test-model"),
):
outcome = spec.specify_task(tid)
assert outcome.ok is False
assert "LLM error" in outcome.reason
with kb.connect() as conn:
assert kb.get_task(conn, tid).status == "triage"
def test_specify_task_empty_llm_response(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
p, _ = _patch_aux_client("")
with p:
outcome = spec.specify_task(tid)
assert outcome.ok is False
with kb.connect() as conn:
assert kb.get_task(conn, tid).status == "triage"
def test_list_triage_ids(kanban_home):
with kb.connect() as conn:
a = kb.create_task(conn, title="a", triage=True)
b = kb.create_task(conn, title="b", triage=True, tenant="proj-1")
kb.create_task(conn, title="c") # not triage — excluded
ids_all = spec.list_triage_ids()
assert set(ids_all) == {a, b}
ids_tenant = spec.list_triage_ids(tenant="proj-1")
assert ids_tenant == [b]
# ---------------------------------------------------------------------------
# CLI wiring — argparse + _cmd_specify
# ---------------------------------------------------------------------------
def _run_cli(*argv: str) -> int:
"""Invoke the `hermes kanban …` argparse surface directly."""
root = argparse.ArgumentParser()
subp = root.add_subparsers(dest="cmd")
kanban_cli.build_parser(subp)
ns = root.parse_args(["kanban", *argv])
return kanban_cli.kanban_command(ns)
def test_cli_specify_requires_id_or_all(kanban_home, capsys):
rc = _run_cli("specify")
assert rc == 2
err = capsys.readouterr().err
assert "requires a task id or --all" in err
def test_cli_specify_rejects_both_id_and_all(kanban_home, capsys):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
rc = _run_cli("specify", tid, "--all")
assert rc == 2
err = capsys.readouterr().err
assert "either a task id OR --all" in err
def test_cli_specify_single_id_success(kanban_home, capsys):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
content = jsonlib.dumps({"title": "clean", "body": "body"})
p, _ = _patch_aux_client(content)
with p:
rc = _run_cli("specify", tid)
assert rc == 0
out = capsys.readouterr().out
assert tid in out
assert "→ todo" in out or "-> todo" in out or "" in out
def test_cli_specify_all_success_and_json(kanban_home, capsys):
with kb.connect() as conn:
a = kb.create_task(conn, title="a", triage=True)
b = kb.create_task(conn, title="b", triage=True)
content = jsonlib.dumps({"title": "spec", "body": "body"})
p, _ = _patch_aux_client(content)
with p:
rc = _run_cli("specify", "--all", "--json")
assert rc == 0
lines = [l for l in capsys.readouterr().out.strip().splitlines() if l]
# One JSON object per task + nothing else.
assert len(lines) == 2
parsed = [jsonlib.loads(l) for l in lines]
ids = {row["task_id"] for row in parsed}
assert ids == {a, b}
assert all(row["ok"] for row in parsed)
def test_cli_specify_all_empty_triage_column(kanban_home, capsys):
rc = _run_cli("specify", "--all")
assert rc == 0
assert "No triage tasks" in capsys.readouterr().out
def test_cli_specify_all_returns_1_when_every_task_fails(kanban_home, capsys):
with kb.connect() as conn:
kb.create_task(conn, title="a", triage=True)
kb.create_task(conn, title="b", triage=True)
with patch(
"agent.auxiliary_client.get_text_auxiliary_client",
return_value=(None, ""), # no aux client → every task fails
):
rc = _run_cli("specify", "--all")
assert rc == 1
def test_cli_specify_tenant_filter(kanban_home, capsys):
with kb.connect() as conn:
outside = kb.create_task(conn, title="outside", triage=True)
inside = kb.create_task(
conn, title="inside", triage=True, tenant="proj-a",
)
content = jsonlib.dumps({"title": "spec", "body": "body"})
p, _ = _patch_aux_client(content)
with p:
rc = _run_cli("specify", "--all", "--tenant", "proj-a", "--json")
assert rc == 0
lines = [
jsonlib.loads(l)
for l in capsys.readouterr().out.strip().splitlines()
if l
]
ids = {row["task_id"] for row in lines}
assert ids == {inside}
# The outside task stays in triage.
with kb.connect() as conn:
assert kb.get_task(conn, outside).status == "triage"
# The inside task was promoted.
assert kb.get_task(conn, inside).status in {"todo", "ready"}
def test_cli_specify_author_passed_through(kanban_home, capsys):
with kb.connect() as conn:
tid = kb.create_task(conn, title="rough", triage=True)
content = jsonlib.dumps({"title": "fresh title", "body": "fresh body"})
p, _ = _patch_aux_client(content)
with p:
rc = _run_cli("specify", tid, "--author", "custom-agent")
assert rc == 0
with kb.connect() as conn:
comments = kb.list_comments(conn, tid)
assert comments and comments[0].author == "custom-agent"
+184
View File
@@ -0,0 +1,184 @@
"""Tests for kb.specify_triage_task — the DB-layer atomic promotion
from the triage column to todo. LLM-free by design."""
from __future__ import annotations
from pathlib import Path
import pytest
from hermes_cli import kanban_db as kb
@pytest.fixture
def kanban_home(tmp_path, monkeypatch):
"""Isolated HERMES_HOME with an empty kanban DB."""
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
kb.init_db()
return home
def _create_triage(conn, title="rough idea", body=None, assignee=None):
return kb.create_task(
conn,
title=title,
body=body,
assignee=assignee,
triage=True,
)
def test_specify_promotes_triage_to_todo(kanban_home):
with kb.connect() as conn:
tid = _create_triage(conn, title="rough idea")
assert kb.get_task(conn, tid).status == "triage"
with kb.connect() as conn:
ok = kb.specify_triage_task(
conn,
tid,
title="Refined: rough idea",
body="**Goal**\nDo the thing.",
author="specifier-bot",
)
assert ok is True
with kb.connect() as conn:
task = kb.get_task(conn, tid)
# No parents → recompute_ready should have flipped it past todo to ready.
assert task.status == "ready"
assert task.title == "Refined: rough idea"
assert "**Goal**" in (task.body or "")
def test_specify_with_open_parent_lands_in_todo_not_ready(kanban_home):
# Parent-gated specified tasks must not jump the dispatcher — they go
# to todo and wait for parent completion like any other gated task.
with kb.connect() as conn:
parent = kb.create_task(conn, title="parent work")
child = _create_triage(conn, title="child idea")
kb.link_tasks(conn, parent, child)
# After linking with an open parent, triage status should still be
# 'triage' (linking doesn't touch triage tasks).
assert kb.get_task(conn, child).status == "triage"
with kb.connect() as conn:
ok = kb.specify_triage_task(
conn,
child,
body="full spec",
author="specifier",
)
assert ok is True
with kb.connect() as conn:
t = kb.get_task(conn, child)
# Parent still open → specified child sits in 'todo', not 'ready'.
assert t.status == "todo"
def test_specify_refuses_non_triage_task(kanban_home):
with kb.connect() as conn:
tid = kb.create_task(conn, title="normal task")
assert kb.get_task(conn, tid).status == "ready"
with kb.connect() as conn:
ok = kb.specify_triage_task(conn, tid, body="won't apply")
assert ok is False
with kb.connect() as conn:
# Status unchanged.
assert kb.get_task(conn, tid).status == "ready"
def test_specify_returns_false_for_unknown_id(kanban_home):
with kb.connect() as conn:
ok = kb.specify_triage_task(conn, "t_does_not_exist", body="x")
assert ok is False
def test_specify_rejects_blank_title(kanban_home):
with kb.connect() as conn:
tid = _create_triage(conn, title="rough")
with kb.connect() as conn, pytest.raises(ValueError):
kb.specify_triage_task(conn, tid, title=" ", body="ok")
def test_specify_emits_event(kanban_home):
with kb.connect() as conn:
tid = _create_triage(conn, title="rough")
with kb.connect() as conn:
kb.specify_triage_task(
conn, tid, title="new", body="b", author="ace"
)
with kb.connect() as conn:
events = kb.list_events(conn, tid)
kinds = [e.kind for e in events]
assert "specified" in kinds
# The specified event records which fields actually changed as a
# JSON payload under task_events.payload.
spec_ev = next(e for e in events if e.kind == "specified")
assert spec_ev.payload is not None
fields = spec_ev.payload.get("changed_fields") or []
assert "title" in fields
assert "body" in fields
def test_specify_records_audit_comment_only_when_author_given(kanban_home):
# With author → comment added.
with kb.connect() as conn:
tid1 = _create_triage(conn, title="a")
kb.specify_triage_task(
conn, tid1, title="A-spec", body="b", author="ace"
)
comments1 = kb.list_comments(conn, tid1)
assert len(comments1) == 1
assert "Specified" in comments1[0].body
assert comments1[0].author == "ace"
# Without author → no comment (silent).
with kb.connect() as conn:
tid2 = _create_triage(conn, title="b")
kb.specify_triage_task(conn, tid2, title="B-spec", body="b")
comments2 = kb.list_comments(conn, tid2)
assert comments2 == []
def test_specify_skips_comment_when_nothing_changed(kanban_home):
# Create triage task with title and body already set; pass identical
# values to specify. Should promote to todo but skip audit comment.
with kb.connect() as conn:
tid = _create_triage(conn, title="same", body="same body")
with kb.connect() as conn:
ok = kb.specify_triage_task(
conn,
tid,
title="same",
body="same body",
author="ace",
)
assert ok is True
with kb.connect() as conn:
# Promoted.
assert kb.get_task(conn, tid).status in {"todo", "ready"}
# No audit comment because neither field changed.
assert kb.list_comments(conn, tid) == []
def test_specify_with_only_body_preserves_title(kanban_home):
with kb.connect() as conn:
tid = _create_triage(conn, title="keep this title")
with kb.connect() as conn:
kb.specify_triage_task(conn, tid, body="new body only")
with kb.connect() as conn:
t = kb.get_task(conn, tid)
assert t.title == "keep this title"
assert t.body == "new body only"
def test_specify_second_call_noop_false(kanban_home):
# Promoting twice must not crash and the second call returns False
# because the task is no longer in triage.
with kb.connect() as conn:
tid = _create_triage(conn, title="once")
with kb.connect() as conn:
assert kb.specify_triage_task(conn, tid, body="spec") is True
with kb.connect() as conn:
assert kb.specify_triage_task(conn, tid, body="spec again") is False
@@ -0,0 +1,87 @@
"""Regression test: ``hermes mcp add --command`` must not clobber the
top-level ``args.command`` subparser dest.
The top-level argparse parser uses ``dest="command"`` for its subparsers
(``hermes_cli/_parser.py``). The dispatcher in ``hermes_cli/main.py``
reads ``args.command`` to decide which command to run; if it is ``None``
it falls through to interactive chat.
The ``mcp add`` subparser exposes a ``--command`` flag (the stdio command
for an MCP server, e.g. ``npx``). Without an explicit ``dest=``, argparse
derives the dest from the flag name and writes ``args.command = None``
when the flag is omitted, overwriting the top-level ``"mcp"`` value. As a
result, ``hermes mcp add foo --url ...`` silently launches chat instead
of registering an MCP server.
The fix: declare the flag with ``dest="mcp_command"``. The CLI flag name
is unchanged; only the in-memory attribute moves.
We replicate the relevant parser shape here rather than importing the
real builder, mirroring ``test_argparse_flag_propagation.py`` and
``test_subparser_routing_fallback.py``.
"""
import argparse
def _build_parser():
"""Minimal replica of the slice of the hermes parser that exhibits
the bug: top-level subparsers (dest="command") and ``mcp add`` with
its ``--command`` flag.
"""
parser = argparse.ArgumentParser(prog="hermes")
subparsers = parser.add_subparsers(dest="command")
subparsers.add_parser("chat")
mcp_p = subparsers.add_parser("mcp")
mcp_sub = mcp_p.add_subparsers(dest="mcp_action")
mcp_add = mcp_sub.add_parser("add")
mcp_add.add_argument("name")
mcp_add.add_argument("--url")
mcp_add.add_argument("--command", dest="mcp_command")
return parser
class TestMcpAddCommandDest:
def test_url_invocation_preserves_top_level_command(self):
"""`hermes mcp add foo --url ...` must keep args.command == "mcp".
Before the dest fix this was clobbered to None, sending the
dispatcher into the chat fallback.
"""
parser = _build_parser()
args = parser.parse_args(
["mcp", "add", "foo", "--url", "https://example.com/mcp"]
)
assert args.command == "mcp"
assert args.mcp_action == "add"
assert args.name == "foo"
assert args.url == "https://example.com/mcp"
assert args.mcp_command is None
def test_command_flag_writes_to_mcp_command_dest(self):
"""`--command npx` must populate args.mcp_command, not args.command."""
parser = _build_parser()
args = parser.parse_args(
["mcp", "add", "github", "--command", "npx"]
)
assert args.command == "mcp"
assert args.mcp_command == "npx"
def test_bare_mcp_add_does_not_clobber_command(self):
"""Even without --url or --command, args.command stays "mcp".
Catches the regression at the parser layer regardless of which
transport flag the user passes.
"""
parser = _build_parser()
args = parser.parse_args(["mcp", "add", "foo"])
assert args.command == "mcp"
assert args.mcp_command is None
assert args.url is None
+5 -5
View File
@@ -43,7 +43,7 @@ def _make_args(**kwargs):
defaults = {
"name": "test-server",
"url": None,
"command": None,
"mcp_command": None,
"args": None,
"auth": None,
"preset": None,
@@ -233,7 +233,7 @@ class TestMcpAdd:
cmd_mcp_add(_make_args(
name="github",
command="npx",
mcp_command="npx",
args=["@mcp/github"],
))
out = capsys.readouterr().out
@@ -291,7 +291,7 @@ class TestMcpAdd:
cmd_mcp_add(_make_args(
name="github",
command="npx",
mcp_command="npx",
args=["@mcp/github"],
env=["MY_API_KEY=secret123", "DEBUG=true"],
))
@@ -313,7 +313,7 @@ class TestMcpAdd:
cmd_mcp_add(_make_args(
name="github",
command="npx",
mcp_command="npx",
args=["@mcp/github"],
env=["BAD-NAME=value"],
))
@@ -390,7 +390,7 @@ class TestMcpAdd:
cmd_mcp_add(_make_args(
name="custom",
preset="testmcp",
command="uvx",
mcp_command="uvx",
args=["custom-server"],
))
out = capsys.readouterr().out
@@ -506,3 +506,64 @@ def test_lmstudio_picker_skips_probe_when_not_configured(monkeypatch):
)
assert "base_url" not in captured
def test_custom_providers_uses_live_models_for_multi_model_endpoint(monkeypatch):
"""Custom providers with api_key + base_url should prefer live /models.
Custom providers (section 4 of list_authenticated_providers) point at
gateways like Bifrost that expose hundreds of models. Reading only the
static ``models:`` dict from config.yaml leaves the /model picker with
a stale subset. Live discovery fills the picker with all available
models from the endpoint.
"""
monkeypatch.setattr("agent.models_dev.fetch_models_dev", lambda: {})
monkeypatch.setattr("hermes_cli.providers.HERMES_OVERLAYS", {})
calls = []
def fake_fetch_api_models(api_key, base_url):
calls.append((api_key, base_url))
return ["gateway-model-a", "gateway-model-b", "gateway-model-c"]
monkeypatch.setattr("hermes_cli.models.fetch_api_models", fake_fetch_api_models)
custom_providers = [
{
"name": "my-gateway",
"api_key": "sk-gateway-key",
"base_url": "https://gateway.example.com/v1",
"model": "gateway-model-a",
"models": {
"gateway-model-a": {"context_length": 128000},
"gateway-model-b": {"context_length": 128000},
},
}
]
providers = list_authenticated_providers(
current_provider="openrouter",
current_base_url="https://openrouter.ai/api/v1",
custom_providers=custom_providers,
max_models=50,
)
gateway_prov = next(
(
p
for p in providers
if p.get("api_url") == "https://gateway.example.com/v1"
),
None,
)
assert gateway_prov is not None, "Custom provider group not found in results"
assert calls == [("sk-gateway-key", "https://gateway.example.com/v1")], (
"fetch_api_models must be called with the custom provider's credentials"
)
assert gateway_prov["models"] == [
"gateway-model-a",
"gateway-model-b",
"gateway-model-c",
], "Live models must replace the static subset"
assert gateway_prov["total_models"] == 3
+1
View File
@@ -330,6 +330,7 @@ class TestPluginHooks:
assert "post_api_request" in VALID_HOOKS
assert "transform_terminal_output" in VALID_HOOKS
assert "transform_tool_result" in VALID_HOOKS
assert "transform_llm_output" in VALID_HOOKS
def test_valid_hooks_include_pre_gateway_dispatch(self):
assert "pre_gateway_dispatch" in VALID_HOOKS
+113
View File
@@ -33,6 +33,9 @@ from hermes_cli.profiles import (
generate_zsh_completion,
_get_profiles_root,
_get_default_hermes_home,
seed_profile_skills,
has_bundled_skills_opt_out,
NO_BUNDLED_SKILLS_MARKER,
)
@@ -243,6 +246,116 @@ class TestCreateProfile:
assert (profile_dir / "SOUL.md").exists()
# ===================================================================
# TestNoSkillsOptOut
# ===================================================================
class TestNoSkillsOptOut:
"""Tests for `hermes profile create --no-skills` and the opt-out marker."""
def test_no_skills_writes_marker_and_skips_seeding(self, profile_env):
profile_dir = create_profile("orchestrator", no_alias=True, no_skills=True)
# Marker file is present
marker = profile_dir / NO_BUNDLED_SKILLS_MARKER
assert marker.is_file(), "expected .no-bundled-skills marker in profile root"
assert "--no-skills" in marker.read_text()
# has_bundled_skills_opt_out() agrees
assert has_bundled_skills_opt_out(profile_dir) is True
# skills/ dir exists (profile bootstrapping still creates the dir) but
# contains nothing yet because create_profile itself doesn't seed.
assert (profile_dir / "skills").is_dir()
assert list((profile_dir / "skills").iterdir()) == []
def test_no_skills_conflicts_with_clone(self, profile_env):
with pytest.raises(ValueError, match="mutually exclusive"):
create_profile(
"orchestrator",
no_alias=True,
no_skills=True,
clone_config=True,
)
def test_no_skills_conflicts_with_clone_all(self, profile_env):
with pytest.raises(ValueError, match="mutually exclusive"):
create_profile(
"orchestrator",
no_alias=True,
no_skills=True,
clone_all=True,
)
def test_seed_profile_skills_respects_marker(self, profile_env):
"""seed_profile_skills() must no-op on opted-out profiles even when
called directly (e.g. by `hermes update`'s all-profile sync loop)."""
profile_dir = create_profile("orchestrator", no_alias=True, no_skills=True)
# Call seed_profile_skills() directly — it should NOT invoke subprocess,
# NOT modify the skills/ dir, and return a dict with skipped_opt_out=True.
result = seed_profile_skills(profile_dir, quiet=True)
assert result is not None
assert result.get("skipped_opt_out") is True
assert result.get("copied") == []
# skills/ stays empty — no subprocess ran
assert list((profile_dir / "skills").iterdir()) == []
def test_default_profile_gets_skills_seeded(self, profile_env, monkeypatch):
"""Sanity: without --no-skills, seed_profile_skills() runs the real
subprocess path. Mock the subprocess so the test is hermetic, and
just confirm the marker is NOT checked in the non-opt-out case."""
import subprocess as _sp
profile_dir = create_profile("coder", no_alias=True)
# No marker — not opted out
assert not (profile_dir / NO_BUNDLED_SKILLS_MARKER).exists()
assert has_bundled_skills_opt_out(profile_dir) is False
# Mock subprocess.run to avoid actually running skill sync in tests
calls = []
def fake_run(*args, **kwargs):
calls.append(args)
return _sp.CompletedProcess(
args=args, returncode=0, stdout='{"copied": ["x"]}', stderr=""
)
monkeypatch.setattr("subprocess.run", fake_run)
result = seed_profile_skills(profile_dir, quiet=True)
# Subprocess was invoked (the opt-out branch did NOT short-circuit)
assert len(calls) == 1
assert result == {"copied": ["x"]}
def test_delete_marker_re_enables_seeding(self, profile_env, monkeypatch):
"""Deleting .no-bundled-skills opts the profile back in."""
import subprocess as _sp
profile_dir = create_profile("orchestrator", no_alias=True, no_skills=True)
assert has_bundled_skills_opt_out(profile_dir) is True
# First call: opted out, returns skipped dict without touching subprocess
called = []
monkeypatch.setattr(
"subprocess.run",
lambda *a, **kw: (called.append(a), _sp.CompletedProcess(
args=a, returncode=0, stdout='{"copied": []}', stderr=""
))[1],
)
r1 = seed_profile_skills(profile_dir, quiet=True)
assert r1.get("skipped_opt_out") is True
assert called == []
# Delete marker → next call runs the real path
(profile_dir / NO_BUNDLED_SKILLS_MARKER).unlink()
assert has_bundled_skills_opt_out(profile_dir) is False
r2 = seed_profile_skills(profile_dir, quiet=True)
assert r2 == {"copied": []}
assert len(called) == 1
# ===================================================================
# TestDeleteProfile
# ===================================================================
@@ -72,11 +72,13 @@ def test_redact_secrets_false_in_config_yaml_is_honored(tmp_path):
assert "ENV_VAR=false" in result.stdout
def test_redact_secrets_default_false_when_unset(tmp_path):
"""Without the config key, redaction stays OFF by default.
def test_redact_secrets_default_true_when_unset(tmp_path):
"""Without the config key or env var, redaction is ON by default (#17691).
Secret redaction is opt-in users who want it must set
`security.redact_secrets: true` explicitly (or HERMES_REDACT_SECRETS=true).
Secret redaction is a secure default users who need raw credential
values in tool output (e.g. working on the redactor itself) must set
`security.redact_secrets: false` explicitly (or
`HERMES_REDACT_SECRETS=false`).
"""
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
@@ -107,7 +109,7 @@ def test_redact_secrets_default_false_when_unset(tmp_path):
timeout=30,
)
assert result.returncode == 0, f"probe failed: {result.stderr}"
assert "REDACT_ENABLED=False" in result.stdout
assert "REDACT_ENABLED=True" in result.stdout
def test_redact_secrets_true_in_config_yaml_is_honored(tmp_path):
+45
View File
@@ -88,6 +88,51 @@ def test_auth_spotify_status_command_reports_logged_in(capsys, monkeypatch: pyte
assert "client_id: spotify-client" in output
def test_spotify_logout_does_not_reset_model_provider(
tmp_path,
monkeypatch: pytest.MonkeyPatch,
capsys,
) -> None:
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
config_path = tmp_path / "config.yaml"
config_path.write_text(
"model:\n"
" default: gemini-3-flash\n"
" provider: custom:local\n"
" base_url: http://localhost:11434/v1\n"
" api_key: ${LOCAL_API_KEY}\n",
encoding="utf-8",
)
with auth_mod._auth_store_lock():
store = auth_mod._load_auth_store()
auth_mod._store_provider_state(
store,
"spotify",
{
"client_id": "spotify-client",
"access_token": "access-token",
"refresh_token": "refresh-token",
"expires_at": "2099-01-01T00:00:00+00:00",
},
set_active=False,
)
auth_mod._save_auth_store(store)
auth_mod.logout_command(SimpleNamespace(provider="spotify"))
output = capsys.readouterr().out
assert "Logged out of Spotify." in output
assert "Model provider configuration was unchanged." in output
assert auth_mod.get_provider_auth_state("spotify") is None
assert config_path.read_text(encoding="utf-8") == (
"model:\n"
" default: gemini-3-flash\n"
" provider: custom:local\n"
" base_url: http://localhost:11434/v1\n"
" api_key: ${LOCAL_API_KEY}\n"
)
def test_spotify_interactive_setup_persists_client_id(
tmp_path,
@@ -192,13 +192,19 @@ class TestTencentTokenhubCanonicalProvider:
class TestTencentInOpenRouterAndNous:
"""tencent/hy3-preview:free should appear in OpenRouter and Nous curated lists."""
"""tencent/hy3-preview:free and tencent/hy3-preview should appear in OpenRouter and Nous curated lists."""
def test_in_openrouter_fallback(self):
from hermes_cli.models import OPENROUTER_MODELS
ids = [mid for mid, _ in OPENROUTER_MODELS]
assert "tencent/hy3-preview:free" in ids
def test_paid_in_openrouter_fallback(self):
"""tencent/hy3-preview (paid, no :free suffix) should also be in OpenRouter list."""
from hermes_cli.models import OPENROUTER_MODELS
ids = [mid for mid, _ in OPENROUTER_MODELS]
assert "tencent/hy3-preview" in ids
def test_in_nous_provider_models(self):
from hermes_cli.models import _PROVIDER_MODELS
assert "tencent/hy3-preview" in _PROVIDER_MODELS["nous"]
@@ -420,7 +426,7 @@ class TestTencentTokenhubCLIDispatch:
class TestTencentTokenhubModelCatalogJSON:
"""Verify tencent/hy3-preview:free is present in the website model-catalog.json."""
"""Verify tencent/hy3-preview:free and tencent/hy3-preview are present in the website model-catalog.json."""
def test_in_model_catalog_json(self):
catalog_path = os.path.join(
@@ -445,6 +451,7 @@ class TestTencentTokenhubModelCatalogJSON:
for model in provider_entry.get("models", []):
all_ids.add(model.get("id", ""))
assert "tencent/hy3-preview:free" in all_ids
assert "tencent/hy3-preview" in all_ids
# =============================================================================
+28 -10
View File
@@ -323,15 +323,15 @@ def test_cmd_update_retries_optional_extras_individually_when_all_fails(monkeypa
return SimpleNamespace(stdout="main\n", stderr="", returncode=0)
if cmd == ["git", "rev-list", "HEAD..origin/main", "--count"]:
return SimpleNamespace(stdout="1\n", stderr="", returncode=0)
if cmd == ["git", "pull", "origin", "main"]:
if cmd == ["git", "pull", "--ff-only", "origin", "main"]:
return SimpleNamespace(stdout="Updating\n", stderr="", returncode=0)
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[all]", "--quiet"]:
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[all]"]:
raise CalledProcessError(returncode=1, cmd=cmd)
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".", "--quiet"]:
if cmd == ["/usr/bin/uv", "pip", "install", "-e", "."]:
return SimpleNamespace(returncode=0)
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[matrix]", "--quiet"]:
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[matrix]"]:
raise CalledProcessError(returncode=1, cmd=cmd)
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[mcp]", "--quiet"]:
if cmd == ["/usr/bin/uv", "pip", "install", "-e", ".[mcp]"]:
return SimpleNamespace(returncode=0)
# Catch-all must include stdout/stderr so consumers that parse
# output (e.g. the dashboard-restart `ps -A` scan added in the
@@ -344,10 +344,10 @@ def test_cmd_update_retries_optional_extras_individually_when_all_fails(monkeypa
install_cmds = [c for c in recorded if "pip" in c and "install" in c]
assert install_cmds == [
["/usr/bin/uv", "pip", "install", "-e", ".[all]", "--quiet"],
["/usr/bin/uv", "pip", "install", "-e", ".", "--quiet"],
["/usr/bin/uv", "pip", "install", "-e", ".[matrix]", "--quiet"],
["/usr/bin/uv", "pip", "install", "-e", ".[mcp]", "--quiet"],
["/usr/bin/uv", "pip", "install", "-e", ".[all]"],
["/usr/bin/uv", "pip", "install", "-e", "."],
["/usr/bin/uv", "pip", "install", "-e", ".[matrix]"],
["/usr/bin/uv", "pip", "install", "-e", ".[mcp]"],
]
out = capsys.readouterr().out
@@ -371,7 +371,7 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
return SimpleNamespace(stdout="main\n", stderr="", returncode=0)
if cmd == ["git", "rev-list", "HEAD..origin/main", "--count"]:
return SimpleNamespace(stdout="1\n", stderr="", returncode=0)
if cmd == ["git", "pull", "origin", "main"]:
if cmd == ["git", "pull", "--ff-only", "origin", "main"]:
return SimpleNamespace(stdout="Updating\n", stderr="", returncode=0)
return SimpleNamespace(returncode=0, stdout="", stderr="")
@@ -384,6 +384,24 @@ def test_cmd_update_succeeds_with_extras(monkeypatch, tmp_path):
assert ".[all]" in install_cmds[0]
def test_install_heartbeat_prints_when_dependency_install_is_silent(monkeypatch, capsys):
"""Long quiet installs should emit periodic heartbeat lines."""
def fake_run(cmd, **kwargs):
hermes_main._time.sleep(1.2)
return SimpleNamespace(returncode=0)
monkeypatch.setattr(hermes_main.subprocess, "run", fake_run)
hermes_main._run_install_with_heartbeat(
["uv", "pip", "install", "-e", "."],
heartbeat_interval_seconds=1,
)
out = capsys.readouterr().out
assert "still installing dependencies" in out
# ---------------------------------------------------------------------------
# ff-only fallback to reset --hard on diverged history
# ---------------------------------------------------------------------------
+253 -4
View File
@@ -415,7 +415,13 @@ class TestCmdUpdateLaunchdRestart:
pid=12345,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[12345]), \
# ``find_gateway_pids`` is invoked twice: once to enumerate manual
# PIDs to restart, then again ~3s later by the post-restart survivor
# sweep (#17648). Return the live PID first, then an empty list to
# simulate the process actually exiting after the graceful restart
# — otherwise the sweep would SIGKILL pid 12345 even though graceful
# drain succeeded, and ``kill.assert_not_called()`` would fire.
with patch.object(gateway_cli, "find_gateway_pids", side_effect=[[12345], []]), \
patch.object(gateway_cli, "find_profile_gateway_processes", return_value=[process]), \
patch.object(gateway_cli, "launch_detached_profile_gateway_restart", return_value=True) as restart, \
patch.object(gateway_cli, "_graceful_restart_via_sigusr1", return_value=True) as graceful, \
@@ -453,7 +459,11 @@ class TestCmdUpdateLaunchdRestart:
pid=12345,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[12345]), \
# See note in ``test_update_restarts_profile_manual_gateways``: the
# post-restart survivor sweep (#17648) re-queries ``find_gateway_pids``
# ~3s after the restart attempt. Return ``[]`` on the second call so
# the SIGTERM fallback isn't escalated to SIGKILL by the sweep.
with patch.object(gateway_cli, "find_gateway_pids", side_effect=[[12345], []]), \
patch.object(gateway_cli, "find_profile_gateway_processes", return_value=[process]), \
patch.object(gateway_cli, "launch_detached_profile_gateway_restart", return_value=True) as restart, \
patch.object(gateway_cli, "_graceful_restart_via_sigusr1", return_value=False) as graceful, \
@@ -872,15 +882,25 @@ class TestServicePidExclusion:
launchctl_loaded=True,
)
# Survivor sweep (#17648) re-queries ``find_gateway_pids`` after
# SIGTERM. ``os.kill`` is mocked, so the PID never "dies" — track
# the killed-via-SIGTERM PIDs ourselves and exclude them on later
# calls to simulate the OS reaping the process. Without this the
# sweep escalates with SIGKILL and ``manual_kills == 2`` instead of 1.
_killed_pids: set[int] = set()
def fake_find(exclude_pids=None, all_profiles=False):
_exclude = exclude_pids or set()
_exclude = (exclude_pids or set()) | _killed_pids
return [p for p in [SERVICE_PID, MANUAL_PID] if p not in _exclude]
def fake_kill(pid, _sig):
_killed_pids.add(pid)
with patch.object(
gateway_cli, "_get_service_pids", return_value={SERVICE_PID}
), patch.object(
gateway_cli, "find_gateway_pids", side_effect=fake_find,
), patch("os.kill") as mock_kill:
), patch("os.kill", side_effect=fake_kill) as mock_kill:
cmd_update(mock_args)
captured = capsys.readouterr().out
@@ -1336,3 +1356,232 @@ class TestCmdUpdateLegacyGatewayWarning:
assert "Legacy Hermes gateway" in captured
assert "(system scope)" in captured
assert "sudo" in captured
# ---------------------------------------------------------------------------
# cmd_update — reset-failed precedes systemctl restart on fallback path
# ---------------------------------------------------------------------------
def _systemctl_calls(mock_run, subcommand):
"""Return every subprocess.run call that was `systemctl [--user] <subcommand>`."""
out = []
for call in mock_run.call_args_list:
argv = call.args[0]
joined = " ".join(str(c) for c in argv)
if "systemctl" in joined and subcommand in joined:
out.append(argv)
return out
class TestCmdUpdateResetFailedBeforeRestart:
"""`hermes update` must call `systemctl reset-failed` before every
fallback `systemctl restart` so a systemd-parked `failed` state from
earlier auto-restart crashes (CHDIR, OOM, filesystem race) doesn't
permanently strand the unit.
Mirrors the recovery pattern `hermes gateway restart` (systemd_restart)
adopted in PR #20949. Without this, users hit "gateway never comes
back after update" until they manually run `systemctl reset-failed`.
"""
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_reset_failed_runs_before_fallback_restart(
self, mock_run, _mock_which, mock_args, monkeypatch,
):
"""When SIGUSR1 drain times out, the fallback systemctl restart
MUST be preceded by a `reset-failed` call against the same unit."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
mock_run.side_effect = _make_run_side_effect(
commit_count="3",
systemd_active=True,
)
# Force the graceful SIGUSR1 path to report failure so cmd_update
# falls back to systemctl restart.
orig = mock_run.side_effect
def wrapped(cmd, **kwargs):
joined = " ".join(str(c) for c in cmd)
if "systemctl" in joined and "show" in joined and "MainPID" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="4242\n", stderr="")
return orig(cmd, **kwargs)
mock_run.side_effect = wrapped
monkeypatch.setattr(
"hermes_cli.gateway._graceful_restart_via_sigusr1",
lambda pid, drain_timeout: False,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[]):
cmd_update(mock_args)
reset_calls = _systemctl_calls(mock_run, "reset-failed")
restart_calls = _systemctl_calls(mock_run, "restart")
assert any(
"hermes-gateway" in " ".join(str(c) for c in call)
for call in reset_calls
), (
"Expected `systemctl reset-failed hermes-gateway` before the "
"fallback `systemctl restart`, got reset_calls=%r" % (reset_calls,)
)
assert restart_calls, "Fallback systemctl restart should still run"
# Order check: the first reset-failed must come before the first restart.
first_reset_idx = None
first_restart_idx = None
for idx, call in enumerate(mock_run.call_args_list):
joined = " ".join(str(c) for c in call.args[0])
if "systemctl" in joined and "reset-failed" in joined and first_reset_idx is None:
first_reset_idx = idx
if "systemctl" in joined and "restart" in joined and "hermes-gateway" in joined:
if first_restart_idx is None:
first_restart_idx = idx
assert first_reset_idx is not None and first_restart_idx is not None
assert first_reset_idx < first_restart_idx, (
f"reset-failed (call #{first_reset_idx}) must precede "
f"restart (call #{first_restart_idx}) so the unit isn't "
"blocked by systemd's failed-state backoff."
)
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_reset_failed_also_runs_before_retry_restart(
self, mock_run, _mock_which, mock_args, monkeypatch,
):
"""If the first fallback restart spawns a process that dies
immediately (is-active stays inactive), the retry restart must
ALSO be preceded by a reset-failed otherwise the retry races
the unit's own failed-state transition."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
# is-active toggles:
# first call (discovery / check active) -> "active"
# later calls (post-restart verify) -> "inactive"
# Using a state counter so both the initial check and the verify
# loops behave realistically.
is_active_calls = {"n": 0}
def side_effect(cmd, **kwargs):
joined = " ".join(str(c) for c in cmd)
if "rev-parse" in joined and "--abbrev-ref" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="main\n", stderr="")
if "rev-parse" in joined and "--verify" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
if "rev-list" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="3\n", stderr="")
if "systemctl" in joined and "list-units" in joined:
if "--user" in joined:
return subprocess.CompletedProcess(
cmd, 0,
stdout="hermes-gateway.service loaded active running\n",
stderr="",
)
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
if "systemctl" in joined and "is-active" in joined:
is_active_calls["n"] += 1
# First check: the unit is active (so we enter the restart path).
# Subsequent polling: inactive, which drives the retry branch.
if is_active_calls["n"] == 1:
return subprocess.CompletedProcess(cmd, 0, stdout="active\n", stderr="")
return subprocess.CompletedProcess(cmd, 3, stdout="inactive\n", stderr="")
if "systemctl" in joined and "show" in joined and "MainPID" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="4242\n", stderr="")
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
mock_run.side_effect = side_effect
# Force graceful SIGUSR1 to fail → fallback restart path.
monkeypatch.setattr(
"hermes_cli.gateway._graceful_restart_via_sigusr1",
lambda pid, drain_timeout: False,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[]):
cmd_update(mock_args)
reset_calls = _systemctl_calls(mock_run, "reset-failed")
restart_calls = _systemctl_calls(mock_run, "restart")
# Two restart attempts (initial + retry), two reset-failed calls.
gateway_restarts = [
c for c in restart_calls
if "hermes-gateway" in " ".join(str(a) for a in c)
]
gateway_resets = [
c for c in reset_calls
if "hermes-gateway" in " ".join(str(a) for a in c)
]
assert len(gateway_restarts) >= 2, (
f"Expected both initial + retry restart calls, got {len(gateway_restarts)}"
)
assert len(gateway_resets) >= 2, (
f"Expected reset-failed before BOTH restart attempts, "
f"got {len(gateway_resets)} reset-failed call(s)"
)
@patch("shutil.which", return_value=None)
@patch("subprocess.run")
def test_final_failure_message_tells_user_to_reset_failed(
self, mock_run, _mock_which, mock_args, capsys, monkeypatch,
):
"""When both fallback restart attempts fail, the final error
message must include `systemctl reset-failed` as part of the
manual recovery hint not just `systemctl restart` on its own,
which is the step that just failed twice."""
monkeypatch.setattr(gateway_cli, "is_macos", lambda: False)
monkeypatch.setattr(gateway_cli, "supports_systemd_services", lambda: True)
monkeypatch.setattr(gateway_cli, "is_termux", lambda: False)
is_active_calls = {"n": 0}
def side_effect(cmd, **kwargs):
joined = " ".join(str(c) for c in cmd)
if "rev-parse" in joined and "--abbrev-ref" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="main\n", stderr="")
if "rev-parse" in joined and "--verify" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
if "rev-list" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="3\n", stderr="")
if "systemctl" in joined and "list-units" in joined:
if "--user" in joined:
return subprocess.CompletedProcess(
cmd, 0,
stdout="hermes-gateway.service loaded active running\n",
stderr="",
)
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
if "systemctl" in joined and "is-active" in joined:
is_active_calls["n"] += 1
if is_active_calls["n"] == 1:
return subprocess.CompletedProcess(cmd, 0, stdout="active\n", stderr="")
return subprocess.CompletedProcess(cmd, 3, stdout="inactive\n", stderr="")
if "systemctl" in joined and "show" in joined and "MainPID" in joined:
return subprocess.CompletedProcess(cmd, 0, stdout="4242\n", stderr="")
return subprocess.CompletedProcess(cmd, 0, stdout="", stderr="")
mock_run.side_effect = side_effect
monkeypatch.setattr(
"hermes_cli.gateway._graceful_restart_via_sigusr1",
lambda pid, drain_timeout: False,
)
with patch.object(gateway_cli, "find_gateway_pids", return_value=[]):
cmd_update(mock_args)
captured = capsys.readouterr().out
assert "failed to stay running" in captured, (
"Expected the terminal failure message to fire when both "
f"restart attempts don't survive. Got:\n{captured}"
)
assert "reset-failed" in captured, (
"Final recovery hint must include `reset-failed` so users "
"know how to escape systemd's parked failed state. Got:\n"
f"{captured}"
)
assert "hermes-gateway" in captured
+22 -6
View File
@@ -113,11 +113,18 @@ class TestUpdateYesConfigMigration:
args = SimpleNamespace(yes=False)
with patch("builtins.input", return_value="n") as mock_input, patch(
"hermes_cli.main.sys"
) as mock_sys:
mock_sys.stdin.isatty.return_value = True
mock_sys.stdout.isatty.return_value = True
# Patch ``sys.stdin.isatty`` and ``sys.stdout.isatty`` directly on the
# real ``sys`` module instead of replacing ``hermes_cli.main.sys`` with
# a MagicMock. The MagicMock approach was flaky under ``pytest-xdist``
# — a sibling test that imported ``hermes_cli.main`` first could leave
# a different ``sys`` reference resolved inside the function and the
# mock would never be consulted, with CI then taking the
# "Non-interactive session" branch instead of prompting.
import sys as _sys
with patch("builtins.input", return_value="n") as mock_input, patch.object(
_sys.stdin, "isatty", return_value=True
), patch.object(_sys.stdout, "isatty", return_value=True):
cmd_update(args)
# The user was actually prompted.
assert mock_input.called
@@ -156,7 +163,16 @@ class TestUpdateYesStashRestore:
args = SimpleNamespace(yes=True)
cmd_update(args)
# Force a TTY-shaped session so the autostash-restore branch is
# reachable in CI workers regardless of inherited stdio (matches the
# isatty patching strategy in ``test_no_yes_flag_still_prompts_in_tty``
# — ``patch.object`` on the real streams is robust under xdist).
import sys as _sys
with patch.object(_sys.stdin, "isatty", return_value=True), patch.object(
_sys.stdout, "isatty", return_value=True
):
cmd_update(args)
# _restore_stashed_changes was called, and called with prompt_user=False
# every time (so the user never sees "Restore local changes now?").
+218 -1
View File
@@ -309,6 +309,7 @@ class TestContinuousAPI:
# Isolate from any state left behind by other tests in the session.
monkeypatch.setattr(voice, "_continuous_active", False)
monkeypatch.setattr(voice, "_continuous_stopping", False, raising=False)
monkeypatch.setattr(voice, "_continuous_recorder", None)
assert voice.is_continuous_active() is False
@@ -343,11 +344,20 @@ class TestContinuousAPI:
monkeypatch.setattr(voice, "_continuous_recorder", FakeRecorder())
voice.start_continuous(on_transcript=lambda _t: None)
started = voice.start_continuous(on_transcript=lambda _t: None)
# The guard inside start_continuous short-circuits before rec.start()
assert started is True
assert called["n"] == 0
def test_start_returns_false_while_stopping(self, monkeypatch):
import hermes_cli.voice as voice
monkeypatch.setattr(voice, "_continuous_active", False)
monkeypatch.setattr(voice, "_continuous_stopping", True, raising=False)
assert voice.start_continuous(on_transcript=lambda _t: None) is False
class TestContinuousLoopSimulation:
"""End-to-end simulation of the VAD loop with a fake recorder.
@@ -368,6 +378,8 @@ class TestContinuousLoopSimulation:
monkeypatch.setattr(voice, "_continuous_on_transcript", None)
monkeypatch.setattr(voice, "_continuous_on_status", None)
monkeypatch.setattr(voice, "_continuous_on_silent_limit", None)
monkeypatch.setattr(voice, "_continuous_auto_restart", True, raising=False)
monkeypatch.setattr(voice, "_play_beep", lambda *_, **__: None)
class FakeRecorder:
_silence_threshold = 200
@@ -381,13 +393,20 @@ class TestContinuousLoopSimulation:
self.cancelled = 0
# Preset WAV path returned by stop()
self.next_stop_wav = "/tmp/fake.wav"
self.fail_stop = False
self.fail_next_start = False
def start(self, on_silence_stop=None):
if self.fail_next_start:
self.fail_next_start = False
raise RuntimeError("boom")
self.start_calls += 1
self.last_callback = on_silence_stop
self.is_recording = True
def stop(self):
if self.fail_stop:
raise RuntimeError("stop failed")
self.stopped += 1
self.is_recording = False
return self.next_stop_wav
@@ -433,6 +452,204 @@ class TestContinuousLoopSimulation:
voice.stop_continuous()
def test_auto_restart_false_stops_after_first_transcript(self, fake_recorder, monkeypatch):
import hermes_cli.voice as voice
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": "single shot"},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
transcripts = []
statuses = []
voice.start_continuous(
on_transcript=lambda t: transcripts.append(t),
on_status=lambda s: statuses.append(s),
auto_restart=False,
)
fake_recorder.last_callback()
assert transcripts == ["single shot"]
assert fake_recorder.start_calls == 1
assert statuses == ["listening", "transcribing", "idle"]
assert voice.is_continuous_active() is False
def test_auto_restart_false_retains_silent_strikes_across_starts(
self, fake_recorder, monkeypatch
):
import hermes_cli.voice as voice
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": ""},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
silent_limit_fired = []
for _ in range(3):
voice.start_continuous(
on_transcript=lambda _t: None,
on_silent_limit=lambda: silent_limit_fired.append(True),
auto_restart=False,
)
fake_recorder.last_callback()
assert silent_limit_fired == [True]
assert voice.is_continuous_active() is False
assert fake_recorder.start_calls == 3
def test_force_transcribe_stop_delivers_current_buffer(self, fake_recorder, monkeypatch):
import hermes_cli.voice as voice
class ImmediateThread:
def __init__(self, target, daemon=False):
self.target = target
def start(self):
self.target()
monkeypatch.setattr(voice.threading, "Thread", ImmediateThread)
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": "manual stop"},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
transcripts = []
statuses = []
voice.start_continuous(
on_transcript=lambda t: transcripts.append(t),
on_status=lambda s: statuses.append(s),
)
voice.stop_continuous(force_transcribe=True)
assert fake_recorder.stopped == 1
assert transcripts == ["manual stop"]
assert statuses == ["listening", "transcribing", "idle"]
assert voice.is_continuous_active() is False
def test_force_transcribe_empty_single_shots_hit_silent_limit(
self, fake_recorder, monkeypatch
):
import hermes_cli.voice as voice
class ImmediateThread:
def __init__(self, target, daemon=False):
self.target = target
def start(self):
self.target()
monkeypatch.setattr(voice.threading, "Thread", ImmediateThread)
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": ""},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
silent_limit_fired = []
for _ in range(3):
voice.start_continuous(
on_transcript=lambda _t: None,
on_silent_limit=lambda: silent_limit_fired.append(True),
auto_restart=False,
)
voice.stop_continuous(force_transcribe=True)
assert silent_limit_fired == [True]
assert fake_recorder.stopped == 3
assert voice._continuous_no_speech_count == 0
def test_force_transcribe_valid_single_shot_resets_silent_strikes(
self, fake_recorder, monkeypatch
):
import hermes_cli.voice as voice
class ImmediateThread:
def __init__(self, target, daemon=False):
self.target = target
def start(self):
self.target()
monkeypatch.setattr(voice.threading, "Thread", ImmediateThread)
monkeypatch.setattr(voice, "_continuous_no_speech_count", 2)
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": "manual stop"},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
transcripts = []
silent_limit_fired = []
voice.start_continuous(
on_transcript=lambda t: transcripts.append(t),
on_silent_limit=lambda: silent_limit_fired.append(True),
auto_restart=False,
)
voice.stop_continuous(force_transcribe=True)
assert transcripts == ["manual stop"]
assert silent_limit_fired == []
assert voice._continuous_no_speech_count == 0
def test_force_transcribe_stop_failure_cancels_and_clears_stopping(
self, fake_recorder, monkeypatch
):
import hermes_cli.voice as voice
class ImmediateThread:
def __init__(self, target, daemon=False):
self.target = target
def start(self):
self.target()
monkeypatch.setattr(voice.threading, "Thread", ImmediateThread)
fake_recorder.fail_stop = True
statuses = []
voice.start_continuous(
on_transcript=lambda _t: None,
on_status=lambda s: statuses.append(s),
)
voice.stop_continuous(force_transcribe=True)
assert fake_recorder.cancelled == 1
assert statuses == ["listening", "transcribing", "idle"]
assert voice.is_continuous_active() is False
assert voice._continuous_stopping is False
def test_restart_failure_reports_idle(self, fake_recorder, monkeypatch):
import hermes_cli.voice as voice
monkeypatch.setattr(
voice,
"transcribe_recording",
lambda _p: {"success": True, "transcript": "hello world"},
)
monkeypatch.setattr(voice, "is_whisper_hallucination", lambda _t: False)
statuses = []
voice.start_continuous(on_transcript=lambda _t: None, on_status=statuses.append)
fake_recorder.fail_next_start = True
fake_recorder.last_callback()
assert statuses == ["listening", "transcribing", "idle"]
assert voice.is_continuous_active() is False
def test_silent_limit_halts_loop_after_three_strikes(self, fake_recorder, monkeypatch):
import hermes_cli.voice as voice