"""Tests for hermes_cli/goals.py — persistent cross-turn goals.""" from __future__ import annotations import json import time from unittest.mock import patch, MagicMock import pytest # ────────────────────────────────────────────────────────────────────── # Fixtures # ────────────────────────────────────────────────────────────────────── @pytest.fixture def hermes_home(tmp_path, monkeypatch): """Isolated HERMES_HOME so SessionDB.state_meta writes don't clobber the real one.""" from pathlib import Path home = tmp_path / ".hermes" home.mkdir() monkeypatch.setattr(Path, "home", lambda: tmp_path) monkeypatch.setenv("HERMES_HOME", str(home)) # Bust the goal-module's DB cache for each test so it re-resolves HERMES_HOME. from hermes_cli import goals goals._DB_CACHE.clear() yield home goals._DB_CACHE.clear() # ────────────────────────────────────────────────────────────────────── # _parse_judge_response # ────────────────────────────────────────────────────────────────────── class TestParseJudgeResponse: def test_clean_json_done(self): from hermes_cli.goals import _parse_judge_response verdict, reason, _pf, wait = _parse_judge_response('{"done": true, "reason": "all good"}') assert verdict == "done" assert reason == "all good" assert wait is None def test_wait_verdict_with_pid(self): from hermes_cli.goals import _parse_judge_response v, reason, pf, wait = _parse_judge_response( '{"verdict": "wait", "wait_on_pid": 4242, "reason": "CI running"}' ) assert v == "wait" assert pf is False assert wait == {"pid": 4242} assert reason == "CI running" # ────────────────────────────────────────────────────────────────────── # judge_goal — fail-open semantics # ────────────────────────────────────────────────────────────────────── class TestJudgeGoal: def test_api_error_continues(self): """Judge exception → fail-open continue (don't wedge progress on judge bugs).""" from hermes_cli import goals with patch( "agent.auxiliary_client.call_llm", side_effect=RuntimeError("boom"), ): verdict, reason, _, _wd, _tf = goals.judge_goal("goal", "response") assert verdict == "continue" assert "judge error" in reason.lower() def test_judge_says_done(self): from hermes_cli import goals with patch( "agent.auxiliary_client.call_llm", return_value=MagicMock( choices=[MagicMock(message=MagicMock(content='{"done": true, "reason": "achieved"}'))] ), ): verdict, reason, _, _wd, _tf = goals.judge_goal("goal", "agent response") assert verdict == "done" assert reason == "achieved" # ────────────────────────────────────────────────────────────────────── # GoalManager lifecycle + persistence # ────────────────────────────────────────────────────────────────────── class TestGoalManager: def test_set_then_status(self, hermes_home): from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="test-sid-2", default_max_turns=5) state = mgr.set("port the thing") assert state.goal == "port the thing" assert state.status == "active" assert state.max_turns == 5 assert state.turns_used == 0 assert mgr.is_active() assert "active" in mgr.status_line().lower() assert "port the thing" in mgr.status_line() def test_continuation_prompt_shape(self, hermes_home): """The continuation prompt must include the goal text verbatim — and must be safe to inject as a user-role message (prompt-cache invariants: no system-prompt mutation).""" from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="cont-sid") mgr.set("port goal command to hermes") prompt = mgr.next_continuation_prompt() assert prompt is not None assert "port goal command to hermes" in prompt assert prompt.strip() # non-empty # ────────────────────────────────────────────────────────────────────── # Smoke: CommandDef is wired # ────────────────────────────────────────────────────────────────────── def test_goal_command_in_registry(): from hermes_cli.commands import resolve_command cmd = resolve_command("goal") assert cmd is not None assert cmd.name == "goal" def test_goal_command_dispatches_in_cli_registry_helpers(): """goal shows up in autocomplete / help categories alongside other Session cmds.""" from hermes_cli.commands import COMMANDS, COMMANDS_BY_CATEGORY assert "/goal" in COMMANDS session_cmds = COMMANDS_BY_CATEGORY.get("Session", {}) assert "/goal" in session_cmds # ────────────────────────────────────────────────────────────────────── # Auto-pause on consecutive judge parse failures # ────────────────────────────────────────────────────────────────────── class TestJudgeParseFailureAutoPause: """Regression: weak judge models (e.g. deepseek-v4-flash) that return empty strings or non-JSON prose must auto-pause the loop after N turns instead of burning the whole turn budget.""" def test_api_error_does_not_count_as_parse_failure(self): """Transient network/API errors must not trip the auto-pause guard.""" from hermes_cli import goals with patch( "agent.auxiliary_client.call_llm", side_effect=RuntimeError("connection reset"), ): verdict, _, parse_failed, _wd, transport_failed = goals.judge_goal( "goal", "response" ) assert verdict == "continue" assert parse_failed is False assert transport_failed is True def test_auto_pause_after_three_consecutive_parse_failures(self, hermes_home): """N=3 consecutive parse failures → auto-pause with config pointer.""" from hermes_cli import goals from hermes_cli.goals import GoalManager, DEFAULT_MAX_CONSECUTIVE_PARSE_FAILURES assert DEFAULT_MAX_CONSECUTIVE_PARSE_FAILURES == 3 mgr = GoalManager(session_id="parse-fail-sid-1", default_max_turns=20) mgr.set("do a thing") with patch.object( goals, "judge_goal", return_value=("continue", "judge returned empty response", True, None, False) ): d1 = mgr.evaluate_after_turn("step 1") assert d1["should_continue"] is True assert mgr.state.consecutive_parse_failures == 1 d2 = mgr.evaluate_after_turn("step 2") assert d2["should_continue"] is True assert mgr.state.consecutive_parse_failures == 2 d3 = mgr.evaluate_after_turn("step 3") assert d3["should_continue"] is False assert d3["status"] == "paused" assert mgr.state.consecutive_parse_failures == 3 # Message points at the config surface so the user can fix it. assert "auxiliary" in d3["message"] assert "goal_judge" in d3["message"] assert "config.yaml" in d3["message"] # ────────────────────────────────────────────────────────────────────── # /subgoal — user-added criteria # ────────────────────────────────────────────────────────────────────── class TestGoalStateSubgoalsBackcompat: def test_old_state_meta_row_loads_without_subgoals(self): """A goal serialized BEFORE the subgoals field existed must round-trip with an empty list, not crash.""" from hermes_cli.goals import GoalState legacy = json.dumps({ "goal": "do a thing", "status": "active", "turns_used": 2, "max_turns": 20, "created_at": 1.0, "last_turn_at": 2.0, "consecutive_parse_failures": 0, }) state = GoalState.from_json(legacy) assert state.goal == "do a thing" assert state.subgoals == [] class TestMigrateGoalToSession: """migrate_goal_to_session carries a /goal from a parent session to its compression continuation child (#33618). load_goal does a flat per-session lookup with no lineage walk, so without migration an active goal silently dies when compression rotates session_id.""" def test_migrates_active_goal_to_child(self, hermes_home): from hermes_cli.goals import save_goal, load_goal, migrate_goal_to_session, GoalState save_goal("parent-sid", GoalState(goal="ship the feature")) assert migrate_goal_to_session("parent-sid", "child-sid", reason="compression") is True child = load_goal("child-sid") assert child is not None and child.goal == "ship the feature" # Parent row archived (cleared) so only the child is active. parent = load_goal("parent-sid") assert parent is not None and parent.status == "cleared" def test_does_not_clobber_existing_child_goal(self, hermes_home): from hermes_cli.goals import save_goal, load_goal, migrate_goal_to_session, GoalState save_goal("p3", GoalState(goal="parent goal")) save_goal("c3", GoalState(goal="child already has one")) assert migrate_goal_to_session("p3", "c3") is False assert load_goal("c3").goal == "child already has one" class TestGoalManagerSubgoals: def test_add_subgoal(self, hermes_home): from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="sub-add") mgr.set("main goal") text = mgr.add_subgoal(" use bullet points ") assert text == "use bullet points" assert mgr.state.subgoals == ["use bullet points"] def test_remove_subgoal_out_of_range(self, hermes_home): import pytest from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="sub-oob") mgr.set("g") mgr.add_subgoal("only") with pytest.raises(IndexError): mgr.remove_subgoal(5) with pytest.raises(IndexError): mgr.remove_subgoal(0) class TestJudgeGoalWithSubgoals: def test_judge_uses_subgoals_template_when_provided(self, hermes_home): """judge_goal switches templates when subgoals is non-empty. We don't actually call the model — we patch the aux client to capture the prompt that would be sent. """ from unittest.mock import patch from hermes_cli import goals captured = {} class _FakeMsg: content = '{"done": true, "reason": "all done"}' class _FakeChoice: message = _FakeMsg() class _FakeResp: choices = [_FakeChoice()] def _fake_call_llm(**kwargs): captured.update(kwargs) return _FakeResp() with patch("agent.auxiliary_client.call_llm", side_effect=_fake_call_llm): verdict, reason, parse_failed, _wd, _tf = goals.judge_goal( "ship the feature", "ok shipped", subgoals=["write tests", "update docs"], ) # The aux client was called with a prompt that includes the subgoals. sent_messages = captured.get("messages") or [] user_msg = next((m["content"] for m in sent_messages if m["role"] == "user"), "") assert "Additional criteria" in user_msg assert "1. write tests" in user_msg assert "2. update docs" in user_msg assert "every additional criterion" in user_msg assert verdict == "done" def test_judge_uses_original_template_when_no_subgoals(self, hermes_home): from unittest.mock import patch from hermes_cli import goals captured = {} class _FakeMsg: content = '{"done": true, "reason": "ok"}' class _FakeChoice: message = _FakeMsg() class _FakeResp: choices = [_FakeChoice()] def _fake_call_llm(**kwargs): captured.update(kwargs) return _FakeResp() with patch("agent.auxiliary_client.call_llm", side_effect=_fake_call_llm): goals.judge_goal("ship it", "done", subgoals=None) sent_messages = captured.get("messages") or [] user_msg = next((m["content"] for m in sent_messages if m["role"] == "user"), "") assert "Additional criteria" not in user_msg assert "ship it" in user_msg class TestStatusLineSubgoalCount: def test_status_line_with_subgoals(self, hermes_home): from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="sl-with") mgr.set("ship it") mgr.add_subgoal("a") mgr.add_subgoal("b") line = mgr.status_line() assert "2 subgoals" in line # ────────────────────────────────────────────────────────────────────── # Wait barrier — parking the goal loop on a background process # ────────────────────────────────────────────────────────────────────── class TestWaitBarrier: """The /goal wait barrier parks the loop on a live PID and resumes when the process exits, without burning turns or calling the judge.""" @staticmethod def _spawn_sleeper(): """Start a short-lived child process; return its Popen handle.""" import subprocess import sys return subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) @staticmethod def _dead_pid(): """A PID that is essentially guaranteed not to be running.""" return 2_000_000_000 def test_parked_on_live_pid_does_not_continue_or_judge(self, hermes_home): from hermes_cli import goals from hermes_cli.goals import GoalManager proc = self._spawn_sleeper() try: mgr = GoalManager(session_id="wb-live") mgr.set("ship it", max_turns=5) mgr.wait_on(proc.pid, reason="CI green") assert mgr.is_waiting() is True # The judge must NOT be called while parked, and no turn is burned. judge = MagicMock(return_value=("continue", "x", False, None, False)) with patch.object(goals, "judge_goal", judge): decision = mgr.evaluate_after_turn("still waiting on CI") judge.assert_not_called() assert decision["verdict"] == "waiting" assert decision["should_continue"] is False assert decision["continuation_prompt"] is None assert mgr.state.turns_used == 0 # no turn consumed while parked assert "CI green" in decision["message"] assert mgr.state.status == "active" # still active, just parked finally: proc.terminate() proc.wait(timeout=10) def test_stop_waiting_clears_barrier(self, hermes_home): from hermes_cli.goals import GoalManager proc = self._spawn_sleeper() try: mgr = GoalManager(session_id="wb-stop") mgr.set("g") mgr.wait_on(proc.pid) assert mgr.is_waiting() is True assert mgr.stop_waiting() is True assert mgr.state.waiting_on_pid is None assert mgr.is_waiting() is False assert mgr.stop_waiting() is False # idempotent finally: proc.terminate() proc.wait(timeout=10) # ────────────────────────────────────────────────────────────────────── # Judge-driven auto-wait — the judge parks the loop on its own # ────────────────────────────────────────────────────────────────────── class TestJudgeDrivenWait: """The judge returns a `wait` verdict (given live background-process context) and the loop parks automatically — no manual /goal wait.""" @staticmethod def _spawn_sleeper(): import subprocess, sys return subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) def test_judge_wait_pid_parks_loop(self, hermes_home): from hermes_cli import goals from hermes_cli.goals import GoalManager proc = self._spawn_sleeper() try: mgr = GoalManager(session_id="jw-pid", default_max_turns=10) mgr.set("ship the PR") # Judge sees the running process and says wait-on-pid. with patch.object( goals, "judge_goal", return_value=("wait", "CI watcher still running", False, {"pid": proc.pid}, False), ): decision = mgr.evaluate_after_turn( "Pushed the PR, watching CI.", background_processes=[{ "pid": proc.pid, "command": "wait_for_pr_green.sh", "status": "running", "uptime_seconds": 12, }], ) assert decision["verdict"] == "wait" assert decision["should_continue"] is False assert decision["continuation_prompt"] is None assert mgr.state.waiting_on_pid == proc.pid assert mgr.is_waiting() is True # Next turn while still parked: judge must NOT be called again. judge = MagicMock() with patch.object(goals, "judge_goal", judge): d2 = mgr.evaluate_after_turn("still going") judge.assert_not_called() assert d2["verdict"] == "waiting" assert d2["should_continue"] is False finally: proc.terminate() proc.wait(timeout=10) def test_time_barrier_clears_after_deadline(self, hermes_home): from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="jw-deadline") mgr.set("g") mgr.wait_for_seconds(120, reason="backoff") assert mgr.is_waiting() is True # Force the deadline into the past → barrier auto-clears. mgr.state.waiting_until = time.time() - 1 assert mgr.is_waiting() is False assert mgr.state.waiting_until == 0.0 def test_continue_verdict_still_continues_with_background(self, hermes_home): """A running process present but judge says continue → normal loop.""" from hermes_cli import goals from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="jw-cont", default_max_turns=10) mgr.set("do work") with patch.object( goals, "judge_goal", return_value=("continue", "more to do", False, None, False), ): decision = mgr.evaluate_after_turn( "made progress", background_processes=[{"pid": 999999, "command": "x", "status": "running"}], ) assert decision["verdict"] == "continue" assert decision["should_continue"] is True assert mgr.state.waiting_on_pid is None # ────────────────────────────────────────────────────────────────────── # Session/trigger barrier — wait on a process's OWN trigger, not just exit # ────────────────────────────────────────────────────────────────────── class TestSessionTriggerBarrier: """The session barrier (wait_on_session) releases when a process's own trigger fires — a watch_patterns match mid-run (process may never exit) OR exit — not only on PID exit. CI-safe: uses synthetic registry session objects, no real child processes.""" @staticmethod def _inject(sid, *, watch_patterns=None, exited=False): import time as _t from tools.process_registry import process_registry, ProcessSession s = ProcessSession(id=sid, command="watcher.sh", task_id="t", session_key="", cwd="/tmp", started_at=_t.time()) if watch_patterns: s.watch_patterns = list(watch_patterns) s.exited = exited if exited: process_registry._finished[sid] = s else: process_registry._running[sid] = s return s, process_registry def test_registry_releases_on_watch_match_while_alive(self, hermes_home): s, reg = self._inject("proc_t2", watch_patterns=["READY"]) assert reg.is_session_waiting("proc_t2") is True s._watch_hits = 1 # what _check_watch_patterns sets on a match # Released even though the process is STILL running (never exited). assert s.exited is False assert reg.is_session_waiting("proc_t2") is False def test_wait_on_session_validation(self, hermes_home): from hermes_cli.goals import GoalManager mgr = GoalManager(session_id="st-val") # No active goal → RuntimeError try: mgr.wait_on_session("proc_x") assert False, "expected RuntimeError" except RuntimeError: pass mgr.set("g") try: mgr.wait_on_session("") assert False, "expected ValueError" except ValueError: pass # ────────────────────────────────────────────────────────────────────── # Completion contract (Codex-inspired structured goals) # ────────────────────────────────────────────────────────────────────── class TestParseContract: def test_inline_fields_parsed(self): from hermes_cli.goals import parse_contract text = ( "Migrate auth to JWT\n" "verify: the auth test suite passes\n" "constraints: keep the /login response shape unchanged\n" "boundaries: only touch services/auth and its tests\n" "stop when: a schema change needs product sign-off" ) headline, contract = parse_contract(text) assert headline == "Migrate auth to JWT" assert contract.verification == "the auth test suite passes" assert contract.constraints == "keep the /login response shape unchanged" assert contract.boundaries == "only touch services/auth and its tests" assert contract.stop_when == "a schema change needs product sign-off" assert not contract.is_empty() def test_alias_variants(self): from hermes_cli.goals import parse_contract _, c = parse_contract("Goal\nverified by: tests green\npreserve: public API") assert c.verification == "tests green" assert c.constraints == "public API" class TestGoalContractSerialization: def test_roundtrip_with_contract(self): from hermes_cli.goals import GoalState, GoalContract state = GoalState( goal="ship it", contract=GoalContract( verification="pytest passes", constraints="don't break the API", ), ) restored = GoalState.from_json(state.to_json()) assert restored.goal == "ship it" assert restored.contract.verification == "pytest passes" assert restored.contract.constraints == "don't break the API" assert restored.has_contract() def test_old_row_without_contract_loads_clean(self): # A state_meta row written before this feature has no "contract" key. from hermes_cli.goals import GoalState legacy = '{"goal": "old goal", "status": "active", "turns_used": 2}' state = GoalState.from_json(legacy) assert state.goal == "old goal" assert state.turns_used == 2 assert state.contract.is_empty() assert not state.has_contract() def test_render_block_omits_empty_fields(self): from hermes_cli.goals import GoalContract block = GoalContract(outcome="X", verification="Y").render_block() assert "Outcome: X" in block assert "Verification: Y" in block assert "Constraints" not in block class TestGoalManagerContract: def test_set_contract_after_the_fact(self, hermes_home): from hermes_cli.goals import GoalManager, GoalContract mgr = GoalManager(session_id="c-after") mgr.set("ship it") assert not mgr.has_contract() mgr.set_contract(GoalContract(verification="x")) assert mgr.has_contract() # Survives reload. from hermes_cli.goals import GoalManager as GM2 assert GM2(session_id="c-after").has_contract() def test_persistence_roundtrip(self, hermes_home): from hermes_cli.goals import GoalManager, GoalContract GoalManager(session_id="c-persist").set( "ship it", contract=GoalContract(outcome="O", verification="V") ) reloaded = GoalManager(session_id="c-persist") assert reloaded.state.contract.outcome == "O" assert reloaded.state.contract.verification == "V" class TestJudgeWithContract: def _fake_call_llm(self, captured, content='{"done": false, "reason": "more"}'): """judge_goal routes through call_llm (#35566) — capture its kwargs.""" class _FakeMsg: pass _FakeMsg.content = content class _FakeChoice: message = _FakeMsg() class _FakeResp: choices = [_FakeChoice()] def _fake(**kwargs): captured.update(kwargs) return _FakeResp() return _fake def test_judge_uses_contract_template(self, hermes_home): from unittest.mock import patch from hermes_cli import goals from hermes_cli.goals import GoalContract captured = {} with patch("agent.auxiliary_client.call_llm", side_effect=self._fake_call_llm(captured)): goals.judge_goal( "ship it", "I think it's done", contract=GoalContract(verification="pytest -q passes"), ) user_msg = next( (m["content"] for m in (captured.get("messages") or []) if m["role"] == "user"), "" ) assert "completion contract" in user_msg.lower() assert "pytest -q passes" in user_msg assert "concrete evidence" in user_msg class TestDraftContract: def test_draft_parses_json(self, hermes_home): from unittest.mock import patch from hermes_cli import goals class _FakeMsg: content = ( '{"outcome": "auth on JWT", "verification": "auth suite green", ' '"constraints": "no API change", "boundaries": "services/auth", ' '"stop_when": "schema change needed"}' ) class _FakeChoice: message = _FakeMsg() class _FakeResp: choices = [_FakeChoice()] with patch("agent.auxiliary_client.call_llm", return_value=_FakeResp()): contract = goals.draft_contract("Migrate auth to JWT") assert contract is not None assert contract.outcome == "auth on JWT" assert contract.verification == "auth suite green" assert not contract.is_empty() def test_draft_returns_none_when_no_client(self, hermes_home): from unittest.mock import patch from hermes_cli import goals with patch("agent.auxiliary_client.call_llm", side_effect=RuntimeError("No LLM provider configured")): assert goals.draft_contract("anything") is None # ────────────────────────────────────────────────────────────────────── # Compose: completion contract + wait barrier in one judge call # ────────────────────────────────────────────────────────────────────── class TestContractAndBackgroundCompose: """A contract goal blocked on a background process must surface BOTH the contract block and the background-process list to the judge, so it can return either done (evidence met) or wait (parked on the poller).""" def _capture_call_llm(self, captured, content='{"verdict": "wait", "wait_on_pid": 4242, "reason": "CI still running"}'): """judge_goal routes through call_llm (#35566) — capture its kwargs.""" class _FakeMsg: pass _FakeMsg.content = content class _FakeChoice: message = _FakeMsg() class _FakeResp: choices = [_FakeChoice()] def _fake(**kwargs): captured.update(kwargs) return _FakeResp() return _fake def test_judge_prompt_carries_contract_and_background(self, hermes_home): from unittest.mock import patch from hermes_cli import goals from hermes_cli.goals import GoalContract captured = {} bg = [{ "session_id": "ci-watch", "pid": 4242, "status": "running", "command": "wait_for_pr_green.sh 50501", "trigger": "exit", }] with patch("agent.auxiliary_client.call_llm", side_effect=self._capture_call_llm(captured)): verdict, reason, parse_failed, wait_directive, _tf = goals.judge_goal( "ship the PR", "I pushed and started the CI watcher; waiting on it now.", contract=GoalContract(verification="PR CI goes green"), background_processes=bg, ) user_msg = next( (m["content"] for m in (captured.get("messages") or []) if m["role"] == "user"), "" ) # Both surfaces present in one prompt. assert "completion contract" in user_msg.lower() assert "PR CI goes green" in user_msg assert "Background processes" in user_msg assert "4242" in user_msg # The judge can return a wait verdict on a contract goal. assert verdict == "wait" assert wait_directive and wait_directive.get("pid") == 4242