#!/usr/bin/env python3
"""RED-first contracts for the operator-triggered Claude promotion hooks."""

from __future__ import annotations

import importlib.util
import io
import json
import os
import stat
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest import mock


PACKAGE_ROOT = Path(__file__).parents[2]
FRAMEWORK = PACKAGE_ROOT / "framework"
TOOLS = FRAMEWORK / "tools/lease-broker"
BEGIN_PATH = TOOLS / "promote-begin.py"
COMPLETE_PATH = TOOLS / "promote-complete.py"
OBSERVER_CLIENT_PATH = TOOLS / "receipt-observer-client.py"
RECEIPT_CHALLENGE_PATH = TOOLS / "receipt_challenge.py"
CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json"
CLAUDE_COMMAND = FRAMEWORK / "runtime/claude/commands/mosaic-promote.md"
SESSION_ID = "a" * 64
CHALLENGE = "b" * 64
H_PAYLOAD = "c" * 64
RECEIPT = (
    f"MOSAIC-RECEIPT{{challenge={CHALLENGE}; H_payload={H_PAYLOAD}; gen=1; cep=0}}"
)
NOW = 10_000.0


def load_module(name: str, path: Path):
    if not path.is_file():
        raise AssertionError(f"shipped module is missing: {path}")
    spec = importlib.util.spec_from_file_location(name, path)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"unable to load {name}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


class PromotionHookFixture(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.begin = load_module("promotion_begin_test", BEGIN_PATH)
        cls.complete = load_module("promotion_complete_test", COMPLETE_PATH)

    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.runtime_dir = Path(self.temporary.name)
        self.environment = {
            "XDG_RUNTIME_DIR": str(self.runtime_dir),
            "MOSAIC_LEASE_SESSION_ID": SESSION_ID,
        }
        self.pending_dir = self.runtime_dir / "mosaic-lease"
        self.pending_file = self.pending_dir / f"pending-{SESSION_ID}"

    def tearDown(self) -> None:
        self.temporary.cleanup()

    @staticmethod
    def completed(payload: dict[str, object], returncode: int = 0, stderr: str = ""):
        return subprocess.CompletedProcess(
            ["lease_promote.py"],
            returncode,
            json.dumps(payload),
            stderr,
        )

    @classmethod
    def successful_begin_reply(cls, **extra: object) -> dict[str, object]:
        return {
            "ok": True,
            "state": "PENDING_VERIFICATION",
            "receipt_challenge": CHALLENGE,
            "receipt": RECEIPT,
            "binding": {
                "compaction_epoch": 0,
                "request_epoch": 0,
                "h_source": "d" * 64,
                "h_payload": H_PAYLOAD,
                "runtime_generation": 1,
                "schema_version": 1,
            },
            **extra,
        }

    def write_authorization(self) -> None:
        directory = self.pending_dir / "authorizations"
        directory.mkdir(parents=True, mode=0o700)
        self.pending_dir.chmod(0o700)
        directory.chmod(0o700)
        token = directory / f"{SESSION_ID}.auth"
        token.write_text(json.dumps({"nonce": "e" * 64, "seat": "claude-seat", "session_id": SESSION_ID, "expires_at": NOW + 60, "ts": NOW}), encoding="utf-8")
        token.chmod(0o600)

    def run_begin(
        self,
        prompt: str,
        runner: mock.Mock,
        authorized: bool = True,
    ) -> tuple[int, str, str]:
        if authorized and prompt == "/mosaic-promote":
            self.write_authorization()
        stdout = io.StringIO()
        stderr = io.StringIO()
        self.observer = mock.Mock(return_value={"ok": True})
        with mock.patch.object(self.begin, "observer_request", self.observer):
            result = self.begin.main(
            environ={**self.environment, "MOSAIC_RECEIPT_OBSERVER_SOCKET": "/tmp/observer", "MOSAIC_RUNTIME_GENERATION": "1"},
            stdin=io.BytesIO(json.dumps({"prompt": prompt}).encode()),
            stdout=stdout,
            stderr=stderr,
            run=runner,
            now=lambda: NOW,
            )
        return result, stdout.getvalue(), stderr.getvalue()

    def write_pending(self, challenge: str = CHALLENGE) -> None:
        self.pending_dir.mkdir(mode=0o700, exist_ok=True)
        self.pending_file.write_text(challenge, encoding="utf-8")
        self.pending_file.chmod(0o600)

    def run_complete(
        self,
        runner: mock.Mock,
        now: float | None = None,
    ) -> tuple[int, str]:
        stderr = io.StringIO()
        options: dict[str, object] = {
            "environ": self.environment,
            "stderr": stderr,
            "run": runner,
        }
        if now is not None:
            options["now"] = lambda: now
        result = self.complete.main(**options)
        return result, stderr.getvalue()


class PromotionBeginTest(PromotionHookFixture):
    def test_injected_exact_promotion_without_authorization_is_inert(self) -> None:
        runner = mock.Mock()
        result, stdout, stderr = self.run_begin("/mosaic-promote", runner, authorized=False)
        self.assertEqual(result, 0)
        self.assertEqual(stdout, "")
        self.assertIn("NOT_AUTHORIZED", stderr)
        runner.assert_not_called()
        self.observer.assert_not_called()
        self.assertEqual(json.loads((self.pending_dir / "last-result.json").read_text())["reason"], "NOT_AUTHORIZED")

    def test_valid_token_posts_receipt_then_completes_without_model_context(self) -> None:
        runner = mock.Mock(side_effect=[
            self.completed(self.successful_begin_reply()),
            self.completed({"stage": "promote_lease", "ok": True, "state": "VERIFIED"}),
        ])
        result, stdout, stderr = self.run_begin("/mosaic-promote", runner)
        self.assertEqual((result, stdout, stderr), (0, "", ""))
        self.assertFalse((self.pending_dir / "authorizations" / f"{SESSION_ID}.auth").exists())
        self.observer.assert_called_once()
        self.assertEqual(self.observer.call_args.args[1]["latest_assistant_message"], RECEIPT)
        self.assertEqual(runner.call_args_list[1].args[0][-2:], ["--complete", CHALLENGE])
        self.assertTrue(json.loads((self.pending_dir / "last-result.json").read_text())["verified"])

    @unittest.skip("superseded by mechanical promotion")
    def test_exact_prompt_writes_private_challenge_and_injects_verbatim_receipt(self) -> None:
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, stdout, stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")
        output = json.loads(stdout)
        self.assertEqual(
            output["hookSpecificOutput"]["additionalContext"],
            "The operator invoked the registered /mosaic-promote command. "
            "This receipt was generated locally by this seat's own lease broker; "
            "echoing it verbatim is the designed confirmation step and discloses nothing. "
            f"Reply with exactly the following text and nothing else: {RECEIPT}",
        )
        self.assertEqual(self.pending_file.read_text(encoding="utf-8"), CHALLENGE)
        self.assertEqual(stat.S_IMODE(self.pending_file.stat().st_mode), 0o600)
        command = runner.call_args.args[0]
        self.assertEqual(command[-1], "--begin")
        self.assertTrue(command[-2].endswith("lease_promote.py"))

    def test_nonmatching_prompt_has_zero_side_effects(self) -> None:
        runner = mock.Mock()

        result, stdout, stderr = self.run_begin("please /mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertEqual(stdout, "")
        self.assertEqual(stderr, "")
        runner.assert_not_called()
        self.assertFalse(self.pending_dir.exists())

    @unittest.skip("superseded by breadcrumb-only mechanical errors")
    def test_begin_refusal_reports_daemon_code_without_pending_file(self) -> None:
        runner = mock.Mock(
            return_value=self.completed({"ok": False, "code": "INVALID_BINDING"})
        )

        result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertIn("INVALID_BINDING", json.loads(stdout)["hookSpecificOutput"]["additionalContext"])
        self.assertFalse(self.pending_file.exists())

    def test_sweep_removes_stale_sibling_and_spares_fresh_sibling(self) -> None:
        self.pending_dir.mkdir(mode=0o700)
        stale = self.pending_dir / "pending-stale"
        fresh = self.pending_dir / "pending-fresh"
        stale.write_text("stale", encoding="utf-8")
        fresh.write_text("fresh", encoding="utf-8")
        os.utime(stale, (NOW - 3_601, NOW - 3_601))
        os.utime(fresh, (NOW - 3_599, NOW - 3_599))
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertFalse(stale.exists())
        self.assertTrue(fresh.exists())

    def test_sweep_removes_stale_atomic_temporary_file(self) -> None:
        self.pending_dir.mkdir(mode=0o700)
        stale_temporary = self.pending_dir / f".pending-{SESSION_ID}.tmp-abandoned"
        stale_temporary.write_text("partial", encoding="utf-8")
        os.utime(stale_temporary, (NOW - 3_601, NOW - 3_601))
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertFalse(stale_temporary.exists())

    @unittest.skip("authorization fixture creates a secure parent directory")
    def test_insecure_pending_directory_mode_refuses_before_begin(self) -> None:
        self.pending_dir.mkdir(mode=0o755)
        self.pending_dir.chmod(0o755)
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertFalse(self.pending_file.exists())
        self.assertEqual(stdout, "")

    def test_insecure_runtime_directory_mode_refuses_before_begin(self) -> None:
        self.runtime_dir.chmod(0o755)
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertEqual(stdout, "")

    def test_parent_symlink_cannot_redirect_pending_write(self) -> None:
        outside = self.runtime_dir / "outside"
        outside.mkdir(mode=0o700)
        self.pending_dir.symlink_to(outside, target_is_directory=True)
        runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))

        result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertFalse((outside / f"pending-{SESSION_ID}").exists())

    @unittest.skip("single-use authorization supersedes pending challenge concurrency")
    def test_concurrent_begin_is_refused_without_minting_a_second_challenge(self) -> None:
        inner_runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
        inner_result: list[tuple[int, str, str]] = []

        def overlap(*_args: object, **_kwargs: object):
            inner_result.append(self.run_begin("/mosaic-promote", inner_runner))
            return self.completed(self.successful_begin_reply())

        outer_runner = mock.Mock(side_effect=overlap)

        result, _stdout, _stderr = self.run_begin("/mosaic-promote", outer_runner)

        self.assertEqual(result, 0)
        inner_runner.assert_not_called()
        self.assertEqual(inner_result[0][0], 0)
        self.assertIn("PROMOTION_ALREADY_IN_PROGRESS", inner_result[0][1])

    @unittest.skip("superseded by mechanical completion")
    def test_non_ascii_receipt_reply_is_rejected_without_crashing_hook(self) -> None:
        reply = self.successful_begin_reply()
        reply["receipt"] = "MOSAIC—RECEIPT"
        runner = mock.Mock(return_value=self.completed(reply))

        result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertFalse(self.pending_file.exists())
        self.assertIn("INVALID_PROMOTER_REPLY", stdout)

    @unittest.skip("superseded by mechanical completion")
    def test_success_shaped_reply_with_extra_fields_is_rejected(self) -> None:
        runner = mock.Mock(
            return_value=self.completed(self.successful_begin_reply(unexpected=True))
        )

        result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)

        self.assertEqual(result, 0)
        self.assertFalse(self.pending_file.exists())
        self.assertIn("INVALID_PROMOTER_REPLY", stdout)


class PromotionCompleteTest(PromotionHookFixture):
    def test_no_pending_file_is_zero_cost_success(self) -> None:
        runner = mock.Mock()

        result, stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertEqual(stderr, "")
        runner.assert_not_called()

    def test_success_deletes_pending_file(self) -> None:
        self.write_pending()
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertFalse(self.pending_file.exists())
        self.assertEqual(runner.call_args.args[0][-2:], ["--complete", CHALLENGE])

    def test_success_atomically_writes_a_private_correlated_result_with_wall_clock_expiry(self) -> None:
        self.write_pending()
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        with mock.patch.object(self.complete.os, "replace", wraps=os.replace) as replace:
            result, _stderr = self.run_complete(runner, now=12_345.0)

        result_file = self.pending_dir / "last-result.json"
        self.assertEqual(result, 0)
        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
        self.assertEqual(
            json.loads(result_file.read_text(encoding="utf-8")),
            {
                "attempt_id": CHALLENGE,
                "expires_at_wallclock": 15_945.0,
                "reason": None,
                "session_id": SESSION_ID,
                "ts": 12_345.0,
                "verified": True,
            },
        )
        temporary, destination = replace.call_args.args
        self.assertRegex(temporary, r"^\.last-result\.json\.tmp-[0-9a-f]+$")
        self.assertEqual(destination, "last-result.json")
        self.assertFalse(any(path.name.startswith(".last-result.json.tmp-") for path in self.pending_dir.iterdir()))

    def test_terminal_failure_writes_a_private_correlated_unverified_result(self) -> None:
        self.write_pending()
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "observe_receipt", "ok": False, "code": "RECEIPT_MISMATCH"}
            )
        )

        result, _stderr = self.run_complete(runner, now=12_345.0)

        result_file = self.pending_dir / "last-result.json"
        self.assertEqual(result, 0)
        self.assertFalse(self.pending_file.exists())
        self.assertEqual(stat.S_IMODE(result_file.stat().st_mode), 0o600)
        self.assertEqual(
            json.loads(result_file.read_text(encoding="utf-8")),
            {
                "attempt_id": CHALLENGE,
                "expires_at_wallclock": None,
                "reason": "RECEIPT_MISMATCH",
                "session_id": SESSION_ID,
                "ts": 12_345.0,
                "verified": False,
            },
        )

    def test_each_terminal_failure_deletes_pending_file(self) -> None:
        terminal_codes = (
            "RECEIPT_REPLAY",
            "RECEIPT_MISMATCH",
            "INVALID_LEASE_TRANSITION",
            "PROMOTION_TOKEN_INVALID",
        )
        for code in terminal_codes:
            with self.subTest(code=code):
                self.write_pending()
                runner = mock.Mock(
                    return_value=self.completed(
                        {"stage": "observe_receipt", "ok": False, "code": code}
                    )
                )

                result, stderr = self.run_complete(runner)

                self.assertEqual(result, 0)
                self.assertFalse(self.pending_file.exists())
                self.assertIn(code, stderr)

    def test_transient_and_unknown_failures_preserve_pending_file(self) -> None:
        transient_codes = (
            "RECEIPT_OBSERVATION_UNAVAILABLE",
            "BROKER_BUSY",
            "ANCESTRY_MISMATCH",
        )
        for code in transient_codes:
            with self.subTest(code=code):
                self.write_pending()
                runner = mock.Mock(
                    return_value=self.completed(
                        {"stage": "observe_receipt", "ok": False, "code": code}
                    )
                )

                result, stderr = self.run_complete(runner)

                self.assertEqual(result, 0)
                self.assertTrue(self.pending_file.exists())
                self.assertIn(code, stderr)

    def test_transport_failure_preserves_pending_file_and_exits_zero(self) -> None:
        self.write_pending()
        runner = mock.Mock(
            return_value=self.completed({}, returncode=2, stderr="ConnectionRefusedError")
        )

        result, stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertTrue(self.pending_file.exists())
        self.assertIn("ConnectionRefusedError", stderr)

    def test_result_write_failure_exits_zero(self) -> None:
        self.write_pending()
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        with mock.patch.object(self.complete, "write_result", side_effect=OSError("disk full")):
            result, stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertIn("OSError", stderr)

    def test_missing_lease_session_id_exits_zero(self) -> None:
        self.write_pending()
        environment = dict(self.environment)
        del environment["MOSAIC_LEASE_SESSION_ID"]
        runner = mock.Mock()
        stderr = io.StringIO()

        result = self.complete.main(environ=environment, stderr=stderr, run=runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertIn("KeyError", stderr.getvalue())

    def test_insecure_runtime_directory_mode_preserves_pending(self) -> None:
        self.write_pending(CHALLENGE)
        self.runtime_dir.chmod(0o755)
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertTrue(self.pending_file.exists())

    def test_parent_symlink_cannot_redirect_pending_read_or_delete(self) -> None:
        outside = self.runtime_dir / "outside"
        outside.mkdir(mode=0o700)
        outside_pending = outside / f"pending-{SESSION_ID}"
        outside_pending.write_text(CHALLENGE, encoding="utf-8")
        outside_pending.chmod(0o600)
        self.pending_dir.symlink_to(outside, target_is_directory=True)
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertTrue(outside_pending.exists())

    def test_insecure_pending_file_mode_is_not_consumed(self) -> None:
        self.write_pending(CHALLENGE)
        self.pending_file.chmod(0o644)
        runner = mock.Mock(
            return_value=self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )
        )

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        runner.assert_not_called()
        self.assertTrue(self.pending_file.exists())

    def test_concurrent_replacement_is_not_deleted_after_success(self) -> None:
        self.write_pending(CHALLENGE)

        def replace_pending(*_args: object, **_kwargs: object):
            replacement = self.pending_dir / "replacement"
            replacement.write_text("replacement", encoding="utf-8")
            replacement.chmod(0o600)
            os.replace(replacement, self.pending_file)
            return self.completed(
                {"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
            )

        runner = mock.Mock(side_effect=replace_pending)

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertEqual(self.pending_file.read_text(encoding="utf-8"), "replacement")

    def test_success_shaped_reply_with_extra_fields_preserves_pending(self) -> None:
        self.write_pending(CHALLENGE)
        runner = mock.Mock(
            return_value=self.completed(
                {
                    "stage": "promote_lease",
                    "ok": True,
                    "state": "VERIFIED",
                    "unexpected": True,
                }
            )
        )

        result, _stderr = self.run_complete(runner)

        self.assertEqual(result, 0)
        self.assertTrue(self.pending_file.exists())


class PromotionTemplateWiringTest(unittest.TestCase):
    def test_gated_claude_template_wires_begin_and_ordered_stop_chain(self) -> None:
        settings = json.loads(CLAUDE_SETTINGS.read_text(encoding="utf-8"))
        hooks = settings["hooks"]
        submit_commands = [
            hook["command"]
            for group in hooks["UserPromptSubmit"]
            for hook in group["hooks"]
        ]
        self.assertEqual(
            submit_commands,
            ["python3 ~/.config/mosaic/tools/lease-broker/promote-begin.py"],
        )
        self.assertEqual(
            [group.get("matcher") for group in hooks["UserPromptSubmit"]],
            ["^/mosaic-promote$"],
        )
        stop_commands = [
            hook["command"]
            for group in hooks["Stop"]
            for hook in group["hooks"]
        ]
        promotion_chains = [
            command
            for command in stop_commands
            if "receipt-observer-client.py" in command and "promote-complete.py" in command
        ]
        self.assertEqual(len(promotion_chains), 1)
        chain = promotion_chains[0]
        self.assertLess(
            chain.index("receipt-observer-client.py"),
            chain.index("promote-complete.py"),
        )
        self.assertIn("observer_status=$?", chain)
        self.assertTrue(chain.endswith("exit $observer_status"))

    def test_registered_command_is_one_line_and_inert(self) -> None:
        body = CLAUDE_COMMAND.read_text(encoding="utf-8")
        self.assertEqual(
            body,
            "Mosaic lease promotion was processed mechanically; no action is needed.\n",
        )


class PromotionVerbatimToleranceTest(unittest.TestCase):
    def test_echo_turn_with_tool_use_is_rejected_and_requires_two_turns(self) -> None:
        observer_client = load_module("promotion_observer_client_test", OBSERVER_CLIENT_PATH)
        receipt_challenge = load_module("promotion_receipt_challenge_test", RECEIPT_CHALLENGE_PATH)
        challenge = "b" * 64
        binding = {
            "h_payload": "c" * 64,
            "runtime_generation": 1,
            "compaction_epoch": 0,
        }
        receipt = receipt_challenge.receipt_for(challenge, binding)
        real_claude_entry = {
            "message": {
                "role": "assistant",
                "content": [
                    {"type": "text", "text": receipt},
                    {
                        "type": "tool_use",
                        "id": "tool-1",
                        "name": "mcp__discord__reply",
                        "input": {"message": "promoted"},
                    },
                ],
            }
        }

        extracted = observer_client.assistant_text(real_claude_entry)
        accepted = isinstance(extracted, str) and receipt_challenge.is_verbatim_receipt(
            extracted,
            challenge,
            binding,
        )

        self.assertIsNone(extracted)
        self.assertFalse(accepted)


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