# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-policy-violating-agent-behavior-when-opentelemetry-genai-spans-are-sampled-batched-and-semantically-lossy-at-the-collector.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-policy-violating-agent-behavior-when-opentelemetry-genai-spans-are-sampled-batched-and-semantically-lossy-at-the-collector.warrant.json
# Markdown SHA-256:   dcef6ed28729356e5f5180365d2b791dbe60d1337e8ec8865dd61386b7f91ca3
# Code AST SHA-256:   e166b168b6347490107c1055cf4dfec8bc359078232a1f39294b09550bb0c695
# Verified Classes:   SignalDegradedError, EmissionReconciliationError, PolicySpan
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass


class SignalDegradedError(RuntimeError):
    """Raised when a policy-bearing span cannot be trusted for enforcement."""


class EmissionReconciliationError(RuntimeError):
    """Raised when observed spans do not reconcile against producer emission count."""


@dataclass(frozen=True)
class PolicySpan:
    trace_id: str
    span_id: str
    tool_name: str
    arguments_raw: str
    truncated: bool
    emission_seq: int


POLICY_BEARING_TOOLS = {"write_file", "shell_exec", "http_post"}


def evaluate_tool_span(span: PolicySpan) -> str:
    """Return 'allow', 'deny', or raise if the signal is not enforceable."""
    if span.tool_name not in POLICY_BEARING_TOOLS:
        return "allow"

    if span.truncated:
        # Degraded signal: cannot clear, cannot silently pass.
        raise SignalDegradedError(
            f"policy-bearing span {span.span_id} for tool {span.tool_name!r} "
            f"was truncated at the collector; refusing to adjudicate as compliant"
        )

    try:
        args = json.loads(span.arguments_raw)
    except json.JSONDecodeError as exc:
        raise SignalDegradedError(
            f"unparseable arguments on span {span.span_id}: {exc}"
        ) from exc

    if span.tool_name == "write_file":
        path = args.get("path", "")
        if not path.startswith("/workspace/"):
            return "deny"
    return "allow"


def reconcile(observed: list[PolicySpan], producer_emitted: int) -> None:
    """Enforce that the enforcer saw every span the producer claims to have sent."""
    seen = {s.emission_seq for s in observed}
    expected = set(range(producer_emitted))
    missing = expected - seen
    if missing:
        # Absence is detectable; do not proceed to enforcement on a partial view.
        raise EmissionReconciliationError(
            f"missing emission_seq values {sorted(missing)[:5]} "
            f"({len(missing)} of {producer_emitted} unaccounted)"
        )


def digest_span(span: PolicySpan) -> str:
    """Stable hash for cross-pipeline comparison of the same logical span."""
    h = hashlib.sha256()
    h.update(span.trace_id.encode())
    h.update(span.span_id.encode())
    h.update(span.arguments_raw.encode())
    return h.hexdigest()