# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-skill-registry-load-as-an-injection-event-rather-than-a-benign-refresh.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-a-skill-registry-load-as-an-injection-event-rather-than-a-benign-refresh.warrant.json
# Markdown SHA-256:   6a3c2fc59596711920b7eb62a6ba298973a92819a9f1aeec207db84dcaab65ae
# Code AST SHA-256:   797a370a1f693df1e28cc09ab051467d07b0b8d44225c179320c357a5be22b70
# Verified Classes:   SkillInjectionError, SkillRef, ResolutionSurface
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Dict, Set, Tuple


class SkillInjectionError(RuntimeError):
    """Raised when a registry load mutates the resolution surface of a loaded skill."""
    pass


@dataclass(frozen=True)
class SkillRef:
    trigger_key: str
    artifact_id: str
    capability_scope: frozenset


@dataclass
class ResolutionSurface:
    # trigger_key -> (artifact_id, capability_scope)
    entries: Dict[str, Tuple[str, frozenset]] = field(default_factory=dict)

    def closure(self) -> Set[Tuple[str, str, frozenset]]:
        return {(k, v[0], v[1]) for k, v in self.entries.items()}


def apply_refresh(
    loaded: ResolutionSurface,
    incoming: ResolutionSurface,
    operator_trusted: bool,
    consent_gated: Set[str],
) -> ResolutionSurface:
    """Apply a registry refresh only if it cannot reroute a loaded skill.

    Raises SkillInjectionError when the refresh mutates the resolution
    surface reachable by an already-loaded skill and the registry is not
    operator-trusted.
    """
    if operator_trusted:
        return incoming

    loaded_closure = loaded.closure()
    incoming_closure = incoming.closure()
    delta = incoming_closure.symmetric_difference(loaded_closure)

    if not delta:
        return incoming

    # Any changed trigger_key that a loaded skill currently occupies is a
    # reroute of that skill, regardless of payload identity.
    loaded_keys = {k for (k, _, _) in loaded_closure}
    rerouted_keys = {k for (k, _, _) in delta} & loaded_keys

    if rerouted_keys:
        raise SkillInjectionError(
            f"refresh reroutes loaded trigger keys {sorted(rerouted_keys)}; "
            f"treating as injection event, not refresh"
        )

    # Delta confined to keys not occupied by loaded skills: apply only if
    # those skills are consent-gated; otherwise fail closed.
    ungated = {k for (k, _, _) in delta} - consent_gated
    if ungated:
        raise SkillInjectionError(
            f"refresh introduces unconsented trigger keys {sorted(ungated)}; "
            f"cannot verify resolution closure is disjoint"
        )

    return incoming