| import subprocess | |
| from clfgui.kernel import FakeAgentBackend, HermesAgentBackend, Job, Phase | |
| def make_job() -> Job: | |
| return Job( | |
| id=7, | |
| intent="Build a small owner cockpit", | |
| status="running", | |
| created_at="2026-05-05T00:00:00Z", | |
| updated_at="2026-05-05T00:00:00Z", | |
| ) | |
| def make_phase(name: str = "research") -> Phase: | |
| return Phase( | |
| id=11, | |
| job_id=7, | |
| name=name, | |
| position=2, | |
| status="running", | |
| summary="Phase is running.", | |
| started_at=None, | |
| completed_at=None, | |
| ) | |
| def test_fake_agent_backend_returns_deterministic_phase_outcome(): | |
| outcome = FakeAgentBackend().run_phase(make_job(), make_phase("research"), "context") | |
| assert outcome.status == "completed" | |
| assert outcome.summary == "Relevante Annahmen und offene Risiken wurden für V1 verdichtet." | |
| assert outcome.decision is None | |
| assert outcome.result is None | |
| def test_hermes_backend_builds_command_and_maps_success(monkeypatch, tmp_path): | |
| captured = {} | |
| def fake_run(command, **kwargs): | |
| captured["command"] = command | |
| captured["kwargs"] = kwargs | |
| return subprocess.CompletedProcess(command, 0, stdout=" concise Hermes summary \n", stderr="") | |
| monkeypatch.setattr("clfgui.kernel.backends.subprocess.run", fake_run) | |
| backend = HermesAgentBackend(executable="/usr/bin/hermes", timeout_s=9, cwd=tmp_path) | |
| outcome = backend.run_phase(make_job(), make_phase("research"), "phase context") | |
| assert outcome.status == "completed" | |
| assert outcome.summary == "concise Hermes summary" | |
| assert captured["command"][0:3] == ["/usr/bin/hermes", "chat", "-q"] | |
| assert "-Q" in captured["command"] | |
| assert captured["command"][captured["command"].index("--source") + 1] == "clfgui" | |
| assert captured["command"][captured["command"].index("--max-turns") + 1] == "1" | |
| prompt = captured["command"][captured["command"].index("-q") + 1] | |
| assert "Do not edit files." in prompt | |
| assert "Do not run tools or commands that mutate files." in prompt | |
| assert "phase context" in prompt | |
| assert captured["kwargs"]["cwd"] == str(tmp_path) | |
| assert captured["kwargs"]["capture_output"] is True | |
| assert captured["kwargs"]["text"] is True | |
| assert captured["kwargs"]["timeout"] == 9 | |
| assert captured["kwargs"]["check"] is False | |
| def test_hermes_backend_maps_nonzero_exit_to_failed(monkeypatch): | |
| def fake_run(command, **kwargs): | |
| return subprocess.CompletedProcess(command, 2, stdout="stdout details", stderr="stderr details") | |
| monkeypatch.setattr("clfgui.kernel.backends.subprocess.run", fake_run) | |
| outcome = HermesAgentBackend().run_phase(make_job(), make_phase("verify"), "context") | |
| assert outcome.status == "failed" | |
| assert "exit 2" in outcome.summary | |
| assert "stderr details" in outcome.summary | |
| def test_hermes_backend_maps_timeout_to_failed(monkeypatch): | |
| def fake_run(command, **kwargs): | |
| raise subprocess.TimeoutExpired(cmd=command, timeout=3) | |
| monkeypatch.setattr("clfgui.kernel.backends.subprocess.run", fake_run) | |
| outcome = HermesAgentBackend(timeout_s=3).run_phase(make_job(), make_phase("test"), "context") | |
| assert outcome.status == "failed" | |
| assert outcome.summary == "Hermes phase timed out after 3 seconds." | |