60 lines
2.0 KiB
Python
60 lines
2.0 KiB
Python
"""Safety stub extractor.
|
|
|
|
Extracts a tiny, conservative safety stub from a PlanDelta. The stub contains
|
|
only the minimal actions that are safety-related (e.g., safety_critical
|
|
commitments) and a provenance hash linking back to the full delta snapshot.
|
|
|
|
The stub is intentionally tiny and can be executed in constrained runtimes.
|
|
"""
|
|
from typing import Dict, Any, List
|
|
from .plan_delta import PlanDelta
|
|
import json
|
|
import hashlib
|
|
|
|
|
|
def extract_safety_stub(delta: PlanDelta) -> Dict[str, Any]:
|
|
"""Return a compact stub containing safety-critical commitments and a hash.
|
|
|
|
The stub fields:
|
|
- stub_commitments: list of commitments filtered for safety_critical or
|
|
those that match a small whitelist (e.g., type == 'safe-hold')
|
|
- provenance: short hash of the full delta snapshot
|
|
- origin: delta.origin for traceability
|
|
"""
|
|
stub_commitments: List[Dict[str, Any]] = []
|
|
|
|
for c in delta.commitments.value():
|
|
if not isinstance(c, dict):
|
|
continue
|
|
# Keep explicit safety-critical commitments
|
|
if c.get("safety_critical"):
|
|
stub_commitments.append(_minimalize_commitment(c))
|
|
continue
|
|
# Keep very small whitelist of actions that are safe fallbacks
|
|
if c.get("type") in ("safe-hold", "thruster-off", "throttle-back"):
|
|
stub_commitments.append(_minimalize_commitment(c))
|
|
|
|
prov = _short_hash(delta.snapshot())
|
|
|
|
return {
|
|
"stub_commitments": stub_commitments,
|
|
"provenance": prov,
|
|
"origin": getattr(delta, "origin", ""),
|
|
}
|
|
|
|
|
|
def _minimalize_commitment(c: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Return a tiny projection of a commitment preserving key invariants.
|
|
|
|
Keep only fields that are likely necessary to reason about safety.
|
|
"""
|
|
keys = ["task_id", "type", "fuel", "power", "deadline_step", "safety_critical"]
|
|
return {k: c[k] for k in keys if k in c}
|
|
|
|
|
|
def _short_hash(obj: Any) -> str:
|
|
import hashlib
|
|
|
|
raw = json.dumps(obj, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:16]
|