# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-a-sandbox-escape-from-a-legitimate-skill-escalation-when-syscall-level-capability-filters-are-the-only-enforcement.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-a-sandbox-escape-from-a-legitimate-skill-escalation-when-syscall-level-capability-filters-are-the-only-enforcement.warrant.json
# Markdown SHA-256:   6229feed79e3ceaad696e9966cd3a54757a18e0f41b47f473eb488763ef78305
# Code AST SHA-256:   08e4b0ff94e5d2c8b4998dad22eed1120697d7a83120a9edf2febea76875a958
# Verified Classes:   UnattributedCapabilityError, FilterSwapRaceError, Grant, GrantLedger
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import os
import threading
from dataclasses import dataclass, field
from typing import Optional


class UnattributedCapabilityError(RuntimeError):
    """A syscall was permitted by the filter but has no matching grant in the
    current execution epoch. Treat as a sandbox escape until proven otherwise."""
    pass


class FilterSwapRaceError(RuntimeError):
    """A filter mutation could not be proven atomic across all threads."""
    pass


@dataclass(frozen=True)
class Grant:
    skill_id: str
    syscall: str
    arg_tuple: tuple
    epoch: int


@dataclass
class GrantLedger:
    epoch: int = 0
    _grants: dict = field(default_factory=dict)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def begin_mutation(self) -> int:
        with self._lock:
            self.epoch += 1
            return self.epoch

    def record(self, grant: Grant) -> None:
        with self._lock:
            self._grants[(grant.skill_id, grant.syscall, grant.arg_tuple, grant.epoch)] = True

    def is_attributed(self, skill_id: str, syscall: str, arg_tuple: tuple, epoch: int) -> bool:
        with self._lock:
            return (skill_id, syscall, arg_tuple, epoch) in self._grants


def canonicalize(path: str) -> str:
    # Runtime-owned normalization; the skill never supplies the compared form.
    return os.path.realpath(os.path.abspath(path))


def escalate_and_widen(
    ledger: GrantLedger,
    skill_id: str,
    syscall: str,
    raw_arg: str,
    apply_filter_widen,
    thread_barrier,
) -> None:
    """Two-phase commit: ledger-write, widen filter, barrier, resume."""
    arg_tuple = (canonicalize(raw_arg),)
    epoch = ledger.begin_mutation()

    # Phase 1: ledger write (orphan on crash is safe and revocable).
    ledger.record(Grant(skill_id, syscall, arg_tuple, epoch))

    # Phase 2: widen filter, then prove atomicity across threads.
    apply_filter_widen(syscall, arg_tuple)
    if not thread_barrier.wait(timeout=0.5):
        # Cannot prove no syscall was in flight across the swap.
        raise FilterSwapRaceError(
            f"filter swap for {skill_id}:{syscall} not proven atomic; quarantine required"
        )


def check_permitted_syscall(
    ledger: GrantLedger,
    skill_id: str,
    syscall: str,
    raw_arg: str,
    current_epoch: int,
) -> None:
    """Audit pass: every filter-permitted syscall after a mutation must be
    attributable to a grant in the current epoch."""
    arg_tuple = (canonicalize(raw_arg),)
    if not ledger.is_attributed(skill_id, syscall, arg_tuple, current_epoch):
        raise UnattributedCapabilityError(
            f"skill {skill_id} invoked permitted syscall {syscall}{arg_tuple} "
            f"in epoch {current_epoch} with no matching grant; "
            f"treating as escape"
        )