build(agent): weasel-1#856f80 iteration

spacesafeml_certification_benchmark_and_/dsl.py

@@ -67,4 +67,7 @@ class Scenario:
     sensors: Dict[str, Any] = field(default_factory=dict)
     environment: Dict[str, Any] = field(default_factory=dict)
     deterministic_replay: bool = False
+    # Optional seed for deterministic replay. Small, explicit extension to
+    # support reproducible trace IDs and replay across environments.
+    seed: Optional[int] = None
     description: str = ""

spacesafeml_certification_benchmark_and_/verification.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 import hashlib
 import json
 from dataclasses import asdict
-from typing import Dict, Any
+from typing import Dict, Any, Optional
 
 from .dsl import SafetyContract
 
@@ -17,7 +17,7 @@ class VerificationEngine:
     def __init__(self):
         pass
 
-    def certify(self, plan: Dict[str, Any], contract: SafetyContract) -> Dict[str, Any]:
+    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", [])
@@ -42,7 +42,8 @@ class VerificationEngine:
             safe = False
 
         certificate = {
-            "certificate_id": self._trace_id(plan, contract),
+            # 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,
@@ -56,11 +57,26 @@ class VerificationEngine:
         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 _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)

tests/test_trace_determinism.py

@@ -0,0 +1,29 @@
+import json
+
+from spacesafeml_certification_benchmark_and_.verification import VerificationEngine
+from spacesafeml_certification_benchmark_and_.dsl import (
+    SafetyContract,
+    LocalCapabilities,
+    Scenario,
+    ResourceBudgets,
+)
+
+
+def test_trace_id_is_deterministic_and_sensitive_to_seed():
+    plan = {"name": "mission-x", "actions": ["go", "scan"]}
+
+    # Create a contract with a scenario that has an explicit seed
+    sc = Scenario(scenario_id="sc-1", deterministic_replay=True, seed=12345, description="seeded")
+    contract = SafetyContract(contract_id="c-1", local_capabilities=[LocalCapabilities(name="A")], scenarios=[sc])
+
+    engine = VerificationEngine()
+
+    tid1 = engine._trace_id(plan, contract, scenario_id="sc-1")
+    tid2 = engine._trace_id(plan, contract, scenario_id="sc-1")
+    assert tid1 == tid2, "Trace id should be stable for same plan + scenario seed"
+
+    # Changing the seed should change the trace id
+    sc2 = Scenario(scenario_id="sc-1", deterministic_replay=True, seed=54321, description="seeded-different")
+    contract2 = SafetyContract(contract_id="c-1", local_capabilities=[LocalCapabilities(name="A")], scenarios=[sc2])
+    tid3 = engine._trace_id(plan, contract2, scenario_id="sc-1")
+    assert tid3 != tid1, "Different scenario seed should change trace id"