"""Frozen local agent campaigns with durable, non-retrying trial journals.

The manifest attestation is an operator assertion, not an OS sandbox. Episode

wall budgets control admission and the next agent timeout; setup/oracle overhead

can exceed that budget. No dollar or provider-token ceiling is claimed.

"""

from __future__ import annotations

import argparse

import contextlib

from dataclasses import replace

from datetime import datetime, timezone

import fcntl

import hashlib

import json

import math

import os

from pathlib import Path

import random

import re

import sys

import time

import tempfile

from typing import Any

SCHEMA_VERSION = 1

WORKFLOWS = {"wide_config_refactor", "large_asset_manifest"}

LEVELS = {"zero-shot", "cheat-sheet", "full-docs"}

ROOT = Path(__file__).resolve().parents[2]

class CampaignError(ValueError):

    pass

def now() -> str:

    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")

def canonical(value: Any) -> bytes:

    return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()

def digest(value: Any) -> str:

    return hashlib.sha256(canonical(value)).hexdigest()

def file_hash(path: Path) -> str:

    hasher = hashlib.sha256()

    with path.open("rb") as stream:

        for chunk in iter(lambda: stream.read(1024 * 1024), b""):

            hasher.update(chunk)

    return hasher.hexdigest()

def _unique_object(pairs):

    data = {}

    for key, value in pairs:

        if key in data:

            raise CampaignError("duplicate JSON key: " + key)

        data[key] = value

    return data

def read_json(path: Path) -> Any:

    try:

        return json.loads(path.read_text(), object_pairs_hook=_unique_object,

                          parse_constant=lambda value: (_ for _ in ()).throw(CampaignError("nonfinite JSON")))

    except (OSError, ValueError) as exc:

        raise CampaignError(f"cannot read {path}: {exc}") from exc

def _positive(value, name, integer=False):

    valid = type(value) in ((int,) if integer else (int, float))

    try:

        valid = valid and value > 0 and (integer or math.isfinite(value))

    except OverflowError:

        valid = False

    if not valid:

        raise CampaignError(name + " must be positive" + (" integer" if integer else " finite number"))

def _names(value, name, allowed=None):

    if not isinstance(value, list) or not value or any(not isinstance(x, str) or not x for x in value):

        raise CampaignError(name + " must be a nonempty string array")

    if len(set(value)) != len(value):

        raise CampaignError("duplicate " + name)

    if allowed is not None and not set(value) <= allowed:

        raise CampaignError("unsupported " + name + ": " + ", ".join(sorted(set(value) - allowed)))

def load_manifest(path: Path) -> dict:

    return validate_manifest(read_json(path), path.resolve().parent)

def validate_manifest(raw, base: Path) -> dict:

    required = {"schema_version", "campaign_id", "agents_config", "subjects_config", "agents", "subjects",

                "workflows", "instruction_levels", "repetitions", "seed", "max_trials", "max_wall_seconds",

                "trial_timeout_seconds", "track", "agent_environment"}

    if not isinstance(raw, dict) or not required <= set(raw) or set(raw) - required - {"isolation_attestation", "output_limit_bytes"}:

        raise CampaignError("manifest has missing or unknown fields")

    if type(raw["schema_version"]) is not int or raw["schema_version"] != SCHEMA_VERSION:

        raise CampaignError("unsupported campaign schema_version")

    if not isinstance(raw["campaign_id"], str) or not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,79}", raw["campaign_id"]):

        raise CampaignError("invalid campaign_id")

    _names(raw["subjects"], "subjects")

    _names(raw["workflows"], "workflows", WORKFLOWS)

    _names(raw["instruction_levels"], "instruction_levels", LEVELS)

    if raw["track"] not in {"agent-default", "core-equivalent"} or raw["agent_environment"] != "minimal":

        raise CampaignError("campaigns require a supported track and minimal agent_environment")

    for key in ("repetitions", "max_trials"):

        _positive(raw[key], key, True)

    for key in ("trial_timeout_seconds", "max_wall_seconds"):

        _positive(raw[key], key)

    if type(raw["seed"]) is not int:

        raise CampaignError("seed must be an integer")

    if "output_limit_bytes" in raw:

        _positive(raw["output_limit_bytes"], "output_limit_bytes", True)

    agents = raw["agents"]

    if not isinstance(agents, list) or not agents:

        raise CampaignError("agents must be a nonempty array")

    for agent in agents:

        if not isinstance(agent, dict) or set(agent) != {"name", "model"}:

            raise CampaignError("each agent requires exactly name and model")

        if agent["model"] is not None and (not isinstance(agent["model"], str) or not agent["model"].strip()):

            raise CampaignError("model must be null or a nonempty string")

    _names([agent["name"] for agent in agents], "agents")

    for name in raw["subjects"] + [agent["name"] for agent in agents]:

        if not re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_-]{0,79}", name):

            raise CampaignError("agent and subject names must be safe slugs")

    raw = dict(raw)

    for key in ("agents_config", "subjects_config"):

        if not isinstance(raw[key], str) or not raw[key]:

            raise CampaignError(key + " must be a path")

        raw[key] = str((base / Path(raw[key]).expanduser()).resolve())

    return raw

def schedule(manifest: dict) -> list[dict]:

    size = (len(manifest["subjects"]) * len(manifest["agents"]) * len(manifest["workflows"])

            * len(manifest["instruction_levels"]) * manifest["repetitions"])

    if size > manifest["max_trials"] or size > 10000:

        raise CampaignError(f"schedule has {size} trials; exceeds manifest or 10000-trial safety limit")

    rows = []

    rng = random.Random(manifest["seed"])

    groups = [(workflow, agent, level) for workflow in manifest["workflows"]

              for agent in manifest["agents"] for level in manifest["instruction_levels"]]

    for repetition in range(manifest["repetitions"]):

        block_order = list(groups)

        rng.shuffle(block_order)

        for workflow, agent, level in block_order:

            identity = {"campaign_id": manifest["campaign_id"], "workflow": workflow,

                        "agent": agent["name"], "model": agent["model"], "instruction_level": level}

            order = list(manifest["subjects"])

            random.Random(str(manifest["seed"]) + digest(identity)).shuffle(order)

            shift = repetition % len(order)

            order = order[shift:] + order[:shift]

            block = "block-" + digest({**identity, "repetition": repetition})[:24]

            for subject in order:

                rows.append({**identity, "subject": subject, "repetition": repetition,

                             "block_id": block, "trial_id": "trial-" + digest([block, subject])[:24],

                             "ordinal": len(rows)})

    if len(rows) > manifest["max_trials"]:

        raise CampaignError(f"schedule has {len(rows)} trials, exceeds max_trials={manifest['max_trials']}")

    return rows

def resolve_runtime(manifest: dict):

    import agent_workflow

    from oakbench.subjects import load_subjects, subject_details

    try:

        agents = agent_workflow.load_agents(Path(manifest["agents_config"]), ",".join(x["name"] for x in manifest["agents"]))

        subjects = load_subjects(argparse.Namespace(config=Path(manifest["subjects_config"]),

            subjects=",".join(manifest["subjects"]), git_bin=None, oak_installed_bin=None, oak_local_bin=None))

    except (SystemExit, KeyError, OSError) as exc:

        raise CampaignError("configuration preflight failed: " + str(exc)) from exc

    if {x.name for x in subjects} != set(manifest["subjects"]):

        raise CampaignError("unknown selected subject")

    requested = {x["name"]: x["model"] for x in manifest["agents"]}

    if {agent.name for agent in agents} != set(requested):

        raise CampaignError("selected agent names must identify exact configured entries")

    agents = [replace(agent, model=requested[agent.name]) for agent in agents]

    for agent in agents:

        if agent.adapter not in {"deterministic_mock", "codex_exec_json", "claude_print_stream_json", "cursor_agent_stream_json"}:

            raise CampaignError("unsupported agent adapter: " + agent.adapter)

        if agent.adapter == "deterministic_mock" and agent.model is not None:

            raise CampaignError("mock agents require model=null")

        if agent.adapter != "deterministic_mock" and not agent.model:

            raise CampaignError("real agents require an explicit requested model: " + agent.name)

    for subject in subjects:

        if subject.kind not in {"git", "oak"} or subject.git_mode is not None:

            raise CampaignError("only local Git/Oak subjects are supported")

    for binary in [x.bin for x in agents if x.bin is not None] + [x.bin for x in subjects]:

        if not binary.is_file() or not os.access(binary, os.X_OK):

            raise CampaignError("not an executable binary: " + str(binary))

    from oakbench.runlock import measurement_lock_path

    provenance = {

        "measurement_lock_path": str(measurement_lock_path()),

        "configs": {key: {"path": manifest[key], "sha256": file_hash(Path(manifest[key]))}

                    for key in ("agents_config", "subjects_config")},

        "subjects": subject_details(subjects),

        "agents": {agent.name: {"adapter": agent.adapter, "provider": agent.provider, "model": agent.model,

                    "version": agent.version, "bin": str(agent.bin.resolve()) if agent.bin else None,

                    "binary_sha256": file_hash(agent.bin) if agent.bin else None} for agent in agents},

        "harness": {str(path.relative_to(ROOT)): file_hash(path) for path in sorted((ROOT / "scripts").rglob("*.py"))},

        "configuration_and_prompts": {

            str(path.relative_to(ROOT)): file_hash(path)

            for directory in (ROOT / "config", ROOT / "scenarios")

            for path in sorted(directory.rglob("*")) if path.is_file() and not any(part.startswith(".") for part in path.relative_to(directory).parts)

        },

    }

    return {x.name: x for x in agents}, {x.name: x for x in subjects}, provenance

def safe_path(path: Path) -> Path:

    if ".." in path.parts:

        raise CampaignError("destination path traversal is not allowed")

    absolute = path.absolute()

    for item in [absolute, *absolute.parents]:

        if item.is_symlink() and not (str(item) in {"/tmp", "/var"} and str(item.resolve()) == "/private" + str(item)):

            raise CampaignError("destination symlink is not allowed: " + str(item))

    return absolute.resolve()

def artifact_hashes(trial_root: Path) -> dict[str, str]:

    candidates = list((trial_root / "artifacts").rglob("*"))

    candidates.extend((trial_root / "work").rglob("AGENTS.md"))

    result = {}

    for path in candidates:

        if path.is_symlink():

            continue

        if path.is_file():

            try:

                safe_path(path)

                result[path.relative_to(trial_root).as_posix()] = file_hash(path)

            except (OSError, CampaignError):

                continue

    return result

def _sync_directory(path):

    fd = os.open(path, os.O_RDONLY)

    try:

        os.fsync(fd)

    finally:

        os.close(fd)

def atomic_write(path: Path, content: bytes):

    safe_path(path)

    fd, name = tempfile.mkstemp(prefix=path.name + ".tmp-", dir=path.parent)

    temporary = Path(name)

    with os.fdopen(fd, "wb") as stream:

        stream.write(content)

        stream.flush()

        os.fsync(stream.fileno())

    os.replace(temporary, path)

    _sync_directory(path.parent)

def create_plan(manifest_path: Path, directory: Path) -> dict:

    manifest = load_manifest(manifest_path)

    trials = schedule(manifest)

    _, _, provenance = resolve_runtime(manifest)

    from oakbench.runner import runner_fields

    plan = {"schema_version": 1, "campaign_id": manifest["campaign_id"], "created_at": now(),

            "manifest": manifest, "provenance": provenance, "runner": runner_fields(), "schedule": trials,

            "budget_policy": "cumulative episode-wall admission; interrupted episodes reserve their timeout; setup/oracle overhead may exceed cap"}

    plan["plan_sha256"] = digest(plan)

    directory = safe_path(directory)

    if directory.exists():

        raise CampaignError("campaign directory already exists; refusing overwrite: " + str(directory))

    try:

        directory.mkdir(parents=True)

    except FileExistsError as exc:

        raise CampaignError("campaign directory collision") from exc

    atomic_write(directory / "plan.json", canonical(plan) + b"\n")

    return plan

def load_plan(directory: Path) -> dict:

    safe_path(directory / "plan.json")

    plan = read_json(directory / "plan.json")

    if not isinstance(plan, dict) or type(plan.get("schema_version")) is not int or plan.get("schema_version") != 1:

        raise CampaignError("invalid frozen plan")

    body = {key: value for key, value in plan.items() if key != "plan_sha256"}

    if digest(body) != plan.get("plan_sha256"):

        raise CampaignError("frozen plan fingerprint mismatch")

    try:

        for key in ("campaign_id", "manifest", "schedule", "provenance", "runner"):

            if key not in plan:

                raise CampaignError("invalid frozen plan: missing " + key)

        if not isinstance(plan["provenance"], dict) or not isinstance(plan["runner"], dict):

            raise CampaignError("invalid frozen plan metadata")

        manifest = validate_manifest(plan["manifest"], directory.resolve())

        if manifest != plan["manifest"] or plan["campaign_id"] != manifest["campaign_id"]:

            raise CampaignError("invalid frozen manifest identity or paths")

        if not all(isinstance(plan["runner"].get(key), str) for key in ("runner_id", "runner_class")):

            raise CampaignError("invalid frozen runner identity")

        if schedule(manifest) != plan["schedule"]:

            raise CampaignError("frozen schedule mismatch")

    except (KeyError, TypeError, AttributeError) as exc:

        raise CampaignError("invalid frozen plan structure") from exc

    return plan

def validate_trial_row(row: dict, trial: dict, plan: dict):

    import agent_workflow

    if not isinstance(row, dict) or agent_workflow.validate_row(row):

        raise CampaignError("invalid journal terminal row schema")

    expected = {"task_id": trial["trial_id"], "bench_id": plan["campaign_id"],

                "subject": trial["subject"], "scenario": trial["workflow"], "run": trial["repetition"]}

    if any(row.get(key) != value for key, value in expected.items()):

        raise CampaignError("terminal row does not match frozen trial identity")

    expected_agent = {"name": trial["agent"], "model": trial["model"],

                      "instruction_level": trial["instruction_level"],

                      "environment": plan["manifest"]["agent_environment"]}

    pinned_agent = plan["provenance"].get("agents", {}).get(trial["agent"], {})

    expected_agent.update({key: pinned_agent[key] for key in ("adapter", "provider", "version") if key in pinned_agent})

    if any(row["agent"].get(key) != value for key, value in expected_agent.items()):

        raise CampaignError("terminal row agent does not match frozen trial identity")

def load_events(directory: Path, plan: dict, *, tolerate_torn=False) -> list[dict]:

    path = safe_path(directory / "events.jsonl")

    if not path.exists():

        return []

    raw = path.read_bytes()

    tail = b""

    if raw and not raw.endswith(b"\n"):

        if not tolerate_torn:

            raise CampaignError("journal has incomplete final record; resume requires --acknowledge-torn-journal; affected trials are never retried")

        prefix_end = raw.rfind(b"\n") + 1

        raw, tail = raw[:prefix_end], raw[prefix_end:]

    events = []

    known = {trial["trial_id"]: trial for trial in plan["schedule"]}

    started, finished = set(), set()

    for line in raw.splitlines():

        try:

            event = json.loads(line, object_pairs_hook=_unique_object,

                parse_constant=lambda value: (_ for _ in ()).throw(CampaignError("nonfinite journal JSON")))

        except ValueError as exc:

            raise CampaignError("invalid journal record") from exc

        if not isinstance(event, dict):

            raise CampaignError("journal record must be an object")

        if event.get("event") == "journal_recovery":

            if event.get("plan_sha256") != plan["plan_sha256"] or not isinstance(event.get("tail_sha256"), str):

                raise CampaignError("invalid journal recovery marker")

            events.append(event)

            continue

        trial = event.get("trial_id")

        if not isinstance(trial, str):

            raise CampaignError("journal trial_id must be a string")

        if trial not in known or event.get("plan_sha256") != plan["plan_sha256"]:

            raise CampaignError("journal trial or plan mismatch")

        if event.get("event") == "started" and trial not in started:

            _positive(event.get("timeout_seconds"), "journal timeout_seconds")

            started.add(trial)

        elif event.get("event") == "finished" and trial in started and trial not in finished:

            duration = event.get("wall_seconds")

            if type(duration) not in (int, float) or not math.isfinite(duration) or duration < 0:

                raise CampaignError("journal wall_seconds must be finite and nonnegative")

            if event.get("outcome") not in {"pass", "fail", "timeout", "error"}:

                raise CampaignError("invalid journal terminal outcome")

            if event["outcome"] != "error" or event.get("row") is not None:

                row = event.get("row")

                validate_trial_row(row, known[trial], plan)

                if row.get("outcome") != event["outcome"]:

                    raise CampaignError("journal outcome contradicts trial row")

            finished.add(trial)

        else:

            raise CampaignError("duplicate or out-of-order journal event")

        events.append(event)

    if tail:

        # Only the most recent intact start can own this torn terminal.

        # Older interrupted trials do not account for a newly torn start.

        active = bool(events and events[-1].get("event") == "started")

        candidate = next((trial for trial in known if trial not in started), None)

        if not active and candidate:

            events.append({"event": "started", "trial_id": candidate, "plan_sha256": plan["plan_sha256"],

                           "timeout_seconds": plan["manifest"]["trial_timeout_seconds"], "uncertain_torn_start": True})

        events.append({"event": "journal_recovery", "plan_sha256": plan["plan_sha256"],

                       "tail_sha256": hashlib.sha256(tail).hexdigest(), "tail_bytes": len(tail),

                       "acknowledged": False})

    return events

def acknowledge_torn_journal(directory, plan):

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

    if not events or events[-1].get("event") != "journal_recovery" or events[-1].get("acknowledged") is not False:

        return events

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

    archive = directory / ("events-before-recovery-" + hashlib.sha256(raw).hexdigest() + ".jsonl")

    if not archive.exists():

        atomic_write(archive, raw)

    elif safe_path(archive).read_bytes() != raw:

        raise CampaignError("recovery archive collision")

    events[-1].update(acknowledged=True, archive=archive.name, archive_sha256=hashlib.sha256(raw).hexdigest(), timestamp_utc=now())

    atomic_write(directory / "events.jsonl", b"".join(canonical(event) + b"\n" for event in events))

    return events

def append_event(directory: Path, plan: dict, event: dict):

    record = {**event, "plan_sha256": plan["plan_sha256"], "timestamp_utc": now()}

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

        stream.write(canonical(record) + b"\n")

        stream.flush()

        os.fsync(stream.fileno())

    _sync_directory(directory)

    return record

def campaign_status(plan: dict, events: list[dict]) -> dict:

    starts = {x["trial_id"]: x for x in events if x["event"] == "started"}

    finishes = {x["trial_id"]: x for x in events if x["event"] == "finished"}

    interrupted = sorted(set(starts) - set(finishes))

    charged = sum(float(x["wall_seconds"]) for x in finishes.values())

    charged += sum(float(starts[key]["timeout_seconds"]) for key in interrupted)

    return {"campaign_id": plan["campaign_id"], "plan_sha256": plan["plan_sha256"],

            "scheduled": len(plan["schedule"]), "finished": len(finishes),

            "journal_recoveries": [x for x in events if x["event"] == "journal_recovery"],

            "interrupted_trial_ids": interrupted,

            "pending_trial_ids": [x["trial_id"] for x in plan["schedule"] if x["trial_id"] not in starts],

            "charged_episode_seconds": charged,

            "remaining_episode_seconds": max(0.0, plan["manifest"]["max_wall_seconds"] - charged)}

@contextlib.contextmanager

def campaign_lock(directory):

    with safe_path(directory / ".run.lock").open("a") as stream:

        try:

            fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)

        except BlockingIOError as exc:

            raise CampaignError("campaign already has an active runner") from exc

        try:

            yield

        finally:

            fcntl.flock(stream, fcntl.LOCK_UN)

def preflight(plan: dict, directory: Path, enable_real=False):

    agents, subjects, provenance = resolve_runtime(plan["manifest"])

    if provenance != plan["provenance"]:

        changed = sorted(key for key in set(provenance) | set(plan["provenance"])

                         if provenance.get(key) != plan["provenance"].get(key))

        raise CampaignError("provenance drift in " + ", ".join(changed) + "; create a new campaign")

    from oakbench.runner import runner_fields

    current = runner_fields()

    if any(current[key] != plan["runner"][key] for key in ("runner_id", "runner_class")):

        raise CampaignError("runner identity drift; create a new campaign")

    real = any(agent.adapter != "deterministic_mock" for agent in agents.values())

    if real:

        attestation = plan["manifest"].get("isolation_attestation")

        if not enable_real:

            raise CampaignError("real agents require --enable-real; no paid invocation has started")

        if not isinstance(attestation, dict) or set(attestation) != {"worker_root", "unrestricted_cli_acknowledged"}:

            raise CampaignError("real agents require isolated-worker attestation; it is not a sandbox")

        if attestation["unrestricted_cli_acknowledged"] is not True or not Path(attestation["worker_root"]).is_absolute():

            raise CampaignError("invalid isolated-worker attestation")

        if not directory.resolve().is_relative_to(Path(attestation["worker_root"]).resolve()):

            raise CampaignError("campaign directory is outside attested worker root")

    return agents, subjects

def cached_fixture(directory: Path, plan: dict, workflow) -> Path:

    """Generate once; verify the complete cached input before each invocation.

    The campaign writer lock protects this cache. Each trial still receives its

    own copy in run_one. Host-level containment is an operator responsibility.

    """

    import workflow_ab

    from oakbench.conformance import tree_manifest

    fixture = safe_path(directory / "fixtures" / workflow.name)

    record_path = safe_path(directory / "fixture-integrity" / (workflow.name + ".json"))

    expected_identity = {"plan_sha256": plan["plan_sha256"], "fixture_version": workflow_ab.FIXTURE_VERSION}

    if fixture.exists() or record_path.exists():

        if not fixture.is_dir() or not record_path.is_file():

            raise CampaignError("fixture cache is incomplete; preserve evidence and create a new campaign")

        record = read_json(record_path)

        if not isinstance(record, dict) or any(record.get(key) != value for key, value in expected_identity.items()):

            raise CampaignError("fixture cache identity drift")

        if any((fixture / name).exists() or (fixture / name).is_symlink() for name in (".git", ".oak")):

            raise CampaignError("fixture content drift: unexpected repository metadata")

        if tree_manifest(fixture) != record.get("manifest"):

            raise CampaignError("fixture content drift; refusing reuse of changed cached input")

        return fixture

    generated = workflow_ab.prepare_fixture(directory, workflow)

    if generated != fixture:

        raise CampaignError("fixture generator returned an unexpected cache path")

    record_path.parent.mkdir(parents=True, exist_ok=True)

    atomic_write(record_path, canonical({**expected_identity, "manifest": tree_manifest(fixture)}) + b"\n")

    return fixture

def run_campaign(directory: Path, *, resume=False, enable_real=False, limit=None, acknowledge_torn=False) -> dict:

    import agent_workflow

    import workflow_ab

    from oakbench.runlock import measurement_lock

    directory = safe_path(directory)

    plan = load_plan(directory)

    with campaign_lock(directory):

        agents, subjects = preflight(plan, directory, enable_real)

        events = acknowledge_torn_journal(directory, plan) if resume and acknowledge_torn else load_events(directory, plan)

        if events and not resume:

            raise CampaignError("campaign already started; use resume (interrupted trials are not retried)")

        if limit is not None:

            _positive(limit, "limit", True)

        initial_state = campaign_status(plan, events)

        pending = set(initial_state["pending_trial_ids"])

        remaining = initial_state["remaining_episode_seconds"]

        fixtures = {}

        completed_now = 0

        stopped_reason = None

        rows = [event["row"] for event in events if event.get("row") is not None]

        atomic_write(directory / "rows.jsonl", b"".join(canonical(row) + b"\n" for row in rows))

        for trial in plan["schedule"]:

            if trial["trial_id"] not in pending:

                continue

            timeout = plan["manifest"]["trial_timeout_seconds"]

            if remaining < timeout:

                stopped_reason = "insufficient_budget_for_frozen_trial_timeout"

                break

            if limit is not None and completed_now >= limit:

                stopped_reason = "invocation_trial_limit"

                break

            trial_root = safe_path(directory / "trials" / trial["trial_id"])

            if trial_root.exists():

                raise CampaignError("trial path collision; refusing overwrite: " + str(trial_root))

            with measurement_lock("agent-campaign:" + trial["trial_id"]) as lock:

                if not lock.enabled:

                    raise CampaignError("global measurement lock is mandatory for campaigns")

                pinned_lock = plan["provenance"].get("measurement_lock_path")

                if pinned_lock is not None and lock.path != pinned_lock:

                    raise CampaignError("acquired measurement lock differs from frozen provenance")

                started = append_event(directory, plan, {"event": "started", "trial_id": trial["trial_id"],

                    "timeout_seconds": timeout, "ordinal": trial["ordinal"]})

                events.append(started)

                start = time.monotonic()

                print(f"[trial] {trial['ordinal'] + 1}/{len(plan['schedule'])} {trial['trial_id']} {trial['agent']}/{trial['subject']}", file=sys.stderr, flush=True)

                row = None

                try:

                    trial_root.mkdir(parents=True)

                    workflow = workflow_ab.WORKFLOWS[trial["workflow"]]

                    if workflow.name not in fixtures:

                        fixtures[workflow.name] = cached_fixture(directory, plan, workflow)

                    fixture = fixtures[workflow.name]

                    row = agent_workflow.run_one(

                        agent=agents[trial["agent"]], subject=subjects[trial["subject"]], workflow=workflow,

                        fixture=fixture, run_index=trial["repetition"], run_root=trial_root / "work",

                        artifacts_root=trial_root / "artifacts", timeout_seconds=timeout,

                        track=plan["manifest"]["track"], agent_environment=plan["manifest"]["agent_environment"],

                        instruction_level=trial["instruction_level"], bench_id=plan["campaign_id"],

                        timestamp_utc=now(), source={}, host_meta={},

                        output_limit_bytes=plan["manifest"].get("output_limit_bytes", 8 * 1024 * 1024),

                    )

                    row["task_id"] = trial["trial_id"]

                    validate_trial_row(row, trial, plan)

                    terminal = {"event": "finished", "trial_id": trial["trial_id"], "outcome": row["outcome"], "row": row}

                except (Exception, SystemExit) as exc:

                    terminal = {"event": "finished", "trial_id": trial["trial_id"], "outcome": "error",

                                "error": type(exc).__name__ + ": " + str(exc)}

                    if row is not None:

                        try:

                            canonical(row)

                            terminal["rejected_row"] = row

                        except (ValueError, TypeError):

                            terminal["rejected_row_repr"] = repr(row)

                terminal["wall_seconds"] = time.monotonic() - start

                remaining = max(0.0, remaining - terminal["wall_seconds"])

                terminal["artifact_sha256"] = artifact_hashes(trial_root)

                terminal["artifact_integrity_source"] = "sha256_of_regular_trial_artifacts_and_instructions;links_and_missing_files_not_attested"

                terminal["artifact_integrity_status"] = "observed_files_only" if terminal["artifact_sha256"] else "unavailable"

                events.append(append_event(directory, plan, terminal))

                if terminal.get("row") is not None:

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

                        stream.write(canonical(terminal["row"]) + b"\n")

                        stream.flush()

                        os.fsync(stream.fileno())

                completed_now += 1

        # The journal is authoritative. A crash after its final fsync may leave

        # derived rows absent/stale even when there are no pending trials.

        rows = [event["row"] for event in events if event.get("row") is not None]

        atomic_write(directory / "rows.jsonl", b"".join(canonical(row) + b"\n" for row in rows))

        return {**campaign_status(plan, events), "stopped_reason": stopped_reason}