90 lines
3.0 KiB
Python
90 lines
3.0 KiB
Python
from __future__ import annotations
|
|
|
|
"""CatOpt Bridge: lightweight GoC registry and DSL blocks used by tests.
|
|
|
|
Provides:
|
|
- FoundationModelBlock, PlanningPolicyBlock, LocalProblemBlock (minimal dataclasses)
|
|
- GoCRegistry: register adapters, publish contracts, log deltas with ed25519 signatures,
|
|
and verify signatures. Uses deterministic ed25519 signing for replay-safe tests.
|
|
"""
|
|
|
|
from dataclasses import dataclass, asdict
|
|
from typing import Any, Dict, List
|
|
import time
|
|
|
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
|
|
from cryptography.hazmat.primitives import serialization
|
|
import json
|
|
|
|
|
|
@dataclass
|
|
class FoundationModelBlock:
|
|
id: str
|
|
domain: str
|
|
model_type: str = "foundation"
|
|
|
|
|
|
@dataclass
|
|
class PlanningPolicyBlock:
|
|
safety_budget: float = 1.0
|
|
performance_budget: float = 1.0
|
|
|
|
|
|
@dataclass
|
|
class LocalProblemBlock:
|
|
problem_id: str
|
|
objective: str = ""
|
|
|
|
|
|
class GoCRegistry:
|
|
"""In-memory Graph-of-Contracts registry with simple crypto for signing deltas.
|
|
|
|
This class is intentionally small: it stores adapter metadata, contracts, and a
|
|
simple delta log. Deltas are signed with an ed25519 keypair created at init.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.adapters: Dict[str, Dict[str, Any]] = {}
|
|
self.contracts: Dict[str, Dict[str, Any]] = {}
|
|
self.deltas: List[Dict[str, Any]] = []
|
|
|
|
# create an ed25519 keypair for signing/verifying
|
|
self._priv = Ed25519PrivateKey.generate()
|
|
self._pub = self._priv.public_key()
|
|
|
|
def register_adapter(self, adapter_id: str, supported_domains: List[str]) -> Dict[str, Any]:
|
|
info = {"adapter_id": adapter_id, "supported_domains": supported_domains}
|
|
self.adapters[adapter_id] = info
|
|
return info
|
|
|
|
def get_adapter(self, adapter_id: str) -> Dict[str, Any] | None:
|
|
return self.adapters.get(adapter_id)
|
|
|
|
def publish_contract(self, contract_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
|
|
contract = {"contract_id": contract_id, "payload": payload}
|
|
self.contracts[contract_id] = contract
|
|
return contract
|
|
|
|
def log_delta(self, delta: Dict[str, Any], contract_id: str) -> Dict[str, Any]:
|
|
# build a deterministic payload for signing
|
|
ts = time.time()
|
|
payload = {"delta": delta, "contract_id": contract_id, "timestamp": ts}
|
|
# canonicalize to bytes deterministically
|
|
payload_bytes = json.dumps(payload, sort_keys=True).encode()
|
|
sig = self._priv.sign(payload_bytes)
|
|
entry = {"payload": payload, "signature": sig.hex()}
|
|
self.deltas.append(entry)
|
|
return entry
|
|
|
|
def verify(self, message: bytes, signature: bytes) -> bool:
|
|
try:
|
|
# verify against stored public key
|
|
self._pub.verify(signature, message)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
# convenience helpers for tests
|
|
def last_delta(self) -> Dict[str, Any] | None:
|
|
return self.deltas[-1] if self.deltas else None
|