from __future__ import annotations import asyncio import hashlib import json from pathlib import Path from types import SimpleNamespace import pytest import autocad_bench.evaluation.rescore as rescore from autocad_bench.evaluation.rescore import ( RescoreCandidate, _inspection_is_current, _inspection_phase, _launch_prepared_worker, _score_candidates, _terminate_owned_instance, archive_current_evaluation, materialize_cloud_run_inputs, promote_staged_evaluation, select_rollout_candidates, validate_staged_evaluation, ) V3 = "autocad-2019-r23.0.46-v3" V4 = "autocad-2019-r23.0.46-v4" V4_RENDER_CONFIG_HASH = "6" * 64 PNG_HEADER = b"\x89PNG\r\n\x1a\n" def _write_json(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value), encoding="utf-8") def _evaluation( directory: Path, *, evaluator_version: str, attempt: bytes, render_config_hash: str = V4_RENDER_CONFIG_HASH, stage: str = "completed", vision_enabled: bool = True, ) -> None: render = PNG_HEADER + f"render-{evaluator_version}".encode() directory.mkdir(parents=True, exist_ok=True) _write_json( directory / "metadata.json", { "schema_version": "3", "evaluator_version": evaluator_version, "render_config_hash": render_config_hash, "source_sha256": hashlib.sha256(attempt).hexdigest(), "render_sha256": hashlib.sha256(render).hexdigest(), "drawing_type": "3d", "valid": True, }, ) (directory / "render.png").write_bytes(render) _write_json( directory / "score.json", { "task_id": "task-026", "evaluator_version": evaluator_version, "valid": True, "score": 0.75, }, ) _write_json( directory / "status.json", { "task_id": "task-026", "evaluator_version": evaluator_version, "stage": stage, "deterministic_score": 0.75, "vision_enabled": vision_enabled, **( {} if vision_enabled else {"vision_skipped_reason": "disabled"} ), }, ) if vision_enabled: _write_json( directory / "vision-judge" / "score.json", {"overall_score": 0.8, "verdict": "partial"}, ) def _rollout( run_root: Path, task_id: str, *, completed: bool, attempt: bytes | None, ) -> Path: rollout = run_root / "rollouts" / task_id rollout.mkdir(parents=True) _write_json( rollout / "run-state.json", { "task_id": task_id, "state": "completed" if completed else "running", }, ) _write_json( rollout / "result.json", { "task_id": task_id, "completed": completed, "artifact_sha256": ( hashlib.sha256(attempt).hexdigest() if attempt is not None else None ), "evaluation": { "status": "completed", "evaluator_version": V3, }, }, ) if attempt is not None: (rollout / "attempt.dwg").write_bytes(attempt) return rollout def test_archive_preserves_v3_evaluation_and_result_without_mutating_active_files( tmp_path: Path, ) -> None: rollout = tmp_path / "run" / "rollouts" / "task-026" attempt = b"AC1032-v3-candidate" (rollout / "attempt.dwg").parent.mkdir(parents=True) (rollout / "attempt.dwg").write_bytes(attempt) _evaluation( rollout / "evaluation", evaluator_version=V3, attempt=attempt, render_config_hash="3" * 64, ) result = { "task_id": "task-026", "completed": True, "artifact_sha256": hashlib.sha256(attempt).hexdigest(), "evaluation": { "status": "completed", "evaluator_version": V3, "deterministic_score": 0.75, }, } _write_json(rollout / "result.json", result) active_files_before = { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } archive_current_evaluation(rollout, evaluator_version=V3) archived_evaluation = rollout / "evaluation-history" / V3 assert { path.relative_to(archived_evaluation): path.read_bytes() for path in archived_evaluation.rglob("*") if path.is_file() } == active_files_before assert json.loads((rollout / "result-history" / f"{V3}.json").read_text()) == result assert { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } == active_files_before assert (rollout / "attempt.dwg").read_bytes() == attempt assert json.loads((rollout / "result.json").read_text()) == result def test_validate_staged_v4_accepts_only_source_bound_completed_artifacts( tmp_path: Path, ) -> None: attempt = b"AC1032-v4-candidate" staging = tmp_path / ".evaluation-v4-staging" _evaluation(staging, evaluator_version=V4, attempt=attempt) validate_staged_evaluation( staging, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) def test_validate_staged_v4_accepts_completed_deterministic_only_scoring( tmp_path: Path, ) -> None: attempt = b"AC1032-v4-candidate" staging = tmp_path / ".evaluation-v4-no-vision" _evaluation( staging, evaluator_version=V4, attempt=attempt, vision_enabled=False, ) validate_staged_evaluation( staging, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, vision_required=False, ) def test_cached_invalid_inspection_with_empty_render_is_scoreable( tmp_path: Path, ) -> None: run_root = tmp_path / "run" rollout = _rollout( run_root, "task-001", completed=True, attempt=b"AC1032-invalid-candidate", ) candidate = select_rollout_candidates([run_root])[0] staging_root = rollout / ".rescore-staging" / "rescore-v4-test" evaluation = staging_root / "evaluation" _evaluation( evaluation, evaluator_version=V4, attempt=b"AC1032-invalid-candidate", vision_enabled=False, ) metadata = json.loads((evaluation / "metadata.json").read_text()) metadata["valid"] = False metadata["render_sha256"] = None _write_json(evaluation / "metadata.json", metadata) (evaluation / "render.png").write_bytes(b"") settings = SimpleNamespace( evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) assert _inspection_is_current(candidate, staging_root, settings) validate_staged_evaluation( evaluation, attempt_sha256=candidate.attempt_sha256, evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, vision_required=False, ) def test_validate_staged_v4_rejects_non_png_even_when_its_checksum_matches( tmp_path: Path, ) -> None: attempt = b"AC1032-v4-candidate" staging = tmp_path / ".evaluation-v4-invalid-render" _evaluation(staging, evaluator_version=V4, attempt=attempt) invalid_render = b"this is not a PNG" (staging / "render.png").write_bytes(invalid_render) metadata_path = staging / "metadata.json" metadata = json.loads(metadata_path.read_text()) metadata["render_sha256"] = hashlib.sha256(invalid_render).hexdigest() _write_json(metadata_path, metadata) with pytest.raises((RuntimeError, ValueError), match="PNG"): validate_staged_evaluation( staging, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) @pytest.mark.parametrize( ("mutation", "expected_error"), [ ("source_sha256", "source"), ("evaluator_version", "version"), ("render_config_hash", "render"), ("incomplete_status", "completed"), ("missing_render", "render"), ], ) def test_validate_staged_v4_rejects_unbound_or_incomplete_artifacts( tmp_path: Path, mutation: str, expected_error: str, ) -> None: attempt = b"AC1032-v4-candidate" staging = tmp_path / f".evaluation-v4-staging-{mutation}" _evaluation(staging, evaluator_version=V4, attempt=attempt) if mutation in {"source_sha256", "evaluator_version", "render_config_hash"}: metadata_path = staging / "metadata.json" metadata = json.loads(metadata_path.read_text()) metadata[mutation] = "wrong" _write_json(metadata_path, metadata) elif mutation == "incomplete_status": status_path = staging / "status.json" status = json.loads(status_path.read_text()) status["stage"] = "vision_judging" _write_json(status_path, status) else: (staging / "render.png").unlink() with pytest.raises((FileNotFoundError, RuntimeError, ValueError), match=expected_error): validate_staged_evaluation( staging, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) def test_promote_swaps_valid_v4_as_a_directory_and_keeps_v3_history( tmp_path: Path, ) -> None: rollout = tmp_path / "run" / "rollouts" / "task-026" attempt = b"AC1032-candidate" (rollout / "attempt.dwg").parent.mkdir(parents=True) (rollout / "attempt.dwg").write_bytes(attempt) _evaluation( rollout / "evaluation", evaluator_version=V3, attempt=attempt, render_config_hash="3" * 64, ) _write_json( rollout / "result.json", { "task_id": "task-026", "completed": True, "evaluation": {"status": "completed", "evaluator_version": V3}, }, ) archive_current_evaluation(rollout, evaluator_version=V3) staged = rollout / ".evaluation-v4-staging" _evaluation(staged, evaluator_version=V4, attempt=attempt) validate_staged_evaluation( staged, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) promote_staged_evaluation( rollout, staged, prior_evaluator_version=V3, ) assert not staged.exists() active_metadata = json.loads((rollout / "evaluation" / "metadata.json").read_text()) archived_metadata = json.loads( ( rollout / "evaluation-history" / V3 / "metadata.json" ).read_text() ) assert active_metadata["evaluator_version"] == V4 assert archived_metadata["evaluator_version"] == V3 assert (rollout / "evaluation" / "vision-judge" / "score.json").is_file() assert (rollout / "result-history" / f"{V3}.json").is_file() assert (rollout / "attempt.dwg").read_bytes() == attempt def test_promote_refuses_incomplete_staging_without_disturbing_v3( tmp_path: Path, ) -> None: rollout = tmp_path / "run" / "rollouts" / "task-026" attempt = b"AC1032-candidate" (rollout / "attempt.dwg").parent.mkdir(parents=True) (rollout / "attempt.dwg").write_bytes(attempt) _evaluation( rollout / "evaluation", evaluator_version=V3, attempt=attempt, render_config_hash="3" * 64, ) active_before = { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } staged = rollout / ".evaluation-v4-staging" _evaluation( staged, evaluator_version=V4, attempt=attempt, stage="vision_judging", ) with pytest.raises((FileNotFoundError, RuntimeError, ValueError)): promote_staged_evaluation( rollout, staged, prior_evaluator_version=V3, ) assert { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } == active_before assert staged.is_dir() def test_promote_rechecks_source_binding_after_validation_before_swapping( tmp_path: Path, ) -> None: rollout = tmp_path / "run" / "rollouts" / "task-026" attempt = b"AC1032-candidate" (rollout / "attempt.dwg").parent.mkdir(parents=True) (rollout / "attempt.dwg").write_bytes(attempt) _evaluation( rollout / "evaluation", evaluator_version=V3, attempt=attempt, render_config_hash="3" * 64, ) active_before = { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } staged = rollout / ".evaluation-v4-staging" _evaluation(staged, evaluator_version=V4, attempt=attempt) validate_staged_evaluation( staged, attempt_sha256=hashlib.sha256(attempt).hexdigest(), evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) metadata_path = staged / "metadata.json" metadata = json.loads(metadata_path.read_text()) metadata["source_sha256"] = "0" * 64 _write_json(metadata_path, metadata) with pytest.raises((RuntimeError, ValueError), match="source"): promote_staged_evaluation( rollout, staged, prior_evaluator_version=V3, ) assert { path.relative_to(rollout / "evaluation"): path.read_bytes() for path in (rollout / "evaluation").rglob("*") if path.is_file() } == active_before assert staged.is_dir() def test_candidate_selection_defaults_to_completed_execution_and_is_read_only( tmp_path: Path, ) -> None: run_root = tmp_path / "benchmarks" / "run-a" finished = _rollout( run_root, "task-001", completed=True, attempt=b"AC1032-finished", ) active = _rollout( run_root, "task-002", completed=False, attempt=b"AC1032-active", ) _rollout( run_root, "task-003", completed=True, attempt=None, ) active_before = { path.relative_to(active): path.read_bytes() for path in active.rglob("*") if path.is_file() } selected = select_rollout_candidates([run_root]) assert [(item.rollout_dir, item.task_id) for item in selected] == [ (finished, "task-001") ] assert selected[0].attempt_sha256 == hashlib.sha256( b"AC1032-finished" ).hexdigest() assert { path.relative_to(active): path.read_bytes() for path in active.rglob("*") if path.is_file() } == active_before def test_candidate_selection_does_not_follow_rollout_directory_symlinks( tmp_path: Path, ) -> None: external_root = tmp_path / "unrelated-active-run" external = _rollout( external_root, "task-001", completed=True, attempt=b"AC1032-unrelated", ) requested_root = tmp_path / "requested-run" rollouts = requested_root / "rollouts" rollouts.mkdir(parents=True) (rollouts / "task-001").symlink_to(external, target_is_directory=True) assert select_rollout_candidates([requested_root]) == [] def test_cloud_materialization_prefers_complete_authoritative_bundle_over_local( tmp_path: Path, ) -> None: run_root = tmp_path / "sources" / "run-a" local = b"AC1032-local" _rollout(run_root, "task-001", completed=True, attempt=local) cloud = b"AC1032-cloud-authoritative" base = "controller/runs/benchmarks/run-a/rollouts/task-001" objects = { f"{base}/attempt.dwg": cloud, f"{base}/result.json": json.dumps( { "task_id": "task-001", "completed": True, "artifact_sha256": hashlib.sha256(cloud).hexdigest(), } ).encode(), } class FakeS3: def list_objects_v2(self, **request): prefix = request["Prefix"] return { "Contents": [ {"Key": key} for key in objects if key.startswith(prefix) ], "IsTruncated": False, } def download_file(self, _bucket, key, target): Path(target).write_bytes(objects[key]) materialized = materialize_cloud_run_inputs( [run_root], destination_root=tmp_path / "materialized", s3=FakeS3(), bucket="test-bucket", source_prefix="controller/runs/benchmarks", ) target = materialized[0] / "rollouts" / "task-001" assert (target / "attempt.dwg").read_bytes() == cloud assert ( json.loads((target / "result.json").read_text())["artifact_sha256"] == hashlib.sha256(cloud).hexdigest() ) assert (run_root / "rollouts" / "task-001" / "attempt.dwg").read_bytes() == local def test_cloud_materialization_preserves_completed_v4_against_stale_shared_mirror( tmp_path: Path, ) -> None: run_root = tmp_path / "sources" / "run-a" stale = b"AC1032-stale-cloud" base = "controller/runs/benchmarks/run-a/rollouts/task-001" objects = { f"{base}/attempt.dwg": stale, f"{base}/result.json": json.dumps( { "task_id": "task-001", "completed": True, "artifact_sha256": hashlib.sha256(stale).hexdigest(), } ).encode(), } destination = tmp_path / "materialized" target = destination / "run-a" / "rollouts" / "task-001" current = b"AC1032-current-v4" target.mkdir(parents=True) (target / "attempt.dwg").write_bytes(current) _write_json( target / "result.json", { "task_id": "task-001", "completed": True, "artifact_sha256": hashlib.sha256(current).hexdigest(), }, ) _evaluation( target / "evaluation", evaluator_version=V4, attempt=current, ) class FakeS3: def list_objects_v2(self, **request): prefix = request["Prefix"] return { "Contents": [ {"Key": key} for key in objects if key.startswith(prefix) ], "IsTruncated": False, } def download_file(self, _bucket, key, output): Path(output).write_bytes(objects[key]) materialize_cloud_run_inputs( [run_root], destination_root=destination, s3=FakeS3(), bucket="test-bucket", source_prefix="controller/runs/benchmarks", preserve_evaluator_version=V4, ) assert (target / "attempt.dwg").read_bytes() == current assert json.loads((target / "evaluation" / "metadata.json").read_text())[ "evaluator_version" ] == V4 def test_cached_inspection_rejects_a_render_that_no_longer_matches_metadata( tmp_path: Path, ) -> None: rollout = tmp_path / "run" / "rollouts" / "task-001" attempt = b"AC1032-attempt" candidate = RescoreCandidate( run_root=rollout.parents[1], rollout_dir=rollout, rollout_id="task-001", task_id="task-001", attempt_sha256=hashlib.sha256(attempt).hexdigest(), result={"completed": True}, ) staging = rollout / ".rescore-staging" / "job" _evaluation(staging / "evaluation", evaluator_version=V4, attempt=attempt) (staging / "evaluation" / "render.png").write_bytes(b"corrupt") settings = SimpleNamespace( evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) assert not _inspection_is_current(candidate, staging, settings) def test_worker_is_terminated_if_persisting_ownership_fails() -> None: job_id = "rescore-v4-test" class FakeEc2: def describe_instances(self, **_request): return { "Reservations": [ { "Instances": [ { "InstanceId": "i-owned", "Tags": [ {"Key": "BatchId", "Value": job_id}, { "Key": "ManagedBy", "Value": "autocad-bench-sandbox", }, {"Key": "Ephemeral", "Value": "true"}, ], } ] } ] } class FakeManager: ec2 = FakeEc2() def __init__(self) -> None: self.terminated: list[str] = [] self.prepared: list[str] = [] def launch_instance(self, _spec): return "i-owned" def prepare(self, instance_id): self.prepared.append(instance_id) def terminate_instance(self, instance_id): self.terminated.append(instance_id) class FailingJournal: def own_instance(self, _instance_id): raise RuntimeError("journal upload failed") def release_instance(self, _instance_id): pass manager = FakeManager() settings = SimpleNamespace( image_id="ami-v4", subnet_id="subnet-a", security_group_id="sg-a", instance_profile_name="profile-a", instance_type="m7i.xlarge", job_id=job_id, ) with pytest.raises(RuntimeError, match="journal upload failed"): _launch_prepared_worker(manager, FailingJournal(), settings, 0) assert manager.prepared == [] assert manager.terminated == ["i-owned"] def test_termination_refuses_an_instance_without_exact_job_ownership_tags() -> None: class FakeEc2: def describe_instances(self, **_request): return { "Reservations": [ { "Instances": [ { "InstanceId": "i-unrelated", "Tags": [ {"Key": "BatchId", "Value": "another-job"}, { "Key": "ManagedBy", "Value": "autocad-bench-sandbox", }, {"Key": "Ephemeral", "Value": "true"}, ], } ] } ] } class FakeManager: ec2 = FakeEc2() def __init__(self) -> None: self.terminated: list[str] = [] def terminate_instance(self, instance_id): self.terminated.append(instance_id) class FakeJournal: def release_instance(self, _instance_id): raise AssertionError("unowned instance was released") manager = FakeManager() with pytest.raises(RuntimeError, match="refusing to terminate unowned"): _terminate_owned_instance( manager, FakeJournal(), instance_id="i-unrelated", job_id="rescore-v4-test", ) assert manager.terminated == [] def test_resume_repairs_result_if_crash_happened_after_evaluation_promotion( tmp_path: Path, ) -> None: run_root = tmp_path / "run" rollout = run_root / "rollouts" / "task-001" attempt = b"AC1032-candidate" rollout.mkdir(parents=True) (rollout / "attempt.dwg").write_bytes(attempt) _evaluation(rollout / "evaluation", evaluator_version=V4, attempt=attempt) _write_json( rollout / "result.json", { "task_id": "task-001", "completed": True, "artifact_sha256": hashlib.sha256(attempt).hexdigest(), "evaluation": {"status": "completed", "evaluator_version": V3}, }, ) candidate = RescoreCandidate( run_root=run_root, rollout_dir=rollout, rollout_id="task-001", task_id="task-001", attempt_sha256=hashlib.sha256(attempt).hexdigest(), result=json.loads((rollout / "result.json").read_text()), ) class FakeJournal: def __init__(self) -> None: self.states: list[str] = [] def task(self, _candidate, state, **_details): self.states.append(state) journal = FakeJournal() settings = SimpleNamespace( job_id="rescore-v4-test", vision_concurrency=1, evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, vision_enabled=True, publish_prefix=None, ) failures = asyncio.run( _score_candidates( [candidate], journal=journal, settings=settings, s3=None, ) ) assert failures == {} result = json.loads((rollout / "result.json").read_text()) assert result["evaluation"]["evaluator_version"] == V4 assert result["evaluation"]["status"] == "completed" assert journal.states[-1] == "completed" def test_journal_failure_while_queueing_still_terminates_launched_workers( tmp_path: Path, monkeypatch, ) -> None: run_root = tmp_path / "run" rollout = _rollout( run_root, "task-001", completed=True, attempt=b"AC1032-candidate", ) candidate = select_rollout_candidates([run_root])[0] terminated: list[str] = [] monkeypatch.setattr( rescore, "_launch_prepared_worker", lambda *_args, **_kwargs: "i-owned", ) monkeypatch.setattr( rescore, "_terminate_owned_instance", lambda *_args, instance_id, **_kwargs: terminated.append(instance_id), ) class FailingJournal: def task(self, _candidate, _state, **_details): raise RuntimeError("journal upload failed") def set_state(self, _state, **_details): pass settings = SimpleNamespace( job_id="rescore-v4-test", worker_count=1, evaluator_version=V4, render_config_hash=V4_RENDER_CONFIG_HASH, ) with pytest.raises(RuntimeError, match="journal upload failed"): _inspection_phase( [candidate], manager=object(), journal=FailingJournal(), settings=settings, ) assert rollout.is_dir() assert terminated == ["i-owned"]