import fcntl import importlib.util import io import json import os import subprocess import sys import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) def load_module(name, filename): spec = importlib.util.spec_from_file_location(name, HERE / filename) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module listener = load_module("agent_codex_listener", "listener.py") worker = load_module("agent_codex_worker", "worker.py") builder = load_module("agent_codex_builder", "builder.py") deploy_runner = load_module("agent_codex_deploy_runner", "deploy_runner.py") common = sys.modules["common"] class _RecordingStdin: """Records what was written instead of actually closing/discarding it -- unlike io.StringIO, whose buffer becomes unreadable after close(), which real code (run_streaming()) always calls.""" def __init__(self): self.value = "" def write(self, text): self.value += text def close(self): pass class FakeCodexProcess: """Stand-in for subprocess.Popen in common.run_streaming() tests -- yields `lines` (already-newline-terminated JSONL) one at a time from a fake stdout, then EOF, mirroring how a real Codex process's stdout behaves once terminated/exited.""" def __init__(self, lines, returncode=0): self._lines = list(lines) self.stdin = _RecordingStdin() self.stderr = io.StringIO("") self.returncode = returncode self.terminated = False self.killed = False def readline(self): if self._lines: return self._lines.pop(0) return "" @property def stdout(self): return self def close(self): pass def terminate(self): self.terminated = True def kill(self): self.killed = True def wait(self, timeout=None): return self.returncode class HangingCodexProcess(FakeCodexProcess): """readline() blocks past any reasonable test timeout instead of hitting EOF -- exercises run_streaming()'s own wall-clock deadline, the "zusätzliche Notbremse" independent of the iteration limit.""" def readline(self): import time as _time _time.sleep(0.2) return "" class CommonRunStreamingTests(unittest.TestCase): def test_runtime_validation_is_exact(self): self.assertEqual(common.validate_runtime("codex"), "codex") self.assertEqual(common.validate_runtime("claude"), "claude") self.assertEqual(common.validate_runtime("openrouter"), "openrouter") with self.assertRaises(ValueError): common.validate_runtime("codex --model gpt") def test_openrouter_model_falls_back_and_accepts_provider_slash(self): command = [] with mock.patch.dict( os.environ, {"TEST_OPENROUTER_MODEL": "deepseek/deepseek-v3.2"}, clear=False, ): common.append_openrouter_options(command, "TEST_OPENROUTER_MODEL", model="") self.assertEqual(command[command.index("--model") + 1], "deepseek/deepseek-v3.2") self.assertIn('model_provider="openrouter"', command) self.assertIn( 'model_providers.openrouter.base_url="https://openrouter.ai/api/v1"', command, ) self.assertIn('model_providers.openrouter.env_key="OPENROUTER_API_KEY"', command) def test_wall_clock_timeout_kills_process_independent_of_iteration_limit(self): fake_process = HangingCodexProcess([]) with mock.patch.object(common.subprocess, "Popen", return_value=fake_process): with self.assertRaises(subprocess.TimeoutExpired): common.run_streaming(["codex"], "body", {}, timeout=0.05, iteration_limit=20) self.assertTrue(fake_process.killed) class CommonRunClaudeStreamingTests(unittest.TestCase): """FakeCodexProcess/HangingCodexProcess are runtime-agnostic Popen stand-ins (they just replay JSONL lines) despite the name -- reused here for common.run_claude_streaming()/parse_claude_events() rather than duplicating them under a Claude-specific name.""" def test_wall_clock_timeout_kills_process_independent_of_iteration_limit(self): fake_process = HangingCodexProcess([]) with mock.patch.object(common.subprocess, "Popen", return_value=fake_process): with self.assertRaises(subprocess.TimeoutExpired): common.run_claude_streaming( ["claude"], "body", {}, "/repo/agents", timeout=0.05, iteration_limit=20 ) self.assertTrue(fake_process.killed) def test_cwd_is_passed_to_popen_explicitly(self): # Claude has no -C-equivalent CLI flag -- unlike Codex, cwd has to # be asserted via the Popen kwarg, not an argv token. fake_process = FakeCodexProcess( [ '{"type":"result","subtype":"success","is_error":false,' '"result":"ok","session_id":"session-1"}\n', ] ) with mock.patch.object( common.subprocess, "Popen", return_value=fake_process ) as popen: common.run_claude_streaming( ["claude"], "body", {}, "/repo/agents", timeout=5, iteration_limit=20 ) self.assertEqual(popen.call_args.kwargs["cwd"], "/repo/agents") def test_parse_claude_events_returns_result_text_and_session_id(self): output = ( '{"type":"assistant","message":{"content":[{"type":"text","text":"..."}]}}\n' '{"type":"result","subtype":"success","is_error":false,' '"result":"Testantwort","session_id":"session-1"}\n' ) response, session_id = common.parse_claude_events(output) self.assertEqual(session_id, "session-1") self.assertEqual(response, "Testantwort") def test_parse_claude_events_raises_without_session_id(self): output = '{"type":"result","subtype":"success","is_error":false,"result":"x"}\n' with self.assertRaises(RuntimeError): common.parse_claude_events(output) def test_parse_claude_events_raises_on_is_error(self): output = ( '{"type":"result","subtype":"error_during_execution","is_error":true,' '"result":"Kontingent aufgebraucht","session_id":"session-1"}\n' ) with self.assertRaises(RuntimeError) as context: common.parse_claude_events(output) self.assertIn("Kontingent aufgebraucht", str(context.exception)) def test_parse_claude_telemetry_uses_final_aggregate_without_prompt_content(self): output = ( '{"type":"assistant","message":{"model":"claude-sonnet-5",' '"content":[{"type":"tool_use","name":"Read",' '"input":{"file_path":"/secret/prompt"}}],' '"usage":{"input_tokens":999}}}\n' '{"type":"result","session_id":"claude-session","num_turns":3,' '"duration_ms":7200,"duration_api_ms":6100,"ttft_ms":900,' '"usage":{"input_tokens":12,"cache_creation_input_tokens":100,' '"cache_read_input_tokens":800,"output_tokens":44,' '"output_tokens_details":{"thinking_tokens":7}},' '"modelUsage":{"claude-haiku":{"costUSD":0.001},' '"claude-sonnet-5":{"costUSD":0.02}}}\n' ) telemetry = common.parse_claude_telemetry(output) self.assertEqual(telemetry["session_id"], "claude-session") self.assertEqual(telemetry["actual_profile"]["model"], "claude-sonnet-5") self.assertEqual( telemetry["usage"], { "input_tokens": 912, "cached_input_tokens": 800, "cache_creation_input_tokens": 100, "output_tokens": 44, "reasoning_tokens": 7, "total_tokens": 956, }, ) self.assertEqual( telemetry["timing"], { "duration_seconds": 7.2, "model_duration_seconds": 6.1, "ttft_seconds": 0.9, }, ) self.assertEqual(telemetry["iterations"], 3) self.assertNotIn("secret", json.dumps(telemetry)) def test_build_claude_partial_response_concatenates_text_blocks(self): output = ( '{"type":"assistant","message":{"content":' '[{"type":"text","text":"Erster Teil."}]}}\n' '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n' '{"type":"assistant","message":{"content":' '[{"type":"text","text":"Zweiter Teil."}]}}\n' ) response = common.build_claude_partial_response(output) self.assertIn("Erster Teil.", response) self.assertIn("Zweiter Teil.", response) def test_build_claude_partial_response_is_empty_on_no_text(self): self.assertEqual(common.build_claude_partial_response(""), "") self.assertEqual(common.build_claude_partial_response("not json\n"), "") class ListenerTests(unittest.TestCase): def test_multiline_message_is_queued_for_allowlisted_sender(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_ALLOWED_SENDERS": "lukas@0ff.space", "AGENT_QUEUE_DIR": temporary, } with mock.patch.dict(os.environ, environment, clear=False): messages = listener.MessageBuffer(listener.enqueue) messages.feed( "2026-07-29T20:00:00+02:00 lukas@0ff.space: erste Zeile" ) messages.feed("zweite Zeile") messages.flush() jobs = list(Path(temporary).glob("*.json")) self.assertEqual(len(jobs), 1) payload = json.loads(jobs[0].read_text()) self.assertEqual(payload["sender"], "lukas@0ff.space") self.assertEqual(payload["body"], "erste Zeile\nzweite Zeile") def test_non_allowlisted_sender_is_ignored(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_ALLOWED_SENDERS": "lukas@0ff.space", "AGENT_QUEUE_DIR": temporary, } with mock.patch.dict(os.environ, environment, clear=False): listener.enqueue("other@0ff.space", "nicht ausführen") self.assertEqual(list(Path(temporary).glob("*.json")), []) def test_domain_wildcard_allows_any_sender_on_that_domain(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_ALLOWED_SENDERS": "@0ff.space", "AGENT_QUEUE_DIR": temporary, } with mock.patch.dict(os.environ, environment, clear=False): listener.enqueue("alfo@0ff.space", "hallo") listener.enqueue("someone@elsewhere.example", "sollte ignoriert werden") jobs = list(Path(temporary).glob("*.json")) self.assertEqual(len(jobs), 1) payload = json.loads(jobs[0].read_text()) self.assertEqual(payload["sender"], "alfo@0ff.space") def test_debouncer_combines_messages_within_window(self): queued = [] now = [10.0] debouncer = listener.MessageDebouncer( lambda sender, body: queued.append((sender, body)), delay=3, clock=lambda: now[0], ) debouncer.add("lukas@0ff.space", "erste Hälfte") now[0] = 12.0 debouncer.add("lukas@0ff.space", "zweite Hälfte") now[0] = 14.9 debouncer.flush_due() self.assertEqual(queued, []) now[0] = 15.0 debouncer.flush_due() self.assertEqual( queued, [("lukas@0ff.space", "erste Hälfte\nzweite Hälfte")], ) def test_new_command_flushes_pending_message_first(self): queued = [] debouncer = listener.MessageDebouncer( lambda sender, body: queued.append((sender, body)), delay=3, ) debouncer.add("lukas@0ff.space", "alte Anfrage") debouncer.add("lukas@0ff.space", "/neu") self.assertEqual( queued, [ ("lukas@0ff.space", "alte Anfrage"), ("lukas@0ff.space", "/neu"), ], ) class WorkerTests(unittest.TestCase): def test_parse_codex_events_concatenates_multiple_agent_messages(self): # A single turn can emit more than one item.completed/agent_message # (e.g. an intermediate update before the final reply). Keeping # only the last one silently dropped earlier text -- looked like a # long reply getting cut off with no error (found 2026-08-01). output = ( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Erster Teil."}}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Zweiter Teil."}}\n' ) response, session_id = worker.parse_codex_events(output) self.assertEqual(session_id, "session-1") self.assertIn("Erster Teil.", response) self.assertIn("Zweiter Teil.", response) def test_parse_codex_events_tolerates_malformed_intermediate_event(self): output = ( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"text":"abgeschnitten}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Fertige Antwort."}}\n' ) response, session_id = worker.parse_codex_events(output) self.assertEqual(session_id, "session-1") self.assertEqual(response, "Fertige Antwort.") def test_parse_codex_events_rejects_malformed_output_without_answer(self): output = ( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"text":"abgeschnitten}\n' ) with self.assertRaisesRegex(RuntimeError, "Ungültige Codex-JSONL-Ausgabe"): worker.parse_codex_events(output) def test_parse_codex_telemetry_sums_completed_turn_usage(self): output = ( '{"type":"thread.started","thread_id":"codex-session"}\n' '{"type":"item.completed","item":{"type":"command_execution",' '"command":"cat /secret/prompt"}}\n' '{"type":"turn.completed","usage":{"input_tokens":1000,' '"cached_input_tokens":700,"output_tokens":90,' '"output_tokens_details":{"reasoning_tokens":30}}}\n' ) telemetry = common.parse_codex_telemetry(output) self.assertEqual(telemetry["session_id"], "codex-session") self.assertEqual( telemetry["usage"], { "input_tokens": 1000, "cached_input_tokens": 700, "output_tokens": 90, "reasoning_tokens": 30, "total_tokens": 1090, }, ) self.assertEqual(telemetry["iterations"], 1) self.assertNotIn("secret", json.dumps(telemetry)) def test_claim_moves_jobs_to_processing(self): with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) inbox = root / "inbox" processing = root / "processing" inbox.mkdir() source = inbox / "1.json" source.write_text("{}") claimed = list(worker.claim_jobs(inbox, processing)) self.assertEqual(claimed, [processing / "1.json"]) self.assertFalse(source.exists()) self.assertTrue(claimed[0].exists()) def test_codex_approval_option_precedes_exec_subcommand(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_CODEX_MODEL": "gpt-5.6-terra", "AGENT_CODEX_REASONING_EFFORT": "medium", } completed = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=completed) as run, ): response, session_id = worker.run_codex({"body": "Test"}) command = run.call_args.args[0] self.assertLess(command.index("--ask-for-approval"), command.index("exec")) self.assertEqual(response, "Testantwort") self.assertEqual(session_id, "session-1") self.assertNotIn("--ephemeral", command) self.assertIn("--json", command) self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-terra") reasoning = next(i for i, value in enumerate(command) if value.startswith("model_reasoning_effort=")) self.assertEqual(command[reasoning - 1], "-c") self.assertEqual(command[reasoning], 'model_reasoning_effort="medium"') def test_codex_model_controls_precede_exec_and_reject_invalid_values(self): command = ["codex"] with mock.patch.dict( os.environ, {"AGENT_CODEX_MODEL": "gpt-5.6-terra", "AGENT_CODEX_REASONING_EFFORT": "high"}, clear=False, ): worker.append_model_options(command) self.assertEqual(command, ["codex", "--model", "gpt-5.6-terra", "-c", 'model_reasoning_effort="high"']) with mock.patch.dict(os.environ, {"AGENT_CODEX_MODEL": "bad model"}, clear=False): with self.assertRaises(ValueError): worker.append_model_options([]) def test_existing_session_uses_resume_subcommand(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", } completed = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Fortsetzung"}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=completed) as run, ): response, session_id = worker.run_codex( {"body": "Weiter"}, session_id="session-1" ) command = run.call_args.args[0] self.assertEqual(command[command.index("exec") + 1], "resume") self.assertIn("session-1", command) self.assertEqual(response, "Fortsetzung") self.assertEqual(session_id, "session-1") def test_agent_gets_read_only_add_dir_for_readable_repos(self): # The Agent must be able to see the actual platform repos to answer # questions about current state (found 2026-07-31: it previously # only ever saw agents/ itself) -- but --sandbox stays read-only, # unlike the Builder's workspace-write. environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_READABLE_DIRS": "/repo/nixos-config:/repo/website", "AGENT_CODEX_MODEL": "gpt-5.6-terra", "AGENT_CODEX_REASONING_EFFORT": "medium", } completed = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=completed) as run, ): worker.run_codex({"body": "Test"}) command = run.call_args.args[0] sandbox_index = command.index("--sandbox") self.assertEqual(command[sandbox_index + 1], "read-only") add_dir_indices = [i for i, arg in enumerate(command) if arg == "--add-dir"] added_dirs = [command[i + 1] for i in add_dir_indices] self.assertEqual(added_dirs, ["/repo/nixos-config", "/repo/website"]) def test_agent_without_readable_dirs_configured_gets_no_add_dir(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", } completed = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=completed) as run, ): worker.run_codex({"body": "Test"}) command = run.call_args.args[0] self.assertNotIn("--add-dir", command) def test_first_reply_notice_is_only_added_once_per_sender(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), "AGENT_BUILDER_STATE_DIR": str(Path(temporary) / "state"), } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): first_response, marker = worker.add_first_reply_notice( sender, "Antwort eins" ) self.assertIn("innerhalb von drei Sekunden", first_response) self.assertIsNotNone(marker) worker.mark_introduced(marker) second_response, marker = worker.add_first_reply_notice( sender, "Antwort zwei" ) self.assertEqual(second_response, "Antwort zwei") self.assertIsNone(marker) def test_expired_session_is_not_loaded(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), "AGENT_SESSION_IDLE_SECONDS": "1800", } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): worker.db.save_session(worker.ROLE, sender, "session-1", updated_at=100) session_id = worker.load_session(sender, now=lambda: 1901) stored_session = worker.db.load_session(worker.ROLE, sender) self.assertIsNone(session_id) self.assertIsNone(stored_session) def test_extract_handoff_returns_unchanged_response_without_marker(self): response, spec = worker.extract_handoff("Ganz normale Chat-Antwort.") self.assertEqual(response, "Ganz normale Chat-Antwort.") self.assertIsNone(spec) def test_extract_handoff_strips_marker_and_returns_spec(self): response = ( "Alles klar, ich übergebe das jetzt.\n" "===BUILDER-HANDOFF===\n" "Ändere X in Y.\n" "===END-HANDOFF===\n" "Ich sag dir Bescheid." ) visible, spec = worker.extract_handoff(response) self.assertEqual(spec, "Ändere X in Y.") self.assertIn("Alles klar, ich übergebe das jetzt.", visible) self.assertIn("Ich sag dir Bescheid.", visible) self.assertNotIn("BUILDER-HANDOFF", visible) def test_extract_handoff_uses_default_confirmation_without_surrounding_text(self): response = "===BUILDER-HANDOFF===\nÄndere X in Y.\n===END-HANDOFF===" visible, spec = worker.extract_handoff(response) self.assertEqual(spec, "Ändere X in Y.") self.assertEqual(visible, worker.DEFAULT_HANDOFF_CONFIRMATION) def test_enqueue_builder_job_addresses_owner_and_records_requester(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_BUILDER_QUEUE_DIR": temporary, "AGENT_BUILDER_ALLOWED_SENDERS": "lukas@0ff.space", } with mock.patch.dict(os.environ, environment, clear=False): worker.enqueue_builder_job("alfo@0ff.space", "Ändere X in Y.") jobs = list(Path(temporary).glob("*.json")) self.assertEqual(len(jobs), 1) payload = json.loads(jobs[0].read_text(encoding="utf-8")) self.assertEqual(payload["sender"], "lukas@0ff.space") self.assertEqual(payload["requested_by"], "alfo@0ff.space") self.assertEqual(payload["body"], "Ändere X in Y.") def test_extract_task_edit_returns_unchanged_response_without_marker(self): response, spec = worker.extract_task_edit("Ganz normale Chat-Antwort.") self.assertEqual(response, "Ganz normale Chat-Antwort.") self.assertIsNone(spec) def test_extract_task_edit_strips_marker_and_returns_spec(self): response = ( "Klar, ich passe die Liste an.\n" "===TASK-EDIT===\n" "Füg 'Wäsche waschen' hinzu.\n" "===END-TASK-EDIT===\n" "Bin gleich fertig." ) visible, spec = worker.extract_task_edit(response) self.assertEqual(spec, "Füg 'Wäsche waschen' hinzu.") self.assertIn("Klar, ich passe die Liste an.", visible) self.assertIn("Bin gleich fertig.", visible) self.assertNotIn("TASK-EDIT", visible) def test_extract_calendar_events_returns_unchanged_response_without_marker(self): response, events = worker.extract_calendar_events("Ganz normale Chat-Antwort.") self.assertEqual(response, "Ganz normale Chat-Antwort.") self.assertEqual(events, []) def test_extract_calendar_events_parses_well_formed_block(self): response = ( "Alles klar, ich trag das ein.\n" "===CALENDAR-EVENT===\n" "TITLE: Zahnarzt\n" "START: 2026-08-05T14:00\n" "END: 2026-08-05T15:00\n" "DESCRIPTION: Kontrolle\n" "===END-CALENDAR-EVENT===\n" ) visible, events = worker.extract_calendar_events(response) self.assertEqual( events, [ { "title": "Zahnarzt", "start": "2026-08-05T14:00", "end": "2026-08-05T15:00", "description": "Kontrolle", } ], ) self.assertIn("Alles klar, ich trag das ein.", visible) self.assertNotIn("CALENDAR-EVENT", visible) def test_extract_calendar_events_reports_error_when_required_field_missing(self): # No END time -- must not crash, and must not silently guess a # duration (see agent.md's "Umgang mit Unsicherheiten"). response = ( "===CALENDAR-EVENT===\n" "TITLE: Zahnarzt\n" "START: 2026-08-05T14:00\n" "===END-CALENDAR-EVENT===\n" ) visible, events = worker.extract_calendar_events(response) self.assertEqual(events, ["Kalendereintrag konnte nicht angelegt werden: TITLE, START und END müssen alle gesetzt sein."]) self.assertEqual(visible, "") def test_extract_calendar_events_parses_multiple_blocks_in_one_reply(self): response = ( "Trag ich beide ein.\n" "===CALENDAR-EVENT===\n" "TITLE: Zahnarzt\n" "START: 2026-08-05T14:00\n" "END: 2026-08-05T15:00\n" "===END-CALENDAR-EVENT===\n" "===CALENDAR-EVENT===\n" "TITLE: Friseur\n" "START: 2026-08-06T10:00\n" "END: 2026-08-06T11:00\n" "===END-CALENDAR-EVENT===\n" ) visible, events = worker.extract_calendar_events(response) self.assertEqual(len(events), 2) self.assertEqual(events[0]["title"], "Zahnarzt") self.assertEqual(events[1]["title"], "Friseur") self.assertIn("Trag ich beide ein.", visible) self.assertNotIn("CALENDAR-EVENT", visible) def test_task_edit_codex_uses_workspace_write_sandbox_scoped_to_tasks_dir_only(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_TASKS_REPO_DIR": "/home/drone/tasks", "AGENT_READABLE_DIRS": "/repo/nixos-config:/repo/website", "AGENT_CODEX_MODEL": "gpt-5.6-terra", "AGENT_CODEX_REASONING_EFFORT": "medium", } completed = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=completed) as run, ): worker.run_task_edit_codex("Füg einen Task hinzu.") command = run.call_args.args[0] sandbox_index = command.index("--sandbox") self.assertEqual(command[sandbox_index + 1], "workspace-write") add_dir_indices = [i for i, arg in enumerate(command) if arg == "--add-dir"] add_dirs = [command[i + 1] for i in add_dir_indices] # Exactly the tasks dir, never any of AGENT_READABLE_DIRS. self.assertEqual(add_dirs, ["/home/drone/tasks"]) self.assertNotIn("sandbox_workspace_write.network_access=true", command) self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-terra") # -C must be inside the one --add-dir, not AGENT_ROLE_DIR -- # Codex's own sandbox fails outright ("bwrap: Can't mkdir # .../.git: Read-only file system") setting up its .git # write-protection bind mount if the working directory isn't # inside a writable --add-dir (found live 2026-08-01, a real # production failure -- Builder's identical-looking -C works # only because nixos-config is itself one of ITS --add-dir # targets, this sub-invocation has no such overlap). c_index = command.index("-C") self.assertEqual(command[c_index + 1], "/home/drone/tasks") self.assertEqual(run.call_args.kwargs["input"], "Füg einen Task hinzu.") def test_run_calendar_event_invokes_wrapper_with_env_credentials(self): with tempfile.TemporaryDirectory() as temporary: password_file = Path(temporary) / "pw" password_file.write_text("s3cret\n", encoding="utf-8") environment = { "AGENT_CALENDAR_ADD_EVENT_BIN": "/bin/calendar-add-event", "AGENT_CALENDAR_URL": "https://calendar.0ff.space/calendar/public", "AGENT_CALENDAR_USER": "calendar", "AGENT_CALENDAR_PASSWORD_FILE": str(password_file), } completed = SimpleNamespace(returncode=0, stdout="", stderr="") with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker.subprocess, "run", return_value=completed ) as run, ): result = worker.run_calendar_event( {"title": "Zahnarzt", "start": "2026-08-05T14:00", "end": "2026-08-05T15:00"} ) command = run.call_args.args[0] self.assertEqual( command, [ "/bin/calendar-add-event", "--title", "Zahnarzt", "--start", "2026-08-05T14:00", "--end", "2026-08-05T15:00", ], ) call_env = run.call_args.kwargs["env"] self.assertEqual(call_env["CALENDAR_PASSWORD"], "s3cret") self.assertEqual(call_env["CALENDAR_USER"], "calendar") self.assertIn("angelegt", result) def _init_tasks_repo(self, directory): subprocess.run(["git", "init", "-q", directory], check=True) subprocess.run( ["git", "-C", directory, "config", "user.email", "test@example.com"], check=True, ) subprocess.run( ["git", "-C", directory, "config", "user.name", "Test"], check=True ) (Path(directory) / "index.html").write_text("\n") subprocess.run(["git", "-C", directory, "add", "-A"], check=True) subprocess.run( ["git", "-C", directory, "commit", "-q", "-m", "init"], check=True ) def test_run_task_edit_commits_when_marker_present(self): with tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir: subprocess.run(["git", "init", "-q", "--bare", bare_dir], check=True) subprocess.run(["git", "clone", "-q", bare_dir, clone_dir], check=True) self._init_tasks_repo(clone_dir) subprocess.run( ["git", "-C", clone_dir, "checkout", "-q", "-B", "main"], check=True ) subprocess.run( ["git", "-C", clone_dir, "push", "-q", "origin", "main"], check=True ) def fake_run_task_edit_codex(spec): (Path(clone_dir) / "index.html").write_text("\n") return ( "Erledigt.\n" f"{worker.COMMIT_START}\n" "tasks: add item\n" f"{worker.COMMIT_END}\n" ) # tasks_repo_dir() reads AGENT_TASKS_REPO_DIR itself, so drive # run_task_edit() through the real env var instead of a direct # argument (it takes none -- matches run_and_commit()'s use of # writable_dirs() in builder.py). environment = {"AGENT_TASKS_REPO_DIR": clone_dir} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker, "run_task_edit_codex", side_effect=fake_run_task_edit_codex ), ): result = worker.run_task_edit("Füg einen Task hinzu.") self.assertIn("committed und nach main gepusht", result) # Explicit ref, not bare HEAD -- the bare repo's symbolic HEAD # still points at whatever init.defaultBranch is (often # "master", never created here), so `git log -1` without a ref # would resolve an unborn branch and fail with "does not have # any commits yet" even though refs/heads/main is populated. remote_log = subprocess.run( ["git", "--git-dir", bare_dir, "log", "-1", "--format=%s", "refs/heads/main"], capture_output=True, text=True, check=True, ) self.assertEqual(remote_log.stdout.strip(), "tasks: add item") def test_run_task_edit_skips_commit_without_marker(self): with tempfile.TemporaryDirectory() as temporary: self._init_tasks_repo(temporary) environment = {"AGENT_TASKS_REPO_DIR": temporary} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker, "run_task_edit_codex", return_value="Ich hab noch eine Rückfrage, bevor ich das ändere.", ), ): result = worker.run_task_edit("Mach irgendwas.") self.assertEqual( result, "Ich hab noch eine Rückfrage, bevor ich das ändere." ) log = subprocess.run( ["git", "-C", temporary, "log", "--oneline"], capture_output=True, text=True, check=True, ) self.assertEqual(log.stdout.strip().count("\n"), 0) class BuilderTests(unittest.TestCase): # No builder-level parse_codex_events test here anymore: builder.py # dropped that import when it moved to Claude (run_claude() uses # common.parse_claude_events() instead, covered by # CommonRunClaudeStreamingTests below). common.parse_codex_events() # itself is still fully covered via WorkerTests, which exercises the # untouched Agent/Codex path. def test_context_round_trip(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.save_context(sender, "Zwischenstand eins") context = builder.load_context(sender, now=lambda: 100) self.assertEqual(context, "Zwischenstand eins") def test_context_is_scoped_by_board_slug_not_shared_across_cards(self): # Every board job shares one fixed owner sender identity -- an # unscoped context would let two concurrently-running, unrelated # jobs (or a board job and direct XMPP chat) bleed into each # other now that agent-codex-builder runs more than one instance # at once (found live 2026-09-09, # board-worktree-parallelism-smoke-test-a). with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.save_context(sender, "card A's interim report", scope="card-a") builder.save_context(sender, "unrelated direct chat reply", scope="") builder.save_context(sender, "card B's interim report", scope="card-b") context_a = builder.load_context(sender, scope="card-a") context_chat = builder.load_context(sender, scope="") context_b = builder.load_context(sender, scope="card-b") self.assertEqual(context_a, "card A's interim report") self.assertEqual(context_chat, "unrelated direct chat reply") self.assertEqual(context_b, "card B's interim report") def test_run_and_commit_scopes_context_by_board_slug(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as temporary, ): subprocess.run(["git", "init", "-q", repo_dir], check=True) sender = "lukas@0ff.space" environment = { "AGENT_BUILDER_WRITABLE_DIRS": repo_dir, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "run_runtime", return_value=("Erledigt A", False)), ): builder.run_and_commit( "job-a", sender, sender, "Auftrag A", payload={"id": "job-a", "sender": sender, "board_slug": "card-a"}, ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "run_runtime", return_value=("Erledigt B", False)), ): builder.run_and_commit( "job-b", sender, sender, "Auftrag B", payload={"id": "job-b", "sender": sender, "board_slug": "card-b"}, ) with mock.patch.dict(os.environ, environment, clear=False): self.assertEqual(builder.load_context(sender, scope="card-a"), "Erledigt A") self.assertEqual(builder.load_context(sender, scope="card-b"), "Erledigt B") def test_context_table_migrates_pre_scope_rows(self): with tempfile.TemporaryDirectory() as temporary: db_path = str(Path(temporary) / "agent.db") connection = builder.db.sqlite3.connect(db_path) connection.executescript( """ CREATE TABLE context ( role TEXT NOT NULL, sender TEXT NOT NULL, context_text TEXT NOT NULL, updated_at REAL NOT NULL, PRIMARY KEY (role, sender) ); INSERT INTO context VALUES ('builder', 'lukas@0ff.space', 'vor der Migration', 100); """ ) connection.commit() connection.close() environment = {"AGENT_DB_PATH": db_path} with mock.patch.dict(os.environ, environment, clear=False): migrated = builder.load_context("lukas@0ff.space", now=lambda: 100) self.assertEqual(migrated, "vor der Migration") def test_expired_context_is_not_loaded(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), "AGENT_SESSION_IDLE_SECONDS": "1800", } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.db.save_context(builder.ROLE, sender, "alt", updated_at=100) context = builder.load_context(sender, now=lambda: 1901) stored_context = builder.db.load_context(builder.ROLE, sender) self.assertIsNone(context) self.assertIsNone(stored_context) def test_neu_command_resets_context_not_session(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.save_context(sender, "Zwischenstand") builder.reset_context(sender) context = builder.load_context(sender) self.assertIsNone(context) def test_first_reply_notice_is_only_added_once_per_sender(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): first_response, marker = builder.add_first_reply_notice( sender, "Antwort eins" ) self.assertIn("nixos-rebuild switch", first_response) self.assertIsNotNone(marker) builder.mark_introduced(marker) second_response, marker = builder.add_first_reply_notice( sender, "Antwort zwei" ) self.assertEqual(second_response, "Antwort zwei") self.assertIsNone(marker) def test_extract_commit_message_returns_unchanged_response_without_marker(self): response, message = builder.extract_commit_message("Nur eine Antwort, kein Commit.") self.assertEqual(response, "Nur eine Antwort, kein Commit.") self.assertIsNone(message) def test_extract_commit_message_strips_marker_and_returns_message(self): response = ( "Ich habe X geändert.\n" "===COMMIT-MESSAGE===\n" "fix: X anpassen\n" "===END-COMMIT-MESSAGE===" ) visible, message = builder.extract_commit_message(response) self.assertEqual(message, "fix: X anpassen") self.assertIn("Ich habe X geändert.", visible) self.assertNotIn("COMMIT-MESSAGE", visible) def _init_repo(self, directory): subprocess.run(["git", "init", "-q", directory], check=True) subprocess.run( ["git", "-C", directory, "config", "user.email", "test@example.com"], check=True, ) subprocess.run( ["git", "-C", directory, "config", "user.name", "Test"], check=True ) (Path(directory) / "committed.txt").write_text("original\n") subprocess.run(["git", "-C", directory, "add", "-A"], check=True) subprocess.run( ["git", "-C", directory, "commit", "-q", "-m", "init"], check=True ) def test_commit_and_push_uses_checked_out_branch_not_hardcoded_dev(self): # dashboard/launcher only have a `main` branch (no `dev`), and their # deploy webhooks pull whatever branch is checked out -- pushing to # a hardcoded "dev" would silently create an orphan branch nobody # deploys instead of updating `main`. with tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir: subprocess.run(["git", "init", "-q", "--bare", bare_dir], check=True) subprocess.run( ["git", "clone", "-q", bare_dir, clone_dir], check=True ) self._init_repo(clone_dir) subprocess.run( ["git", "-C", clone_dir, "checkout", "-q", "-B", "main"], check=True ) subprocess.run( ["git", "-C", clone_dir, "push", "-q", "origin", "main"], check=True ) (Path(clone_dir) / "changed.txt").write_text("builder change\n") ok, _detail, branch = builder.commit_and_push( Path(clone_dir), "test: branch-aware push" ) self.assertTrue(ok) self.assertEqual(branch, "main") remote_branches = subprocess.run( ["git", "--git-dir", bare_dir, "branch"], capture_output=True, text=True, check=True, ) self.assertIn("main", remote_branches.stdout) self.assertNotIn("dev", remote_branches.stdout) def test_stash_and_restore_foreign_changes(self): with tempfile.TemporaryDirectory() as temporary: self._init_repo(temporary) repo_dir = Path(temporary) foreign_path = repo_dir / "foreign.txt" foreign_path.write_text("pre-existing scratch work\n") stashed = builder.stash_foreign_changes(repo_dir, "test-job") self.assertTrue(stashed) self.assertFalse(foreign_path.exists()) builder_path = repo_dir / "builder-change.txt" builder_path.write_text("builder wrote this\n") subprocess.run(["git", "-C", str(repo_dir), "add", "-A"], check=True) subprocess.run( ["git", "-C", str(repo_dir), "commit", "-q", "-m", "builder change"], check=True, ) log = subprocess.run( ["git", "-C", str(repo_dir), "show", "--stat", "-1"], capture_output=True, text=True, check=True, ) self.assertIn("builder-change.txt", log.stdout) self.assertNotIn("foreign.txt", log.stdout) builder.restore_foreign_changes(repo_dir) self.assertTrue(foreign_path.exists()) self.assertEqual(foreign_path.read_text(), "pre-existing scratch work\n") def test_stashed_untracked_intent_is_temporarily_absent(self): with tempfile.TemporaryDirectory() as temporary: self._init_repo(temporary) intent_path = Path(temporary) / "docs" / "intent" / "slug.md" intent_path.parent.mkdir(parents=True) intent_path.write_text("# Intent\n") self.assertTrue(builder.stash_foreign_changes(Path(temporary), "test-job")) self.assertFalse(intent_path.exists()) self.assertTrue(builder.restore_foreign_changes(Path(temporary))) self.assertTrue(intent_path.exists()) def test_run_and_commit_reports_failed_restore(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as temporary, ): subprocess.run(["git", "init", "-q", repo_dir], check=True) environment = { "AGENT_BUILDER_WRITABLE_DIRS": repo_dir, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "stash_foreign_changes", return_value=True), mock.patch.object(builder, "restore_foreign_changes", return_value=False), mock.patch.object(builder, "run_runtime", return_value=("Erledigt", False)), ): response, _ = builder.run_and_commit( "job-restore", "lukas@0ff.space", "lukas@0ff.space", "Auftrag" ) self.assertIn("nicht wiederhergestellt", response) def test_stash_foreign_changes_is_noop_on_clean_repo(self): with tempfile.TemporaryDirectory() as temporary: self._init_repo(temporary) stashed = builder.stash_foreign_changes(Path(temporary), "test-job") self.assertFalse(stashed) def _init_bare_and_dev_clone(self, bare_dir, clone_dir): """Bare "origin" plus a clone checked out on `dev` and pushed -- the isolated-worktree tests below need a real remote, since create_worktree() branches from origin/dev, not the clone's own current HEAD.""" subprocess.run(["git", "init", "-q", "--bare", bare_dir], check=True) subprocess.run(["git", "clone", "-q", bare_dir, clone_dir], check=True) self._init_repo(clone_dir) subprocess.run(["git", "-C", clone_dir, "checkout", "-q", "-B", "dev"], check=True) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) def test_create_worktree_and_remove_worktree_roundtrip(self): with ( tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir, tempfile.TemporaryDirectory() as worktree_root, tempfile.TemporaryDirectory() as lock_dir, ): self._init_bare_and_dev_clone(bare_dir, clone_dir) repo_dir = Path(clone_dir) environment = { "AGENT_BUILDER_WORKTREE_DIR": worktree_root, "AGENT_BUILDER_LOCK_DIR": lock_dir, } with mock.patch.dict(os.environ, environment, clear=False): worktree_path = builder.create_worktree(repo_dir, "job-wt-1") self.assertTrue(worktree_path.exists()) self.assertEqual(worktree_path.name, repo_dir.name) branches = subprocess.run( ["git", "-C", str(repo_dir), "branch", "--list", "builder-job-job-wt-1"], capture_output=True, text=True, check=True, ) self.assertIn("builder-job-job-wt-1", branches.stdout) builder.remove_worktree(repo_dir, worktree_path) self.assertFalse(worktree_path.exists()) branches_after = subprocess.run( ["git", "-C", str(repo_dir), "branch", "--list", "builder-job-job-wt-1"], capture_output=True, text=True, check=True, ) self.assertNotIn("builder-job-job-wt-1", branches_after.stdout) def test_run_and_commit_isolates_flake_repo_in_worktree(self): with ( tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir, tempfile.TemporaryDirectory() as worktree_root, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as deploy_queue_dir, tempfile.TemporaryDirectory() as temporary, ): self._init_bare_and_dev_clone(bare_dir, clone_dir) (Path(clone_dir) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", clone_dir, "add", "-A"], check=True) subprocess.run(["git", "-C", clone_dir, "commit", "-q", "-m", "add flake"], check=True) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) captured_directories = [] captured_role_dirs = [] captured_payloads = [] def fake_run_runtime(payload, runtime, directories=None, role_dir=None): captured_payloads.append(dict(payload)) captured_directories.extend(directories) captured_role_dirs.append(role_dir) for directory in directories: if directory.name == Path(clone_dir).name: (directory / "new-file.txt").write_text("from the model\n") return ( f"Erledigt.\n\n{builder.COMMIT_START}\ntest commit\n{builder.COMMIT_END}", False, ) environment = { "AGENT_BUILDER_WRITABLE_DIRS": clone_dir, "AGENT_BUILDER_WORKTREE_DIR": worktree_root, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_BUILDER_STATE_DIR": state_dir, "AGENT_BUILDER_DEPLOY_QUEUE_DIR": deploy_queue_dir, "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "run_runtime", side_effect=fake_run_runtime), ): response, limit_reached = builder.run_and_commit( "job-wt-2", "lukas@0ff.space", "lukas@0ff.space", "Auftrag", payload={ "id": "job-wt-2", "sender": "lukas@0ff.space", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 17, }, ) self.assertFalse(limit_reached) self.assertIn("committed und nach dev gepusht", response) # The model wrote into an isolated worktree, never the shared checkout. self.assertEqual(len(captured_directories), 1) self.assertNotEqual(captured_directories[0], Path(clone_dir)) self.assertFalse((Path(clone_dir) / "new-file.txt").exists()) self.assertFalse(captured_directories[0].exists()) # -C must resolve inside a --add-dir target (see role_dir's own # comment in run_and_commit()) -- the worktree's own agents/, # not AGENT_ROLE_DIR pointing at the now-unwritable main checkout. self.assertEqual(captured_role_dirs, [captured_directories[0] / "agents"]) self.assertEqual(captured_payloads[0]["model"], "gpt-5.6-luna") self.assertEqual(captured_payloads[0]["reasoning_effort"], "low") self.assertEqual(captured_payloads[0]["iteration_limit"], 17) self.assertEqual(captured_payloads[0]["body"], "Auftrag") log = subprocess.run( ["git", "--git-dir", bare_dir, "log", "dev", "--name-only", "-1"], capture_output=True, text=True, check=True, ) self.assertIn("new-file.txt", log.stdout) def test_apply_commit_worktree_push_conflict_preserves_recoverable_work(self): with ( tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir, tempfile.TemporaryDirectory() as worktree_root, tempfile.TemporaryDirectory() as lock_dir, ): self._init_bare_and_dev_clone(bare_dir, clone_dir) (Path(clone_dir) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", clone_dir, "add", "-A"], check=True) subprocess.run(["git", "-C", clone_dir, "commit", "-q", "-m", "add flake"], check=True) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) environment = { "AGENT_BUILDER_WRITABLE_DIRS": clone_dir, "AGENT_BUILDER_WORKTREE_DIR": worktree_root, "AGENT_BUILDER_LOCK_DIR": lock_dir, } with mock.patch.dict(os.environ, environment, clear=False): worktree_path = builder.create_worktree(Path(clone_dir), "job-wt-3") # Someone else's commit lands on dev after the worktree branched. (Path(clone_dir) / "other-change.txt").write_text("someone else\n") subprocess.run(["git", "-C", clone_dir, "add", "-A"], check=True) subprocess.run( ["git", "-C", clone_dir, "commit", "-q", "-m", "other change"], check=True ) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) (worktree_path / "job-change.txt").write_text("this job\n") result = builder.apply_commit("test commit", dirs=[worktree_path]) self.assertIn("Commit/Push fehlgeschlagen", result) self.assertTrue(worktree_path.exists()) self.assertEqual((worktree_path / "job-change.txt").read_text(), "this job\n") def test_queue_for_approval_and_freigeben_worktree_roundtrip(self): with ( tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir, tempfile.TemporaryDirectory() as worktree_root, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as pending_dir, tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as deploy_queue_dir, ): self._init_bare_and_dev_clone(bare_dir, clone_dir) (Path(clone_dir) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", clone_dir, "add", "-A"], check=True) subprocess.run(["git", "-C", clone_dir, "commit", "-q", "-m", "add flake"], check=True) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) environment = { "AGENT_BUILDER_WRITABLE_DIRS": clone_dir, "AGENT_BUILDER_WORKTREE_DIR": worktree_root, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_BUILDER_PENDING_DIR": pending_dir, "AGENT_BUILDER_STATE_DIR": state_dir, "AGENT_BUILDER_DEPLOY_QUEUE_DIR": deploy_queue_dir, } with mock.patch.dict(os.environ, environment, clear=False): worktree_path = builder.create_worktree(Path(clone_dir), "job-wt-4") (worktree_path / "handoff-change.txt").write_text("from a handoff\n") message = builder.queue_for_approval( "job-wt-4", "alfo@0ff.space", "test commit", dirs=[worktree_path] ) self.assertIn("wartet auf deine Freigabe", message) # Kept alive, not cleaned up, while awaiting /freigeben. self.assertTrue(worktree_path.exists()) result = builder.handle_pending_command("/freigeben job-wt-4") self.assertIn("committed und nach dev gepusht", result) self.assertFalse(worktree_path.exists()) log = subprocess.run( ["git", "--git-dir", bare_dir, "log", "dev", "--name-only", "-1"], capture_output=True, text=True, check=True, ) self.assertIn("handoff-change.txt", log.stdout) def test_requeue_interrupted_job_cleans_up_leftover_worktree(self): with ( tempfile.TemporaryDirectory() as bare_dir, tempfile.TemporaryDirectory() as clone_dir, tempfile.TemporaryDirectory() as worktree_root, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as processing_dir, ): self._init_bare_and_dev_clone(bare_dir, clone_dir) (Path(clone_dir) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", clone_dir, "add", "-A"], check=True) subprocess.run(["git", "-C", clone_dir, "commit", "-q", "-m", "add flake"], check=True) subprocess.run(["git", "-C", clone_dir, "push", "-q", "origin", "dev"], check=True) environment = { "AGENT_BUILDER_WRITABLE_DIRS": clone_dir, "AGENT_BUILDER_WORKTREE_DIR": worktree_root, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_BUILDER_QUEUE_DIR": inbox_dir, } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "send_reply"), ): worktree_path = builder.create_worktree(Path(clone_dir), "job-wt-5") self.assertTrue(worktree_path.exists()) job_path = Path(processing_dir) / "job-wt-5.json" builder.write_payload( job_path, { "id": "job-wt-5", "sender": "lukas@0ff.space", "requested_by": "lukas@0ff.space", "body": "x", }, ) builder.requeue_interrupted_job(job_path) self.assertFalse(worktree_path.exists()) self.assertTrue((Path(inbox_dir) / "job-wt-5.json").exists()) def test_locked_repos_holds_exclusive_lock_and_releases_after(self): with tempfile.TemporaryDirectory() as repo, tempfile.TemporaryDirectory() as lock_dir: repo_dir = Path(repo) environment = {"AGENT_BUILDER_LOCK_DIR": lock_dir} with mock.patch.dict(os.environ, environment, clear=False): with builder.locked_repos([repo_dir]): probe = open(builder.lock_path(repo_dir), "w") with self.assertRaises(BlockingIOError): fcntl.flock(probe, fcntl.LOCK_EX | fcntl.LOCK_NB) probe.close() released = open(builder.lock_path(repo_dir), "w") fcntl.flock(released, fcntl.LOCK_EX | fcntl.LOCK_NB) fcntl.flock(released, fcntl.LOCK_UN) released.close() def test_queue_for_approval_stashes_and_records_pending(self): with tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as pending_dir: self._init_repo(temporary) repo_dir = Path(temporary) (repo_dir / "changed.txt").write_text("builder change\n") environment = { "AGENT_BUILDER_WRITABLE_DIRS": str(repo_dir), "AGENT_BUILDER_PENDING_DIR": pending_dir, } with mock.patch.dict(os.environ, environment, clear=False): result = builder.queue_for_approval( "job-1", "alfo@0ff.space", "test: pending change" ) record = builder.load_pending_approval("job-1") self.assertIn("job-1", result) self.assertFalse((repo_dir / "changed.txt").exists()) self.assertEqual(record["requested_by"], "alfo@0ff.space") self.assertEqual(record["commit_message"], "test: pending change") def test_handle_pending_command_freigeben_commits_stashed_change(self): with tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as pending_dir: self._init_repo(temporary) repo_dir = Path(temporary) (repo_dir / "changed.txt").write_text("builder change\n") environment = { "AGENT_BUILDER_WRITABLE_DIRS": str(repo_dir), "AGENT_BUILDER_PENDING_DIR": pending_dir, } with mock.patch.dict(os.environ, environment, clear=False): builder.queue_for_approval("job-2", "alfo@0ff.space", "test: approved change") with mock.patch.object( builder, "commit_and_push", return_value=(True, "abc123", "dev") ): result = builder.handle_pending_command("/freigeben job-2") record = builder.load_pending_approval("job-2") self.assertIn("committed", result) self.assertIsNone(record) def test_handle_pending_command_verwerfen_drops_change(self): with tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as pending_dir: self._init_repo(temporary) repo_dir = Path(temporary) (repo_dir / "changed.txt").write_text("builder change\n") environment = { "AGENT_BUILDER_WRITABLE_DIRS": str(repo_dir), "AGENT_BUILDER_PENDING_DIR": pending_dir, } with mock.patch.dict(os.environ, environment, clear=False): builder.queue_for_approval("job-3", "alfo@0ff.space", "test: discarded change") result = builder.handle_pending_command("/verwerfen job-3") record = builder.load_pending_approval("job-3") listing = subprocess.run( ["git", "-C", str(repo_dir), "stash", "list"], capture_output=True, text=True, check=True, ) self.assertIn("verworfen", result) self.assertIsNone(record) self.assertFalse((repo_dir / "changed.txt").exists()) self.assertNotIn("job-3", listing.stdout) def test_apply_commit_only_deploys_where_flake_nix_exists(self): with tempfile.TemporaryDirectory() as no_flake, tempfile.TemporaryDirectory() as has_flake: self._init_repo(no_flake) self._init_repo(has_flake) (Path(has_flake) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", has_flake, "add", "-A"], check=True) subprocess.run( ["git", "-C", has_flake, "commit", "-q", "-m", "add flake"], check=True ) (Path(no_flake) / "index.html").write_text("changed\n") (Path(has_flake) / "flake.nix").write_text("{ changed = true; }\n") environment = { "AGENT_BUILDER_WRITABLE_DIRS": f"{no_flake}:{has_flake}", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( builder, "commit_and_push", return_value=(True, "abc123", "dev") ) as commit, ): result = builder.apply_commit("test commit") self.assertEqual(commit.call_count, 2) self.assertIn(Path(no_flake).name, result) self.assertIn(Path(has_flake).name, result) no_flake_section, has_flake_section = result.split(f"{Path(has_flake).name}:") self.assertNotIn("Deploy wird", no_flake_section) self.assertIn("Deploy wird", has_flake_section) def test_apply_commit_defers_board_deployment_until_review(self): with tempfile.TemporaryDirectory() as has_flake: self._init_repo(has_flake) (Path(has_flake) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", has_flake, "add", "-A"], check=True) subprocess.run( ["git", "-C", has_flake, "commit", "-q", "-m", "add flake"], check=True ) (Path(has_flake) / "flake.nix").write_text("{ changed = true; }\n") with ( mock.patch.object( builder, "commit_and_push", return_value=(True, "abc123", "dev") ), mock.patch.object(builder, "queue_deployment") as queue_deployment, ): result = builder.apply_commit( "test commit", dirs=[Path(has_flake)], payload={"id": "board-card-1", "defer_deploy_until_review": True}, ) self.assertIn("Deploy ist bis zum erfolgreichen separaten Review gesperrt", result) queue_deployment.assert_not_called() def test_deploy_runner_rejects_wrong_activated_closure(self): commands = [] def run(command, **kwargs): commands.append(command) output = ("a" * 40 if "rev-parse" in command else "/nix/store/expected" if command[:2] == ["nix", "build"] else "/nix/store/different" if command[0] == "readlink" else "active" if "is-active" in command else "ok") return subprocess.CompletedProcess(command, 0, output, "") request = {"host": "blunix", "repo": "/home/drone/nixos-config", "commit_hash": "a" * 40} ok, output = deploy_runner.run_switch(request, { "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", }, run) self.assertFalse(ok) self.assertIn("does not match", output) self.assertNotIn("deployed_commit_hash", request) self.assertFalse(any("checkout" in command or "merge" in command for command in commands)) def test_deploy_runner_refuses_automatic_rollback(self): with tempfile.TemporaryDirectory() as state: deploy_runner.write_payload(Path(state) / "last-verified-deployment.json", {"deployed_commit_hash": "b" * 40}) calls = [] def run(command, **kwargs): calls.append(command) return subprocess.CompletedProcess(command, 1 if "merge-base" in command else 0, "a" * 40, "") ok, output = deploy_runner.run_switch({"host": "blunix", "repo": "/home/drone/nixos-config", "commit_hash": "a" * 40}, {"AGENT_BUILDER_STATE_DIR": state}, run) self.assertFalse(ok) self.assertIn("roll back", output) self.assertEqual(len(calls), 2) def test_deploy_runner_rejects_other_hosts(self): with self.assertRaises(ValueError): deploy_runner.run_switch( {"host": "tnix", "repo": "/repo"}, { "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", }, lambda *args, **kwargs: None, ) def test_deploy_runner_uses_independent_switch_command(self): calls = [] def fake_run(command, **kwargs): calls.append((command, kwargs)) if "rev-parse" in command: return subprocess.CompletedProcess(command, 0, "abc1234" + "0" * 33, "") if command[0] == "readlink" or command[:2] == ["nix", "build"]: return subprocess.CompletedProcess(command, 0, "/nix/store/test-system", "") if "is-active" in command: return subprocess.CompletedProcess(command, 0, "active\n", "") return subprocess.CompletedProcess(command, 0, "switch complete", "") request = {"job_id": "board-x-1", "host": "blunix", "commit_hash": "abc1234", "repo": "/home/drone/nixos-config"} environment = { "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", } ok, output = deploy_runner.run_switch(request, environment, fake_run) self.assertTrue(ok) self.assertIn("switch complete", output) self.assertIn("active", output) # calls[0:3] are sync_repo_to_dev()'s fetch/checkout/merge --ff-only. self.assertEqual(calls[2][0][-2:], ["--flake", "git+file:///home/drone/nixos-config?rev=" + "abc1234" + "0" * 33 + "#blunix"]) self.assertEqual( calls[3][0][-2:], ["is-active", "board-agent-trigger.service"], ) def test_deploy_runner_unresolved_revision_never_attempts_switch(self): # Builder jobs now commit from an isolated worktree, not this # persistent runner's own shared checkout (see # docs/intent/board-builder-worktree-parallelism.md) -- a dirty or # diverged checkout must fail the deploy loudly instead of # building whatever happens to already be on disk there. calls = [] def fake_run(command, **kwargs): calls.append(command) if command[:3] == ["git", "-C", "/home/drone/nixos-config"] and "checkout" in command: return subprocess.CompletedProcess( command, 1, "", "error: Your local changes would be overwritten" ) return subprocess.CompletedProcess(command, 0, "ok", "") ok, output = deploy_runner.run_switch( {"host": "blunix", "commit_hash": "abc1234", "repo": "/home/drone/nixos-config"}, { "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", }, fake_run, ) self.assertFalse(ok) self.assertIn("unavailable or ambiguous", output) self.assertTrue(all("nixos-rebuild" not in " ".join(call) for call in calls)) def test_deploy_runner_fails_when_board_is_not_active_after_switch(self): def fake_run(command, **kwargs): if "rev-parse" in command: return subprocess.CompletedProcess(command, 0, "abc1234" + "0" * 33, "") if command[0] == "readlink" or command[:2] == ["nix", "build"]: return subprocess.CompletedProcess(command, 0, "/nix/store/test-system", "") if "is-active" in command: return subprocess.CompletedProcess(command, 3, "inactive\n", "") return subprocess.CompletedProcess(command, 0, "switch complete", "") ok, output = deploy_runner.run_switch( { "job_id": "board-x-2", "host": "blunix", "commit_hash": "abc1234", "repo": "/home/drone/nixos-config", }, { "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", }, fake_run, ) self.assertFalse(ok) self.assertIn("inactive", output) def test_apply_commit_reports_deploy_failure_but_keeps_the_commit(self): # No pre-commit verification anymore (2026-09-06 simplification) -- # a deploy failure is reported alongside the already-pushed commit, # never rolled back. Recovery is git revert / a previous system # generation, not this script. with tempfile.TemporaryDirectory() as has_flake: self._init_repo(has_flake) (Path(has_flake) / "flake.nix").write_text("{}\n") subprocess.run(["git", "-C", has_flake, "add", "-A"], check=True) subprocess.run( ["git", "-C", has_flake, "commit", "-q", "-m", "add flake"], check=True ) (Path(has_flake) / "flake.nix").write_text("{ changed = true; }\n") environment = {"AGENT_BUILDER_WRITABLE_DIRS": has_flake} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( builder, "commit_and_push", return_value=(True, "abc123", "dev") ) as commit, ): result = builder.apply_commit("test commit") commit.assert_called_once() self.assertIn("Deploy wird", result) self.assertIn("abc123", result) def test_deploy_runner_uses_sudo_and_nixos_rebuild_bins(self): environment = { "AGENT_BUILDER_SUDO_BIN": "/run/wrappers/bin/sudo", "AGENT_BUILDER_ENV_BIN": "/nix/store/x-coreutils/bin/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/nix/store/x-nixos-rebuild/bin/nixos-rebuild", "AGENT_BUILDER_DEPLOY_CHECK_HOST": "blunix", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(deploy_runner.subprocess, "run") as run, ): run.side_effect = [ subprocess.CompletedProcess(args=[], returncode=0, stdout="abc1234" + "0" * 33, stderr=""), subprocess.CompletedProcess(args=[], returncode=0, stdout="/nix/store/test-system", stderr=""), subprocess.CompletedProcess( args=[], returncode=0, stdout="ok", stderr="" ), subprocess.CompletedProcess( args=[], returncode=0, stdout="active\n", stderr="" ), subprocess.CompletedProcess(args=[], returncode=0, stdout="/nix/store/test-system", stderr=""), ] environment["AGENT_BUILDER_DEPLOY_REPO"] = "/repo/nixos-config" ok, output = deploy_runner.run_switch( {"host": "blunix", "commit_hash": "abc1234", "repo": "/repo/nixos-config"}, environment ) command = run.call_args_list[2].args[0] self.assertEqual( command, [ "/run/wrappers/bin/sudo", "/nix/store/x-coreutils/bin/env", "HOME=/tmp", "/nix/store/x-nixos-rebuild/bin/nixos-rebuild", "switch", "--flake", "git+file:///repo/nixos-config?rev=" + "abc1234" + "0" * 33 + "#blunix", ], ) self.assertTrue(ok) self.assertEqual(output, "ok\nactive") def test_codex_builder_uses_luna_none_and_workspace_write(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_CODEX_MODEL": "gpt-5.6-luna", "AGENT_BUILDER_CODEX_REASONING_EFFORT": "none", } fake_process = FakeCodexProcess( [ '{"type":"thread.started","thread_id":"session-1"}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n', '{"type":"turn.completed","usage":{"input_tokens":120,' '"cached_input_tokens":80,"output_tokens":12}}\n', ] ) payload = {"body": "Test"} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): response, limit_reached = builder.run_codex(payload) command = popen.call_args.args[0] self.assertEqual(command[command.index("--sandbox") + 1], "workspace-write") self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-luna") reasoning = next(value for value in command if value.startswith("model_reasoning_effort=")) self.assertEqual(reasoning, 'model_reasoning_effort="none"') self.assertEqual(response, "Testantwort") self.assertFalse(limit_reached) self.assertEqual(payload["_telemetry"]["session_id"], "session-1") self.assertEqual( payload["_telemetry"]["actual_profile"], { "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "none", "iteration_limit": 20, }, ) self.assertEqual(payload["_telemetry"]["usage"]["total_tokens"], 132) def test_run_codex_role_dir_override_takes_precedence_over_agent_role_dir(self): # -C must resolve inside a --add-dir target, same reasoning as # worker.py's run_task_edit() test above: once the flake repo is # worktree-isolated, AGENT_ROLE_DIR (the now-unwritable main # checkout's agents/) no longer satisfies that -- run_and_commit() # passes the worktree's own agents/ copy instead (found live # 2026-09-09, board-worktree-parallelism-smoke-test-a). environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/worktrees/job-1/nixos-config", } fake_process = FakeCodexProcess( [ '{"type":"thread.started","thread_id":"session-1"}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): builder.run_codex( {"body": "Test"}, role_dir=Path("/worktrees/job-1/nixos-config/agents") ) command = popen.call_args.args[0] self.assertEqual( command[command.index("-C") + 1], "/worktrees/job-1/nixos-config/agents" ) # The prompt text itself must describe the worktree path, not the # hardcoded main-checkout literal it used to be -- a stale literal # actively misled the model into trying the now-unwritable main # checkout first (found live 2026-09-09, # board-worktree-parallelism-smoke-test-a). prompt = command[-1] self.assertIn("/worktrees/job-1/nixos-config", prompt) self.assertNotIn("/home/drone/nixos-config", prompt) def test_codex_payload_overrides_model_reasoning_effort_and_iteration_limit(self): environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_CODEX_MODEL": "gpt-5.6-luna", "AGENT_BUILDER_CODEX_REASONING_EFFORT": "none", "AGENT_BUILDER_ITERATION_LIMIT": "20", } fake_process = FakeCodexProcess( [ '{"type":"thread.started","thread_id":"session-1"}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Erster Schritt."}}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Zweiter Schritt."}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): response, limit_reached = builder.run_codex( { "body": "Test", "model": "gpt-5.6-payload", "reasoning_effort": "high", "iteration_limit": 1, } ) self.assertTrue(limit_reached) self.assertIn("Iterationslimit (1)", response) command = popen.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-payload") reasoning = next(value for value in command if value.startswith("model_reasoning_effort=")) self.assertEqual(reasoning, 'model_reasoning_effort="high"') def test_openrouter_builder_uses_codex_harness_and_child_only_secret(self): with tempfile.TemporaryDirectory() as temporary: secret_path = Path(temporary) / "openrouter-key" secret_path.write_text("test-openrouter-key\n", encoding="utf-8") environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_OPENROUTER_MODEL": "deepseek/deepseek-v3.2", "AGENT_BUILDER_OPENROUTER_REASONING_EFFORT": "", "AGENT_BUILDER_OPENROUTER_API_KEY_FILE": str(secret_path), } fake_process = FakeCodexProcess( [ '{"type":"thread.started","thread_id":"session-1"}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Testantwort"}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=True), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): response, limited = builder.run_runtime( {"body": "Test", "model": ""}, "openrouter" ) self.assertNotIn("OPENROUTER_API_KEY", os.environ) command = popen.call_args.args[0] child_environment = popen.call_args.kwargs["env"] self.assertEqual(command[command.index("--model") + 1], "deepseek/deepseek-v3.2") self.assertIn('model_provider="openrouter"', command) self.assertEqual(child_environment["OPENROUTER_API_KEY"], "test-openrouter-key") self.assertNotIn("test-openrouter-key", " ".join(command)) self.assertEqual(response, "Testantwort") self.assertFalse(limited) def test_openrouter_builder_preserves_codex_iteration_limit_handling(self): with tempfile.TemporaryDirectory() as temporary: secret_path = Path(temporary) / "openrouter-key" secret_path.write_text("test-openrouter-key\n", encoding="utf-8") environment = { "AGENT_CODEX_BIN": "/bin/codex", "AGENT_CODEX_HOME": "/var/lib/agent-codex/codex-home", "AGENT_HOME": "/home/agent", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_OPENROUTER_MODEL": "deepseek/deepseek-v3.2", "AGENT_BUILDER_OPENROUTER_API_KEY_FILE": str(secret_path), } fake_process = FakeCodexProcess( [ '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Erster Schritt."}}\n', '{"type":"item.completed","item":{"type":"agent_message",' '"text":"Zweiter Schritt."}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=True), mock.patch.object(common.subprocess, "Popen", return_value=fake_process), ): response, limited = builder.run_openrouter( {"body": "Test", "iteration_limit": 1} ) self.assertTrue(limited) self.assertIn("Iterationslimit (1)", response) self.assertIn("Erster Schritt.", response) def test_claude_uses_settings_and_add_dir(self): environment = { "AGENT_BUILDER_CLAUDE_BIN": "/bin/claude", "AGENT_BUILDER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/builder-settings.json", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config:/repo/website", "AGENT_BUILDER_CLAUDE_MODEL": "sonnet", "AGENT_BUILDER_CLAUDE_REASONING_EFFORT": "medium", } fake_process = FakeCodexProcess( [ '{"type":"result","subtype":"success","is_error":false,' '"result":"Testantwort","session_id":"session-1"}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): response, limit_reached = builder.run_claude({"body": "Test"}) command = popen.call_args.args[0] self.assertEqual( command[command.index("--settings") + 1], "/repo/agents/claude-settings/builder-settings.json", ) add_dir_indices = [i for i, arg in enumerate(command) if arg == "--add-dir"] add_dirs = [command[i + 1] for i in add_dir_indices] self.assertEqual(add_dirs, ["/repo/nixos-config", "/repo/website"]) self.assertEqual(command[command.index("--model") + 1], "sonnet") self.assertEqual(command[command.index("--effort") + 1], "medium") self.assertEqual(command[command.index("--permission-mode") + 1], "dontAsk") self.assertEqual(popen.call_args.kwargs["cwd"], "/repo/agents") self.assertEqual(response, "Testantwort") self.assertFalse(limit_reached) self.assertIn("Test", fake_process.stdin.value) # Codex's exec argv token has no Claude equivalent -- # the fixed instructions are folded into the same stdin payload as # the job body instead, so the marker contract they describe # should show up there too. self.assertIn("===COMMIT-MESSAGE===", fake_process.stdin.value) def test_claude_payload_overrides_model_reasoning_effort_and_iteration_limit(self): environment = { "AGENT_BUILDER_CLAUDE_BIN": "/bin/claude", "AGENT_BUILDER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/builder-settings.json", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_CLAUDE_MODEL": "sonnet", "AGENT_BUILDER_CLAUDE_REASONING_EFFORT": "medium", "AGENT_BUILDER_ITERATION_LIMIT": "20", } fake_process = FakeCodexProcess( [ '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n', '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): response, limit_reached = builder.run_claude( { "body": "Test", "model": "opus", "reasoning_effort": "high", "iteration_limit": 1, } ) self.assertTrue(limit_reached) self.assertIn("Iterationslimit (1)", response) command = popen.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "opus") self.assertEqual(command[command.index("--effort") + 1], "high") def test_run_claude_never_resumes_a_session(self): environment = { "AGENT_BUILDER_CLAUDE_BIN": "/bin/claude", "AGENT_BUILDER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/builder-settings.json", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", } fake_process = FakeCodexProcess( [ '{"type":"result","subtype":"success","is_error":false,' '"result":"Antwort","session_id":"session-1"}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process) as popen, ): builder.run_claude({"body": "Weiter"}) command = popen.call_args.args[0] self.assertNotIn("--resume", command) def test_iteration_limit_terminates_process_and_builds_interim_report(self): environment = { "AGENT_BUILDER_CLAUDE_BIN": "/bin/claude", "AGENT_BUILDER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/builder-settings.json", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_ITERATION_LIMIT": "2", } fake_process = FakeCodexProcess( [ '{"type":"assistant","message":{"content":' '[{"type":"text","text":"Erster Schritt."}]}}\n', '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n', '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n', # Never reached: the process is terminated once the limit # (2) is exceeded by the third tool_use block above. '{"type":"assistant","message":{"content":' '[{"type":"tool_use","name":"Bash","input":{}}]}}\n', ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process), ): response, limit_reached = builder.run_claude({"body": "Sehr langer Auftrag"}) self.assertTrue(limit_reached) self.assertTrue(fake_process.terminated) self.assertIn("Iterationslimit (2)", response) self.assertIn("Erster Schritt.", response) def test_iteration_limit_counts_multiple_tool_use_blocks_in_one_turn(self): # A single assistant turn can carry more than one parallel tool_use # block -- each one is its own unit of work and must count on its # own, not once per JSONL line (unlike Codex's item.completed, # which is already one-event-per-item). environment = { "AGENT_BUILDER_CLAUDE_BIN": "/bin/claude", "AGENT_BUILDER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/builder-settings.json", "AGENT_ROLE_DIR": "/repo/agents", "AGENT_BUILDER_WRITABLE_DIRS": "/repo/nixos-config", "AGENT_BUILDER_ITERATION_LIMIT": "1", } fake_process = FakeCodexProcess( [ '{"type":"assistant","message":{"content":[' '{"type":"tool_use","name":"Bash","input":{}},' '{"type":"tool_use","name":"Bash","input":{}}' "]}}\n", ] ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(common.subprocess, "Popen", return_value=fake_process), ): response, limit_reached = builder.run_claude({"body": "Auftrag"}) self.assertTrue(limit_reached) self.assertTrue(fake_process.terminated) self.assertIn("Iterationslimit (1)", response) def test_run_and_commit_skips_commit_handling_on_limit_reached(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as temporary, ): subprocess.run(["git", "init", "-q", repo_dir], check=True) environment = { "AGENT_BUILDER_WRITABLE_DIRS": repo_dir, "AGENT_BUILDER_LOCK_DIR": lock_dir, "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), } interim = "[Automatisch erzeugter Zwischenstandsbericht -- Iterationslimit (2) erreicht, Claude wurde beendet]" with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( builder, "run_runtime", return_value=(interim, True) ), mock.patch.object(builder, "extract_commit_message") as extract, mock.patch.object(builder, "apply_commit") as apply, ): response, limit_reached = builder.run_and_commit( "job-x", "lukas@0ff.space", "lukas@0ff.space", "Auftrag" ) extract.assert_not_called() apply.assert_not_called() self.assertEqual(response, interim) self.assertTrue(limit_reached) def test_run_and_commit_prepends_prior_context_and_saves_new_one(self): with ( tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as lock_dir, ): subprocess.run(["git", "init", "-q", repo_dir], check=True) environment = { "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), "AGENT_BUILDER_WRITABLE_DIRS": repo_dir, "AGENT_BUILDER_LOCK_DIR": lock_dir, } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.save_context(sender, "Bisheriger Stand") with mock.patch.object( builder, "run_runtime", return_value=("Neue Antwort", False) ) as run_runtime: response, limit_reached = builder.run_and_commit( "job-y", sender, sender, "Mach weiter" ) sent_body = run_runtime.call_args.args[0]["body"] self.assertIn("Bisheriger Stand", sent_body) self.assertIn("Mach weiter", sent_body) self.assertEqual(builder.load_context(sender), "Neue Antwort") self.assertEqual(response, "Neue Antwort") def test_agent_session_and_builder_context_do_not_collide(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): worker.save_session(sender, "agent-session") builder.save_context(sender, "builder-context") self.assertEqual(worker.load_session(sender), "agent-session") self.assertEqual(builder.load_context(sender), "builder-context") def test_queue_for_start_approval_records_pending_without_running_codex(self): with tempfile.TemporaryDirectory() as start_pending_dir: environment = {"AGENT_BUILDER_START_PENDING_DIR": start_pending_dir} with mock.patch.dict(os.environ, environment, clear=False): result = builder.queue_for_start_approval( "job-4", "lukas@0ff.space", "alfo@0ff.space", "Trek OIDC Client einrichten" ) record = builder.load_start_pending("job-4") self.assertIn("job-4", result) self.assertIn("/annehmen", result) self.assertIn("/ablehnen", result) self.assertEqual(record["sender"], "lukas@0ff.space") self.assertEqual(record["requested_by"], "alfo@0ff.space") self.assertEqual(record["body"], "Trek OIDC Client einrichten") self.assertEqual(record["runtime"], "codex") def test_process_job_gates_third_party_handoff_without_running_claude(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as start_pending_dir, tempfile.TemporaryDirectory() as temporary, ): job_path = Path(processing_dir) / "job-5.json" builder.write_payload( job_path, { "id": "job-5", "sender": "lukas@0ff.space", "requested_by": "alfo@0ff.space", "body": "Trek OIDC Client einrichten", }, ) environment = { "AGENT_BUILDER_START_PENDING_DIR": start_pending_dir, "AGENT_DB_PATH": str(Path(temporary) / "agent.db"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "run_claude") as run_claude, mock.patch.object(builder, "send_reply") as send_reply, ): builder.process_job(job_path, Path(failed_dir)) record = builder.load_start_pending("job-5") run_claude.assert_not_called() send_reply.assert_called_once() sent_to, sent_body = send_reply.call_args[0] self.assertEqual(sent_to, "lukas@0ff.space") self.assertIn("job-5", sent_body) self.assertIn("/annehmen", sent_body) self.assertFalse(job_path.exists()) self.assertEqual(record["requested_by"], "alfo@0ff.space") def test_extract_commit_records_parses_apply_commit_summary_lines(self): response = ( "nixos-config: committed und nach dev gepusht (Commit abc1234).\n\n" "website: committed und nach dev gepusht (Commit ffeeaa9)." ) records = builder.extract_commit_records(response) self.assertEqual( records, [ {"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}, {"repo": "website", "branch": "dev", "hash": "ffeeaa9"}, ], ) def test_completion_status_limit_reached_ignores_response_content(self): status, records = builder.completion_status("irrelevant", True) self.assertEqual(status, "limit_reached") self.assertEqual(records, []) def test_completion_status_committed_when_commit_record_present(self): response = "nixos-config: committed und nach dev gepusht (Commit abc1234)." status, records = builder.completion_status(response, False) self.assertEqual(status, "committed") self.assertEqual(records, [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}]) def test_completion_status_check_failed_when_not_committed_marker_present(self): response = "nixos-config: `nix flake check` ist fehlgeschlagen, NICHT committed." status, records = builder.completion_status(response, False) self.assertEqual(status, "check_failed") self.assertEqual(records, []) def test_completion_status_no_changes_otherwise(self): response = "Hinweis: keine Änderungen in den schreibbaren Verzeichnissen gefunden, nichts committed." status, records = builder.completion_status(response, False) self.assertEqual(status, "no_changes") self.assertEqual(records, []) def test_fixer_outcome_is_strict_and_requires_concrete_question(self): result = { "schema_version": 1, "outcome": "needs_input", "summary": "Blockiert", "commit_hash": None, "pushed_branch": None, "tests": [], "question": "Welche Zieloption soll gelten?", "required_action": "Option nennen", "review_required": False, "failure": None, } response, outcome = builder.extract_fixer_outcome( f"===FIX_RESULT===\n{json.dumps(result)}\n===END_FIX_RESULT===" ) self.assertEqual(response, "") self.assertEqual(outcome["outcome"], "needs_input") self.assertEqual(outcome["question"], "Welche Zieloption soll gelten?") self.assertNotIn("target_phase", outcome) with self.assertRaises(ValueError): builder.extract_fixer_outcome( "===FIX_RESULT===\nOUTCOME: needs_input\nQUESTION:\n" "SUMMARY: unklar\n===END_FIX_RESULT===" ) with self.assertRaises(ValueError): builder.extract_fixer_outcome(f"===FIX_RESULT===\n{json.dumps(result | {'extra': 1})}\n===END_FIX_RESULT===") def test_fixer_outcome_gets_one_format_correction(self): corrected_result = { "schema_version": 1, "outcome": "needs_input", "summary": "unklar", "commit_hash": None, "pushed_branch": None, "tests": [], "question": "Welche Datei soll geändert werden?", "required_action": "Datei nennen", "review_required": False, "failure": None, } corrected = f"===FIX_RESULT===\n{json.dumps(corrected_result)}\n===END_FIX_RESULT===" correction_run = mock.Mock(return_value=(corrected, False)) response, outcome = builder.resolve_fixer_outcome( "Antwort ohne Marker", correction_run ) correction_run.assert_called_once() self.assertEqual(response, "") self.assertEqual(outcome["outcome"], "needs_input") self.assertNotIn("target_phase", outcome) def test_repaired_fixer_accepts_exact_role_json_plus_wrapper_commit_block(self): result = { "schema_version": 1, "outcome": "repaired", "summary": "Defekt behoben", "commit_hash": "a" * 40, "pushed_branch": "dev", "tests": [{"command": "python3 -m unittest", "result": "passed"}], "question": None, "required_action": None, "review_required": True, "failure": None, } response, parsed = builder.extract_fixer_outcome( f"===FIX_RESULT===\n{json.dumps(result)}\n===END_FIX_RESULT===\n" "===COMMIT-MESSAGE===\nfix: repair defect\n===END-COMMIT-MESSAGE===" ) self.assertEqual(parsed["outcome"], "repaired") self.assertIn("===COMMIT-MESSAGE===", response) def test_fixer_rejects_legacy_key_value_format(self): with self.assertRaises(ValueError): builder.extract_fixer_outcome( "===FIX_RESULT===\nOUTCOME: needs_input\nQUESTION: Was?\n===END_FIX_RESULT===" ) def test_invalid_fixer_format_falls_back_to_question(self): correction_run = mock.Mock(return_value=("weiterhin kaputt", False)) _response, outcome = builder.resolve_fixer_outcome( "Antwort ohne Marker", correction_run ) correction_run.assert_called_once() self.assertEqual(outcome["outcome"], "needs_input") self.assertIn("maschinenlesbare Entscheidung", outcome["question"]) self.assertNotIn("target_phase", outcome) def test_fixer_prompt_forbids_direct_deploy_and_other_hosts(self): prompt = builder.fixed_prompt("/tmp/worktree", "fixer") self.assertIn("danach separat erfolgreicher Review", prompt) self.assertIn("niemals auf tnix oder pinix", prompt) self.assertIn("kannst nicht committen, pushen, sudo oder nixos-rebuild", prompt) self.assertNotIn("TARGET_PHASE", prompt) self.assertIn("===FIX_RESULT===", prompt) def test_write_completion_record_is_noop_without_on_complete_dir(self): with tempfile.TemporaryDirectory() as result_dir: builder.write_completion_record({"id": "job-1"}, "committed", []) self.assertEqual(list(Path(result_dir).glob("*.json")), []) def test_write_completion_record_writes_status_and_commits(self): with tempfile.TemporaryDirectory() as result_dir: builder.write_completion_record( {"id": "job-1", "on_complete_dir": result_dir}, "committed", [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], ) record = json.loads((Path(result_dir) / "job-1.json").read_text(encoding="utf-8")) self.assertEqual(record["job_id"], "job-1") self.assertEqual(record["status"], "committed") self.assertEqual(record["commits"], [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}]) self.assertEqual(record["report"], "") def test_write_completion_record_hands_off_pipeline_profile_and_telemetry(self): with tempfile.TemporaryDirectory() as result_dir: telemetry = { "session_id": "session-1", "actual_profile": { "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "low", }, "usage": {"input_tokens": 100, "output_tokens": 10}, "timing": {"duration_seconds": 1.234}, "iterations": 4, } builder.write_completion_record( { "id": "job-telemetry", "on_complete_dir": result_dir, "pipeline_run_id": "pipeline-42", "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20, "_telemetry": telemetry, }, "no_changes", ) record = json.loads( (Path(result_dir) / "job-telemetry.json").read_text(encoding="utf-8") ) self.assertEqual(record["pipeline_run_id"], "pipeline-42") self.assertEqual( record["configured_profile"], { "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20, }, ) self.assertEqual(record["telemetry"], telemetry) def test_write_completion_record_stores_and_truncates_report(self): with tempfile.TemporaryDirectory() as result_dir: builder.write_completion_record( {"id": "job-2", "on_complete_dir": result_dir}, "no_changes", [], report=" " + ("x" * 5000) + " ", ) record = json.loads((Path(result_dir) / "job-2.json").read_text(encoding="utf-8")) self.assertEqual(len(record["report"]), 4000) self.assertEqual(record["report"], "x" * 4000) def test_process_job_writes_completion_record_for_board_triggered_job(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as result_dir, ): job_path = Path(processing_dir) / "board-job-1.json" owner = "lukas@0ff.space" builder.write_payload( job_path, { "id": "board-job-1", "sender": owner, "requested_by": owner, "body": "Setze Intent XY um", "on_complete_dir": result_dir, }, ) environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} committed = "nixos-config: committed und nach dev gepusht (Commit abc1234)." with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "run_and_commit", return_value=(committed, False)), mock.patch.object(builder, "load_job_state", return_value={"commit_hash": "abc1234"}), mock.patch.object(builder, "send_reply"), ): builder.process_job(job_path, Path(failed_dir)) self.assertFalse((Path(result_dir) / "board-job-1.json").exists()) def test_board_job_keeps_no_changes_when_optional_reply_fails(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as result_dir, ): job_path = Path(processing_dir) / "board-job-no-changes.json" owner = "lukas@0ff.space" builder.write_payload( job_path, { "id": "board-job-no-changes", "sender": owner, "requested_by": owner, "body": "Fuehre einen No-op-Smoke-Test aus", "on_complete_dir": result_dir, }, ) environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} no_changes = "Keine Änderungen gefunden, nichts committed." with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( builder, "run_and_commit", return_value=(no_changes, False) ), mock.patch.object( builder, "send_reply", side_effect=RuntimeError("XMPP send_reply endete mit Status 1."), ), ): builder.process_job(job_path, Path(failed_dir)) record = json.loads( (Path(result_dir) / "board-job-no-changes.json").read_text( encoding="utf-8" ) ) self.assertEqual(record["status"], "no_changes") self.assertEqual(record["commits"], []) self.assertEqual(record["report"], no_changes) self.assertFalse(job_path.exists()) self.assertEqual(list(Path(failed_dir).iterdir()), []) def test_process_job_writes_failed_completion_record_on_exception(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as temporary, tempfile.TemporaryDirectory() as result_dir, ): job_path = Path(processing_dir) / "board-job-2.json" owner = "lukas@0ff.space" builder.write_payload( job_path, { "id": "board-job-2", "sender": owner, "requested_by": owner, "body": "Setze Intent XY um", "on_complete_dir": result_dir, }, ) environment = {"AGENT_DB_PATH": str(Path(temporary) / "agent.db")} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( builder, "run_and_commit", side_effect=RuntimeError("Codex endete mit Status 1.") ), ): builder.process_job(job_path, Path(failed_dir)) record = json.loads((Path(result_dir) / "board-job-2.json").read_text(encoding="utf-8")) self.assertEqual(record["status"], "failed") self.assertEqual(record["report"], "Codex endete mit Status 1.") def test_annehmen_runs_pending_job_and_clears_record(self): with tempfile.TemporaryDirectory() as start_pending_dir: environment = {"AGENT_BUILDER_START_PENDING_DIR": start_pending_dir} with mock.patch.dict(os.environ, environment, clear=False): builder.save_start_pending( "job-6", "lukas@0ff.space", "alfo@0ff.space", "Trek OIDC Client einrichten" ) with mock.patch.object( builder, "run_and_commit", return_value=("erledigt", False) ) as run_and_commit: response = builder.handle_start_pending_command("/annehmen job-6") record = builder.load_start_pending("job-6") run_and_commit.assert_called_once_with( "job-6", "lukas@0ff.space", "alfo@0ff.space", "Trek OIDC Client einrichten", "codex" ) self.assertEqual(response, "erledigt") self.assertIsNone(record) def test_ablehnen_drops_pending_job_without_running_codex(self): with tempfile.TemporaryDirectory() as start_pending_dir: environment = {"AGENT_BUILDER_START_PENDING_DIR": start_pending_dir} with mock.patch.dict(os.environ, environment, clear=False): builder.save_start_pending( "job-7", "lukas@0ff.space", "alfo@0ff.space", "Trek OIDC Client einrichten" ) with mock.patch.object(builder, "run_and_commit") as run_and_commit: response = builder.handle_start_pending_command("/ablehnen job-7") record = builder.load_start_pending("job-7") run_and_commit.assert_not_called() self.assertIn("abgelehnt", response) self.assertIsNone(record) def test_handle_start_pending_command_reports_unknown_id(self): with tempfile.TemporaryDirectory() as start_pending_dir: environment = {"AGENT_BUILDER_START_PENDING_DIR": start_pending_dir} with mock.patch.dict(os.environ, environment, clear=False): response = builder.handle_start_pending_command("/annehmen unknown-id") self.assertIn("Kein wartender Auftrag", response) def test_requeue_interrupted_job_moves_file_back_and_notifies_owner(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as processing_dir, ): job_path = Path(processing_dir) / "job-8.json" builder.write_payload( job_path, { "id": "job-8", "sender": "lukas@0ff.space", "requested_by": "alfo@0ff.space", "body": "Trek OIDC Client einrichten", }, ) environment = {"AGENT_BUILDER_QUEUE_DIR": inbox_dir, "AGENT_BUILDER_WRITABLE_DIRS": ""} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "send_reply") as send_reply, ): builder.requeue_interrupted_job(job_path) self.assertFalse(job_path.exists()) self.assertTrue((Path(inbox_dir) / "job-8.json").exists()) send_reply.assert_called_once() sent_to, sent_body = send_reply.call_args[0] self.assertEqual(sent_to, "lukas@0ff.space") self.assertIn("job-8", sent_body) self.assertIn("unterbrochen", sent_body) def test_handle_sigterm_requeues_active_job_and_exits(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as processing_dir, ): job_path = Path(processing_dir) / "job-9.json" builder.write_payload( job_path, { "id": "job-9", "sender": "lukas@0ff.space", "requested_by": "lukas@0ff.space", "body": "Trek OIDC Client einrichten", }, ) environment = {"AGENT_BUILDER_QUEUE_DIR": inbox_dir, "AGENT_BUILDER_WRITABLE_DIRS": ""} builder._ACTIVE_JOB_PATH = job_path try: with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "send_reply"), ): with self.assertRaises(SystemExit): builder._handle_sigterm(15, None) finally: builder._ACTIVE_JOB_PATH = None self.assertFalse(job_path.exists()) self.assertTrue((Path(inbox_dir) / "job-9.json").exists()) def test_sigterm_after_saved_commit_does_not_requeue(self): with tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as state_dir: job_path = Path(processing_dir) / "job-10.json" builder.write_payload(job_path, {"id": "job-10", "sender": "lukas@0ff.space", "body": "x"}) environment = { "AGENT_BUILDER_QUEUE_DIR": inbox_dir, "AGENT_BUILDER_STATE_DIR": state_dir, } builder._ACTIVE_JOB_PATH = job_path try: with mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(builder, "send_reply"): builder.save_job_state("job-10", "committed", commit_hash="abc123") with self.assertRaises(SystemExit): builder._handle_sigterm(15, None) finally: builder._ACTIVE_JOB_PATH = None self.assertTrue(job_path.exists()) def test_completion_record_is_idempotent(self): with tempfile.TemporaryDirectory() as result_dir: payload = {"id": "job-11", "on_complete_dir": result_dir} builder.write_completion_record(payload, "deployed", []) first = (Path(result_dir) / "job-11.json").read_text() builder.write_completion_record(payload, "deploy_failed", []) self.assertEqual(first, (Path(result_dir) / "job-11.json").read_text()) def test_queue_deployment_recreates_missing_queue_file(self): with tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as queue_dir: environment = { "AGENT_BUILDER_STATE_DIR": state_dir, "AGENT_BUILDER_DEPLOY_QUEUE_DIR": queue_dir, "AGENT_BUILDER_DEPLOY_CHECK_HOST": "blunix", } with mock.patch.dict(os.environ, environment, clear=False): builder.save_job_state("job-12", "committed", commit_hash="abc123", repo="/home/drone/nixos-config") builder.queue_deployment( {"id": "job-12", "runtime": "claude"}, Path("/home/drone/nixos-config"), "abc123", ) self.assertTrue((Path(queue_dir) / "job-12.json").exists()) def test_deploy_runner_writes_one_completion_with_runtime_and_commit(self): with tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as result_dir: path = Path(queue_dir) / "job-13.json" deploy_runner.write_payload( path, { "job_id": "job-13", "host": "blunix", "repo": "/home/drone/nixos-config", "commit_hash": "abc1234", "runtime": "claude", "on_complete_dir": result_dir, "pipeline_run_id": "pipeline-13", "configured_profile": { "runtime": "claude", "model": "claude-sonnet-5", "reasoning_effort": "low", }, "telemetry": { "session_id": "session-13", "usage": {"total_tokens": 123}, }, "retry_count": 0, }, ) environment = { "AGENT_BUILDER_STATE_DIR": state_dir, "AGENT_BUILDER_DEPLOY_REPO": "/home/drone/nixos-config", "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", "AGENT_BUILDER_DEPLOY_BACKOFF_SECONDS": "0", } def successful_runner(command, **kwargs): output = ("abc1234" + "0" * 33 if "rev-parse" in command else "/nix/store/test-system" if command[0] == "readlink" or command[:2] == ["nix", "build"] else "active\n" if "is-active" in command else "ok") return subprocess.CompletedProcess(command, 0, output, "") deploy_runner.process(path, environment, successful_runner) records = list(Path(result_dir).glob("*.json")) self.assertEqual(len(records), 1) record = json.loads(records[0].read_text()) self.assertEqual(record["status"], "committed") self.assertEqual(record["deployment_status"], "deployed") self.assertEqual(record["runtime"], "claude") self.assertEqual(record["commits"][0]["hash"], "abc1234") self.assertEqual(record["pipeline_run_id"], "pipeline-13") self.assertEqual(record["configured_profile"]["model"], "claude-sonnet-5") self.assertEqual(record["telemetry"]["usage"]["total_tokens"], 123) def test_deploy_runner_retries_and_persists_queue_count(self): with tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as state_dir: path = Path(queue_dir) / "job-14.json" deploy_runner.write_payload( path, { "job_id": "job-14", "host": "blunix", "repo": "/home/drone/nixos-config", "commit_hash": "abc1234", "retry_count": 0, }, ) attempts = [] def failing_runner(*args, **kwargs): attempts.append(1) return subprocess.CompletedProcess(args[0], 1, "", "failed") result = deploy_runner.process( path, { "AGENT_BUILDER_STATE_DIR": state_dir, "AGENT_BUILDER_DEPLOY_REPO": "/home/drone/nixos-config", "AGENT_BUILDER_SUDO_BIN": "/sudo", "AGENT_BUILDER_ENV_BIN": "/env", "AGENT_BUILDER_NIXOS_REBUILD_BIN": "/rebuild", "AGENT_BUILDER_DEPLOY_MAX_RETRIES": "3", "AGENT_BUILDER_DEPLOY_BACKOFF_SECONDS": "0", }, failing_runner, ) self.assertEqual(len(attempts), 3) self.assertEqual(result["state"], "deploy_failed") self.assertEqual(result["retry_count"], 3) self.assertFalse(path.exists()) if __name__ == "__main__": unittest.main()