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-30 13:19:27 -05:00
157 changed files with 10059 additions and 831 deletions
+39
View File
@@ -39,6 +39,45 @@ def mock_args():
return SimpleNamespace()
class TestCmdUpdatePip:
"""Regression tests for pip-install update flows."""
@patch("shutil.which", return_value="/usr/bin/uv")
@patch("subprocess.run")
def test_update_pip_exports_virtualenv_from_sys_prefix(
self, mock_run, _mock_which, mock_args, monkeypatch
):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(hm.sys, "prefix", "/tmp/hermes-launcher-venv")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
hm._cmd_update_pip(mock_args)
assert mock_run.call_count == 1
assert mock_run.call_args.args[0] == ["/usr/bin/uv", "pip", "install", "--upgrade", "hermes-agent"]
assert mock_run.call_args.kwargs["env"]["VIRTUAL_ENV"] == "/tmp/hermes-launcher-venv"
@patch("shutil.which", return_value="/usr/bin/uv")
@patch("subprocess.run")
def test_update_pip_does_not_export_virtualenv_for_system_python(
self, mock_run, _mock_which, mock_args, monkeypatch
):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(hm.sys, "prefix", "/usr")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
hm._cmd_update_pip(mock_args)
assert mock_run.call_count == 1
assert "env" not in mock_run.call_args.kwargs
class TestCmdUpdateBranchFallback:
"""cmd_update falls back to main when current branch has no remote counterpart."""
@@ -6,25 +6,6 @@ from unittest.mock import patch
from hermes_cli.model_switch import list_authenticated_providers
@patch.dict(os.environ, {"GH_TOKEN": "test-key"}, clear=False)
def test_copilot_picker_keeps_curated_copilot_models_when_live_catalog_unavailable():
with patch("agent.models_dev.fetch_models_dev", return_value={}), \
patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value="gh-token"), \
patch("hermes_cli.models._fetch_github_models", return_value=None):
providers = list_authenticated_providers(current_provider="openrouter", max_models=50)
copilot = next((p for p in providers if p["slug"] == "copilot"), None)
assert copilot is not None
assert "gpt-5.4" in copilot["models"]
assert "claude-sonnet-4.6" in copilot["models"]
assert "claude-sonnet-4" in copilot["models"]
assert "claude-sonnet-4.5" in copilot["models"]
assert "claude-haiku-4.5" in copilot["models"]
assert "gemini-3.1-pro-preview" in copilot["models"]
assert "claude-opus-4.6" not in copilot["models"]
@patch.dict(os.environ, {"GH_TOKEN": "test-key"}, clear=False)
def test_copilot_picker_uses_live_catalog_when_available():
live_models = ["gpt-5.4", "claude-sonnet-4.6", "gemini-3.1-pro-preview"]
@@ -80,6 +80,25 @@ def loopback_app():
web_server.app.state.auth_required = prev_required
@pytest.fixture
def insecure_public_app():
"""web_server.app configured for all-interfaces insecure mode."""
_reset_for_tests()
clear_providers()
prev_host = getattr(web_server.app.state, "bound_host", None)
prev_port = getattr(web_server.app.state, "bound_port", None)
prev_required = getattr(web_server.app.state, "auth_required", None)
web_server.app.state.bound_host = "0.0.0.0"
web_server.app.state.bound_port = 9120
web_server.app.state.auth_required = False
client = TestClient(web_server.app, base_url="http://192.168.0.222:9120")
yield client
_reset_for_tests()
web_server.app.state.bound_host = prev_host
web_server.app.state.bound_port = prev_port
web_server.app.state.auth_required = prev_required
def _logged_in(client: TestClient) -> None:
"""Drive the stub OAuth round trip so the client holds session cookies."""
r1 = client.get("/auth/login?provider=stub", follow_redirects=False)
@@ -143,6 +162,30 @@ class TestWsTicketEndpoint:
# ---------------------------------------------------------------------------
@pytest.fixture
def insecure_explicit_host_app():
"""web_server.app bound to an explicit non-loopback host (--insecure).
Models `--host 100.64.0.10 --insecure` (e.g. a Tailscale IP behind
`tailscale serve`) — a specific address rather than the all-interfaces
0.0.0.0 wildcard.
"""
_reset_for_tests()
clear_providers()
prev_host = getattr(web_server.app.state, "bound_host", None)
prev_port = getattr(web_server.app.state, "bound_port", None)
prev_required = getattr(web_server.app.state, "auth_required", None)
web_server.app.state.bound_host = "100.64.0.10"
web_server.app.state.bound_port = 9119
web_server.app.state.auth_required = False
client = TestClient(web_server.app, base_url="http://100.64.0.10:9119")
yield client
_reset_for_tests()
web_server.app.state.bound_host = prev_host
web_server.app.state.bound_port = prev_port
web_server.app.state.auth_required = prev_required
def _fake_ws(*, query: dict, client_host: str = "127.0.0.1", path: str = "/api/pty"):
"""Build a stand-in for starlette.WebSocket good enough for _ws_auth_ok."""
@@ -281,6 +324,48 @@ class TestWsRequestIsAllowedGated:
ws.headers = {"host": "127.0.0.1:8080"}
assert web_server._ws_request_is_allowed(ws) is True
def test_non_loopback_peer_allowed_in_insecure_public_mode(self, insecure_public_app):
"""`--host 0.0.0.0 --insecure` is an explicit LAN/public opt-in.
Regression coverage for the dashboard `/chat` breakage where the
HTML shell loaded on 9120 but every WebSocket upgrade was rejected
with 403 because the loopback-only peer guard still ran even though
the operator intentionally exposed the dashboard on all interfaces.
"""
ws = _fake_ws(query={}, client_host="192.168.0.55")
ws.headers = {
"host": "192.168.0.222:9120",
"origin": "http://192.168.0.222:9120",
}
assert web_server._ws_request_is_allowed(ws) is True
def test_peer_allowed_on_explicit_non_loopback_bind(self, insecure_explicit_host_app):
"""`--host 100.64.0.10 --insecure` (Tailscale/LAN IP) is an explicit
non-loopback opt-in too — not just the 0.0.0.0 wildcard.
Regression coverage: the merged 0.0.0.0/:: fix did not cover binding
directly to a specific tailnet/LAN address, so `/chat` HTML loaded but
WS upgrades were still rejected by the loopback-only peer guard.
"""
ws = _fake_ws(query={}, client_host="100.64.0.99")
ws.headers = {
"host": "100.64.0.10:9119",
"origin": "http://100.64.0.10:9119",
}
assert web_server._ws_request_is_allowed(ws) is True
def test_rebinding_host_rejected_on_explicit_non_loopback_bind(
self, insecure_explicit_host_app
):
"""Lifting the peer-IP gate for an explicit bind must NOT lift the
DNS-rebinding Host guard: a mismatched Host header is still rejected,
because an explicit non-loopback bind requires an exact Host match in
`_is_accepted_host` (unlike the 0.0.0.0 wildcard, which accepts any).
"""
ws = _fake_ws(query={}, client_host="100.64.0.99")
ws.headers = {"host": "evil.example.com"}
assert web_server._ws_request_is_allowed(ws) is False
def test_host_origin_guard_still_runs_in_gated_mode(self, gated_app):
"""Bypassing the peer-IP check must not bypass the DNS-rebinding
Host header guard — that one still protects against attacker
-13
View File
@@ -80,14 +80,6 @@ class TestGmiConfigRegistry:
class TestGmiModelCatalog:
def test_static_model_fallback_exists(self):
assert "gmi" in _PROVIDER_MODELS
models = _PROVIDER_MODELS["gmi"]
assert "zai-org/GLM-5.1-FP8" in models
assert "deepseek-ai/DeepSeek-V3.2" in models
assert "moonshotai/Kimi-K2.5" in models
assert "anthropic/claude-sonnet-4.6" in models
def test_canonical_provider_entry(self):
slugs = [p.slug for p in CANONICAL_PROVIDERS]
assert "gmi" in slugs
@@ -267,11 +259,6 @@ class TestGmiModelMetadata:
class TestGmiAuxiliary:
def test_aux_default_model(self):
from agent.auxiliary_client import _get_aux_model_for_provider
assert _get_aux_model_for_provider("gmi") == "google/gemini-3.1-flash-lite-preview"
def test_resolve_provider_client_uses_gmi_aux_default(self, monkeypatch):
monkeypatch.setenv("GMI_API_KEY", "gmi-test-key")
+21 -10
View File
@@ -106,20 +106,30 @@ def test_worker_block_on_child_with_done_parents_is_still_sticky(kanban_home: Pa
def test_circuit_breaker_block_still_auto_promotes(kanban_home: Path) -> None:
"""A child that was put into ``blocked`` *without* a worker-issued
``kanban_block`` (e.g. circuit-breaker after repeated spawn
failures, manual DB triage) must still get auto-promoted when its
parents complete — preserves the pre-#28712 recovery semantics."""
``kanban_block`` (e.g. a transient crash, manual DB triage) and whose
``consecutive_failures`` is still *below* the circuit-breaker limit
must get auto-promoted when its parents complete — preserves the
pre-#28712 recovery semantics for genuinely transient failures.
The complementary case — a block whose failure count has *reached*
the limit must stay blocked — is covered by
``test_kanban_db.py::test_recompute_ready_skips_tasks_at_failure_limit``
(#35072). Together they pin the contract: ``recompute_ready`` defers
the give-up decision to the same effective limit the breaker uses, so
the two never disagree.
"""
with kb.connect() as conn:
parent = kb.create_task(conn, title="parent")
child = kb.create_task(conn, title="child", parents=[parent])
kb.complete_task(conn, parent, result="ok")
# Simulate a circuit-breaker / direct triage that flips status
# without emitting a ``blocked`` event — exactly what
# ``_record_task_failure`` does after a ``gave_up``.
# Simulate a transient circuit-breaker / direct triage that flips
# status without emitting a ``blocked`` event — exactly what
# ``_record_task_failure`` does below the limit. One failure is
# under the default limit (2), so recovery is still correct.
conn.execute(
"UPDATE tasks SET status='blocked', consecutive_failures=5, "
"last_failure_error='persistent error' WHERE id=?",
"UPDATE tasks SET status='blocked', consecutive_failures=1, "
"last_failure_error='transient error' WHERE id=?",
(child,),
)
conn.commit()
@@ -128,8 +138,9 @@ def test_circuit_breaker_block_still_auto_promotes(kanban_home: Path) -> None:
assert promoted == 1
task = kb.get_task(conn, child)
assert task.status == "ready"
assert task.consecutive_failures == 0
assert task.last_failure_error is None
# Counter is preserved across recovery (not reset) so the breaker
# can still accumulate if the task keeps failing (#35072).
assert task.consecutive_failures == 1
def test_gave_up_event_alone_does_not_make_block_sticky(kanban_home: Path) -> None:
+150 -6
View File
@@ -307,7 +307,8 @@ def test_recompute_ready_cascades_through_chain(kanban_home):
def test_recompute_ready_promotes_blocked_with_done_parents(kanban_home):
"""blocked tasks with all parents done should be promoted to ready."""
"""blocked tasks with all parents done should be promoted to ready,
unless the circuit-breaker failure limit has been reached."""
with kb.connect() as conn:
parent = kb.create_task(conn, title="parent", assignee="a")
child = kb.create_task(
@@ -316,16 +317,16 @@ def test_recompute_ready_promotes_blocked_with_done_parents(kanban_home):
# Complete the parent
kb.claim_task(conn, parent)
kb.complete_task(conn, parent, result="ok")
# Manually block the child (simulates a worker that failed
# after the parent finished)
# Manually block the child with zero failures (simulates a
# dependency block, not a circuit-breaker block).
conn.execute(
"UPDATE tasks SET status='blocked', consecutive_failures=5, "
"last_failure_error='persistent error' WHERE id=?",
"UPDATE tasks SET status='blocked', consecutive_failures=0, "
"last_failure_error=NULL WHERE id=?",
(child,),
)
conn.commit()
assert kb.get_task(conn, child).status == "blocked"
# recompute_ready should promote blocked → ready and reset failures
# recompute_ready should promote blocked → ready
promoted = kb.recompute_ready(conn)
assert promoted == 1
task = kb.get_task(conn, child)
@@ -815,6 +816,149 @@ def test_unblock_resets_failure_counters(kanban_home):
assert task.last_failure_error is None
def test_recompute_ready_skips_tasks_at_failure_limit(kanban_home):
"""recompute_ready must not auto-recover tasks whose consecutive_failures
has reached the circuit-breaker limit (#35072).
Without this guard, a task that repeatedly exhausts its iteration
budget would cycle forever: block → auto-recover (counter reset)
→ respawn → budget exhausted → block → …
"""
with kb.connect() as conn:
parent = kb.create_task(conn, title="parent", assignee="a")
child = kb.create_task(conn, title="child", assignee="a",
parents=[parent])
# Complete the parent so the child's dependencies are satisfied.
kb.claim_task(conn, parent)
kb.complete_task(conn, parent, summary="done")
# Simulate the child having exhausted its budget twice,
# hitting the default failure limit (2).
kb.claim_task(conn, child)
kb._record_task_failure(
conn, child, error="budget exhausted 1",
outcome="timed_out", release_claim=True, end_run=True,
failure_limit=2,
)
kb._record_task_failure(
conn, child, error="budget exhausted 2",
outcome="timed_out", release_claim=True, end_run=True,
failure_limit=2,
)
task = kb.get_task(conn, child)
assert task.status == "blocked"
assert task.consecutive_failures >= 2
# recompute_ready must NOT promote this task — the circuit
# breaker has tripped and it should stay blocked.
promoted = kb.recompute_ready(conn)
assert promoted == 0
assert kb.get_task(conn, child).status == "blocked"
# Explicit unblock should still work and reset the counter.
assert kb.unblock_task(conn, child)
task = kb.get_task(conn, child)
assert task.status == "ready"
assert task.consecutive_failures == 0
def test_recompute_ready_recovers_below_limit(kanban_home):
"""recompute_ready auto-recovers blocked tasks that haven't hit the
failure limit yet — the counter is preserved across recovery."""
with kb.connect() as conn:
t = kb.create_task(conn, title="task", assignee="a")
kb.claim_task(conn, t)
# One failure, below the default limit of 2.
kb._record_task_failure(
conn, t, error="budget exhausted 1",
outcome="timed_out", release_claim=True, end_run=True,
failure_limit=2,
)
task = kb.get_task(conn, t)
assert task.status == "ready"
assert task.consecutive_failures == 1
# Simulate being blocked by something else (not circuit breaker).
conn.execute(
"UPDATE tasks SET status = 'blocked' WHERE id = ?", (t,),
)
conn.commit()
promoted = kb.recompute_ready(conn)
assert promoted == 1
task = kb.get_task(conn, t)
assert task.status == "ready"
# Counter must be preserved, not reset.
assert task.consecutive_failures == 1
def test_recompute_ready_honours_dispatcher_failure_limit(kanban_home):
"""The guard's effective limit must follow the same resolution order
as the circuit breaker (#35072): per-task max_retries → dispatcher
failure_limit → DEFAULT_FAILURE_LIMIT.
Without threading the dispatcher's ``kanban.failure_limit`` through,
the guard falls back to DEFAULT_FAILURE_LIMIT and disagrees with the
breaker — sticking a task prematurely (config limit > default) or
letting a tripped task escape (config limit < default).
"""
with kb.connect() as conn:
# Config allows MORE retries than the default. A task blocked
# with failures below the configured limit must still recover.
t = kb.create_task(conn, title="lenient", assignee="a")
conn.execute(
"UPDATE tasks SET status='blocked', consecutive_failures=? "
"WHERE id=?",
(kb.DEFAULT_FAILURE_LIMIT, t),
)
conn.commit()
# Default-limit call would stick it (failures >= default).
assert kb.recompute_ready(conn) == 0
assert kb.get_task(conn, t).status == "blocked"
# Dispatcher configured a higher limit → recover, preserve counter.
promoted = kb.recompute_ready(
conn, failure_limit=kb.DEFAULT_FAILURE_LIMIT + 2
)
assert promoted == 1
task = kb.get_task(conn, t)
assert task.status == "ready"
assert task.consecutive_failures == kb.DEFAULT_FAILURE_LIMIT
# Config allows FEWER retries than the default. A task at the
# stricter limit must stay blocked even though it's below default.
t2 = kb.create_task(conn, title="strict", assignee="a")
conn.execute(
"UPDATE tasks SET status='blocked', consecutive_failures=1 "
"WHERE id=?",
(t2,),
)
conn.commit()
# Default-limit (2) would recover it (1 < 2).
# Stricter config limit (1) must keep it blocked (1 >= 1).
assert kb.recompute_ready(conn, failure_limit=1) == 0
assert kb.get_task(conn, t2).status == "blocked"
def test_recompute_ready_per_task_max_retries_overrides_dispatcher(kanban_home):
"""A per-task ``max_retries`` wins over the dispatcher failure_limit,
matching ``_record_task_failure``'s resolution order."""
with kb.connect() as conn:
t = kb.create_task(conn, title="per-task", assignee="a")
# Per-task allows 4 retries; dispatcher config says 2.
conn.execute(
"UPDATE tasks SET status='blocked', consecutive_failures=2, "
"max_retries=4 WHERE id=?",
(t,),
)
conn.commit()
# failures(2) < per-task limit(4) → recover, despite dispatcher=2.
promoted = kb.recompute_ready(conn, failure_limit=2)
assert promoted == 1
task = kb.get_task(conn, t)
assert task.status == "ready"
assert task.consecutive_failures == 2
# ---------------------------------------------------------------------------
# Parent-completion invariant at the claim gate (RCA t_a6acd07d)
# ---------------------------------------------------------------------------
+139
View File
@@ -1,11 +1,74 @@
from __future__ import annotations
import sqlite3
import threading
from pathlib import Path
from hermes_cli import kanban_db as kb
def _make_legacy_db(path: Path) -> None:
"""Write a kanban DB with the pre-AUTOINCREMENT (TEXT PK) schema for the
four tables #35096 affects, keeping every other table current so the
additive-column migration runs cleanly on top.
"""
conn = sqlite3.connect(str(path))
conn.executescript(kb.SCHEMA_SQL)
conn.executescript(
"""
DROP TABLE task_events;
DROP TABLE task_comments;
DROP TABLE task_runs;
DROP TABLE kanban_notify_subs;
CREATE TABLE task_comments (id TEXT PRIMARY KEY, task_id TEXT NOT NULL,
author TEXT NOT NULL, body TEXT NOT NULL, created_at INTEGER NOT NULL);
CREATE TABLE task_events (id TEXT PRIMARY KEY, task_id TEXT NOT NULL,
kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL);
CREATE TABLE task_runs (id TEXT PRIMARY KEY, task_id TEXT NOT NULL,
profile TEXT, status TEXT NOT NULL, started_at INTEGER NOT NULL);
CREATE TABLE kanban_notify_subs (task_id TEXT NOT NULL, platform TEXT NOT NULL,
chat_id TEXT NOT NULL, thread_id TEXT NOT NULL DEFAULT '', user_id TEXT,
created_at INTEGER NOT NULL, last_event_id TEXT,
PRIMARY KEY (task_id, platform, chat_id, thread_id));
"""
)
conn.execute("INSERT INTO tasks (id, title, status, created_at) VALUES ('task-1', 'T', 'done', 1000)")
conn.execute("INSERT INTO task_comments VALUES ('c-1', 'task-1', 'agent', 'hi', 1500)")
conn.execute("INSERT INTO task_events VALUES ('e-1', 'task-1', 'completed', NULL, 2000)")
conn.execute("INSERT INTO task_events VALUES ('e-2', 'task-1', 'blocked', NULL, 2100)")
conn.execute("INSERT INTO task_runs VALUES ('r-1', 'task-1', 'default', 'done', 1000)")
conn.execute(
"INSERT INTO kanban_notify_subs (task_id, platform, chat_id, created_at, last_event_id) "
"VALUES ('task-1', 'telegram', '123', 1000, 'e-1')"
)
conn.commit()
conn.close()
def _setup_home(tmp_path, monkeypatch) -> Path:
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
db_path = kb.kanban_db_path(board="legacy")
db_path.parent.mkdir(parents=True, exist_ok=True)
kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
return db_path
def _table_struct(conn: sqlite3.Connection, table: str):
cols = [
(r["name"], (r["type"] or "").upper(), r["notnull"], r["pk"])
for r in conn.execute(f"PRAGMA table_info({table})")
]
idx = sorted(
r["name"]
for r in conn.execute(f"PRAGMA index_list({table})")
if not r["name"].startswith("sqlite_")
)
return cols, idx
def test_connect_initialization_is_thread_safe(tmp_path, monkeypatch):
home = tmp_path / ".hermes"
home.mkdir()
@@ -36,3 +99,79 @@ def test_connect_initialization_is_thread_safe(tmp_path, monkeypatch):
with kb.connect(board="default") as conn:
cols = {row["name"] for row in conn.execute("PRAGMA table_info(tasks)")}
assert "max_retries" in cols
def test_legacy_text_pk_tables_rebuilt_to_integer_autoincrement(tmp_path, monkeypatch):
"""A pre-AUTOINCREMENT DB is migrated in place: id columns become INTEGER
PKs, ``last_event_id`` becomes INTEGER, data is preserved, and indexes
are recreated (DROP TABLE would otherwise take them down)."""
db_path = _setup_home(tmp_path, monkeypatch)
_make_legacy_db(db_path)
with kb.connect(db_path) as conn:
for table in ("task_events", "task_comments", "task_runs"):
id_col = {r["name"]: r for r in conn.execute(f"PRAGMA table_info({table})")}["id"]
assert id_col["type"].upper() == "INTEGER" and id_col["pk"] == 1
lei = {r["name"]: r for r in conn.execute("PRAGMA table_info(kanban_notify_subs)")}
assert lei["last_event_id"]["type"].upper() == "INTEGER"
# Data preserved across the rebuild.
assert len(conn.execute("SELECT * FROM task_events").fetchall()) == 2
assert conn.execute("SELECT body FROM task_comments").fetchone()["body"] == "hi"
assert len(conn.execute("SELECT * FROM task_runs").fetchall()) == 1
# Non-numeric legacy cursor ("e-1") casts to 0.
assert conn.execute("SELECT last_event_id FROM kanban_notify_subs").fetchone()["last_event_id"] == 0
# Indexes restored, including idx_events_run (added by the additive pass).
indexes = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='index'")}
for name in ("idx_events_task", "idx_events_run", "idx_comments_task",
"idx_runs_task", "idx_runs_status", "idx_notify_task"):
assert name in indexes
# AUTOINCREMENT actually works after the rebuild.
conn.execute("INSERT INTO task_events (task_id, kind, created_at) VALUES ('task-1', 'completed', 3000)")
new_id = conn.execute("SELECT id FROM task_events ORDER BY id DESC LIMIT 1").fetchone()["id"]
assert isinstance(new_id, int) and new_id >= 1
def test_rebuilt_schema_matches_fresh_db(tmp_path, monkeypatch):
"""The rebuilt tables must be structurally identical to a fresh DB, so the
hand-written DDL in ``_REBUILD_SPECS`` can't silently drift from SCHEMA_SQL."""
legacy_path = _setup_home(tmp_path, monkeypatch)
_make_legacy_db(legacy_path)
fresh_path = kb.kanban_db_path(board="fresh")
fresh_path.parent.mkdir(parents=True, exist_ok=True)
kb._INITIALIZED_PATHS.discard(str(fresh_path.resolve()))
with kb.connect(legacy_path) as migrated, kb.connect(fresh_path) as fresh:
for table in ("task_events", "task_comments", "task_runs", "kanban_notify_subs"):
assert _table_struct(migrated, table) == _table_struct(fresh, table)
def test_migration_is_idempotent(tmp_path, monkeypatch):
"""Re-opening an already-migrated DB is a no-op and leaves data intact."""
db_path = _setup_home(tmp_path, monkeypatch)
_make_legacy_db(db_path)
with kb.connect(db_path):
pass
kb._INITIALIZED_PATHS.discard(str(db_path.resolve()))
with kb.connect(db_path) as conn:
id_col = {r["name"]: r for r in conn.execute("PRAGMA table_info(task_events)")}["id"]
assert id_col["type"].upper() == "INTEGER"
assert len(conn.execute("SELECT * FROM task_events").fetchall()) == 2
def test_unseen_events_for_sub_survives_migrated_db(tmp_path, monkeypatch):
"""The crash that motivated #35096 — ``int(None)`` on a NULL cursor — is
gone after migration; the notifier query returns an integer cursor."""
db_path = _setup_home(tmp_path, monkeypatch)
_make_legacy_db(db_path)
with kb.connect(db_path) as conn:
cursor, events = kb.unseen_events_for_sub(
conn, task_id="task-1", platform="telegram", chat_id="123"
)
assert isinstance(cursor, int)
assert isinstance(events, list)
+166
View File
@@ -0,0 +1,166 @@
"""Regression tests for bounded/lazy CLI MCP startup."""
from __future__ import annotations
from argparse import Namespace
import sys
import threading
import time
import types
import pytest
import cli as cli_mod
from hermes_cli import main as main_mod
from hermes_cli import mcp_startup
@pytest.fixture(autouse=True)
def _reset_mcp_startup_state():
saved_started = mcp_startup._mcp_discovery_started
saved_thread = mcp_startup._mcp_discovery_thread
try:
mcp_startup._mcp_discovery_started = False
mcp_startup._mcp_discovery_thread = None
yield
finally:
thread = mcp_startup._mcp_discovery_thread
if thread is not None and thread.is_alive():
thread.join(timeout=1.0)
mcp_startup._mcp_discovery_started = saved_started
mcp_startup._mcp_discovery_thread = saved_thread
def _agent_args(**overrides) -> Namespace:
base = {
"accept_hooks": False,
"command": "chat",
"cron_command": None,
"gateway_command": None,
"mcp_action": None,
"tui": False,
}
base.update(overrides)
return Namespace(**base)
def test_prepare_agent_startup_backgrounds_blocking_mcp_for_chat(monkeypatch):
stop = threading.Event()
calls = {"mcp": 0}
def _blocking_discover():
calls["mcp"] += 1
stop.wait()
monkeypatch.setitem(
sys.modules,
"hermes_cli.plugins",
types.SimpleNamespace(discover_plugins=lambda: None),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.config",
types.SimpleNamespace(
read_raw_config=lambda: {"mcp_servers": {"demo": {"transport": "stdio"}}},
load_config=lambda: {},
),
)
monkeypatch.setitem(
sys.modules,
"agent.shell_hooks",
types.SimpleNamespace(register_from_config=lambda *_a, **_k: None),
)
monkeypatch.setitem(
sys.modules,
"tools.mcp_tool",
types.SimpleNamespace(discover_mcp_tools=_blocking_discover),
)
try:
start = time.monotonic()
main_mod._prepare_agent_startup(_agent_args())
elapsed = time.monotonic() - start
assert elapsed < 0.2
assert calls["mcp"] == 1
assert mcp_startup._mcp_discovery_thread is not None
assert mcp_startup._mcp_discovery_thread.is_alive()
finally:
stop.set()
def test_prepare_agent_startup_skips_mcp_bootstrap_for_tui_chat(monkeypatch):
calls = {"mcp": 0}
monkeypatch.setitem(
sys.modules,
"hermes_cli.plugins",
types.SimpleNamespace(discover_plugins=lambda: None),
)
monkeypatch.setitem(
sys.modules,
"hermes_cli.config",
types.SimpleNamespace(load_config=lambda: {}),
)
monkeypatch.setitem(
sys.modules,
"agent.shell_hooks",
types.SimpleNamespace(register_from_config=lambda *_a, **_k: None),
)
monkeypatch.setitem(
sys.modules,
"tools.mcp_tool",
types.SimpleNamespace(
discover_mcp_tools=lambda: calls.__setitem__("mcp", calls["mcp"] + 1)
),
)
main_mod._prepare_agent_startup(_agent_args(tui=True))
assert calls["mcp"] == 0
assert mcp_startup._mcp_discovery_thread is None
def test_cli_get_tool_definitions_briefly_waits_for_fast_mcp_thread(monkeypatch):
thread = threading.Thread(target=lambda: time.sleep(0.05), daemon=True)
thread.start()
mcp_startup._mcp_discovery_thread = thread
monkeypatch.setitem(
sys.modules,
"model_tools",
types.SimpleNamespace(get_tool_definitions=lambda *_a, **_k: ["ok"]),
)
start = time.monotonic()
result = cli_mod.get_tool_definitions(enabled_toolsets=["web"], quiet_mode=True)
elapsed = time.monotonic() - start
assert result == ["ok"]
assert elapsed >= 0.04
assert not thread.is_alive()
def test_init_agent_waits_for_mcp_discovery_before_agent_build(monkeypatch):
waited = {"done": False}
cli = cli_mod.HermesCLI(compact=True)
cli._session_db = object()
cli._resumed = False
cli.conversation_history = []
cli._install_tool_callbacks = lambda: None
cli._ensure_tirith_security = lambda: None
cli._ensure_runtime_credentials = lambda: True
monkeypatch.setattr(
mcp_startup,
"wait_for_mcp_discovery",
lambda timeout=0.75: waited.__setitem__("done", True),
)
def _fake_agent(*_a, **_k):
assert waited["done"] is True
return types.SimpleNamespace()
monkeypatch.setattr(cli_mod, "AIAgent", _fake_agent)
assert cli._init_agent() is True
@@ -0,0 +1,50 @@
"""Tests for `hermes memory setup [provider]` routing.
The `memory setup` subcommand accepts an optional positional ``provider`` so a
fresh install can configure a specific provider directly (e.g.
``hermes memory setup honcho``) without the interactive picker — which matters
because the per-provider ``hermes <provider>`` subcommand is only registered
once that provider is active.
"""
from types import SimpleNamespace
from unittest.mock import patch
from hermes_cli import memory_setup
class TestMemorySetupProviderRouting:
def test_setup_with_provider_arg_skips_picker(self):
"""`memory setup honcho` routes straight to cmd_setup_provider."""
args = SimpleNamespace(memory_command="setup", provider="honcho")
with patch.object(memory_setup, "cmd_setup_provider") as direct, \
patch.object(memory_setup, "cmd_setup") as picker:
memory_setup.memory_command(args)
direct.assert_called_once_with("honcho")
picker.assert_not_called()
def test_setup_without_provider_runs_picker(self):
"""`memory setup` (no provider) runs the interactive picker."""
args = SimpleNamespace(memory_command="setup", provider=None)
with patch.object(memory_setup, "cmd_setup_provider") as direct, \
patch.object(memory_setup, "cmd_setup") as picker:
memory_setup.memory_command(args)
picker.assert_called_once_with(args)
direct.assert_not_called()
def test_setup_with_missing_provider_attr_runs_picker(self):
"""A SimpleNamespace lacking `provider` must not crash — fall back to picker."""
args = SimpleNamespace(memory_command="setup")
with patch.object(memory_setup, "cmd_setup_provider") as direct, \
patch.object(memory_setup, "cmd_setup") as picker:
memory_setup.memory_command(args)
picker.assert_called_once_with(args)
direct.assert_not_called()
def test_unknown_provider_reports_and_returns_early(self, capsys):
"""An unknown provider name surfaces a helpful message and returns
before any config load/save (the not-found guard precedes those imports)."""
memory_setup.cmd_setup_provider("notaprovider")
out = capsys.readouterr().out
assert "not found" in out
assert "hermes memory setup" in out
-32
View File
@@ -142,10 +142,6 @@ class TestCuratedModelsForProvider:
assert len(models) > 0
assert any("claude" in m[0] for m in models)
def test_zai_returns_glm_models(self):
models = curated_models_for_provider("zai")
assert any("glm" in m[0] for m in models)
def test_unknown_provider_returns_empty(self):
assert curated_models_for_provider("totally-unknown") == []
@@ -199,9 +195,6 @@ class TestProviderModelIds:
def test_unknown_provider_returns_empty(self):
assert provider_model_ids("some-unknown-provider") == []
def test_zai_returns_glm_models(self):
assert "glm-5" in provider_model_ids("zai")
def test_stepfun_prefers_live_catalog(self):
with patch(
"hermes_cli.auth.resolve_api_key_provider_credentials",
@@ -222,31 +215,6 @@ class TestProviderModelIds:
patch("hermes_cli.models._fetch_github_models", return_value=["gpt-5.4", "claude-sonnet-4.6"]):
assert provider_model_ids("copilot-acp") == ["gpt-5.4", "claude-sonnet-4.6"]
def test_copilot_falls_back_to_curated_defaults_without_stale_opus(self):
with patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value="gh-token"), \
patch("hermes_cli.models._fetch_github_models", return_value=None):
ids = provider_model_ids("copilot")
assert "gpt-5.4" in ids
assert "claude-sonnet-4.6" in ids
assert "claude-sonnet-4" in ids
assert "claude-sonnet-4.5" in ids
assert "claude-haiku-4.5" in ids
assert "gemini-3.1-pro-preview" in ids
assert "claude-opus-4.6" not in ids
def test_copilot_acp_falls_back_to_copilot_defaults(self):
with patch("hermes_cli.models._resolve_copilot_catalog_api_key", return_value="gh-token"), \
patch("hermes_cli.models._fetch_github_models", return_value=None):
ids = provider_model_ids("copilot-acp")
assert "gpt-5.4" in ids
assert "claude-sonnet-4.6" in ids
assert "claude-sonnet-4" in ids
assert "gemini-3.1-pro-preview" in ids
assert "copilot-acp" not in ids
assert "claude-opus-4.6" not in ids
# -- fetch_api_models --------------------------------------------------------
-4
View File
@@ -56,10 +56,6 @@ class TestOpenRouterModels:
assert isinstance(mid, str) and len(mid) > 0
assert isinstance(desc, str)
def test_at_least_5_models(self):
"""Sanity check that the models list hasn't been accidentally truncated."""
assert len(OPENROUTER_MODELS) >= 5
class TestFetchOpenRouterModels:
def test_live_fetch_recomputes_free_tags(self, monkeypatch):
@@ -231,3 +231,93 @@ def test_get_gateway_eligible_tools_ignores_quoted_false_opt_in(monkeypatch):
assert "web" in has_direct
assert "web" not in already_managed
assert set(unconfigured) == {"image_gen", "video_gen", "tts", "browser"}
def test_apply_nous_managed_defaults_writes_video_gen_config(monkeypatch):
"""apply_nous_managed_defaults must write video_gen.provider and
video_gen.use_gateway when a Nous subscriber selects video_gen
without a direct FAL_KEY."""
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda **kw: True)
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setattr(ns, "fal_key_is_configured", lambda: False)
monkeypatch.setattr(
ns, "get_nous_portal_account_info",
lambda **kw: _account(logged_in=True, paid=True),
)
config = {"model": {"provider": "nous"}}
changed = ns.apply_nous_managed_defaults(
config, enabled_toolsets=["video_gen"],
)
assert "video_gen" in changed
assert config["video_gen"]["provider"] == "fal"
assert config["video_gen"]["use_gateway"] is True
def test_apply_nous_managed_defaults_writes_image_gen_config(monkeypatch):
"""apply_nous_managed_defaults must write image_gen.use_gateway
when a Nous subscriber selects image_gen without a direct FAL_KEY."""
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda **kw: True)
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setattr(ns, "fal_key_is_configured", lambda: False)
monkeypatch.setattr(
ns, "get_nous_portal_account_info",
lambda **kw: _account(logged_in=True, paid=True),
)
config = {"model": {"provider": "nous"}}
changed = ns.apply_nous_managed_defaults(
config, enabled_toolsets=["image_gen"],
)
assert "image_gen" in changed
assert config["image_gen"]["use_gateway"] is True
def test_apply_nous_managed_defaults_skips_fal_tools_when_key_present(monkeypatch):
"""When FAL_KEY is set, apply_nous_managed_defaults should not touch
image_gen or video_gen config the user's direct key takes precedence."""
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda **kw: True)
monkeypatch.setenv("FAL_KEY", "fal-direct-key")
monkeypatch.setattr(ns, "fal_key_is_configured", lambda: True)
monkeypatch.setattr(
ns, "get_nous_portal_account_info",
lambda **kw: _account(logged_in=True, paid=True),
)
config = {"model": {"provider": "nous"}}
changed = ns.apply_nous_managed_defaults(
config, enabled_toolsets=["image_gen", "video_gen"],
)
assert "image_gen" not in changed
assert "video_gen" not in changed
assert "image_gen" not in config
assert "video_gen" not in config
def test_apply_nous_managed_defaults_preserves_existing_video_gen_section(monkeypatch):
"""When video_gen config already exists as a dict, the function should
update it in-place rather than replacing it."""
monkeypatch.setattr(ns, "managed_nous_tools_enabled", lambda **kw: True)
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setattr(ns, "fal_key_is_configured", lambda: False)
monkeypatch.setattr(
ns, "get_nous_portal_account_info",
lambda **kw: _account(logged_in=True, paid=True),
)
config = {
"model": {"provider": "nous"},
"video_gen": {"model": "pixverse-v6"},
}
changed = ns.apply_nous_managed_defaults(
config, enabled_toolsets=["video_gen"],
)
assert "video_gen" in changed
assert config["video_gen"]["provider"] == "fal"
assert config["video_gen"]["use_gateway"] is True
# Pre-existing keys should be preserved
assert config["video_gen"]["model"] == "pixverse-v6"
@@ -495,12 +495,3 @@ class TestOllamaCloudSuffixStripping:
assert _strip_ollama_cloud_suffix("qwen3-coder:480b-cloud") == "qwen3-coder:480b"
assert _strip_ollama_cloud_suffix("nemotron-3-nano:30b") == "nemotron-3-nano:30b"
assert _strip_ollama_cloud_suffix("") == ""
# ── Auxiliary Model ──
class TestOllamaCloudAuxiliary:
def test_aux_model_defined(self):
from agent.auxiliary_client import _API_KEY_PROVIDER_AUX_MODELS
assert "ollama-cloud" in _API_KEY_PROVIDER_AUX_MODELS
assert _API_KEY_PROVIDER_AUX_MODELS["ollama-cloud"] == "nemotron-3-nano:30b"
+22 -2
View File
@@ -48,12 +48,32 @@ def test_stamp_file_takes_precedence(tmp_path):
assert detect_install_method(project_root=tmp_path) == "docker"
def test_docker_detected_via_dockerenv(tmp_path):
def test_container_without_stamp_is_not_docker(tmp_path):
"""An unstamped install in a generic container must NOT be flagged as docker.
Regression for issue #34397. The two supported installs both stamp
``.install_method`` (the curl installer -> ``git``, covered by
``test_stamp_file_takes_precedence``; the published image -> ``docker``),
so neither hits this path. An unsupported manual install dropped into a
container has no stamp and was wrongly classified as the published Docker
image, so ``hermes update`` refused to run. With a ``.git`` checkout it
must resolve to ``git``.
"""
(tmp_path / ".git").mkdir()
with patch("hermes_cli.config.get_managed_system", return_value=None), \
patch("hermes_cli.config.get_hermes_home", return_value=tmp_path), \
patch("hermes_constants.is_container", return_value=True):
from hermes_cli.config import detect_install_method
assert detect_install_method(project_root=tmp_path) == "docker"
assert detect_install_method(project_root=tmp_path) == "git"
def test_container_pip_install_without_stamp_is_pip(tmp_path):
"""Container + no .git + no stamp -> pip, not docker (issue #34397)."""
with patch("hermes_cli.config.get_managed_system", return_value=None), \
patch("hermes_cli.config.get_hermes_home", return_value=tmp_path), \
patch("hermes_constants.is_container", return_value=True):
from hermes_cli.config import detect_install_method
assert detect_install_method(project_root=tmp_path) == "pip"
def test_recommended_update_command_docker():
+6 -6
View File
@@ -754,8 +754,8 @@ class TestRenameProfile:
cfg = json.loads(honcho_path.read_text())
assert "hermes.ssi_health" not in cfg["hosts"]
assert cfg["hosts"]["hermes.heimdall"]["aiPeer"] == "ssi_health"
assert cfg["hosts"]["hermes.heimdall"]["peerName"] == "user-peer"
assert cfg["hosts"]["hermes_heimdall"]["aiPeer"] == "ssi_health"
assert cfg["hosts"]["hermes_heimdall"]["peerName"] == "user-peer"
def test_pins_ai_peer_when_absent_on_honcho_host_rename(self, profile_env):
tmp_path = profile_env
@@ -772,8 +772,8 @@ class TestRenameProfile:
cfg = json.loads(honcho_path.read_text())
assert "hermes.ssi_health" not in cfg["hosts"]
assert cfg["hosts"]["hermes.heimdall"]["aiPeer"] == "ssi_health"
assert cfg["hosts"]["hermes.heimdall"]["workspace"] == "hermes"
assert cfg["hosts"]["hermes_heimdall"]["aiPeer"] == "ssi_health"
assert cfg["hosts"]["hermes_heimdall"]["workspace"] == "hermes"
def test_does_not_overwrite_existing_honcho_host_on_rename(self, profile_env):
tmp_path = profile_env
@@ -782,7 +782,7 @@ class TestRenameProfile:
honcho_path.write_text(json.dumps({
"hosts": {
"hermes.ssi_health": {"aiPeer": "ssi_health"},
"hermes.heimdall": {"aiPeer": "heimdall"},
"hermes_heimdall": {"aiPeer": "heimdall"},
}
}))
@@ -791,7 +791,7 @@ class TestRenameProfile:
cfg = json.loads(honcho_path.read_text())
assert cfg["hosts"]["hermes.ssi_health"]["aiPeer"] == "ssi_health"
assert cfg["hosts"]["hermes.heimdall"]["aiPeer"] == "heimdall"
assert cfg["hosts"]["hermes_heimdall"]["aiPeer"] == "heimdall"
def test_default_raises_value_error(self, profile_env):
with pytest.raises(ValueError, match="default"):
+118
View File
@@ -0,0 +1,118 @@
"""Tests for the ``hermes prompt-size`` diagnostic (issue #34667)."""
import json
import pytest
from hermes_cli.prompt_size import (
_SKILLS_BLOCK_RE,
compute_prompt_breakdown,
render_breakdown,
)
def _seed_memory(hermes_home, memory_text="", user_text=""):
mem_dir = hermes_home / "memories"
mem_dir.mkdir(parents=True, exist_ok=True)
if memory_text:
(mem_dir / "MEMORY.md").write_text(memory_text, encoding="utf-8")
if user_text:
(mem_dir / "USER.md").write_text(user_text, encoding="utf-8")
def _seed_skill(hermes_home, name, description):
skill_dir = hermes_home / "skills" / "demo" / name
skill_dir.mkdir(parents=True, exist_ok=True)
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: {description}\n---\n# {name}\nbody\n",
encoding="utf-8",
)
@pytest.fixture
def isolated_home(tmp_path, monkeypatch):
hermes_home = tmp_path / ".hermes"
hermes_home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(hermes_home))
monkeypatch.chdir(tmp_path) # avoid picking up the repo's AGENTS.md
return hermes_home
def test_breakdown_keys_and_shape(isolated_home):
"""The breakdown exposes every documented key with int byte/char counts."""
data = compute_prompt_breakdown("cli")
assert set(data) >= {
"platform",
"model",
"system_prompt",
"skills_index",
"memory",
"user_profile",
"tools",
"sections",
}
assert data["platform"] == "cli"
for key in ("system_prompt", "skills_index", "memory", "user_profile"):
assert data[key]["bytes"] >= 0
assert data[key]["chars"] >= 0
assert data["tools"]["count"] >= 0
assert data["tools"]["json_bytes"] >= 0
# System prompt is non-trivial even with empty home (identity + guidance).
assert data["system_prompt"]["bytes"] > 0
def test_runs_offline_without_credentials(isolated_home, monkeypatch):
"""No provider credentials configured → still produces a breakdown."""
for var in ("OPENROUTER_API_KEY", "OPENAI_API_KEY", "NOUS_API_KEY",
"ANTHROPIC_API_KEY"):
monkeypatch.delenv(var, raising=False)
data = compute_prompt_breakdown("cli")
assert data["system_prompt"]["bytes"] > 0
def test_skills_index_reflects_installed_skills(isolated_home):
"""Installing a skill makes the skills-index block non-empty.
Note: the skills prompt is cached per-process (in-process LRU + disk
snapshot), so we seed the skill BEFORE the first build rather than
comparing before/after within one process.
"""
_seed_skill(isolated_home, "hello", "a demo skill for size testing")
data = compute_prompt_breakdown("cli")
assert data["skills_index"]["bytes"] > 0
def test_memory_and_profile_are_attributed(isolated_home):
"""Memory and user-profile blocks are measured separately."""
_seed_memory(
isolated_home,
memory_text="Project uses pytest.\n",
user_text="User is a developer.\n",
)
data = compute_prompt_breakdown("cli")
assert data["memory"]["bytes"] > 0
assert data["user_profile"]["bytes"] > 0
def test_skills_block_regex_matches_tagged_block():
text = "preamble\n<available_skills>\n cat:\n - a: b\n</available_skills>\ntail"
m = _SKILLS_BLOCK_RE.search(text)
assert m is not None
assert m.group(0).startswith("<available_skills>")
assert m.group(0).endswith("</available_skills>")
def test_render_breakdown_is_plain_text(isolated_home):
data = compute_prompt_breakdown("cli")
out = render_breakdown(data)
assert "System prompt total" in out
assert "skills index" in out
assert "Tool schemas" in out
# Plain text — no JSON braces leaking in.
assert not out.strip().startswith("{")
def test_json_serializable(isolated_home):
data = compute_prompt_breakdown("cli")
# Round-trips cleanly for ``--json`` output.
assert json.loads(json.dumps(data)) == json.loads(json.dumps(data))
+118
View File
@@ -0,0 +1,118 @@
"""Tests for provider-group folding (display-only picker grouping).
These are invariant tests, not catalog snapshots: they assert how
``group_providers`` folds a flat slug list and how member slugs relate to
``PROVIDER_GROUPS`` / ``CANONICAL_PROVIDERS`` not the specific set of
vendors, which is expected to change over time.
"""
from hermes_cli.models import (
CANONICAL_PROVIDERS,
PROVIDER_GROUPS,
group_providers,
provider_group_for_slug,
)
def _slugs(rows):
"""Flatten picker rows back to the concrete slugs they expose."""
out = []
for r in rows:
if r["kind"] == "single":
out.append(r["slug"])
else:
out.extend(r["members"])
return out
def test_groups_reference_real_canonical_slugs():
"""Every group member must be an actual provider slug. Guards typos and
stale group entries after a provider is renamed/removed."""
canonical = {p.slug for p in CANONICAL_PROVIDERS}
for gid, (label, members) in PROVIDER_GROUPS.items():
assert label, f"group {gid} has empty label"
assert len(members) >= 1
for m in members:
assert m in canonical, f"group {gid} member {m!r} is not a canonical slug"
def test_member_slugs_are_unique_across_groups():
"""A slug may belong to at most one group."""
seen = {}
for gid, (_label, members) in PROVIDER_GROUPS.items():
for m in members:
assert m not in seen, f"{m!r} in both {seen[m]!r} and {gid!r}"
seen[m] = gid
def test_reverse_index_matches_groups():
for gid, (_label, members) in PROVIDER_GROUPS.items():
for m in members:
assert provider_group_for_slug(m) == gid
assert provider_group_for_slug("openrouter") == ""
assert provider_group_for_slug("") == ""
def test_ungrouped_providers_pass_through_in_order():
rows = group_providers(["nous", "openrouter", "deepseek"])
assert all(r["kind"] == "single" for r in rows)
assert [r["slug"] for r in rows] == ["nous", "openrouter", "deepseek"]
def test_multi_member_group_folds_to_one_row():
rows = group_providers(["minimax", "minimax-oauth", "minimax-cn"])
assert len(rows) == 1
row = rows[0]
assert row["kind"] == "group"
assert row["group_id"] == "minimax"
assert row["members"] == ["minimax", "minimax-oauth", "minimax-cn"]
def test_group_appears_at_first_member_position():
"""The group row takes the slot of its earliest-listed present member,
and later members do not re-emit."""
rows = group_providers(["nous", "minimax", "deepseek", "minimax-cn"])
kinds = [(r["kind"], r.get("group_id") or r.get("slug")) for r in rows]
assert kinds == [
("single", "nous"),
("group", "minimax"),
("single", "deepseek"),
]
# both minimax members folded into the single group row
assert rows[1]["members"] == ["minimax", "minimax-cn"]
def test_single_present_member_degrades_to_single_row():
"""A group with only one present member shows no submenu."""
rows = group_providers(["xai"]) # xai-oauth absent
assert len(rows) == 1
assert rows[0]["kind"] == "single"
assert rows[0]["slug"] == "xai"
def test_member_order_follows_declaration_not_input():
"""Inside a folded group, members are ordered by PROVIDER_GROUPS, not by
the order they appeared in the input list."""
rows = group_providers(["minimax-cn", "minimax", "minimax-oauth"])
assert rows[0]["members"] == ["minimax", "minimax-oauth", "minimax-cn"]
def test_duplicate_slugs_ignored():
rows = group_providers(["nous", "nous", "minimax", "minimax"])
assert [r.get("slug") or r["group_id"] for r in rows] == ["nous", "minimax"]
def test_fold_is_lossless_for_present_slugs():
"""Every input slug (deduped) must still be reachable through the folded
rows grouping hides nothing."""
flat = [p.slug for p in CANONICAL_PROVIDERS]
rows = group_providers(flat)
assert set(_slugs(rows)) == set(flat)
def test_canonical_fold_row_count_shrinks():
"""Folding the full canonical list produces fewer top-level rows than the
flat list (proves grouping actually consolidates)."""
flat = [p.slug for p in CANONICAL_PROVIDERS]
rows = group_providers(flat)
assert len(rows) < len(flat)
+60
View File
@@ -757,8 +757,68 @@ def test_first_install_nous_auto_configures_managed_defaults(monkeypatch):
assert config["web"]["backend"] == "firecrawl"
assert config["tts"]["provider"] == "openai"
assert config["browser"]["cloud_provider"] == "browser-use"
assert config["image_gen"]["use_gateway"] is True
assert configured == []
def test_first_install_nous_auto_configures_video_gen(monkeypatch):
"""When a Nous subscriber checks video_gen in the toolset checklist,
apply_nous_managed_defaults must write video_gen.provider and
video_gen.use_gateway so the FAL plugin can route through the gateway
at runtime. Regression test for the bug where video_gen was marked as
auto-configured but no config was actually written."""
monkeypatch.setattr("hermes_cli.nous_subscription.managed_nous_tools_enabled", lambda: True)
config = {
"model": {"provider": "nous"},
"platform_toolsets": {"cli": []},
}
for env_var in (
"VOICE_TOOLS_OPENAI_KEY",
"OPENAI_API_KEY",
"ELEVENLABS_API_KEY",
"FIRECRAWL_API_KEY",
"FIRECRAWL_API_URL",
"TAVILY_API_KEY",
"PARALLEL_API_KEY",
"BROWSERBASE_API_KEY",
"BROWSERBASE_PROJECT_ID",
"BROWSER_USE_API_KEY",
"FAL_KEY",
):
monkeypatch.delenv(env_var, raising=False)
monkeypatch.setattr(
"hermes_cli.tools_config._prompt_toolset_checklist",
lambda *args, **kwargs: {"video_gen"},
)
monkeypatch.setattr("hermes_cli.tools_config.save_config", lambda config: None)
monkeypatch.setattr(
"hermes_cli.tools_config._get_enabled_platforms",
lambda: ["cli"],
)
monkeypatch.setattr(
"hermes_cli.nous_subscription.get_nous_portal_account_info",
lambda *args, **kwargs: NousPortalAccountInfo(
logged_in=True,
source="jwt",
fresh=False,
paid_service_access=True,
),
)
configured = []
monkeypatch.setattr(
"hermes_cli.tools_config._configure_toolset",
lambda ts_key, config: configured.append(ts_key),
)
tools_command(first_install=True, config=config)
assert config["video_gen"]["provider"] == "fal"
assert config["video_gen"]["use_gateway"] is True
# video_gen should NOT appear in the manual configure list — it's auto-configured
assert "video_gen" not in configured
# ── Platform / toolset consistency ────────────────────────────────────────────
+54
View File
@@ -638,6 +638,60 @@ def test_oneshot_rejects_invalid_only_toolsets(monkeypatch, capsys):
assert "did not contain any valid toolsets" in err
def test_oneshot_fails_closed_on_empty_final_response(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.oneshot as oneshot_mod
monkeypatch.setattr(oneshot_mod, "_run_agent", lambda *_args, **_kwargs: "")
assert oneshot_mod.run_oneshot("hello") == 1
captured = capsys.readouterr()
assert captured.out == ""
assert "no final response" in captured.err
def test_oneshot_prints_nonempty_final_response(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.oneshot as oneshot_mod
monkeypatch.setattr(oneshot_mod, "_run_agent", lambda *_args, **_kwargs: "done")
assert oneshot_mod.run_oneshot("hello") == 0
captured = capsys.readouterr()
assert captured.out == "done\n"
assert captured.err == ""
def test_oneshot_fails_closed_on_agent_exception(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.oneshot as oneshot_mod
def _boom(*_args, **_kwargs):
raise OSError("not a TTY")
monkeypatch.setattr(oneshot_mod, "_run_agent", _boom)
assert oneshot_mod.run_oneshot("hello") == 1
captured = capsys.readouterr()
assert captured.out == ""
assert "agent failed" in captured.err
assert "not a TTY" in captured.err
def test_oneshot_reraises_keyboard_interrupt(monkeypatch):
_stub_plugin_discovery(monkeypatch)
import hermes_cli.oneshot as oneshot_mod
import pytest as _pytest
def _interrupt(*_args, **_kwargs):
raise KeyboardInterrupt
monkeypatch.setattr(oneshot_mod, "_run_agent", _interrupt)
with _pytest.raises(KeyboardInterrupt):
oneshot_mod.run_oneshot("hello")
def test_oneshot_filters_invalid_toolsets_before_redirect(monkeypatch, capsys):
_stub_plugin_discovery(monkeypatch)
from hermes_cli.oneshot import _validate_explicit_toolsets
@@ -128,24 +128,31 @@ def test_detect_concurrent_is_noop_off_windows(_winp, tmp_path):
def _fake_psutil_with_parent_chain(
parent_chain: list[int],
proc_iter_rows: list,
*,
ancestor_exe: str | None = None,
):
"""Build a psutil stand-in that has Process()/parent() AND process_iter().
"""Build a psutil stand-in that has Process()/parents()/exe() AND process_iter().
``parent_chain`` is the list of PIDs returned by successive ``.parent()``
calls starting from the seed (``os.getpid()``); the last entry's
``.parent()`` returns ``None`` to terminate the walk.
``parent_chain`` is the ordered list of ancestor PIDs (closest first)
returned by ``proc.parents()`` on the seed (``os.getpid()``).
``ancestor_exe`` is the executable path reported by each ancestor's
``.exe()``; when it matches one of our shim paths the ancestor is
excluded (the launcher-shim case). Pass ``None`` to model an ancestor
whose exe can't be read (psutil error) — it stays in the candidate set.
"""
class _FakeProc:
def __init__(self, pid: int, chain: list[int]):
def __init__(self, pid: int, exe_path: str | None):
self.pid = pid
self._chain = chain
self._exe = exe_path
def parent(self):
if not self._chain:
return None
next_pid = self._chain[0]
return _FakeProc(next_pid, self._chain[1:])
def exe(self):
if self._exe is None:
raise OSError("exe unavailable")
return self._exe
def parents(self):
return [_FakeProc(p, ancestor_exe) for p in parent_chain]
class _NoSuchProcess(Exception):
pass
@@ -153,8 +160,8 @@ def _fake_psutil_with_parent_chain(
class _AccessDenied(Exception):
pass
def _process(pid):
return _FakeProc(pid, list(parent_chain))
def _process(pid=None):
return _FakeProc(pid if pid is not None else os.getpid(), ancestor_exe)
return types.SimpleNamespace(
Process=_process,
@@ -185,6 +192,7 @@ def test_detect_concurrent_excludes_parent_chain(_winp, tmp_path):
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[launcher_pid],
proc_iter_rows=rows,
ancestor_exe=str(shim),
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
@@ -211,6 +219,7 @@ def test_detect_concurrent_still_finds_unrelated_other_hermes(_winp, tmp_path):
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[launcher_pid],
proc_iter_rows=rows,
ancestor_exe=str(shim),
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
@@ -238,6 +247,7 @@ def test_detect_concurrent_parent_chain_walks_deep(_winp, tmp_path):
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[parent_pid, grandparent_pid, greatgrandparent_pid],
proc_iter_rows=rows,
ancestor_exe=str(shim),
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
@@ -246,25 +256,38 @@ def test_detect_concurrent_parent_chain_walks_deep(_winp, tmp_path):
@patch.object(cli_main, "_is_windows", return_value=True)
def test_detect_concurrent_parent_walk_handles_cycle(_winp, tmp_path):
"""A PID cycle in the parent chain must not hang the walk."""
def test_detect_concurrent_parents_call_robust_to_one_bad_hop(_winp, tmp_path):
"""The launcher shim is still excluded even when an ancestor exe is unreadable.
Field regression (issues #29341, #34795): the old per-hop ``parent()``
walk bailed on the FIRST psutil error, so an AccessDenied on any hop left
the launcher shim in the candidate set and re-triggered the false
positive. ``parents()`` returns the whole list at once; we evaluate each
ancestor independently, so one unreadable hop never strands the launcher.
"""
scripts_dir = tmp_path
shim = scripts_dir / "hermes.exe"
shim.write_bytes(b"")
me = os.getpid()
bogus_loop_pid = me + 1
launcher_pid = me + 100
rows = [_make_proc(me, str(shim), "python.exe")]
# Chain that points back to ``me`` — the loop-detection branch must break.
rows = [
_make_proc(me, str(shim), "python.exe"),
_make_proc(launcher_pid, str(shim), "hermes.exe"),
]
# ancestor_exe=None → every ancestor's .exe() raises OSError. The helper
# must swallow it per-ancestor and not crash; the launcher won't be
# excluded in this degenerate case, but a real run reads the shim exe.
fake_psutil = _fake_psutil_with_parent_chain(
parent_chain=[bogus_loop_pid, me, bogus_loop_pid],
parent_chain=[launcher_pid],
proc_iter_rows=rows,
ancestor_exe=None,
)
with patch.dict(sys.modules, {"psutil": fake_psutil}):
result = cli_main._detect_concurrent_hermes_instances(scripts_dir)
# No crash, no hang; self + bogus_loop_pid excluded; no others reported.
assert result == []
# No crash; helper completes. (Degenerate stub: launcher exe unreadable.)
assert result == [(launcher_pid, "hermes.exe")]
@patch.object(cli_main, "_is_windows", return_value=True)
@@ -310,6 +333,11 @@ def test_format_message_mentions_pids_and_remediation(tmp_path):
assert "--force" in msg
# Mentions the file that would have been overwritten
assert str(tmp_path / "hermes.exe") in msg
# Self-service kill command targets the exact stale PIDs (issue #34795).
assert "taskkill" in msg
assert "/PID 1234" in msg
assert "/PID 5678" in msg
assert "/F" in msg
# ---------------------------------------------------------------------------
+311
View File
@@ -0,0 +1,311 @@
"""Tests for uv-tool install detection in the update path (issue #29700).
``uv tool install hermes-agent`` lives outside any venv, so the previous
``uv pip install --upgrade`` update path failed with ``No virtual
environment found``. ``is_uv_tool_install`` should detect this layout and
both the user-facing recommended command and the actual
``_cmd_update_pip`` subprocess invocation should switch to
``uv tool upgrade hermes-agent``.
Detection is restricted to properties of the running interpreter
(``sys.prefix`` / ``sys.executable``) so a pip/venv install on a machine
that also has ``uv tool install hermes-agent`` does not get misclassified.
"""
from __future__ import annotations
import subprocess
from types import SimpleNamespace
from unittest.mock import patch
import pytest
# ---------------------------------------------------------------------------
# is_uv_tool_install
# ---------------------------------------------------------------------------
class TestIsUvToolInstall:
def test_returns_true_when_sys_prefix_matches_uv_tool_layout(self):
from hermes_cli import config
with patch.object(config.sys, "prefix", "/home/user/.local/share/uv/tools/hermes-agent"):
assert config.is_uv_tool_install() is True
def test_returns_true_when_sys_executable_matches_uv_tool_layout(self):
"""Some uv-tool layouts surface the marker on ``sys.executable`` (bin/python)."""
from hermes_cli import config
with patch.object(config.sys, "prefix", "/some/unrelated/venv"), \
patch.object(
config.sys,
"executable",
"/home/user/.local/share/uv/tools/hermes-agent/bin/python",
):
assert config.is_uv_tool_install() is True
def test_returns_false_when_neither_prefix_nor_executable_matches(self):
from hermes_cli import config
with patch.object(config.sys, "prefix", "/some/unrelated/venv"), \
patch.object(config.sys, "executable", "/usr/bin/python3"):
assert config.is_uv_tool_install() is False
def test_does_not_consult_uv_tool_list(self):
"""Detection must NOT shell out: ``uv tool list`` would false-positive
when the active install is pip/venv but the machine also has
``uv tool install hermes-agent`` somewhere on disk. Copilot review on
PR #29703 flagged this; the fix is to never call ``uv tool list``
from the detection path."""
from hermes_cli import config
with patch.object(config.sys, "prefix", "/some/unrelated/venv"), \
patch.object(config.sys, "executable", "/usr/bin/python3"), \
patch("subprocess.run") as mock_run:
assert config.is_uv_tool_install() is False
mock_run.assert_not_called()
def test_case_insensitive_match(self):
"""Match must be case-insensitive — Windows paths preserve case
(e.g. ``...AppData\\Local\\UV\\Tools\\hermes-agent``) and a case-sensitive
check would miss them. We exercise the lower-cased compare path here
without monkey-patching ``os.sep``, which would break the whole suite."""
from hermes_cli import config
with patch.object(
config.sys, "prefix", "/HOME/USER/.local/share/UV/Tools/hermes-agent"
):
assert config.is_uv_tool_install() is True
def test_handles_empty_executable(self):
from hermes_cli import config
with patch.object(config.sys, "prefix", "/some/unrelated/venv"), \
patch.object(config.sys, "executable", ""):
assert config.is_uv_tool_install() is False
# ---------------------------------------------------------------------------
# recommended_update_command_for_method
# ---------------------------------------------------------------------------
class TestRecommendedUpdateCommandForUvTool:
def test_uv_tool_install_recommends_uv_tool_upgrade(self):
from hermes_cli import config
with patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch.object(config, "is_uv_tool_install", return_value=True):
cmd = config.recommended_update_command_for_method("pip")
assert cmd == "uv tool upgrade hermes-agent"
def test_uv_tool_install_recommends_uv_tool_upgrade_even_without_uv_on_path(self):
"""Recommendation reflects the *install method*, not whether ``uv`` is
currently on PATH the user needs to know the right command to run."""
from hermes_cli import config
with patch("shutil.which", return_value=None), \
patch.object(config, "is_uv_tool_install", return_value=True):
cmd = config.recommended_update_command_for_method("pip")
assert cmd == "uv tool upgrade hermes-agent"
def test_uv_pip_install_keeps_legacy_recommendation(self):
"""Existing behavior: uv is on PATH but Hermes is a regular pip install."""
from hermes_cli import config
with patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch.object(config, "is_uv_tool_install", return_value=False):
cmd = config.recommended_update_command_for_method("pip")
assert cmd == "uv pip install --upgrade hermes-agent"
def test_no_uv_falls_back_to_plain_pip(self):
from hermes_cli import config
with patch("shutil.which", return_value=None), \
patch.object(config, "is_uv_tool_install", return_value=False):
cmd = config.recommended_update_command_for_method("pip")
assert cmd == "pip install --upgrade hermes-agent"
def test_recommendation_does_not_spawn_subprocess(self):
"""Computing the recommendation string must be cheap — no ``uv tool list``
spawn. Copilot review on PR #29703 flagged the prior subprocess hop
as adding overhead and a multi-second timeout window for what is
purely a display string."""
from hermes_cli import config
with patch.object(config.sys, "prefix", "/some/unrelated/venv"), \
patch.object(config.sys, "executable", "/usr/bin/python3"), \
patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch("subprocess.run") as mock_run:
cmd = config.recommended_update_command_for_method("pip")
mock_run.assert_not_called()
assert cmd == "uv pip install --upgrade hermes-agent"
# ---------------------------------------------------------------------------
# _cmd_update_pip subprocess command
# ---------------------------------------------------------------------------
class TestCmdUpdatePipUsesUvTool:
@patch("subprocess.run")
def test_runs_uv_tool_upgrade_when_uv_tool_install(self, mock_run):
"""The actual subprocess invocation must switch to ``uv tool upgrade``."""
from hermes_cli.main import _cmd_update_pip
mock_run.return_value = subprocess.CompletedProcess(["uv"], 0, stdout="", stderr="")
with patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch("hermes_cli.config.is_uv_tool_install", return_value=True):
_cmd_update_pip(SimpleNamespace())
assert mock_run.call_args[0][0] == ["/usr/local/bin/uv", "tool", "upgrade", "hermes-agent"]
@patch("subprocess.run")
def test_runs_uv_pip_install_when_not_uv_tool(self, mock_run):
"""Existing behavior preserved when uv is present but Hermes isn't a tool install."""
from hermes_cli.main import _cmd_update_pip
mock_run.return_value = subprocess.CompletedProcess(["uv"], 0, stdout="", stderr="")
with patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
_cmd_update_pip(SimpleNamespace())
assert mock_run.call_args[0][0] == [
"/usr/local/bin/uv",
"pip",
"install",
"--upgrade",
"hermes-agent",
]
@patch("subprocess.run")
def test_falls_back_to_pip_when_no_uv(self, mock_run):
from hermes_cli.main import _cmd_update_pip
mock_run.return_value = subprocess.CompletedProcess(["pip"], 0, stdout="", stderr="")
with patch("shutil.which", return_value=None), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
_cmd_update_pip(SimpleNamespace())
cmd = mock_run.call_args[0][0]
assert cmd[1:] == ["-m", "pip", "install", "--upgrade", "hermes-agent"]
@patch("subprocess.run")
def test_exits_nonzero_on_subprocess_failure(self, mock_run):
from hermes_cli.main import _cmd_update_pip
mock_run.return_value = subprocess.CompletedProcess(["uv"], 1, stdout="", stderr="")
with patch("shutil.which", return_value="/usr/local/bin/uv"), \
patch("hermes_cli.config.is_uv_tool_install", return_value=True):
with pytest.raises(SystemExit) as exc_info:
_cmd_update_pip(SimpleNamespace())
assert exc_info.value.code == 1
@patch("subprocess.run")
def test_uv_tool_install_without_uv_on_path_exits_with_hint(self, mock_run):
"""If the running interpreter looks like a uv-tool install but ``uv`` is
somehow missing from PATH, surface a clear hint instead of silently
falling back to ``python -m pip``, which would either fail (no venv)
or upgrade the wrong copy."""
from hermes_cli.main import _cmd_update_pip
with patch("shutil.which", return_value=None), \
patch("hermes_cli.config.is_uv_tool_install", return_value=True):
with pytest.raises(SystemExit) as exc_info:
_cmd_update_pip(SimpleNamespace())
assert exc_info.value.code == 1
mock_run.assert_not_called()
# ---------------------------------------------------------------------------
# pipx-managed installs, --system fallback, and VIRTUAL_ENV overlay
# (issue #29700 / #35031 family — consolidated update-path handling)
# ---------------------------------------------------------------------------
class TestCmdUpdatePipInstallLayouts:
"""The uv pip path must adapt to where the running interpreter lives:
- inside a venv (launcher shim) -> export VIRTUAL_ENV, no ``--system``
- bare pip outside any venv -> add ``--system``, no overlay
- pipx-managed -> ``pipx upgrade``
"""
@patch("subprocess.run")
def test_pipx_managed_uses_pipx_upgrade(self, mock_run, monkeypatch):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
monkeypatch.setattr(hm.sys, "prefix", "/home/u/.local/pipx/venvs/hermes-agent")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
def _which(name):
return {"uv": "/usr/bin/uv", "pipx": "/usr/bin/pipx"}.get(name)
with patch("shutil.which", side_effect=_which), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
hm._cmd_update_pip(SimpleNamespace())
assert mock_run.call_args[0][0] == ["/usr/bin/pipx", "upgrade", "hermes-agent"]
# pipx upgrade ignores VIRTUAL_ENV; we must not set it.
assert "env" not in mock_run.call_args.kwargs
@patch("subprocess.run")
def test_pipx_layout_without_pipx_binary_treated_as_venv(
self, mock_run, monkeypatch
):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
monkeypatch.setattr(hm.sys, "prefix", "/home/u/.local/pipx/venvs/hermes-agent")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
# pipx layout detected via prefix, but pipx binary missing on PATH.
def _which(name):
return "/usr/bin/uv" if name == "uv" else None
with patch("shutil.which", side_effect=_which), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
hm._cmd_update_pip(SimpleNamespace())
# prefix != base_prefix, so this is treated as a venv -> overlay, no --system.
assert mock_run.call_args[0][0] == [
"/usr/bin/uv", "pip", "install", "--upgrade", "hermes-agent",
]
assert mock_run.call_args.kwargs["env"]["VIRTUAL_ENV"].endswith("hermes-agent")
@patch("subprocess.run")
def test_bare_pip_outside_venv_adds_system(self, mock_run, monkeypatch):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
# No venv: prefix == base_prefix.
monkeypatch.setattr(hm.sys, "prefix", "/usr")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
with patch("shutil.which", return_value="/usr/bin/uv"), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
hm._cmd_update_pip(SimpleNamespace())
assert mock_run.call_args[0][0] == [
"/usr/bin/uv", "pip", "install", "--system", "--upgrade", "hermes-agent",
]
assert "env" not in mock_run.call_args.kwargs
@patch("subprocess.run")
def test_venv_exports_virtualenv_and_omits_system(self, mock_run, monkeypatch):
from hermes_cli import main as hm
mock_run.return_value = subprocess.CompletedProcess([], 0, stdout="", stderr="")
monkeypatch.delenv("VIRTUAL_ENV", raising=False)
monkeypatch.setattr(hm.sys, "prefix", "/home/u/.hermes/hermes-agent/venv")
monkeypatch.setattr(hm.sys, "base_prefix", "/usr")
with patch("shutil.which", return_value="/usr/bin/uv"), \
patch("hermes_cli.config.is_uv_tool_install", return_value=False):
hm._cmd_update_pip(SimpleNamespace())
cmd = mock_run.call_args[0][0]
assert "--system" not in cmd
assert cmd == ["/usr/bin/uv", "pip", "install", "--upgrade", "hermes-agent"]
assert mock_run.call_args.kwargs["env"]["VIRTUAL_ENV"] == "/home/u/.hermes/hermes-agent/venv"