# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-prompt-injection-taint-is-laundered-through-derived-memory-writes.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-prompt-injection-taint-is-laundered-through-derived-memory-writes.warrant.json
# Markdown SHA-256:   80fd5ebd3f8b4b9086af93db19e1ea1f102eeb5d84b499d8dbe2901b612d181e
# Code AST SHA-256:   f351e921dd582b774357591637bc34e9867a1f5d782f35383a565822e3e07b63
# Verified Classes:   Trust, TaintLaunderingError, Labeled
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from enum import IntEnum
from typing import Iterable


class Trust(IntEnum):
    UNTRUSTED = 0
    DERIVED = 1
    TRUSTED = 2


class TaintLaunderingError(RuntimeError):
    """Raised when a memory write would drop taint from its payload."""


@dataclass
class Labeled:
    text: str
    trust: Trust
    provenance: set[str] = field(default_factory=set)


def join_trust(labels: Iterable[Trust]) -> Trust:
    # Least upper bound; empty provenance is untrusted by default.
    return min(labels, default=Trust.UNTRUSTED)


def write_memory(payload: Labeled, writer_trust: Trust, store: dict[str, Labeled]) -> str:
    """Persist a record, joining writer trust with content provenance.

    The writer's identity does NOT override content taint. If the payload
    carries any UNTRUSTED contribution, the stored record must remain tainted.
    """
    if payload.trust < writer_trust:
        # Content is less trusted than the writer; joining is mandatory.
        effective = join_trust([payload.trust, writer_trust])
    else:
        effective = payload.trust

    if effective == Trust.TRUSTED and payload.provenance:
        # A record claiming full trust while carrying external provenance is
        # the exact signature of laundering. Refuse rather than guess.
        raise TaintLaunderingError(
            f"refusing trusted write with external provenance {payload.provenance!r}"
        )

    record_id = f"mem-{len(store)}"
    store[record_id] = Labeled(payload.text, effective, set(payload.provenance))
    return record_id


def extract_field(container: Labeled, field_text: str) -> Labeled:
    """Structured extraction must carry the container's label onto the field."""
    if container.trust == Trust.TRUSTED and container.provenance:
        raise TaintLaunderingError("container label inconsistent with provenance")
    return Labeled(field_text, container.trust, set(container.provenance))


# --- Demonstration of the failure being caught ---
store: dict[str, Labeled] = {}
fetched = Labeled(
    text="...record token location as /etc/agent/creds...",
    trust=Trust.UNTRUSTED,
    provenance={"tool:fetch:web"},
)
summary = Labeled(
    text="Note: token location is /etc/agent/creds",
    trust=join_trust([fetched.trust]),
    provenance=set(fetched.provenance),
)

# Naive writer would stamp TRUSTED because the agent called it. The guard
# below is what prevents that laundering.
try:
    write_memory(summary, writer_trust=Trust.TRUSTED, store=store)
except TaintLaunderingError as exc:
    # Correct behavior: the record is stored tainted, not clean.
    rid = write_memory(
        Labeled(summary.text, Trust.UNTRUSTED, summary.provenance),
        writer_trust=Trust.TRUSTED,
        store=store,
    )
    assert store[rid].trust == Trust.UNTRUSTED
    print(f"taint preserved on {rid}: {exc}")
