83 lines
3.4 KiB
Python
83 lines
3.4 KiB
Python
"""Verification harness (MVP).
|
|
|
|
Provides a tiny, deterministic verifier that can produce a safety certificate
|
|
for a given plan against a SafetyContract.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from dataclasses import asdict
|
|
from typing import Dict, Any, Optional
|
|
|
|
from .dsl import SafetyContract
|
|
|
|
|
|
class VerificationEngine:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def certify(self, plan: Dict[str, Any], contract: SafetyContract, scenario_id: Optional[str] = None) -> Dict[str, Any]:
|
|
# Very small, deterministic check:
|
|
# - Ensure plan contains an 'actions' list
|
|
actions = plan.get("actions", [])
|
|
safe = True
|
|
details = []
|
|
if not isinstance(actions, list) or not actions:
|
|
safe = False
|
|
details.append("No actions provided")
|
|
|
|
# Simple budgets check if provided
|
|
budget_ok = True
|
|
budgets = plan.get("budgets", {})
|
|
if contract.budgets and budgets:
|
|
req = contract.budgets
|
|
# Compare a subset similarity
|
|
for k in ("cpu_cores", "memory_gb", "energy_wh", "time_seconds"):
|
|
if k in budgets and getattr(req, k, 0) > 0:
|
|
if budgets[k] > getattr(req, k):
|
|
budget_ok = False
|
|
details.append(f"Budget {k} exceeded: {budgets[k]} > {getattr(req, k)}")
|
|
if not budget_ok:
|
|
safe = False
|
|
|
|
certificate = {
|
|
# include an explicit, scenario-aware trace id for reproducibility
|
|
"certificate_id": self._trace_id(plan, contract, scenario_id),
|
|
"status": "safe" if safe else "unsafe",
|
|
"plan_hash": self._hash_plan(plan),
|
|
"details": details,
|
|
"version": contract.version if contract else "1.0",
|
|
}
|
|
# Attach a minimal certificate payload
|
|
return certificate
|
|
|
|
@staticmethod
|
|
def _hash_plan(plan: Dict[str, Any]) -> str:
|
|
return hashlib.sha256(json.dumps(plan, sort_keys=True).encode("utf-8")).hexdigest()
|
|
|
|
@staticmethod
|
|
def _trace_id(plan: Dict[str, Any], contract: SafetyContract, scenario_id: Optional[str] = None) -> str:
|
|
# Build a compact, deterministic seed that is stable across runs.
|
|
# Prefer an explicit scenario seed when available so replay can be reproduced.
|
|
seed_obj = {"plan": plan, "contract_id": getattr(contract, "contract_id", None)}
|
|
if scenario_id and contract and getattr(contract, "scenarios", None):
|
|
# find the scenario and include its seed (if any) in the trace seed
|
|
for sc in contract.scenarios:
|
|
if sc.scenario_id == scenario_id:
|
|
seed_obj["scenario_id"] = sc.scenario_id
|
|
if getattr(sc, "seed", None) is not None:
|
|
seed_obj["scenario_seed"] = sc.seed
|
|
break
|
|
# Fallback: include full contract dict for compatibility if no scenario matched
|
|
if "scenario_id" not in seed_obj:
|
|
seed_obj["contract"] = asdict(contract) if contract else {}
|
|
|
|
return hashlib.sha256(json.dumps(seed_obj, sort_keys=True).encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def generate_certificate(plan: Dict[str, Any], contract: SafetyContract) -> Dict[str, Any]:
|
|
engine = VerificationEngine()
|
|
# keep public API backward compatible: no scenario_id by default
|
|
return engine.certify(plan, contract)
|