diff --git a/catopt_flow_category_theoretic_compositi/__init__.py b/catopt_flow_category_theoretic_compositi/__init__.py index 5d27fd5..ebd384d 100644 --- a/catopt_flow_category_theoretic_compositi/__init__.py +++ b/catopt_flow_category_theoretic_compositi/__init__.py @@ -1,4 +1,5 @@ from .core import LocalProblem, GlobalProblem, Object, Morphism, Functor, Planner, ADMMNode, DeltaSyncRegistry, run_admm +from .contract_registry import AdapterContract, ContractRegistry __all__ = [ "LocalProblem", @@ -10,4 +11,6 @@ __all__ = [ "ADMMNode", "DeltaSyncRegistry", "run_admm", + "AdapterContract", + "ContractRegistry", ] diff --git a/catopt_flow_category_theoretic_compositi/contract_registry.py b/catopt_flow_category_theoretic_compositi/contract_registry.py new file mode 100644 index 0000000..f831b89 --- /dev/null +++ b/catopt_flow_category_theoretic_compositi/contract_registry.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Dict, List, Optional + + +@dataclass(frozen=True) +class AdapterContract: + """Describes a plug-and-play adapter contract for an ML framework/back-end. + + This is a lightweight descriptor used by the Graph-of-Contracts registry to + advertise capabilities and interface expectations without pulling in the + actual adapter implementation. + """ + platform: str # e.g., "pytorch", "tensorflow", "jax" + name: str # e.g., "cuda-gpu-adapter" + version: str + capabilities: List[str] = field(default_factory=list) + contract: Dict[str, object] = field(default_factory=dict) + + def __post_init__(self): + # Normalize common keys if present to avoid KeyErrors downstream + if self.contract is None: + object.__setattr__(self, "contract", {}) + + +class ContractRegistry: + """A tiny registry for adapter contracts (Graph-of-Contracts skeleton). + + This registry stores AdapterContract descriptors by a string key. It is + intentionally small but ready to evolve into a richer registry backed by a + database or remote service in the future. + """ + + def __init__(self) -> None: + self._registry: Dict[str, AdapterContract] = {} + + def register(self, key: str, contract: AdapterContract) -> None: + self._registry[key] = contract + + def get(self, key: str) -> Optional[AdapterContract]: + return self._registry.get(key) + + def list(self) -> List[AdapterContract]: + return list(self._registry.values()) + + +__all__ = ["AdapterContract", "ContractRegistry"] diff --git a/tests/test_contract_registry.py b/tests/test_contract_registry.py new file mode 100644 index 0000000..d9e504d --- /dev/null +++ b/tests/test_contract_registry.py @@ -0,0 +1,16 @@ +from catopt_flow_category_theoretic_compositi import ContractRegistry, AdapterContract + + +def test_registry_basic(): + reg = ContractRegistry() + ac = AdapterContract( + platform="pytorch", + name="gpu-adapter", + version="0.1.0", + capabilities=["training", "autograd"], + contract={"interface": "local_problem"}, + ) + reg.register("pytorch/gpu-adapter", ac) + assert reg.get("pytorch/gpu-adapter") is ac + items = reg.list() + assert len(items) == 1