1036 lines
47 KiB
Python
1036 lines
47 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import sys
|
|
import types
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
SPEC = importlib.util.spec_from_file_location("multiscene_eval_runner_tested", SCRIPTS / "multiscene_eval_runner.py")
|
|
EVAL = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader is not None
|
|
SPEC.loader.exec_module(EVAL)
|
|
|
|
|
|
def _sha(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _write_x(path: Path, value) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("x", encoding="utf-8") as stream:
|
|
json.dump(value, stream)
|
|
|
|
|
|
class RealFactory:
|
|
pass
|
|
|
|
|
|
class SyntheticFactory:
|
|
_rc05_execution_mode = "synthetic"
|
|
|
|
|
|
def _plan(tmp_path, count=3):
|
|
source = tmp_path / "source.py"
|
|
source.write_text("frozen\n", encoding="utf-8")
|
|
cases = [{"case_id": f"case-{index}", "provider_call_budget": 1,
|
|
"kind": "supplemental", "operation": "draft",
|
|
"operations": ["draft"]}
|
|
for index in range(count)]
|
|
images = [{"role": role, "repository": f"example/{role}",
|
|
"manifest_digest": f"sha256:{index + 1:064x}",
|
|
"config_digest": f"sha256:{index + 11:064x}"}
|
|
for index, role in enumerate(("contract-web", "contract-api",
|
|
"ai-platform-api", "ai-worker"))]
|
|
candidate = {"schema_version": "multiscene-candidate-binding.v1",
|
|
"commit": "1" * 40, "tag": "test-candidate",
|
|
"source_manifest": {"schema_version": "yuqei.release-source-manifest.v1",
|
|
"raw_sha256": "2" * 64, "payload_sha256": "3" * 64},
|
|
"image_identities": {"schema_version": "yuqei.rc06-image-identities.v1",
|
|
"raw_sha256": "4" * 64,
|
|
"identity_set_sha256": _sha(json.dumps(
|
|
images, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
).encode("utf-8")), "images": images}}
|
|
return {
|
|
"cases": cases,
|
|
"operation_limits": {"draft": count},
|
|
"source_pins": {"source.py": _sha(source.read_bytes())},
|
|
"candidate": candidate,
|
|
"release_controls": {"schema_version": "multiscene-release-controls.v1",
|
|
"canary_case_ids": ["case-0"],
|
|
"failure_classes": {case["case_id"]: "draft" for case in cases}},
|
|
}
|
|
|
|
|
|
def _install_modules(monkeypatch, plan, digest="a" * 64, calls=None):
|
|
calls = calls if calls is not None else []
|
|
plan_module = types.ModuleType("multiscene_eval_plan")
|
|
plan_module.validate_plan = lambda *_args: (plan, digest)
|
|
plan_module.build_plan = lambda *_args: plan
|
|
plan_module.load_candidate_binding = lambda *_args, **_kwargs: (plan["candidate"], {})
|
|
plan_module.verify_source_pins_at_candidate = lambda *_args, **_kwargs: None
|
|
supplemental = types.ModuleType("multiscene_supplemental_cases")
|
|
|
|
def run_case(case, directory, _factory, state, *, ledger, runner):
|
|
calls.append(case["case_id"])
|
|
directory.mkdir(parents=True)
|
|
token = ledger.claim(case["case_id"], "draft", "b" * 64)
|
|
state["provider_call_count"] = 1
|
|
ledger.finish(token, "succeeded", "c" * 64)
|
|
_write_x(directory / "01-draft-after.json", {
|
|
"case_id": case["case_id"], "operation": "draft",
|
|
"request_sha256": "b" * 64, "response_sha256": "c" * 64,
|
|
"provider_invocation_count": 1,
|
|
"execution_elapsed_seconds": 2.5,
|
|
"job": {"id": "job-current", "operation": "contract_draft",
|
|
"status": "succeeded", "provider_status": "succeeded",
|
|
"audit_run_id": "run-current", "error_code": None,
|
|
"error_message": None},
|
|
"audit": {"id": "run-current", "provider": "codex-cli-development",
|
|
"model": "gpt-5.6-sol", "provider_status": "succeeded",
|
|
"provider_error": None},
|
|
})
|
|
return {"case_id": case["case_id"], "status": "completed", "passed": True,
|
|
"provider_call_count": 1, "elapsed_seconds": 3.0,
|
|
"automated_quality_passed": True,
|
|
"automated_model_quality_release_passed": True,
|
|
"content_quality_release_passed": True,
|
|
"human_calibration_completed": False,
|
|
"legal_approval_completed": False}
|
|
|
|
supplemental.run_supplemental_case = run_case
|
|
monkeypatch.setitem(sys.modules, "multiscene_eval_plan", plan_module)
|
|
monkeypatch.setitem(sys.modules, "multiscene_supplemental_cases", supplemental)
|
|
return calls
|
|
|
|
|
|
def _runner(tmp_path):
|
|
return types.SimpleNamespace(
|
|
ROOT=tmp_path,
|
|
CodexCliDevelopmentProviderAdapter=RealFactory,
|
|
sha256=_sha,
|
|
write_x=_write_x,
|
|
)
|
|
|
|
|
|
def _candidate_inputs():
|
|
return {"commit": "1" * 40, "tag": "test-candidate",
|
|
"source_manifest_path": Path("source.json"),
|
|
"source_manifest_sha256": "2" * 64,
|
|
"source_manifest_payload_sha256": "3" * 64,
|
|
"image_identities_path": Path("images.json"),
|
|
"image_identities_sha256": "4" * 64,
|
|
"image_identity_set_sha256": "5" * 64}
|
|
|
|
|
|
def test_selected_execution_keeps_full_denominator_and_cannot_make_all_passed(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
summary = EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "a" * 64, tmp_path / "out",
|
|
SyntheticFactory, case_ids=["case-1"], execution_mode="synthetic",
|
|
)
|
|
assert calls == ["case-1"]
|
|
assert summary["case_count"] == 3
|
|
assert [row["status"] for row in summary["cases"]] == ["not_attempted", "completed", "not_attempted"]
|
|
assert summary["selected_passed"] is True
|
|
assert summary["complete"] is False
|
|
assert summary["passed"] is False
|
|
assert summary["cases"][1]["elapsed_seconds"] == 3.0
|
|
assert summary["operation_statistics"]["draft"]["elapsed_seconds"] == {
|
|
"observed": 1, "p50": 2.5, "p95": 2.5,
|
|
}
|
|
|
|
|
|
def test_resume_never_resends_completed_failed_or_unknown(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=4)
|
|
digest = "a" * 64
|
|
calls = _install_modules(monkeypatch, plan, digest)
|
|
out = tmp_path / "out"
|
|
ledger = EVAL.CallLedger(out, digest, plan["operation_limits"],
|
|
{case["case_id"]: 1 for case in plan["cases"]}, plan["source_pins"], False)
|
|
for case_id, filename, status in (
|
|
("case-0", "result.json", "completed"),
|
|
("case-1", "failure.json", "stopped"),
|
|
):
|
|
token = ledger.claim(case_id, "draft", "b" * 64)
|
|
ledger.finish(token, "succeeded" if status == "completed" else "failed", "c" * 64)
|
|
outcome = out / case_id / filename
|
|
_write_x(outcome, {"case_id": case_id, "status": status,
|
|
"passed": status == "completed", "provider_call_count": 1,
|
|
**({"automated_quality_passed": True,
|
|
"automated_model_quality_release_passed": True,
|
|
"content_quality_release_passed": True}
|
|
if status == "completed" else {})})
|
|
ledger.bind_case_outcome(
|
|
case_id, filename, _sha(outcome.read_bytes()), {}, ["draft"])
|
|
ledger.claim("case-2", "draft", "b" * 64)
|
|
(out / "case-2").mkdir()
|
|
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", digest, out,
|
|
SyntheticFactory, resume=True, execution_mode="synthetic")
|
|
assert calls == ["case-3"]
|
|
assert [row["status"] for row in summary["cases"]] == ["completed", "stopped", "unknown", "completed"]
|
|
assert summary["ledger"]["unknown"] == 1
|
|
assert summary["passed"] is False
|
|
|
|
|
|
def test_unknown_or_duplicate_selection_is_rejected_before_calls(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
for selection in (["unknown"], ["case-0", "case-0"], []):
|
|
with pytest.raises(ValueError, match="selection"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / f"out-{len(selection)}-{selection[0] if selection else 'empty'}",
|
|
SyntheticFactory, case_ids=selection, execution_mode="synthetic")
|
|
assert calls == []
|
|
|
|
|
|
def test_source_drift_stops_before_case_or_ledger_claim(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
(tmp_path / "source.py").write_text("drifted\n", encoding="utf-8")
|
|
out = tmp_path / "out"
|
|
with pytest.raises(ValueError, match="source drift before next case"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64, out,
|
|
SyntheticFactory, execution_mode="synthetic")
|
|
assert calls == []
|
|
snapshot = EVAL.CallLedger(out, "a" * 64, plan["operation_limits"],
|
|
{"case-0": 1}, plan["source_pins"], False).snapshot()
|
|
assert snapshot["attempted"] == 0
|
|
|
|
|
|
def test_source_drift_after_last_case_blocks_final_summary(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
|
|
def drifting(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
(tmp_path / "source.py").write_text("drifted-after-call\n", encoding="utf-8")
|
|
return result
|
|
|
|
supplemental.run_supplemental_case = drifting
|
|
with pytest.raises(ValueError, match="source drift before final summary"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
|
|
|
|
def test_execution_mode_and_factory_attestation_reject_wrapper_bypass(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
|
|
class WrappedReal:
|
|
def __call__(self, *args, **kwargs):
|
|
return RealFactory(*args, **kwargs)
|
|
|
|
with pytest.raises(ValueError, match="execution_mode"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "missing-mode", SyntheticFactory)
|
|
with pytest.raises(ValueError, match="attested synthetic"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "wrapped", WrappedReal(), execution_mode="synthetic")
|
|
|
|
|
|
def _historical_usage():
|
|
return {
|
|
"provider_call_count": 4,
|
|
"repair_budget_charged": 4,
|
|
"attempts": [
|
|
{
|
|
"attempt_id": f"prior-{index}",
|
|
"plan_sha256": f"{index + 10:064x}",
|
|
"case_id": f"prior-case-{index}",
|
|
"operation": "draft" if index < 3 else "semantic_review",
|
|
"provider_status": "failed" if index < 2 else "succeeded",
|
|
"business_status": "failed" if index < 2 or index == 3 else "succeeded",
|
|
"elapsed_seconds": float(index + 1),
|
|
"timed_out": False,
|
|
"request_sha256": f"{index + 20:064x}",
|
|
"response_sha256": f"{index + 30:064x}",
|
|
"provider": "codex-cli-development", "model": "gpt-5.6-sol",
|
|
"job_id": f"job-{index}", "audit_run_id": f"run-{index}",
|
|
"identity_path": f"identity-{index}.json",
|
|
"identity_file_sha256": f"{index + 40:064x}",
|
|
"ledger_identity_sha256": f"{index + 50:064x}",
|
|
"claim_path": f"claim-{index}.json",
|
|
"claim_sha256": f"{index + 60:064x}",
|
|
"finish_path": f"finish-{index}.json",
|
|
"finish_sha256": f"{index + 70:064x}",
|
|
"before_path": f"before-{index}.json",
|
|
"before_sha256": f"{index + 80:064x}",
|
|
"after_path": f"after-{index}.json",
|
|
"after_sha256": f"{index + 90:064x}",
|
|
"disposition": {"resolution_status": (
|
|
"resolved" if index == 2 else "requires_replacement"),
|
|
"release_eligible": index == 2,
|
|
"reason": "successful prior evidence" if index == 2 else "failure retained",
|
|
"replacement_case_id": None if index == 2 else f"prior-case-{index}",
|
|
"replacement_operation": None if index == 2 else (
|
|
"draft" if index < 3 else "semantic_review")},
|
|
}
|
|
for index in range(4)
|
|
],
|
|
"receipt_sha256": "e" * 64,
|
|
}
|
|
|
|
|
|
def _replacement_usage(case_id="case-0", operation="draft"):
|
|
usage = _historical_usage()
|
|
for attempt in usage["attempts"]:
|
|
if not attempt["disposition"]["release_eligible"]:
|
|
attempt["disposition"]["replacement_case_id"] = case_id
|
|
attempt["disposition"]["replacement_operation"] = operation
|
|
return usage
|
|
|
|
|
|
def _remediation_ledger(tmp_path, usage, *, business_status="succeeded"):
|
|
out = tmp_path / "remediation"
|
|
book = EVAL.CallLedger(
|
|
out, "a" * 64, {"draft": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, False, historical_usage=usage,
|
|
)
|
|
token = book.claim("case-0", "draft", "c" * 64)
|
|
book.finish(token, "succeeded", "d" * 64)
|
|
_write_x(out / "case-0/01-draft-after.json", {
|
|
"case_id": "case-0", "operation": "draft",
|
|
"request_sha256": "c" * 64, "response_sha256": "d" * 64,
|
|
"provider_invocation_count": 1, "execution_elapsed_seconds": 1,
|
|
"job": {"operation": "contract_draft", "status": business_status,
|
|
"provider_status": "succeeded", "error_code": None,
|
|
"error_message": None}, "audit": {},
|
|
})
|
|
rows = [{"case_id": "case-0", "status": "completed", "passed": True,
|
|
"automated_model_quality_release_passed": True,
|
|
"content_quality_release_passed": True}]
|
|
return out, book, rows
|
|
|
|
|
|
def test_historical_failure_without_replacement_remains_blocking(tmp_path):
|
|
usage = _replacement_usage()
|
|
out = tmp_path / "no-replacement"
|
|
book = EVAL.CallLedger(
|
|
out, "a" * 64, {"draft": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, False, historical_usage=usage,
|
|
)
|
|
passed, evidence = EVAL._historical_remediation(out, book, [])
|
|
assert passed is False
|
|
assert any(item["status"] == "replacement_required" for item in evidence)
|
|
|
|
|
|
def test_successful_same_key_replacement_remediates_but_keeps_history(tmp_path):
|
|
usage = _replacement_usage()
|
|
out, book, rows = _remediation_ledger(tmp_path, usage)
|
|
passed, evidence = EVAL._historical_remediation(out, book, rows)
|
|
assert passed is True
|
|
assert sum(item["status"] == "replacement_verified" for item in evidence) == 3
|
|
assert book.snapshot()["budget"]["historical_provider_call_count"] == 4
|
|
|
|
|
|
@pytest.mark.parametrize("wrong_key,business_status", [
|
|
(True, "succeeded"), (False, "failed"),
|
|
])
|
|
def test_wrong_key_or_business_failed_replacement_still_blocks(
|
|
tmp_path, wrong_key, business_status,
|
|
):
|
|
usage = _replacement_usage("other-case" if wrong_key else "case-0")
|
|
out, book, rows = _remediation_ledger(tmp_path, usage,
|
|
business_status=business_status)
|
|
passed, evidence = EVAL._historical_remediation(out, book, rows)
|
|
assert passed is False
|
|
unresolved = [item for item in evidence if item["status"] == "replacement_required"]
|
|
assert unresolved
|
|
if business_status == "failed" and not wrong_key:
|
|
assert all(item["business_passed"] is False for item in unresolved)
|
|
|
|
|
|
def test_real_execution_requires_and_binds_known_historical_usage(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
registry = tmp_path.parent / f"{tmp_path.name}-rc05-registry"
|
|
registry.mkdir(parents=True)
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry))
|
|
wrong_registry = tmp_path / "another-registry"
|
|
wrong_registry.mkdir()
|
|
with pytest.raises(ValueError, match="single configured campaign registry"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "wrong-registry", RealFactory,
|
|
historical_usage=_historical_usage(), execution_mode="real",
|
|
run_registry_dir=wrong_registry, case_ids=["case-0"],
|
|
candidate_inputs=_candidate_inputs())
|
|
with pytest.raises(ValueError, match="historical usage receipt"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "missing-history", RealFactory, execution_mode="real",
|
|
run_registry_dir=registry, case_ids=["case-0"],
|
|
candidate_inputs=_candidate_inputs())
|
|
with pytest.raises(ValueError, match="historical usage declaration"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "wrong-history", RealFactory,
|
|
historical_usage={**_historical_usage(), "provider_call_count": 3},
|
|
execution_mode="real", run_registry_dir=registry, case_ids=["case-0"],
|
|
candidate_inputs=_candidate_inputs())
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "bound-history", RealFactory, historical_usage=_historical_usage(),
|
|
execution_mode="real", run_registry_dir=registry, case_ids=["case-0"],
|
|
candidate_inputs=_candidate_inputs())
|
|
assert calls == ["case-0"]
|
|
assert summary["current_selection_passed"] is True
|
|
assert summary["selected_passed"] is False
|
|
assert summary["overall_release_passed"] is False
|
|
assert summary["ledger"]["budget"]["new_provider_call_limit"] == 276
|
|
assert summary["ledger"]["budget"]["new_repair_call_limit"] == 12
|
|
assert summary["combined_call_counts"] == {
|
|
"attempted": 5, "terminal": 5, "unknown": 0,
|
|
}
|
|
assert summary["historical_by_operation"]["draft"]["provider_outcomes"] == {
|
|
"succeeded": 1, "failed": 2, "cancelled": 0, "timeout": 0,
|
|
}
|
|
assert summary["historical_by_operation"]["draft"]["business_outcomes"] == {
|
|
"succeeded": 1, "failed": 2, "cancelled": 0,
|
|
"awaiting_human": 0, "other": 0,
|
|
}
|
|
assert summary["historical_by_operation"]["draft"]["elapsed_seconds"] == {
|
|
"observed": 3, "p50": 2.0, "p95": 2.9,
|
|
}
|
|
assert summary["operation_statistics"]["draft"]["combined"]["provider_outcomes"] == {
|
|
"succeeded": 2, "failed": 2, "cancelled": 0, "timeout": 0,
|
|
}
|
|
assert summary["operation_statistics"]["draft"]["combined"]["elapsed_seconds"] == {
|
|
"observed": 4, "p50": 2.25, "p95": 2.925,
|
|
}
|
|
identity_path = Path(summary["ledger"]["authority"]["registry_root"]) / summary["ledger"]["authority"]["run_key"] / "identity.json"
|
|
identity = json.loads(identity_path.read_text())
|
|
assert identity["historical_usage"]["provider_call_count"] == 4
|
|
assert identity["historical_usage"]["attempts"] == _historical_usage()["attempts"]
|
|
assert len(identity["historical_usage"]["attempts_sha256"]) == 64
|
|
|
|
|
|
def test_real_execution_rejects_unbound_candidate_and_initial_full_dispatch(
|
|
monkeypatch, tmp_path,
|
|
):
|
|
plan = _plan(tmp_path, count=2)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
registry = tmp_path.parent / f"{tmp_path.name}-rc05-preflight-registry"
|
|
registry.mkdir()
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry))
|
|
with pytest.raises(ValueError, match="complete immutable candidate inputs"):
|
|
EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "unbound", RealFactory,
|
|
historical_usage=_historical_usage(), execution_mode="real",
|
|
run_registry_dir=registry, case_ids=["case-0"],
|
|
)
|
|
assert calls == []
|
|
with pytest.raises(EVAL.CallLedgerError, match="explicit predefined canary"):
|
|
EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "initial-full", RealFactory,
|
|
historical_usage=_historical_usage(), execution_mode="real",
|
|
run_registry_dir=registry, candidate_inputs=_candidate_inputs(),
|
|
)
|
|
assert calls == []
|
|
|
|
|
|
def test_authoritative_campaign_registry_requires_explicit_empty_directory(
|
|
monkeypatch, tmp_path,
|
|
):
|
|
plan = _plan(tmp_path, count=1)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
registry = tmp_path.parent / f"{tmp_path.name}-rc05-registry"
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry))
|
|
with pytest.raises(ValueError, match="must be pre-created"):
|
|
EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", RealFactory, historical_usage=_historical_usage(),
|
|
execution_mode="real", run_registry_dir=registry, case_ids=["case-0"],
|
|
candidate_inputs=_candidate_inputs())
|
|
assert calls == []
|
|
|
|
|
|
def test_authoritative_campaign_registry_requires_external_absolute_configuration(
|
|
monkeypatch, tmp_path,
|
|
):
|
|
runner = _runner(tmp_path)
|
|
external = tmp_path.parent / f"{tmp_path.name}-rc05-registry"
|
|
external.mkdir()
|
|
|
|
monkeypatch.delenv(EVAL.REAL_RUN_REGISTRY_ENV, raising=False)
|
|
with pytest.raises(ValueError, match=EVAL.REAL_RUN_REGISTRY_ENV):
|
|
EVAL._authoritative_registry(runner, external)
|
|
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, "relative/registry")
|
|
with pytest.raises(ValueError, match="absolute path outside"):
|
|
EVAL._authoritative_registry(runner, external)
|
|
|
|
inside = tmp_path / "registry"
|
|
inside.mkdir()
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(inside))
|
|
with pytest.raises(ValueError, match="outside the source worktree"):
|
|
EVAL._authoritative_registry(runner, inside)
|
|
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(external))
|
|
with pytest.raises(ValueError, match="single configured campaign registry"):
|
|
EVAL._authoritative_registry(runner, Path("relative/registry"))
|
|
|
|
assert EVAL._authoritative_registry(runner, external) == external.resolve()
|
|
|
|
|
|
def test_historical_usage_receipt_is_strict_and_raw_hash_bound(tmp_path):
|
|
runner = _runner(tmp_path)
|
|
path = tmp_path / "history.json"
|
|
run = tmp_path / "historical-run"
|
|
ledger_root = run / ".call-ledger"
|
|
identity = {
|
|
"schema_version": "multiscene-call-ledger.v1", "real_provider": True,
|
|
"plan_sha256": "1" * 64,
|
|
"case_limits": {f"prior-case-{index}": 1 for index in range(4)},
|
|
"operation_limits": {"draft": 3, "semantic_review": 1},
|
|
}
|
|
identity_path = ledger_root / "identity.json"
|
|
_write_x(identity_path, identity)
|
|
identity_file_sha = _sha(identity_path.read_bytes())
|
|
identity_digest = _sha(json.dumps(identity, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":")).encode("utf-8"))
|
|
attempts = []
|
|
for index in range(4):
|
|
operation = "draft" if index < 3 else "semantic_review"
|
|
provider_status = "failed" if index < 2 else "succeeded"
|
|
business_status = "failed" if index < 2 or index == 3 else "succeeded"
|
|
case_id = f"prior-case-{index}"
|
|
job_operation = ("contract_draft" if operation == "draft"
|
|
else "contract_draft_semantic_review")
|
|
job_id, audit_id = f"job-{index}", f"run-{index}"
|
|
request = {"case": index}
|
|
request_sha = _sha(json.dumps(request, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":")).encode("utf-8"))
|
|
response_raw = f"response-{index}"
|
|
response_sha = _sha(response_raw.encode("utf-8"))
|
|
claim = {"schema_version": "multiscene-call-claim.v1",
|
|
"ledger_identity_sha256": identity_digest, "claim_id": f"{index:032x}",
|
|
"case_id": case_id, "operation": operation,
|
|
"request_sha256": request_sha, "attempt_number": 1}
|
|
finish = {"schema_version": "multiscene-call-finish.v1",
|
|
"ledger_identity_sha256": identity_digest, "claim_id": f"{index:032x}",
|
|
"case_id": case_id, "operation": operation, "status": provider_status,
|
|
"response_sha256": response_sha}
|
|
claim_path = ledger_root / "attempts" / f"{case_id}--{operation}.claim.json"
|
|
finish_path = ledger_root / "attempts" / f"{case_id}--{operation}.finish.json"
|
|
_write_x(claim_path, claim)
|
|
_write_x(finish_path, finish)
|
|
case_dir = run / case_id
|
|
before = case_dir / "before.json"
|
|
_write_x(before, {"request": request, "request_sha256": request_sha,
|
|
"job": {"id": job_id, "operation": job_operation}})
|
|
output = {"result": index}
|
|
execution_evidence = ({
|
|
"input_sha256": request_sha,
|
|
"execution_identity": {"output_sha256": _sha(json.dumps(
|
|
output, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":")).encode("utf-8"))},
|
|
} if operation == "draft" else {
|
|
"request_sha256": request_sha,
|
|
"wire_output_raw": response_raw,
|
|
"wire_output_sha256": response_sha,
|
|
"output_sha256": _sha(json.dumps(output, ensure_ascii=False,
|
|
sort_keys=True, separators=(",", ":")).encode("utf-8")),
|
|
})
|
|
after = case_dir / "after.json"
|
|
_write_x(after, {
|
|
"execution_elapsed_seconds": float(index + 1),
|
|
"job": {
|
|
"id": job_id, "operation": job_operation, "audit_run_id": audit_id,
|
|
"status": business_status,
|
|
"provider_status": provider_status,
|
|
"error_code": "provider_run_failed" if index == 0 else None,
|
|
"error_message": "provider failed" if index == 0 else None,
|
|
},
|
|
"audit": {"id": audit_id, "provider": "codex-cli-development",
|
|
"model": "gpt-5.6-sol", "provider_status": provider_status,
|
|
"provider_error": "failure" if index == 0 else None},
|
|
"output": output, "evidence": execution_evidence,
|
|
})
|
|
attempts.append({
|
|
"attempt_id": f"prior-{index}",
|
|
"plan_sha256": "1" * 64, "case_id": case_id,
|
|
"operation": operation,
|
|
"provider_status": provider_status,
|
|
"business_status": business_status,
|
|
"elapsed_seconds": float(index + 1),
|
|
"timed_out": False,
|
|
"request_sha256": request_sha, "response_sha256": response_sha,
|
|
"provider": "codex-cli-development", "model": "gpt-5.6-sol",
|
|
"job_id": job_id, "audit_run_id": audit_id,
|
|
"identity_path": str(identity_path),
|
|
"identity_file_sha256": identity_file_sha,
|
|
"ledger_identity_sha256": identity_digest,
|
|
"claim_path": str(claim_path), "claim_sha256": _sha(claim_path.read_bytes()),
|
|
"finish_path": str(finish_path), "finish_sha256": _sha(finish_path.read_bytes()),
|
|
"before_path": str(before), "before_sha256": _sha(before.read_bytes()),
|
|
"after_path": str(after), "after_sha256": _sha(after.read_bytes()),
|
|
"disposition": {"resolution_status": (
|
|
"resolved" if business_status == "succeeded" else "requires_replacement"),
|
|
"release_eligible": business_status == "succeeded",
|
|
"reason": "verified success" if business_status == "succeeded" else "retained failure",
|
|
"replacement_case_id": None if business_status == "succeeded" else case_id,
|
|
"replacement_operation": None if business_status == "succeeded" else operation},
|
|
})
|
|
payload = {
|
|
"schema_version": "rc05-historical-provider-usage.v3",
|
|
"attempts": attempts,
|
|
}
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
usage = EVAL.load_historical_usage(runner, path)
|
|
assert usage == {
|
|
"provider_call_count": 4,
|
|
"repair_budget_charged": 4,
|
|
"attempts": payload["attempts"],
|
|
"receipt_sha256": _sha(path.read_bytes()),
|
|
}
|
|
path.write_text(json.dumps({**payload, "attempts": payload["attempts"][:3]}), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="four known historical calls"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
unknown = json.loads(json.dumps(payload))
|
|
unknown["attempts"][0]["provider_status"] = "unknown"
|
|
path.write_text(json.dumps(unknown), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="provider status is invalid"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
failed_but_resolved = json.loads(json.dumps(payload))
|
|
failed_but_resolved["attempts"][0]["business_status"] = "resolved"
|
|
failed_but_resolved["attempts"][0]["disposition"] = {
|
|
"resolution_status": "resolved", "release_eligible": True,
|
|
"reason": "must not hide provider failure",
|
|
"replacement_case_id": None, "replacement_operation": None,
|
|
}
|
|
path.write_text(json.dumps(failed_but_resolved), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="provider/business failure"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
mismatched = json.loads(json.dumps(payload))
|
|
mismatched["attempts"][0]["elapsed_seconds"] = 999
|
|
path.write_text(json.dumps(mismatched), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="declaration differs from evidence"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
synthetic_identity = {**identity, "real_provider": False}
|
|
identity_path.write_text(json.dumps(synthetic_identity), encoding="utf-8")
|
|
synthetic = json.loads(json.dumps(payload))
|
|
synthetic["attempts"][0]["identity_file_sha256"] = _sha(identity_path.read_bytes())
|
|
path.write_text(json.dumps(synthetic), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="matching real run"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
identity_path.unlink()
|
|
_write_x(identity_path, identity)
|
|
mixed = json.loads(json.dumps(payload))
|
|
mixed["attempts"][0]["after_path"] = payload["attempts"][1]["after_path"]
|
|
mixed["attempts"][0]["after_sha256"] = payload["attempts"][1]["after_sha256"]
|
|
path.write_text(json.dumps(mixed), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="before/after|job_id|declaration differs"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
Path(payload["attempts"][0]["after_path"]).write_text("{}", encoding="utf-8")
|
|
path.write_text(json.dumps(payload), encoding="utf-8")
|
|
with pytest.raises(ValueError, match="SHA-256 mismatch"):
|
|
EVAL.load_historical_usage(runner, path)
|
|
|
|
|
|
def test_operation_statistics_report_outcomes_timeout_and_percentiles(tmp_path):
|
|
out = tmp_path / "statistics"
|
|
ledger = EVAL.CallLedger(
|
|
out, "a" * 64, {"draft": 2}, {"case-0": 1, "case-1": 1},
|
|
{"source.py": "b" * 64}, False,
|
|
)
|
|
first = ledger.claim("case-0", "draft", "c" * 64)
|
|
ledger.finish(first, "succeeded", "d" * 64)
|
|
second = ledger.claim("case-1", "draft", "c" * 64)
|
|
ledger.finish(second, "failed")
|
|
_write_x(out / "case-0/01-draft-after.json", {
|
|
"case_id": "case-0", "operation": "draft",
|
|
"request_sha256": "c" * 64, "response_sha256": "d" * 64,
|
|
"provider_invocation_count": 1,
|
|
"execution_elapsed_seconds": 10,
|
|
"job": {"operation": "contract_draft", "status": "succeeded",
|
|
"provider_status": "succeeded", "error_code": None, "error_message": None},
|
|
"audit": {"provider_error": None},
|
|
})
|
|
_write_x(out / "case-1/01-draft-after.json", {
|
|
"case_id": "case-1", "operation": "draft",
|
|
"request_sha256": "c" * 64, "response_sha256": None,
|
|
"provider_invocation_count": 1,
|
|
"execution_elapsed_seconds": 20,
|
|
"job": {"operation": "contract_draft", "status": "failed",
|
|
"provider_status": "failed", "error_code": "provider_run_failed",
|
|
"error_message": "provider timed out"},
|
|
"audit": {"provider_error": "timeout"},
|
|
})
|
|
statistics, errors = EVAL._operation_statistics(out, ledger)
|
|
assert errors == []
|
|
draft = statistics["draft"]
|
|
assert {key: draft[key] for key in (
|
|
"limit", "attempted", "terminal", "unknown", "succeeded", "failed",
|
|
"cancelled", "awaiting_human", "other", "timeout", "business_outcomes",
|
|
"provider_outcomes", "elapsed_seconds",
|
|
)} == {
|
|
"limit": 2, "attempted": 2, "terminal": 2, "unknown": 0,
|
|
"succeeded": 1, "failed": 1, "cancelled": 0,
|
|
"awaiting_human": 0, "other": 0, "timeout": 1,
|
|
"business_outcomes": {
|
|
"succeeded": 1, "failed": 1, "cancelled": 0,
|
|
"awaiting_human": 0, "other": 0,
|
|
},
|
|
"provider_outcomes": {
|
|
"succeeded": 1, "failed": 1, "cancelled": 0, "timeout": 1,
|
|
},
|
|
"elapsed_seconds": {"observed": 2, "p50": 15.0, "p95": 19.5},
|
|
}
|
|
assert draft["historical"] == {
|
|
"attempted": 0, "terminal": 0, "unknown": 0,
|
|
"provider_outcomes": {
|
|
"succeeded": 0, "failed": 0, "cancelled": 0, "timeout": 0,
|
|
},
|
|
"business_outcomes": {
|
|
"succeeded": 0, "failed": 0, "cancelled": 0,
|
|
"awaiting_human": 0, "other": 0,
|
|
},
|
|
"elapsed_seconds": {"observed": 0, "p50": None, "p95": None},
|
|
}
|
|
assert draft["combined"] == {
|
|
"attempted": 2, "terminal": 2, "unknown": 0,
|
|
"provider_outcomes": {
|
|
"succeeded": 1, "failed": 1, "cancelled": 0, "timeout": 1,
|
|
},
|
|
"business_outcomes": {
|
|
"succeeded": 1, "failed": 1, "cancelled": 0,
|
|
"awaiting_human": 0, "other": 0,
|
|
},
|
|
"elapsed_seconds": {"observed": 2, "p50": 15.0, "p95": 19.5},
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("raw", [
|
|
b'{"schema_version":"rc05-historical-provider-usage.v3","attempts":[],"attempts":[]}',
|
|
b'{"schema_version":"rc05-historical-provider-usage.v3","attempts":NaN}',
|
|
])
|
|
def test_historical_receipt_rejects_duplicate_keys_and_nonfinite_json(tmp_path, raw):
|
|
receipt = tmp_path / "invalid.json"
|
|
receipt.write_bytes(raw)
|
|
with pytest.raises(ValueError, match="not valid JSON"):
|
|
EVAL.load_historical_usage(_runner(tmp_path), receipt)
|
|
|
|
|
|
def test_business_failure_is_not_hidden_by_provider_success(tmp_path):
|
|
out = tmp_path / "business-failure"
|
|
ledger = EVAL.CallLedger(
|
|
out, "a" * 64, {"semantic_review": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, False,
|
|
)
|
|
token = ledger.claim("case-0", "semantic_review", "c" * 64)
|
|
ledger.finish(token, "succeeded", "d" * 64)
|
|
_write_x(out / "case-0/02-semantic-after.json", {
|
|
"case_id": "case-0", "operation": "semantic_review",
|
|
"request_sha256": "c" * 64, "response_sha256": "d" * 64,
|
|
"provider_invocation_count": 1,
|
|
"execution_elapsed_seconds": 12.0,
|
|
"job": {
|
|
"operation": "contract_draft_semantic_review",
|
|
"status": "failed",
|
|
"provider_status": "succeeded",
|
|
"error_code": "semantic_review_validation_failed",
|
|
"error_message": "citation validation failed",
|
|
},
|
|
"audit": {"provider_error": None},
|
|
})
|
|
statistics, errors = EVAL._operation_statistics(out, ledger)
|
|
assert errors == []
|
|
semantic = statistics["semantic_review"]
|
|
assert semantic["succeeded"] == 0
|
|
assert semantic["failed"] == 1
|
|
assert semantic["provider_outcomes"] == {
|
|
"succeeded": 1, "failed": 0, "cancelled": 0, "timeout": 0,
|
|
}
|
|
|
|
|
|
def test_after_evidence_wrong_job_operation_cannot_count_as_success(tmp_path):
|
|
out = tmp_path / "wrong-operation"
|
|
ledger = EVAL.CallLedger(
|
|
out, "a" * 64, {"draft": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, False,
|
|
)
|
|
token = ledger.claim("case-0", "draft", "c" * 64)
|
|
ledger.finish(token, "succeeded", "d" * 64)
|
|
_write_x(out / "case-0/01-draft-after.json", {
|
|
"case_id": "case-0", "operation": "draft",
|
|
"request_sha256": "c" * 64, "response_sha256": "d" * 64,
|
|
"provider_invocation_count": 1, "execution_elapsed_seconds": 1,
|
|
"job": {"operation": "contract_intake", "status": "succeeded",
|
|
"provider_status": "succeeded", "error_code": None,
|
|
"error_message": None},
|
|
"audit": {},
|
|
})
|
|
statistics, errors = EVAL._operation_statistics(out, ledger)
|
|
assert errors == [{"case_id": "case-0", "operation": "draft",
|
|
"error": "after_evidence_operation_mismatch"}]
|
|
assert statistics["draft"]["succeeded"] == 0
|
|
|
|
|
|
def test_real_after_evidence_requires_matching_complete_audit_identity(tmp_path):
|
|
out = tmp_path / "real-audit-mismatch"
|
|
ledger = EVAL.CallLedger(
|
|
out, "a" * 64, {"draft": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, True,
|
|
)
|
|
token = ledger.claim("case-0", "draft", "c" * 64)
|
|
ledger.finish(token, "succeeded", "d" * 64)
|
|
evidence = {
|
|
"case_id": "case-0", "operation": "draft",
|
|
"request_sha256": "c" * 64, "response_sha256": "d" * 64,
|
|
"provider_invocation_count": 1, "execution_elapsed_seconds": 1,
|
|
"job": {"id": "job-1", "operation": "contract_draft",
|
|
"status": "succeeded", "provider_status": "succeeded",
|
|
"audit_run_id": "audit-1", "error_code": None,
|
|
"error_message": None},
|
|
"audit": {"id": "audit-1", "provider": "codex-cli-development",
|
|
"model": "gpt-5.6-sol", "provider_status": "failed"},
|
|
}
|
|
after = out / "case-0/01-draft-after.json"
|
|
_write_x(after, evidence)
|
|
statistics, errors = EVAL._operation_statistics(out, ledger)
|
|
assert errors == [{"case_id": "case-0", "operation": "draft",
|
|
"error": "after_evidence_real_provider_mismatch"}]
|
|
assert EVAL._current_operation_outcomes_passed(statistics) is False
|
|
|
|
after.unlink()
|
|
evidence["audit"]["provider_status"] = "succeeded"
|
|
evidence["job"]["id"] = ""
|
|
_write_x(after, evidence)
|
|
statistics, errors = EVAL._operation_statistics(out, ledger)
|
|
assert errors[0]["error"] == "after_evidence_real_provider_mismatch"
|
|
assert EVAL._current_operation_outcomes_passed(statistics) is False
|
|
|
|
|
|
def test_matrix_execution_and_release_quality_are_separate(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
|
|
def uncalibrated(*args, **kwargs):
|
|
return {**original(*args, **kwargs),
|
|
"automated_quality_passed": True,
|
|
"human_calibration_completed": False,
|
|
"content_quality_release_passed": False}
|
|
|
|
supplemental.run_supplemental_case = uncalibrated
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
assert summary["matrix_execution_passed"] is True
|
|
assert summary["current_run_passed"] is True
|
|
assert summary["release_quality_gate_passed"] is False
|
|
assert summary["overall_release_passed"] is False
|
|
|
|
|
|
@pytest.mark.parametrize("business_status,outcome", [
|
|
("failed", "failed"), ("resolved", "other"),
|
|
])
|
|
def test_business_failure_in_after_evidence_cannot_green_release(
|
|
monkeypatch, tmp_path, business_status, outcome,
|
|
):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
|
|
def inconsistent(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
directory = Path(args[1])
|
|
after = directory / "01-draft-after.json"
|
|
payload = json.loads(after.read_text(encoding="utf-8"))
|
|
payload["job"]["status"] = business_status
|
|
payload["job"]["error_code"] = "business_validation_failed"
|
|
after.write_text(json.dumps(payload), encoding="utf-8")
|
|
return result
|
|
|
|
supplemental.run_supplemental_case = inconsistent
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
assert summary["operation_statistics"]["draft"]["business_outcomes"][outcome] == 1
|
|
assert summary["matrix_execution_passed"] is False
|
|
assert summary["current_selection_passed"] is False
|
|
assert summary["passed"] is False
|
|
|
|
|
|
def test_missing_explicit_quality_fields_cannot_default_green(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
|
|
def missing_quality(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
for field in (
|
|
"automated_quality_passed", "automated_model_quality_release_passed",
|
|
"content_quality_release_passed",
|
|
):
|
|
result.pop(field)
|
|
return result
|
|
|
|
supplemental.run_supplemental_case = missing_quality
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
assert summary["matrix_execution_passed"] is True
|
|
assert summary["release_quality_gate_passed"] is False
|
|
assert summary["selected_release_quality_passed"] is False
|
|
assert summary["passed"] is False
|
|
|
|
|
|
def test_completed_case_with_zero_provider_calls_cannot_green_matrix(
|
|
monkeypatch, tmp_path,
|
|
):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
|
|
def forged_without_call(case, directory, _factory, state, *, ledger, runner):
|
|
directory.mkdir(parents=True)
|
|
return {
|
|
"case_id": case["case_id"], "status": "completed", "passed": True,
|
|
"provider_call_count": 0, "elapsed_seconds": 0.1,
|
|
"automated_quality_passed": True,
|
|
"automated_model_quality_release_passed": True,
|
|
"content_quality_release_passed": True,
|
|
}
|
|
|
|
supplemental.run_supplemental_case = forged_without_call
|
|
summary = EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
assert summary["complete"] is True
|
|
assert summary["provider_call_count"] == 0
|
|
assert summary["case_operation_coverage_passed"] is False
|
|
assert summary["matrix_execution_passed"] is False
|
|
assert summary["overall_release_passed"] is False
|
|
|
|
|
|
def test_missing_mandatory_operation_cannot_green_matrix(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
plan["cases"][0].update(
|
|
provider_call_budget=2,
|
|
operations=["draft", "semantic_review"],
|
|
)
|
|
plan["operation_limits"] = {"draft": 1, "semantic_review": 1}
|
|
_install_modules(monkeypatch, plan)
|
|
summary = EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
coverage = summary["case_operation_coverage"][0]
|
|
assert coverage["actual_operations"] == ["draft"]
|
|
assert coverage["mandatory_operations"] == ["draft", "semantic_review"]
|
|
assert coverage["passed"] is False
|
|
assert summary["overall_release_passed"] is False
|
|
|
|
|
|
def test_foreign_plan_campaign_attempt_blocks_current_plan_green(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
usage = _historical_usage()
|
|
registry = tmp_path.parent / f"{tmp_path.name}-rc05-registry"
|
|
registry.mkdir(parents=True)
|
|
monkeypatch.setenv(EVAL.REAL_RUN_REGISTRY_ENV, str(registry))
|
|
old = EVAL.CallLedger(
|
|
tmp_path / "old-output", "1" * 64, plan["operation_limits"],
|
|
{"case-0": 1}, plan["source_pins"], True,
|
|
historical_usage=usage, registry_root=registry,
|
|
)
|
|
token = old.claim("case-0", "draft", "b" * 64)
|
|
old.finish(token, "failed")
|
|
|
|
_install_modules(monkeypatch, plan, digest="2" * 64)
|
|
summary = EVAL.execute(
|
|
_runner(tmp_path), tmp_path / "plan.json", "2" * 64,
|
|
tmp_path / "current-output", RealFactory,
|
|
historical_usage=usage, execution_mode="real", run_registry_dir=registry,
|
|
case_ids=["case-0"], candidate_inputs=_candidate_inputs(),
|
|
)
|
|
assert summary["ledger"]["campaign"]["attempted"] == 2
|
|
assert summary["ledger"]["campaign"]["foreign_plan_attempted"] == 1
|
|
assert summary["current_plan_provider_call_count"] == 1
|
|
assert summary["current_selection_passed"] is False
|
|
assert summary["matrix_execution_passed"] is False
|
|
|
|
|
|
def test_resume_rejects_result_tamper_without_new_provider_call(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
calls = _install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
|
|
def rejected(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
return {**result, "passed": False, "automated_quality_passed": False,
|
|
"automated_model_quality_release_passed": False,
|
|
"content_quality_release_passed": False}
|
|
|
|
supplemental.run_supplemental_case = rejected
|
|
out = tmp_path / "out"
|
|
first = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
out, SyntheticFactory, execution_mode="synthetic")
|
|
assert first["passed"] is False
|
|
result_path = out / "case-0/result.json"
|
|
tampered = json.loads(result_path.read_text(encoding="utf-8"))
|
|
tampered.update(passed=True, automated_quality_passed=True,
|
|
automated_model_quality_release_passed=True,
|
|
content_quality_release_passed=True)
|
|
result_path.write_text(json.dumps(tampered), encoding="utf-8")
|
|
|
|
resumed = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
out, SyntheticFactory, resume=True, execution_mode="synthetic")
|
|
assert calls == ["case-0"]
|
|
assert resumed["cases"][0]["status"] == "unknown"
|
|
assert resumed["passed"] is False
|
|
|
|
|
|
def test_source_change_and_restore_during_case_is_detected(monkeypatch, tmp_path):
|
|
plan = _plan(tmp_path, count=1)
|
|
_install_modules(monkeypatch, plan)
|
|
supplemental = sys.modules["multiscene_supplemental_cases"]
|
|
original = supplemental.run_supplemental_case
|
|
source = tmp_path / "source.py"
|
|
|
|
def transient_drift(*args, **kwargs):
|
|
result = original(*args, **kwargs)
|
|
source.write_text("temporary drift\n", encoding="utf-8")
|
|
source.write_text("frozen\n", encoding="utf-8")
|
|
return result
|
|
|
|
supplemental.run_supplemental_case = transient_drift
|
|
summary = EVAL.execute(_runner(tmp_path), tmp_path / "plan.json", "a" * 64,
|
|
tmp_path / "out", SyntheticFactory, execution_mode="synthetic")
|
|
failure = json.loads((tmp_path / "out/case-0/failure.json").read_text(encoding="utf-8"))
|
|
assert "source drift during case case-0" in failure["error"]
|
|
assert summary["cases"][0]["status"] == "unknown"
|
|
assert summary["passed"] is False
|
|
|
|
|
|
def test_provider_failed_resolved_history_requires_replacement(tmp_path):
|
|
usage = _historical_usage()
|
|
attempt = usage["attempts"][0]
|
|
attempt["business_status"] = "resolved"
|
|
attempt["disposition"] = {
|
|
"resolution_status": "resolved", "release_eligible": True,
|
|
"reason": "must not hide provider failure",
|
|
"replacement_case_id": None, "replacement_operation": None,
|
|
}
|
|
with pytest.raises(ValueError, match="provider/business failure"):
|
|
EVAL.CallLedger(
|
|
tmp_path / "out", "a" * 64, {"draft": 1}, {"case-0": 1},
|
|
{"source.py": "b" * 64}, False, historical_usage=usage,
|
|
)
|
|
|
|
fake_ledger = types.SimpleNamespace(
|
|
historical_usage={"attempts": [attempt]}, attempt_records=lambda: [])
|
|
passed, remediation = EVAL._historical_remediation(
|
|
tmp_path, fake_ledger, [])
|
|
assert passed is False
|
|
assert remediation[0]["evidence_error"] == "historical_release_eligibility_invalid"
|