| """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() |
|
|
