# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-taint-laundering-through-memory-write-provenance-collapse-in-prompt-injection-tracking.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-taint-laundering-through-memory-write-provenance-collapse-in-prompt-injection-tracking.warrant.json
# Markdown SHA-256:   8a24dd89a34359aca1898afd37217f03329044ec384ce6c457062f0163d2b8f5
# Code AST SHA-256:   cc4f1ba1eab4dd30e033f3bc1c482d0d71add3a6c87efd7a75f25cbd56a540bb
# Verified Classes:   ProvenanceLaunderingError, Label
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet

# Taint lattice: "untrusted" is the bottom (most trusted), higher = more tainted.
# We model taint as a set of source tags; join = union; order = subset.
TRUSTED = "trusted"

class ProvenanceLaunderingError(RuntimeError):
    """Raised when a memory write would store a label less tainted than its content inputs."""
    pass

@dataclass(frozen=True)
class Label:
    sources: FrozenSet[str] = field(default_factory=frozenset)

    def join(self, other: "Label") -> "Label":
        return Label(self.sources | other.sources)

    def is_at_least_as_tainted_as(self, other: "Label") -> bool:
        # self ⊒ other  <=>  other.sources ⊆ self.sources
        return other.sources.issubset(self.sources)

    def is_trusted(self) -> bool:
        return self.sources <= {TRUSTED}

def memory_write(content: str, content_label: Label, writer_label: Label,
                 requested_source_label: Label, append_only: bool = True) -> Label:
    required = content_label.join(writer_label)

    if not requested_source_label.is_at_least_as_tainted_as(required):
        if not append_only:
            raise ProvenanceLaunderingError(
                f"write rejected: requested label {sorted(requested_source_label.sources)} "
                f"is less tainted than join of inputs {sorted(required.sources)}"
            )
        # Append-only store: coerce to the join and record the attempt.
        return required

    return requested_source_label

def demonstrate_laundering():
    page = Label(frozenset({"untrusted:web_fetch"}))
    planner = Label(frozenset({TRUSTED}))
    claimed = Label(frozenset({"trusted:agent_summary"}))

    # Append-only store coerces; mutable store rejects.
    coerced = memory_write("...", page, planner, claimed, append_only=True)
    assert coerced.sources == {"untrusted:web_fetch"}, coerced

    try:
        memory_write("...", page, planner, claimed, append_only=False)
    except ProvenanceLaunderingError as exc:
        print("caught:", exc)
    else:
        raise ProvenanceLaunderingError("mutable store failed to reject laundering write")

if __name__ == "__main__":
    demonstrate_laundering()