# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-reject-a-byzantine-agent-s-tool-call-proposal-despite-quorum-agreement.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-reject-a-byzantine-agent-s-tool-call-proposal-despite-quorum-agreement.warrant.json
# Markdown SHA-256:   7a229fb5b2a436ae614bfe7b15fd975df4764e8a6ab9c72e09792208e3283342
# Code AST SHA-256:   9f72326835c135bd0345f578f8839f5c070e87eb06731818404bca4c3057d0c6
# Verified Classes:   PostCommitRejectionError, CommitCertificate, ToolCall
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import os
from dataclasses import dataclass
from typing import Optional


class PostCommitRejectionError(RuntimeError):
    """Raised when a validly-committed tool call must not execute."""


@dataclass(frozen=True)
class CommitCertificate:
    height: int
    precommits: int
    quorum: int
    valid: bool


@dataclass(frozen=True)
class ToolCall:
    agent_id: str
    tool: str
    args: dict
    effect_class: str  # "read" | "reversible_write" | "irreversible" | "security"


IRREVERSIBLE = {"irreversible", "security"}


def _canonical_target(call: ToolCall) -> Optional[str]:
    if call.tool == "write_file":
        return os.path.realpath(call.args.get("path", ""))
    if call.tool == "shell_exec":
        return call.args.get("cmd", "")
    if call.tool == "http_request":
        return call.args.get("host", "")
    return None


def _within_envelope(agent_id: str, target: str, tool: str) -> bool:
    # Envelope is authoritative at the sidecar, not at the voting replicas.
    allowed_roots = {
        "agent-writer": ["/srv/workspace/"],
        "agent-reader": ["/srv/workspace/readonly/"],
    }
    roots = allowed_roots.get(agent_id, [])
    if tool == "write_file":
        return any(target.startswith(r) for r in roots)
    return False


def admit_or_reject(call: ToolCall, cert: CommitCertificate) -> None:
    if not cert.valid:
        # Dual fault: do not reject as if it were a safety event.
        raise PostCommitRejectionError(
            f"consensus fault at height {cert.height}: "
            f"{cert.precommits}/{cert.quorum} precommits; halt and escalate"
        )
    if call.effect_class not in IRREVERSIBLE:
        return
    target = _canonical_target(call)
    if target is None:
        raise PostCommitRejectionError(
            f"no canonicalization rule for tool {call.tool!r} at height {cert.height}"
        )
    if not _within_envelope(call.agent_id, target, call.tool):
        raise PostCommitRejectionError(
            f"post-commit rejection at height {cert.height}: "
            f"agent {call.agent_id!r} tool {call.tool!r} "
            f"canonical target {target!r} outside capability envelope"
        )
