# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-skill-injection-attacks-when-the-loader-has-already-resolved-trust.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-skill-injection-attacks-when-the-loader-has-already-resolved-trust.warrant.json
# Markdown SHA-256:   0359e1e608fba3126011e755ff8ca7f77ca46a1553a792e84835b0528aacd07b
# Code AST SHA-256:   e7ccf5153dbdceacd9a12f9471f0878616984cec08602f0e50f512c38b37a257
# Verified Classes:   CapabilityViolation, LoaderGrantDefect, Capability, Skill
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet, Tuple


class CapabilityViolation(RuntimeError):
    """Raised when exercised capability exceeds declared capability."""
    pass


class LoaderGrantDefect(RuntimeError):
    """Raised when an enforced grant is broader than its declared capability."""
    pass


@dataclass(frozen=True)
class Capability:
    kind: str          # "fs:read", "fs:write", "net:post", ...
    scope: str         # subtree, endpoint, etc.


@dataclass
class Skill:
    name: str
    declared: FrozenSet[Capability]
    exercised: set = field(default_factory=set)


def _subsumes(enforced: Capability, declared: Capability) -> bool:
    # enforced grant must be within declared scope for the same kind
    if enforced.kind != declared.kind:
        return False
    return enforced.scope == declared.scope or enforced.scope.startswith(
        declared.scope.rstrip("/") + "/"
    )


def check_loader_grant(skill: Skill, enforced: Capability) -> None:
    if not any(_subsumes(enforced, d) for d in skill.declared):
        raise LoaderGrantDefect(
            f"skill {skill.name!r}: enforced grant {enforced} is not subsumed "
            f"by any declared capability {sorted(skill.declared)}"
        )


def check_exercised(skill: Skill) -> None:
    for cap in skill.exercised:
        if not any(_subsumes(cap, d) for d in skill.declared):
            raise CapabilityViolation(
                f"skill {skill.name!r}: exercised {cap} outside declared "
                f"surface {sorted(skill.declared)}"
            )


def check_flow(src: Skill, dst: Skill, value_origin: Capability,
               value_sink: Capability) -> None:
    # data originating under src's read set flowing into dst's write/net set
    if value_origin not in src.exercised:
        raise CapabilityViolation(
            f"flow origin {value_origin} not in {src.name!r} exercised set"
        )
    if value_sink not in dst.exercised:
        raise CapabilityViolation(
            f"flow sink {value_sink} not in {dst.name!r} exercised set"
        )
    covered_src = any(_subsumes(value_origin, d) for d in src.declared)
    covered_dst = any(_subsumes(value_sink, d) for d in dst.declared)
    if not (covered_src and covered_dst):
        raise CapabilityViolation(
            f"undeclared data flow {src.name!r}:{value_origin} -> "
            f"{dst.name!r}:{value_sink}"
        )


def load_and_verify(skill: Skill, enforced_grants: Tuple[Capability, ...]) -> None:
    for g in enforced_grants:
        check_loader_grant(skill, g)
    check_exercised(skill)