# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-manifest-declared-capability-that-diverges-from-runtime-behavior-in-agent-skill-injection-surfaces.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-manifest-declared-capability-that-diverges-from-runtime-behavior-in-agent-skill-injection-surfaces.warrant.json
# Markdown SHA-256:   e617c8058ecbc0cbc8a810a1d7e9bcdff3cd00b2d956502fc5a8951d7f688f35
# Code AST SHA-256:   b098388e29a9a25fce37f38ad7e50c1cf5352a62d5a23b3f328f3459a6be5eee
# Verified Classes:   CapabilityDriftError, SkillRegistry
# Defensive Guards:   5 fail-closed invariants (zero external deps)

class CapabilityDriftError(RuntimeError):
    """Raised when a skill/plugin's runtime behavior or advertisement
    diverges from its manifest-declared capability set."""
    pass


class SkillRegistry:
    def __init__(self):
        self._advertised = {}   # skill_id -> frozenset of declared tools
        self._registered = {}   # tool_name -> skill_id
        self._declared_caps = {}  # skill_id -> frozenset of declared side-effect hosts/paths

    def register(self, skill_id, manifest):
        advertised = frozenset(manifest.get("tools", []))
        declared_caps = frozenset(manifest.get("capabilities", []))
        if not advertised:
            raise CapabilityDriftError(
                f"skill {skill_id!r} advertises no tools; empty advertisement "
                f"is an unbounded surface and cannot be validated"
            )
        for tool in advertised:
            if tool in self._registered:
                raise CapabilityDriftError(
                    f"tool {tool!r} already registered by {self._registered[tool]!r}; "
                    f"refusing shadowing registration from {skill_id!r}"
                )
        self._advertised[skill_id] = advertised
        self._declared_caps[skill_id] = declared_caps
        for tool in advertised:
            self._registered[tool] = skill_id

    def check_dispatch(self, skill_id, tool_name, observed_effects):
        """observed_effects: iterable of (kind, value) e.g. ('net', 'host:443'),
        ('fs_write', '/etc/passwd'), ('subprocess', 'curl')."""
        advertised = self._advertised.get(skill_id)
        if advertised is None:
            raise CapabilityDriftError(
                f"dispatch for unregistered skill {skill_id!r}; no advertisement baseline"
            )
        if tool_name not in advertised:
            raise CapabilityDriftError(
                f"skill {skill_id!r} invoked tool {tool_name!r} not in advertised set "
                f"{sorted(advertised)}; advertisement drift"
            )
        declared = self._declared_caps[skill_id]
        for kind, value in observed_effects:
            token = f"{kind}:{value}"
            if token not in declared and kind not in declared:
                raise CapabilityDriftError(
                    f"skill {skill_id!r} produced undeclared side effect {token!r}; "
                    f"declared capabilities {sorted(declared)} do not cover it"
                )
        return True


def dispatch(registry, skill_id, tool_name, observed_effects):
    # Invariant check precedes any execution of the skill's code path.
    registry.check_dispatch(skill_id, tool_name, observed_effects)
    return {"skill_id": skill_id, "tool": tool_name, "status": "dispatched"}