bundle/v0/tests/test_batch_control.py

from __future__ import annotations

import copy
import json
import hashlib
import math
import subprocess
import sys
import threading
import time
from pathlib import Path

import pytest
import loss_aware_dro_repro.batch_control as batch_control
import scripts.run_local_batch as batch_cli

from loss_aware_dro_repro.batch_control import (
    STOPPING_SAMPLE_EXECUTION_REQUEST,
    _record_success,
    _task_runtime_contract,
    aggregate_validated_successes,
    build_batch_manifest,
    build_stopping_sample_batch_manifest,
    completion_projection_gate_v1,
    lane_file_snapshot,
    reserve_task,
    run_local_batch,
    validate_batch_manifest,
    validate_success_output,
    verify_live_binding,
    write_batch_plan,
)
from loss_aware_dro_repro.core import ContractError, canonical_bytes, load_plan, sha256_value
from loss_aware_dro_repro.matrix import expand_tasks
from loss_aware_dro_repro.task_dispatch import run_task
from scripts.run_local_batch import _result_exit_code


TASKS = [
    "portfolio_gaussian_main/d001/r00/n010",
    "portfolio_gaussian_main/d001/r00/n020",
    "portfolio_gaussian_main/d001/r00/n030",
    "portfolio_gaussian_main/d001/r00/n040",
]


def _snapshot(commit: str = "a" * 40):
    source = lane_file_snapshot()
    return {
        "commit": commit,
        "clean": True,
        "lane_source_hash": source["hash"],
        "files": source["files"],
    }


def _manifest(task_ids=None, workers=2):
    manifest, receipt = build_batch_manifest(
        task_ids=task_ids or TASKS[:1],
        worker_count=workers,
        snapshot_provider=_snapshot,
    )
    return manifest, receipt


_POLICY_MANIFEST_CACHE = None


def _policy_manifest(workers=8):
    global _POLICY_MANIFEST_CACHE
    assert workers == 8
    if _POLICY_MANIFEST_CACHE is None:
        _POLICY_MANIFEST_CACHE = build_stopping_sample_batch_manifest(
            snapshot_provider=_snapshot,
        )
    return copy.deepcopy(_POLICY_MANIFEST_CACHE)


class _GateClock:
    def __init__(self, elapsed):
        self.values = iter((10_000.0, 10_000.0 + elapsed))

    def __call__(self):
        return next(self.values)


def _fake_success_validator(output_dir, task_entry, frozen_manifest):
    return {
        "result": {
            "task_id": task_entry["task_id"],
            "run_identity": task_entry["scientific_run_identity"],
        },
        "result_sha256": task_entry["task_hash"].removeprefix("sha256:"),
        "artifact_hashes": {},
    }


def _write_manifest(tmp_path: Path, manifest: dict, receipt: dict):
    manifest_path = tmp_path / "plan" / "manifest.json"
    receipt_path = tmp_path / "plan" / "receipt.json"
    write_batch_plan(manifest_path, receipt_path, manifest, receipt)
    return manifest_path, receipt_path


def test_manifest_subset_shard_order_and_dry_run_receipt(tmp_path):
    manifest, receipt = build_batch_manifest(
        task_ids=list(reversed(TASKS)),
        shard_count=2,
        shard_index=0,
        worker_count=3,
        snapshot_provider=_snapshot,
    )
    assert [row["task_id"] for row in manifest["tasks"]] == [TASKS[0], TASKS[2]]
    assert receipt["projected_pre_shard_task_count"] == 4
    assert receipt["projected_task_count"] == 2
    assert receipt["worker_count"] == 3
    assert receipt["scientific_tasks_launched"] == 0
    assert all(value is False for value in receipt["authority"].values())
    manifest_path, receipt_path = _write_manifest(tmp_path, manifest, receipt)
    stored_receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
    assert stored_receipt["manifest_file_sha256"]
    with pytest.raises(ContractError, match="must both be new"):
        write_batch_plan(manifest_path, tmp_path / "second.json", manifest, receipt)


def test_lane_snapshot_binds_stopping_sample_design():
    assert (
        "configs/stopping_storage_sample_v1.json"
        in lane_file_snapshot()["files"]
    )


def test_dedicated_stopping_sample_plan_binds_exact_sample_and_request():
    from loss_aware_dro_repro.stopping_sample import select_stopping_sample

    manifest, receipt = _policy_manifest()
    binding = manifest["stopping_sample"]
    expected_ids = [task["task_id"] for task in select_stopping_sample()]
    config_path = Path(__file__).parents[1] / binding["sample_config_path"]

    assert manifest["execution"]["worker_count"] == 8
    assert (
        manifest["execution"]["requested_execution_config"]
        == STOPPING_SAMPLE_EXECUTION_REQUEST
    )
    assert binding["sample_id"] == "loss-aware-stopping-storage-sample-v1"
    assert binding["ordered_task_ids"] == expected_ids
    assert binding["ordered_task_id_list_sha256"] == hashlib.sha256(
        json.dumps(expected_ids, separators=(",", ":")).encode("utf-8")
    ).hexdigest()
    assert binding["sample_config_file_sha256"] == hashlib.sha256(
        config_path.read_bytes()
    ).hexdigest()
    assert [row["task_id"] for row in manifest["tasks"]] == sorted(expected_ids)
    assert receipt["stopping_sample"] == binding
    assert receipt["projected_task_count"] == 400


def test_generic_completion_gate_cannot_bind_an_arbitrary_400_task_set():
    task_ids = sorted(task["task_id"] for task in expand_tasks(load_plan()))[:400]
    with pytest.raises(ContractError, match="stopping sample launch binding"):
        build_batch_manifest(
            task_ids=task_ids,
            worker_count=8,
            requested_execution_config=dict(STOPPING_SAMPLE_EXECUTION_REQUEST),
            issuance_policy=completion_projection_gate_v1(8),
            snapshot_provider=_snapshot,
        )


@pytest.mark.parametrize(
    ("field", "replacement"),
    [
        ("sample_id", "forged-sample"),
        ("sample_config_file_sha256", "0" * 64),
        ("ordered_task_id_list_sha256", "0" * 64),
    ],
)
def test_manifest_validation_recomputes_stopping_sample_binding(field, replacement):
    manifest, _ = _policy_manifest()
    manifest["stopping_sample"][field] = replacement
    manifest["manifest_hash"] = sha256_value(
        {key: value for key, value in manifest.items() if key != "manifest_hash"}
    )
    with pytest.raises(ContractError, match="frozen sample"):
        validate_batch_manifest(manifest)


def test_live_binding_recomputes_stopping_sample_semantics(monkeypatch):
    manifest, _ = _policy_manifest()
    drifted = {
        **manifest["stopping_sample"],
        "sample_config_file_sha256": "0" * 64,
    }
    monkeypatch.setattr(
        batch_control, "_expected_stopping_sample_binding", lambda plan: drifted
    )
    with pytest.raises(ContractError, match="frozen sample"):
        verify_live_binding(
            manifest, snapshot_provider=lambda: manifest["repository"]
        )


def test_stopping_sample_gate_rejects_request_worker_and_membership_drift():
    valid, _ = _policy_manifest()
    binding = valid["stopping_sample"]
    task_ids = list(binding["ordered_task_ids"])

    with pytest.raises(ContractError, match="exactly 8 workers"):
        build_batch_manifest(
            task_ids=task_ids,
            worker_count=7,
            requested_execution_config=dict(STOPPING_SAMPLE_EXECUTION_REQUEST),
            issuance_policy=completion_projection_gate_v1(7),
            stopping_sample_binding=binding,
            snapshot_provider=_snapshot,
        )
    with pytest.raises(ContractError, match="exact execution request"):
        build_batch_manifest(
            task_ids=task_ids,
            worker_count=8,
            requested_execution_config={
                **STOPPING_SAMPLE_EXECUTION_REQUEST,
                "max_outer_iterations": 4999,
            },
            issuance_policy=completion_projection_gate_v1(8),
            stopping_sample_binding=binding,
            snapshot_provider=_snapshot,
        )
    replacement = next(
        task["task_id"]
        for task in expand_tasks(load_plan())
        if task["task_id"] not in set(task_ids)
    )
    with pytest.raises(ContractError, match="exact sample membership"):
        build_batch_manifest(
            task_ids=[*task_ids[:-1], replacement],
            worker_count=8,
            requested_execution_config=dict(STOPPING_SAMPLE_EXECUTION_REQUEST),
            issuance_policy=completion_projection_gate_v1(8),
            stopping_sample_binding=binding,
            snapshot_provider=_snapshot,
        )


def test_write_plan_rejects_receipt_without_exact_sample_binding(tmp_path):
    manifest, receipt = _policy_manifest()
    receipt["stopping_sample"] = {
        **receipt["stopping_sample"],
        "ordered_task_id_list_sha256": "0" * 64,
    }
    with pytest.raises(ContractError, match="receipt stopping sample"):
        write_batch_plan(
            tmp_path / "manifest.json", tmp_path / "receipt.json", manifest, receipt
        )


def test_duplicate_reservation_is_denied_and_ledger_is_append_only(tmp_path):
    manifest, _ = _manifest()
    task = manifest["tasks"][0]
    control = tmp_path / "control"
    reserve_task(control, manifest, task)
    with pytest.raises(ContractError, match="already exists"):
        reserve_task(control, manifest, task)
    lines = (control / "launch-ledger.jsonl").read_text(encoding="utf-8").splitlines()
    assert len(lines) == 1
    assert json.loads(lines[0])["event"] == "reserved"


@pytest.mark.parametrize("field", ["commit", "lane_source_hash", "files"])
def test_live_binding_rejects_commit_or_source_drift(field):
    manifest, _ = _manifest()
    drifted = _snapshot()
    if field == "commit":
        drifted[field] = "b" * 40
    elif field == "lane_source_hash":
        drifted[field] = "sha256:" + "0" * 64
    else:
        drifted[field] = {**drifted[field], "forged.py": "0" * 64}
    with pytest.raises(ContractError, match="source drift"):
        verify_live_binding(manifest, snapshot_provider=lambda: drifted)


@pytest.mark.parametrize("mutation", ["uppercase_commit", "dirty", "bad_source_hash", "empty_files"])
def test_manifest_rejects_malformed_repository_binding(mutation):
    manifest, _ = _manifest()
    if mutation == "uppercase_commit":
        manifest["repository"]["commit"] = "A" * 40
    elif mutation == "dirty":
        manifest["repository"]["clean"] = False
    elif mutation == "bad_source_hash":
        manifest["repository"]["lane_source_hash"] = "sha256:" + "0" * 64
    else:
        manifest["repository"]["files"] = {}
    manifest["manifest_hash"] = sha256_value({key: value for key, value in manifest.items() if key != "manifest_hash"})
    with pytest.raises(ContractError, match="repository"):
        validate_batch_manifest(manifest)


def test_batch_contract_matches_real_runner_identity(tmp_path):
    from loss_aware_dro_repro.paper_task import RUNNER_SCHEMA_VERSION, _repository_state

    task_id = "portfolio_gaussian_main/d001/r00/n010"
    plan = load_plan()
    task = next(task for task in expand_tasks(plan) if task["task_id"] == task_id)
    requested = {
        "schema_version": RUNNER_SCHEMA_VERSION,
        "execution_scale": "test",
        "max_outer_iterations": 1,
    }
    commit, clean = _repository_state()
    runtime = _task_runtime_contract(
        task,
        plan,
        requested,
        {"commit": commit, "clean": clean},
    )
    result = run_task(
        task_id,
        tmp_path / "real-run",
        requested_execution_config=requested,
        plan=plan,
    )
    assert runtime["runner_schema_version"] == RUNNER_SCHEMA_VERSION
    assert runtime["scientific_run_identity"] == result["run_identity"]


def test_forged_success_is_rejected(tmp_path):
    manifest, _ = _manifest()
    task = manifest["tasks"][0]
    output = tmp_path / "forged"
    output.mkdir()
    (output / "result.json").write_text('{"status":"success"}\n', encoding="utf-8")
    (output / "lineage_receipt.json").write_text("{}\n", encoding="utf-8")
    with pytest.raises(ContractError, match="schema violation"):
        validate_success_output(output, task, manifest)


def test_partial_output_is_rejected(tmp_path):
    manifest, _ = _manifest()
    task = manifest["tasks"][0]
    output = tmp_path / "partial"
    output.mkdir()
    (output / "result.json").write_text("{}\n", encoding="utf-8")
    with pytest.raises(ContractError, match="partial task output"):
        validate_success_output(output, task, manifest)


def test_worker_failure_is_recorded_once_and_never_relaunched(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "batch"
    calls = []

    def failing_launcher(command, environment):
        calls.append(command)
        assert all(environment[name] == "1" for name in manifest["execution"]["thread_environment"])
        return subprocess.CompletedProcess(command, 9, stdout="worker output", stderr="failure")

    first = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=failing_launcher,
    )
    assert first["status_counts"] == {"failed": 1}
    assert len(calls) == 1
    failure_lines = (output / "control" / "failure-ledger.jsonl").read_text(encoding="utf-8").splitlines()
    assert len(failure_lines) == 1
    failure = json.loads(failure_lines[0])
    assert (output / "control" / failure["stdout_path"]).read_text(encoding="utf-8") == "worker output"
    assert (output / "control" / failure["stderr_path"]).read_text(encoding="utf-8") == "failure"
    assert first["artifact_hashes"][failure["stderr_path"]]

    task = manifest["tasks"][0]
    inserted = output / task["output_relpath"]
    inserted.mkdir(parents=True)
    (inserted / "result.json").write_text('{"status":"success"}\n', encoding="utf-8")

    def forbidden_launcher(command, environment):
        raise AssertionError("a failed scientific identity must never relaunch")

    def forbidden_validator(*args):
        raise AssertionError("a failed identity must be terminal before output validation")

    second = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=forbidden_launcher,
        success_validator=forbidden_validator,
    )
    assert second["status_counts"] == {"skipped_failed_identity": 1}
    assert len((output / "control" / "failure-ledger.jsonl").read_text(encoding="utf-8").splitlines()) == 1


def test_launcher_exception_records_terminal_failure_and_cannot_retry(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "launcher-exception"
    launches = 0

    def exploding_launcher(command, environment):
        nonlocal launches
        launches += 1
        raise RuntimeError("synthetic launcher failure")

    first = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=exploding_launcher,
    )
    assert first["status_counts"] == {"failed": 1}
    failure = json.loads(
        (output / "control" / "failure-ledger.jsonl")
        .read_text(encoding="utf-8")
        .splitlines()[0]
    )
    assert failure["returncode"] is None
    assert failure["reason"].startswith("launcher_exception:RuntimeError")
    assert "synthetic launcher failure" in (
        output / "control" / failure["stderr_path"]
    ).read_text(encoding="utf-8")

    second = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=lambda *_: (_ for _ in ()).throw(
            AssertionError("failed identity must not relaunch")
        ),
    )
    assert second["status_counts"] == {"skipped_failed_identity": 1}
    assert launches == 1


def test_unhandled_worker_future_exception_gets_failure_receipt(
    tmp_path, monkeypatch
):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "future-exception"

    def explode_after_validation(*args, **kwargs):
        raise RuntimeError("synthetic post-validation failure")

    monkeypatch.setattr(batch_control, "_record_success", explode_after_validation)
    summary = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=lambda command, environment: subprocess.CompletedProcess(
            command, 0, stdout="", stderr=""
        ),
        success_validator=_fake_success_validator,
    )
    assert summary["status_counts"] == {"failed": 1}
    failure = json.loads(
        (output / "control" / "failure-ledger.jsonl")
        .read_text(encoding="utf-8")
        .splitlines()[0]
    )
    assert failure["reason"].startswith("future_exception:RuntimeError")


def test_reserved_identity_skips_only_independently_validated_success(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "batch"
    task = manifest["tasks"][0]
    reserve_task(output / "control", manifest, task)

    def validated(output_dir, task_entry, frozen_manifest):
        return {
            "result": {"task_id": task_entry["task_id"], "run_identity": task_entry["scientific_run_identity"]},
            "result_sha256": "f" * 64,
            "artifact_hashes": {},
        }

    _record_success(
        output / "control",
        manifest,
        task,
        {"result_sha256": "f" * 64},
    )

    result = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=lambda *_: (_ for _ in ()).throw(AssertionError("must not launch")),
        success_validator=validated,
    )
    assert result["status_counts"] == {"skipped_validated_success": 1}


def test_aggregation_rejects_mismatched_validator_result(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)

    task = manifest["tasks"][0]
    reserve_task(tmp_path / "batch" / "control", manifest, task)
    success = {
        "schema_version": 1,
        "event": "success",
        "manifest_hash": manifest["manifest_hash"],
        "scientific_run_identity": task["scientific_run_identity"],
        "task_id": task["task_id"],
        "result_sha256": "0" * 64,
    }
    with (tmp_path / "batch" / "control" / "launch-ledger.jsonl").open("ab") as handle:
        handle.write(canonical_bytes(success) + b"\n")

    def mismatched(output_dir, task_entry, frozen_manifest):
        return {
            "result": {"task_id": "forged/task", "run_identity": task_entry["scientific_run_identity"]},
            "result_sha256": "0" * 64,
            "artifact_hashes": {},
        }

    aggregate = aggregate_validated_successes(
        manifest_path,
        tmp_path / "batch",
        success_validator=mismatched,
    )
    assert aggregate["validated_success_count"] == 0
    assert aggregate["complete"] is False
    assert len(aggregate["rejected"]) == 1
    assert "mismatched task" in aggregate["rejected"][0]["reason"]


def test_aggregation_rejects_valid_looking_output_without_launch_provenance(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)

    def valid_looking(output_dir, task_entry, frozen_manifest):
        return {
            "result": {"task_id": task_entry["task_id"], "run_identity": task_entry["scientific_run_identity"]},
            "result_sha256": "1" * 64,
            "artifact_hashes": {},
        }

    aggregate = aggregate_validated_successes(
        manifest_path,
        tmp_path / "batch",
        success_validator=valid_looking,
    )
    assert aggregate["validated_success_count"] == 0
    assert "reservation receipt" in aggregate["rejected"][0]["reason"]


def test_aggregation_requires_exactly_one_matching_success_and_no_failure(tmp_path):
    manifest, receipt = _manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "batch"
    task = manifest["tasks"][0]
    reserve_task(output / "control", manifest, task)
    success = {
        "schema_version": 1,
        "event": "success",
        "manifest_hash": manifest["manifest_hash"],
        "scientific_run_identity": task["scientific_run_identity"],
        "task_id": task["task_id"],
        "result_sha256": "2" * 64,
    }
    ledger = output / "control" / "launch-ledger.jsonl"
    with ledger.open("ab") as handle:
        handle.write(canonical_bytes(success) + b"\n")
        handle.write(canonical_bytes(success) + b"\n")

    def valid(output_dir, task_entry, frozen_manifest):
        return {
            "result": {"task_id": task_entry["task_id"], "run_identity": task_entry["scientific_run_identity"]},
            "result_sha256": "2" * 64,
            "artifact_hashes": {},
        }

    duplicate = aggregate_validated_successes(manifest_path, output, success_validator=valid)
    assert duplicate["validated_success_count"] == 0
    assert "exactly one matching success" in duplicate["rejected"][0]["reason"]

    second_output = tmp_path / "batch-failed"
    reserve_task(second_output / "control", manifest, task)
    failure = {
        "schema_version": 1,
        "event": "failed",
        "manifest_hash": "sha256:" + "f" * 64,
        "scientific_run_identity": task["scientific_run_identity"],
        "task_id": task["task_id"],
    }
    with (second_output / "control" / "failure-ledger.jsonl").open("wb") as handle:
        handle.write(canonical_bytes(failure) + b"\n")
    with (second_output / "control" / "launch-ledger.jsonl").open("ab") as handle:
        handle.write(canonical_bytes(success) + b"\n")
    failed = aggregate_validated_successes(manifest_path, second_output, success_validator=valid)
    assert failed["validated_success_count"] == 0
    assert "failure event" in failed["rejected"][0]["reason"]


def test_worker_sets_threads_before_scientific_import():
    worker = Path(__file__).parents[1] / "scripts" / "local_batch_worker.py"
    source = worker.read_text(encoding="utf-8")
    assignment = source.index('os.environ[_name] = "1"')
    scientific_import = source.index("from loss_aware_dro_repro.task_dispatch import run_task")
    assert assignment < scientific_import


def test_completion_projection_gate_passes_at_exact_boundary_and_bounds_in_flight(tmp_path):
    manifest, receipt = _policy_manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    active = 0
    maximum_active = 0
    lock = threading.Lock()

    def launcher(command, environment):
        nonlocal active, maximum_active
        with lock:
            active += 1
            maximum_active = max(maximum_active, active)
        time.sleep(0.001)
        with lock:
            active -= 1
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    summary = run_local_batch(
        manifest_path,
        tmp_path / "pass-batch",
        snapshot_provider=lambda: manifest["repository"],
        launcher=launcher,
        success_validator=_fake_success_validator,
        wall_clock=_GateClock(300.0),
    )
    evaluation = summary["issuance_policy_evaluation"]
    assert summary["partial_blocker"] is False
    assert summary["issued_task_count"] == 400
    assert summary["unissued_task_count"] == 0
    assert summary["maximum_in_flight_observed"] == 8
    assert maximum_active <= 8
    assert evaluation["evaluated_after_completed_tasks"] == 100
    assert evaluation["conservative_upper_wall_seconds"] == 1500.0
    assert all(
        math.isfinite(evaluation[key])
        for key in (
            "elapsed_wall_seconds",
            "point_projected_wall_seconds",
            "conservative_upper_wall_seconds",
        )
    )
    assert set(summary["scientific_verdicts"].values()) == {"HOLD"}
    assert any(path.endswith(".passed.json") for path in summary["artifact_hashes"])

    aggregate = aggregate_validated_successes(
        manifest_path,
        tmp_path / "pass-batch",
        success_validator=_fake_success_validator,
    )
    assert aggregate["complete"] is True
    passed_path = next(
        (tmp_path / "pass-batch" / "control" / "issuance-policy").glob(
            "*.passed.json"
        )
    )
    assert aggregate["issuance_pass_receipt_sha256"] == hashlib.sha256(
        passed_path.read_bytes()
    ).hexdigest()


def test_policy_aggregation_rejects_missing_or_semantically_forged_pass(tmp_path):
    manifest, receipt = _policy_manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "policy-aggregate"
    with pytest.raises(ContractError, match="issuance-passed"):
        aggregate_validated_successes(
            manifest_path, output, success_validator=_fake_success_validator
        )

    run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=lambda command, environment: subprocess.CompletedProcess(
            command, 0, stdout="", stderr=""
        ),
        success_validator=_fake_success_validator,
        wall_clock=_GateClock(300.0),
    )
    passed_path = next(
        (output / "control" / "issuance-policy").glob("*.passed.json")
    )
    forged = json.loads(passed_path.read_text(encoding="utf-8"))
    forged["conservative_upper_wall_seconds"] = 1.0
    unhashed = {key: value for key, value in forged.items() if key != "receipt_hash"}
    forged["receipt_hash"] = sha256_value(unhashed)
    passed_path.write_bytes(canonical_bytes(forged) + b"\n")
    with pytest.raises(ContractError, match="arithmetic"):
        aggregate_validated_successes(
            manifest_path, output, success_validator=_fake_success_validator
        )


def test_completion_projection_gate_blocks_issuance_and_prevents_resume(tmp_path):
    manifest, receipt = _policy_manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    launch_count = 0
    lock = threading.Lock()

    def launcher(command, environment):
        nonlocal launch_count
        with lock:
            launch_count += 1
        time.sleep(0.001)
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    output = tmp_path / "blocked-batch"
    summary = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=launcher,
        success_validator=_fake_success_validator,
        wall_clock=_GateClock(300.1),
    )
    assert summary["partial_blocker"] is True
    assert summary["issued_task_count"] == launch_count
    assert 100 <= launch_count <= 108
    assert summary["unissued_task_count"] == 400 - launch_count
    assert summary["maximum_in_flight_observed"] <= 8
    assert summary["issuance_policy_evaluation"]["conservative_upper_wall_seconds"] > 1500.0
    assert any(path.endswith(".blocked.json") for path in summary["artifact_hashes"])
    assert set(summary["scientific_verdicts"].values()) == {"HOLD"}

    with pytest.raises(ContractError, match="issuance blocker"):
        aggregate_validated_successes(
            manifest_path, output, success_validator=_fake_success_validator
        )

    with pytest.raises(ContractError, match="blocker prevents continuation"):
        run_local_batch(
            manifest_path,
            output,
            snapshot_provider=lambda: manifest["repository"],
            launcher=lambda *_: (_ for _ in ()).throw(AssertionError("blocked manifest must not issue")),
            success_validator=_fake_success_validator,
            wall_clock=lambda: 0.0,
        )

    blocker_path = next(
        (output / "control" / "issuance-policy").glob("*.blocked.json")
    )
    forged = json.loads(blocker_path.read_text(encoding="utf-8"))
    forged["block_reason"] = "unsupported_reason"
    unhashed = {key: value for key, value in forged.items() if key != "receipt_hash"}
    forged["receipt_hash"] = sha256_value(unhashed)
    blocker_path.write_bytes(canonical_bytes(forged) + b"\n")
    with pytest.raises(ContractError, match="unsupported issuance-policy"):
        aggregate_validated_successes(
            manifest_path, output, success_validator=_fake_success_validator
        )


def test_manifest_rejects_forged_issuance_policy():
    manifest, _ = _policy_manifest()
    manifest["execution"]["issuance_policy"] = {
        **completion_projection_gate_v1(8),
        "upper_multiplier": 1.0,
    }
    manifest["manifest_hash"] = sha256_value(
        {key: value for key, value in manifest.items() if key != "manifest_hash"}
    )
    with pytest.raises(ContractError, match="forged or unsupported"):
        validate_batch_manifest(manifest)


def test_policy_preflight_blocks_orphan_reservation_before_new_issuance(tmp_path):
    manifest, receipt = _policy_manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "orphaned-batch"
    reserve_task(output / "control", manifest, manifest["tasks"][137])

    summary = run_local_batch(
        manifest_path,
        output,
        snapshot_provider=lambda: manifest["repository"],
        launcher=lambda *_: (_ for _ in ()).throw(
            AssertionError("an orphaned sample must not issue new work")
        ),
        success_validator=_fake_success_validator,
        wall_clock=lambda: 10_000.0,
    )

    assert summary["partial_blocker"] is True
    assert summary["issued_task_count"] == 1
    assert summary["unissued_task_count"] == 399
    assert summary["maximum_in_flight_observed"] == 0
    assert summary["status_counts"] == {"skipped_reserved_identity": 1}
    assert summary["issuance_policy_evaluation"]["block_reason"] == "nonrecoverable_task_status"
    assert summary["issuance_policy_evaluation"]["blocking_status"] == "skipped_reserved_identity"

    with pytest.raises(ContractError, match="blocker prevents continuation"):
        run_local_batch(
            manifest_path,
            output,
            snapshot_provider=lambda: manifest["repository"],
            launcher=lambda *_: (_ for _ in ()).throw(
                AssertionError("a blocked manifest must not relaunch")
            ),
            success_validator=_fake_success_validator,
            wall_clock=lambda: 10_000.0,
        )


def test_policy_worker_failure_halts_after_only_bounded_in_flight_work(tmp_path):
    manifest, receipt = _policy_manifest()
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    launched = 0
    lock = threading.Lock()

    def failing_launcher(command, environment):
        nonlocal launched
        with lock:
            launched += 1
        return subprocess.CompletedProcess(
            command, 1, stdout="bounded failure", stderr="diagnostic"
        )

    summary = run_local_batch(
        manifest_path,
        tmp_path / "failed-policy-batch",
        snapshot_provider=lambda: manifest["repository"],
        launcher=failing_launcher,
        success_validator=_fake_success_validator,
        wall_clock=lambda: 10_000.0,
    )

    assert summary["partial_blocker"] is True
    assert 1 <= launched <= 8
    assert summary["issued_task_count"] == launched
    assert summary["unissued_task_count"] == 400 - launched
    assert summary["issuance_policy_evaluation"]["block_reason"] == "nonrecoverable_task_status"
    assert summary["issuance_policy_evaluation"]["blocking_status"] == "failed"
    assert summary["issuance_policy_evaluation"].get("evaluated_after_completed_tasks") is None
    assert set(summary["status_counts"]) == {"failed"}


def test_controller_lease_prevents_concurrent_owners_and_releases(tmp_path):
    manifest, receipt = _manifest(workers=1)
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "leased-batch"
    launched = threading.Event()
    release = threading.Event()
    first_results = []
    first_errors = []

    def blocking_launcher(command, environment):
        launched.set()
        assert release.wait(timeout=5)
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    def first_controller():
        try:
            first_results.append(
                run_local_batch(
                    manifest_path,
                    output,
                    snapshot_provider=lambda: manifest["repository"],
                    launcher=blocking_launcher,
                    success_validator=_fake_success_validator,
                )
            )
        except Exception as exc:
            first_errors.append(exc)

    thread = threading.Thread(target=first_controller)
    thread.start()
    assert launched.wait(timeout=5)
    with pytest.raises(ContractError, match="another controller owns"):
        run_local_batch(
            manifest_path,
            output,
            snapshot_provider=lambda: manifest["repository"],
            launcher=lambda *_: (_ for _ in ()).throw(
                AssertionError("second controller must not launch")
            ),
            success_validator=_fake_success_validator,
        )
    release.set()
    thread.join(timeout=5)
    assert not thread.is_alive()
    assert first_errors == []
    assert first_results[0]["status_counts"] == {"success": 1}
    assert list((output / "control" / "controller-leases").glob("*.json")) == []


def test_controller_lease_releases_when_run_raises(tmp_path):
    manifest, receipt = _manifest(workers=1)
    manifest_path, _ = _write_manifest(tmp_path, manifest, receipt)
    output = tmp_path / "raised-batch"

    def broken_snapshot():
        raise RuntimeError("snapshot probe failed")

    with pytest.raises(RuntimeError, match="snapshot probe failed"):
        run_local_batch(
            manifest_path,
            output,
            snapshot_provider=broken_snapshot,
        )
    assert list((output / "control" / "controller-leases").glob("*.json")) == []


@pytest.mark.parametrize(
    ("command", "result", "expected"),
    [
        (
            "run",
            {
                "partial_blocker": True,
                "task_count": 100,
                "manifest_task_count": 400,
                "unissued_task_count": 300,
                "status_counts": {"success": 100},
            },
            2,
        ),
        (
            "run",
            {
                "partial_blocker": False,
                "task_count": 4,
                "manifest_task_count": 4,
                "unissued_task_count": 0,
                "status_counts": {"success": 3, "failed": 1},
            },
            2,
        ),
        ("aggregate", {"complete": False}, 2),
        ("aggregate", {"complete": True}, 0),
    ],
)
def test_cli_exit_code_reflects_incomplete_or_failed_work(command, result, expected):
    assert _result_exit_code(command, result) == expected


def test_cli_main_returns_nonzero_for_partial_run(monkeypatch, tmp_path, capsys):
    monkeypatch.setattr(
        batch_cli,
        "run_local_batch",
        lambda manifest, output: {
            "partial_blocker": True,
            "task_count": 100,
            "manifest_task_count": 400,
            "unissued_task_count": 300,
            "status_counts": {"success": 100},
        },
    )
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "run_local_batch.py",
            "run",
            "--manifest",
            str(tmp_path / "manifest.json"),
            "--output-root",
            str(tmp_path / "output"),
        ],
    )
    assert batch_cli.main() == 2
    assert '"partial_blocker": true' in capsys.readouterr().out