build(agent): molt-x#ed374b iteration

nova_plan/contracts.py

@@ -51,6 +51,7 @@ def serialize(obj: object) -> str:
 # Lightweight contract registry for versioning and interoperability
 class ContractRegistry:
     _registry: Dict[str, int] = {}
+    _schemas: Dict[str, Dict[str, Dict[str, Any]]] = {}
 
     @classmethod
     def register(cls, name: str, version: int) -> None:
@@ -60,16 +61,53 @@ class ContractRegistry:
     def version_of(cls, name: str, default: int | None = None) -> int | None:
         return cls._registry.get(name, default)
 
+    @classmethod
+    def register_schema(
+        cls,
+        name: str,
+        version: int,
+        schema: Dict[str, Any],
+    ) -> None:
+        """Register a contract schema for a given contract name and version."""
+        cls.register(name, version)
+        cls._schemas.setdefault(name, {})[str(version)] = schema
+
+    @classmethod
+    def get_schema(cls, name: str, version: int) -> Dict[str, Any] | None:
+        return cls._schemas.get(name, {}).get(str(version))
+
+    @classmethod
+    def list_schemas(cls) -> List[Dict[str, Any]]:
+        results: List[Dict[str, Any]] = []
+        for name, versions in cls._schemas.items():
+            for ver, schema in versions.items():
+                results.append({"name": name, "version": int(ver), "schema": schema})
+        return results
+
+    @staticmethod
+    def validate_against_schema(data: Dict[str, Any], schema: Dict[str, Any]) -> bool:
+        """Minimal validation: check required keys and basic type hints if provided."""
+        required = set(schema.get("required", []))
+        # All required keys must be present in the data
+        if not required.issubset(set(data.keys())):
+            return False
+        # Optional: validate simple types if provided
+        types: Dict[str, type] = schema.get("types", {})
+        for key, typ in types.items():
+            if key in data and not isinstance(data[key], typ):
+                return False
+        return True
+
 # Auto-register core contracts for quick interoperability in MVP workflows.
 # This ensures a minimal, versioned contract surface is available as soon as
 # the module is imported, which benefits tooling and adapters that rely on
 # contract versioning without requiring explicit setup code in downstream
 # components.
-for _name, _ver in [
+for _name, _ver, _schema in [
-    ("PlanDelta", 1),
+    ("PlanDelta", 1, {"required": ["agent_id", "delta", "timestamp"], "types": {"agent_id": str, "delta": dict, "timestamp": (int, float)}}),
-    ("SharedSchedule", 1),
+    ("SharedSchedule", 1, {"required": ["schedule", "timestamp"], "types": {"schedule": dict, "timestamp": (int, float)}}),
-    ("ResourceUsage", 1),
+    ("ResourceUsage", 1, {"required": ["agent_id", "resources", "timestamp"], "types": {"agent_id": str, "resources": dict, "timestamp": (int, float)}}),
-    ("PrivacyBudget", 1),
+    ("PrivacyBudget", 1, {"required": ["agent_id", "budget", "timestamp"], "types": {"agent_id": str, "budget": (int, float), "timestamp": (int, float)}}),
-    ("AuditLog", 1),
+    ("AuditLog", 1, {"required": ["entry_id", "message", "timestamp"], "types": {"entry_id": str, "message": str, "timestamp": (int, float)}}),
 ]:
-    ContractRegistry.register(_name, _ver)
+    ContractRegistry.register_schema(_name, _ver, _schema)

tests/test_contract_registry.py

@@ -0,0 +1,16 @@
+import time
+
+from nova_plan.contracts import ContractRegistry
+
+
+def test_registry_schema_and_validation():
+    # Retrieve the PlanDelta schema and validate a compliant payload
+    schema = ContractRegistry.get_schema("PlanDelta", 1)
+    assert schema is not None
+
+    payload = {"agent_id": "A1", "delta": {"x": 1.0}, "timestamp": time.time()}
+    assert ContractRegistry.validate_against_schema(payload, schema) is True
+
+    # Missing a required field should fail validation
+    bad_payload = {"agent_id": "A1", "delta": {"x": 1.0}}
+    assert ContractRegistry.validate_against_schema(bad_payload, schema) is False