File size: 4,907 Bytes
3e94469 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 | """Local, development-only real-patch validation and forecast interface.
The forecast path reads only the model-visible request. Oracle scoring is a
separate explicit operation, never part of forecast().
"""
from __future__ import annotations
import argparse
import hashlib
import importlib.util
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parent / "capsule/code_review"
CONTRACT = ROOT / "contracts/real_patch_prediction_v1/validator.py"
EXACT = ROOT / "scripts/run_real_patch_visible_exact_v1.py"
GATE = ROOT / "scripts/run_real_patch_baseline_gate_v2.py"
def _module(path: Path, name: str):
spec = importlib.util.spec_from_file_location(name, path)
if spec is None or spec.loader is None:
raise RuntimeError(f"packaged module cannot be loaded: {name}")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _sha(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def validate(request_path: Path) -> dict:
contract = _module(CONTRACT, "laxmi_real_patch_contract")
request = contract.load_forecast_request(request_path)
return {
"schema_version": "laxmi.real-patch-local-api.v1",
"status": "ADMITTED",
"case_id": request["case_id"],
"input_sha256": contract.input_sha256(request),
"request_file_sha256": _sha(request_path),
"evaluation_split": request["evaluation_split"],
"custody_verified": False,
"acceptance_eligible": False,
"executed": False,
}
def forecast(request_path: Path, output_dir: Path) -> dict:
admitted = validate(request_path)
exact = _module(EXACT, "laxmi_real_patch_visible_exact")
gate = _module(GATE, "laxmi_real_patch_gate")
output_dir.mkdir(mode=0o700, parents=True, exist_ok=False)
prediction, execution = exact.run(request_path, output_dir / "prediction")
sealed = gate.seal(request_path, output_dir / "prediction/predictions.json", output_dir / "seal")
if prediction["input_sha256"] != admitted["input_sha256"] or sealed["input_sha256"] != admitted["input_sha256"]:
raise ValueError("prediction changed the admitted input identity")
row = prediction["comparators"]["EXACT_VISIBLE_SOURCE"]
return {
"schema_version": "laxmi.real-patch-local-api.v1",
"status": "PREDICTED" if row["status"] == "predicted" else "UNSUPPORTED",
"engine_id": "EXACT_VISIBLE_SOURCE",
"information_tier": "SAME_INFORMATION",
"case_id": admitted["case_id"],
"input_sha256": admitted["input_sha256"],
"method_sha256": row.get("method_sha256"),
"reason": row.get("reason"),
"prediction_file_sha256": _sha(output_dir / "prediction/predictions.json"),
"seal_file_sha256": _sha(output_dir / "seal/seal.json"),
"oracle_read": execution["oracle_read"],
"custody_verified": False,
"acceptance_eligible": False,
"development_only": True,
}
def score(request_path: Path, oracle_path: Path, output_dir: Path) -> dict:
"""Privileged local evaluation call; never invoked by forecast()."""
gate = _module(GATE, "laxmi_real_patch_gate")
receipt = gate.score(request_path, oracle_path, output_dir / "seal")
return {
"schema_version": "laxmi.real-patch-local-api.v1",
"status": "SCORED_DEVELOPMENT_ORACLE",
"case_id": receipt["case_id"],
"input_sha256": receipt["input_sha256"],
"receipt_file_sha256": _sha(output_dir / "seal/receipt.json"),
"same_information_scored_count": receipt["same_information_scored_count"],
"development_only": True,
}
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
commands = parser.add_subparsers(dest="command", required=True)
validation = commands.add_parser("validate")
validation.add_argument("--request", type=Path, required=True)
prediction = commands.add_parser("forecast")
prediction.add_argument("--request", type=Path, required=True)
prediction.add_argument("--output-dir", type=Path, required=True)
scoring = commands.add_parser("score")
scoring.add_argument("--request", type=Path, required=True)
scoring.add_argument("--oracle", type=Path, required=True)
scoring.add_argument("--output-dir", type=Path, required=True)
args = parser.parse_args()
try:
if args.command == "validate":
result = validate(args.request)
elif args.command == "forecast":
result = forecast(args.request, args.output_dir)
else:
result = score(args.request, args.oracle, args.output_dir)
except (OSError, ValueError, RuntimeError) as exc:
parser.exit(2, f"Laxmi real-patch refused: {exc}\n")
print(json.dumps(result, sort_keys=True, allow_nan=False))
if __name__ == "__main__":
main()
|