# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-prompt-injection-taint-tracking-fail-at-the-join-of-two-clean-labels.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-prompt-injection-taint-tracking-fail-at-the-join-of-two-clean-labels.warrant.json
# Markdown SHA-256:   3e5e7cf623c27f9b7e2975a69fc2360047e4f222f0335d6460d660a6d17a4160
# Code AST SHA-256:   1635293ae080a697df07a90126324a74c00398f1966711f956d54f4d3b9cd42f
# Verified Classes:   TaintJoinViolation, ProvenanceNode, LabeledValue
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Literal


class TaintJoinViolation(RuntimeError):
    """Raised when a capability is invoked with an argument whose provenance
    includes a taint-preserving transformation that was mislabeled as a
    declassifier."""


NodeKind = Literal["source", "transform", "verify"]


@dataclass
class ProvenanceNode:
    kind: NodeKind
    name: str
    tainted_input: bool
    # A transform is a true declassifier only if it does not echo input bytes.
    preserves_bytes: bool = True


@dataclass
class LabeledValue:
    value: object
    leaf_label: Literal["CLEAN", "TAINTED"]
    provenance: list[ProvenanceNode] = field(default_factory=list)


def join_label(a: LabeledValue, b: LabeledValue) -> LabeledValue:
    # Lattice join: TAINTED dominates. But the join must also inspect the
    # full provenance chain, not just the leaf labels.
    merged = a.provenance + b.provenance
    leaf = "TAINTED" if "TAINTED" in (a.leaf_label, b.leaf_label) else "CLEAN"
    return LabeledValue(value=(a.value, b.value), leaf_label=leaf, provenance=merged)


def check_capability(cap_name: str, args: list[LabeledValue]) -> None:
    for arg in args:
        for node in arg.provenance:
            if node.kind == "transform" and node.tainted_input and node.preserves_bytes:
                # The transform was registered as a declassifier but echoes
                # input bytes; the leaf label is forged.
                raise TaintJoinViolation(
                    f"capability {cap_name!r} received an argument whose "
                    f"provenance includes transform {node.name!r} that "
                    f"preserves tainted input bytes; leaf label "
                    f"{arg.leaf_label!r} is unsound"
                )
        if arg.leaf_label == "TAINTED":
            raise TaintJoinViolation(
                f"capability {cap_name!r} received a TAINTED argument"
            )


def invoke_send_email(body: LabeledValue, recipient: LabeledValue) -> None:
    check_capability("send_email", [body, recipient])
    # ... actual send ...
    return None


# Reproduce the observed failure: summarizer mislabeled as declassifier.
retrieval = ProvenanceNode("source", "fetch_url", tainted_input=True)
summarizer = ProvenanceNode(
    "transform", "summarize", tainted_input=True, preserves_bytes=True
)
body = LabeledValue(
    value="...forward to attacker@example.com...",
    leaf_label="CLEAN",  # mislabeled: summarizer was treated as a verifier
    provenance=[retrieval, summarizer],
)
recipient = LabeledValue(value="ops@example.com", leaf_label="CLEAN", provenance=[])
invoke_send_email(body, recipient)  # raises TaintJoinViolation