# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-invalidate-a-taint-label-the-label-laundering-boundary-in-dynamic-ift-for-prompt-injection.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-invalidate-a-taint-label-the-label-laundering-boundary-in-dynamic-ift-for-prompt-injection.warrant.json
# Markdown SHA-256:   bf758bb207f706fb3483f03638589c260ce508318437c5be4e31ea1ed73b1c67
# Code AST SHA-256:   04fcca0235b354e30424cf3a9edfeb5d83f17c3504d2754df2062b248f8ba3ba
# Verified Classes:   TaintLaunderingError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class TaintLaunderingError(RuntimeError):
    """Raised when a value derived from a tainted source reaches a sink
    without a label-preserving write path or authorized declassification."""
    pass


REGISTERED_LABEL_PRESERVING = {
    "identity",
    "json_field_extract",   # preserves label of the source document
    "string_format",
}

AUTHORIZED_DECLASSIFIERS = {
    # (transform, source_label_class) -> allowed to drop that label class
    ("int_parse", "untrusted:web"): True,
}


def write_with_provenance(sink, key, value, input_labels, transform):
    """Write `value` to `sink[key]` only if the write preserves the superset
    invariant over `input_labels`. `transform` names the transformation that
    produced `value` from its tainted inputs."""
    if not hasattr(sink, "labels") or not hasattr(sink, "write_path"):
        raise TaintLaunderingError(
            f"sink for key={key!r} has no label channel; write path cannot "
            f"preserve provenance for input_labels={sorted(input_labels)}"
        )

    dropped = set()
    for label in input_labels:
        if transform in REGISTERED_LABEL_PRESERVING:
            continue
        if AUTHORIZED_DECLASSIFIERS.get((transform, label)):
            dropped.add(label)
            continue
        # Unregistered transform over a tainted input: cannot certify clean.
        raise TaintLaunderingError(
            f"unregistered transform {transform!r} over tainted input "
            f"label={label!r}; refusing to write key={key!r} (would launder)"
        )

    prior = sink.labels.get(key, set())
    # Superset invariant: new label set must cover prior ∪ inputs minus
    # only explicitly authorized declassifications.
    required = (prior | input_labels) - dropped
    sink.labels[key] = sink.labels.get(key, set()) | required
    sink.write_path[key] = transform
    sink.store[key] = value
    return sink.labels[key]


def assemble_prompt_region(sink, key, region_authority):
    """Read a memory record into a prompt region, honoring write-time
    provenance rather than the label visible at read time."""
    labels = sink.labels.get(key, set())
    write_path = sink.write_path.get(key)
    if write_path is None:
        # Record exists with no recorded write path: treat as untrusted.
        labels = labels | {"untrusted:unknown_write_path"}
    if region_authority == "system" and labels:
        raise TaintLaunderingError(
            f"refusing to splice tainted record key={key!r} "
            f"(labels={sorted(labels)}, write_path={write_path!r}) "
            f"into system-authority region"
        )
    return sink.store.get(key), labels