#!/usr/bin/env python3
"""Red-first tests for issue #869 Point-1 C4 — the enforcement/activation
version-coupling assertion at the `launch-runtime.py` seam.

Root cause under test (#828 restated): the lease broker's ENFORCEMENT half
(this toolkit) and its ACTIVATION half (`execLeaseGatedRuntime()` in
`launch.ts`, chained through `launch-runtime.py`) shipped on different
channels and drifted. C1 (`lease-activation-probe.ts`) gave the activation
half a versioned, machine-checkable identity
(`LEASE_ACTIVATION_CAPABILITY`, printed via the hidden CLI subcommand
`mosaic __lease-capability`). C4 (this module + `activation_version_gate.py`)
is the assertion that actually USES that identity: enforcement must refuse
to proceed — loudly, with an actionable remediation message, never a
silent pass — unless the activation capability it observes exactly matches
what enforcement expects.

Every case here drives the seam with injected fakes/stubs (a fake
`probe_activation_capability` callable at the `launch-runtime.py` level, or
a fake `run` transport at the `activation_version_gate` level) — never a
real broker, a real installed CLI, or a real `mosaic` binary on PATH.
"""

from __future__ import annotations

import importlib.util
import io
import os
import shlex
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock


TOOLS_DIR = Path(__file__).parents[2] / "framework/tools/lease-broker"
if str(TOOLS_DIR) not in sys.path:
    sys.path.insert(0, str(TOOLS_DIR))


def load_tool(module_name: str, filename: str):
    spec = importlib.util.spec_from_file_location(module_name, TOOLS_DIR / filename)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"unable to load {filename}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


# Loaded under distinct module names from runtime_tools_unittest.py's own
# LAUNCHER/GATE loads — importlib.util.module_from_spec() gives each load a
# fresh module object regardless of name collisions, but distinct names keep
# tracebacks/debugging unambiguous when both files run in the same process.
LAUNCHER = load_tool("lease_runtime_launcher_version_coupling", "launch-runtime.py")
VERSION_GATE = load_tool("lease_activation_version_gate_test", "activation_version_gate.py")


def matching_capability() -> dict[str, object]:
    return dict(VERSION_GATE.EXPECTED_ACTIVATION_CAPABILITY)


def write_fake_mosaic(directory: Path, marker: Path) -> Path:
    directory.mkdir(parents=True, exist_ok=True)
    executable = directory / "mosaic"
    executable.write_text(
        "#!/bin/sh\n"
        f"printf '%s\\n' executed >> {shlex.quote(str(marker))}\n"
        "printf '%s\\n' "
        "'{\"name\":\"lease-runtime-activation\",\"version\":1}'\n",
        encoding="utf-8",
    )
    executable.chmod(0o755)
    return executable


class AssertActivationCapabilityMatchesTest(unittest.TestCase):
    """Unit-level coverage of `activation_version_gate.py`'s own assertion,
    isolated from the launch-runtime.py seam it is wired into below."""

    def test_matching_capability_passes_silently(self) -> None:
        VERSION_GATE.assert_activation_capability_matches(matching_capability())
        # No exception is the assertion; nothing further to check.

    def test_absent_capability_fails_closed_not_silent_pass(self) -> None:
        with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
            VERSION_GATE.assert_activation_capability_matches(None)
        message = str(raised.exception)
        self.assertIn("#869", message)
        self.assertIn("upgrade", message.lower())

    def test_version_mismatch_message_is_actionable(self) -> None:
        expected = {"name": "lease-runtime-activation", "version": 1}
        mismatched = {"name": "lease-runtime-activation", "version": 2}
        with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
            VERSION_GATE.assert_activation_capability_matches(mismatched, expected)
        message = str(raised.exception)
        self.assertIn("v2", message)
        self.assertIn("v1", message)
        self.assertIn("#869", message)
        self.assertIn("upgrade", message.lower())
        self.assertIn("version skew", message.lower())

    def test_name_mismatch_fails_loud(self) -> None:
        expected = {"name": "lease-runtime-activation", "version": 1}
        mismatched = {"name": "some-other-capability", "version": 1}
        with self.assertRaises(VERSION_GATE.VersionCouplingError) as raised:
            VERSION_GATE.assert_activation_capability_matches(mismatched, expected)
        message = str(raised.exception)
        self.assertIn("some-other-capability", message)
        self.assertIn("lease-runtime-activation", message)
        self.assertIn("#869", message)

    def test_reversed_drift_newer_activation_than_enforcement_expects_also_fails(self) -> None:
        # A build/deploy where ACTIVATION shipped ahead of ENFORCEMENT is
        # exactly as much version skew as the reverse (#828's actual shape
        # was enforcement ahead of activation) — the assertion must not special
        # case direction.
        expected = {"name": "lease-runtime-activation", "version": 1}
        newer_activation = {"name": "lease-runtime-activation", "version": 2}
        with self.assertRaises(VERSION_GATE.VersionCouplingError):
            VERSION_GATE.assert_activation_capability_matches(newer_activation, expected)


class ProbeActivationCapabilityTest(unittest.TestCase):
    """Coverage of the probe's command resolution and fail-closed transport
    handling — never spawns a real `mosaic` process."""

    def test_returns_none_when_mosaic_is_not_resolvable_on_path(self) -> None:
        # Keep even a deliberate ambient-lookup mutation away from any host
        # installation. The dedicated hermeticity tests below provide fake
        # ambient executables and markers.
        with mock.patch.dict(
            os.environ, {"PATH": "/nonexistent-ambient-bin-dir-for-869-c4-test"}
        ):
            result = VERSION_GATE.default_probe_activation_capability(
                {"PATH": "/nonexistent-bin-dir-for-869-c4-test"}
            )
        self.assertIsNone(result)

    def test_supplied_path_wins_over_ambient_process_path(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supplied_marker = root / "supplied.marker"
            ambient_marker = root / "ambient.marker"
            supplied_bin = root / "supplied-bin"
            ambient_bin = root / "ambient-bin"
            write_fake_mosaic(supplied_bin, supplied_marker)
            write_fake_mosaic(ambient_bin, ambient_marker)

            with mock.patch.dict(os.environ, {"PATH": str(ambient_bin)}):
                result = VERSION_GATE.default_probe_activation_capability(
                    {"PATH": str(supplied_bin)}
                )

            self.assertEqual(result, matching_capability())
            self.assertTrue(supplied_marker.exists())
            self.assertFalse(ambient_marker.exists())

    def test_absent_or_empty_supplied_path_never_falls_back_or_executes(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            ambient_marker = root / "ambient.marker"
            current_directory_marker = root / "current-directory.marker"
            ambient_bin = root / "ambient-bin"
            current_directory = root / "current-directory"
            write_fake_mosaic(ambient_bin, ambient_marker)
            write_fake_mosaic(current_directory, current_directory_marker)
            original_directory = Path.cwd()

            try:
                os.chdir(current_directory)
                with mock.patch.dict(os.environ, {"PATH": str(ambient_bin)}):
                    for supplied_environment in ({}, {"PATH": ""}):
                        with self.subTest(environ=supplied_environment):
                            result = VERSION_GATE.default_probe_activation_capability(
                                supplied_environment
                            )
                            self.assertIsNone(result)
                            self.assertFalse(ambient_marker.exists())
                            self.assertFalse(current_directory_marker.exists())
            finally:
                os.chdir(original_directory)

    def test_valid_override_wins_and_invalid_override_does_not_fall_back_to_path(
        self,
    ) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supplied_marker = root / "supplied.marker"
            ambient_marker = root / "ambient.marker"
            override_marker = root / "override.marker"
            supplied_bin = root / "supplied-bin"
            ambient_bin = root / "ambient-bin"
            override_bin = root / "override-bin"
            write_fake_mosaic(supplied_bin, supplied_marker)
            write_fake_mosaic(ambient_bin, ambient_marker)
            override_executable = write_fake_mosaic(override_bin, override_marker)

            with mock.patch.dict(os.environ, {"PATH": str(ambient_bin)}):
                result = VERSION_GATE.default_probe_activation_capability(
                    {
                        "PATH": str(supplied_bin),
                        VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: str(override_executable),
                    }
                )
                self.assertEqual(result, matching_capability())
                self.assertTrue(override_marker.exists())
                self.assertFalse(supplied_marker.exists())
                self.assertFalse(ambient_marker.exists())

                override_marker.unlink()
                result = VERSION_GATE.default_probe_activation_capability(
                    {
                        "PATH": str(supplied_bin),
                        VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: str(
                            root / "invalid-override" / "mosaic"
                        ),
                    }
                )
                self.assertIsNone(result)
                self.assertFalse(override_marker.exists())
                self.assertFalse(supplied_marker.exists())
                self.assertFalse(ambient_marker.exists())

    def test_override_command_is_parsed_and_the_probe_subcommand_is_not_double_appended(
        self,
    ) -> None:
        captured: list[list[str]] = []

        class FakeCompleted:
            returncode = 0
            stdout = '{"name": "lease-runtime-activation", "version": 1}'

        def fake_run(argv: list[str], **_kwargs: object) -> FakeCompleted:
            captured.append(argv)
            return FakeCompleted()

        result = VERSION_GATE.default_probe_activation_capability(
            {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic __lease-capability"},
            run=fake_run,
        )
        self.assertEqual(result, {"name": "lease-runtime-activation", "version": 1})
        self.assertEqual(captured, [["/fake/mosaic", "__lease-capability"]])

    def test_probe_passes_ten_second_timeout_to_runner(self) -> None:
        captured_argv: list[str] = []
        captured_kwargs: dict[str, object] = {}

        class FakeCompleted:
            returncode = 0
            stdout = '{"name": "lease-runtime-activation", "version": 1}'

        def fake_run(argv: list[str], **kwargs: object) -> FakeCompleted:
            captured_argv.extend(argv)
            captured_kwargs.update(kwargs)
            return FakeCompleted()

        result = VERSION_GATE.default_probe_activation_capability(
            {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"},
            run=fake_run,
        )

        self.assertEqual(result, matching_capability())
        self.assertEqual(captured_argv, ["/fake/mosaic"])
        self.assertEqual(captured_kwargs["timeout"], 10.0)
        self.assertEqual(captured_kwargs["check"], False)

    def test_fails_closed_on_nonzero_exit_malformed_json_and_missing_fields(self) -> None:
        class NonZeroExit:
            returncode = 1
            stdout = '{"name": "lease-runtime-activation", "version": 1}'

        class MalformedOutput:
            returncode = 0
            stdout = "not-json"

        class MissingVersion:
            returncode = 0
            stdout = '{"name": "lease-runtime-activation"}'

        class WrongShapeVersion:
            returncode = 0
            stdout = '{"name": "lease-runtime-activation", "version": "1"}'

        class BooleanVersion:
            # bool is a subclass of int in Python; must not be accepted as
            # a version number.
            returncode = 0
            stdout = '{"name": "lease-runtime-activation", "version": true}'

        for fake in (
            NonZeroExit(),
            MalformedOutput(),
            MissingVersion(),
            WrongShapeVersion(),
            BooleanVersion(),
        ):
            with self.subTest(stdout=fake.stdout, returncode=fake.returncode):
                result = VERSION_GATE.default_probe_activation_capability(
                    {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"},
                    run=lambda *_a, fake=fake, **_kw: fake,
                )
                self.assertIsNone(result)

    def test_fails_closed_on_timeout_and_transport_error(self) -> None:
        def timeout_run(*_args: object, **_kwargs: object) -> None:
            raise subprocess.TimeoutExpired(cmd="mosaic", timeout=10.0)

        def oserror_run(*_args: object, **_kwargs: object) -> None:
            raise OSError("no such file or directory")

        for run_fake in (timeout_run, oserror_run):
            with self.subTest(run=run_fake.__name__):
                result = VERSION_GATE.default_probe_activation_capability(
                    {VERSION_GATE.MOSAIC_COMMAND_OVERRIDE_VAR: "/fake/mosaic"},
                    run=run_fake,
                )
                self.assertIsNone(result)


class LaunchRuntimeVersionCouplingSeamTest(unittest.TestCase):
    """End-to-end (still fully faked) coverage of the seam as wired into
    `launch-runtime.py`'s `main()` — the strongest natural enforcement point
    per the C4 card, run before any broker registration."""

    def _run(self, *, probe):
        calls: dict[str, object] = {}

        def request(_path: Path, payload: dict[str, object]) -> dict[str, object]:
            calls["registered"] = True
            calls["request"] = payload
            return {"ok": True, "session_id": "a" * 64}

        def execute(command: str, argv: list[str], environment: dict[str, str]) -> None:
            calls["executed"] = (command, argv, environment)

        def initialize_generation(_path: Path, _generation: int) -> None:
            calls["generation_initialized"] = True

        stderr = io.StringIO()
        with redirect_stderr(stderr):
            result = LAUNCHER.main(
                ["--runtime", "claude", "--", "claude", "--print", "hello"],
                environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"},
                request=request,
                execute=execute,
                initialize_generation=initialize_generation,
                probe_activation_capability=probe,
            )
        return result, stderr.getvalue(), calls

    def test_matching_activation_version_passes_and_the_gate_proceeds(self) -> None:
        result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: matching_capability())
        self.assertEqual(result, 0)
        self.assertEqual(stderr_text, "")
        self.assertTrue(calls.get("registered"))
        self.assertIn("executed", calls)

    def test_version_mismatch_fails_loud_denies_and_never_registers_or_execs(self) -> None:
        expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY
        mismatched = {"name": expected["name"], "version": expected["version"] + 1}
        result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched)

        self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
        self.assertNotEqual(result, 0)
        self.assertIn("#869", stderr_text)
        self.assertIn(f"v{mismatched['version']}", stderr_text)
        self.assertIn(f"v{expected['version']}", stderr_text)
        self.assertIn("upgrade", stderr_text.lower())
        # Never reaches broker registration or exec — the version gate is a
        # hard stop, not advisory.
        self.assertNotIn("registered", calls)
        self.assertNotIn("executed", calls)

    def test_name_mismatch_fails_loud(self) -> None:
        expected = LAUNCHER.EXPECTED_ACTIVATION_CAPABILITY
        mismatched = {"name": "some-other-capability", "version": expected["version"]}
        result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: mismatched)

        self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
        self.assertIn("#869", stderr_text)
        self.assertIn("some-other-capability", stderr_text)
        self.assertNotIn("registered", calls)
        self.assertNotIn("executed", calls)

    def test_absent_activation_capability_fails_closed_not_a_silent_pass(self) -> None:
        result, stderr_text, calls = self._run(probe=lambda *_a, **_kw: None)

        self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)
        self.assertNotEqual(result, 0)
        self.assertIn("#869", stderr_text)
        self.assertNotIn("registered", calls)
        self.assertNotIn("executed", calls)

    def test_version_gate_runs_before_and_independently_of_broker_registration(self) -> None:
        def request_must_not_be_called(*_args: object, **_kwargs: object) -> dict[str, object]:
            self.fail("broker must not be contacted when activation version is mismatched")

        stderr = io.StringIO()
        with redirect_stderr(stderr):
            result = LAUNCHER.main(
                ["--runtime", "claude", "--", "claude"],
                environ={"MOSAIC_LEASE_BROKER_SOCKET": "/run/test/broker.sock"},
                request=request_must_not_be_called,
                probe_activation_capability=lambda *_a, **_kw: None,
            )
        self.assertEqual(result, LAUNCHER.EXIT_VERSION_SKEW)

    def test_dedicated_exit_code_never_collides_with_usage_or_registration_codes(self) -> None:
        # Distinctness guard: a version-skew denial must never be mistaken
        # for the pre-existing usage error (64) or registration/exec
        # fail-closed code (1) this script already owns.
        self.assertNotIn(LAUNCHER.EXIT_VERSION_SKEW, (0, 1, 64))


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