build(agent): weasel-1#856f80 iteration
This commit is contained in:
parent
df8bbb15e6
commit
22c31fc61a
|
|
@ -0,0 +1,55 @@
|
||||||
|
"""Small demo runner that certifies a plan and records the attestation to the ledger.
|
||||||
|
|
||||||
|
This demonstrates how the DSL, verifier, and ledger tie together and can be
|
||||||
|
used as a starting point for real adapters and HIL integrations.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import json
|
||||||
|
from typing import Dict, Any
|
||||||
|
|
||||||
|
from ..dsl import SafetyContract, LocalCapabilities, ResourceBudgets, DataSharingPolicy
|
||||||
|
from ..verification import VerificationEngine
|
||||||
|
from ..governance.ledger import Ledger
|
||||||
|
|
||||||
|
|
||||||
|
def run_demo(contract: SafetyContract, plan: Dict[str, Any]) -> Dict[str, Any]:
|
||||||
|
"""Run a deterministic demo: certify plan and add certificate to ledger.
|
||||||
|
|
||||||
|
Returns a dict with certificate and the ledger entry dict.
|
||||||
|
"""
|
||||||
|
engine = VerificationEngine()
|
||||||
|
ledger = Ledger()
|
||||||
|
|
||||||
|
cert = engine.certify(plan, contract)
|
||||||
|
|
||||||
|
# Record attestation in the ledger with minimal metadata
|
||||||
|
payload = {
|
||||||
|
"contract_id": getattr(contract, "contract_id", None),
|
||||||
|
"plan_hash": cert.get("plan_hash"),
|
||||||
|
"certificate_id": cert.get("certificate_id"),
|
||||||
|
"status": cert.get("status"),
|
||||||
|
}
|
||||||
|
entry = ledger.add_entry(payload)
|
||||||
|
|
||||||
|
return {"certificate": cert, "ledger_entry": entry.__dict__, "chain": ledger.get_chain()}
|
||||||
|
|
||||||
|
|
||||||
|
def demo_cli() -> None:
|
||||||
|
"""Create a sample contract/plan, run the demo and print JSON output."""
|
||||||
|
contract = SafetyContract(
|
||||||
|
contract_id="demo-contract-1",
|
||||||
|
local_capabilities=[LocalCapabilities(name="Planner", capabilities=["plan"])],
|
||||||
|
budgets=ResourceBudgets(cpu_cores=1.0, memory_gb=1.0, energy_wh=50.0, time_seconds=600),
|
||||||
|
data_policy=DataSharingPolicy(policy_id="demo-policy", allowed_data=["telemetry"]),
|
||||||
|
version="1.0",
|
||||||
|
)
|
||||||
|
|
||||||
|
plan = {"name": "demo-mission", "actions": ["scan", "navigate"], "budgets": {"cpu_cores": 0.5}}
|
||||||
|
|
||||||
|
out = run_demo(contract, plan)
|
||||||
|
print(json.dumps(out, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
demo_cli()
|
||||||
|
|
@ -0,0 +1,27 @@
|
||||||
|
from spacesafeml_certification_benchmark_and_.examples.demo import run_demo
|
||||||
|
from spacesafeml_certification_benchmark_and_.dsl import (
|
||||||
|
SafetyContract,
|
||||||
|
LocalCapabilities,
|
||||||
|
ResourceBudgets,
|
||||||
|
DataSharingPolicy,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_demo_records_certificate_and_ledger_entry():
|
||||||
|
contract = SafetyContract(
|
||||||
|
contract_id="demo-test-1",
|
||||||
|
local_capabilities=[LocalCapabilities(name="Planner")],
|
||||||
|
budgets=ResourceBudgets(cpu_cores=1.0, memory_gb=1.0, energy_wh=50.0, time_seconds=600),
|
||||||
|
data_policy=DataSharingPolicy(policy_id="p1", allowed_data=["telemetry"]),
|
||||||
|
)
|
||||||
|
|
||||||
|
plan = {"name": "demo-plan", "actions": ["a1", "a2"], "budgets": {"cpu_cores": 0.5}}
|
||||||
|
|
||||||
|
out = run_demo(contract, plan)
|
||||||
|
|
||||||
|
assert "certificate" in out
|
||||||
|
assert "ledger_entry" in out
|
||||||
|
assert out["certificate"]["status"] in ("safe", "unsafe")
|
||||||
|
# chain should contain one entry
|
||||||
|
assert isinstance(out["chain"], list)
|
||||||
|
assert len(out["chain"]) == 1
|
||||||
Loading…
Reference in New Issue