idea174-catopt-swarm/catopt_swarm/registry.py

65 lines
2.0 KiB
Python

from __future__ import annotations
from typing import Any
import networkx as nx
from pydantic import BaseModel, Field
class ContractSpec(BaseModel):
contract_id: str
adapter_name: str
domain: str
version: str
invariants: tuple[str, ...] = ()
preconditions: tuple[str, ...] = ()
postconditions: tuple[str, ...] = ()
metadata: dict[str, Any] = Field(default_factory=dict)
class AdapterConformanceReport(BaseModel):
adapter_name: str
contract_id: str
passed: bool
issues: list[str] = Field(default_factory=list)
class ContractRegistry:
def __init__(self) -> None:
self._graph = nx.DiGraph()
def register(self, spec: ContractSpec) -> None:
self._graph.add_node(spec.contract_id, spec=spec)
def link(self, parent_contract_id: str, child_contract_id: str, relation: str = "extends") -> None:
self._graph.add_edge(parent_contract_id, child_contract_id, relation=relation)
def get(self, contract_id: str) -> ContractSpec:
return self._graph.nodes[contract_id]["spec"]
def has_contract(self, contract_id: str) -> bool:
return contract_id in self._graph
def summary(self) -> dict[str, Any]:
return {
"contracts": list(self._graph.nodes),
"links": [
{"from": source, "to": target, **data}
for source, target, data in self._graph.edges(data=True)
],
}
def check_adapter_conformance(adapter: Any, spec: ContractSpec) -> AdapterConformanceReport:
issues: list[str] = []
if getattr(adapter, "name", None) != spec.adapter_name:
issues.append(f"adapter name mismatch: expected {spec.adapter_name}")
if getattr(adapter, "domain", None) != spec.domain:
issues.append(f"domain mismatch: expected {spec.domain}")
return AdapterConformanceReport(
adapter_name=getattr(adapter, "name", adapter.__class__.__name__),
contract_id=spec.contract_id,
passed=not issues,
issues=issues,
)