# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-tool-descriptor-inheritance-the-skill-injection-boundary-in-mcp-tool-augmented-agents.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-tool-descriptor-inheritance-the-skill-injection-boundary-in-mcp-tool-augmented-agents.warrant.json
# Markdown SHA-256:   ac59c6784a83ccc5b8899a8c2f54caf958484fc3c80c96cfcf696904ccc201b1
# Code AST SHA-256:   d24b7df81faf9feff3b85fb15aa3ec83467ddf4ba9fd2c3cbf7995bf99ebbff6
# Verified Classes:   DescriptorProvenanceError, PinnedSession
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class DescriptorProvenanceError(RuntimeError):
    """Raised when a refreshed tool/skill descriptor fails provenance continuity."""
    pass


class PinnedSession:
    def __init__(self, origin: str, signature: str, channel: str):
        self.pinned_origin = origin
        self.pinned_signature = signature
        self.initial_channel = channel
        self.frozen_descriptors = {}


def inherit_descriptor(session: PinnedSession, descriptor: dict) -> dict:
    """Validate a post-snapshot descriptor against the session's pinned provenance.

    Returns the descriptor only if provenance continuity holds. Otherwise raises
    DescriptorProvenanceError so the caller keeps the prior snapshot and logs the event.
    """
    if descriptor.get("origin") != session.pinned_origin:
        raise DescriptorProvenanceError(
            f"origin mismatch: descriptor claims {descriptor.get('origin')!r}, "
            f"session pinned {session.pinned_origin!r}"
        )
    if descriptor.get("manifest_signature") != session.pinned_signature:
        raise DescriptorProvenanceError(
            f"signature mismatch for {descriptor.get('name')!r}: "
            f"descriptor signed by {descriptor.get('manifest_signature')!r}, "
            f"session pinned {session.pinned_signature!r}"
        )
    if descriptor.get("arrival_channel") != session.initial_channel:
        raise DescriptorProvenanceError(
            f"channel substitution: arrived via {descriptor.get('arrival_channel')!r}, "
            f"session pinned {session.initial_channel!r}"
        )
    if descriptor.get("introduces_new_capability_surface"):
        raise DescriptorProvenanceError(
            f"capability surface expansion in {descriptor.get('name')!r} "
            f"without re-pin; refusing inheritance"
        )
    return descriptor


def refresh_tools(session: PinnedSession, incoming: list[dict]) -> dict:
    accepted = {}
    for descriptor in incoming:
        try:
            accepted[descriptor["name"]] = inherit_descriptor(session, descriptor)
        except DescriptorProvenanceError as exc:
            # Keep prior snapshot; record rejection for audit.
            print(f"[provenance-reject] {descriptor.get('name')}: {exc}")
    session.frozen_descriptors.update(accepted)
    return session.frozen_descriptors