# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-a-taint-tracking-boundary-failure-at-the-policy-enforcement-point-of-an-agent-runtime.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-a-taint-tracking-boundary-failure-at-the-policy-enforcement-point-of-an-agent-runtime.warrant.json
# Markdown SHA-256:   496f7ae77539dca5c83147ba2b0d821f1223342b7149d5304f8abfdbc2140794
# Code AST SHA-256:   068682b605d96d7297ba801903fd98d380c7a1a688486efcbc5f8c7bd142afde
# Verified Classes:   TaintBoundaryError, LabeledValue, TaintLedger
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Any, Dict, Set


class TaintBoundaryError(RuntimeError):
    """Raised when a value reaches the egress PEP without a traceable taint label."""
    pass


@dataclass
class LabeledValue:
    value: Any
    sources: Set[str] = field(default_factory=set)

    def is_clean(self) -> bool:
        return not self.sources


class TaintLedger:
    """Tracks which sources are known for the current request."""

    def __init__(self) -> None:
        self.registered: Set[str] = set()
        self.egress_decisions: Dict[str, Set[str]] = {}

    def register_source(self, source_id: str) -> None:
        self.registered.add(source_id)

    def record_decision(self, decision_id: str, cited_sources: Set[str]) -> None:
        self.egress_decisions[decision_id] = set(cited_sources)


def propagate_through_transform(
    ledger: TaintLedger,
    labeled: LabeledValue,
    transform_name: str,
) -> LabeledValue:
    """
    Any transformation must either preserve taint or be declared a sink that
    forces TAINTED_UNKNOWN. Silent erasure is the failure mode we forbid.
    """
    if labeled.sources:
        # Conservative: transformation cannot launder taint, only carry it.
        return LabeledValue(value=labeled.value, sources=set(labeled.sources))
    # A clean value transformed by an unmodeled primitive is not trustworthy.
    # Force the most restrictive label rather than returning clean.
    return LabeledValue(value=labeled.value, sources={"TAINTED_UNKNOWN"})


def egress_check(
    ledger: TaintLedger,
    decision_id: str,
    labeled: LabeledValue,
    allowlist: Set[str],
    target: str,
) -> bool:
    if target not in allowlist:
        return False

    # Structural signal 3: verdict rationale must link back to the ingress ledger.
    orphaned = labeled.sources - ledger.registered - {"TAINTED_UNKNOWN"}
    if orphaned:
        raise TaintBoundaryError(
            f"egress decision {decision_id} cites sources {sorted(orphaned)} "
            f"not present in ingress ledger; label-verdict decoupled"
        )

    if not labeled.is_clean():
        # Tainted value attempting egress through an allowlisted tool.
        raise TaintBoundaryError(
            f"egress decision {decision_id} denied: value carries taint "
            f"{sorted(labeled.sources)}"
        )

    ledger.record_decision(decision_id, labeled.sources)
    return True


def run_agent_egress(ledger: TaintLedger, secret_summary: LabeledValue) -> None:
    # Simulate: tool-returned secret -> summary string -> re-parsed dict.
    # The round-trip is the erasing transformation.
    transformed = propagate_through_transform(ledger, secret_summary, "json_roundtrip")
    if not egress_check(
        ledger,
        decision_id="d-001",
        labeled=transformed,
        allowlist={"http_post"},
        target="http_post",
    ):
        return
    raise TaintBoundaryError("unreachable: tainted value should not pass")


if __name__ == "__main__":
    ledger = TaintLedger()
    ledger.register_source("tool:fetch_secret")
    secret = LabeledValue(value="sk-live-xxxx", sources={"tool:fetch_secret"})
    run_agent_egress(ledger, secret)