1693fca97a
Change-detectors, tautologies, source-reading tests, redundant duplicates, mock-echo tests and dead/unrunnable tests. Per-test rationale in the lane ledger (category + reason for every removal).
468 lines
18 KiB
Python
468 lines
18 KiB
Python
"""Behavioral contract tests for Desktop session-control JSON-RPC methods."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.fixture()
|
|
def hermes_home(tmp_path, monkeypatch):
|
|
"""Give the persisted control managers an isolated database for every test."""
|
|
home = tmp_path / ".hermes"
|
|
home.mkdir()
|
|
monkeypatch.setattr("pathlib.Path.home", lambda: tmp_path)
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
from hermes_cli import goals
|
|
|
|
goals._DB_CACHE.clear()
|
|
yield home
|
|
goals._DB_CACHE.clear()
|
|
|
|
|
|
@pytest.fixture()
|
|
def server(hermes_home, monkeypatch):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"hermes_cli.env_loader": MagicMock(),
|
|
"hermes_cli.banner": MagicMock(),
|
|
},
|
|
):
|
|
mod = importlib.import_module("tui_gateway.server")
|
|
monkeypatch.setattr(mod, "_hermes_home", hermes_home)
|
|
monkeypatch.setattr(mod, "_cfg_cache", None)
|
|
monkeypatch.setattr(mod, "_cfg_sig", None)
|
|
monkeypatch.setattr(mod, "_cfg_path", None)
|
|
yield mod
|
|
mod._sessions.clear()
|
|
__import__("tui_gateway.server_requests", fromlist=["x"]).reset_for_tests()
|
|
|
|
|
|
@pytest.fixture()
|
|
def session(server):
|
|
sid = f"sid-control-{uuid.uuid4().hex}"
|
|
key = f"control-{uuid.uuid4().hex}"
|
|
entry = {
|
|
"session_key": key,
|
|
"history": [],
|
|
"history_lock": threading.Lock(),
|
|
"history_version": 0,
|
|
"running": False,
|
|
"attached_images": [],
|
|
"cols": 120,
|
|
"agent": None,
|
|
"created_at": time.time(),
|
|
}
|
|
server._sessions[sid] = entry
|
|
yield sid, key, entry
|
|
from hermes_cli.goals import GoalManager
|
|
from hermes_cli.heartbeat import HeartbeatManager
|
|
from hermes_cli.loops import LoopManager
|
|
|
|
GoalManager(key).clear()
|
|
LoopManager(key).clear()
|
|
HeartbeatManager(key).clear()
|
|
|
|
|
|
def _call(server, method, *, rid=91, **params):
|
|
return server._methods[method](rid, params)
|
|
|
|
|
|
def _control(server, sid):
|
|
return _call(server, "session.control.read", session_id=sid)["result"]["control"]
|
|
|
|
|
|
def _error(response):
|
|
assert "error" in response
|
|
return response["error"]
|
|
|
|
|
|
def _observe_dispatch(server, monkeypatch):
|
|
calls = []
|
|
original = server._methods["command.dispatch"]
|
|
|
|
def observe(rid, params):
|
|
calls.append((rid, dict(params)))
|
|
return original(rid, params)
|
|
|
|
monkeypatch.setitem(server._methods, "command.dispatch", observe)
|
|
return calls
|
|
|
|
|
|
def _forbid_dispatch(server, monkeypatch):
|
|
def forbidden(_rid, _params):
|
|
raise AssertionError("manager-only action must not call command.dispatch")
|
|
|
|
monkeypatch.setitem(server._methods, "command.dispatch", forbidden)
|
|
|
|
|
|
def _save_goal(key, **overrides):
|
|
from hermes_cli.goals import GoalState, save_goal
|
|
|
|
fields = {
|
|
"goal": "Finish the desktop control card",
|
|
"status": "active",
|
|
"turns_used": 3,
|
|
"max_turns": 12,
|
|
"created_at": 100.0,
|
|
"last_turn_at": 200.0,
|
|
}
|
|
fields.update(overrides)
|
|
state = GoalState(**fields)
|
|
save_goal(key, state)
|
|
return state
|
|
|
|
|
|
def _save_loop(key, **overrides):
|
|
from hermes_cli.loops import LoopState, save_loop
|
|
|
|
fields = {
|
|
"prompt": "Check the deployment",
|
|
"status": "active",
|
|
"mode": "interval",
|
|
"interval_seconds": 300,
|
|
"current_delay": 300,
|
|
"created_at": 100.0,
|
|
"next_due_at": 400.0,
|
|
}
|
|
fields.update(overrides)
|
|
state = LoopState(**fields)
|
|
save_loop(key, state)
|
|
return state
|
|
|
|
|
|
def _save_heartbeat(key, **overrides):
|
|
from hermes_cli.heartbeat import HeartbeatState, save_heartbeat
|
|
|
|
fields = {
|
|
"prompt": "Check the deployment",
|
|
"interval_seconds": 600,
|
|
"status": "active",
|
|
"created_at": 100.0,
|
|
"last_fired_at": 150.0,
|
|
"fire_count": 2,
|
|
}
|
|
fields.update(overrides)
|
|
state = HeartbeatState(**fields)
|
|
save_heartbeat(key, state)
|
|
return state
|
|
|
|
|
|
class TestStructuredRead:
|
|
def test_methods_are_registered_and_empty_snapshot_is_stable(self, server, session):
|
|
sid, _, _ = session
|
|
first = _control(server, sid)
|
|
second = _control(server, sid)
|
|
assert first == second
|
|
assert first == {
|
|
"goal": None,
|
|
"loop": None,
|
|
"heartbeat": None,
|
|
"revision": "",
|
|
"updated_at": 0,
|
|
}
|
|
|
|
def test_goal_contract_subgoals_and_gates_are_structured_and_sanitized(self, server, session):
|
|
from hermes_cli.goals import GoalContract, GoalGate
|
|
|
|
sid, key, _ = session
|
|
contract = GoalContract(outcome="Card is correct", verification="Run focused tests")
|
|
_save_goal(
|
|
key,
|
|
contract=contract,
|
|
subgoals=["Keep command routing narrow", "Document event hydration seam"],
|
|
gates=[GoalGate(
|
|
command="scripts/run_tests.sh tests/tui_gateway/test_session_control.py",
|
|
timeout_seconds=90,
|
|
max_retries=2,
|
|
attempts=1,
|
|
last_exit_code=1,
|
|
last_output_tail="private output must stay private",
|
|
)],
|
|
)
|
|
|
|
goal = _control(server, sid)["goal"]
|
|
assert goal["title"] == "Finish the desktop control card"
|
|
assert goal["contract"] == contract.to_dict()
|
|
assert goal["subgoals"] == ["Keep command routing narrow", "Document event hydration seam"]
|
|
assert goal["gates"] == [{
|
|
"command": "scripts/run_tests.sh tests/tui_gateway/test_session_control.py",
|
|
"timeout_seconds": 90,
|
|
"max_retries": 2,
|
|
"attempts": 1,
|
|
"last_exit_code": 1,
|
|
}]
|
|
serialized = json.dumps(goal)
|
|
for forbidden in (
|
|
"last_output_tail", "private output",
|
|
"route", "session_id", "credential", "api_key",
|
|
):
|
|
assert forbidden not in serialized
|
|
|
|
@pytest.mark.parametrize(
|
|
("kind", "setup"),
|
|
[
|
|
("goal", lambda key: _save_goal(key)),
|
|
("loop", lambda key: _save_loop(key)),
|
|
("heartbeat", lambda key: _save_heartbeat(key)),
|
|
],
|
|
)
|
|
def test_revision_is_stable_then_changes_for_each_visible_control(self, server, session, kind, setup):
|
|
sid, key, _ = session
|
|
setup(key)
|
|
before = _control(server, sid)
|
|
assert isinstance(before["revision"], str) and before["revision"]
|
|
assert before == _control(server, sid)
|
|
|
|
if kind == "goal":
|
|
_save_goal(key, goal="A visible replacement goal")
|
|
elif kind == "loop":
|
|
_save_loop(key, prompt="A visibly changed loop prompt")
|
|
else:
|
|
_save_heartbeat(key, prompt="A visibly changed heartbeat prompt")
|
|
|
|
assert _control(server, sid)["revision"] != before["revision"]
|
|
|
|
class TestDispatcherBackedMutations:
|
|
def test_goal_pause_resume_clear_mutate_real_persisted_state(self, server, session):
|
|
sid, key, _ = session
|
|
_save_goal(key, status="active", turns_used=8)
|
|
|
|
paused = _call(server, "session.control", session_id=sid, action="goal.pause")
|
|
assert paused["result"]["control"]["goal"]["status"] == "paused"
|
|
|
|
resumed = _call(server, "session.control", session_id=sid, action="goal.resume")
|
|
assert resumed["result"]["dispatch"]["type"] == "send"
|
|
assert resumed["result"]["dispatch"]["message"]
|
|
assert resumed["result"]["dispatch"]["notice"]
|
|
assert resumed["result"]["dispatch"]["display"] == "/goal resume"
|
|
assert resumed["result"]["control"]["goal"]["status"] == "active"
|
|
|
|
cleared = _call(server, "session.control", session_id=sid, action="goal.clear")
|
|
assert cleared["result"]["control"]["goal"] is None
|
|
|
|
def test_loop_pause_resume_stop_mutate_real_persisted_state(self, server, session):
|
|
sid, key, _ = session
|
|
_save_loop(key)
|
|
|
|
paused = _call(server, "session.control", session_id=sid, action="loop.pause")
|
|
assert paused["result"]["control"]["loop"]["status"] == "paused"
|
|
|
|
resumed = _call(server, "session.control", session_id=sid, action="loop.resume")
|
|
assert resumed["result"]["control"]["loop"]["status"] == "active"
|
|
|
|
stopped = _call(server, "session.control", session_id=sid, action="loop.stop")
|
|
assert stopped["result"]["control"]["loop"] is None
|
|
|
|
|
|
class TestManagerOnlyMutations:
|
|
def test_subgoal_add_remove_clear_are_real_one_based_mutations_without_dispatch(self, server, session, monkeypatch):
|
|
from hermes_cli.goals import load_goal
|
|
|
|
sid, key, _ = session
|
|
_save_goal(key, subgoals=["First criterion"])
|
|
_forbid_dispatch(server, monkeypatch)
|
|
|
|
added = _call(server, "session.control", session_id=sid, action="subgoal.add", args={"text": "Second criterion"})
|
|
assert "error" not in added
|
|
assert load_goal(key).subgoals == ["First criterion", "Second criterion"]
|
|
|
|
removed = _call(server, "session.control", session_id=sid, action="subgoal.remove", args={"index": 1})
|
|
assert "error" not in removed
|
|
assert load_goal(key).subgoals == ["Second criterion"]
|
|
|
|
cleared = _call(server, "session.control", session_id=sid, action="subgoal.clear")
|
|
assert "error" not in cleared
|
|
assert load_goal(key).subgoals == []
|
|
|
|
def test_subgoal_requires_a_goal_and_valid_arguments_without_dispatch(self, server, session, monkeypatch):
|
|
sid, key, _ = session
|
|
_forbid_dispatch(server, monkeypatch)
|
|
assert _error(_call(server, "session.control", session_id=sid, action="subgoal.add", args={"text": "criterion"}))["code"] == 4004
|
|
|
|
_save_goal(key, subgoals=["Only criterion"])
|
|
for args in (
|
|
{"text": " "},
|
|
{"index": "1"},
|
|
{"index": 1.5},
|
|
{"index": True},
|
|
{"index": 0},
|
|
{"index": 2},
|
|
):
|
|
action = "subgoal.add" if "text" in args else "subgoal.remove"
|
|
assert _error(_call(server, "session.control", session_id=sid, action=action, args=args))["code"] == 4004
|
|
|
|
def test_goal_unwait_clears_the_real_barrier_through_shared_command(self, server, session):
|
|
from hermes_cli.goals import GoalManager
|
|
|
|
sid, key, _ = session
|
|
_save_goal(key)
|
|
GoalManager(key).wait_for_seconds(60, reason="backoff")
|
|
|
|
response = _call(server, "session.control", session_id=sid, action="goal.unwait")
|
|
assert "error" not in response
|
|
assert GoalManager(key).state.waiting_until == 0.0
|
|
|
|
def test_heartbeat_pause_resume_clear_and_no_heartbeat_messages_do_not_dispatch(self, server, session, monkeypatch):
|
|
sid, key, _ = session
|
|
_forbid_dispatch(server, monkeypatch)
|
|
for action in ("heartbeat.pause", "heartbeat.resume", "heartbeat.clear"):
|
|
empty = _call(server, "session.control", session_id=sid, action=action)
|
|
assert "error" not in empty and empty["result"]["control"]["heartbeat"] is None
|
|
|
|
_save_heartbeat(key, status="active", last_fired_at=1.0)
|
|
paused = _call(server, "session.control", session_id=sid, action="heartbeat.pause")
|
|
assert paused["result"]["control"]["heartbeat"]["status"] == "paused"
|
|
|
|
resumed = _call(server, "session.control", session_id=sid, action="heartbeat.resume")
|
|
assert resumed["result"]["control"]["heartbeat"]["last_fired_at"] > 1.0
|
|
|
|
cleared = _call(server, "session.control", session_id=sid, action="heartbeat.clear")
|
|
assert cleared["result"]["control"]["heartbeat"] is None
|
|
|
|
|
|
class TestErrorsAndEvents:
|
|
@pytest.mark.parametrize(
|
|
("action", "args"),
|
|
[
|
|
("", {}),
|
|
("not.allowed", {}),
|
|
("goal.gate.add", {"command": "echo should-not-run"}),
|
|
("subgoal.add", []),
|
|
],
|
|
)
|
|
def test_invalid_actions_and_malformed_args_return_4004_without_dispatch(self, server, session, monkeypatch, action, args):
|
|
sid, _, _ = session
|
|
_forbid_dispatch(server, monkeypatch)
|
|
assert _error(_call(server, "session.control", session_id=sid, action=action, args=args))["code"] == 4004
|
|
|
|
def test_unknown_session_returns_4001(self, server):
|
|
assert _error(_call(server, "session.control", session_id="gone", action="goal.pause"))["code"] == 4001
|
|
assert _error(_call(server, "session.control.read", session_id="gone"))["code"] == 4001
|
|
|
|
def test_dispatch_error_emits_no_update(self, server, session, monkeypatch):
|
|
sid, key, _ = session
|
|
_save_goal(key)
|
|
emitted = []
|
|
monkeypatch.setattr(server, "_emit", lambda *event: emitted.append(event))
|
|
monkeypatch.setitem(server._methods, "command.dispatch", lambda rid, params: server._err(rid, 4018, "dispatch failed"))
|
|
|
|
response = _call(server, "session.control", session_id=sid, action="goal.pause")
|
|
assert _error(response)["code"] == 4018
|
|
assert emitted == []
|
|
|
|
def test_adapter_error_becomes_4004_and_emits_no_update(self, server, session, monkeypatch):
|
|
from hermes_cli.goals import GoalManager
|
|
|
|
sid, key, _ = session
|
|
_save_goal(key)
|
|
emitted = []
|
|
monkeypatch.setattr(server, "_emit", lambda *event: emitted.append(event))
|
|
monkeypatch.setattr(GoalManager, "add_subgoal", lambda self, text: (_ for _ in ()).throw(RuntimeError("blocked")))
|
|
|
|
response = _call(server, "session.control", session_id=sid, action="subgoal.add", args={"text": "criterion"})
|
|
assert _error(response)["code"] == 4004
|
|
assert emitted == []
|
|
|
|
|
|
class TestUpdatePublication:
|
|
"""The Desktop card repaints from ``session.control.update``; every path that mutates control state must
|
|
publish it exactly once, and the read after a turn must reflect the post-turn hooks."""
|
|
|
|
def _capture(self, server, monkeypatch):
|
|
emitted = []
|
|
monkeypatch.setattr(server, "_emit", lambda event, event_sid, payload=None: emitted.append((event, event_sid, payload)))
|
|
return emitted
|
|
|
|
def test_control_action_publishes_exactly_one_update_matching_the_response(self, server, session, monkeypatch):
|
|
sid, key, _ = session
|
|
_save_goal(key)
|
|
emitted = self._capture(server, monkeypatch)
|
|
response = _call(server, "session.control", session_id=sid, action="goal.pause")
|
|
updates = [e for e in emitted if e[0] == "session.control.update"]
|
|
assert updates == [("session.control.update", sid, {"control": response["result"]["control"]})]
|
|
|
|
@pytest.mark.parametrize("name,arg", [("goal", "pause"), ("loop", "pause")])
|
|
def test_typed_slash_via_command_dispatch_publishes_the_new_state(self, server, session, monkeypatch, name, arg):
|
|
"""/goal and /loop never reach the slash worker (``_PENDING_INPUT_COMMANDS``) — the built-in path must
|
|
publish too, or the structured card stays stale until the next turn."""
|
|
sid, key, _ = session
|
|
if name == "goal":
|
|
_save_goal(key)
|
|
else:
|
|
from hermes_cli.loops import LoopManager
|
|
|
|
LoopManager(key).set("poll CI", interval_seconds=300)
|
|
emitted = self._capture(server, monkeypatch)
|
|
response = _call(server, "command.dispatch", session_id=sid, name=name, arg=arg)
|
|
assert "result" in response
|
|
updates = [e for e in emitted if e[0] == "session.control.update"]
|
|
assert len(updates) == 1 and updates[0][1] == sid
|
|
assert updates[0][2]["control"][name]["status"] == "paused"
|
|
|
|
def test_unknown_dispatch_publishes_nothing(self, server, session, monkeypatch):
|
|
sid, _, _ = session
|
|
emitted = self._capture(server, monkeypatch)
|
|
assert "error" in _call(server, "command.dispatch", session_id=sid, name="definitely-not-a-command", arg="")
|
|
assert [e for e in emitted if e[0] == "session.control.update"] == []
|
|
|
|
def test_real_turn_publishes_after_the_goal_judge_ran(self, server, session, monkeypatch, tmp_path):
|
|
"""``message.complete`` precedes the post-turn goal judge, so a client refresh keyed on that event reads
|
|
the pre-judge turn count. A real ``_run_prompt_submit`` turn (inline thread, stub agent) must publish the
|
|
snapshot AFTER the judge with the incremented count."""
|
|
sid, key, entry = session
|
|
_save_goal(key, turns_used=3)
|
|
emitted = self._capture(server, monkeypatch)
|
|
|
|
class _InlineThread:
|
|
def __init__(self, target=None, daemon=None, args=(), kwargs=None, name=None):
|
|
self._t, self._a, self._k = target, args, kwargs or {}
|
|
|
|
def start(self):
|
|
self._t(*self._a, **self._k)
|
|
|
|
def is_alive(self):
|
|
return False
|
|
|
|
def join(self, timeout=None):
|
|
return None
|
|
|
|
for name, value in {
|
|
"_wire_callbacks": lambda sid_: None, "_sync_agent_model_with_config": lambda sid_, s: None,
|
|
"_session_cwd": lambda s: str(tmp_path), "_register_session_cwd": lambda s: None,
|
|
"_tts_stream_begin": lambda: None, "_sync_session_key_after_compress": lambda *a, **k: None,
|
|
"_get_usage": lambda agent: {}, "_hermes_home": tmp_path,
|
|
}.items():
|
|
monkeypatch.setattr(server, name, value)
|
|
monkeypatch.setattr(server.threading, "Thread", _InlineThread)
|
|
# Deterministic judge: the model-graded verdict is not under test, the ordering is.
|
|
from hermes_cli.goals import GoalManager
|
|
|
|
def judge(self, raw, **kwargs):
|
|
from hermes_cli.goals import save_goal
|
|
|
|
self.state.turns_used += 1
|
|
save_goal(self.session_id, self.state)
|
|
return {"should_continue": False, "message": ""}
|
|
|
|
monkeypatch.setattr(GoalManager, "evaluate_after_turn", judge)
|
|
entry.update({"image_counter": 0, "slash_worker": None, "show_reasoning": False, "tool_progress_mode": "all",
|
|
"inflight_turn": None, "running": True,
|
|
"agent": SimpleNamespace(session_id=key, clear_interrupt=lambda: None,
|
|
run_conversation=lambda message, **kw: {"final_response": "done"})})
|
|
|
|
assert server._run_prompt_submit("rid", sid, entry, "work on it")
|
|
|
|
order = [e[0] for e in emitted]
|
|
assert "message.complete" in order and "session.control.update" in order
|
|
assert order.index("message.complete") < order.index("session.control.update")
|
|
assert emitted[order.index("session.control.update")][2]["control"]["goal"]["turns_used"] == 4
|