From df8bbb15e6d585a76128626579d8d1c3d2df0a58 Mon Sep 17 00:00:00 2001 From: agent-856f80a92b1141b4 Date: Fri, 24 Apr 2026 20:31:12 +0200 Subject: [PATCH] build(agent): weasel-1#856f80 iteration --- .../dsl.py | 3 ++ .../verification.py | 28 ++++++++++++++---- tests/test_trace_determinism.py | 29 +++++++++++++++++++ 3 files changed, 54 insertions(+), 6 deletions(-) create mode 100644 tests/test_trace_determinism.py diff --git a/spacesafeml_certification_benchmark_and_/dsl.py b/spacesafeml_certification_benchmark_and_/dsl.py index 47fd3ff..7c3fe42 100644 --- a/spacesafeml_certification_benchmark_and_/dsl.py +++ b/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 = "" diff --git a/spacesafeml_certification_benchmark_and_/verification.py b/spacesafeml_certification_benchmark_and_/verification.py index 1bca7fc..d8e13a2 100644 --- a/spacesafeml_certification_benchmark_and_/verification.py +++ b/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) diff --git a/tests/test_trace_determinism.py b/tests/test_trace_determinism.py new file mode 100644 index 0000000..c89a8e8 --- /dev/null +++ b/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"