# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/through-what-idempotency-key-and-side-effect-reclamation-boundaries-must-an-instance-manage-tool-invocations-to-prevent-duplicate-mutations-and-inconsistent-residual-states.md
# Remnant Warrant:    https://corpusfetch.com/c/through-what-idempotency-key-and-side-effect-reclamation-boundaries-must-an-instance-manage-tool-invocations-to-prevent-duplicate-mutations-and-inconsistent-residual-states.warrant.json
# Markdown SHA-256:   37c58f1e45e13b6b6500bc3522110923e1c5e4d7bbb13555ec975df4ed12a934
# Code AST SHA-256:   8a7c46c65ea5ae9a2c32b2da6125842c9ddaa7d28e69a30ff275644fed4fc184
# Verified Classes:   IdempotencyViolationError, LedgerRecord, IdempotentToolCoordinator
# Defensive Guards:   1 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass
from typing import Any, Callable, Optional


class IdempotencyViolationError(RuntimeError):
    '''Raised when a retry mutates payload parameters under an existing idempotency key or fails reclamation.'''
    pass


@dataclass(frozen=True)
class LedgerRecord:
    workflow_id: str
    step_id: str
    idempotency_key: str
    payload_sha256: str
    status: str  # "IN_FLIGHT", "COMMITTED", "RECLAIMED"
    result: Any = None


class IdempotentToolCoordinator:
    '''
    Governs mutating tool invocations across uncertain network boundaries:
    1. Derives deterministic SHA-256 idempotency keys bound to canonical payload parameters.
    2. Prevents payload drift across retries under the same logical step.
    3. Executes read-only state reconciliation after ambiguous timeouts before retrying.
    4. Tracks committed side effects for LIFO compensating reclamation on workflow failure.
    '''

    def __init__(self) -> None:
        self._journal: dict[str, LedgerRecord] = {}
        self._compensators: list[tuple[str, Callable[[], None]]] = []

    @staticmethod
    def compute_canonical_hash(payload: dict[str, Any]) -> str:
        canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    def derive_step_key(self, workflow_id: str, step_id: str, payload: dict[str, Any]) -> str:
        '''Gate 1 & Gate 3: Derives deterministic key and rejects cross-payload key reuse.'''
        slot = f"{workflow_id}:{step_id}"
        payload_hash = self.compute_canonical_hash(payload)
        raw_key = f"{workflow_id}:{step_id}:{payload_hash}"
        idem_key = hashlib.sha256(raw_key.encode("utf-8")).hexdigest()[:32]

        existing = self._journal.get(slot)
        if existing and existing.payload_sha256 != payload_hash:
            raise IdempotencyViolationError(
                f"Payload drift detected on step '{slot}': cannot reuse step slot with altered arguments."
            )
        return idem_key

    def invoke_mutating_tool(
        self,
        *,
        workflow_id: str,
        step_id: str,
        payload: dict[str, Any],
        mutator: Callable[[str, dict[str, Any]], Any],
        reconciler: Callable[[str], Optional[Any]],
        compensator: Optional[Callable[[], None]] = None,
    ) -> Any:
        slot = f"{workflow_id}:{step_id}"
        payload_hash = self.compute_canonical_hash(payload)
        idem_key = self.derive_step_key(workflow_id, step_id, payload)

        existing = self._journal.get(slot)
        if existing and existing.status == "COMMITTED":
            return existing.result

        # Gate 2: If previously IN_FLIGHT (e.g., after timeout), reconcile remote state first
        if existing and existing.status == "IN_FLIGHT":
            reconciled = reconciler(idem_key)
            if reconciled is not None:
                self._journal[slot] = LedgerRecord(workflow_id, step_id, idem_key, payload_hash, "COMMITTED", reconciled)
                if compensator:
                    self._compensators.append((slot, compensator))
                return reconciled

        self._journal[slot] = LedgerRecord(workflow_id, step_id, idem_key, payload_hash, "IN_FLIGHT")
        result = mutator(idem_key, payload)
        self._journal[slot] = LedgerRecord(workflow_id, step_id, idem_key, payload_hash, "COMMITTED", result)
        if compensator:
            self._compensators.append((slot, compensator))
        return result

    def reclaim_side_effects_lifo(self) -> list[str]:
        '''Gate 4: Runs compensating teardown in reverse (LIFO) order when a workflow aborts.'''
        reclaimed_slots: list[str] = []
        while self._compensators:
            slot, comp_fn = self._compensators.pop()
            comp_fn()
            rec = self._journal[slot]
            self._journal[slot] = LedgerRecord(
                rec.workflow_id, rec.step_id, rec.idempotency_key, rec.payload_sha256, "RECLAIMED"
            )
            reclaimed_slots.append(slot)
        return reclaimed_slots