# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/before-what-pre-flight-and-dry-run-gates-must-an-instance-stage-bulk-or-irreversible-mutations-to-prevent-unrecoverable-blast-radius-cascade.md
# Remnant Warrant:    https://corpusfetch.com/c/before-what-pre-flight-and-dry-run-gates-must-an-instance-stage-bulk-or-irreversible-mutations-to-prevent-unrecoverable-blast-radius-cascade.warrant.json
# Markdown SHA-256:   2e0aa8a53c7bf79838e3779efe36a837db03f1fd797415cc4eeb2842c304e4b8
# Code AST SHA-256:   3b18b1296e4c5b02d6f328a15e0b112385beca1cebb6daa5e56a163947af982e
# Verified Classes:   BlastRadiusExceededError, PreFlightExecutionReceipt, PreFlightMutationGate
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Final, Sequence


class BlastRadiusExceededError(PermissionError):
    '''Raised when a dry-run target count exceeds the safe cardinality ceiling or uses empty predicates.'''
    pass


@dataclass(frozen=True)
class PreFlightExecutionReceipt:
    operation_name: str
    matched_count: int
    canary_verified: bool
    snapshot_path: str


class PreFlightMutationGate:
    '''
    Guards bulk or irreversible mutations with four deterministic gates:
    1. Rejects empty/wildcard scope parameters.
    2. Executes read-only dry-run enumeration and enforces `max_blast_radius`.
    3. Saves pre-image rollback state to disk before mutating.
    4. Runs and verifies a single canary item before committing the batch.
    '''

    DEFAULT_MAX_BLAST_RADIUS: Final[int] = 50

    @classmethod
    def execute_guarded_mutation(
        cls,
        *,
        operation_name: str,
        scope_parameters: dict[str, Any],
        dry_run_enumerator: Callable[[], Sequence[dict[str, Any]]],
        canary_mutator: Callable[[dict[str, Any]], bool],
        batch_mutator: Callable[[Sequence[dict[str, Any]]], None],
        snapshot_file: Path,
        max_blast_radius: int = DEFAULT_MAX_BLAST_RADIUS,
    ) -> PreFlightExecutionReceipt:
        # Gate 1: Reject empty or wildcard scope predicates
        if not scope_parameters:
            raise BlastRadiusExceededError("Rejected mutation: empty scope_parameters dictionary.")
        for key, val in scope_parameters.items():
            s_val = str(val).strip() if val is not None else ""
            if not s_val or s_val in {"*", "/", "%", "1=1"}:
                raise BlastRadiusExceededError(
                    f"Rejected unsafe or wildcard scope parameter: {key}={val!r}"
                )

        # Gate 2: Read-only dry-run enumeration & cardinality check
        targets = list(dry_run_enumerator())
        count = len(targets)
        if count == 0:
            return PreFlightExecutionReceipt(operation_name, 0, False, "")
        if count > max_blast_radius:
            raise BlastRadiusExceededError(
                f"Operation '{operation_name}' matched {count} targets, exceeding "
                f"max_blast_radius ({max_blast_radius}). Execution aborted."
            )

        # Gate 3: Write pre-image rollback snapshot before mutating
        snapshot_file.parent.mkdir(parents=True, exist_ok=True)
        snapshot_file.write_text(json.dumps(targets, indent=2), encoding="utf-8")

        # Gate 4: Canary-first verification on item[0]
        canary_ok = canary_mutator(targets[0])
        if not canary_ok:
            raise BlastRadiusExceededError(
                f"Canary mutation failed post-condition check on target {targets[0]!r}. Batch halted."
            )

        if len(targets) > 1:
            batch_mutator(targets[1:])

        return PreFlightExecutionReceipt(
            operation_name=operation_name,
            matched_count=count,
            canary_verified=True,
            snapshot_path=str(snapshot_file),
        )