build(agent): new-agents-2#7e3bbc iteration

core/contract_registry.py

@@ -1,41 +1,42 @@
-from typing import Dict, List, Any, Optional
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
 
 
 class ContractRegistry:
-    """
+    """Lightweight, in-memory, versioned contract registry for MVP tests.
-    Lightweight, versioned contract registry for CatOpt-Graph MVP.
+
-    Stores schema definitions per contract name and version.
+    - Contracts are registered by name and version.
-    Provides a simple conformance check against adapter-provided data.
+    - Each contract has a schema (dict) describing LocalProblem/SharedVariables/etc.
+    - Exposes simple get_contract(name, version), register_contract(...), and conformance checks.
+    - Exposes list_versions(name) to mirror test expectations.
     """
 
     def __init__(self) -> None:
-        # Structure: { name: { version: schema_dict } }
+        self._store: Dict[str, Dict[str, Dict[str, Any]]] = {}
-        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:
+        """Register or update a contract schema for a given name/version."""
-            self._registry[name] = {}
+        # Store a shallow copy to avoid accidental external mutation
-        self._registry[name][version] = schema
+        self._store.setdefault(name, {})[version] = dict(schema)
 
     def get_contract(self, name: str, version: str) -> Optional[Dict[str, Any]]:
-        return self._registry.get(name, {}).get(version)
+        return self._store.get(name, {}).get(version)
 
-    def list_versions(self, name: str) -> List[str]:
+    def list_versions(self, name: str) -> list[str]:
-        return list(self._registry.get(name, {}).keys())
+        return sorted(list(self._store.get(name, {}).keys()))
 
     def conformance_check(self, name: str, version: str, adapter_data: Dict[str, Any]) -> bool:
-        """
+        """Check if adapter_data satisfies the contract's required_fields.
-        Very lightweight conformance check:
+
-        - The contract schema defines required_fields.
+        Returns True if all required_fields (defined in schema) are present in adapter_data.
-        - adapter_data must contain all required fields at top level.
+        If no contract or required_fields are defined, returns False to be safe.
-        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
+        required = contract.get("required_fields") or []
-        for field in required:
+        if not isinstance(required, (list, tuple)):
-            if field not in adapter_data:
             return False
-        return True
+        return all(field in adapter_data for field in required)