mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
f0b763c74f
Reported during the TUI v2 blitz test: switching from openrouter to anthropic via `/model <name> --provider anthropic` appeared to succeed, but the next turn kept hitting openrouter — the provider the user was deliberately moving away from. Two gaps caused this: 1. `Agent.switch_model` reset `_fallback_activated` / `_fallback_index` but left `_fallback_chain` intact. The chain was seeded from `fallback_providers:` at agent init for the *original* primary, so when the new primary returned 401 (invalid/expired Anthropic key), `_try_activate_fallback()` picked the old provider back up without informing the user. Prune entries matching either the old primary (user is moving away) or the new primary (redundant) whenever the primary provider actually changes. 2. `_apply_model_switch` persisted `HERMES_MODEL` but never updated `HERMES_INFERENCE_PROVIDER`. Any ambient re-resolution of the runtime (credential pool refresh, compressor rebuild, aux clients) falls through to that env var in `resolve_requested_provider`, so it kept reporting the original provider even after an in-memory switch. Adds three regression tests: fallback-chain prune on primary change, no-op on same-provider model swap, and env-var sync on explicit switch.
1259 lines
47 KiB
Python
1259 lines
47 KiB
Python
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import time
|
|
import types
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from tui_gateway import server
|
|
|
|
|
|
class _ChunkyStdout:
|
|
def __init__(self):
|
|
self.parts: list[str] = []
|
|
|
|
def write(self, text: str) -> int:
|
|
for ch in text:
|
|
self.parts.append(ch)
|
|
time.sleep(0.0001)
|
|
return len(text)
|
|
|
|
def flush(self) -> None:
|
|
return None
|
|
|
|
|
|
class _BrokenStdout:
|
|
def write(self, text: str) -> int:
|
|
raise BrokenPipeError
|
|
|
|
def flush(self) -> None:
|
|
return None
|
|
|
|
|
|
def test_write_json_serializes_concurrent_writes(monkeypatch):
|
|
out = _ChunkyStdout()
|
|
monkeypatch.setattr(server, "_real_stdout", out)
|
|
|
|
threads = [
|
|
threading.Thread(target=server.write_json, args=({"seq": i, "text": "x" * 24},))
|
|
for i in range(8)
|
|
]
|
|
|
|
for t in threads:
|
|
t.start()
|
|
|
|
for t in threads:
|
|
t.join()
|
|
|
|
lines = "".join(out.parts).splitlines()
|
|
|
|
assert len(lines) == 8
|
|
assert {json.loads(line)["seq"] for line in lines} == set(range(8))
|
|
|
|
|
|
def test_write_json_returns_false_on_broken_pipe(monkeypatch):
|
|
monkeypatch.setattr(server, "_real_stdout", _BrokenStdout())
|
|
|
|
assert server.write_json({"ok": True}) is False
|
|
|
|
|
|
def test_status_callback_emits_kind_and_text():
|
|
with patch("tui_gateway.server._emit") as emit:
|
|
cb = server._agent_cbs("sid")["status_callback"]
|
|
cb("context_pressure", "85% to compaction")
|
|
|
|
emit.assert_called_once_with(
|
|
"status.update",
|
|
"sid",
|
|
{"kind": "context_pressure", "text": "85% to compaction"},
|
|
)
|
|
|
|
|
|
def test_status_callback_accepts_single_message_argument():
|
|
with patch("tui_gateway.server._emit") as emit:
|
|
cb = server._agent_cbs("sid")["status_callback"]
|
|
cb("thinking...")
|
|
|
|
emit.assert_called_once_with(
|
|
"status.update",
|
|
"sid",
|
|
{"kind": "status", "text": "thinking..."},
|
|
)
|
|
|
|
|
|
def _session(agent=None, **extra):
|
|
return {
|
|
"agent": agent if agent is not None else types.SimpleNamespace(),
|
|
"session_key": "session-key",
|
|
"history": [],
|
|
"history_lock": threading.Lock(),
|
|
"history_version": 0,
|
|
"running": False,
|
|
"attached_images": [],
|
|
"image_counter": 0,
|
|
"cols": 80,
|
|
"slash_worker": None,
|
|
"show_reasoning": False,
|
|
"tool_progress_mode": "all",
|
|
**extra,
|
|
}
|
|
|
|
|
|
def test_config_set_yolo_toggles_session_scope():
|
|
from tools.approval import clear_session, is_session_yolo_enabled
|
|
|
|
server._sessions["sid"] = _session()
|
|
try:
|
|
resp_on = server.handle_request({"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "yolo"}})
|
|
assert resp_on["result"]["value"] == "1"
|
|
assert is_session_yolo_enabled("session-key") is True
|
|
|
|
resp_off = server.handle_request({"id": "2", "method": "config.set", "params": {"session_id": "sid", "key": "yolo"}})
|
|
assert resp_off["result"]["value"] == "0"
|
|
assert is_session_yolo_enabled("session-key") is False
|
|
finally:
|
|
clear_session("session-key")
|
|
server._sessions.clear()
|
|
|
|
|
|
def test_enable_gateway_prompts_sets_gateway_env(monkeypatch):
|
|
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
|
|
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
|
|
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
|
|
|
|
server._enable_gateway_prompts()
|
|
|
|
assert server.os.environ["HERMES_GATEWAY_SESSION"] == "1"
|
|
assert server.os.environ["HERMES_EXEC_ASK"] == "1"
|
|
assert server.os.environ["HERMES_INTERACTIVE"] == "1"
|
|
|
|
|
|
def test_setup_status_reports_provider_config(monkeypatch):
|
|
monkeypatch.setattr("hermes_cli.main._has_any_provider_configured", lambda: False)
|
|
|
|
resp = server.handle_request({"id": "1", "method": "setup.status", "params": {}})
|
|
|
|
assert resp["result"]["provider_configured"] is False
|
|
|
|
|
|
def test_config_set_reasoning_updates_live_session_and_agent(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(server, "_hermes_home", tmp_path)
|
|
agent = types.SimpleNamespace(reasoning_config=None)
|
|
server._sessions["sid"] = _session(agent=agent)
|
|
|
|
resp_effort = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "reasoning", "value": "low"}}
|
|
)
|
|
assert resp_effort["result"]["value"] == "low"
|
|
assert agent.reasoning_config == {"enabled": True, "effort": "low"}
|
|
|
|
resp_show = server.handle_request(
|
|
{"id": "2", "method": "config.set", "params": {"session_id": "sid", "key": "reasoning", "value": "show"}}
|
|
)
|
|
assert resp_show["result"]["value"] == "show"
|
|
assert server._sessions["sid"]["show_reasoning"] is True
|
|
|
|
|
|
def test_config_set_verbose_updates_session_mode_and_agent(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(server, "_hermes_home", tmp_path)
|
|
agent = types.SimpleNamespace(verbose_logging=False)
|
|
server._sessions["sid"] = _session(agent=agent)
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "verbose", "value": "cycle"}}
|
|
)
|
|
|
|
assert resp["result"]["value"] == "verbose"
|
|
assert server._sessions["sid"]["tool_progress_mode"] == "verbose"
|
|
assert agent.verbose_logging is True
|
|
|
|
|
|
def test_config_set_model_uses_live_switch_path(monkeypatch):
|
|
server._sessions["sid"] = _session()
|
|
seen = {}
|
|
|
|
def _fake_apply(sid, session, raw):
|
|
seen["args"] = (sid, session["session_key"], raw)
|
|
return {"value": "new/model", "warning": "catalog unreachable"}
|
|
|
|
monkeypatch.setattr(server, "_apply_model_switch", _fake_apply)
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "model", "value": "new/model"}}
|
|
)
|
|
|
|
assert resp["result"]["value"] == "new/model"
|
|
assert resp["result"]["warning"] == "catalog unreachable"
|
|
assert seen["args"] == ("sid", "session-key", "new/model")
|
|
|
|
|
|
def test_config_set_model_global_persists(monkeypatch):
|
|
class _Agent:
|
|
provider = "openrouter"
|
|
model = "old/model"
|
|
base_url = ""
|
|
api_key = "sk-old"
|
|
|
|
def switch_model(self, **kwargs):
|
|
return None
|
|
|
|
result = types.SimpleNamespace(
|
|
success=True,
|
|
new_model="anthropic/claude-sonnet-4.6",
|
|
target_provider="anthropic",
|
|
api_key="sk-new",
|
|
base_url="https://api.anthropic.com",
|
|
api_mode="anthropic_messages",
|
|
warning_message="",
|
|
)
|
|
seen = {}
|
|
saved = {}
|
|
|
|
def _switch_model(**kwargs):
|
|
seen.update(kwargs)
|
|
return result
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
monkeypatch.setattr("hermes_cli.model_switch.switch_model", _switch_model)
|
|
monkeypatch.setattr(server, "_restart_slash_worker", lambda session: None)
|
|
monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr("hermes_cli.config.save_config", lambda cfg: saved.update(cfg))
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "model", "value": "anthropic/claude-sonnet-4.6 --global"}}
|
|
)
|
|
|
|
assert resp["result"]["value"] == "anthropic/claude-sonnet-4.6"
|
|
assert seen["is_global"] is True
|
|
assert saved["model"]["default"] == "anthropic/claude-sonnet-4.6"
|
|
assert saved["model"]["provider"] == "anthropic"
|
|
assert saved["model"]["base_url"] == "https://api.anthropic.com"
|
|
|
|
|
|
def test_config_set_model_syncs_inference_provider_env(monkeypatch):
|
|
"""After an explicit provider switch, HERMES_INFERENCE_PROVIDER must
|
|
reflect the user's choice so ambient re-resolution (credential pool
|
|
refresh, aux clients) picks up the new provider instead of the original
|
|
one persisted in config or shell env.
|
|
|
|
Regression: a TUI user switched openrouter → anthropic and the TUI kept
|
|
trying openrouter because the env-var-backed resolvers still saw the old
|
|
provider.
|
|
"""
|
|
class _Agent:
|
|
provider = "openrouter"
|
|
model = "old/model"
|
|
base_url = ""
|
|
api_key = "sk-or"
|
|
|
|
def switch_model(self, **_kwargs):
|
|
return None
|
|
|
|
result = types.SimpleNamespace(
|
|
success=True,
|
|
new_model="claude-sonnet-4.6",
|
|
target_provider="anthropic",
|
|
api_key="sk-ant",
|
|
base_url="https://api.anthropic.com",
|
|
api_mode="anthropic_messages",
|
|
warning_message="",
|
|
)
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
monkeypatch.setenv("HERMES_INFERENCE_PROVIDER", "openrouter")
|
|
monkeypatch.setattr("hermes_cli.model_switch.switch_model", lambda **_kwargs: result)
|
|
monkeypatch.setattr(server, "_restart_slash_worker", lambda session: None)
|
|
monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
|
|
|
|
server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "model", "value": "claude-sonnet-4.6 --provider anthropic"}}
|
|
)
|
|
|
|
assert os.environ["HERMES_INFERENCE_PROVIDER"] == "anthropic"
|
|
|
|
|
|
def test_config_set_personality_rejects_unknown_name(monkeypatch):
|
|
monkeypatch.setattr(server, "_available_personalities", lambda cfg=None: {"helpful": "You are helpful."})
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"key": "personality", "value": "bogus"}}
|
|
)
|
|
|
|
assert "error" in resp
|
|
assert "Unknown personality" in resp["error"]["message"]
|
|
|
|
|
|
def test_config_set_personality_resets_history_and_returns_info(monkeypatch):
|
|
session = _session(agent=types.SimpleNamespace(), history=[{"role": "user", "text": "hi"}], history_version=4)
|
|
new_agent = types.SimpleNamespace(model="x")
|
|
emits = []
|
|
|
|
server._sessions["sid"] = session
|
|
monkeypatch.setattr(server, "_available_personalities", lambda cfg=None: {"helpful": "You are helpful."})
|
|
monkeypatch.setattr(server, "_make_agent", lambda sid, key, session_id=None: new_agent)
|
|
monkeypatch.setattr(server, "_session_info", lambda agent: {"model": getattr(agent, "model", "?")})
|
|
monkeypatch.setattr(server, "_restart_slash_worker", lambda session: None)
|
|
monkeypatch.setattr(server, "_emit", lambda *args: emits.append(args))
|
|
monkeypatch.setattr(server, "_write_config_key", lambda path, value: None)
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "personality", "value": "helpful"}}
|
|
)
|
|
|
|
assert resp["result"]["history_reset"] is True
|
|
assert resp["result"]["info"] == {"model": "x"}
|
|
assert session["history"] == []
|
|
assert session["history_version"] == 5
|
|
assert ("session.info", "sid", {"model": "x"}) in emits
|
|
|
|
|
|
def test_session_compress_uses_compress_helper(monkeypatch):
|
|
agent = types.SimpleNamespace()
|
|
server._sessions["sid"] = _session(agent=agent)
|
|
|
|
monkeypatch.setattr(server, "_compress_session_history", lambda session, focus_topic=None: (2, {"total": 42}))
|
|
monkeypatch.setattr(server, "_session_info", lambda _agent: {"model": "x"})
|
|
|
|
with patch("tui_gateway.server._emit") as emit:
|
|
resp = server.handle_request({"id": "1", "method": "session.compress", "params": {"session_id": "sid"}})
|
|
|
|
assert resp["result"]["removed"] == 2
|
|
assert resp["result"]["usage"]["total"] == 42
|
|
emit.assert_called_once_with("session.info", "sid", {"model": "x"})
|
|
|
|
|
|
def test_prompt_submit_sets_approval_session_key(monkeypatch):
|
|
from tools.approval import get_current_session_key
|
|
|
|
captured = {}
|
|
|
|
class _Agent:
|
|
def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
|
|
captured["session_key"] = get_current_session_key(default="")
|
|
return {"final_response": "ok", "messages": [{"role": "assistant", "content": "ok"}]}
|
|
|
|
class _ImmediateThread:
|
|
def __init__(self, target=None, daemon=None):
|
|
self._target = target
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
|
|
monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
|
|
monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
|
|
|
|
resp = server.handle_request({"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "ping"}})
|
|
|
|
assert resp["result"]["status"] == "streaming"
|
|
assert captured["session_key"] == "session-key"
|
|
|
|
|
|
def test_prompt_submit_expands_context_refs(monkeypatch):
|
|
captured = {}
|
|
|
|
class _Agent:
|
|
model = "test/model"
|
|
base_url = ""
|
|
api_key = ""
|
|
|
|
def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
|
|
captured["prompt"] = prompt
|
|
return {"final_response": "ok", "messages": [{"role": "assistant", "content": "ok"}]}
|
|
|
|
class _ImmediateThread:
|
|
def __init__(self, target=None, daemon=None):
|
|
self._target = target
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
fake_ctx = types.ModuleType("agent.context_references")
|
|
fake_ctx.preprocess_context_references = lambda message, **kwargs: types.SimpleNamespace(
|
|
blocked=False, message="expanded prompt", warnings=[], references=[], injected_tokens=0
|
|
)
|
|
fake_meta = types.ModuleType("agent.model_metadata")
|
|
fake_meta.get_model_context_length = lambda *args, **kwargs: 100000
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
|
|
monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
|
|
monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
|
|
monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
|
|
monkeypatch.setitem(sys.modules, "agent.context_references", fake_ctx)
|
|
monkeypatch.setitem(sys.modules, "agent.model_metadata", fake_meta)
|
|
|
|
server.handle_request({"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "@diff"}})
|
|
|
|
assert captured["prompt"] == "expanded prompt"
|
|
|
|
|
|
def test_image_attach_appends_local_image(monkeypatch):
|
|
fake_cli = types.ModuleType("cli")
|
|
fake_cli._IMAGE_EXTENSIONS = {".png"}
|
|
fake_cli._detect_file_drop = lambda raw: {
|
|
"path": Path("/tmp/cat.png"),
|
|
"is_image": True,
|
|
"remainder": "",
|
|
}
|
|
fake_cli._split_path_input = lambda raw: (raw, "")
|
|
fake_cli._resolve_attachment_path = lambda raw: Path("/tmp/cat.png")
|
|
|
|
server._sessions["sid"] = _session()
|
|
monkeypatch.setitem(sys.modules, "cli", fake_cli)
|
|
|
|
resp = server.handle_request({"id": "1", "method": "image.attach", "params": {"session_id": "sid", "path": "/tmp/cat.png"}})
|
|
|
|
assert resp["result"]["attached"] is True
|
|
assert resp["result"]["name"] == "cat.png"
|
|
assert len(server._sessions["sid"]["attached_images"]) == 1
|
|
|
|
|
|
def test_image_attach_accepts_unquoted_screenshot_path_with_spaces(monkeypatch):
|
|
screenshot = Path("/tmp/Screenshot 2026-04-21 at 1.04.43 PM.png")
|
|
fake_cli = types.ModuleType("cli")
|
|
fake_cli._IMAGE_EXTENSIONS = {".png"}
|
|
fake_cli._detect_file_drop = lambda raw: {
|
|
"path": screenshot,
|
|
"is_image": True,
|
|
"remainder": "",
|
|
}
|
|
fake_cli._split_path_input = lambda raw: ("/tmp/Screenshot", "2026-04-21 at 1.04.43 PM.png")
|
|
fake_cli._resolve_attachment_path = lambda raw: None
|
|
|
|
server._sessions["sid"] = _session()
|
|
monkeypatch.setitem(sys.modules, "cli", fake_cli)
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "image.attach", "params": {"session_id": "sid", "path": str(screenshot)}}
|
|
)
|
|
|
|
assert resp["result"]["attached"] is True
|
|
assert resp["result"]["path"] == str(screenshot)
|
|
assert resp["result"]["remainder"] == ""
|
|
assert len(server._sessions["sid"]["attached_images"]) == 1
|
|
|
|
|
|
def test_commands_catalog_surfaces_quick_commands(monkeypatch):
|
|
monkeypatch.setattr(server, "_load_cfg", lambda: {"quick_commands": {
|
|
"build": {"type": "exec", "command": "npm run build"},
|
|
"git": {"type": "alias", "target": "/shell git"},
|
|
"notes": {"type": "exec", "command": "cat NOTES.md", "description": "Open design notes"},
|
|
}})
|
|
|
|
resp = server.handle_request({"id": "1", "method": "commands.catalog", "params": {}})
|
|
|
|
pairs = dict(resp["result"]["pairs"])
|
|
assert "npm run build" in pairs["/build"]
|
|
assert pairs["/git"].startswith("alias →")
|
|
assert pairs["/notes"] == "Open design notes"
|
|
|
|
user_cat = next(c for c in resp["result"]["categories"] if c["name"] == "User commands")
|
|
user_pairs = dict(user_cat["pairs"])
|
|
assert set(user_pairs) == {"/build", "/git", "/notes"}
|
|
|
|
assert resp["result"]["canon"]["/build"] == "/build"
|
|
assert resp["result"]["canon"]["/notes"] == "/notes"
|
|
|
|
|
|
def test_command_dispatch_exec_nonzero_surfaces_error(monkeypatch):
|
|
monkeypatch.setattr(server, "_load_cfg", lambda: {"quick_commands": {"boom": {"type": "exec", "command": "boom"}}})
|
|
monkeypatch.setattr(
|
|
server.subprocess,
|
|
"run",
|
|
lambda *args, **kwargs: types.SimpleNamespace(returncode=1, stdout="", stderr="failed"),
|
|
)
|
|
|
|
resp = server.handle_request({"id": "1", "method": "command.dispatch", "params": {"name": "boom"}})
|
|
|
|
assert "error" in resp
|
|
assert "failed" in resp["error"]["message"]
|
|
|
|
|
|
def test_plugins_list_surfaces_loader_error(monkeypatch):
|
|
with patch("hermes_cli.plugins.get_plugin_manager", side_effect=Exception("boom")):
|
|
resp = server.handle_request({"id": "1", "method": "plugins.list", "params": {}})
|
|
|
|
assert "error" in resp
|
|
assert "boom" in resp["error"]["message"]
|
|
|
|
|
|
def test_complete_slash_surfaces_completer_error(monkeypatch):
|
|
with patch("hermes_cli.commands.SlashCommandCompleter", side_effect=Exception("no completer")):
|
|
resp = server.handle_request({"id": "1", "method": "complete.slash", "params": {"text": "/mo"}})
|
|
|
|
assert "error" in resp
|
|
assert "no completer" in resp["error"]["message"]
|
|
|
|
|
|
def test_input_detect_drop_attaches_image(monkeypatch):
|
|
fake_cli = types.ModuleType("cli")
|
|
fake_cli._detect_file_drop = lambda raw: {
|
|
"path": Path("/tmp/cat.png"),
|
|
"is_image": True,
|
|
"remainder": "",
|
|
}
|
|
|
|
server._sessions["sid"] = _session()
|
|
monkeypatch.setitem(sys.modules, "cli", fake_cli)
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "input.detect_drop", "params": {"session_id": "sid", "text": "/tmp/cat.png"}}
|
|
)
|
|
|
|
assert resp["result"]["matched"] is True
|
|
assert resp["result"]["is_image"] is True
|
|
assert resp["result"]["text"] == "[User attached image: cat.png]"
|
|
|
|
|
|
def test_rollback_restore_resolves_number_and_file_path():
|
|
calls = {}
|
|
|
|
class _Mgr:
|
|
enabled = True
|
|
|
|
def list_checkpoints(self, cwd):
|
|
return [{"hash": "aaa111"}, {"hash": "bbb222"}]
|
|
|
|
def restore(self, cwd, target, file_path=None):
|
|
calls["args"] = (cwd, target, file_path)
|
|
return {"success": True, "message": "done"}
|
|
|
|
server._sessions["sid"] = _session(agent=types.SimpleNamespace(_checkpoint_mgr=_Mgr()), history=[])
|
|
resp = server.handle_request(
|
|
{
|
|
"id": "1",
|
|
"method": "rollback.restore",
|
|
"params": {"session_id": "sid", "hash": "2", "file_path": "src/app.tsx"},
|
|
}
|
|
)
|
|
|
|
assert resp["result"]["success"] is True
|
|
assert calls["args"][1] == "bbb222"
|
|
assert calls["args"][2] == "src/app.tsx"
|
|
|
|
|
|
# ── session.steer ────────────────────────────────────────────────────
|
|
|
|
|
|
def test_session_steer_calls_agent_steer_when_agent_supports_it():
|
|
"""The TUI RPC method must call agent.steer(text) and return a
|
|
queued status without touching interrupt state.
|
|
"""
|
|
calls = {}
|
|
|
|
class _Agent:
|
|
def steer(self, text):
|
|
calls["steer_text"] = text
|
|
return True
|
|
|
|
def interrupt(self, *args, **kwargs):
|
|
calls["interrupt_called"] = True
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
try:
|
|
resp = server.handle_request(
|
|
{
|
|
"id": "1",
|
|
"method": "session.steer",
|
|
"params": {"session_id": "sid", "text": "also check auth.log"},
|
|
}
|
|
)
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
assert "result" in resp, resp
|
|
assert resp["result"]["status"] == "queued"
|
|
assert resp["result"]["text"] == "also check auth.log"
|
|
assert calls["steer_text"] == "also check auth.log"
|
|
assert "interrupt_called" not in calls # must NOT interrupt
|
|
|
|
|
|
def test_session_steer_rejects_empty_text():
|
|
server._sessions["sid"] = _session(agent=types.SimpleNamespace(steer=lambda t: True))
|
|
try:
|
|
resp = server.handle_request(
|
|
{
|
|
"id": "1",
|
|
"method": "session.steer",
|
|
"params": {"session_id": "sid", "text": " "},
|
|
}
|
|
)
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
assert "error" in resp, resp
|
|
assert resp["error"]["code"] == 4002
|
|
|
|
|
|
def test_session_steer_errors_when_agent_has_no_steer_method():
|
|
server._sessions["sid"] = _session(agent=types.SimpleNamespace()) # no steer()
|
|
try:
|
|
resp = server.handle_request(
|
|
{
|
|
"id": "1",
|
|
"method": "session.steer",
|
|
"params": {"session_id": "sid", "text": "hi"},
|
|
}
|
|
)
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
assert "error" in resp, resp
|
|
assert resp["error"]["code"] == 4010
|
|
|
|
|
|
def test_session_info_includes_mcp_servers(monkeypatch):
|
|
fake_status = [
|
|
{"name": "github", "transport": "http", "tools": 12, "connected": True},
|
|
{"name": "filesystem", "transport": "stdio", "tools": 4, "connected": True},
|
|
{"name": "broken", "transport": "stdio", "tools": 0, "connected": False},
|
|
]
|
|
fake_mod = types.ModuleType("tools.mcp_tool")
|
|
fake_mod.get_mcp_status = lambda: fake_status
|
|
monkeypatch.setitem(sys.modules, "tools.mcp_tool", fake_mod)
|
|
|
|
info = server._session_info(types.SimpleNamespace(tools=[], model=""))
|
|
|
|
assert info["mcp_servers"] == fake_status
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# History-mutating commands must reject while session.running is True.
|
|
# Without these guards, prompt.submit's post-run history write either
|
|
# clobbers the mutation (version matches) or silently drops the agent's
|
|
# output (version mismatch) — both produce UI<->backend state desync.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_session_undo_rejects_while_running():
|
|
"""Fix for TUI silent-drop #1: /undo must not mutate history
|
|
while the agent is mid-turn — would either clobber the undo or
|
|
cause prompt.submit to silently drop the agent's response."""
|
|
server._sessions["sid"] = _session(running=True, history=[
|
|
{"role": "user", "content": "hi"},
|
|
{"role": "assistant", "content": "hello"},
|
|
])
|
|
try:
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "session.undo", "params": {"session_id": "sid"}}
|
|
)
|
|
assert resp.get("error"), "session.undo should reject while running"
|
|
assert resp["error"]["code"] == 4009
|
|
assert "session busy" in resp["error"]["message"]
|
|
# History must be unchanged
|
|
assert len(server._sessions["sid"]["history"]) == 2
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_session_undo_allowed_when_idle():
|
|
"""Regression guard: when not running, /undo still works."""
|
|
server._sessions["sid"] = _session(running=False, history=[
|
|
{"role": "user", "content": "hi"},
|
|
{"role": "assistant", "content": "hello"},
|
|
])
|
|
try:
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "session.undo", "params": {"session_id": "sid"}}
|
|
)
|
|
assert resp.get("result"), f"got error: {resp.get('error')}"
|
|
assert resp["result"]["removed"] == 2
|
|
assert server._sessions["sid"]["history"] == []
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_session_compress_rejects_while_running(monkeypatch):
|
|
server._sessions["sid"] = _session(running=True)
|
|
try:
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "session.compress", "params": {"session_id": "sid"}}
|
|
)
|
|
assert resp.get("error")
|
|
assert resp["error"]["code"] == 4009
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_rollback_restore_rejects_full_history_while_running(monkeypatch):
|
|
"""Full-history rollback must reject; file-scoped rollback still allowed."""
|
|
server._sessions["sid"] = _session(running=True)
|
|
try:
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "rollback.restore", "params": {"session_id": "sid", "hash": "abc"}}
|
|
)
|
|
assert resp.get("error"), "full-history rollback should reject while running"
|
|
assert resp["error"]["code"] == 4009
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_prompt_submit_history_version_mismatch_surfaces_warning(monkeypatch):
|
|
"""Fix for TUI silent-drop #2: the defensive backstop at prompt.submit
|
|
must attach a 'warning' to message.complete when history was
|
|
mutated externally during the turn (instead of silently dropping
|
|
the agent's output)."""
|
|
# Agent bumps history_version itself mid-run to simulate an external
|
|
# mutation slipping past the guards.
|
|
session_ref = {"s": None}
|
|
|
|
class _RacyAgent:
|
|
def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
|
|
# Simulate: something external bumped history_version
|
|
# while we were running.
|
|
with session_ref["s"]["history_lock"]:
|
|
session_ref["s"]["history_version"] += 1
|
|
return {"final_response": "agent reply", "messages": [{"role": "assistant", "content": "agent reply"}]}
|
|
|
|
class _ImmediateThread:
|
|
def __init__(self, target=None, daemon=None):
|
|
self._target = target
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
server._sessions["sid"] = _session(agent=_RacyAgent())
|
|
session_ref["s"] = server._sessions["sid"]
|
|
emits: list[tuple] = []
|
|
try:
|
|
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
|
|
monkeypatch.setattr(server, "_get_usage", lambda _a: {})
|
|
monkeypatch.setattr(server, "render_message", lambda _t, _c: "")
|
|
monkeypatch.setattr(server, "_emit", lambda *a: emits.append(a))
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "hi"}}
|
|
)
|
|
assert resp.get("result"), f"got error: {resp.get('error')}"
|
|
|
|
# History should NOT contain the agent's output (version mismatch)
|
|
assert server._sessions["sid"]["history"] == []
|
|
|
|
# message.complete must carry a 'warning' so the UI / operator
|
|
# knows the output was not persisted.
|
|
complete_calls = [a for a in emits if a[0] == "message.complete"]
|
|
assert len(complete_calls) == 1
|
|
_, _, payload = complete_calls[0]
|
|
assert "warning" in payload, (
|
|
"message.complete must include a 'warning' field on "
|
|
"history_version mismatch — otherwise the UI silently "
|
|
"shows output that was never persisted"
|
|
)
|
|
assert "not saved" in payload["warning"].lower() or "changed" in payload["warning"].lower()
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_prompt_submit_history_version_match_persists_normally(monkeypatch):
|
|
"""Regression guard: the backstop does not affect the happy path."""
|
|
class _Agent:
|
|
def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
|
|
return {"final_response": "reply", "messages": [{"role": "assistant", "content": "reply"}]}
|
|
|
|
class _ImmediateThread:
|
|
def __init__(self, target=None, daemon=None):
|
|
self._target = target
|
|
|
|
def start(self):
|
|
self._target()
|
|
|
|
server._sessions["sid"] = _session(agent=_Agent())
|
|
emits: list[tuple] = []
|
|
try:
|
|
monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
|
|
monkeypatch.setattr(server, "_get_usage", lambda _a: {})
|
|
monkeypatch.setattr(server, "render_message", lambda _t, _c: "")
|
|
monkeypatch.setattr(server, "_emit", lambda *a: emits.append(a))
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "hi"}}
|
|
)
|
|
assert resp.get("result")
|
|
|
|
# History was written
|
|
assert server._sessions["sid"]["history"] == [{"role": "assistant", "content": "reply"}]
|
|
assert server._sessions["sid"]["history_version"] == 1
|
|
|
|
# No warning should be attached
|
|
complete_calls = [a for a in emits if a[0] == "message.complete"]
|
|
assert len(complete_calls) == 1
|
|
_, _, payload = complete_calls[0]
|
|
assert "warning" not in payload
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# session.interrupt must only cancel pending prompts owned by the calling
|
|
# session — it must not blast-resolve clarify/sudo/secret prompts on
|
|
# unrelated sessions sharing the same tui_gateway process. Without
|
|
# session scoping the other sessions' prompts silently resolve to empty
|
|
# strings, unblocking their agent threads as if the user cancelled.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_interrupt_only_clears_own_session_pending():
|
|
"""session.interrupt on session A must NOT release pending prompts
|
|
that belong to session B."""
|
|
import types
|
|
|
|
session_a = _session()
|
|
session_a["agent"] = types.SimpleNamespace(interrupt=lambda: None)
|
|
session_b = _session()
|
|
session_b["agent"] = types.SimpleNamespace(interrupt=lambda: None)
|
|
server._sessions["sid_a"] = session_a
|
|
server._sessions["sid_b"] = session_b
|
|
|
|
try:
|
|
# Simulate pending prompts on both sessions (what _block creates
|
|
# while a clarify/sudo/secret request is outstanding).
|
|
ev_a = threading.Event()
|
|
ev_b = threading.Event()
|
|
server._pending["rid-a"] = ("sid_a", ev_a)
|
|
server._pending["rid-b"] = ("sid_b", ev_b)
|
|
server._answers.clear()
|
|
|
|
# Interrupt session A.
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "session.interrupt", "params": {"session_id": "sid_a"}}
|
|
)
|
|
assert resp.get("result"), f"got error: {resp.get('error')}"
|
|
|
|
# Session A's pending must be released to empty.
|
|
assert ev_a.is_set(), "sid_a pending Event should be set after interrupt"
|
|
assert server._answers.get("rid-a") == ""
|
|
|
|
# Session B's pending MUST remain untouched — no cross-session blast.
|
|
assert not ev_b.is_set(), (
|
|
"CRITICAL: session.interrupt on sid_a released a pending prompt "
|
|
"belonging to sid_b — other sessions' clarify/sudo/secret "
|
|
"prompts are being silently cancelled"
|
|
)
|
|
assert "rid-b" not in server._answers
|
|
finally:
|
|
server._sessions.pop("sid_a", None)
|
|
server._sessions.pop("sid_b", None)
|
|
server._pending.pop("rid-a", None)
|
|
server._pending.pop("rid-b", None)
|
|
server._answers.pop("rid-a", None)
|
|
server._answers.pop("rid-b", None)
|
|
|
|
|
|
def test_interrupt_clears_multiple_own_pending():
|
|
"""When a single session has multiple pending prompts (uncommon but
|
|
possible via nested tool calls), interrupt must release all of them."""
|
|
import types
|
|
|
|
sess = _session()
|
|
sess["agent"] = types.SimpleNamespace(interrupt=lambda: None)
|
|
server._sessions["sid"] = sess
|
|
|
|
try:
|
|
ev1, ev2 = threading.Event(), threading.Event()
|
|
server._pending["r1"] = ("sid", ev1)
|
|
server._pending["r2"] = ("sid", ev2)
|
|
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "session.interrupt", "params": {"session_id": "sid"}}
|
|
)
|
|
assert resp.get("result")
|
|
assert ev1.is_set() and ev2.is_set()
|
|
assert server._answers.get("r1") == "" and server._answers.get("r2") == ""
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
for key in ("r1", "r2"):
|
|
server._pending.pop(key, None)
|
|
server._answers.pop(key, None)
|
|
|
|
|
|
def test_clear_pending_without_sid_clears_all():
|
|
"""_clear_pending(None) is the shutdown path — must still release
|
|
every pending prompt regardless of owning session."""
|
|
ev1, ev2, ev3 = threading.Event(), threading.Event(), threading.Event()
|
|
server._pending["a"] = ("sid_x", ev1)
|
|
server._pending["b"] = ("sid_y", ev2)
|
|
server._pending["c"] = ("sid_z", ev3)
|
|
try:
|
|
server._clear_pending(None)
|
|
assert ev1.is_set() and ev2.is_set() and ev3.is_set()
|
|
finally:
|
|
for key in ("a", "b", "c"):
|
|
server._pending.pop(key, None)
|
|
server._answers.pop(key, None)
|
|
|
|
|
|
def test_respond_unpacks_sid_tuple_correctly():
|
|
"""After the (sid, Event) tuple change, _respond must still work."""
|
|
ev = threading.Event()
|
|
server._pending["rid-x"] = ("sid_x", ev)
|
|
try:
|
|
resp = server.handle_request(
|
|
{"id": "1", "method": "clarify.respond",
|
|
"params": {"request_id": "rid-x", "answer": "the answer"}}
|
|
)
|
|
assert resp.get("result")
|
|
assert ev.is_set()
|
|
assert server._answers.get("rid-x") == "the answer"
|
|
finally:
|
|
server._pending.pop("rid-x", None)
|
|
server._answers.pop("rid-x", None)
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# /model switch and other agent-mutating commands must reject while the
|
|
# session is running. agent.switch_model() mutates self.model, self.provider,
|
|
# self.base_url, self.client etc. in place — the worker thread running
|
|
# agent.run_conversation is reading those on every iteration. Same class of
|
|
# bug as the session.undo / session.compress mid-run silent-drop; same fix
|
|
# pattern: reject with 4009 while running.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_config_set_model_rejects_while_running(monkeypatch):
|
|
"""/model via config.set must reject during an in-flight turn."""
|
|
seen = {"called": False}
|
|
|
|
def _fake_apply(sid, session, raw):
|
|
seen["called"] = True
|
|
return {"value": raw, "warning": ""}
|
|
|
|
monkeypatch.setattr(server, "_apply_model_switch", _fake_apply)
|
|
|
|
server._sessions["sid"] = _session(running=True)
|
|
try:
|
|
resp = server.handle_request({
|
|
"id": "1", "method": "config.set",
|
|
"params": {"session_id": "sid", "key": "model", "value": "anthropic/claude-sonnet-4.6"},
|
|
})
|
|
assert resp.get("error")
|
|
assert resp["error"]["code"] == 4009
|
|
assert "session busy" in resp["error"]["message"]
|
|
assert not seen["called"], (
|
|
"_apply_model_switch was called mid-turn — would race with "
|
|
"the worker thread reading agent.model / agent.client"
|
|
)
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_config_set_model_allowed_when_idle(monkeypatch):
|
|
"""Regression guard: idle sessions can still switch models."""
|
|
seen = {"called": False}
|
|
|
|
def _fake_apply(sid, session, raw):
|
|
seen["called"] = True
|
|
return {"value": "newmodel", "warning": ""}
|
|
|
|
monkeypatch.setattr(server, "_apply_model_switch", _fake_apply)
|
|
|
|
server._sessions["sid"] = _session(running=False)
|
|
try:
|
|
resp = server.handle_request({
|
|
"id": "1", "method": "config.set",
|
|
"params": {"session_id": "sid", "key": "model", "value": "newmodel"},
|
|
})
|
|
assert resp.get("result")
|
|
assert resp["result"]["value"] == "newmodel"
|
|
assert seen["called"]
|
|
finally:
|
|
server._sessions.pop("sid", None)
|
|
|
|
|
|
def test_mirror_slash_side_effects_rejects_mutating_commands_while_running(monkeypatch):
|
|
"""Slash worker passthrough (e.g. /model, /personality, /prompt,
|
|
/compress) must reject during an in-flight turn. Same race as
|
|
config.set — mutates live agent state while run_conversation is
|
|
reading it."""
|
|
import types
|
|
|
|
applied = {"model": False, "compress": False}
|
|
|
|
def _fake_apply_model(sid, session, arg):
|
|
applied["model"] = True
|
|
return {"value": arg, "warning": ""}
|
|
|
|
def _fake_compress(session, focus):
|
|
applied["compress"] = True
|
|
return (0, {})
|
|
|
|
monkeypatch.setattr(server, "_apply_model_switch", _fake_apply_model)
|
|
monkeypatch.setattr(server, "_compress_session_history", _fake_compress)
|
|
|
|
session = _session(running=True)
|
|
session["agent"] = types.SimpleNamespace(model="x")
|
|
|
|
for cmd, expected_name in [
|
|
("/model new/model", "model"),
|
|
("/personality default", "personality"),
|
|
("/prompt", "prompt"),
|
|
("/compress", "compress"),
|
|
]:
|
|
warning = server._mirror_slash_side_effects("sid", session, cmd)
|
|
assert "session busy" in warning, (
|
|
f"{cmd} should have returned busy warning, got: {warning!r}"
|
|
)
|
|
assert f"/{expected_name}" in warning
|
|
|
|
# None of the mutating side-effect helpers should have fired.
|
|
assert not applied["model"], "model switch fired despite running session"
|
|
assert not applied["compress"], "compress fired despite running session"
|
|
|
|
|
|
def test_mirror_slash_side_effects_allowed_when_idle(monkeypatch):
|
|
"""Regression guard: idle session still runs the side effects."""
|
|
import types
|
|
|
|
applied = {"model": False}
|
|
|
|
def _fake_apply_model(sid, session, arg):
|
|
applied["model"] = True
|
|
return {"value": arg, "warning": ""}
|
|
|
|
monkeypatch.setattr(server, "_apply_model_switch", _fake_apply_model)
|
|
|
|
session = _session(running=False)
|
|
session["agent"] = types.SimpleNamespace(model="x")
|
|
|
|
warning = server._mirror_slash_side_effects("sid", session, "/model foo")
|
|
# Should NOT contain "session busy" — the switch went through.
|
|
assert "session busy" not in warning
|
|
assert applied["model"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# session.create / session.close race: fast /new churn must not orphan the
|
|
# slash_worker subprocess or the global approval-notify registration.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_session_create_close_race_does_not_orphan_worker(monkeypatch):
|
|
"""Regression guard: if session.close runs while session.create's
|
|
_build thread is still constructing the agent, the build thread
|
|
must detect the orphan and clean up the slash_worker + notify
|
|
registration it's about to install. Without the cleanup those
|
|
resources leak — the subprocess stays alive until atexit and the
|
|
notify callback lingers in the global registry."""
|
|
import threading
|
|
|
|
closed_workers: list[str] = []
|
|
unregistered_keys: list[str] = []
|
|
|
|
class _FakeWorker:
|
|
def __init__(self, key, model):
|
|
self.key = key
|
|
self._closed = False
|
|
|
|
def close(self):
|
|
self._closed = True
|
|
closed_workers.append(self.key)
|
|
|
|
class _FakeAgent:
|
|
def __init__(self):
|
|
self.model = "x"
|
|
self.provider = "openrouter"
|
|
self.base_url = ""
|
|
self.api_key = ""
|
|
|
|
# Make _build block until we release it — simulates slow agent init
|
|
release_build = threading.Event()
|
|
|
|
def _slow_make_agent(sid, key):
|
|
release_build.wait(timeout=3.0)
|
|
return _FakeAgent()
|
|
|
|
# Stub everything _build touches
|
|
monkeypatch.setattr(server, "_make_agent", _slow_make_agent)
|
|
monkeypatch.setattr(server, "_SlashWorker", _FakeWorker)
|
|
monkeypatch.setattr(server, "_get_db", lambda: types.SimpleNamespace(create_session=lambda *a, **kw: None))
|
|
monkeypatch.setattr(server, "_session_info", lambda _a: {"model": "x"})
|
|
monkeypatch.setattr(server, "_probe_credentials", lambda _a: None)
|
|
monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
|
|
monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
|
|
|
|
# Shim register/unregister to observe leaks
|
|
import tools.approval as _approval
|
|
monkeypatch.setattr(_approval, "register_gateway_notify",
|
|
lambda key, cb: None)
|
|
monkeypatch.setattr(_approval, "unregister_gateway_notify",
|
|
lambda key: unregistered_keys.append(key))
|
|
monkeypatch.setattr(_approval, "load_permanent_allowlist", lambda: None)
|
|
|
|
# Start: session.create spawns _build thread, returns synchronously
|
|
resp = server.handle_request({
|
|
"id": "1", "method": "session.create", "params": {"cols": 80},
|
|
})
|
|
assert resp.get("result"), f"got error: {resp.get('error')}"
|
|
sid = resp["result"]["session_id"]
|
|
|
|
# Build thread is blocked in _slow_make_agent. Close the session
|
|
# NOW — this pops _sessions[sid] before _build can install the
|
|
# worker/notify.
|
|
close_resp = server.handle_request({
|
|
"id": "2", "method": "session.close", "params": {"session_id": sid},
|
|
})
|
|
assert close_resp.get("result", {}).get("closed") is True
|
|
|
|
# At this point session.close saw slash_worker=None (not yet
|
|
# installed) so it didn't close anything. Release the build thread
|
|
# and let it finish — it should detect the orphan and clean up the
|
|
# worker it just allocated + unregister the notify.
|
|
release_build.set()
|
|
|
|
# Give the build thread a moment to run through its finally.
|
|
for _ in range(100):
|
|
if closed_workers:
|
|
break
|
|
import time
|
|
time.sleep(0.02)
|
|
|
|
assert len(closed_workers) == 1, (
|
|
f"orphan worker was not cleaned up — closed_workers={closed_workers}"
|
|
)
|
|
# Notify may be unregistered by both session.close (unconditional)
|
|
# and the orphan-cleanup path; the key guarantee is that the build
|
|
# thread does at least one unregister call (any prior close
|
|
# already popped the callback; the duplicate is a no-op).
|
|
assert len(unregistered_keys) >= 1, (
|
|
f"orphan notify registration was not unregistered — "
|
|
f"unregistered_keys={unregistered_keys}"
|
|
)
|
|
|
|
|
|
def test_session_create_no_race_keeps_worker_alive(monkeypatch):
|
|
"""Regression guard: when session.close does NOT race, the build
|
|
thread must install the worker + notify normally and leave them
|
|
alone (no over-eager cleanup)."""
|
|
closed_workers: list[str] = []
|
|
unregistered_keys: list[str] = []
|
|
|
|
class _FakeWorker:
|
|
def __init__(self, key, model):
|
|
self.key = key
|
|
|
|
def close(self):
|
|
closed_workers.append(self.key)
|
|
|
|
class _FakeAgent:
|
|
def __init__(self):
|
|
self.model = "x"
|
|
self.provider = "openrouter"
|
|
self.base_url = ""
|
|
self.api_key = ""
|
|
|
|
monkeypatch.setattr(server, "_make_agent", lambda sid, key: _FakeAgent())
|
|
monkeypatch.setattr(server, "_SlashWorker", _FakeWorker)
|
|
monkeypatch.setattr(server, "_get_db", lambda: types.SimpleNamespace(create_session=lambda *a, **kw: None))
|
|
monkeypatch.setattr(server, "_session_info", lambda _a: {"model": "x"})
|
|
monkeypatch.setattr(server, "_probe_credentials", lambda _a: None)
|
|
monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
|
|
monkeypatch.setattr(server, "_emit", lambda *a, **kw: None)
|
|
|
|
import tools.approval as _approval
|
|
monkeypatch.setattr(_approval, "register_gateway_notify", lambda key, cb: None)
|
|
monkeypatch.setattr(_approval, "unregister_gateway_notify",
|
|
lambda key: unregistered_keys.append(key))
|
|
monkeypatch.setattr(_approval, "load_permanent_allowlist", lambda: None)
|
|
|
|
resp = server.handle_request({
|
|
"id": "1", "method": "session.create", "params": {"cols": 80},
|
|
})
|
|
sid = resp["result"]["session_id"]
|
|
|
|
# Wait for the build to finish (ready event inside session dict).
|
|
session = server._sessions[sid]
|
|
session["agent_ready"].wait(timeout=2.0)
|
|
|
|
# Build finished without a close race — nothing should have been
|
|
# cleaned up by the orphan check.
|
|
assert closed_workers == [], (
|
|
f"build thread closed its own worker despite no race: {closed_workers}"
|
|
)
|
|
assert unregistered_keys == [], (
|
|
f"build thread unregistered its own notify despite no race: {unregistered_keys}"
|
|
)
|
|
|
|
# Session should have the live worker installed.
|
|
assert session.get("slash_worker") is not None
|
|
|
|
# Cleanup
|
|
server._sessions.pop(sid, None)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# model.options — curated-list parity with `hermes model` and classic /model
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
def test_model_options_does_not_overwrite_curated_models(monkeypatch):
|
|
"""The TUI model.options handler must surface the same curated model
|
|
list as `hermes model` and the classic CLI /model picker.
|
|
|
|
Regression: earlier versions of this handler unconditionally replaced
|
|
each provider's curated ``models`` field with ``provider_model_ids()``
|
|
(live /models catalog). That pulled in hundreds of non-agentic models
|
|
for providers like Nous whose /models endpoint returns image/video
|
|
generators, rerankers, embeddings, and TTS models alongside chat models.
|
|
"""
|
|
curated_providers = [
|
|
{
|
|
"slug": "nous",
|
|
"name": "Nous",
|
|
"models": ["moonshotai/kimi-k2.5", "anthropic/claude-opus-4.7"],
|
|
"total_models": 30,
|
|
"source": "built-in",
|
|
"is_current": False,
|
|
"is_user_defined": False,
|
|
},
|
|
]
|
|
|
|
monkeypatch.setattr(
|
|
server,
|
|
"_load_cfg",
|
|
lambda: {"providers": {}, "custom_providers": []},
|
|
)
|
|
|
|
with patch(
|
|
"hermes_cli.model_switch.list_authenticated_providers",
|
|
return_value=curated_providers,
|
|
) as listing:
|
|
# If provider_model_ids gets called at all, the handler is still
|
|
# overwriting curated with live — that's the regression we're
|
|
# guarding against.
|
|
with patch("hermes_cli.models.provider_model_ids") as live_fetch:
|
|
resp = server._methods["model.options"](99, {"session_id": ""})
|
|
|
|
assert "result" in resp, resp
|
|
providers = resp["result"]["providers"]
|
|
nous = next((p for p in providers if p.get("slug") == "nous"), None)
|
|
assert nous is not None
|
|
assert nous["models"] == [
|
|
"moonshotai/kimi-k2.5",
|
|
"anthropic/claude-opus-4.7",
|
|
]
|
|
assert nous["total_models"] == 30
|
|
# Handler must not consult the live catalog — curated is the truth.
|
|
live_fetch.assert_not_called()
|
|
# list_authenticated_providers is the single source.
|
|
assert listing.call_count == 1
|
|
|
|
|
|
def test_model_options_propagates_list_exception(monkeypatch):
|
|
"""If list_authenticated_providers itself raises, surface as an RPC
|
|
error rather than swallowing to a blank picker."""
|
|
monkeypatch.setattr(
|
|
server,
|
|
"_load_cfg",
|
|
lambda: {"providers": {}, "custom_providers": []},
|
|
)
|
|
with patch(
|
|
"hermes_cli.model_switch.list_authenticated_providers",
|
|
side_effect=RuntimeError("catalog blew up"),
|
|
):
|
|
resp = server._methods["model.options"](77, {"session_id": ""})
|
|
assert "error" in resp
|
|
assert resp["error"]["code"] == 5033
|
|
assert "catalog blew up" in resp["error"]["message"]
|