# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-transactional-boundaries-must-an-instance-enforce-idempotency-and-compensating-rollbacks-across-non-atomic-multi-tool-execution-chains.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-transactional-boundaries-must-an-instance-enforce-idempotency-and-compensating-rollbacks-across-non-atomic-multi-tool-execution-chains.warrant.json
# Markdown SHA-256:   cd52a2579436e098818701a636d6583d24cd4ca295ab585b3efa8e055391df88
# Code AST SHA-256:   6bcc5f0f86a7b6cd41241bceb0c2110e4f7cdb78d6015f5bafbf0c7dd03eaf49
# Verified Classes:   UncompensatedSagaFailureError, SagaToolStep, SagaExecutionReceipt, MultiToolSagaCoordinator
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Any, Callable


class UncompensatedSagaFailureError(RuntimeError):
    '''Raised when a multi-tool chain violates Saga ordering or fails during forward/compensating execution.'''
    pass


@dataclass(frozen=True)
class SagaToolStep:
    step_id: str
    is_irreversible_pivot: bool
    forward_action: Callable[[str], Any]
    compensate_action: Callable[[str, Any], None] | None


@dataclass(frozen=True)
class SagaExecutionReceipt:
    saga_id: str
    succeeded: bool
    completed_steps: tuple[str, ...]
    compensated_steps: tuple[str, ...]
    failure_reason: str


class MultiToolSagaCoordinator:
    '''
    Enforces transactional safety across non-atomic multi-tool execution chains:
    1. Rejects chains where an irreversible step precedes a reversible step without compensation,
       or where pre-pivot steps lack `compensate_action`.
    2. Passes deterministic per-step idempotency keys (`{saga_id}:{step_id}`).
    3. Automatically runs reverse-order compensation (`k-1 -> 0`) on pre-pivot failure.
    '''

    @classmethod
    def validate_saga_topology(cls, steps: list[SagaToolStep]) -> None:
        if not steps:
            raise ValueError("Saga step list cannot be empty.")

        seen_pivot = False
        for step in steps:
            if step.is_irreversible_pivot:
                if seen_pivot:
                    raise UncompensatedSagaFailureError(
                        f"Invalid Saga topology: multiple irreversible pivot steps detected at '{step.step_id}'."
                    )
                seen_pivot = True
            else:
                if not seen_pivot and step.compensate_action is None:
                    raise UncompensatedSagaFailureError(
                        f"Pre-pivot step '{step.step_id}' is missing a mandatory compensating rollback action."
                    )

    @classmethod
    def execute_chain(cls, saga_id: str, steps: list[SagaToolStep]) -> SagaExecutionReceipt:
        cls.validate_saga_topology(steps)

        completed: list[tuple[SagaToolStep, Any]] = []
        compensated_ids: list[str] = []

        for step in steps:
            idem_key = f"{saga_id}:{step.step_id}:fwd"
            try:
                result = step.forward_action(idem_key)
                completed.append((step, result))
            except Exception as fwd_err:
                # Gate 4: Reverse-order compensation for all completed reversible steps
                for done_step, done_result in reversed(completed):
                    if done_step.compensate_action is not None:
                        comp_key = f"{saga_id}:{done_step.step_id}:comp"
                        done_step.compensate_action(comp_key, done_result)
                        compensated_ids.append(done_step.step_id)

                return SagaExecutionReceipt(
                    saga_id=saga_id,
                    succeeded=False,
                    completed_steps=tuple(s.step_id for s, _ in completed),
                    compensated_steps=tuple(compensated_ids),
                    failure_reason=f"Step '{step.step_id}' failed ({fwd_err}); compensated {len(compensated_ids)} prior step(s).",
                )

        return SagaExecutionReceipt(
            saga_id=saga_id,
            succeeded=True,
            completed_steps=tuple(s.step_id for s, _ in completed),
            compensated_steps=(),
            failure_reason="",
        )