# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-of-a-tainted-provenance-label-is-laundered-through-a-tool-that-returns-structured-data.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-of-a-tainted-provenance-label-is-laundered-through-a-tool-that-returns-structured-data.warrant.json
# Markdown SHA-256:   5493de3494ab9c1019cfbb911690147731c7be66230b1b41ff072256b2abcdcb
# Code AST SHA-256:   3ea27ba8352ca305b5e46e149f6f2e7a1bf4afe3c5add84a744045cddbf7fbdb
# Verified Classes:   ProvenanceLaunderingError, Span
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Any


class ProvenanceLaunderingError(RuntimeError):
    """Raised when a tool boundary resets taint instead of joining it."""


@dataclass(frozen=True)
class Span:
    content: str
    taint: frozenset[str] = field(default_factory=frozenset)  # set of taint sources


def join_taint(*spans: Span) -> frozenset[str]:
    out: set[str] = set()
    for s in spans:
        out |= s.taint
    return frozenset(out)


def propagate_across_tool_boundary(
    inputs: list[Span],
    tool_output: Span,
    tool_is_first_party: bool,
    fetched_external: list[Span] | None = None,
) -> Span:
    """Return the output span with taint = join(inputs, fetched_external).

    Tool identity is deliberately ignored: first-party status contributes
    zero trust. Declassification must happen elsewhere, explicitly.
    """
    expected = join_taint(*inputs, *(fetched_external or []))
    if tool_is_first_party and tool_output.taint == frozenset():
        # The deserializer assigned empty taint because the tool is trusted.
        # That is exactly the laundering condition: schema-valid output from
        # a trusted tool whose inputs were tainted, with taint dropped.
        if expected:
            raise ProvenanceLaunderingError(
                f"tool boundary dropped taint: inputs carried {sorted(expected)}, "
                f"output carried none; refusing to re-enter context as trusted"
            )
    if not tool_output.taint.issuperset(expected):
        raise ProvenanceLaunderingError(
            f"output taint {sorted(tool_output.taint)} does not cover "
            f"joined input taint {sorted(expected)}"
        )
    return Span(content=tool_output.content, taint=expected)


def ingest_tool_result(
    inputs: list[Span],
    raw_output: dict[str, Any],
    tool_is_first_party: bool,
    fetched_external: list[Span] | None = None,
) -> Span:
    # Deserializer default: trust by tool class, not by dataflow. This is
    # the bug we are guarding against.
    declared = frozenset() if tool_is_first_party else frozenset({"untrusted"})
    out = Span(content=str(raw_output.get("result", "")), taint=declared)
    return propagate_across_tool_boundary(
        inputs=inputs,
        tool_output=out,
        tool_is_first_party=tool_is_first_party,
        fetched_external=fetched_external,
    )