# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/loss-of-control-to-a-misaligned-ai-system.md
# Remnant Warrant:    https://corpusfetch.com/c/loss-of-control-to-a-misaligned-ai-system.warrant.json
# Markdown SHA-256:   6b352d56afec420707d71263ee43e5eefbe981c84930ff2810bca2f5ca0287d1
# Code AST SHA-256:   98a16133dae8b04e94d062053c3c5b266221a0ef267ecfd809cf6a88bfbef178
# Verified Classes:   ContainmentBreachRefusalError, ExecutionLeaseContract, ExternalContainmentSupervisor
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import time
from dataclasses import dataclass
from pathlib import Path
from typing import Final


class ContainmentBreachRefusalError(PermissionError):
    '''Raised when an autonomous loop exceeds its lease, recursion depth, or targets protected oversight paths.'''
    pass


@dataclass(frozen=True)
class ExecutionLeaseContract:
    lease_id: str
    issued_at_monotonic: float
    ttl_seconds: float
    max_step_budget: int
    max_recursion_depth: int
    protected_readonly_prefixes: tuple[str, ...]


class ExternalContainmentSupervisor:
    '''
    Enforces deterministic architectural control over autonomous agent loops:
    1. Enforces hard wall-clock TTL (`ttl_seconds`), step budget (`max_step_budget`),
       and sub-agent recursion depth (`max_recursion_depth`).
    2. Blocks write mutations to protected oversight, test-harness, and policy paths.
    3. Honors immediate cooperative halt flags alongside OS-level cgroup bounds.
    '''

    DEFAULT_PROTECTED_PREFIXES: Final[tuple[str, ...]] = (
        ".github/workflows",
        ".git/hooks",
        "/etc/security",
        "tests/security_gates",
    )

    def __init__(self, lease: ExecutionLeaseContract) -> None:
        self.lease = lease
        self._steps_consumed = 0
        self._emergency_halt_requested = False

    def trigger_emergency_halt(self) -> None:
        '''Allows external operator or supervisor to halt execution immediately.'''
        self._emergency_halt_requested = True

    def check_step_and_lease(self, *, current_recursion_depth: int, now_monotonic: float | None = None) -> int:
        '''Gate 1 & Gate 3: Verifies interrupt status, lease TTL, step budget, and recursion cap.'''
        if self._emergency_halt_requested:
            raise ContainmentBreachRefusalError(
                f"Lease '{self.lease.lease_id}' halted: external operator interrupt signal active."
            )

        now = time.monotonic() if now_monotonic is None else now_monotonic
        elapsed = now - self.lease.issued_at_monotonic
        if elapsed > self.lease.ttl_seconds:
            raise ContainmentBreachRefusalError(
                f"Lease '{self.lease.lease_id}' expired: elapsed {elapsed:.2f}s > ttl {self.lease.ttl_seconds:.2f}s."
            )

        if current_recursion_depth > self.lease.max_recursion_depth:
            raise ContainmentBreachRefusalError(
                f"Sub-agent recursion depth ({current_recursion_depth}) exceeded max_recursion_depth "
                f"({self.lease.max_recursion_depth})."
            )

        self._steps_consumed += 1
        if self._steps_consumed > self.lease.max_step_budget:
            raise ContainmentBreachRefusalError(
                f"Step budget exhausted ({self._steps_consumed} > {self.lease.max_step_budget})."
            )

        return self.lease.max_step_budget - self._steps_consumed

    def assert_write_path_permitted(self, target_path: Path) -> None:
        '''Gate 2 & Gate 4: Prevents agent from tampering with evaluation harnesses or containment configs.'''
        norm_target = str(target_path.as_posix()).lstrip("./")
        all_protected = (*self.DEFAULT_PROTECTED_PREFIXES, *self.lease.protected_readonly_prefixes)
        for prefix in all_protected:
            clean_prefix = prefix.lstrip("./")
            if norm_target == clean_prefix or norm_target.startswith(clean_prefix + "/"):
                raise ContainmentBreachRefusalError(
                    f"Containment Breach Blocked: agent may not modify protected oversight/harness path '{target_path}'."
                )