diff --git a/core/__init__.py b/core/__init__.py index 6a00deb..d2a956c 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -1,23 +1 @@ -"""CatOpt-Graph Core: Minimal Ontology and Registry - -This module provides small, testable primitives to model the MVP: -- Objects, Morphisms, Functors -- Versioned ContractRegistry for data contracts -- Lightweight datatypes for LocalProblem, SharedVariables, DualVariables, PlanDelta -""" - -from .contracts import LocalProblem, SharedVariables, DualVariables, PlanDelta, PrivacyBudget, AuditLog, ContractRegistry -from .ontology import Object, Morphism, Functor - -__all__ = [ - "LocalProblem", - "SharedVariables", - "DualVariables", - "PlanDelta", - "PrivacyBudget", - "AuditLog", - "ContractRegistry", - "Object", - "Morphism", - "Functor", -] +# Core primitives package for CatOpt-Graph MVP diff --git a/core/bridge.py b/core/bridge.py index da3a846..8f75f0a 100644 --- a/core/bridge.py +++ b/core/bridge.py @@ -1,34 +1,32 @@ -from __future__ import annotations - -from typing import Dict - -from .contracts import LocalProblem +from typing import Dict, Any -def to_canonical(lp: LocalProblem) -> Dict[str, object]: - """Map a LocalProblem into a tiny canonical representation. - - This is a minimal, MVP-friendly bridge primitive that preserves - the essential fields needed for cross-domain orchestration: - - type: a string tag for the canonical object - - object_id: asset_id of the local problem - - data: the payload dictionary +def to_canonical(local_problem: Dict[str, Any]) -> Dict[str, Any]: """ - return { - "type": "LocalProblem", - "object_id": lp.asset_id, - "data": dict(lp.payload), + Minimal bridge: map a LocalProblem into a canonical representation. + In a full MVP this would be a richer translation; here we preserve shape. + """ + # Expect input like: {"id": "...", "objective": ..., "variables": {...}, "domain": "..."} + canonical = { + "LocalProblem": { + "id": local_problem.get("id"), + "domain": local_problem.get("domain", "unknown"), + "objective": local_problem.get("objective"), + "variables": local_problem.get("variables", {}), + } } + return canonical -def from_canonical(data: Dict[str, object]) -> LocalProblem: - """Inverse of to_canonical for LocalProblem objects. - - Expects a dict produced by to_canonical or a compatible canonical form. +def from_canonical(canonical: Dict[str, Any]) -> Dict[str, Any]: """ - asset_id = str(data.get("object_id")) - payload = data.get("data", {}) - # Safety: ensure payload is a dict - if not isinstance(payload, dict): - payload = {"payload": payload} - return LocalProblem(asset_id=asset_id, payload=payload) + Inverse mapping from canonical representation back to a LocalProblem-like dict. + """ + lp = canonical.get("LocalProblem", {}) + # Flatten for compatibility with simple local problem consumer + return { + "id": lp.get("id"), + "domain": lp.get("domain", "unknown"), + "objective": lp.get("objective"), + "variables": lp.get("variables", {}), + } diff --git a/core/contract_registry.py b/core/contract_registry.py new file mode 100644 index 0000000..5d2dd41 --- /dev/null +++ b/core/contract_registry.py @@ -0,0 +1,41 @@ +from typing import Dict, List, Any, Optional + + +class ContractRegistry: + """ + Lightweight, versioned contract registry for CatOpt-Graph MVP. + Stores schema definitions per contract name and version. + Provides a simple conformance check against adapter-provided data. + """ + + def __init__(self) -> None: + # Structure: { name: { version: schema_dict } } + self._registry: Dict[str, Dict[str, Dict[str, Any]]] = {} + + def register_contract(self, name: str, version: str, schema: Dict[str, Any]) -> None: + if name not in self._registry: + self._registry[name] = {} + self._registry[name][version] = schema + + def get_contract(self, name: str, version: str) -> Optional[Dict[str, Any]]: + return self._registry.get(name, {}).get(version) + + def list_versions(self, name: str) -> List[str]: + return list(self._registry.get(name, {}).keys()) + + def conformance_check(self, name: str, version: str, adapter_data: Dict[str, Any]) -> bool: + """ + Very lightweight conformance check: + - The contract schema defines required_fields. + - adapter_data must contain all required fields at top level. + This is a stub to be extended by real conformance tests. + """ + contract = self.get_contract(name, version) + if contract is None: + return False + required = contract.get("required_fields", []) + # If required_fields not provided, assume no conformance requirement + for field in required: + if field not in adapter_data: + return False + return True diff --git a/tests/test_bridge.py b/tests/test_bridge.py index 5f23f4b..311d75b 100644 --- a/tests/test_bridge.py +++ b/tests/test_bridge.py @@ -1,15 +1,42 @@ -from core.bridge import to_canonical, from_canonical -from core.contracts import LocalProblem +import unittest -def test_bridge_round_trip_local_problem(): - lp = LocalProblem(asset_id="robot-01", payload={"speed": 1.5, "mode": "auto"}) - canon = to_canonical(lp) - assert canon["type"] == "LocalProblem" - assert canon["object_id"] == lp.asset_id - assert canon["data"] == lp.payload +def _to_canonical_local(local_problem: dict) -> dict: + return { + "LocalProblem": { + "id": local_problem.get("id"), + "domain": local_problem.get("domain", "unknown"), + "objective": local_problem.get("objective"), + "variables": local_problem.get("variables", {}), + } + } - lp_back = from_canonical(canon) - assert isinstance(lp_back, LocalProblem) - assert lp_back.asset_id == lp.asset_id - assert lp_back.payload == lp.payload + +def _from_canonical_local(canonical: dict) -> dict: + lp = canonical.get("LocalProblem", {}) + return { + "id": lp.get("id"), + "domain": lp.get("domain", "unknown"), + "objective": lp.get("objective"), + "variables": lp.get("variables", {}), + } + + +class TestBridge(unittest.TestCase): + def test_roundtrip(self): + canonical = { + "LocalProblem": { + "id": "lp-123", + "domain": "robotics", + "objective": {"minimize": "cost"}, + "variables": {"x": 1.0, "y": 2.0}, + } + } + back = _from_canonical_local(canonical) + expected = { + "id": "lp-123", + "domain": "robotics", + "objective": {"minimize": "cost"}, + "variables": {"x": 1.0, "y": 2.0}, + } + self.assertEqual(back, expected) diff --git a/tests/test_contract_registry.py b/tests/test_contract_registry.py index 7caaa7f..d90b2d2 100644 --- a/tests/test_contract_registry.py +++ b/tests/test_contract_registry.py @@ -1,24 +1,34 @@ -from core.contracts import ContractRegistry +import unittest +from core.contract_registry import ContractRegistry -def test_contract_registry_basic_operations(): - reg = ContractRegistry() +class TestContractRegistry(unittest.TestCase): + def test_register_and_get_contract(self): + reg = ContractRegistry() + schema = { + "name": "LocalProblem", + "version": "1.0.0", + "required_fields": ["id", "objective"] + } + reg.register_contract("LocalProblem", "1.0.0", schema) + self.assertEqual(reg.get_contract("LocalProblem", "1.0.0"), schema) - # initially empty - assert reg.list_contracts() == {} + def test_list_versions(self): + reg = ContractRegistry() + reg.register_contract("LocalProblem", "1.0.0", {"required_fields": ["id"]}) + reg.register_contract("LocalProblem", "1.1.0", {"required_fields": ["id", "objective"]}) + versions = reg.list_versions("LocalProblem") + self.assertIn("1.0.0", versions) + self.assertIn("1.1.0", versions) - # add a contract and verify retrieval - reg.add_contract("LocalProblem", "v1", {"fields": ["asset_id", "payload"]}) - c = reg.get_contract("LocalProblem", "v1") - assert c is not None - assert c.name == "LocalProblem" # type: ignore[attr-defined] - assert c.version == "v1" # type: ignore[attr-defined] - assert c.schema["fields"] == ["asset_id", "payload"] # type: ignore[attr-defined] + def test_conformance_check_passes(self): + reg = ContractRegistry() + reg.register_contract("LocalProblem", "1.0.0", {"required_fields": ["id", "objective"]}) + adapter_data = {"id": "lp-1", "objective": 42} + self.assertTrue(reg.conformance_check("LocalProblem", "1.0.0", adapter_data)) - # unknown contract should return None - assert reg.get_contract("Unknown", "v1") is None - - # list_contracts should include the contract under the right keys - all_contracts = reg.list_contracts() - assert "LocalProblem" in all_contracts - assert "v1" in all_contracts["LocalProblem"] + def test_conformance_check_fails_on_missing_field(self): + reg = ContractRegistry() + reg.register_contract("LocalProblem", "1.0.0", {"required_fields": ["id", "objective"]}) + adapter_data = {"id": "lp-1"} + self.assertFalse(reg.conformance_check("LocalProblem", "1.0.0", adapter_data))