"""Tests for /goal quality gates (GoalGate, run_gate, GoalManager gate flow).""" import json import subprocess import sys from unittest.mock import patch import pytest from hermes_cli.goals import ( DEFAULT_GATE_MAX_RETRIES, DEFAULT_GATE_TIMEOUT_SECONDS, GoalGate, GoalManager, GoalState, run_gate, ) # ────────────────────────────────────────────────────────────────────── # GoalGate serialization # ────────────────────────────────────────────────────────────────────── def test_gate_roundtrip_through_goalstate_json(): state = GoalState(goal="ship it", status="active") state.gates.append(GoalGate(command="echo ok", timeout_seconds=42, max_retries=7)) raw = state.to_json() loaded = GoalState.from_json(raw) assert len(loaded.gates) == 1 g = loaded.gates[0] assert g.command == "echo ok" assert g.timeout_seconds == 42 assert g.max_retries == 7 assert g.attempts == 0 def test_gate_from_dict_defaults_and_garbage(): g = GoalGate.from_dict({"command": "true"}) assert g.timeout_seconds == DEFAULT_GATE_TIMEOUT_SECONDS assert g.max_retries == DEFAULT_GATE_MAX_RETRIES assert GoalGate.from_dict(None).command == "" assert GoalGate.from_dict("nonsense").command == "" def test_old_state_rows_without_gates_load_clean(): """Backwards compatibility: pre-gates state_meta rows load with no gates.""" old = {"goal": "legacy", "status": "active"} state = GoalState.from_json(json.dumps(old)) assert state.gates == [] # ────────────────────────────────────────────────────────────────────── # run_gate # ────────────────────────────────────────────────────────────────────── def test_run_gate_pass(): passed, code, out = run_gate(GoalGate(command="echo hello")) assert passed is True assert code == 0 assert "hello" in out @pytest.mark.platforms("linux") def test_run_gate_fail_captures_output(): # POSIX shell syntax (`>&2`, `;`, `exit`) — cmd.exe (shell=True on Windows) # doesn't parse it, so the gate "passes" instead of failing. passed, code, out = run_gate(GoalGate(command="echo broken >&2; exit 3")) assert passed is False assert code == 3 assert "broken" in out def test_run_gate_timeout(): passed, code, out = run_gate(GoalGate(command="sleep 5", timeout_seconds=1)) assert passed is False assert code == -1 assert "timed out" in out def test_run_gate_keeps_diagnostics_when_a_byte_will_not_decode(tmp_path): """A gate's output tail must survive bytes the decoder rejects. A gate runs whatever the operator configured, so its output is arbitrary bytes — a test runner's checkmarks or CJK on a non-UTF-8 Windows console, or stray binary. Decoding strictly means one bad byte kills subprocess's reader thread, stdout comes back None, and the tail lands empty: the agent is told the gate failed with nothing to act on, so it burns every retry and the goal auto-pauses. """ script = tmp_path / "gate.py" script.write_text( "import os, sys\n" "os.write(1, b'FAILED: 3 tests broken \\x90\\x8d rerun me\\n')\n" "sys.exit(1)\n", encoding="utf-8", ) passed, code, out = run_gate( GoalGate(command=f'"{sys.executable}" "{script}"'), ) assert passed is False assert code == 1 assert "FAILED: 3 tests broken" in out, ( f"gate diagnostics were lost to a decode failure (tail={out!r})" ) # ────────────────────────────────────────────────────────────────────── # GoalManager gate management # ────────────────────────────────────────────────────────────────────── def _mgr_with_goal(session_id="gate-test-sid"): mgr = GoalManager(session_id=session_id) mgr.set("test goal") return mgr def test_add_remove_clear_gates(): mgr = _mgr_with_goal("gate-mgmt-sid") mgr.add_gate("echo one") mgr.add_gate("echo two") assert len(mgr.state.gates) == 2 assert "echo one" in mgr.render_gates() removed = mgr.remove_gate(1) assert removed == "echo one" assert len(mgr.state.gates) == 1 assert mgr.clear_gates() == 1 assert mgr.state.gates == [] def test_add_gate_requires_active_goal(): mgr = GoalManager(session_id="gate-nogoal-sid") with pytest.raises(RuntimeError): mgr.add_gate("echo nope") def test_gates_persist_and_reload(): mgr = _mgr_with_goal("gate-persist-sid") mgr.add_gate("echo persisted") reloaded = GoalManager(session_id="gate-persist-sid") assert len(reloaded.state.gates) == 1 assert reloaded.state.gates[0].command == "echo persisted" # ────────────────────────────────────────────────────────────────────── # evaluate_after_turn integration # ────────────────────────────────────────────────────────────────────── def test_failing_gate_short_circuits_judge(): mgr = _mgr_with_goal("gate-fail-sid") mgr.add_gate("exit 5") with patch("hermes_cli.goals.judge_goal") as mock_judge: decision = mgr.evaluate_after_turn("I think it's done!") mock_judge.assert_not_called() assert decision["verdict"] == "gate_failed" assert decision["should_continue"] is True assert "exit 5" in decision["continuation_prompt"] assert "quality gate" in decision["continuation_prompt"].lower() def test_passing_gates_fall_through_to_judge(): mgr = _mgr_with_goal("gate-pass-sid") mgr.add_gate("true") with patch( "hermes_cli.goals.judge_goal", return_value=("done", "all good", False, None, False), ) as mock_judge: decision = mgr.evaluate_after_turn("finished") mock_judge.assert_called_once() assert decision["verdict"] == "done" # Passing run resets attempt bookkeeping. assert mgr.state.gates[0].attempts == 0 assert mgr.state.gates[0].last_exit_code == 0 def test_gate_retry_exhaustion_pauses_goal(): mgr = _mgr_with_goal("gate-exhaust-sid") mgr.add_gate("exit 1") mgr.state.gates[0].max_retries = 2 with patch("hermes_cli.goals.judge_goal") as mock_judge: d1 = mgr.evaluate_after_turn("attempt one") d2 = mgr.evaluate_after_turn("attempt two") d3 = mgr.evaluate_after_turn("attempt three") mock_judge.assert_not_called() assert d1["should_continue"] is True assert d2["should_continue"] is True assert d3["status"] == "paused" assert d3["should_continue"] is False assert mgr.state.status == "paused" assert "gate" in (mgr.state.paused_reason or "") def test_failed_gate_reruns_when_untracked_file_content_changes(tmp_path, monkeypatch): """#110649: `git status --porcelain` reports the same `?? untracked/` for `before` and `after`, so a status-based cache replayed the stale failure; the gate must execute again.""" for argv in (["init", "-q"], ["config", "user.email", "t@example.com"], ["config", "user.name", "T"], ["commit", "-q", "--allow-empty", "-m", "baseline"]): subprocess.run(["git", *argv], cwd=tmp_path, check=True, capture_output=True) result = tmp_path / "untracked" / "result.txt" result.parent.mkdir() result.write_text("before", encoding="utf-8") monkeypatch.chdir(tmp_path) mgr = _mgr_with_goal("gate-content-sid") mgr.add_gate(f"grep -q after {result}") with patch("hermes_cli.goals.judge_goal", return_value=("done", "ok", False, None, False)) as judge: d1 = mgr.evaluate_after_turn("turn 1") result.write_text("after", encoding="utf-8") d2 = mgr.evaluate_after_turn("turn 2") assert d1["verdict"] == "gate_failed" assert d2["verdict"] == "done" judge.assert_called_once() assert mgr.state.gates[0].attempts == 0 def test_gate_continuation_respects_turn_budget(): mgr = GoalManager(session_id="gate-budget-sid", default_max_turns=1) mgr.set("budget goal") mgr.add_gate("exit 1") with patch("hermes_cli.goals.judge_goal"): decision = mgr.evaluate_after_turn("only turn") assert decision["status"] == "paused" assert decision["should_continue"] is False assert "turns used" in decision["message"] def test_no_gates_behaves_exactly_as_before(): mgr = _mgr_with_goal("gate-none-sid") with patch( "hermes_cli.goals.judge_goal", return_value=("continue", "keep going", False, None, False), ) as mock_judge: decision = mgr.evaluate_after_turn("wip") mock_judge.assert_called_once() assert decision["verdict"] == "continue" assert decision["should_continue"] is True # ────────────────────────────────────────────────────────────────────── # Gate working directory (#125369) # ────────────────────────────────────────────────────────────────────── @pytest.fixture def backend_and_session(tmp_path, monkeypatch): """A backend started in a project whose check passes, serving a session whose check fails.""" from agent.runtime_cwd import reset_session_cwd, set_session_cwd backend, session = tmp_path / "backend", tmp_path / "session" for folder, code in ((backend, 0), (session, 1)): folder.mkdir() (folder / "check.sh").write_text(f"pwd\nexit {code}\n", encoding="utf-8") monkeypatch.chdir(backend) monkeypatch.delenv("TERMINAL_CWD", raising=False) def bind(cwd): token = set_session_cwd(str(cwd)) return lambda: reset_session_cwd(token) return backend, session, bind def _evaluate_with_done_judge(mgr): with patch("hermes_cli.goals.judge_goal", return_value=("done", "all good", False, None, False)) as judge: return mgr.evaluate_after_turn("ready"), judge def test_gate_runs_in_the_session_workspace_not_the_backend_directory(backend_and_session): backend, session, bind = backend_and_session mgr = _mgr_with_goal("gate-cwd-sid") mgr.add_gate("sh check.sh") unbind = bind(session) try: decision, judge = _evaluate_with_done_judge(mgr) finally: unbind() judge.assert_not_called() assert decision["verdict"] == "gate_failed" assert mgr.state.gates[0].last_exit_code == 1 assert str(session.resolve()) in mgr.state.gates[0].last_output_tail def test_missing_session_workspace_pauses_instead_of_running_elsewhere(backend_and_session, tmp_path): # A deleted, remote or container workspace: the backend's passing check must not stand in for it, # and no retry can fix it, so the goal pauses on the first check with the reason and no attempt charged. backend, _session, bind = backend_and_session missing = tmp_path / "gone" mgr = _mgr_with_goal("gate-missing-cwd-sid") mgr.add_gate("sh check.sh") unbind = bind(missing) try: with patch("hermes_cli.goals.run_gate") as run: decision, judge = _evaluate_with_done_judge(mgr) finally: unbind() run.assert_not_called() judge.assert_not_called() assert decision["status"] == "paused" and decision["should_continue"] is False assert str(missing) in decision["message"] and "still failing" not in decision["message"] assert str(missing) in (mgr.state.paused_reason or "") gate = mgr.state.gates[0] assert (gate.attempts, gate.last_exit_code) == (0, None) def test_gate_without_a_session_workspace_keeps_the_launch_directory(backend_and_session): from agent.runtime_cwd import clear_session_cwd backend, _session, _bind = backend_and_session clear_session_cwd() mgr = _mgr_with_goal("gate-launch-cwd-sid") mgr.add_gate("sh check.sh") decision, judge = _evaluate_with_done_judge(mgr) judge.assert_called_once() assert decision["verdict"] == "done" assert str(backend.resolve()) in mgr.state.gates[0].last_output_tail