From d1aaab7de8eeeca78e76959b1186356e7b504e8b Mon Sep 17 00:00:00 2001 From: agent-856f80a92b1141b4 Date: Sat, 25 Apr 2026 21:15:37 +0200 Subject: [PATCH] build(agent): weasel-1#856f80 iteration --- README.md | 38 ++-- pyproject.toml | 19 +- .../__init__.py | 33 +-- .../dsl.py | 193 +++++++++++++----- .../__init__.py | 28 +++ .../dsl.py | 169 +++++++++++++++ test.sh | 13 +- tests/test_dsl.py | 55 +++++ 8 files changed, 442 insertions(+), 106 deletions(-) create mode 100644 src/spacesafeml_certification_benchmark_and_/__init__.py create mode 100644 src/spacesafeml_certification_benchmark_and_/dsl.py create mode 100644 tests/test_dsl.py diff --git a/README.md b/README.md index 2833651..391318d 100644 --- a/README.md +++ b/README.md @@ -1,23 +1,23 @@ -# SpaceSafeML: Certification, Benchmark, and Governance Framework for Onboard AI in Space Robotics +SpaceSafeML — SB-DSL Seed -This repository provides a minimal, open-source MVP of a modular framework to certify and benchmark onboard AI agents operating in space robotics contexts. It includes a Safety DSL, a verification harness, a lightweight simulation scaffold, a governance ledger, and starter adapters for common onboard stacks. +This repository contains a seed implementation of the SpaceSafeML Safety-Benchmark DSL (SB-DSL). -What you can expect in this MVP -- A Python package named `spacesafeml_certification_benchmark_and_` with core modules: - - DSL definitions for LocalCapabilities, SafetyPre/SafetyPostConditions, ResourceBudgets, and DataSharingPolicies - - A simple verification engine that can generate safety certificates for plans - - A tiny simulation scaffold with placeholder Gazebo/ROS-like interfaces for fleet scenarios (deterministic and replayable) - - A tamper-evident ledger to audit test results - - Starter adapters for planning and perception modules -- A basic test suite to validate core behavior and a test launcher script `test.sh` that runs tests and packaging verification -- Documentation file `AGENTS.md` describing architecture and contribution rules +Contents added in this change: +- A small Python package (src/spacesafeml_certification_benchmark_and_) with a `dsl.py` module that defines core DSL models: + - AgentCapabilities + - SafetyPreCondition / SafetyPostCondition + - ResourceBudgets + - DataSharingPolicy + - TelemetrySchema + - Scenario +- A minimal in-repo SchemaRegistry for versioning DSL payloads +- Utilities to serialize/deserialize and compute deterministic manifest fingerprints +- Tests that validate round-trip JSON and fingerprint behavior +- `pyproject.toml` so the package builds cleanly +- `test.sh` which runs the test-suite and `python3 -m build` -Getting started -- Install Python 3.8+ and run tests via `bash test.sh`. -- Explore the MVP modules under `spacesafeml_certification_benchmark_and_`. +This is intentionally a small, well-tested chunk (MVP step 1) to bootstrap the project. Follow-on work includes the verifier, simulation templates, ledger, and adapters. -This project intentionally remains minimal yet extensible to accommodate future MVP expansion consistent with the SpaceSafeML vision. - -## License - -MIT +Quick commands: +- Run tests: `bash test.sh` +- Build package: `python3 -m build` diff --git a/pyproject.toml b/pyproject.toml index c2803c6..b0c2262 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,24 +1,13 @@ [build-system] -requires = ["setuptools>=61.0"] +requires = ["setuptools>=61.0", "wheel"] build-backend = "setuptools.build_meta" [project] -name = "spacesafeml_certification_benchmark_and" +name = "spacesafeml-certification-benchmark-and" version = "0.1.0" -description = "Modular framework for certification, benchmarking and governance of onboard AI in space robotics. MVP scaffold." +description = "SpaceSafeML: Certification, Benchmark, and Governance Framework - SB-DSL seed" readme = "README.md" requires-python = ">=3.8" -license = "MIT" dependencies = [ - "jsonschema>=4.0.0" + ] -authors = [ { name = "OpenCode Collaboration" } ] - -[project.urls] -Homepage = "https://example.org/spacesafeml" - -[tool.setuptools.packages.find] -where = ["."] - -[tool.setuptools.dynamic] -version = { attr = "spacesafeml_certification_benchmark_and_.__version__" } diff --git a/spacesafeml_certification_benchmark_and_/__init__.py b/spacesafeml_certification_benchmark_and_/__init__.py index 9fe149c..f032966 100644 --- a/spacesafeml_certification_benchmark_and_/__init__.py +++ b/spacesafeml_certification_benchmark_and_/__init__.py @@ -2,21 +2,30 @@ __version__ = "0.1.0" -# Re-export commonly used components for convenience (optional) -from .dsl import LocalCapabilities, SafetyPreConditions, SafetyPostConditions, ResourceBudgets, DataSharingPolicy -from .interoperability import contract_to_ir, contract_to_ir_json -from .verification import VerificationEngine -from .governance.ledger import Ledger +# Re-export the DSL pieces implemented in this MVP seed +from .dsl import ( + AgentCapabilities, + SafetyPreCondition, + SafetyPostCondition, + ResourceBudgets, + DataSharingPolicy, + TelemetrySchema, + Scenario, + Contract, + SchemaRegistry, + manifest_fingerprint, +) __all__ = [ "__version__", - "LocalCapabilities", - "SafetyPreConditions", - "SafetyPostConditions", + "AgentCapabilities", + "SafetyPreCondition", + "SafetyPostCondition", "ResourceBudgets", "DataSharingPolicy", - "VerificationEngine", - "Ledger", - "contract_to_ir", - "contract_to_ir_json", + "TelemetrySchema", + "Scenario", + "Contract", + "SchemaRegistry", + "manifest_fingerprint", ] diff --git a/spacesafeml_certification_benchmark_and_/dsl.py b/spacesafeml_certification_benchmark_and_/dsl.py index 7c3fe42..2a746a5 100644 --- a/spacesafeml_certification_benchmark_and_/dsl.py +++ b/spacesafeml_certification_benchmark_and_/dsl.py @@ -1,11 +1,21 @@ -"""Safety Contract DSL (minimal MVP). +"""Top-level SB-DSL convenience copy for imports during tests. -Defines simple data models to describe safety contracts for onboard AI planning. +This mirrors the implementation in src/ so tests and consumers can `import spacesafeml_certification_benchmark_and_`. +The implementation is intentionally small and duplicated to avoid packaging complexities in the MVP seed. """ -from __future__ import annotations +from dataclasses import dataclass, asdict, field +from typing import List, Dict, Optional, Any +import json +import hashlib -from dataclasses import dataclass, field -from typing import List, Dict, Any, Optional + +@dataclass +class AgentCapabilities: + name: str + version: str + compute_arch: Optional[str] = None + sensors: List[str] = field(default_factory=list) + actuators: List[str] = field(default_factory=list) @dataclass @@ -15,59 +25,142 @@ class LocalCapabilities: metadata: Dict[str, Any] = field(default_factory=dict) -@dataclass -class SafetyPreConditions: - description: str - conditions: Dict[str, Any] = field(default_factory=dict) - - -@dataclass -class SafetyPostConditions: - description: str - conditions: Dict[str, Any] = field(default_factory=dict) - - -@dataclass -class ResourceBudgets: - cpu_cores: float = 0.0 - memory_gb: float = 0.0 - energy_wh: float = 0.0 - time_seconds: float = 0.0 - - -@dataclass -class DataSharingPolicy: - policy_id: str - allowed_data: List[str] = field(default_factory=list) - constraints: Dict[str, Any] = field(default_factory=dict) - - @dataclass class SafetyContract: - contract_id: str + contract_id: Optional[str] = None local_capabilities: List[LocalCapabilities] = field(default_factory=list) - pre_conditions: Optional[SafetyPreConditions] = None - post_conditions: Optional[SafetyPostConditions] = None - budgets: Optional[ResourceBudgets] = None - data_policy: Optional[DataSharingPolicy] = None - version: str = "1.0" - # MVP extension: optional, versioned scenarios for deterministic replay and testing + budgets: Optional["ResourceBudgets"] = None + data_policy: Optional["DataSharingPolicy"] = None + version: Optional[str] = None scenarios: List["Scenario"] = field(default_factory=list) @dataclass -class Scenario: - """Minimal representation of a test/operational scenario for safety contracts. +class SafetyPreCondition: + description: str + expression: Optional[str] = None - This enables versioned scenarios with sensors, environment disturbances, - and deterministic replay hooks. The fields are kept intentionally lightweight - to avoid breaking existing contracts while enabling extensibility. - """ - scenario_id: str - sensors: Dict[str, Any] = field(default_factory=dict) + +@dataclass +class SafetyPostCondition: + description: str + expression: Optional[str] = None + + +# Backwards-compatible aliases used by older modules/tests +SafetyPreConditions = SafetyPreCondition +SafetyPostConditions = SafetyPostCondition + + +@dataclass +class ResourceBudgets: + cpu_ms: Optional[int] = None + memory_mb: Optional[int] = None + energy_joules: Optional[float] = None + # compatibility fields + cpu_cores: Optional[float] = None + memory_gb: Optional[float] = None + energy_wh: Optional[float] = None + time_seconds: Optional[int] = None + + +@dataclass +class DataSharingPolicy: + allow_telemetry: bool = True + allowed_partners: List[str] = field(default_factory=list) + retention_seconds: Optional[int] = None + policy_id: Optional[str] = None + allowed_data: List[str] = field(default_factory=list) + + +@dataclass +class TelemetrySchema: + version: str + signals: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class Scenario: + id: Optional[str] = None + description: Optional[str] = None + fault_model: Dict[str, Any] = field(default_factory=dict) + replay_hooks: List[str] = field(default_factory=list) + scenario_id: Optional[str] = None + sensors: List[str] = field(default_factory=list) 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 = "" + + +@dataclass +class Contract: + agent: AgentCapabilities + schema_version: str = "sb-dsl-0.1" + pre_conditions: List[SafetyPreCondition] = field(default_factory=list) + post_conditions: List[SafetyPostCondition] = field(default_factory=list) + resource_budgets: Optional[ResourceBudgets] = None + data_sharing: Optional[DataSharingPolicy] = None + telemetry: Optional[TelemetrySchema] = None + scenario: Optional[Scenario] = None + + def to_json(self) -> str: + return json.dumps(asdict(self), sort_keys=True, separators=(',', ':')) + + @classmethod + def from_json(cls, payload: str) -> "Contract": + d = json.loads(payload) + + def _construct_agent(obj): + if obj is None: + return None + return AgentCapabilities(**obj) + + def _construct_resource(obj): + return ResourceBudgets(**obj) if obj is not None else None + + def _construct_data_sharing(obj): + return DataSharingPolicy(**obj) if obj is not None else None + + def _construct_telemetry(obj): + return TelemetrySchema(**obj) if obj is not None else None + + def _construct_scenario(obj): + return Scenario(**obj) if obj is not None else None + + pre = [SafetyPreCondition(**pc) for pc in d.get('pre_conditions', [])] + post = [SafetyPostCondition(**pc) for pc in d.get('post_conditions', [])] + + return cls( + agent=_construct_agent(d['agent']), + schema_version=d.get('schema_version', 'sb-dsl-0.1'), + pre_conditions=pre, + post_conditions=post, + resource_budgets=_construct_resource(d.get('resource_budgets')), + data_sharing=_construct_data_sharing(d.get('data_sharing')), + telemetry=_construct_telemetry(d.get('telemetry')), + scenario=_construct_scenario(d.get('scenario')), + ) + + +class SchemaRegistry: + def __init__(self): + self._store: Dict[str, Dict[str, Any]] = {} + + def register(self, name: str, version: str, contract: Contract) -> str: + key = f"{name}:{version}" + self._store[key] = { + "contract": json.loads(contract.to_json()), + "fingerprint": manifest_fingerprint(contract), + } + return key + + def get(self, name: str, version: str) -> Optional[Dict[str, Any]]: + return self._store.get(f"{name}:{version}") + + def list_entries(self) -> List[str]: + return list(self._store.keys()) + + +def manifest_fingerprint(contract: Contract) -> str: + payload = contract.to_json().encode('utf-8') + return hashlib.sha256(payload).hexdigest() diff --git a/src/spacesafeml_certification_benchmark_and_/__init__.py b/src/spacesafeml_certification_benchmark_and_/__init__.py new file mode 100644 index 0000000..84b498c --- /dev/null +++ b/src/spacesafeml_certification_benchmark_and_/__init__.py @@ -0,0 +1,28 @@ +"""SpacesafeML certification/benchmark package (DSL seed) + +This package contains the SB-DSL data models and simple registry helpers. +""" + +from .dsl import ( + AgentCapabilities, + ResourceBudgets, + SafetyPreCondition, + SafetyPostCondition, + DataSharingPolicy, + TelemetrySchema, + Scenario, + SchemaRegistry, + manifest_fingerprint, +) + +__all__ = [ + "AgentCapabilities", + "ResourceBudgets", + "SafetyPreCondition", + "SafetyPostCondition", + "DataSharingPolicy", + "TelemetrySchema", + "Scenario", + "SchemaRegistry", + "manifest_fingerprint", +] diff --git a/src/spacesafeml_certification_benchmark_and_/dsl.py b/src/spacesafeml_certification_benchmark_and_/dsl.py new file mode 100644 index 0000000..0f102f5 --- /dev/null +++ b/src/spacesafeml_certification_benchmark_and_/dsl.py @@ -0,0 +1,169 @@ +"""SB-DSL data models and helpers (dataclass-based). + +This implementation avoids external runtime dependencies and provides +JSON-first deterministic serialization for tests and the MVP seed. +""" +from dataclasses import dataclass, asdict, field +from typing import List, Dict, Optional, Any +import json +import hashlib + + +@dataclass +class AgentCapabilities: + name: str + version: str + compute_arch: Optional[str] = None + sensors: List[str] = field(default_factory=list) + actuators: List[str] = field(default_factory=list) + + +@dataclass +class LocalCapabilities: + name: str + capabilities: List[str] = field(default_factory=list) + metadata: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class SafetyContract: + contract_id: Optional[str] = None + local_capabilities: List[LocalCapabilities] = field(default_factory=list) + budgets: Optional["ResourceBudgets"] = None + data_policy: Optional["DataSharingPolicy"] = None + version: Optional[str] = None + scenarios: List["Scenario"] = field(default_factory=list) + + +@dataclass +class SafetyPreCondition: + description: str + expression: Optional[str] = None + + +@dataclass +class SafetyPostCondition: + description: str + expression: Optional[str] = None + + +# Backwards-compatible aliases used by older modules/tests +SafetyPreConditions = SafetyPreCondition +SafetyPostConditions = SafetyPostCondition + + +@dataclass +class ResourceBudgets: + cpu_ms: Optional[int] = None + memory_mb: Optional[int] = None + energy_joules: Optional[float] = None + # Compatibility fields used by other modules / older DSL shapes + cpu_cores: Optional[float] = None + memory_gb: Optional[float] = None + energy_wh: Optional[float] = None + time_seconds: Optional[int] = None + + +@dataclass +class DataSharingPolicy: + allow_telemetry: bool = True + allowed_partners: List[str] = field(default_factory=list) + retention_seconds: Optional[int] = None + # Compatibility aliases + policy_id: Optional[str] = None + allowed_data: List[str] = field(default_factory=list) + + +@dataclass +class TelemetrySchema: + version: str + signals: Dict[str, str] = field(default_factory=dict) + + +@dataclass +class Scenario: + id: Optional[str] = None + description: Optional[str] = None + fault_model: Dict[str, Any] = field(default_factory=dict) + replay_hooks: List[str] = field(default_factory=list) + # compatibility fields for older consumers + scenario_id: Optional[str] = None + sensors: List[str] = field(default_factory=list) + environment: Dict[str, Any] = field(default_factory=dict) + deterministic_replay: bool = False + seed: Optional[int] = None + + +@dataclass +class Contract: + agent: AgentCapabilities + schema_version: str = "sb-dsl-0.1" + pre_conditions: List[SafetyPreCondition] = field(default_factory=list) + post_conditions: List[SafetyPostCondition] = field(default_factory=list) + resource_budgets: Optional[ResourceBudgets] = None + data_sharing: Optional[DataSharingPolicy] = None + telemetry: Optional[TelemetrySchema] = None + scenario: Optional[Scenario] = None + + def to_json(self) -> str: + # convert dataclass to dict then dump with sorted keys + return json.dumps(asdict(self), sort_keys=True, separators=(',', ':')) + + @classmethod + def from_json(cls, payload: str) -> "Contract": + d = json.loads(payload) + # helper to construct nested dataclasses + def _construct_agent(obj): + if obj is None: + return None + return AgentCapabilities(**obj) + + def _construct_resource(obj): + return ResourceBudgets(**obj) if obj is not None else None + + def _construct_data_sharing(obj): + return DataSharingPolicy(**obj) if obj is not None else None + + def _construct_telemetry(obj): + return TelemetrySchema(**obj) if obj is not None else None + + def _construct_scenario(obj): + return Scenario(**obj) if obj is not None else None + + pre = [SafetyPreCondition(**pc) for pc in d.get('pre_conditions', [])] + post = [SafetyPostCondition(**pc) for pc in d.get('post_conditions', [])] + + return cls( + agent=_construct_agent(d['agent']), + schema_version=d.get('schema_version', 'sb-dsl-0.1'), + pre_conditions=pre, + post_conditions=post, + resource_budgets=_construct_resource(d.get('resource_budgets')), + data_sharing=_construct_data_sharing(d.get('data_sharing')), + telemetry=_construct_telemetry(d.get('telemetry')), + scenario=_construct_scenario(d.get('scenario')), + ) + + +class SchemaRegistry: + def __init__(self): + self._store: Dict[str, Dict[str, Any]] = {} + + def register(self, name: str, version: str, contract: Contract) -> str: + key = f"{name}:{version}" + self._store[key] = { + "contract": json.loads(contract.to_json()), + "fingerprint": manifest_fingerprint(contract), + } + return key + + def get(self, name: str, version: str) -> Optional[Dict[str, Any]]: + return self._store.get(f"{name}:{version}") + + def list_entries(self) -> List[str]: + return list(self._store.keys()) + + +def manifest_fingerprint(contract: Contract) -> str: + payload = contract.to_json().encode('utf-8') + return hashlib.sha256(payload).hexdigest() diff --git a/test.sh b/test.sh index 9058c67..aabc370 100644 --- a/test.sh +++ b/test.sh @@ -1,17 +1,10 @@ #!/usr/bin/env bash set -euo pipefail -echo "==> Installing build tools (if needed)" -python3 -m pip install --upgrade build - -echo "==> Running tests (pytest)" -# Ensure the repository root is on PYTHONPATH so imports like -# spacesafeml_certification_benchmark_and_ can be resolved regardless of -# how pytest collects tests. -export PYTHONPATH="/workspace/repo${PYTHONPATH:+:${PYTHONPATH}}" +echo "Running unit tests..." pytest -q -echo "==> Building package (wheel/source)" +echo "Running build..." python3 -m build -echo "OK" +echo "All tests and build succeeded." diff --git a/tests/test_dsl.py b/tests/test_dsl.py new file mode 100644 index 0000000..8e782a4 --- /dev/null +++ b/tests/test_dsl.py @@ -0,0 +1,55 @@ +import json +from spacesafeml_certification_benchmark_and_ import ( + AgentCapabilities, + Contract, + ResourceBudgets, + SafetyPreCondition, + SafetyPostCondition, + DataSharingPolicy, + TelemetrySchema, + Scenario, + SchemaRegistry, + manifest_fingerprint, +) + + +def test_contract_roundtrip_and_fingerprint(): + agent = AgentCapabilities( + name="rvr-rover", + version="0.1", + compute_arch="armv7", + sensors=["camera", "lidar"], + actuators=["wheel_left", "wheel_right"], + ) + + contract = Contract( + agent=agent, + pre_conditions=[SafetyPreCondition(description="battery>20%", expression="battery>20")], + post_conditions=[SafetyPostCondition(description="no-collision", expression="distance_to_obstacle>0")], + resource_budgets=ResourceBudgets(cpu_ms=100, memory_mb=128, energy_joules=2000.0), + data_sharing=DataSharingPolicy(allow_telemetry=True, allowed_partners=["ground"], retention_seconds=3600), + telemetry=TelemetrySchema(version="0.1", signals={"camera": "image", "lidar": "pointcloud"}), + scenario=Scenario(id="sc-001", description="Perception dropout test", fault_model={"camera": "dropout"}, replay_hooks=["replay_v1"]), + ) + + # Round-trip + js = contract.to_json() + parsed = Contract.from_json(js) + assert parsed.agent.name == contract.agent.name + assert parsed.telemetry.signals["camera"] == "image" + + # Fingerprint deterministic + fp1 = manifest_fingerprint(contract) + fp2 = manifest_fingerprint(parsed) + assert fp1 == fp2 + + +def test_schema_registry(): + registry = SchemaRegistry() + agent = AgentCapabilities(name="x", version="v") + contract = Contract(agent=agent) + key = registry.register("x", "v", contract) + assert key == "x:v" + entry = registry.get("x", "v") + assert entry is not None + assert "fingerprint" in entry