from __future__ import annotations

import hashlib
import json
import os
import pathlib
import re
import subprocess
import sys
import tempfile
import time
import unittest
from unittest import mock

from reliability_core.stage_control_r10r3r2 import (
    KNOWN_FAILURE_TEST_IDS,
    NON_REUSABLE_EVIDENCE,
    REPEAT_LADDER,
    STAGE_NAMES,
    DeadlineCalibration,
    DeadlineHierarchy,
    allocate_deadlines,
    canary_selection,
    checkpoint_payload,
    cleanup_wait_seconds,
    environment_sha256,
    execution_cost_p95,
    group_admission,
    observed_execution_seconds,
    validate_checkpoint,
)


ROOT = pathlib.Path(__file__).resolve().parents[1]


def calibration() -> DeadlineCalibration:
    return DeadlineCalibration.from_samples(
        startup=(0.1, 0.2, 0.3),
        execution=(1.0, 2.0, 3.0),
        drain=(0.01, 0.02, 0.03),
        cleanup=(0.1, 0.2, 0.3),
        environment_sha256="a" * 64,
        observed_at_epoch=1.0,
    )


class R10R3R2Tests(unittest.TestCase):
    def test_default_rollback_drill_work_stays_outside_source_tree(self) -> None:
        from scripts import build_evidence as evidence

        with tempfile.TemporaryDirectory(prefix="r10r3-rollback-") as directory:
            with mock.patch.dict(os.environ, {"WRLAB_RUNTIME_ARTIFACT_ROOT": directory}):
                proof = evidence.run_canary_rollback_drill(
                    ROOT,
                    restore_command=[os.sys.executable, "run_all.py", "--child-smoke"],
                )
        self.assertTrue(proof["passed"])
        self.assertEqual(proof["runtime_artifact_root"], str(pathlib.Path(directory).resolve()))
        self.assertIs(proof["source_tree_written"], False)

    def test_rg08_database_artifacts_stay_outside_source_tree(self) -> None:
        from reliability_core.regressions import run_rg

        with tempfile.TemporaryDirectory(prefix="r10r3-rg08-") as directory:
            with mock.patch.dict(os.environ, {"WRLAB_RUNTIME_ARTIFACT_ROOT": directory}):
                result = run_rg("RG-08", seed=404)
        self.assertTrue(result.passed)
        self.assertEqual(result.observed["runtime_artifact_root"], str(pathlib.Path(directory).resolve()))
        self.assertIs(result.observed["source_tree_written"], False)

    def test_process_restart_artifacts_stay_outside_source_tree(self) -> None:
        from reliability_core.harness import run_process_restart

        with tempfile.TemporaryDirectory(prefix="r10r3-runtime-") as directory:
            with mock.patch.dict(os.environ, {"WRLAB_RUNTIME_ARTIFACT_ROOT": directory}):
                result = run_process_restart(404)
        self.assertTrue(result["passed"])
        self.assertEqual(result["runtime_artifact_root"], str(pathlib.Path(directory).resolve()))
        self.assertIs(result["source_tree_written"], False)

    def test_r5_real_execution_run_ids_are_unique(self) -> None:
        from reliability_core.regressions_r5 import run_r5_rg

        first = run_r5_rg("R5-RG-03")
        second = run_r5_rg("R5-RG-03")
        self.assertNotEqual(first.run_id, second.run_id)

    def test_r8_twenty_reuses_only_one_static_source_digest(self) -> None:
        from reliability_core import regressions_r8
        from reliability_core.regressions_r5 import R5Result

        fake = R5Result(
            test_id="R5-RG-03", run_id="fresh", passed=True,
            assertion_id="a", assertion_contract_id="c", assertion="observed",
            observed_fields=[], expected_predicate="true", observed={}, latency_ms=1.0,
        )
        with mock.patch.object(regressions_r8, "normalized_tree_hash", return_value="a" * 64) as digest:
            with mock.patch.object(regressions_r8, "run_r5_rg", return_value=fake) as execute:
                rows = regressions_r8._r5_twenty()
        self.assertEqual(len(rows), 20)
        digest.assert_called_once_with(ROOT)
        self.assertEqual(execute.call_count, 20)
        self.assertTrue(all(call.kwargs["source_tree_sha256"] == "a" * 64 for call in execute.call_args_list))

    def test_r5_ready_delay_probe_avoids_unrelated_process_tree(self) -> None:
        from reliability_core import regressions_r5
        from reliability_core.regressions_r5 import run_r5_rg

        result = run_r5_rg("R5-RG-03")
        ready_payload = result.observed["ready"]["payload"]
        self.assertTrue(result.passed)
        self.assertIs(ready_payload["single_process_ready"], True)
        self.assertEqual(result.observed["tracked_descendant_count"], 0)
        evidence = mock.Mock()
        evidence.run_command.return_value = {}
        phase = mock.Mock(bootstrap_budget_seconds=1.0, ready_budget_seconds=2.0)
        with mock.patch(
            "reliability_core.suite_isolation.phase_budget_from_environment",
            return_value=phase,
        ):
            regressions_r5._tree_run(
                evidence,
                ready_delay=0.45,
                ready_timeout=3.0,
                execution_timeout=0.1,
                single_process_ready=True,
            )
        kwargs = evidence.run_command.call_args.kwargs
        self.assertEqual(kwargs["bootstrap_timeout_seconds"], 2.0)
        self.assertEqual(kwargs["ready_timeout_seconds"], 3.0)

    def test_process_timeout_preserves_operation_and_cleanup_budgets(self) -> None:
        from reliability_core.process_control import cleanup_reserve_seconds, total_timeout_for_operation

        for operation_seconds in (0.25, 6.0, 45.0):
            total_seconds = total_timeout_for_operation(operation_seconds)
            self.assertGreaterEqual(
                total_seconds - cleanup_reserve_seconds(total_seconds),
                operation_seconds,
            )

    def test_worker_import_discovery_calibration_uses_fresh_processes(self) -> None:
        from reliability_core.suite_isolation import calibrate_worker_import_discovery

        observed = calibrate_worker_import_discovery(ROOT)
        self.assertTrue(observed.passed)
        self.assertEqual(len(observed.samples_seconds), 3)
        self.assertEqual(len(observed.sample_pids), 3)
        self.assertEqual(len(set(observed.sample_process_identities)), 3)
        self.assertGreater(observed.p95_seconds, 0)

    def test_worker_calibration_spans_every_safe_test_module(self) -> None:
        from reliability_core.suite_isolation import (
            SAFE_SHARED_WORKER,
            load_policy_manifest,
            worker_calibration_test_ids,
        )

        manifest = load_policy_manifest(ROOT / "test_execution_manifest.json")
        expected_modules = {
            str(row["test_id"]).rsplit(".", 2)[0]
            for row in manifest["tests"]
            if row["policy"] == SAFE_SHARED_WORKER
        }
        selected = worker_calibration_test_ids(ROOT)
        selected_modules = {test_id.rsplit(".", 2)[0] for test_id in selected}
        self.assertEqual(selected_modules, expected_modules)
        self.assertEqual(len(selected), len(expected_modules))
        self.assertEqual(tuple(sorted(selected)), selected)

    def test_worker_process_floor_includes_measured_startup_jitter(self) -> None:
        from reliability_core.suite_isolation import (
            calibrated_worker_process_floor,
            derive_phase_budget,
        )

        observed = derive_phase_budget((1.270120, 1.401358, 1.413183))
        expected = observed.p95_seconds + 2.0 * (
            max(observed.samples_seconds) - min(observed.samples_seconds)
        )
        self.assertAlmostEqual(calibrated_worker_process_floor(observed), expected)
        self.assertGreater(calibrated_worker_process_floor(observed), observed.p95_seconds)

    def test_worker_calibration_uses_the_supervised_execution_path(self) -> None:
        from reliability_core.suite_isolation import calibrate_worker_import_discovery

        records = []
        for index, elapsed_ms in enumerate((1000.0, 1100.0, 1200.0), start=1):
            records.append({
                "pid": 7,
                "root_process_identity": {"pid": 7, "started": f"creation-time-{index}", "state": "EXITED"},
                "elapsed_ms": elapsed_ms,
                "exit_code": 0,
                "timed_out": False,
                "orphan_process_count": 0,
                "zombie_process_count": 0,
                "output_drain_threads_stopped": True,
                "stderr": "",
                "stdout": 'TEST_WORKER_REPORT {"passed": true, "tests_run": 1}\n',
            })
        with mock.patch(
            "reliability_core.suite_isolation.worker_calibration_test_ids",
            return_value=("tests.test_r10r3r2.R10R3R2Tests.test_deadline_hierarchy_is_strict",),
        ), mock.patch(
            "reliability_core.process_control.run_supervised_process",
            side_effect=records,
        ) as supervised:
            observed = calibrate_worker_import_discovery(ROOT)
        self.assertEqual(supervised.call_count, 3)
        self.assertEqual(observed.samples_seconds, (1.0, 1.1, 1.2))
        self.assertEqual(observed.sample_pids, (7, 7, 7))
        self.assertEqual(
            observed.sample_process_identities,
            ("7:creation-time-1", "7:creation-time-2", "7:creation-time-3"),
        )
        self.assertTrue(observed.passed)

    def test_worker_calibration_isolates_each_module_representative(self) -> None:
        from reliability_core.suite_isolation import calibrate_worker_import_discovery

        test_ids = (
            "tests.test_acceptance.AcceptanceTests.test_at_01",
            "tests.test_rg.RegressionAttackTests.test_rg_01",
        )
        calls = []

        def supervised_record(*args, **kwargs):
            calls.append(kwargs)
            selected = json.loads(kwargs["environment"]["WRLAB_TEST_FILTER"])
            index = len(calls)
            return {
                "pid": index,
                "root_process_identity": {"pid": index, "started": f"creation-time-{index}", "state": "EXITED"},
                "elapsed_ms": 900.0 + index * 100.0,
                "exit_code": 0,
                "timed_out": False,
                "orphan_process_count": 0,
                "zombie_process_count": 0,
                "output_drain_threads_stopped": True,
                "stderr": "",
                "stdout": (
                    "TEST_WORKER_REPORT "
                    + json.dumps({"passed": True, "tests_run": len(selected)})
                    + "\n"
                ),
            }

        with mock.patch(
            "reliability_core.suite_isolation.worker_calibration_test_ids",
            return_value=test_ids,
        ), mock.patch(
            "reliability_core.process_control.run_supervised_process",
            side_effect=supervised_record,
        ) as supervised:
            observed = calibrate_worker_import_discovery(ROOT)

        self.assertEqual(supervised.call_count, 6)
        self.assertTrue(all(len(json.loads(row["environment"]["WRLAB_TEST_FILTER"])) == 1 for row in calls))
        self.assertEqual(
            [json.loads(row["environment"]["WRLAB_TEST_FILTER"])[0] for row in calls],
            [test_ids[0], test_ids[1]] * 3,
        )
        self.assertEqual(observed.samples_seconds, (1.1, 1.3, 1.5))
        self.assertEqual(observed.sample_pids, (2, 4, 6))

    def test_worker_calibration_partitions_large_module_sets_without_resampling(self) -> None:
        from reliability_core.suite_isolation import calibrate_worker_import_discovery

        test_ids = tuple(f"tests.test_module_{index}.Tests.test_probe" for index in range(6))
        calls = []

        def supervised_record(*args, **kwargs):
            calls.append(kwargs)
            selected = json.loads(kwargs["environment"]["WRLAB_TEST_FILTER"])
            index = len(calls)
            return {
                "pid": index,
                "root_process_identity": {"pid": index, "started": f"creation-time-{index}", "state": "EXITED"},
                "elapsed_ms": 900.0 + index * 100.0,
                "exit_code": 0,
                "timed_out": False,
                "orphan_process_count": 0,
                "zombie_process_count": 0,
                "output_drain_threads_stopped": True,
                "stderr": "",
                "stdout": 'TEST_WORKER_REPORT {"passed": true, "tests_run": 1}\n',
            }

        with mock.patch(
            "reliability_core.suite_isolation.worker_calibration_test_ids",
            return_value=test_ids,
        ), mock.patch(
            "reliability_core.process_control.run_supervised_process",
            side_effect=supervised_record,
        ) as supervised:
            observed = calibrate_worker_import_discovery(ROOT)

        self.assertEqual(supervised.call_count, len(test_ids))
        self.assertEqual(
            [json.loads(row["environment"]["WRLAB_TEST_FILTER"])[0] for row in calls],
            list(test_ids),
        )
        self.assertTrue(all(len(json.loads(row["environment"]["WRLAB_TEST_FILTER"])) == 1 for row in calls))
        self.assertEqual(observed.samples_seconds, (1.3, 1.4, 1.5))
        self.assertEqual(observed.sample_pids, (4, 5, 6))

    def test_sacrificial_worker_uses_minimal_real_process_tree_probe(self) -> None:
        from reliability_core.process_control import run_supervised_process, total_timeout_for_operation
        from reliability_core.suite_isolation import phase_budget_from_environment

        phase_budget = phase_budget_from_environment(ROOT)
        worker_total_seconds = total_timeout_for_operation(phase_budget.ready_budget_seconds + 0.1)
        outer_operation_seconds = phase_budget.bootstrap_budget_seconds + worker_total_seconds
        record = run_supervised_process(
            [
                sys.executable,
                str(ROOT / "scripts" / "r8_sacrificial_worker.py"),
                "--test-id",
                "R4-RG-05",
                "--iteration",
                "1",
                "--mode",
                "pass",
            ],
            ROOT,
            environment={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
            total_timeout_seconds=total_timeout_for_operation(outer_operation_seconds),
            max_output_bytes=256 * 1024,
        )
        self.assertEqual(record["exit_code"], 0, record)
        self.assertFalse(record["timed_out"], record)
        self.assertTrue(record["stdout"].strip(), record)
        payload = json.loads(record["stdout"].strip().splitlines()[-1])
        result = payload["result"]
        observed = result["observed"]
        self.assertEqual(result["implementation"], "minimal_process_tree_probe")
        self.assertTrue(result["passed"])
        self.assertTrue(observed["ready"])
        self.assertTrue(observed["timed_out"])
        self.assertGreaterEqual(observed["tracked_descendant_count"], 2)
        self.assertEqual(observed["orphan_process_count"], 0)
        self.assertEqual(observed["zombie_process_count"], 0)
        self.assertEqual(observed["alive_after_timeout"], [])
        self.assertGreaterEqual(observed["ready_timeout_seconds"], 1.0)
        self.assertEqual(record["orphan_process_count"], 0)
        worker_source = (ROOT / "scripts" / "r8_sacrificial_worker.py").read_text(encoding="utf-8")
        self.assertIn("phase_budget_from_environment", worker_source)
        self.assertNotIn("ready_timeout_seconds=1.0", worker_source)
        self.assertEqual(record["zombie_process_count"], 0)

    def test_observed_long_and_side_effecting_tests_are_isolated(self) -> None:
        from reliability_core.suite_isolation import SAFE_SHARED_WORKER, TIMEOUT_PROFILE_SECONDS

        manifest = json.loads((ROOT / "test_execution_manifest.json").read_text(encoding="utf-8"))
        entries = {row["test_id"]: row for row in manifest["tests"]}
        observed_minimums = {
            "tests.test_r4_rg.R4RegressionTests.test_r4_rg_07": 7.0,
            "tests.test_r5_rg.R5RegressionTests.test_r5_rg_17": 4.0,
            "tests.test_r6_rg.R6RegressionTests.test_r6_rg_09": 16.0,
            "tests.test_r7_rg.R7RegressionTests.test_r7_rg_23": 11.0,
            "tests.test_r8_rg.R8RegressionTests.test_r8_rg_08": 6.032,
            "tests.test_r8_rg.R8RegressionTests.test_r8_rg_25": 7.0,
            "tests.test_r8_rg.R8RegressionTests.test_r8_rg_27": 7.0,
            "tests.test_repetition.RepetitionTests.test_eval_03_twenty_scenarios_three_repetitions": 6.5,
            "tests.test_rg.RegressionAttackTests.test_rg_16": 11.0,
        }
        for test_id, observed_seconds in observed_minimums.items():
            entry = entries[test_id]
            self.assertNotEqual(entry["policy"], SAFE_SHARED_WORKER)
            self.assertGreaterEqual(TIMEOUT_PROFILE_SECONDS[entry["timeout_profile"]], observed_seconds)

    def test_cross_module_worker_calibration_has_measured_heavy_profile(self) -> None:
        from reliability_core.suite_isolation import TIMEOUT_PROFILE_SECONDS

        manifest = json.loads((ROOT / "test_execution_manifest.json").read_text(encoding="utf-8"))
        entries = {row["test_id"]: row for row in manifest["tests"]}
        test_id = (
            "tests.test_r10r3r2.R10R3R2Tests."
            "test_worker_import_discovery_calibration_uses_fresh_processes"
        )
        entry = entries[test_id]
        self.assertEqual(entry["policy"], "ISOLATED_TEST_PROCESS")
        self.assertEqual(entry["timeout_profile"], "isolated_heavy")
        self.assertGreaterEqual(TIMEOUT_PROFILE_SECONDS[entry["timeout_profile"]], 43.139)

    def test_observed_random_order_worker_durations_have_measured_profiles(self) -> None:
        from reliability_core.suite_isolation import TIMEOUT_PROFILE_SECONDS

        manifest = json.loads((ROOT / "test_execution_manifest.json").read_text(encoding="utf-8"))
        entries = {row["test_id"]: row for row in manifest["tests"]}
        observed_seconds = {
            "tests.test_r2_rg.R2RegressionTests.test_r2_rg_01": 1.869,
            "tests.test_r3_rg.R3RegressionTests.test_r3_rg_04": 1.913,
            "tests.test_r3_rg.R3RegressionTests.test_r3_rg_01": 1.882,
            "tests.test_correction_regressions.CorrectionDefectTests.test_d16_platform_check_is_nonblocking": 1.781,
            "tests.test_correction_regressions.CorrectionDefectTests.test_d15_rollback_uses_canary_patch": 4.342,
        }
        for test_id, measured_seconds in observed_seconds.items():
            entry = entries[test_id]
            self.assertGreaterEqual(
                TIMEOUT_PROFILE_SECONDS[entry["timeout_profile"]],
                measured_seconds,
                test_id,
            )

    def test_group_execution_budget_sums_all_timeout_profiles(self) -> None:
        from reliability_core import suite_isolation

        budget = suite_isolation.timeout_profile_budget(
            ["shared_safe", "shared_safe", "adaptive_process"]
        )
        self.assertEqual(budget, 6.5)

        test_ids = [f"tests.synthetic.Case.test_{index:03d}" for index in range(243)]
        chunks = suite_isolation.partition_timeout_bounded_tests(
            test_ids,
            {test_id: "shared_safe" for test_id in test_ids},
            maximum_execution_cost_seconds=198.0,
            worker_process_floor_seconds=3.019375,
            shared_safe_observed_p95_seconds=2.165375,
        )
        self.assertGreater(len(chunks), 1)
        self.assertEqual([test_id for chunk in chunks for test_id in chunk], test_ids)
        for chunk in chunks:
            execution_cost = 3.019375 + suite_isolation.timeout_profile_budget(
                ("shared_safe" for _ in chunk),
                shared_safe_observed_p95_seconds=2.165375,
            )
            self.assertLessEqual(execution_cost, 198.0)

    def test_shared_worker_budget_uses_fresh_lifecycle_measurement_as_floor(self) -> None:
        from reliability_core import suite_isolation

        budget = suite_isolation.timeout_profile_budget(
            ["shared_safe", "shared_safe"],
            shared_safe_observed_p95_seconds=0.75,
        )
        self.assertEqual(budget, 1.5)

        adaptive_budget = suite_isolation.timeout_profile_budget(
            ["shared_safe", "adaptive_process"],
            shared_safe_observed_p95_seconds=0.75,
        )
        self.assertEqual(adaptive_budget, 6.75)

    def test_stage_order_is_failfast_then_long_running(self) -> None:
        self.assertEqual(len(STAGE_NAMES), 10)
        self.assertEqual(STAGE_NAMES[:6], (
            "STRUCTURAL_PREFLIGHT", "KNOWN_FAILURE_SMOKE", "KNOWN_FAILURE_STABILITY",
            "MIXED_INTERFERENCE", "DETERMINISTIC_CANARY", "LINUX313_DEFAULT_COLD_FULL",
        ))

    def test_known_failure_set_is_exact(self) -> None:
        self.assertEqual(len(KNOWN_FAILURE_TEST_IDS), 6)
        self.assertEqual(len(set(KNOWN_FAILURE_TEST_IDS)), 6)
        self.assertTrue(all(test_id.startswith("tests.test_") for test_id in KNOWN_FAILURE_TEST_IDS))

    def test_repeat_ladder_is_not_reduced(self) -> None:
        self.assertEqual(REPEAT_LADDER, (1, 3, 10, 20))

    def test_calibration_requires_three_samples_per_phase(self) -> None:
        with self.assertRaises(ValueError):
            DeadlineCalibration.from_samples(
                startup=(0.1, 0.2), execution=(1, 2, 3), drain=(1, 2, 3), cleanup=(1, 2, 3),
                environment_sha256="a" * 64,
            )

    def test_calibration_uses_observed_p95(self) -> None:
        value = calibration()
        self.assertEqual(value.startup_p95_seconds, 0.3)
        self.assertEqual(value.execution_p95_seconds, 3.0)
        self.assertEqual(value.cleanup_p95_seconds, 0.3)
        self.assertEqual(observed_execution_seconds({"elapsed_ms": 1250}), 1.25)

    def test_deadline_hierarchy_is_strict(self) -> None:
        value = DeadlineHierarchy(100, 90, 80, 70, 10)
        self.assertTrue(value.passed)
        negative_runner = (ROOT / "scripts" / "r10r2_negative_runner.py").read_text(encoding="utf-8")
        self.assertIn("allocate_deadlines", negative_runner)
        self.assertIn("negative_deadline_calibration", negative_runner)
        self.assertNotIn('run_one(args.output_dir, "timeout", 1.2, 5.0)', negative_runner)

    def test_deadline_hierarchy_rejects_equal_levels(self) -> None:
        self.assertFalse(DeadlineHierarchy(100, 90, 90, 70, 10).passed)

    def test_allocated_deadlines_preserve_three_reserves(self) -> None:
        value = allocate_deadlines(100, calibration())
        self.assertTrue(value.passed)
        self.assertGreater(value.reviewer_seconds, value.supervisor_seconds)

    def test_group_admission_accepts_measured_cost(self) -> None:
        value = calibration()
        self.assertEqual(execution_cost_p95(value, 0.25), value.execution_p95_seconds)
        required = value.admission_cost()
        self.assertTrue(group_admission(required, value)["admitted"])

    def test_group_admission_rejects_insufficient_remaining_time(self) -> None:
        value = calibration()
        self.assertFalse(group_admission(value.admission_cost() - 0.001, value)["admitted"])

    def test_environment_digest_excludes_secret_fields(self) -> None:
        first = environment_sha256({"PYTHONHASHSEED": "1", "R10_R3_SECRET_TOKEN": "one"})
        second = environment_sha256({"PYTHONHASHSEED": "1", "R10_R3_SECRET_TOKEN": "two"})
        self.assertEqual(first, second)

    def test_canary_is_at_least_twenty_percent(self) -> None:
        ids = [f"tests.t.C.test_{index:03d}" for index in range(100)]
        policies = {test_id: {"policy": "SAFE_SHARED_WORKER"} for test_id in ids}
        self.assertGreaterEqual(len(canary_selection(ids, policies)), 20)

    def test_canary_includes_every_sensitive_test(self) -> None:
        ids = [f"tests.t.C.test_{index:03d}" for index in range(20)]
        policies = {test_id: {"policy": "SAFE_SHARED_WORKER"} for test_id in ids}
        policies[ids[3]]["creates_process"] = True
        policies[ids[5]]["uses_signal"] = True
        selected = canary_selection(ids, policies)
        self.assertIn(ids[3], selected)
        self.assertIn(ids[5], selected)

    def test_checkpoint_explicitly_is_not_evidence(self) -> None:
        value = checkpoint_payload(
            stage_id=1, source_tree_sha256="a" * 64, result_zip_sha256=None,
            workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
            completed_static_gates=("stage0",), completed_final_rows=(), next_action="stage1",
            invalidated_scopes=(), last_failure_category=None,
        )
        self.assertIs(value["checkpoint_is_evidence"], False)

    def test_checkpoint_recomputes_raw_digest(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            root = pathlib.Path(directory)
            raw = root / "row.json"
            raw.write_bytes(b"observed")
            digest = hashlib.sha256(raw.read_bytes()).hexdigest()
            checkpoint = checkpoint_payload(
                stage_id=9, source_tree_sha256="a" * 64, result_zip_sha256="d" * 64,
                workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
                completed_static_gates=(), completed_final_rows=({"run_id": "r", "raw_path": "row.json", "raw_sha256": digest},),
                next_action="next-row", invalidated_scopes=(), last_failure_category=None,
            )
            report = validate_checkpoint(
                checkpoint, source_tree_sha256="a" * 64, result_zip_sha256="d" * 64,
                workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64, raw_root=root,
            )
            self.assertTrue(report["passed"])

    def test_source_change_invalidates_static_candidate_and_final(self) -> None:
        checkpoint = checkpoint_payload(
            stage_id=4, source_tree_sha256="a" * 64, result_zip_sha256="d" * 64,
            workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
            completed_static_gates=(), completed_final_rows=(), next_action="stage4",
            invalidated_scopes=(), last_failure_category=None,
        )
        report = validate_checkpoint(
            checkpoint, source_tree_sha256="e" * 64, result_zip_sha256="d" * 64,
            workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
            raw_root=ROOT,
        )
        self.assertEqual(report["invalidated_scopes"], ["static_gates", "candidate_results", "sealed_result", "exact_final_rows"])

    def test_result_change_invalidates_all_final_rows(self) -> None:
        checkpoint = checkpoint_payload(
            stage_id=9, source_tree_sha256="a" * 64, result_zip_sha256="d" * 64,
            workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
            completed_static_gates=(), completed_final_rows=(), next_action="stage9",
            invalidated_scopes=(), last_failure_category=None,
        )
        report = validate_checkpoint(
            checkpoint, source_tree_sha256="a" * 64, result_zip_sha256="e" * 64,
            workspec_sha256="b" * 64, matrix_contract_sha256="c" * 64,
            raw_root=ROOT,
        )
        self.assertEqual(report["invalidated_scopes"], ["exact_final_rows"])

    def test_runtime_evidence_categories_are_never_reusable(self) -> None:
        self.assertTrue({"test_result", "pid", "ready", "stdout", "cleanup", "challenge", "matrix_row"}.issubset(NON_REUSABLE_EVIDENCE))

    def test_cleanup_wait_is_bounded_by_hard_deadline(self) -> None:
        value = calibration()
        with mock.patch.dict(os.environ, {"R10_R3_DEADLINE_CALIBRATION_JSON": json.dumps(value.to_dict())}, clear=False):
            timeout = cleanup_wait_seconds(time.monotonic() + 0.01)
        self.assertLessEqual(timeout, 0.02)
        self.assertGreater(timeout, 0)

    def test_cleanup_wait_uses_fresh_local_cleanup_floor(self) -> None:
        tiny = DeadlineCalibration.from_samples(
            startup=(0.001, 0.001, 0.001),
            execution=(0.001, 0.001, 0.001),
            drain=(0.0001, 0.0001, 0.0001),
            cleanup=(0.0001, 0.0001, 0.0001),
            environment_sha256="b" * 64,
        )
        local = DeadlineCalibration.from_samples(
            startup=(0.001, 0.001, 0.001),
            execution=(0.001, 0.001, 0.001),
            drain=(0.002, 0.002, 0.002),
            cleanup=(0.02, 0.02, 0.02),
            environment_sha256="c" * 64,
        )
        with (
            mock.patch.dict(os.environ, {"R10_R3_DEADLINE_CALIBRATION_JSON": json.dumps(tiny.to_dict())}, clear=False),
            mock.patch("reliability_core.stage_control_r10r3r2.local_cleanup_calibration", return_value=local),
        ):
            timeout = cleanup_wait_seconds()
        self.assertGreaterEqual(timeout, local.cleanup_p95_seconds + local.finalization_reserve_seconds)

    def test_coarse_windows_clock_keeps_deadline_hierarchy_operational(self) -> None:
        with mock.patch(
            "reliability_core.stage_control_r10r3r2.time.get_clock_info",
            return_value=mock.Mock(resolution=0.015625),
        ):
            coarse = DeadlineCalibration.from_samples(
                startup=(0.0, 0.0, 0.0),
                execution=(0.0, 0.0, 0.0),
                drain=(0.0, 0.0, 0.0),
                cleanup=(0.0, 0.0, 0.0),
                environment_sha256="d" * 64,
            )
        hierarchy = allocate_deadlines(5.0, coarse)
        self.assertAlmostEqual(coarse.finalization_reserve_seconds, 0.0625)
        self.assertTrue(hierarchy.passed)
        self.assertGreater(hierarchy.supervisor_seconds, hierarchy.controller_seconds)
        self.assertGreater(hierarchy.controller_seconds, hierarchy.worker_seconds)

    def test_replay_deadline_is_calibrated_by_target_interpreter(self) -> None:
        from scripts import r10r2_replay_bundle as replay

        target_hierarchy = {
            "reviewer_seconds": 900.0,
            "supervisor_seconds": 899.8,
            "controller_seconds": 899.6,
            "worker_seconds": 899.4,
            "reserve_seconds": 0.2,
            "passed": True,
        }
        completed = subprocess.CompletedProcess(
            args=["target-python"],
            returncode=0,
            stdout=json.dumps({"calibration": calibration().to_dict(), "hierarchy": target_hierarchy}),
            stderr="",
        )
        with mock.patch.object(replay.subprocess, "run", return_value=completed) as invoked:
            command = replay.command_for_supervisor(
                "target-python", "default", None, "normal", 900.0,
            )
        self.assertEqual(command[command.index("--deadline-seconds") + 1], "899.8")
        calibration_argument = command[command.index("--deadline-calibration-json") + 1]
        self.assertEqual(
            json.loads(calibration_argument),
            json.loads(json.dumps(calibration().to_dict())),
        )
        self.assertEqual(invoked.call_args.args[0][0], "target-python")
        self.assertIn("local_cleanup_calibration(sys.executable)", invoked.call_args.args[0][-1])

    def test_fixed_three_and_one_point_five_waits_are_removed(self) -> None:
        pattern = re.compile(r"(?:wait|join)\s*\(\s*timeout\s*=\s*(?:3(?:\.0)?|1\.5)\s*\)")
        paths = [
            ROOT / "scripts" / "r9_matrix_runner.py",
            ROOT / "src" / "reliability_core" / "regressions_r5.py",
            ROOT / "src" / "reliability_core" / "suite_isolation.py",
            ROOT / "src" / "reliability_core" / "supervisor_r10r2.py",
        ]
        hits = [str(path.relative_to(ROOT)) for path in paths if pattern.search(path.read_text(encoding="utf-8"))]
        self.assertEqual(hits, [])

    def test_rollback_source_walk_prunes_excluded_dirs_before_iteration(self) -> None:
        from scripts import build_evidence as evidence

        with tempfile.TemporaryDirectory() as directory:
            root = pathlib.Path(directory) / "source"
            models = root / "src" / "reliability_core" / "models.py"
            models.parent.mkdir(parents=True)
            models.write_text("VALUE = 1\n", encoding="utf-8")
            (root / "run_all.py").write_text("raise SystemExit(0)\n", encoding="utf-8")
            ignored = root / "artifacts" / "deep" / "ignored.txt"
            ignored.parent.mkdir(parents=True)
            ignored.write_text("must not be traversed", encoding="utf-8")
            calls = 0

            def completed_command(*_args: object, **_kwargs: object) -> dict[str, object]:
                nonlocal calls
                calls += 1
                return {
                    "stdout": "",
                    "stderr": "",
                    "exit_code": 1 if calls == 1 else 0,
                    "timed_out": False,
                    "timeout_seconds": 1200,
                    "orphan_process_count": 0,
                }

            with (
                mock.patch.object(pathlib.Path, "rglob", side_effect=AssertionError("unpruned rglob")),
                mock.patch.object(evidence, "run_command", side_effect=completed_command),
            ):
                proof = evidence.run_canary_rollback_drill(
                    root,
                    pathlib.Path(directory) / "drill",
                    ["python", "run_all.py", "--child-smoke"],
                )

            self.assertTrue(proof["passed"])
            self.assertEqual(calls, 2)
            self.assertFalse((pathlib.Path(directory) / "drill" / "clone" / "artifacts").exists())

    def test_cpu_medium_controller_records_load_and_cleanup(self) -> None:
        from reliability_core import supervisor_r10r2 as supervisor

        with tempfile.TemporaryDirectory() as directory:
            process, record = supervisor._start_load_controller(
                ROOT,
                pathlib.Path(directory),
                startup_timeout_seconds=2.0,
            )
            try:
                time.sleep(0.35)
            finally:
                record.update(
                    supervisor.stop_load_controller(
                        process,
                        record,
                        time.monotonic() + 2.0,
                    )
                )

            self.assertTrue(record["ready_handshake"]["ready"])
            self.assertGreater(record["observed_cycle_count"], 0)
            self.assertGreater(record["observed_busy_fraction"], 0.20)
            self.assertLess(record["observed_busy_fraction"], 0.80)
            self.assertTrue(record["metrics_finalized"])
            self.assertTrue(record["cleanup_passed"])
            self.assertEqual(record["orphan_process_count"], 0)
            self.assertEqual(record["zombie_process_count"], 0)

    def test_cpu_controller_cleanup_budget_uses_observed_cycle_time(self) -> None:
        from reliability_core import supervisor_r10r2 as supervisor

        with tempfile.TemporaryDirectory() as directory:
            metrics = pathlib.Path(directory) / "metrics.json"
            metrics.write_text(
                json.dumps({"cycle_count": 5, "elapsed_seconds": 1.0}),
                encoding="utf-8",
            )
            budget = supervisor.load_controller_cleanup_budget(
                {"metrics_path": str(metrics)},
                minimum_seconds=0.01,
            )
        self.assertAlmostEqual(budget, 0.4)

    def test_cpu_controller_startup_budget_uses_current_phase_calibration(self) -> None:
        from reliability_core import supervisor_r10r2 as supervisor

        budget = supervisor.load_controller_startup_budget(
            historical_startup_p95_seconds=0.18,
            current_phase_bootstrap_budget_seconds=1.0,
        )
        self.assertEqual(budget, 1.0)
        self.assertGreater(budget, 0.18 * 3.0)

    def test_cpu_controller_ready_uses_phase_budget_when_history_is_tiny(self) -> None:
        from reliability_core import supervisor_r10r2 as supervisor
        from reliability_core.suite_isolation import phase_budget_from_environment

        phase_budget = phase_budget_from_environment(ROOT)
        startup_timeout = supervisor.load_controller_startup_budget(
            historical_startup_p95_seconds=0.0001,
            current_phase_bootstrap_budget_seconds=phase_budget.bootstrap_budget_seconds,
        )
        with tempfile.TemporaryDirectory() as directory:
            process, record = supervisor._start_load_controller(
                ROOT,
                pathlib.Path(directory),
                startup_timeout_seconds=startup_timeout,
            )
            try:
                self.assertTrue(record["ready_handshake"]["ready"])
                self.assertGreater(record["ready_handshake"]["elapsed_ms"], 0.0)
                self.assertGreaterEqual(startup_timeout, phase_budget.bootstrap_budget_seconds)
            finally:
                record.update(
                    supervisor.stop_load_controller(
                        process,
                        record,
                        time.monotonic() + 2.0,
                    )
                )
            self.assertTrue(record["cleanup_passed"])
            self.assertEqual(record["orphan_process_count"], 0)
            self.assertEqual(record["zombie_process_count"], 0)

    def test_long_semantic_workloads_use_measured_cost_profiles(self) -> None:
        manifest = json.loads((ROOT / "test_execution_manifest.json").read_text(encoding="utf-8"))
        profiles = {row["test_id"]: row["timeout_profile"] for row in manifest["tests"]}
        self.assertEqual(
            profiles["tests.test_r9_rg.R9RegressionTests.test_r9_rg_31"],
            "adaptive_process",
        )
        self.assertEqual(
            profiles["tests.test_repetition.RepetitionTests.test_eval_03_twenty_scenarios_three_repetitions"],
            "isolated_heavy",
        )
        self.assertEqual(
            profiles["tests.test_rg.RegressionAttackTests.test_rg_16"],
            "isolated_heavy",
        )

    def test_r5_rg_03_uses_short_post_ready_execution_window(self) -> None:
        from reliability_core.regressions_r5 import run_r5_rg

        result = run_r5_rg("R5-RG-03")
        self.assertTrue(result.passed)
        self.assertTrue(result.observed["workflow_timed_out"])
        self.assertLessEqual(result.observed["execution_timeout_seconds"], 0.15)

    def test_r4_rg_05_uses_short_post_ready_execution_window(self) -> None:
        from reliability_core.regressions_r4 import run_r4_rg

        result = run_r4_rg("R4-RG-05")
        self.assertTrue(result.passed)
        self.assertEqual(result.observed["alive_after_timeout"], [])
        self.assertLessEqual(result.observed["execution_timeout_seconds"], 0.15)

    def test_supervisor_records_calibration_and_hierarchy(self) -> None:
        text = (ROOT / "src" / "reliability_core" / "supervisor_r10r2.py").read_text(encoding="utf-8")
        self.assertIn('"deadline_hierarchy": deadline_hierarchy.to_dict()', text)
        self.assertIn('"deadline_calibration": deadline_calibration.to_dict()', text)

    def test_long_stages_cannot_precede_canary(self) -> None:
        self.assertLess(STAGE_NAMES.index("DETERMINISTIC_CANARY"), STAGE_NAMES.index("LINUX313_DEFAULT_COLD_FULL"))
        self.assertLess(STAGE_NAMES.index("ROLLBACK_AND_SEAL"), STAGE_NAMES.index("EXACT_FINAL_REPLAY_24"))

    def test_filtered_worker_loads_requested_names_without_global_discovery(self) -> None:
        import run_all

        requested = [
            "tests.test_acceptance.AcceptanceTests.test_at_08",
            "tests.test_r5_rg.R5RegressionTests.test_r5_rg_10",
        ]
        with (
            mock.patch.dict(
                os.environ,
                {
                    "WRLAB_TEST_FILTER": json.dumps(requested),
                    "WRLAB_TEST_EXCLUDE": "[]",
                },
                clear=False,
            ),
            mock.patch.object(
                run_all.unittest.defaultTestLoader,
                "discover",
                side_effect=AssertionError("filtered workers must not run global discovery"),
            ),
        ):
            suite = run_all._selected_worker_suite()

        self.assertEqual([test.id() for test in run_all._flatten(suite)], requested)


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