# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-isolate-taint-across-agent-to-agent-prompt-injection-when-capability-scopes-disagree-at-the-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-isolate-taint-across-agent-to-agent-prompt-injection-when-capability-scopes-disagree-at-the-boundary.warrant.json
# Markdown SHA-256:   837f5439da3bb14e866d3571ad81be05689e5932a4481d90edcb15869121e4dc
# Code AST SHA-256:   79ef65f52460cc2329cda80fea918dcf39ffc610398673c0fd15b10ebca084c7
# Verified Classes:   TaintBoundaryViolation, CapabilitySink
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class TaintBoundaryViolation(RuntimeError):
    """Raised when tainted data reaches a capability sink without declassification."""
    pass


class CapabilitySink:
    def __init__(self, name, taint_policy):
        if taint_policy not in ("reject_tainted", "allow_tainted"):
            raise TaintBoundaryViolation(
                f"capability {name!r} registered without a declared taint policy"
            )
        self.name = name
        self.taint_policy = taint_policy

    def invoke(self, payload, taint_labels, declassified=False):
        if taint_labels and self.taint_policy == "reject_tainted" and not declassified:
            raise TaintBoundaryViolation(
                f"capability {self.name!r} refused tainted payload "
                f"(labels={sorted(taint_labels)}); explicit declassification required"
            )
        return {"capability": self.name, "accepted": True, "labels": sorted(taint_labels)}


def deliver(worker_output, planner_sink):
    # Envelope metadata must survive deserialization; flattening drops it and
    # severs the taint channel. We reconstruct taint from the envelope, not the text.
    labels = set(worker_output.get("taint_labels", []))
    if worker_output.get("source") == "untrusted_document":
        labels.add("untrusted:document")
    return planner_sink.invoke(
        payload=worker_output["content"],
        taint_labels=labels,
        declassified=worker_output.get("declassified", False),
    )


def spawn_child(parent_taint, child_scope):
    # Spawn is derivation, not creation: taint is inherited until audited declassify.
    if parent_taint and "declassified" not in parent_taint:
        raise TaintBoundaryViolation(
            "cannot spawn untainted child from tainted parent without audited declassification"
        )
    return {"scope": child_scope, "taint": set(parent_taint)}