import copy

from pathlib import Path

import sys

import unittest

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))

from oakbench.campaign_report import report_campaign, render_markdown

def plan(repetitions=2):

    return {"schema_version": 1, "campaign_id": "example", "plan_sha256": "digest", "manifest": {},

            "schedule": [dict(trial_id=f"{rep}-{subject}", block_id=f"block-{rep}", ordinal=rep * 2 + index,

                              repetition=rep, workflow="wide_config_refactor", instruction_level="zero-shot",

                              agent="codex", model="pinned-model", subject=subject)

                         for rep in range(repetitions) for index, subject in enumerate(("git", "oak"))]}

def event(trial_id, kind, **extra):

    return dict(event=kind, trial_id=trial_id, plan_sha256="digest", timestamp_utc="2026-09-10T00:00:00Z", **extra)

def outcome(trial_id, name="pass", tokens=10):

    return [event(trial_id, "started"), event(trial_id, "finished", row={

        "outcome": name, "success": name == "pass", "wall_clock_metrics": {"agent_active_ms": 20, "total_ms": 30},

        "token_metrics": {"total_tokens_reported": tokens, "cost_weighted_total": 40},

        "turn_metrics": {"assistant_turns_total": 2}, "tool_call_metrics": {"tool_calls_total": 3},

    })]

class CampaignReportTest(unittest.TestCase):

    def report(self, configuration, events):

        return report_campaign(configuration, events, bootstrap_samples=100, seed=7)

    def test_schedule_and_started_denominators_survive_interruption(self):

        summary = self.report(plan(), outcome("0-git") + outcome("0-oak", "fail") + [event("1-git", "started")])

        self.assertEqual(summary["counts"], dict(scheduled=4, started=3, finished=2, budget_truncated=0, success=1, failure=1,

                         timeout=0, infrastructure_failure=0, interrupted=1, skipped=0, unstarted=1))

        subjects = {item["subject"]: item for item in summary["groups"][0]["subjects"]}

        self.assertEqual(subjects["git"]["verified_success_fraction_of_started"], .5)

        self.assertIsNone(subjects["git"]["success_wilson_interval"])

        self.assertIsNone(subjects["git"]["resources"]["reported_tokens"]["complete_total"])

        self.assertEqual(summary["groups"][0]["paired_comparisons"][0]["incomplete_pairs"], 1)

        self.assertFalse(summary["claim_eligible"])

    def test_legacy_shortened_timeout_is_not_a_complete_pair(self):

        configuration = plan(1)

        configuration["manifest"]["trial_timeout_seconds"] = 60

        events = outcome("0-git") + outcome("0-oak", "timeout")

        events[0]["timeout_seconds"] = 60

        events[2]["timeout_seconds"] = .15

        summary = self.report(configuration, events)

        self.assertEqual(summary["counts"]["budget_truncated"], 1)

        self.assertEqual(summary["counts"]["interrupted"], 1)

        pair = summary["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["complete_pairs"], 0)

        self.assertIsNone(pair["success_delta_percentage_points"])

    def test_missing_instrument_is_unmeasured_not_a_paired_failure(self):

        events = outcome("0-git", "error") + outcome("0-oak")

        events[1]["row"].update(returncode=77, skip_reason="required oracle unavailable")

        summary = self.report(plan(1), events)

        self.assertEqual(summary["counts"]["skipped"], 1)

        self.assertEqual(summary["counts"]["started"], 2)

        pair = summary["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["complete_pairs"], 0)

        self.assertIsNone(pair["success_delta_percentage_points"])

        git = next(subject for subject in summary["groups"][0]["subjects"] if subject["subject"] == "git")

        self.assertIsNone(git["success_wilson_interval"])

    def test_fixture_invalidated_trial_is_not_a_complete_paired_failure(self):

        events = outcome("0-git") + [event("0-oak", "started"), event("0-oak", "finished",

            outcome="error", measurement_invalidated=True, invalidation_reason="fixture content drift",

            rejected_row={"outcome": "pass", "success": True})]

        summary = self.report(plan(1), events)

        self.assertEqual(summary["counts"]["skipped"], 1)

        self.assertEqual(summary["invalidated_count"], 1)

        self.assertEqual(summary["invalidated_trials"][0]["reason"], "fixture content drift")

        self.assertTrue(summary["invalidated_trials"][0]["rejected_row_retained"])

        self.assertIn("0-oak: fixture content drift", render_markdown(summary))

        self.assertEqual(summary["counts"]["infrastructure_failure"], 0)

        pair = summary["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["complete_pairs"], 0)

        self.assertIsNone(pair["success_delta_percentage_points"])

    def test_infrastructure_attempts_stay_resources_not_success_pairs(self):

        events = outcome("0-git") + outcome("1-git")

        for trial in ("0-oak", "1-oak"):

            events.extend([event(trial, "started"), event(trial, "finished", outcome="error", error="harness setup failed")])

        summary = self.report(plan(2), events)

        pair = summary["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["complete_pairs"], 0)

        self.assertIsNone(pair["success_delta_percentage_points"])

        self.assertIn("schedule_or_attempt_accounting_incomplete", summary["claim_ineligibility_reasons"])

        oak = next(item for item in summary["groups"][0]["subjects"] if item["subject"] == "oak")

        self.assertIsNone(oak["success_wilson_interval"])

        self.assertEqual(oak["resources"]["reported_tokens"]["attempted_n"], 2)

    def test_measured_precommit_failure_keeps_pair_but_success_proof_classes_must_match(self):

        events = outcome("0-git") + outcome("0-oak", "fail")

        events[1]["row"].update(workflow_integrity_passed=True, workflow_integrity_evidence_class="independent_committed_tree")

        events[3]["row"].update(workflow_integrity_passed=False, workflow_integrity_evidence_class="trusted_worktree_payload", agent_launched=True, outcome_attribution="agent")

        summary = self.report(plan(1), events)

        pair = summary["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["complete_pairs"], 1)

        self.assertEqual(pair["success_delta_percentage_points"], -100)

        events[3]["row"].update(outcome="pass", success=True, workflow_integrity_passed=True)

        with self.assertRaisesRegex(ValueError, "workflow_integrity_evidence_class"):

            self.report(plan(1), events)

    def test_every_outcome_remains_visible(self):

        configuration = plan(4)

        events = (outcome("0-git") + outcome("0-oak", "fail") + outcome("1-git", "timeout") +

                  [event("1-oak", "started"), event("1-oak", "finished", outcome="error", error="crash"),

                   event("2-git", "started"), event("2-oak", "finished", outcome="skipped")])

        counts = self.report(configuration, events)["counts"]

        for field in ("success", "failure", "timeout", "infrastructure_failure", "interrupted", "skipped"):

            self.assertEqual(counts[field], 1)

        self.assertEqual(counts["unstarted"], 2)

        self.assertEqual(sum(counts[name] for name in ("success", "failure", "timeout", "infrastructure_failure", "interrupted", "skipped", "unstarted")), counts["scheduled"])

    def test_paired_bootstrap_preserves_block_pairs_and_is_seeded(self):

        events = outcome("0-git") + outcome("0-oak", "fail") + outcome("1-git", "fail") + outcome("1-oak")

        first = self.report(plan(), events)

        second = self.report(plan(), events)

        self.assertEqual(first, second)

        pair = first["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["success_delta_percentage_points"], 0)

        self.assertEqual(pair["confidence_interval"], [-100, 100])

        self.assertEqual(pair["complete_pairs"], 2)

        self.assertFalse(pair["multiplicity_adjusted"])

    def test_singleton_pair_does_not_get_zero_width_interval(self):

        pair = self.report(plan(1), outcome("0-git") + outcome("0-oak"))["groups"][0]["paired_comparisons"][0]

        self.assertEqual(pair["success_delta_percentage_points"], 0)

        self.assertIsNone(pair["confidence_interval"])

    def test_missing_usage_does_not_become_zero_or_complete_total(self):

        summary = self.report(plan(), outcome("0-git", tokens=None) + outcome("1-git", tokens=0))

        values = summary["groups"][0]["subjects"][0]["resources"]["reported_tokens"]

        self.assertEqual(values["measured_n"], 1)

        self.assertEqual(values["missing_n"], 1)

        self.assertEqual(values["observed_sum"], 0)

        self.assertIsNone(values["complete_total"])

        self.assertIsNone(values["observed_cost_per_success"])

        self.assertEqual(values["success_conditioned"]["median"], 0)

    def test_partial_capture_counters_are_not_complete_usage(self):

        for changes in ({"outcome": "timeout", "success": False}, {"output_truncated": True},

                        {"source": {"agent_execution": {"capture_complete": False}}},

                        {"source": {"agent_execution": {"usage_complete": False}}},

                        {"source": {"agent_execution": {"counters_coverage": "observed_lower_bounds"}}},

                        {"source": {"agent_execution": {"output_limited": True}}}):

            events = outcome("0-git")

            events[-1]["row"].update(changes)

            resources = self.report(plan(1), events)["groups"][0]["subjects"][0]["resources"]

            for metric in ("reported_tokens", "assistant_turns", "tool_calls"):

                self.assertIsNone(resources[metric]["complete_total"])

                self.assertEqual(resources[metric]["missing_n"], 1)

            self.assertEqual(resources["agent_active_ms"]["complete_total"], 20)

    def test_capture_error_row_keeps_infrastructure_denominator_and_time(self):

        events = outcome("0-git")

        events[-1]["outcome"] = "error"

        events[-1]["row"].update(outcome="error", success=False, output_truncated=True)

        summary = self.report(plan(1), events)

        self.assertEqual(summary["counts"]["infrastructure_failure"], 1)

        resources = summary["groups"][0]["subjects"][0]["resources"]

        self.assertEqual(resources["agent_active_ms"]["complete_total"], 20)

        self.assertIsNone(resources["tool_calls"]["complete_total"])

    def test_failures_contribute_resource_cost_but_not_success_latency(self):

        events = outcome("0-git", tokens=10) + outcome("1-git", "fail", tokens=50)

        subject = self.report(plan(), events)["groups"][0]["subjects"][0]

        values = subject["resources"]["reported_tokens"]

        self.assertEqual(values["complete_total"], 60)

        self.assertEqual(values["observed_cost_per_success"], 60)

        self.assertEqual(values["success_conditioned"]["median"], 10)

        self.assertAlmostEqual(subject["success_wilson_interval"][0], .094531, places=5)

        self.assertAlmostEqual(subject["success_wilson_interval"][1], .905469, places=5)

    def test_zero_success_or_no_attempts_keeps_undefined_metrics_null(self):

        summary = self.report(plan(1), outcome("0-git", "fail"))

        git, oak = summary["groups"][0]["subjects"]

        self.assertIsNone(git["resources"]["reported_tokens"]["observed_cost_per_success"])

        self.assertIsNone(oak["verified_success_fraction_of_started"])

        self.assertIsNone(oak["resources"]["reported_tokens"]["observed_sum"])

        self.assertIn("unmeasured", render_markdown(summary))

    def test_instruction_and_model_conditions_never_pool(self):

        configuration = plan()

        configuration["schedule"][2]["instruction_level"] = "full-docs"

        configuration["schedule"][3]["model"] = "other-model"

        summary = self.report(configuration, [])

        self.assertEqual(len(summary["groups"]), 3)

    def test_duplicate_or_foreign_journal_evidence_rejected(self):

        for events in ([event("0-git", "started")] * 2,

                       outcome("0-git") + [outcome("0-git")[1]],

                       [event("unknown", "started")],

                       [dict(event("0-git", "started"), plan_sha256="wrong")],

                       [outcome("0-git")[1]],

                       [event("0-git", "other")]):

            with self.subTest(events=events), self.assertRaises(ValueError):

                self.report(plan(), events)

    def test_contradictory_result_does_not_count_as_success(self):

        events = outcome("0-git")

        events[1]["row"]["outcome"] = "fail"

        with self.assertRaises(ValueError):

            self.report(plan(), events)

        events = outcome("0-git")

        events[1]["row"]["agent"] = {"model": "unplanned"}

        with self.assertRaises(ValueError):

            self.report(plan(), events)

    def test_mock_plan_never_becomes_claim_eligible(self):

        configuration = plan(1)

        for trial in configuration["schedule"]:

            trial.update(agent="mock", model=None)

        summary = self.report(configuration, outcome("0-git") + outcome("0-oak"))

        self.assertFalse(summary["claim_eligible"])

        self.assertIn("mock_agent_measurements_are_instrument_checks", summary["claim_ineligibility_reasons"])

    def test_duplicate_scheduled_subject_block_is_rejected(self):

        configuration = plan()

        duplicate = copy.deepcopy(configuration["schedule"][0])

        duplicate["trial_id"] = "new-id-same-block-subject"

        configuration["schedule"].append(duplicate)

        with self.assertRaises(ValueError):

            self.report(configuration, [])

    def test_runtime_environment_and_version_drift_cannot_pool_or_pair(self):

        events = outcome("0-git") + outcome("1-git")

        events[-1]["row"]["agent"] = {"environment": "local-default"}

        with self.assertRaisesRegex(ValueError, "incompatible campaign population"):

            self.report(plan(), events)

    def test_row_cannot_be_reassigned_to_another_trial(self):

        for field, value in (("task_id", "other"), ("bench_id", "other"), ("run", 99)):

            events = outcome("0-git")

            events[-1]["row"][field] = value

            with self.subTest(field=field), self.assertRaisesRegex(ValueError, "row trial identity"):

                self.report(plan(), events)

        events = outcome("0-git")

        events[-1]["outcome"] = "fail"

        with self.assertRaisesRegex(ValueError, "event outcome contradicts"):

            self.report(plan(), events)

    def test_efficiency_is_paired_and_only_joint_success_with_same_sources(self):

        events = outcome("0-git", tokens=10) + outcome("0-oak", tokens=20) + outcome("1-git", tokens=100) + outcome("1-oak", tokens=110)

        for entry in events:

            if "row" in entry:

                entry["row"]["token_metrics"]["token_source"] = "provider_reported"

                entry["row"]["wall_clock_metrics"]["measurement_source"] = "outer_cli"

        pair = self.report(plan(), events)["groups"][0]["paired_comparisons"][0]

        tokens = pair["success_conditioned_efficiency"]["reported_tokens"]

        self.assertEqual(tokens["candidate_minus_reference_mean_delta"], 10)

        self.assertEqual(tokens["confidence_interval"], [10, 10])

        self.assertEqual(tokens["measured_pairs"], 2)

        events[-1]["row"]["token_metrics"]["token_source"] = "other_convention"

        tokens = self.report(plan(), events)["groups"][0]["paired_comparisons"][0]["success_conditioned_efficiency"]["reported_tokens"]

        self.assertEqual(tokens["measured_pairs"], 1)

        self.assertEqual(tokens["incompatible_or_unknown_source_pairs"], 1)

        self.assertIsNone(tokens["confidence_interval"])

        events[-1]["row"].update(success=False, outcome="fail")

        tokens = self.report(plan(), events)["groups"][0]["paired_comparisons"][0]["success_conditioned_efficiency"]["reported_tokens"]

        self.assertEqual(tokens["jointly_successful_pairs"], 1)

        events = outcome("0-git") + outcome("0-oak")

        events[-1]["row"]["oracle_version"] = "different"

        with self.assertRaisesRegex(ValueError, "incompatible paired evidence"):

            self.report(plan(), events)

if __name__ == "__main__":

    unittest.main()