oak/benchmarks

"""Campaign contracts and transaction boundaries; no paid agent invocations."""

from __future__ import annotations

import copy

import json

import os

from pathlib import Path

import sys

import tempfile

import unittest

from unittest import mock

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

from oakbench import agent_campaign as campaign

class CampaignTest(unittest.TestCase):

    def setUp(self):

        self.tmp = tempfile.TemporaryDirectory()

        self.root = Path(self.tmp.name).resolve()

        self.manifest_path = self.root / "manifest.json"

        self.raw = json.loads((campaign.ROOT / "config/agent-campaign-smoke.json").read_text())

        self.raw.update(agents_config=str(campaign.ROOT / "config/agents.toml"),

                        subjects_config=str(campaign.ROOT / "config/subjects.toml"))

        self.write_manifest()

    def tearDown(self):

        self.tmp.cleanup()

    def write_manifest(self):

        self.manifest_path.write_text(json.dumps(self.raw))

    def manifest(self):

        self.write_manifest()

        return campaign.load_manifest(self.manifest_path)

    def plan(self):

        with mock.patch.object(campaign, "resolve_runtime", return_value=({}, {}, {})):

            return campaign.create_plan(self.manifest_path, self.root / "campaign")

    def test_oracle_instrument_identity_is_frozen_and_drift_rejected(self):

        from oakbench import workflow_oracles

        original = {"git_head_instrument_binary": "/instrument/git", "git_head_instrument_sha256": "a" * 64,

                    "git_head_instrument_version": "git version synthetic"}

        with mock.patch.object(workflow_oracles, 'resolve_git_instrument', return_value=('/instrument/git', original)), \

             mock.patch.dict(os.environ, {'OAK_BENCH_OAK_READER_PYTHON': ''}):

            pinned = campaign.oracle_instrument_provenance()

        changed = copy.deepcopy(pinned)

        changed['git']['git_head_instrument_sha256'] = 'b' * 64

        plan = self.plan()

        plan['provenance'] = {'oracle_instruments': pinned}

        with mock.patch.object(campaign, 'resolve_runtime', return_value=({}, {}, {'oracle_instruments': changed})):

            with self.assertRaisesRegex(campaign.CampaignError, 'oracle_instruments'):

                campaign.preflight(plan, self.root / 'campaign')

    def test_optional_reader_identity_records_isolated_dependency_probe(self):

        from types import SimpleNamespace

        binary = self.root / 'reader-python'

        binary.write_bytes(b'configured interpreter bytes')

        metadata = {'python': '3.test', 'dependencies': {'blake3': '1.test', 'zstandard': '2.test'}}

        with mock.patch.dict(os.environ, {'OAK_BENCH_OAK_READER_PYTHON': str(binary)}), \

             mock.patch.object(campaign.shutil, 'which', return_value=str(binary)), \

             mock.patch.object(campaign.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout=json.dumps(metadata), stderr='')) as probe:

            from oakbench import workflow_oracles

            with mock.patch.object(workflow_oracles, 'resolve_git_instrument', return_value=(None, {})):

                identity = campaign.oracle_instrument_provenance()['oak_reader']

        self.assertEqual(identity['metadata'], metadata)

        self.assertEqual(identity['sha256'], campaign.file_hash(binary))

        self.assertEqual(probe.call_args.args[0][:3], [str(binary), '-I', '-c'])

        self.assertEqual(identity['reader_module_sha256'], campaign.file_hash(campaign.ROOT / 'scripts/oakbench/oak_sqlite_reader.py'))

    def test_schedule_determinism_balance_and_identity(self):

        first = campaign.schedule(self.manifest())

        self.assertEqual(first, campaign.schedule(self.manifest()))

        self.assertEqual(len({trial["trial_id"] for trial in first}), len(first))

        self.assertEqual([x["subject"] for x in first[:2]], list(reversed([x["subject"] for x in first[2:]])))

        self.assertEqual(first[0]["block_id"], first[1]["block_id"])

        self.assertNotEqual(first[0]["block_id"], first[2]["block_id"])

        self.raw["instruction_levels"] = ["full-docs", "zero-shot"]

        self.raw["max_trials"] = 8

        self.assertEqual(len(campaign.schedule(self.manifest())), 8)

    def test_invalid_limits_unknown_fields_duplicates_and_paths(self):

        for key, value in (("schema_version", True), ("repetitions", True), ("repetitions", 0),

                           ("max_trials", -1), ("max_wall_seconds", float("nan")),

                           ("trial_timeout_seconds", float("inf")), ("seed", True),

                           ("subjects", ["git", "git"]), ("subjects", ["../escape"]),

                           ("workflows", ["history_archaeology"]), ("typo", 1)):

            with self.subTest(key=key, value=value):

                raw = dict(self.raw)

                raw[key] = value

                self.manifest_path.write_text(json.dumps(raw))

                with self.assertRaises(campaign.CampaignError):

                    campaign.load_manifest(self.manifest_path)

        self.manifest_path.write_text('{"schema_version":1,"schema_version":1}')

        with self.assertRaises(campaign.CampaignError):

            campaign.load_manifest(self.manifest_path)

    def test_huge_schedule_rejected_before_materialization(self):

        self.raw["repetitions"] = 10 ** 30

        with self.assertRaises(campaign.CampaignError):

            campaign.schedule(self.manifest())

    def test_freeze_collision_and_tamper_rejected(self):

        plan = self.plan()

        directory = self.root / "campaign"

        self.assertEqual(campaign.load_plan(directory), plan)

        with mock.patch.object(campaign, "resolve_runtime", return_value=({}, {}, {})):

            with self.assertRaises(campaign.CampaignError):

                campaign.create_plan(self.manifest_path, directory)

        plan["schedule"][0]["subject"] = "wrong"

        (directory / "plan.json").write_text(json.dumps(plan))

        with self.assertRaises(campaign.CampaignError):

            campaign.load_plan(directory)

    def test_interrupted_trial_is_never_pending_and_reserves_budget(self):

        plan = self.plan()

        directory = self.root / "campaign"

        trial = plan["schedule"][0]["trial_id"]

        campaign.append_event(directory, plan, {"event": "started", "trial_id": trial, "timeout_seconds": 60})

        status = campaign.campaign_status(plan, campaign.load_events(directory, plan))

        self.assertEqual(status["interrupted_trial_ids"], [trial])

        self.assertNotIn(trial, status["pending_trial_ids"])

        self.assertEqual(status["charged_episode_seconds"], 60)

    def test_torn_duplicate_unknown_nonfinite_and_bad_duration_journal_fail_closed(self):

        plan = self.plan()

        directory = self.root / "campaign"

        trial = plan["schedule"][0]["trial_id"]

        start = {"event": "started", "trial_id": trial, "plan_sha256": plan["plan_sha256"], "timeout_seconds": 60}

        variants = [b'{"event":', b'[]\n', (json.dumps(start) + '\n') .encode() * 2,

                    (json.dumps({**start, "trial_id": "other"}) + '\n').encode(),

                    (json.dumps({**start, "timeout_seconds": float("nan")}) + '\n').encode()]

        for wall in (True, -1, float("inf")):

            finished = {"event": "finished", "trial_id": trial, "plan_sha256": plan["plan_sha256"],

                        "outcome": "error", "error": "test", "wall_seconds": wall}

            variants.append((json.dumps(start) + '\n' + json.dumps(finished) + '\n').encode())

        for content in variants:

            with self.subTest(content=content):

                (directory / "events.jsonl").write_bytes(content)

                with self.assertRaises(campaign.CampaignError):

                    campaign.load_events(directory, plan)

                self.assertEqual((directory / "events.jsonl").read_bytes(), content)

    def test_torn_new_start_after_old_interruption_reserves_next_trial(self):

        plan = self.plan()

        directory = self.root / "campaign"

        first, second, third = plan["schedule"][:3]

        for trial in (first, second):

            campaign.append_event(directory, plan, {"event": "started", "trial_id": trial["trial_id"], "timeout_seconds": 60})

        campaign.append_event(directory, plan, {"event": "finished", "trial_id": second["trial_id"],

            "outcome": "error", "wall_seconds": 1})

        with (directory / "events.jsonl").open("ab") as stream:

            stream.write(b'{"event":"started"')

        events = campaign.load_events(directory, plan, tolerate_torn=True)

        self.assertEqual(events[-2]["trial_id"], third["trial_id"])

        self.assertTrue(events[-2]["uncertain_torn_start"])

        state = campaign.campaign_status(plan, events)

        self.assertNotIn(third["trial_id"], state["pending_trial_ids"])

    def test_system_tmp_alias_is_accepted_but_arbitrary_symlinks_are_not(self):

        self.assertEqual(campaign.safe_path(Path("/tmp") / "campaign-safe-path-check"),

                         (Path("/tmp") / "campaign-safe-path-check").resolve())

    def test_terminal_row_identity_cannot_be_rebound(self):

        from agent_metrics_schema import example_row

        plan = self.plan()

        directory = self.root / "campaign"

        trial = plan["schedule"][0]

        row = example_row()

        row.update(task_id=trial["trial_id"], bench_id=plan["campaign_id"], subject=trial["subject"],

                   scenario=trial["workflow"], run=trial["repetition"], outcome="pass")

        row["agent"].update(name=trial["agent"], model=trial["model"],

                             instruction_level=trial["instruction_level"], environment="minimal")

        campaign.validate_trial_row(row, trial, plan)

        for key, value in (("subject", "wrong"), ("scenario", "wrong"), ("run", 999), ("bench_id", "other")):

            wrong = copy.deepcopy(row)

            wrong[key] = value

            with self.subTest(key=key), self.assertRaises(campaign.CampaignError):

                campaign.validate_trial_row(wrong, trial, plan)

        for key in ("name", "model", "instruction_level"):

            wrong = copy.deepcopy(row)

            wrong["agent"][key] = "wrong"

            with self.subTest(agent_key=key), self.assertRaises(campaign.CampaignError):

                campaign.validate_trial_row(wrong, trial, plan)

        campaign.append_event(directory, plan, {"event": "started", "trial_id": trial["trial_id"], "timeout_seconds": 60})

        row["subject"] = "wrong"

        campaign.append_event(directory, plan, {"event": "finished", "trial_id": trial["trial_id"],

            "outcome": "pass", "row": row, "wall_seconds": 1})

        with self.assertRaises(campaign.CampaignError):

            campaign.load_events(directory, plan)

    def test_symlink_destinations_and_lock_rejected(self):

        alias = self.root / "alias"

        alias.symlink_to(self.root, target_is_directory=True)

        with self.assertRaises(campaign.CampaignError):

            campaign.safe_path(alias / "campaign")

        plan = self.plan()

        directory = self.root / "campaign"

        for name in ("events.jsonl", ".run.lock", "trials"):

            path = directory / name

            path.symlink_to(self.root / "outside")

            with self.assertRaises(campaign.CampaignError):

                campaign.safe_path(path)

            path.unlink()

    def test_exclusive_runner_ownership(self):

        self.plan()

        with campaign.campaign_lock(self.root / "campaign"):

            with self.assertRaises(campaign.CampaignError):

                with campaign.campaign_lock(self.root / "campaign"):

                    pass

    def test_preflight_drift_real_optin_and_attestation(self):

        plan = self.plan()

        directory = self.root / "campaign"

        with mock.patch.object(campaign, "resolve_runtime", return_value=({}, {}, {"changed": True})):

            with self.assertRaises(campaign.CampaignError):

                campaign.preflight(plan, directory)

        from agent_workflow import AgentConfig

        real = AgentConfig("real", "codex_exec_json", "openai", Path("/binary"), "pinned", "1")

        with mock.patch.object(campaign, "resolve_runtime", return_value=({"real": real}, {}, {})):

            with self.assertRaises(campaign.CampaignError):

                campaign.preflight(plan, directory)

            with self.assertRaises(campaign.CampaignError):

                campaign.preflight(plan, directory, enable_real=True)

            plan["manifest"]["isolation_attestation"] = {

                "worker_root": str(self.root), "unrestricted_cli_acknowledged": True}

            self.assertEqual(campaign.preflight(plan, directory, enable_real=True)[0]["real"], real)

    def test_lock_provenance_drift_names_the_key(self):

        plan = self.plan()

        plan["provenance"] = {"measurement_lock_path": "/old/lock"}

        with mock.patch.object(campaign, "resolve_runtime", return_value=({}, {}, {"measurement_lock_path": "/new/lock"})):

            with self.assertRaisesRegex(campaign.CampaignError, "measurement_lock_path"):

                campaign.preflight(plan, self.root / "campaign")

    def test_fixture_cache_reuses_generation_and_rejects_mutated_payload(self):

        import workflow_ab

        from dataclasses import replace

        counter = mock.Mock(side_effect=workflow_ab.WORKFLOWS["wide_config_refactor"].make_fixture)

        workflow = replace(workflow_ab.WORKFLOWS["wide_config_refactor"], make_fixture=counter)

        plan = self.plan()

        directory = self.root / "campaign"

        fixture = campaign.cached_fixture(directory, plan, workflow)

        self.assertEqual(fixture, directory / "fixtures" / workflow.name)

        self.assertEqual(campaign.cached_fixture(directory, plan, workflow), fixture)

        self.assertEqual(counter.call_count, 1)

        payload = next(fixture.glob("services/*/config.toml"))

        payload.chmod(0o600)

        payload.write_text("corrupt fixture")

        with self.assertRaisesRegex(campaign.CampaignError, "fixture.*drift"):

            campaign.cached_fixture(directory, plan, workflow)

        self.assertEqual(counter.call_count, 1)

    def test_cache_permissions_allow_rebuild_and_plain_rmtree(self):

        import workflow_ab

        import shutil

        plan = self.plan()

        directory = self.root / "campaign"

        workflow = workflow_ab.WORKFLOWS["wide_config_refactor"]

        fixture = campaign.cached_fixture(directory, plan, workflow)

        self.assertTrue(fixture.stat().st_mode & 0o200)

        self.assertTrue(all(path.stat().st_mode & 0o200 for path in fixture.rglob("*") if path.is_dir()))

        self.assertTrue(all(not path.stat().st_mode & 0o222 for path in fixture.rglob("*") if path.is_file()))

        marker = fixture / ".fixture-ready"

        marker.chmod(0o600)

        marker.write_text("old fixture version")

        rebuilt = workflow_ab.prepare_fixture(directory, workflow)

        self.assertEqual((rebuilt / ".fixture-ready").read_text().strip(), workflow_ab.FIXTURE_VERSION)

        campaign.fixture_permissions(rebuilt, writable=False)

        shutil.rmtree(directory)

        self.assertFalse(directory.exists())

    def test_permissions_do_not_require_chmod_nofollow_support(self):

        fixture = self.root / "permission-fixture"

        fixture.mkdir()

        (fixture / "file").write_text("content")

        chmod = os.chmod

        def portable(path, mode, **kwargs):

            if kwargs.get("follow_symlinks") is False:

                raise NotImplementedError("chmod nofollow unavailable on this platform")

            return chmod(path, mode, **kwargs)

        with mock.patch.object(os, "chmod", side_effect=portable):

            campaign.fixture_permissions(fixture, writable=False)

            campaign.fixture_permissions(fixture, writable=True)

    def test_fifo_cache_is_a_named_campaign_error(self):

        import workflow_ab

        plan = self.plan()

        directory = self.root / "campaign"

        fixture = campaign.cached_fixture(directory, plan, workflow_ab.WORKFLOWS["wide_config_refactor"])

        fixture.chmod(0o700)

        os.mkfifo(fixture / "unexpected-pipe")

        with self.assertRaisesRegex(campaign.CampaignError, "wide_config_refactor"):

            campaign.cached_fixture(directory, plan, workflow_ab.WORKFLOWS["wide_config_refactor"])

    def test_corrupted_cache_aborts_before_any_trial_admission(self):

        import workflow_ab

        plan = self.plan()

        directory = self.root / "campaign"

        fixture = campaign.cached_fixture(directory, plan, workflow_ab.WORKFLOWS["wide_config_refactor"])

        payload = next(fixture.glob("services/*/config.toml"))

        payload.chmod(0o600)

        payload.write_text("bad cached baseline")

        with mock.patch.object(campaign, "preflight", return_value=({}, {})), mock.patch("agent_workflow.run_one") as run:

            with self.assertRaisesRegex(campaign.CampaignError, "fixture.*drift"):

                campaign.run_campaign(directory, resume=True)

        run.assert_not_called()

        self.assertFalse((directory / "events.jsonl").exists())

        self.assertEqual(len(campaign.campaign_status(plan, [])['pending_trial_ids']), 4)

    def test_generation_crash_and_incomplete_cache_preserve_all_pending_trials(self):

        import workflow_ab

        plan = self.plan()

        directory = self.root / "campaign"

        def crash(root, workflow):

            path = root / "fixtures" / workflow.name

            path.mkdir(parents=True)

            (path / "partial").write_text("interrupted generation")

            raise KeyboardInterrupt("crash before fixture integrity record")

        with mock.patch.object(campaign, "preflight", return_value=({}, {})), mock.patch.object(workflow_ab, "prepare_fixture", side_effect=crash):

            with self.assertRaises(KeyboardInterrupt):

                campaign.run_campaign(directory)

        self.assertFalse((directory / "events.jsonl").exists())

        with mock.patch.object(campaign, "preflight", return_value=({}, {})):

            with self.assertRaisesRegex(campaign.CampaignError, "fixture cache is incomplete"):

                campaign.run_campaign(directory, resume=True)

        self.assertFalse((directory / "events.jsonl").exists())

        self.assertEqual(len(campaign.campaign_status(plan, [])['pending_trial_ids']), 4)

    def test_real_git_mock_cache_poisoning_invalidates_attempt_and_preserves_next(self):

        def mutate(directory):

            payload = next((directory / "fixtures" / "wide_config_refactor").glob("services/*/config.toml"))

            payload.chmod(0o600)  # Same-uid processes can defeat read-only files.

            payload.write_text("poisoned shared baseline")

        self.assert_midtrial_fixture_invalidation(mutate)

    def test_real_git_mock_deleted_cache_record_invalidates_attempt(self):

        self.assert_midtrial_fixture_invalidation(lambda directory:

            (directory / "fixture-integrity/wide_config_refactor.json").unlink())

    def test_real_git_mock_corrupt_cache_record_invalidates_attempt(self):

        self.assert_midtrial_fixture_invalidation(lambda directory:

            (directory / "fixture-integrity/wide_config_refactor.json").write_text("invalid JSON"))

    def assert_midtrial_fixture_invalidation(self, mutate):

        import agent_workflow

        import workflow_ab

        import shutil

        if not shutil.which("git"):

            self.skipTest("Git required")

        self.raw.update(subjects=["git"], repetitions=2)

        self.write_manifest()

        plan = self.plan()

        directory = self.root / "campaign"

        agent = agent_workflow.AgentConfig("mock", "deterministic_mock", "local", None, None, "1")

        subject = workflow_ab.Subject("git", "git", "Git", Path(shutil.which("git")))

        original = agent_workflow.run_mock_agent

        def poison(*args, **kwargs):

            execution = original(*args, **kwargs)

            mutate(directory)

            return execution

        with mock.patch.object(campaign, "preflight", return_value=({"mock": agent}, {"git": subject})), \

             mock.patch.object(agent_workflow, "run_mock_agent", side_effect=poison) as execute:

            state = campaign.run_campaign(directory)

        self.assertEqual(execute.call_count, 1)

        self.assertEqual(len(state["pending_trial_ids"]), 1)

        self.assertEqual(state["stopped_reason"], "fixture_integrity_drift")

        events = campaign.load_events(directory, plan)

        self.assertEqual(len(events), 2)

        self.assertTrue(events[-1]["measurement_invalidated"])

        self.assertTrue(events[-1]["rejected_row"]["success"])

        self.assertNotIn("row", events[-1])

        self.assertEqual((directory / "rows.jsonl").read_bytes(), b"")

        from oakbench.campaign_report import report_campaign

        report = report_campaign(plan, events)

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

        self.assertEqual(report["counts"]["unstarted"], 1)

    def test_copied_fixture_race_is_rejected_before_real_agent_launch(self):

        import agent_workflow

        import workflow_ab

        import shutil

        if not shutil.which("git"):

            self.skipTest("Git required")

        self.raw.update(subjects=["git"], repetitions=2)

        self.write_manifest()

        plan = self.plan()

        directory = self.root / "campaign"

        agent = agent_workflow.AgentConfig("mock", "deterministic_mock", "local", None, None, "1")

        subject = workflow_ab.Subject("git", "git", "Git", Path(shutil.which("git")))

        original = agent_workflow.copy_fixture

        def corrupt_copy(src, dest):

            original(src, dest)

            payload = next(dest.glob("services/*/config.toml"))

            payload.chmod(0o600)

            payload.write_text("changed during copy")

        with mock.patch.object(campaign, "preflight", return_value=({"mock": agent}, {"git": subject})), \

             mock.patch.object(agent_workflow, "copy_fixture", side_effect=corrupt_copy), \

             mock.patch.object(agent_workflow, "run_mock_agent") as execute:

            with self.assertRaisesRegex(campaign.CampaignError, "copied fixture identity drift"):

                campaign.run_campaign(directory)

        self.assertEqual(execute.call_count, 0)

        self.assertFalse((directory / "events.jsonl").exists())

        self.assertEqual(len(campaign.campaign_status(plan, [])["pending_trial_ids"]), 2)

        with mock.patch.object(campaign, "preflight", return_value=({"mock": agent}, {"git": subject})):

            state = campaign.run_campaign(directory, resume=True)

        self.assertEqual(state["finished"], 2)

        self.assertEqual(state["pending_trial_ids"], [])

        self.assertTrue(all(event["row"]["success"] for event in campaign.load_events(directory, plan) if event["event"] == "finished"))

    def test_journal_fixture_binding_rejects_missing_or_conflicting_attempt_hashes(self):

        plan = self.plan()

        trial = plan["schedule"][0]

        record = {"manifest": {"config": {"sha256": "a" * 64}}}

        fingerprint = campaign.digest(record["manifest"])

        start = {"event": "started", "trial_id": trial["trial_id"], "fixture_manifest_sha256": fingerprint}

        finish = {"event": "finished", "trial_id": trial["trial_id"], "fixture_manifest_sha256": fingerprint}

        campaign.verify_journal_fixture_binding(trial["workflow"], record, plan, [start, finish])

        for value in (None, "b" * 64):

            for index in (0, 1):

                events = [dict(start), dict(finish)]

                events[index]["fixture_manifest_sha256"] = value

                with self.assertRaisesRegex(campaign.CampaignError, "journal fixture binding"):

                    campaign.verify_journal_fixture_binding(trial["workflow"], record, plan, events)

    def test_completed_infrastructure_error_not_retried_on_resume(self):

        self.plan()

        directory = self.root / "campaign"

        import agent_workflow

        import workflow_ab

        from oakbench.runlock import LockInfo

        from contextlib import contextmanager

        @contextmanager

        def locked(label):

            yield LockInfo(0, True)

        agent = agent_workflow.AgentConfig("mock", "deterministic_mock", "local", None, None, "1")

        subject = workflow_ab.Subject("git", "git", "Git", Path("/git"))

        with mock.patch.object(campaign, "preflight", return_value=({"mock": agent}, {"git": subject, "oak_installed": subject})), \

             mock.patch("oakbench.runlock.measurement_lock", locked), \

             mock.patch.object(agent_workflow, "run_one", side_effect=RuntimeError("fixture setup failed")) as run:

            state = campaign.run_campaign(directory, limit=1)

            self.assertEqual(state["finished"], 1)

            campaign.run_campaign(directory, resume=True, limit=1)

            self.assertEqual(run.call_count, 2)

            events = campaign.load_events(directory, campaign.load_plan(directory))

            self.assertEqual(len({event["trial_id"] for event in events if event["event"] == "started"}), 2)

            self.assertEqual([event["outcome"] for event in events if event["event"] == "finished"], ["error", "error"])

    def test_staging_move_crash_after_durable_start_is_not_pending(self):

        from contextlib import contextmanager

        from oakbench.runlock import LockInfo

        self.plan()

        directory = self.root / "campaign"

        @contextmanager

        def locked(label):

            yield LockInfo(0, True)

        original = Path.rename

        def crash(path, target):

            if target.parent.name == "trials":

                raise KeyboardInterrupt("simulated process crash")

            return original(path, target)

        with mock.patch.object(campaign, "preflight", return_value=({}, {})), \

             mock.patch("oakbench.runlock.measurement_lock", locked), \

             mock.patch.object(Path, "rename", crash):

            with self.assertRaises(KeyboardInterrupt):

                campaign.run_campaign(directory)

        plan = campaign.load_plan(directory)

        events = campaign.load_events(directory, plan)

        self.assertEqual(len(events), 1)

        state = campaign.campaign_status(plan, events)

        self.assertEqual(state["interrupted_trial_ids"], [plan["schedule"][0]["trial_id"]])

        self.assertNotIn(plan["schedule"][0]["trial_id"], state["pending_trial_ids"])

    def test_schema_rejected_row_remains_in_terminal_evidence(self):

        import agent_workflow

        from contextlib import contextmanager

        from oakbench.runlock import LockInfo

        plan = self.plan()

        directory = self.root / "campaign"

        @contextmanager

        def locked(label):

            yield LockInfo(0, True)

        subjects = {trial["subject"]: object() for trial in plan["schedule"]}

        with mock.patch.object(campaign, "preflight", return_value=({"mock": object()}, subjects)), \

             mock.patch("oakbench.runlock.measurement_lock", locked), \

             mock.patch.object(agent_workflow, "run_one", return_value={"invalid": "original evidence"}):

            campaign.run_campaign(directory, limit=1)

        events = campaign.load_events(directory, plan)

        self.assertEqual(events[1]["outcome"], "error")

        self.assertEqual(events[1]["rejected_row"]["invalid"], "original evidence")

        self.assertEqual((directory / "rows.jsonl").read_bytes(), b"")

    def test_noop_resume_repairs_missing_or_stale_rows_from_finished_journal(self):

        from agent_metrics_schema import example_row

        self.raw.update(subjects=["git"], repetitions=1)

        self.write_manifest()

        plan = self.plan()

        directory = self.root / "campaign"

        trial = plan["schedule"][0]

        import workflow_ab

        campaign.cached_fixture(directory, plan, workflow_ab.WORKFLOWS[trial["workflow"]])

        record = campaign.read_json(directory / "fixture-integrity" / (trial["workflow"] + ".json"))

        fingerprint = campaign.digest(record["manifest"])

        row = example_row()

        row.update(task_id=trial["trial_id"], bench_id=plan["campaign_id"], subject=trial["subject"],

                   scenario=trial["workflow"], run=trial["repetition"], outcome="pass")

        row["agent"].update(name=trial["agent"], model=trial["model"],

                             instruction_level=trial["instruction_level"], environment="minimal")

        campaign.append_event(directory, plan, {"event": "started", "trial_id": trial["trial_id"], "timeout_seconds": 60, "fixture_manifest_sha256": fingerprint})

        campaign.append_event(directory, plan, {"event": "finished", "trial_id": trial["trial_id"],

            "outcome": "pass", "row": row, "wall_seconds": 1, "fixture_manifest_sha256": fingerprint})

        journal = (directory / "events.jsonl").read_bytes()

        for state in ("missing", "stale"):

            with self.subTest(state=state):

                if state == "stale":

                    (directory / "rows.jsonl").write_bytes(b"stale output")

                with mock.patch.object(campaign, "preflight", return_value=({}, {})), \

                     mock.patch("agent_workflow.run_one") as run:

                    status = campaign.run_campaign(directory, resume=True)

                    run.assert_not_called()

                self.assertEqual(status["finished"], 1)

                self.assertEqual((directory / "rows.jsonl").read_bytes(), campaign.canonical(row) + b"\n")

                self.assertEqual((directory / "events.jsonl").read_bytes(), journal)

    def test_disabled_measurement_lock_starts_nothing(self):

        self.plan()

        directory = self.root / "campaign"

        with mock.patch.object(campaign, "preflight", return_value=({}, {})), \

             mock.patch.dict(os.environ, {"OAK_BENCH_LOCK": "off"}):

            with self.assertRaises(campaign.CampaignError):

                campaign.run_campaign(directory)

        self.assertFalse((directory / "events.jsonl").exists())

    def test_artifact_hashes_exclude_links(self):

        trial = self.root / "trial"

        (trial / "artifacts").mkdir(parents=True)

        (trial / "artifacts/prompt.txt").write_text("fixed prompt")

        (trial / "artifacts/escape").symlink_to(self.manifest_path)

        hashes = campaign.artifact_hashes(trial)

        self.assertEqual(set(hashes), {"artifacts/prompt.txt"})

        self.assertEqual(hashes["artifacts/prompt.txt"], campaign.file_hash(trial / "artifacts/prompt.txt"))

if __name__ == "__main__":

    unittest.main()