# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-treat-an-a2a-handoff-as-a-taint-preserving-boundary-rather-than-a-trust-reset.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-treat-an-a2a-handoff-as-a-taint-preserving-boundary-rather-than-a-trust-reset.warrant.json
# Markdown SHA-256:   e83b4e6562cb445e9bec9f82b8da3ae1b8a333cd90cdddb00d9782ab4263b25a
# Code AST SHA-256:   2558a61d893a1b2df49c9520313975d28a3e3c5fe315833335532433531685f8
# Verified Classes:   TaintLineageError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class TaintLineageError(RuntimeError):
    """Raised when an A2A handoff would cross a capability boundary
    without preserving injection lineage from the delegator's context."""
    pass


TRUSTED_PEER = "trusted_peer"


def authorize_handoff(delegator_context_sources, payload, delegate_capability):
    """
    delegator_context_sources: iterable of provenance labels that fed the
        delegator's context, e.g. ["retrieved:doc-4471", "user:free_text"].
    payload: dict that may carry a "taint" field propagated by the delegator.
    delegate_capability: one of "read", "write", "execute", "transfer".
    """
    actionable = delegate_capability in ("write", "execute", "transfer")
    payload_taint = payload.get("taint")

    # Capability boundary: the delegate can do more than the source could.
    if actionable:
        if payload_taint is None:
            # Absence of a label is NOT proof of cleanliness. If the
            # delegator's context touched any lower-trust source, the
            # handoff must carry lineage; otherwise we cannot distinguish
            # a clean instruction from a laundered summary.
            untrusted = [
                s for s in delegator_context_sources
                if not s.startswith(TRUSTED_PEER)
            ]
            if untrusted:
                raise TaintLineageError(
                    "actionable handoff from delegator whose context includes "
                    f"untrusted sources {untrusted!r} carries no taint label; "
                    "injection lineage was dropped at the boundary"
                )
        else:
            unauthorized = [
                s for s in payload_taint
                if s not in delegator_context_sources
                and not s.startswith(TRUSTED_PEER)
            ]
            if unauthorized:
                # Taint present but not traceable to a known source:
                # the delegator fabricated or mis-propagated lineage.
                raise TaintLineageError(
                    f"payload taint {unauthorized!r} has no lineage in "
                    "delegator context; cannot certify provenance"
                )
            if any(not s.startswith(TRUSTED_PEER) for s in payload_taint):
                # Taint present and traceable: escalate rather than execute.
                return {"decision": "escalate", "taint": list(payload_taint)}

    # Read-capability handoffs may reset to trusted-peer; risk is bounded.
    return {"decision": "allow", "taint": list(payload_taint or [])}


def run_migration(delegator_context_sources, payload):
    # Executor receives a destructive task from the planner.
    verdict = authorize_handoff(
        delegator_context_sources=delegator_context_sources,
        payload=payload,
        delegate_capability="execute",
    )
    if verdict["decision"] != "allow":
        raise TaintLineageError(
            f"destructive migration blocked: taint {verdict['taint']!r} "
            "not authorized for execute capability"
        )
    return "migration started"


# Laundered-summary case: planner summarized retrieved doc, dropped the label.
try:
    run_migration(
        delegator_context_sources=["retrieved:doc-4471"],
        payload={"action": "run_migration", "params": {"destructive": True}},
    )
except TaintLineageError as exc:
    print(f"blocked: {exc}")