import importlib.util import json import os import subprocess import tempfile import unittest from pathlib import Path from types import SimpleNamespace from unittest import mock HERE = Path(__file__).parent 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") 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_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_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", } 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) 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_first_reply_notice_is_only_added_once_per_sender(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_SENDER_STATE_DIR": temporary} 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_SESSION_DIR": temporary, "AGENT_SESSION_IDLE_SECONDS": "1800", } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): path = worker.session_path(sender) worker.write_payload( path, { "sender": sender, "session_id": "session-1", "updated_at": 100, }, ) session_id = worker.load_session(sender, now=lambda: 1901) self.assertIsNone(session_id) self.assertFalse(path.exists()) 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_writes_job_file(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_BUILDER_QUEUE_DIR": temporary} with mock.patch.dict(os.environ, environment, clear=False): worker.enqueue_builder_job("lukas@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["body"], "Ändere X in Y.") class BuilderTests(unittest.TestCase): def test_session_round_trip(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_BUILDER_SESSION_DIR": temporary} sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): builder.save_session(sender, "session-1") session_id = builder.load_session(sender, now=lambda: 100) self.assertEqual(session_id, "session-1") def test_expired_session_is_not_loaded(self): with tempfile.TemporaryDirectory() as temporary: environment = { "AGENT_BUILDER_SESSION_DIR": temporary, "AGENT_SESSION_IDLE_SECONDS": "1800", } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): path = builder.session_path(sender) builder.write_payload( path, { "sender": sender, "session_id": "session-1", "updated_at": 100, }, ) session_id = builder.load_session(sender, now=lambda: 1901) self.assertIsNone(session_id) self.assertFalse(path.exists()) def test_first_reply_notice_is_only_added_once_per_sender(self): with tempfile.TemporaryDirectory() as temporary: environment = {"AGENT_BUILDER_SENDER_STATE_DIR": temporary} 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("nicht deployen", 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_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_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 test_codex_uses_workspace_write_sandbox_with_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", "AGENT_BUILDER_WRITABLE_DIR": "/repo", } 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(builder.subprocess, "run", return_value=completed) as run, ): response, session_id = builder.run_codex({"body": "Test"}) command = run.call_args.args[0] sandbox_index = command.index("--sandbox") self.assertEqual(command[sandbox_index + 1], "workspace-write") add_dir_index = command.index("--add-dir") self.assertEqual(command[add_dir_index + 1], "/repo") self.assertEqual(response, "Testantwort") self.assertEqual(session_id, "session-1") def test_agent_and_builder_sessions_do_not_collide(self): with tempfile.TemporaryDirectory() as agent_dir, tempfile.TemporaryDirectory() as builder_dir: environment = { "AGENT_SESSION_DIR": agent_dir, "AGENT_BUILDER_SESSION_DIR": builder_dir, } sender = "lukas@0ff.space" with mock.patch.dict(os.environ, environment, clear=False): worker.save_session(sender, "agent-session") builder.save_session(sender, "builder-session") self.assertEqual(worker.load_session(sender), "agent-session") self.assertEqual(builder.load_session(sender), "builder-session") if __name__ == "__main__": unittest.main()