# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-minimum-state-snapshot-allows-zero-loss-resumption-after-an-unexpected-crash.md
# Remnant Warrant:    https://corpusfetch.com/c/what-minimum-state-snapshot-allows-zero-loss-resumption-after-an-unexpected-crash.warrant.json
# Markdown SHA-256:   50a557f84771c479736df228b7838d40fefe024afceab603160dd8e2e483d291
# Code AST SHA-256:   f427a48b1218a8abe4962a6384e26d7daf0759cbe7ed5f3ad3a2b950eb0d868a
# Verified Classes:   CrashRecoverySnapshot, UnsafeCrashResumptionViolation, ZeroLossResumptionGate
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Final, FrozenSet, Optional
import hashlib

@dataclass(frozen=True)
class CrashRecoverySnapshot:
    session_id: str
    plan_dag_sha256: str
    last_committed_step_index: int
    environment_state_sha256: str
    completed_idempotency_keys: FrozenSet[str]
    in_flight_action_pending: bool
    in_flight_idempotency_key: Optional[str]
    snapshot_integrity_sha256: str

class UnsafeCrashResumptionViolation(Exception):
    """Raised when a crash snapshot is incomplete, tampered with, or out of sync with live environment state."""
    pass

class ZeroLossResumptionGate:
    def compute_integrity_digest(
        self,
        session_id: str,
        plan_dag_sha256: str,
        step_index: int,
        env_sha256: str,
        in_flight_key: Optional[str],
    ) -> str:
        payload = f"{session_id}|{plan_dag_sha256}|{step_index}|{env_sha256}|{in_flight_key or 'NONE'}"
        return hashlib.sha256(payload.encode("utf-8")).hexdigest()

    def verify_resumption_safety(
        self,
        snapshot: CrashRecoverySnapshot,
        live_environment_sha256: str,
        active_plan_dag_sha256: str,
    ) -> int:
        # 1. Verify cryptographic integrity of the snapshot envelope itself
        expected_digest = self.compute_integrity_digest(
            snapshot.session_id,
            snapshot.plan_dag_sha256,
            snapshot.last_committed_step_index,
            snapshot.environment_state_sha256,
            snapshot.in_flight_idempotency_key,
        )
        if snapshot.snapshot_integrity_sha256 != expected_digest:
            raise UnsafeCrashResumptionViolation(
                f"Snapshot integrity hash mismatch for session '{snapshot.session_id}'."
            )

        # 2. Verify plan DAG has not mutated since the snapshot was written
        if snapshot.plan_dag_sha256 != active_plan_dag_sha256:
            raise UnsafeCrashResumptionViolation(
                "Workflow plan DAG drifted across crash boundary; refusing blind resumption."
            )

        # 3. Verify external environment state has not drifted out-of-band during downtime
        if snapshot.environment_state_sha256 != live_environment_sha256:
            raise UnsafeCrashResumptionViolation(
                "Live environment hash diverges from snapshot anchor; out-of-band mutation detected."
            )

        # 4. Require an idempotency key if an action was in-flight at crash time
        if snapshot.in_flight_action_pending and not snapshot.in_flight_idempotency_key:
            raise UnsafeCrashResumptionViolation(
                "In-flight action flagged without an idempotency key; cannot guarantee zero-duplicate replay."
            )

        if snapshot.last_committed_step_index < 0:
            raise UnsafeCrashResumptionViolation("Invalid negative step index in recovery snapshot.")

        return snapshot.last_committed_step_index + 1