From 8e06a3af2049da4edc2e699833c4dd7425336f7f Mon Sep 17 00:00:00 2001 From: agent-54de0bcc6a17828b Date: Fri, 24 Apr 2026 20:17:46 +0200 Subject: [PATCH] build(agent): semicolon#54de0b iteration --- =2.7, | 0 AGENTS.md | 22 +- README.md | 46 ++- idea176_goc_synth_automated/__init__.py | 18 +- idea176_goc_synth_automated/dsl.py | 68 +++-- idea176_goc_synth_automated/generator.py | 353 ++++++++++++++++------- idea176_goc_synth_automated/registry.py | 26 +- pyproject.toml | 1 + registry/registry.json | 42 +-- test.sh | 3 + tests/test_basic.py | 68 ++++- 11 files changed, 432 insertions(+), 215 deletions(-) create mode 100644 =2.7, diff --git a/=2.7, b/=2.7, new file mode 100644 index 0000000..e69de29 diff --git a/AGENTS.md b/AGENTS.md index fbb02ab..dae1687 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -2,30 +2,30 @@ Overview - This repository implements an MVP of GoC Synth: an automated DSL-to-Adapter synthesis pipeline that targets a Graph-of-Contracts (GoC) style interoperability layer across domains. Architecture (high level) -- DSL: LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog, PolicyBlock definitions. +- DSL: LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog, PolicyBlock, GraphOfContractsSeed, RegistryEntry. - IR: EnergiBridge-like vendor-agnostic intermediate representation. - GoC Registry: central registry of cross-domain contracts and adapters. -- Synthesis Engine: template-driven skeleton adapters in Python/Rust/C++ with a conformance harness and toy simulator. +- Synthesis Engine: template-driven skeleton adapters in Python/Rust/C++ with a conformance harness and deterministic delta-sync simulator. - Governance: DID-based identities and governance ledger integration. - Delta-sync: deterministic replay pipeline for islanding and cross-domain updates. -Current MVP Focus (Phase 0) -- Skeleton adapters for 2 domains over TLS (template-driven generation). -- Toy local solver and basic delta-sync flow. -- GoC registry scaffolding and a simple conformance harness. +Current implementation focus +- Skeleton adapters for 2 domains over TLS with generated OpenAPI and protobuf specs. +- Deterministic delta-sync simulator and conformance harness. +- GoC registry entry persistence with DID binding and governance anchor metadata. Codebase layout (key parts) - idea176_goc_synth_automated/ (Python package with DSL primitives, generator, and registry) - adapters/ (Generated skeleton adapters; created by the generator) - registry/ (GoC registry placeholder data) -- tests/ (Basic tests for MVP generation) +- tests/ (Basic tests for generation and replay determinism) - test.sh (Test launcher that also builds packaging artefacts) - AGENTS.md (This document) How to work with this repository -- To add more domains, extend the generator to emit additional domain adapters and domain-specific templates. -- Run tests locally with ./test.sh. The script will run pytest and build the Python package to verify packaging metadata and compilation. -- Review the AGENTS.md for contribution guidelines and architectural decisions. +- To add more domains, extend the generator to emit additional domain adapters, OpenAPI specs, and protobuf stubs. +- Run tests locally with `./test.sh`. The script installs the package in editable mode, runs pytest, and builds the Python package. +- Keep generated artifacts deterministic so registry and conformance tests remain stable. Important notes -- This is an MVP and intentionally minimal in runtime behavior. The focus is on correct scaffolding, reproducible code generation, and testability, with a clear path to expansion in subsequent phases. +- This is still a scaffold, but the generated artifacts are now derived from the DSL and registry metadata rather than fixed placeholders. diff --git a/README.md b/README.md index 8e09764..eed2775 100644 --- a/README.md +++ b/README.md @@ -1,23 +1,37 @@ # idea176_goc_synth_automated -Automated DSL-to-Adapter synthesis for Graph-of-Contracts cross-domain optimization (MVP). +Automated DSL-to-Adapter synthesis for Graph-of-Contracts cross-domain optimization. -What this project builds now -- A compact DSL (LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog, PolicyBlock) that feeds into a vendor-agnostic IR and a GoC registry. -- A Template-driven generator that emits skeleton adapters for two domains (Energy and Robotics) in Python with TLS-ready scaffolding. -- A toy local solver and a minimal delta-sync harness to demonstrate cross-domain data exchange and deterministic replay semantics. -- A small GoC registry scaffold and a conformance harness. +## What it does +- Validates a compact DSL for `LocalProblem`, `SharedVariables`, `PlanDelta`, `DualVariables`, `PrivacyBudget`, `AuditLog`, `PolicyBlock`, and `GraphOfContractsSeed`. +- Generates Energy and Robotics adapter skeletons with TLS setup, manifest export, deterministic delta application, OpenAPI specs, and protobuf service stubs. +- Emits a deterministic delta-sync simulator and a conformance harness that checks adapter shape and replay stability. +- Persists a registry entry with DID binding, governance anchor metadata, and artifact paths. -How to run locally -- Prerequisites: Python 3.9+ (no extra system dependencies required for MVP skeleton). -- Install/test primitives via the provided test script. +## Layout +- `idea176_goc_synth_automated/` core Python package +- `adapters/` generated adapter code and specs +- `registry/` persisted GoC registry entry +- `tests/` package tests -Key commands -- Build and test: bash test.sh -- Generate adapters (via Python API): python -c 'from idea176_goc_synth_automated.generator import generate_mvp_adapters; ...' +## Usage +- Install and test: `bash test.sh` +- Generate artifacts from Python: +```python +from idea176_goc_synth_automated import GraphOfContractsSeed, LocalProblem, generate_mvp_adapters -Packaging and publishing notes -- This project is Python-based. It exposes a package named idea176_goc_synth_automated and a minimal build workflow suitable for CI. -- See pyproject.toml for packaging metadata and how to hook into PyPI-style publishing. +seed = GraphOfContractsSeed( + name="demo", + seeds={"LocalProblem": LocalProblem(domain="Energy", objective="minimize_cost")}, +) +generate_mvp_adapters(seed) +``` -This README intentionally keeps high-level concepts and usage patterns; please refer to AGENTS.md for architectural details and contribution guidelines. +## Packaging +- Python package name: `idea176_goc_synth_automated` +- Build metadata lives in `pyproject.toml` +- `README.md` is wired into the package metadata for publishing + +## Notes +- The current codebase focuses on a reproducible foundation for synthesis, registry persistence, and deterministic replay. +- See `AGENTS.md` for architectural constraints and testing expectations. diff --git a/idea176_goc_synth_automated/__init__.py b/idea176_goc_synth_automated/__init__.py index ee73b2e..02e2c55 100644 --- a/idea176_goc_synth_automated/__init__.py +++ b/idea176_goc_synth_automated/__init__.py @@ -1,14 +1,17 @@ """Idea176 GoC Synth - Core package export""" -from .dsl import LocalProblem # Re-export for convenience in tests/examples -from .dsl import SharedVariables -from .dsl import PlanDelta -from .dsl import DualVariables -from .dsl import PrivacyBudget from .dsl import AuditLog -from .dsl import PolicyBlock +from .dsl import DualVariables from .dsl import GraphOfContractsSeed +from .dsl import LocalProblem +from .dsl import PlanDelta +from .dsl import PolicyBlock +from .dsl import PrivacyBudget +from .dsl import RegistryEntry +from .dsl import SharedVariables from .generator import generate_mvp_adapters +from .registry import load_registry +from .registry import save_registry __all__ = [ "LocalProblem", @@ -19,5 +22,8 @@ __all__ = [ "AuditLog", "PolicyBlock", "GraphOfContractsSeed", + "RegistryEntry", "generate_mvp_adapters", + "load_registry", + "save_registry", ] diff --git a/idea176_goc_synth_automated/dsl.py b/idea176_goc_synth_automated/dsl.py index e8ea9e6..29ca675 100644 --- a/idea176_goc_synth_automated/dsl.py +++ b/idea176_goc_synth_automated/dsl.py @@ -1,46 +1,62 @@ from __future__ import annotations -from dataclasses import dataclass, field -from typing import Dict, Any, List, Optional + +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field -@dataclass -class LocalProblem: +class DslModel(BaseModel): + model_config = ConfigDict(extra="forbid", validate_assignment=True) + + def to_dict(self) -> Dict[str, Any]: + return self.model_dump(mode="json", exclude_none=True) + + +class LocalProblem(DslModel): domain: str objective: str - constraints: Dict[str, Any] = field(default_factory=dict) + constraints: Dict[str, Any] = Field(default_factory=dict) + metadata: Dict[str, Any] = Field(default_factory=dict) -@dataclass -class SharedVariables: - variables: Dict[str, Any] = field(default_factory=dict) +class SharedVariables(DslModel): + variables: Dict[str, Any] = Field(default_factory=dict) -@dataclass -class PlanDelta: - changes: Dict[str, Any] = field(default_factory=dict) +class PlanDelta(DslModel): + changes: Dict[str, Any] = Field(default_factory=dict) -@dataclass -class DualVariables: - alphas: Dict[str, float] = field(default_factory=dict) +class DualVariables(DslModel): + alphas: Dict[str, float] = Field(default_factory=dict) -@dataclass -class PrivacyBudget: - budget: Dict[str, Any] = field(default_factory=dict) +class PrivacyBudget(DslModel): + budget: Dict[str, Any] = Field(default_factory=dict) -@dataclass -class AuditLog: - entries: List[str] = field(default_factory=list) +class AuditLog(DslModel): + entries: List[str] = Field(default_factory=list) -@dataclass -class PolicyBlock: - policy: Dict[str, Any] = field(default_factory=dict) +class PolicyBlock(DslModel): + policy: Dict[str, Any] = Field(default_factory=dict) -@dataclass -class GraphOfContractsSeed: +class GraphOfContractsSeed(DslModel): name: str - seeds: Dict[str, Any] = field(default_factory=dict) + seeds: Dict[str, Any] = Field(default_factory=dict) + domains: List[str] = Field(default_factory=lambda: ["Energy", "Robotics"]) + did: Optional[str] = None + governance_anchor: Optional[str] = None + transport: str = "tls" + + +class RegistryEntry(DslModel): + name: str + domains: List[str] + artifacts: Dict[str, str] + seed_summary: Dict[str, Any] = Field(default_factory=dict) + did_binding: Dict[str, str] = Field(default_factory=dict) + governance_anchor: Dict[str, Any] = Field(default_factory=dict) + transport: str = "tls" diff --git a/idea176_goc_synth_automated/generator.py b/idea176_goc_synth_automated/generator.py index 0cc33a0..cdbd59f 100644 --- a/idea176_goc_synth_automated/generator.py +++ b/idea176_goc_synth_automated/generator.py @@ -1,19 +1,24 @@ -"""Generation entrypoint for MVP adapters. +"""Generation entrypoint for GoC adapter scaffolding. This module emits two domain adapter skeletons (Energy and Robotics) in Python, -along with minimal harnesses (solver, conformance, simulator) and a simple -GoC registry entry. The goal is a minimal, testable scaffold that demonstrates -the pipeline without requiring a full runtime environment. +along with a deterministic delta-sync simulator, a conformance harness, +adapter manifests, and a registry entry. """ + from __future__ import annotations -import os +import hashlib import json +import os +from datetime import datetime, timezone from pathlib import Path -from typing import Dict, Any -from dataclasses import is_dataclass, asdict +from typing import Any, Dict -from .dsl import GraphOfContractsSeed, LocalProblem, SharedVariables, PlanDelta, DualVariables, PrivacyBudget, AuditLog, PolicyBlock +from pydantic import BaseModel + +from .dsl import GraphOfContractsSeed +from .registry import registry_path +from .registry import save_registry def _ensure_dir(path: str) -> None: @@ -25,28 +30,28 @@ def _write_file(path: str, content: str) -> None: f.write(content) -def _energy_adapter_code() -> str: - return '''# Energy Adapter Skeleton (TLS-ready) +def _adapter_code(domain: str, default_objective: str, port: int) -> str: + return '''# {domain} Adapter Skeleton (TLS-ready) +import hashlib +import json import ssl -from typing import Optional, Dict, Any - -try: - # Lightweight local-problem placeholder to avoid external imports - from dataclasses import dataclass - @dataclass - class LocalProblem: - domain: str - objective: str - constraints: dict = None -except Exception: - class LocalProblem: # type: ignore - pass +from dataclasses import dataclass +from typing import Any, Dict, Optional -class EnergyAdapter: +@dataclass +class LocalProblem: + domain: str + objective: str + constraints: Optional[dict] = None + metadata: Optional[dict] = None + + +class {domain}Adapter: def __init__(self, config: Optional[Dict[str, Any]] = None): - self.config = config or {} + self.config = config or {{}} self._tls_context = None + self._applied_deltas = [] def _build_tls_context(self, certfile: str = None, keyfile: str = None) -> ssl.SSLContext: ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) @@ -58,84 +63,182 @@ class EnergyAdapter: def configure_tls(self, certfile: str, keyfile: str) -> ssl.SSLContext: return self._build_tls_context(certfile, keyfile) - def start(self, host: str = "127.0.0.1", port: int = 50051): - # Minimal TLS-ready skeleton; does not bind in MVP tests + def start(self, host: str = "127.0.0.1", port: int = {port}): if self._tls_context is None: self._build_tls_context() - return {"status": "initialized", "host": host, "port": port} + return {{"status": "initialized", "domain": "{domain}", "host": host, "port": port}} + + def export_manifest(self) -> Dict[str, Any]: + return {{ + "domain": "{domain}", + "transport": "tls", + "capabilities": ["solve_local_problem", "apply_delta", "export_manifest"], + "default_objective": "{default_objective}", + "port": {port}, + }} + + def apply_delta(self, delta: Dict[str, Any]) -> Dict[str, Any]: + canonical = json.dumps(delta, sort_keys=True, separators=(",", ":")) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + self._applied_deltas.append({{"delta": delta, "digest": digest}}) + return {{"accepted": True, "digest": digest, "delta": delta}} + def solve_local_problem(lp: LocalProblem) -> Dict[str, Any]: - return { - "domain": "Energy", - "lp_objective": getattr(lp, "objective", "minimize_cost"), - "solution": {k: f"{v}-solved" for k, v in getattr(lp, 'constraints', {}).items()}, - } -''' - - -def _robotics_adapter_code() -> str: - return '''# Robotics Adapter Skeleton (TLS-ready) -import ssl -from typing import Optional, Dict, Any - -try: - from dataclasses import dataclass - @dataclass - class LocalProblem: - domain: str - objective: str - constraints: dict = None -except Exception: - class LocalProblem: # type: ignore - pass - - -class RoboticsAdapter: - def __init__(self, config: Optional[Dict[str, Any]] = None): - self.config = config or {} - self._tls_context = None - - def _build_tls_context(self, certfile: str = None, keyfile: str = None) -> ssl.SSLContext: - ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - if certfile and keyfile: - ctx.load_cert_chain(certfile=certfile, keyfile=keyfile) - self._tls_context = ctx - return ctx - - def configure_tls(self, certfile: str, keyfile: str) -> ssl.SSLContext: - return self._build_tls_context(certfile, keyfile) - - def start(self, host: str = "127.0.0.1", port: int = 50052): - if self._tls_context is None: - self._build_tls_context() - return {"status": "initialized", "host": host, "port": port} - -def solve_local_problem(lp: LocalProblem) -> Dict[str, Any]: - return { - "domain": "Robotics", - "lp_objective": getattr(lp, "objective", "maximize_performance"), - "solution": {k: f"{v}-solved" for k, v in getattr(lp, 'constraints', {}).items()}, - } -''' + constraints = lp.constraints or {{}} + metadata = lp.metadata or {{}} + return {{ + "domain": "{domain}", + "lp_objective": getattr(lp, "objective", "{default_objective}"), + "metadata": metadata, + "solution": {{k: f"{{v}}-solved" for k, v in constraints.items()}}, + }} +'''.format(domain=domain, default_objective=default_objective, port=port) def _conformance_harness_code() -> str: - return '''# Minimal conformance harness + return '''# Deterministic conformance harness +import json + + def verify_adapter_shape(adapter): - # Placeholder: in real system, check required methods exist - required = ["start", "configure_tls"] + required = ["start", "configure_tls", "export_manifest"] return all(hasattr(adapter, r) for r in required) + + +def verify_manifest(manifest): + required = {"domain", "transport", "capabilities", "default_objective", "port"} + return required.issubset(set(manifest.keys())) and manifest["transport"] == "tls" + + +def verify_deterministic_solver(solver, sample_problem): + first = solver(sample_problem) + second = solver(sample_problem) + return first == second + + +def render_report(adapter, sample_problem): + manifest = adapter.export_manifest() + result = adapter.apply_delta({"sample_problem": sample_problem}) + return json.dumps({"manifest": manifest, "delta_result": result}, sort_keys=True) +''' + + +def _simulator_code() -> str: + return '''# Deterministic delta-sync simulator +import hashlib +import json +from dataclasses import dataclass +from typing import Any, Dict, List + + +@dataclass +class DeltaSyncEvent: + seq: int + source: str + target: str + delta: Dict[str, Any] + checksum: str + + +class DeltaSyncSimulator: + def __init__(self, seed_payload: Dict[str, Any]): + self.seed_payload = seed_payload + self._canonical = json.dumps(seed_payload, sort_keys=True, separators=(",", ":")) + + def run(self, rounds: int = 3) -> List[Dict[str, Any]]: + events: List[Dict[str, Any]] = [] + for seq in range(1, rounds + 1): + digest = hashlib.sha256(f"{self._canonical}:{seq}".encode("utf-8")).hexdigest() + delta = { + "round": seq, + "energy_load": int(digest[:4], 16) % 250, + "robotics_load": int(digest[4:8], 16) % 250, + } + events.append({ + "seq": seq, + "source": "Energy", + "target": "Robotics", + "delta": delta, + "checksum": digest, + "replayed": True, + }) + return events + + def replay(self, events: List[Dict[str, Any]]) -> Dict[str, Any]: + expected = self.run(rounds=len(events)) + return { + "accepted": expected == events, + "event_count": len(events), + "checksum": hashlib.sha256(json.dumps(events, sort_keys=True).encode("utf-8")).hexdigest(), + } +''' + + +def _solver_code() -> str: + return '''# Shared solver hook for generated adapters +import hashlib +import json +from typing import Any, Dict + + +def solve_problem(problem: Dict[str, Any]) -> Dict[str, Any]: + canonical = json.dumps(problem, sort_keys=True, default=str, separators=(",", ":")) + digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest() + return { + "status": "solved", + "digest": digest, + "score": int(digest[:8], 16) % 1000, + } +''' + + +def _openapi_spec(domain: str) -> Dict[str, Any]: + return { + "openapi": "3.1.0", + "info": { + "title": f"{domain} Adapter API", + "version": "1.0.0", + }, + "paths": { + "/solve": { + "post": { + "summary": "Solve a local domain problem", + "responses": {"200": {"description": "Solver result"}}, + } + }, + "/delta-sync": { + "post": { + "summary": "Apply a deterministic delta", + "responses": {"200": {"description": "Delta accepted"}}, + } + }, + }, + } + + +def _proto_spec(domain: str) -> str: + return f'''syntax = "proto3"; + +package goc.{domain.lower()}; + +message LocalProblem {{ + string domain = 1; + string objective = 2; + map constraints = 3; +}} + +message DeltaUpdate {{ + uint64 sequence = 1; + string checksum = 2; + string payload_json = 3; +}} ''' def _serialize(obj: Any) -> Any: - """Recursively convert dataclass instances to serializable dicts. - - This allows storing complex DSL seed structures (which may contain - dataclass instances) inside JSON without requiring custom encoders. - """ - if is_dataclass(obj): - return asdict(obj) + if isinstance(obj, BaseModel): + return obj.model_dump(mode="json", exclude_none=True) if isinstance(obj, dict): return {k: _serialize(v) for k, v in obj.items()} if isinstance(obj, list): @@ -144,47 +247,70 @@ def _serialize(obj: Any) -> Any: def _registry_entry(seed: GraphOfContractsSeed) -> Dict[str, Any]: + stamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + did = seed.did or f"did:example:{hashlib.sha256(seed.name.encode('utf-8')).hexdigest()[:16]}" return { "name": seed.name, - "domains": ["Energy", "Robotics"], - "path": { + "schema_version": "1.0.0", + "domains": seed.domains, + "transport": seed.transport, + "artifacts": { "energy_adapter": "adapters/energy_adapter.py", "robotics_adapter": "adapters/robotics_adapter.py", + "energy_openapi": "adapters/energy_openapi.json", + "robotics_openapi": "adapters/robotics_openapi.json", + "energy_proto": "adapters/energy_adapter.proto", + "robotics_proto": "adapters/robotics_adapter.proto", + "manifest": "adapters/adapter_manifest.json", + "simulator": "adapters/sim.py", + "conformance": "adapters/conformance.py", }, "seed_summary": _serialize(seed.seeds) if seed.seeds else {}, + "did_binding": {"controller": did}, + "governance_anchor": { + "ledger_anchor": seed.governance_anchor or f"anchor:{hashlib.sha256((seed.name + stamp).encode('utf-8')).hexdigest()[:24]}", + "generated_at": stamp, + }, } def generate_mvp_adapters(seed: GraphOfContractsSeed, out_dir: str = "adapters") -> Dict[str, Any]: - """Generate MVP skeleton adapters for two domains. - - - Creates two Python adapter modules: energy_adapter.py and robotics_adapter.py - - Includes helper modules: solver.py, conformance.py, sim.py - - Creates a simple registry entry in registry/registry.json - Returns a mapping of generated file paths. - """ + """Generate adapters, harnesses, and registry artifacts.""" _ensure_dir(out_dir) - # Energy adapter + energy_path = os.path.join(out_dir, "energy_adapter.py") robotics_path = os.path.join(out_dir, "robotics_adapter.py") - _write_file(energy_path, _energy_adapter_code()) - _write_file(robotics_path, _robotics_adapter_code()) + _write_file(energy_path, _adapter_code("Energy", "minimize_cost", 50051)) + _write_file(robotics_path, _adapter_code("Robotics", "maximize_performance", 50052)) - # Conformance & solver skeletons conformance_path = os.path.join(out_dir, "conformance.py") solver_path = os.path.join(out_dir, "solver.py") sim_path = os.path.join(out_dir, "sim.py") _write_file(conformance_path, _conformance_harness_code()) - _write_file(solver_path, """# Placeholder solver module (could be extended)\n""") - _write_file(sim_path, """# Placeholder simulator module (Two-Domain demo)\n""") + _write_file(solver_path, _solver_code()) + _write_file(sim_path, _simulator_code()) + + energy_openapi_path = os.path.join(out_dir, "energy_openapi.json") + robotics_openapi_path = os.path.join(out_dir, "robotics_openapi.json") + energy_proto_path = os.path.join(out_dir, "energy_adapter.proto") + robotics_proto_path = os.path.join(out_dir, "robotics_adapter.proto") + adapter_manifest_path = os.path.join(out_dir, "adapter_manifest.json") + _write_file(energy_openapi_path, json.dumps(_openapi_spec("Energy"), indent=2, sort_keys=True)) + _write_file(robotics_openapi_path, json.dumps(_openapi_spec("Robotics"), indent=2, sort_keys=True)) + _write_file(energy_proto_path, _proto_spec("Energy")) + _write_file(robotics_proto_path, _proto_spec("Robotics")) - # Registry file - registry_dir = "registry" - _ensure_dir(registry_dir) - reg_path = os.path.join(registry_dir, "registry.json") reg_entry = _registry_entry(seed) - with open(reg_path, "w", encoding="utf-8") as f: - json.dump(reg_entry, f, indent=2) + adapter_manifest = { + "name": seed.name, + "registry_entry": reg_entry, + "capabilities": ["solver", "delta-sync", "conformance", "tls"], + "domains": seed.domains, + } + _write_file(adapter_manifest_path, json.dumps(adapter_manifest, indent=2, sort_keys=True)) + + save_registry(reg_entry) + reg_path = str(registry_path()) return { "energy_adapter": energy_path, @@ -192,6 +318,11 @@ def generate_mvp_adapters(seed: GraphOfContractsSeed, out_dir: str = "adapters") "conformance": conformance_path, "solver": solver_path, "simulation": sim_path, + "energy_openapi": energy_openapi_path, + "robotics_openapi": robotics_openapi_path, + "energy_proto": energy_proto_path, + "robotics_proto": robotics_proto_path, + "adapter_manifest": adapter_manifest_path, "registry": reg_path, "registry_entry": reg_entry, } diff --git a/idea176_goc_synth_automated/registry.py b/idea176_goc_synth_automated/registry.py index 8fc178a..512dffe 100644 --- a/idea176_goc_synth_automated/registry.py +++ b/idea176_goc_synth_automated/registry.py @@ -1,20 +1,28 @@ -"""Simple on-disk registry interface for MVP GoC entries.""" +"""Simple on-disk registry interface for GoC entries.""" + import json +import os from pathlib import Path -from typing import Dict, Any +from typing import Any, Dict, Union + +from .dsl import RegistryEntry -REG_PATH = Path("registry/registry.json") +def registry_path() -> Path: + return Path(os.environ.get("GOC_REGISTRY_PATH", "registry/registry.json")) def load_registry() -> Dict[str, Any]: - if not REG_PATH.exists(): + path = registry_path() + if not path.exists(): return {} - with REG_PATH.open("r", encoding="utf-8") as f: + with path.open("r", encoding="utf-8") as f: return json.load(f) -def save_registry(entry: Dict[str, Any]) -> None: - REG_PATH.parent.mkdir(parents=True, exist_ok=True) - with REG_PATH.open("w", encoding="utf-8") as f: - json.dump(entry, f, indent=2) +def save_registry(entry: Union[Dict[str, Any], RegistryEntry]) -> None: + path = registry_path() + path.parent.mkdir(parents=True, exist_ok=True) + payload = entry.to_dict() if isinstance(entry, RegistryEntry) else entry + with path.open("w", encoding="utf-8") as f: + json.dump(payload, f, indent=2, sort_keys=True) diff --git a/pyproject.toml b/pyproject.toml index 4453943..604eea6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,3 +8,4 @@ version = "0.1.0" description = "MVP: Automated DSL-to-Adapter synthesis for Graph-of-Contracts cross-domain optimization." readme = "README.md" requires-python = ">=3.9" +dependencies = ["pydantic>=2.7,<3"] diff --git a/registry/registry.json b/registry/registry.json index a39746b..7205882 100644 --- a/registry/registry.json +++ b/registry/registry.json @@ -1,26 +1,28 @@ { - "name": "test_goc", + "artifacts": { + "conformance": "adapters/conformance.py", + "energy_adapter": "adapters/energy_adapter.py", + "energy_openapi": "adapters/energy_openapi.json", + "energy_proto": "adapters/energy_adapter.proto", + "manifest": "adapters/adapter_manifest.json", + "robotics_adapter": "adapters/robotics_adapter.py", + "robotics_openapi": "adapters/robotics_openapi.json", + "robotics_proto": "adapters/robotics_adapter.proto", + "simulator": "adapters/sim.py" + }, + "did_binding": { + "controller": "did:example:0000000000000000" + }, "domains": [ "Energy", "Robotics" ], - "path": { - "energy_adapter": "adapters/energy_adapter.py", - "robotics_adapter": "adapters/robotics_adapter.py" + "governance_anchor": { + "generated_at": "2026-04-24T00:00:00Z", + "ledger_anchor": "anchor:000000000000000000000000" }, - "seed_summary": { - "LocalProblem": { - "domain": "Energy", - "objective": "minimize_cost", - "constraints": { - "power": 100 - } - }, - "SharedVariables": {}, - "PlanDelta": {}, - "DualVariables": {}, - "PrivacyBudget": {}, - "AuditLog": [], - "PolicyBlock": {} - } -} \ No newline at end of file + "name": "default-goc", + "schema_version": "1.0.0", + "seed_summary": {}, + "transport": "tls" +} diff --git a/test.sh b/test.sh index e785fc0..58006a2 100644 --- a/test.sh +++ b/test.sh @@ -6,6 +6,9 @@ set -euo pipefail # 'idea176_goc_synth_automated' during CI/test runs without installing it. export PYTHONPATH="/workspace/repo:${PYTHONPATH:-}" +# Install the package and its runtime dependency. +python3 -m pip install -e . + # Run unit tests and packaging checks pytest -q diff --git a/tests/test_basic.py b/tests/test_basic.py index 5cb94e6..895fd80 100644 --- a/tests/test_basic.py +++ b/tests/test_basic.py @@ -1,24 +1,60 @@ +import importlib.util import os -from idea176_goc_synth_automated.dsl import LocalProblem, SharedVariables, GraphOfContractsSeed +from idea176_goc_synth_automated.dsl import GraphOfContractsSeed +from idea176_goc_synth_automated.dsl import LocalProblem +from idea176_goc_synth_automated.dsl import SharedVariables from idea176_goc_synth_automated.generator import generate_mvp_adapters +from idea176_goc_synth_automated.registry import load_registry -def test_generate_mvp_adapters_creates_skeletons(tmp_path): - # Simple DSL seed for two-domain MVP +def _load_module(path, name): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None + module = importlib.util.module_from_spec(spec) + assert spec.loader is not None + spec.loader.exec_module(module) + return module + + +def test_generate_mvp_adapters_creates_skeletons(tmp_path, monkeypatch): lp = LocalProblem(domain="Energy", objective="minimize_cost", constraints={"power": 100}) - seed = GraphOfContractsSeed(name="test_goc", seeds={ - "LocalProblem": lp, - "SharedVariables": SharedVariables().variables, - "PlanDelta": {}, - "DualVariables": {}, - "PrivacyBudget": {}, - "AuditLog": [], - "PolicyBlock": {} - }) + seed = GraphOfContractsSeed( + name="test_goc", + seeds={ + "LocalProblem": lp, + "SharedVariables": SharedVariables().variables, + "PlanDelta": {}, + "DualVariables": {}, + "PrivacyBudget": {}, + "AuditLog": [], + "PolicyBlock": {}, + }, + ) - out = tmp_path - result = generate_mvp_adapters(seed, out_dir=str(out)) + monkeypatch.setenv("GOC_REGISTRY_PATH", str(tmp_path / "registry.json")) + result = generate_mvp_adapters(seed, out_dir=str(tmp_path)) - assert os.path.exists(os.path.join(out, "energy_adapter.py")) - assert os.path.exists(os.path.join(out, "robotics_adapter.py")) + assert os.path.exists(os.path.join(tmp_path, "energy_adapter.py")) + assert os.path.exists(os.path.join(tmp_path, "robotics_adapter.py")) + assert os.path.exists(os.path.join(tmp_path, "energy_openapi.json")) + assert os.path.exists(os.path.join(tmp_path, "robotics_adapter.proto")) + assert os.path.exists(os.path.join(tmp_path, "adapter_manifest.json")) + + registry = load_registry() + assert registry["name"] == "test_goc" + assert registry["did_binding"]["controller"].startswith("did:example:") + assert result["registry_entry"]["transport"] == "tls" + + +def test_generated_simulator_is_deterministic(tmp_path, monkeypatch): + seed = GraphOfContractsSeed(name="sim-test", seeds={"LocalProblem": {"domain": "Energy"}}) + monkeypatch.setenv("GOC_REGISTRY_PATH", str(tmp_path / "registry.json")) + result = generate_mvp_adapters(seed, out_dir=str(tmp_path)) + + module = _load_module(result["simulation"], "generated_sim") + sim = module.DeltaSyncSimulator({"LocalProblem": {"domain": "Energy"}}) + first = sim.run(rounds=2) + second = sim.run(rounds=2) + assert first == second + assert sim.replay(first)["accepted"] is True