nixos-config/hosts/blunix/localServices/agent-codex/test_agent_codex.py

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()