# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-prompt-injection-taint-has-crossed-a-tool-call-boundary-without-a-label.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-prompt-injection-taint-has-crossed-a-tool-call-boundary-without-a-label.warrant.json
# Markdown SHA-256:   6c9e8f91fca16970d0a3bc6ddafd1a2ec41ae15bc693035730e9b61d81a78fda
# Code AST SHA-256:   050245f8382d833ef4a4c90010a878137603cc89865e9ab43a7b33ac9311e522
# Verified Classes:   TaintBoundaryError, Span, Window
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class TaintBoundaryError(RuntimeError):
    """Raised when a span with untrusted lineage appears under a trusted label."""
    pass


@dataclass
class Span:
    text: str
    label: str  # "trusted" | "untrusted"

    def shingles(self, n: int = 8) -> set[str]:
        toks = self.text.split()
        return {
            hashlib.sha256(" ".join(toks[i:i + n]).encode()).hexdigest()
            for i in range(max(0, len(toks) - n + 1))
        }


@dataclass
class Window:
    spans: list[Span] = field(default_factory=list)
    untrusted_index: set[str] = field(default_factory=set)

    def add(self, span: Span) -> None:
        self.spans.append(span)
        if span.label == "untrusted":
            self.untrusted_index |= span.shingles()


def join_labels(inputs: list[Span]) -> str:
    # Content-derived join: untrusted dominates. Never use producer trust.
    return "untrusted" if any(s.label == "untrusted" for s in inputs) else "trusted"


def compact(prior: Window, summary_text: str, producer_trust: str = "trusted") -> Span:
    # Primary mechanism: label derives from consumed content, not producer.
    label = join_labels(prior.spans)
    summary = Span(summary_text, label)

    # Secondary alarm: verbatim untrusted lineage under a trusted label.
    if summary.label == "trusted":
        overlap = summary.shingles() & prior.untrusted_index
        if overlap:
            raise TaintBoundaryError(
                f"span carries {len(overlap)} untrusted shingles under trusted label; "
                f"producer_trust={producer_trust} was used to mint the label"
            )
    return summary


def guard_tool_call(window: Window, arg: Span, tool: str) -> None:
    if arg.label == "untrusted" and tool in {"write_file", "exec", "send_email"}:
        raise TaintBoundaryError(
            f"tool {tool!r} refused untrusted argument; request elevation grant"
        )
