#!/usr/bin/env python3
"""Exercise the native Codex package at its JSON/process boundaries."""

import importlib.util
import io
import json
import os
import re
import select
import shutil
import signal
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from types import ModuleType
from typing import IO, cast
from unittest.mock import patch

ROOT = Path(__file__).resolve().parents[2]
ADAPTER = ROOT / "hooks/codex-session-start.py"
EVENT = {"hook_event_name": "SessionStart", "cwd": "/fixture/repository"}


class CodexSessionStart(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="codex-policy-test-")
        self.addCleanup(self.temp.cleanup)
        self.root = (Path(self.temp.name) / "plugin with spaces").resolve()
        for directory in ("hooks", "rules", ".tessl-plugin"):
            (self.root / directory).mkdir(parents=True)
        shutil.copy2(ADAPTER, self.root / "hooks/codex-session-start.py")
        self.manifest = {
            "version": "1.2.3",
            "rules": ["rules/always.md", "rules/scoped.md"],
            "hooks": {
                "SessionStart": [
                    {
                        "hooks": [
                            {
                                "args": ["${TESSL_PLUGIN_DIR}/hooks/session-start.sh"],
                            }
                        ]
                    }
                ]
            },
        }
        self.write_manifest()
        (self.root / "hooks/session-start-hooks.txt").write_text("check.sh\n")
        (self.root / "rules/always.md").write_text(
            "---\nalwaysApply: true\n---\n# Always\n"
        )
        (self.root / "rules/scoped.md").write_text(
            '---\nalwaysApply: false\napplyTo: "hooks/** — editing hooks"\n---\n# Scoped\n'
        )

    def write_manifest(self):
        (self.root / ".tessl-plugin/plugin.json").write_text(json.dumps(self.manifest))

    def run_hook(
        self, *args: str, event: object = EVENT
    ) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            [sys.executable, str(self.root / "hooks/codex-session-start.py"), *args],
            input=json.dumps(event),
            capture_output=True,
            text=True,
            env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
        )

    def test_context_names_installed_version_paths_and_scopes(self):
        result = self.run_hook()
        self.assertEqual(result.returncode, 0, result.stderr)
        output = json.loads(result.stdout)["hookSpecificOutput"]
        self.assertEqual(output["hookEventName"], "SessionStart")
        text = output["additionalContext"]
        self.assertIn("gamussa/coding-policy 1.2.3", text)
        self.assertIn(f"{self.root}/rules/always.md (always-on)", text)
        self.assertIn(
            f"{self.root}/rules/scoped.md (conditional; read frontmatter scope)", text
        )
        self.assertIn("use this installed plugin root instead", text)

    def test_other_events_are_silent(self):
        result = self.run_hook(event={"hook_event_name": "Stop"})
        self.assertEqual((result.returncode, result.stdout, result.stderr), (0, "", ""))

    def test_invalid_input_is_reported(self):
        result = self.run_hook(event=[])
        self.assertEqual(result.returncode, 1)
        self.assertIn("JSON object", result.stderr)

    def test_missing_rule_is_reported(self):
        (self.root / "rules/always.md").unlink()
        result = self.run_hook()
        self.assertEqual(result.returncode, 1)
        self.assertIn("inspect the plugin installation", result.stderr)
        self.assertEqual(result.stdout, "")

    def test_invalid_scope_is_reported(self):
        (self.root / "rules/scoped.md").write_text("---\nalwaysApply: false\n---\n")
        result = self.run_hook()
        self.assertEqual(result.returncode, 1)
        self.assertIn("applyTo", result.stderr)

    def test_rule_path_cannot_escape_plugin(self):
        self.manifest["rules"] = ["../outside.md"]
        self.write_manifest()
        result = self.run_hook()
        self.assertEqual(result.returncode, 1)
        self.assertIn("escapes the plugin root", result.stderr)

    def test_undeclared_hook_is_not_executed(self):
        result = self.run_hook("../check.sh")
        self.assertEqual(result.returncode, 1)
        self.assertIn("session-start-hooks.txt", result.stderr)

    def test_manifest_entry_outside_the_list_is_not_executed(self):
        (self.root / "hooks/session-start.sh").write_text(
            "touch \"$TESSL_PLUGIN_DIR/ran\"\n"
        )
        result = self.run_hook("session-start.sh")
        self.assertEqual(result.returncode, 1)
        self.assertIn("session-start-hooks.txt", result.stderr)
        self.assertFalse((self.root / "ran").exists())

    def test_tessl_context_and_stdin_are_preserved(self):
        (self.root / "hooks/check.sh").write_text(
            "set -euo pipefail\n"
            "python3 -c 'import json, os, sys; event = json.load(sys.stdin); "
            'print(json.dumps({"additionalContext": event["cwd"] + "|" + os.environ["TESSL_PLUGIN_DIR"]}))\'\n'
        )
        result = self.run_hook("check.sh")
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(
            json.loads(result.stdout)["hookSpecificOutput"],
            {
                "hookEventName": "SessionStart",
                "additionalContext": f"/fixture/repository|{self.root}",
            },
        )

    def test_silent_child_stays_silent(self):
        (self.root / "hooks/check.sh").write_text("exit 0\n")
        result = self.run_hook("check.sh")
        self.assertEqual((result.returncode, result.stdout), (0, ""))

    def test_child_failure_is_preserved(self):
        (self.root / "hooks/check.sh").write_text(
            "echo 'fixture failure' >&2\nexit 7\n"
        )
        result = self.run_hook("check.sh")
        self.assertEqual(result.returncode, 7)
        self.assertIn("fixture failure", result.stderr)
        self.assertEqual(result.stdout, "")

    def test_malformed_child_output_is_reported(self):
        for output in ("not JSON", "[]", '{"additionalContext":false}'):
            with self.subTest(output=output):
                (self.root / "hooks/check.sh").write_text(
                    f"printf '%s\\n' '{output}'\n"
                )
                result = self.run_hook("check.sh")
                self.assertEqual(result.returncode, 1)
                self.assertIn("inspect the plugin installation", result.stderr)

    def load_adapter(self) -> ModuleType:
        # Load the installed copy so its ROOT is the fixture plugin root.
        spec = importlib.util.spec_from_file_location(
            "codex_session_start", self.root / "hooks/codex-session-start.py"
        )
        assert spec is not None and spec.loader is not None
        adapter = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(adapter)
        return adapter

    def test_child_timeout_is_reported(self):
        (self.root / "hooks/check.sh").write_text("set -euo pipefail\nsleep 30\n")
        adapter = self.load_adapter()
        communicate = subprocess.Popen[str].communicate
        calls: list[float | None] = []

        def time_out(
            process: subprocess.Popen[str],
            input: str | None = None,
            timeout: float | None = None,
        ) -> tuple[str, str]:
            calls.append(timeout)
            if len(calls) == 1:
                raise subprocess.TimeoutExpired(process.args, timeout or 0)
            # Bound cleanup in the test, so a regression fails instead of hanging.
            return communicate(process, input=input, timeout=5)

        errors = io.StringIO()
        with (
            patch.object(sys, "argv", ["hook", "check.sh"]),
            patch.object(sys, "stdin", io.StringIO(json.dumps(EVENT))),
            patch.object(subprocess.Popen, "communicate", time_out),
            redirect_stderr(errors),
        ):
            self.assertEqual(adapter.main(), 1)
        self.assertIn("timed out", errors.getvalue())

    def test_timeout_survives_eperm_from_an_unsignalable_group(self):
        # macOS answers EPERM to killpg while the group's members cannot take a
        # signal yet. The first group signal is refused that way; the adapter
        # still reports the timeout and leaves no member of the group behind.
        (self.root / "hooks/check.sh").write_text("set -euo pipefail\nsleep 30\n")
        adapter = self.load_adapter()
        killpg = os.killpg
        groups: list[int] = []

        def refuse_once(pgid: int, sig: int) -> None:
            groups.append(pgid)
            if len(groups) == 1:
                raise PermissionError(1, "Operation not permitted")
            killpg(pgid, sig)

        communicate = subprocess.Popen[str].communicate
        calls: list[float | None] = []

        def time_out(
            process: subprocess.Popen[str],
            input: str | None = None,
            timeout: float | None = None,
        ) -> tuple[str, str]:
            calls.append(timeout)
            if len(calls) == 1:
                raise subprocess.TimeoutExpired(process.args, timeout or 0)
            return communicate(process, input=input, timeout=5)

        errors = io.StringIO()
        with (
            patch.object(sys, "argv", ["hook", "check.sh"]),
            patch.object(sys, "stdin", io.StringIO(json.dumps(EVENT))),
            patch.object(subprocess.Popen, "communicate", time_out),
            patch.object(os, "killpg", refuse_once),
            redirect_stderr(errors),
        ):
            self.assertEqual(adapter.main(), 1)
        self.assertIn("timed out", errors.getvalue())
        self.assertNotIn("Operation not permitted", errors.getvalue())
        # No member left that still runs: zombies awaiting a reaper answer
        # kill(0) on Linux and EPERM on macOS, so ask ps for their state.
        table = subprocess.run(
            ["ps", "-A", "-o", "pgid=,stat="],
            capture_output=True, text=True, check=True,
        ).stdout
        running = [
            row for row in table.splitlines()
            if row.split()[0] == str(groups[0]) and not row.split()[1].startswith("Z")
        ]
        self.assertEqual(running, [])

    def test_timeout_reaps_a_nested_deadline_runners_group(self):
        # A hook whose own bounded-run.sh deadline (30s) outlives the adapter's
        # (patched to 1s) holds an 8s sleeper in the runner's separate process
        # group. SIGKILL alone skips the runner's TERM trap, leaving the sleeper
        # holding stdout. The sleeper writes a marker only if nothing stopped it,
        # so the outcome is checked, never elapsed time.
        shutil.copy2(ROOT / "hooks/bounded-run.sh", self.root / "hooks/bounded-run.sh")
        sleeper_pid = self.root / "sleeper.pid"
        (self.root / "hooks/check.sh").write_text(
            "set -euo pipefail\n"
            'echo "${SESSION_START_CHILD_DEADLINE_SEC-unset}" > "$TESSL_PLUGIN_DIR/deadline"\n'
            'bash "$TESSL_PLUGIN_DIR/hooks/bounded-run.sh" 30 '
            "bash -c 'echo $$ > \"$1\"; sleep 8; touch \"$2\"' _ "
            "\"$TESSL_PLUGIN_DIR/sleeper.pid\" \"$TESSL_PLUGIN_DIR/sleeper.done\"\n"
        )
        adapter = self.load_adapter()
        errors = io.StringIO()
        with (
            patch.object(adapter, "HOOK_TIMEOUT_SECONDS", 1),
            patch.object(sys, "argv", ["hook", "check.sh"]),
            patch.object(sys, "stdin", io.StringIO(json.dumps(EVENT))),
            redirect_stderr(errors),
        ):
            self.assertEqual(adapter.main(), 1)
        pid = int(sleeper_pid.read_text().strip())
        try:
            os.kill(pid, 0)
            alive = True
        except (ProcessLookupError, PermissionError):
            # Gone, or a zombie awaiting its reaper (macOS answers EPERM).
            alive = False
        if alive:
            os.kill(pid, signal.SIGKILL)
        self.assertFalse(
            (self.root / "sleeper.done").exists(),
            "adapter waited until the nested runner's sleeper finished",
        )
        self.assertFalse(alive, "the nested runner's sleeper survived the timeout")
        self.assertIn("timed out", errors.getvalue())
        self.assertEqual((self.root / "deadline").read_text().strip(), "1")

    def test_timeout_closes_a_real_descendants_output_pipe(self):
        # Trigger the timeout only after the descendant reports readiness. Its
        # pipe reaches EOF only when both Bash and the descendant have exited.
        (self.root / "hooks/check.sh").write_text(
            "set -euo pipefail\n"
            "python3 -c 'import os, signal; print(os.getpid(), flush=True); signal.pause()' &\n"
            "wait\n"
        )
        adapter = self.load_adapter()
        communicate = subprocess.Popen[str].communicate
        descendants: list[tuple[int, int]] = []
        calls: list[float | None] = []

        def timeout_when_ready(
            process: subprocess.Popen[str],
            input: str | None = None,
            timeout: float | None = None,
        ) -> tuple[str, str]:
            calls.append(timeout)
            if len(calls) == 1:
                self.assertIsNotNone(process.stdout)
                # assertIsNotNone above proved the pipe exists.
                stdout = cast(IO[str], process.stdout)
                self.assertTrue(
                    select.select([stdout], [], [], 5)[0],
                    "child did not report readiness",
                )
                pid = int(stdout.readline().strip())
                descendants.append((pid, os.dup(stdout.fileno())))
                raise subprocess.TimeoutExpired(process.args, timeout or 0)
            # Bound cleanup in the test, so a regression fails instead of hanging.
            return communicate(process, input=input, timeout=5)

        errors = io.StringIO()
        try:
            with (
                patch.object(sys, "argv", ["hook", "check.sh"]),
                patch.object(sys, "stdin", io.StringIO(json.dumps(EVENT))),
                patch.object(subprocess.Popen, "communicate", timeout_when_ready),
                redirect_stderr(errors),
            ):
                self.assertEqual(adapter.main(), 1)
            self.assertIn("timed out", errors.getvalue())
            self.assertEqual(len(descendants), 1)
            _, pipe = descendants[0]
            self.assertTrue(
                select.select([pipe], [], [], 0)[0],
                "descendant survived timeout and still owns stdout",
            )
            self.assertEqual(os.read(pipe, 1), b"")
        finally:
            for pid, pipe in descendants:
                os.close(pipe)
                try:
                    os.kill(pid, signal.SIGKILL)
                except (ProcessLookupError, PermissionError):
                    # Expected after successful process-group cleanup; macOS
                    # answers EPERM for a zombie awaiting its reaper.
                    continue


# Every source Codex emits: codex-rs/hooks/src/events/session_start.rs
# `SessionStartSource::as_str` (openai/codex rust-v0.160.0).
CODEX_SESSION_START_SOURCES = ("startup", "resume", "clear", "compact", "fork")


def codex_matcher_selects(matcher: str | None, source: str) -> bool:
    """Mirror codex-rs/hooks/src/engine/matcher.rs `HookMatcher::matches`."""
    if matcher is None or matcher in ("", "*"):
        return True
    if all(ch.isascii() and (ch.isalnum() or ch in "_|") for ch in matcher):
        return source in matcher.split("|")
    return re.search(matcher, source) is not None


class CodexPackage(unittest.TestCase):
    def test_every_session_start_source_runs_every_handler(self):
        groups = json.loads((ROOT / "hooks/hooks.json").read_text())["hooks"][
            "SessionStart"
        ]
        registered = [hook["command"] for group in groups for hook in group["hooks"]]
        for source in CODEX_SESSION_START_SOURCES:
            with self.subTest(source=source):
                selected = [
                    hook["command"]
                    for group in groups
                    if codex_matcher_selects(group.get("matcher"), source)
                    for hook in group["hooks"]
                ]
                self.assertIn(
                    'python3 "${PLUGIN_ROOT}/hooks/codex-session-start.py"', selected
                )
                self.assertEqual(selected, registered)

    def test_matcher_model_rejects_unlisted_source(self):
        self.assertFalse(codex_matcher_selects("startup|resume", "fork"))
        self.assertTrue(codex_matcher_selects("startup|fork", "fork"))
        self.assertTrue(codex_matcher_selects(None, "fork"))

    def test_manifest_version_and_skills_match_tessl(self):
        codex = json.loads((ROOT / ".codex-plugin/plugin.json").read_text())
        tessl = json.loads((ROOT / ".tessl-plugin/plugin.json").read_text())
        self.assertEqual(codex["version"].split("+", 1)[0], tessl["version"])
        self.assertEqual(codex["name"], "gamussa-coding-policy")
        discovered = {
            str(path.parent.relative_to(ROOT))
            for path in (ROOT / codex["skills"]).glob("*/SKILL.md")
        }
        self.assertEqual(discovered, set(tessl["skills"]))

    def test_hooks_cover_original_session_and_stop_handlers(self):
        codex = json.loads((ROOT / "hooks/hooks.json").read_text())["hooks"]
        tessl = json.loads((ROOT / ".tessl-plugin/plugin.json").read_text())
        session = [
            hook["command"]
            for group in codex["SessionStart"]
            for hook in group["hooks"]
        ]
        self.assertIn('python3 "${PLUGIN_ROOT}/hooks/codex-session-start.py"', session)
        originals = [
            line.strip()
            for line in (ROOT / "hooks/session-start-hooks.txt").read_text().splitlines()
            if line.strip()
        ]
        self.assertEqual(
            [command.rsplit(" ", 1)[-1] for command in session[1:]], originals
        )
        expected_stop = [
            hook["command"].replace("${TESSL_PLUGIN_DIR}", "${PLUGIN_ROOT}")
            for group in tessl["nativeHooks"]["codex"]["Stop"]
            for hook in group["hooks"]
        ]
        self.assertEqual(
            [hook["command"] for group in codex["Stop"] for hook in group["hooks"]],
            expected_stop,
        )


if __name__ == "__main__":
    unittest.main()