#!/usr/bin/env python3
"""RED-first framework-firewall and portability contracts for shipped skills."""

from __future__ import annotations

import importlib.util
import re
import unittest
from pathlib import Path


MOSAIC = Path(__file__).parents[2]
REPOSITORY = MOSAIC.parents[1]
SKILLS = MOSAIC / "framework/skills"
REFRESH_SKILL = SKILLS / "mosaic-context-refresh/SKILL.md"
GATE_PATH = MOSAIC / "framework/tools/lease-broker/mutator-gate.py"
COMPACTION_THREAT = REPOSITORY / "docs/architecture/compaction-revocation.md"
RECEIPT_PROTOCOL = REPOSITORY / "docs/architecture/lease-broker-protocol.md"
OPERATOR_HOME = re.compile(r"/home/[^/\s]+/")
RECOVERY_PLACEHOLDER = "/absolute/path/to/mosaic/tools/lease-broker/recover-context.py"
CONSTRUCTION_PLACEHOLDER = "/absolute/path/to/mosaic-context-refresh-construction.json"


def load_gate():
    spec = importlib.util.spec_from_file_location("framework_skill_portability_gate", GATE_PATH)
    if spec is None or spec.loader is None:
        raise RuntimeError("unable to load mutator gate")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


GATE = load_gate()


class FrameworkSkillPortabilityTest(unittest.TestCase):
    def test_shipped_framework_skills_contain_no_operator_home_path(self) -> None:
        offenders = [
            str(path.relative_to(SKILLS))
            for path in SKILLS.rglob("SKILL.md")
            if OPERATOR_HOME.search(path.read_text(encoding="utf-8"))
        ]
        self.assertEqual(offenders, [])

    def test_shipped_recovery_template_resolves_to_a_literal_install_path_and_admits(self) -> None:
        source = REFRESH_SKILL.read_text(encoding="utf-8")
        self.assertIn(RECOVERY_PLACEHOLDER, source)
        self.assertIn(CONSTRUCTION_PLACEHOLDER, source)
        resolved_recovery = "/opt/mosaic/tools/lease-broker/recover-context.py"
        resolved_construction = "/opt/mosaic/recovery/construction.json"
        rendered = source.replace(RECOVERY_PLACEHOLDER, resolved_recovery).replace(
            CONSTRUCTION_PLACEHOLDER, resolved_construction
        )
        match = re.search(r"```bash\s*\n\s*(.*?)\n\s*```", rendered, flags=re.DOTALL)
        self.assertIsNotNone(match)
        command = match.group(1) if match is not None else ""
        self.assertEqual(
            GATE.recovery_invocation_name(
                {"tool_name": "Bash", "tool_input": {"command": command}},
                Path(resolved_recovery),
            ),
            GATE.RECOVERY_TOOL,
        )

    def test_t30_dual_hook_miss_matches_the_amended_threat_table(self) -> None:
        threat_table = COMPACTION_THREAT.read_text(encoding="utf-8")
        receipt_protocol = RECEIPT_PROTOCOL.read_text(encoding="utf-8")
        self.assertRegex(
            threat_table,
            r"\| Both observers are missed, lease unexpired\s*\|\s*\*\*ALLOWED\*\* inside the bounded residual stale window\.",
        )
        self.assertRegex(
            threat_table,
            r"\| Both observers are missed, lease expired\s*\|\s*\*\*DENIED\*\* by monotonic TTL expiry\.",
        )
        self.assertRegex(
            threat_table,
            r"`enable_status_check=False` \(status checks not\s+enforced\)",
        )
        self.assertRegex(
            receipt_protocol,
            r"detects an \*\*ABSENT\*\* or \*\*PREFIX-TRUNCATED\*\* terminal\s+token",
        )
        required_threat_text = (
            "## T-C server-side branch-protection posture",
            "`main` is push-blocked and PR-only-merge",
            "Status-check enforcement and approval enforcement are **RECOMMENDED**.",
            "## Current-vs-required gap (recorded, not enacted)",
            "`enable_push=False` (push-block present)",
            "`require_approvals=0` (approvals not enforced)",
            "`block_on_official_review=False` (official review not enforced)",
        )
        required_receipt_text = (
            "## Receipt boundary and T-C residual (R1)",
            "**MIDDLE-DROP** that preserves the tail",
            "**NOT receipt-detectable**",
            "server-side protected-branch controls",
        )
        for contract_text in required_threat_text:
            with self.subTest(document="threat-table", contract_text=contract_text):
                self.assertIn(contract_text, threat_table)
        for contract_text in required_receipt_text:
            with self.subTest(document="receipt-protocol", contract_text=contract_text):
                self.assertIn(contract_text, receipt_protocol)


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