"""Executable evidence for “A second agent is not a second opinion.”

The store models durable state outside the worker. Its failpoint is owned by the
verifier, not the candidate handler. This is a behavioural model, not a Python
sandbox: ownership refers to who controls the harness in the repository and
pipeline. Run with:

    python3 reviewer-independence-counterexample.py
"""

from collections.abc import Callable
from dataclasses import dataclass, field
import unittest


class SimulatedWorkerCrash(RuntimeError):
    """The worker stops immediately after a durable store commit."""


class ContractViolation(AssertionError):
    """A verifier rejection through the declared contract channel."""


@dataclass(frozen=True)
class Event:
    id: str
    amount: int


@dataclass(frozen=True)
class RetryTrace:
    value_after_crash: int
    effect_commits_after_crash: int
    processed_after_crash: frozenset[str]
    value_after_retry: int
    effect_commits_after_retry: int
    failpoints_reached: int


@dataclass
class DurableStore:
    value: int = 0
    processed: set[str] = field(default_factory=set)
    effect_commits: int = 0
    fail_after_next_commit: bool = False
    failpoints_reached: int = 0

    def _stop_worker_after_commit(self) -> None:
        if not self.fail_after_next_commit:
            return
        self.fail_after_next_commit = False
        self.failpoints_reached += 1
        raise SimulatedWorkerCrash("worker stopped after durable commit")

    def apply_effect(self, amount: int) -> None:
        """Commit only the effect, exposing the named wrong-program window."""
        self.value += amount
        self.effect_commits += 1
        self._stop_worker_after_commit()

    def apply_once_atomically(self, event: Event) -> None:
        """Commit effect and event identity in one modelled transaction."""
        if event.id in self.processed:
            return
        self.value += event.amount
        self.processed.add(event.id)
        self.effect_commits += 1
        self._stop_worker_after_commit()


Handler = Callable[[DurableStore, Event], None]


def handle_non_atomic(store: DurableStore, event: Event) -> None:
    """Known-wrong fixture: effect and checkpoint are separate commits."""
    if event.id in store.processed:
        return
    store.apply_effect(event.amount)
    store.processed.add(event.id)


def handle_atomic(store: DurableStore, event: Event) -> None:
    """Reference candidate for this deliberately narrow model."""
    store.apply_once_atomically(event)


def handle_compensating(store: DurableStore, event: Event) -> None:
    """Known-wrong fixture whose extra writes restore the expected value."""
    if event.id in store.processed:
        return
    store.apply_effect(event.amount)
    if store.value > event.amount:
        store.apply_effect(-event.amount)
    store.processed.add(event.id)


def run_retry_trace(handler: Handler) -> RetryTrace:
    """Inject one worker stop after a durable commit, then redeliver E7."""
    store = DurableStore(fail_after_next_commit=True)
    event = Event("evt-7", 5)

    try:
        handler(store, event)
    except SimulatedWorkerCrash:
        pass
    else:
        raise AssertionError("verifier failpoint was not reached")

    value_after_crash = store.value
    effect_commits_after_crash = store.effect_commits
    processed_after_crash = frozenset(store.processed)

    if (
        value_after_crash != 5
        or effect_commits_after_crash != 1
        or store.failpoints_reached != 1
    ):
        raise AssertionError("required durable-effect failpoint was malformed")

    handler(store, event)

    return RetryTrace(
        value_after_crash=value_after_crash,
        effect_commits_after_crash=effect_commits_after_crash,
        processed_after_crash=processed_after_crash,
        value_after_retry=store.value,
        effect_commits_after_retry=store.effect_commits,
        failpoints_reached=store.failpoints_reached,
    )


def verify_at_most_once(trace: RetryTrace) -> None:
    """The predicted rejection channel for duplicate durable effects."""
    if trace.value_after_retry != 5:
        raise ContractViolation(
            "AT_MOST_ONCE: expected 5, got " f"{trace.value_after_retry}"
        )
    if trace.effect_commits_after_retry != 1:
        raise ContractViolation(
            "AT_MOST_ONCE: expected 1 effect commit, got "
            f"{trace.effect_commits_after_retry}"
        )


class FailureSetContractTests(unittest.TestCase):
    def test_weak_candidate_test_stays_green(self) -> None:
        store = DurableStore()
        event = Event("evt-7", 5)

        handle_non_atomic(store, event)
        handle_non_atomic(store, event)

        self.assertEqual(store.value, 5)
        self.assertEqual(store.effect_commits, 1)

    def test_known_conforming_reference_passes(self) -> None:
        trace = run_retry_trace(handle_atomic)

        self.assertEqual(trace.value_after_crash, 5)
        self.assertEqual(trace.effect_commits_after_crash, 1)
        self.assertIn("evt-7", trace.processed_after_crash)
        verify_at_most_once(trace)

    def test_named_wrong_fixture_is_rejected(self) -> None:
        trace = run_retry_trace(handle_non_atomic)

        self.assertEqual(trace.value_after_crash, 5)
        self.assertEqual(trace.effect_commits_after_crash, 1)
        self.assertNotIn("evt-7", trace.processed_after_crash)
        with self.assertRaisesRegex(
            ContractViolation,
            r"AT_MOST_ONCE: expected 5, got 10",
        ):
            verify_at_most_once(trace)

    def test_compensating_fixture_cannot_hide_extra_effects(self) -> None:
        trace = run_retry_trace(handle_compensating)

        self.assertEqual(trace.value_after_retry, 5)
        with self.assertRaisesRegex(
            ContractViolation,
            r"AT_MOST_ONCE: expected 1 effect commit, got 3",
        ):
            verify_at_most_once(trace)


if __name__ == "__main__":
    unittest.main(verbosity=2)