# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-enforce-egress-policy-for-mcp-tool-calls-when-the-audit-sink-is-asynchronous.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-enforce-egress-policy-for-mcp-tool-calls-when-the-audit-sink-is-asynchronous.warrant.json
# Markdown SHA-256:   705bd75050ebc88ea154cec84cd454475a043563a0c78e62252cfc244d630739
# Code AST SHA-256:   08e1e47a210a26bbe6341b3411edfc50e355defc410a067cfe2e4437c389e983
# Verified Classes:   EgressPolicyViolation, GrantToken
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class EgressPolicyViolation(RuntimeError):
    """Raised when egress is attempted without a valid, non-reused grant token."""


class GrantToken:
    __slots__ = ("tool", "decision", "nonce", "consumed")

    def __init__(self, tool: str, decision: str, nonce: str):
        self.tool = tool
        self.decision = decision
        self.nonce = nonce
        self.consumed = False


_ISSUED: dict[str, GrantToken] = {}


def issue_grant(tool: str, nonce: str, allowed: bool) -> GrantToken:
    tok = GrantToken(tool=tool, decision="ALLOW" if allowed else "DENY", nonce=nonce)
    _ISSUED[nonce] = tok
    return tok


def consume_grant(tool: str, nonce: str) -> None:
    tok = _ISSUED.get(nonce)
    if tok is None:
        raise EgressPolicyViolation(f"no grant issued for nonce={nonce!r} tool={tool!r}")
    if tok.tool != tool:
        raise EgressPolicyViolation(
            f"grant for tool={tok.tool!r} presented to tool={tool!r} (nonce={nonce!r})"
        )
    if tok.consumed:
        raise EgressPolicyViolation(f"grant already consumed for nonce={nonce!r} (retry reuse)")
    if tok.decision != "ALLOW":
        raise EgressPolicyViolation(f"grant decision={tok.decision!r} for nonce={nonce!r}")
    tok.consumed = True


def run_subprocess_tool(nonce: str, argv: list[str]) -> int:
    # Enforcement gate is synchronous and precedes any fork.
    consume_grant(tool="subprocess", nonce=nonce)
    # Outbox row would be written here in the same transaction as the fork;
    # the async flusher only delivers, it does not authorize.
    return 0