329 lines
11 KiB
Python
329 lines
11 KiB
Python
"""Generation entrypoint for GoC adapter scaffolding.
|
|
|
|
This module emits two domain adapter skeletons (Energy and Robotics) in Python,
|
|
along with a deterministic delta-sync simulator, a conformance harness,
|
|
adapter manifests, and a registry entry.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from .dsl import GraphOfContractsSeed
|
|
from .registry import registry_path
|
|
from .registry import save_registry
|
|
|
|
|
|
def _ensure_dir(path: str) -> None:
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def _write_file(path: str, content: str) -> None:
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write(content)
|
|
|
|
|
|
def _adapter_code(domain: str, default_objective: str, port: int) -> str:
|
|
return '''# {domain} Adapter Skeleton (TLS-ready)
|
|
import hashlib
|
|
import json
|
|
import ssl
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, Optional
|
|
|
|
|
|
@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._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)
|
|
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 = {port}):
|
|
if self._tls_context is None:
|
|
self._build_tls_context()
|
|
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]:
|
|
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 '''# Deterministic conformance harness
|
|
import json
|
|
|
|
|
|
def verify_adapter_shape(adapter):
|
|
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<string, string> constraints = 3;
|
|
}}
|
|
|
|
message DeltaUpdate {{
|
|
uint64 sequence = 1;
|
|
string checksum = 2;
|
|
string payload_json = 3;
|
|
}}
|
|
'''
|
|
|
|
|
|
def _serialize(obj: Any) -> Any:
|
|
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):
|
|
return [_serialize(v) for v in obj]
|
|
return obj
|
|
|
|
|
|
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,
|
|
"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 adapters, harnesses, and registry artifacts."""
|
|
_ensure_dir(out_dir)
|
|
|
|
energy_path = os.path.join(out_dir, "energy_adapter.py")
|
|
robotics_path = os.path.join(out_dir, "robotics_adapter.py")
|
|
_write_file(energy_path, _adapter_code("Energy", "minimize_cost", 50051))
|
|
_write_file(robotics_path, _adapter_code("Robotics", "maximize_performance", 50052))
|
|
|
|
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, _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"))
|
|
|
|
reg_entry = _registry_entry(seed)
|
|
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,
|
|
"robotics_adapter": robotics_path,
|
|
"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,
|
|
}
|