# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/from-what-entropy-and-randomness-sources-must-an-instance-decouple-deterministic-replay-harnesses-to-prevent-heisenbug-nondeterminism.md
# Remnant Warrant:    https://corpusfetch.com/c/from-what-entropy-and-randomness-sources-must-an-instance-decouple-deterministic-replay-harnesses-to-prevent-heisenbug-nondeterminism.warrant.json
# Markdown SHA-256:   cc25813afe683f4d6485131f6946abbdc0e517499c71d5eb3b21efb55e4e9ffd
# Code AST SHA-256:   9eeec2082514c1b516747ddb9fac114549acfb8d3184c538933709cc34c2411c
# Verified Classes:   AmbientEntropyLeakError, HermeticReplayEntropy, ReplayVerificationReceipt, DeterministicReplayGate
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import ast
import random
import uuid
from dataclasses import dataclass
from typing import Any, Callable, Final


class AmbientEntropyLeakError(RuntimeError):
    '''Raised when replayable code calls ambient clock/random/UUID globals or fails double-run bitwise parity.'''
    pass


class HermeticReplayEntropy:
    '''
    Provides a 100% deterministic, isolated replacement for ambient OS time,
    randomness, and UUID4 generation during replay and reproduction runs.
    '''

    def __init__(self, *, seed: int, initial_epoch_ms: int = 1_700_000_000_000) -> None:
        self.seed = seed
        self._rng = random.Random(seed)
        self._now_ms = initial_epoch_ms

    def now_ms(self, advance_ms: int = 0) -> int:
        self._now_ms += advance_ms
        return self._now_ms

    def randint(self, a: int, b: int) -> int:
        return self._rng.randint(a, b)

    def next_uuid(self) -> str:
        bits = self._rng.getrandbits(128)
        return str(uuid.UUID(int=bits, version=4))


@dataclass(frozen=True)
class ReplayVerificationReceipt:
    seed: int
    run_1_output: Any
    run_2_output: Any
    bitwise_deterministic: bool


class DeterministicReplayGate:
    '''
    Prevents Heisenbug nondeterminism in reproduction and replay harnesses:
    1. Scans Python ASTs to block direct calls to ambient `time.time`, `datetime.now`,
       `uuid.uuid4`, `os.urandom`, and module-level `random.random`/`randint`/`choice`.
    2. Executes the target callable twice with fresh `HermeticReplayEntropy(seed)` instances
       and asserts exact output equality (`out_1 == out_2`).
    '''

    FORBIDDEN_AMBIENT_CALLS: Final[frozenset[str]] = frozenset({
        "time.time",
        "time.monotonic",
        "time.perf_counter",
        "datetime.now",
        "datetime.utcnow",
        "uuid.uuid4",
        "uuid.uuid1",
        "os.urandom",
        "random.random",
        "random.randint",
        "random.choice",
        "random.shuffle",
    })

    @classmethod
    def assert_no_ambient_entropy_in_ast(cls, source_code: str) -> None:
        tree = ast.parse(source_code)
        for node in ast.walk(tree):
            if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute):
                if isinstance(node.func.value, ast.Name):
                    qualified = f"{node.func.value.id}.{node.func.attr}"
                    if qualified in cls.FORBIDDEN_AMBIENT_CALLS:
                        raise AmbientEntropyLeakError(
                            f"Line {node.lineno}: Forbidden ambient entropy call '{qualified}()' inside "
                            "deterministic replay module. Inject HermeticReplayEntropy instead."
                        )

    @classmethod
    def verify_double_run_determinism(
        cls,
        *,
        seed: int,
        replay_fn: Callable[[HermeticReplayEntropy], Any],
    ) -> ReplayVerificationReceipt:
        ctx_1 = HermeticReplayEntropy(seed=seed)
        ctx_2 = HermeticReplayEntropy(seed=seed)

        out_1 = replay_fn(ctx_1)
        out_2 = replay_fn(ctx_2)

        if out_1 != out_2:
            raise AmbientEntropyLeakError(
                f"Heisenbug Nondeterminism Detected under seed={seed}: "
                f"Run 1 produced {out_1!r}, but Run 2 produced {out_2!r}."
            )

        return ReplayVerificationReceipt(
            seed=seed,
            run_1_output=out_1,
            run_2_output=out_2,
            bitwise_deterministic=True,
        )