# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-treats-mcp-elicitation-as-a-synchronous-confirmation-rather-than-a-durable-capability-grant.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-treats-mcp-elicitation-as-a-synchronous-confirmation-rather-than-a-durable-capability-grant.warrant.json
# Markdown SHA-256:   f23cc9b3844db52e6de3e3b3c0f628c273e1328f21b299a00ad2e196e6345fcd
# Code AST SHA-256:   de470ececb94795d99f728cb50800a1652f922219c610ff61bca027357db3b38
# Verified Classes:   GrantViolation, Grant, GrantStore
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
import time
from dataclasses import dataclass
from typing import Optional


class GrantViolation(RuntimeError):
    """Raised when an execution attempt does not hold a valid, unconsumed grant."""
    pass


@dataclass
class Grant:
    grant_id: str
    arg_hash: str
    expires_at: float
    consumed_at: Optional[float] = None


class GrantStore:
    """Server-side authority for elicitation-derived approvals.

    The elicitation response content carries only grant_id and arg_hash.
    Authority lives here. Consumption is atomic with the side effect.
    """

    def __init__(self) -> None:
        self._grants: dict[str, Grant] = {}

    def issue(self, grant_id: str, args: dict, ttl_seconds: float) -> str:
        arg_hash = hashlib.sha256(
            repr(sorted(args.items())).encode()
        ).hexdigest()
        self._grants[grant_id] = Grant(
            grant_id=grant_id,
            arg_hash=arg_hash,
            expires_at=time.monotonic() + ttl_seconds,
        )
        return arg_hash

    def consume(self, grant_id: str, args: dict, now: Optional[float] = None) -> None:
        now = time.monotonic() if now is None else now
        grant = self._grants.get(grant_id)
        if grant is None:
            raise GrantViolation(f"unknown grant_id={grant_id!r}")
        if grant.consumed_at is not None:
            raise GrantViolation(
                f"grant_id={grant_id!r} already consumed at {grant.consumed_at}"
            )
        if now >= grant.expires_at:
            raise GrantViolation(
                f"grant_id={grant_id!r} expired at {grant.expires_at}"
            )
        arg_hash = hashlib.sha256(
            repr(sorted(args.items())).encode()
        ).hexdigest()
        if arg_hash != grant.arg_hash:
            raise GrantViolation(
                f"grant_id={grant_id!r} argument hash mismatch: "
                f"approved {grant.arg_hash[:12]} got {arg_hash[:12]}"
            )
        grant.consumed_at = now


def execute_destructive(store: GrantStore, grant_id: str, args: dict) -> None:
    """Worker-side entry point. Fails closed on retry, expiry, or arg drift."""
    store.consume(grant_id, args)
    # Side effect proceeds only after atomic consumption. A retry after this
    # point raises GrantViolation rather than re-executing.
    _perform_delete(args["path"])


def _perform_delete(path: str) -> None:
    # Placeholder for the actual destructive backend call.
    pass