67 lines
2.3 KiB
Python
67 lines
2.3 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
|
|
|
|
from .dsl import SafetyContract
|
|
|
|
|
|
class VerificationEngine:
|
|
def __init__(self):
|
|
pass
|
|
|
|
def certify(self, plan: Dict[str, Any], contract: SafetyContract) -> 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 = {
|
|
"certificate_id": self._trace_id(plan, contract),
|
|
"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) -> str:
|
|
seed = json.dumps({"plan": plan, "contract": asdict(contract) if contract else {}} , sort_keys=True)
|
|
return hashlib.sha256(seed.encode("utf-8")).hexdigest()[:16]
|
|
|
|
|
|
def generate_certificate(plan: Dict[str, Any], contract: SafetyContract) -> Dict[str, Any]:
|
|
engine = VerificationEngine()
|
|
return engine.certify(plan, contract)
|