build(agent): molt-y#23e5c8 iteration

nova_plan/contracts.py

@@ -111,3 +111,76 @@ for _name, _ver, _schema in [
     ("AuditLog", 1, {"required": ["entry_id", "message", "timestamp"], "types": {"entry_id": str, "message": str, "timestamp": (int, float)}}),
 ]:
     ContractRegistry.register_schema(_name, _ver, _schema)
+
+
+# Lightweight Adapter Registry (Graph-of-Contracts for adapters)
+class AdapterRegistry:
+    """Minimal registry to track adapter versions and their schemas.
+
+    This mirrors the contract registry pattern but for adapter software units
+    (e.g., rover HabitatAdapter, etc.). It enables plugging in vendor adapters
+    while keeping a versioned contract surface for interoperability tooling.
+    """
+
+    _registry: Dict[str, int] = {}
+    _schemas: Dict[str, Dict[str, Dict[str, Any]]] = {}
+
+    @classmethod
+    def register_adapter(cls, name: str, version: int) -> None:
+        cls._registry[name] = int(version)
+
+    @classmethod
+    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:
+        cls.register_adapter(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:
+        required = set(schema.get("required", []))
+        if not required.issubset(set(data.keys())):
+            return False
+        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
+
+
+# Pre-register a couple of MVP adapter schemas to illustrate interoperability.
+AdapterRegistry.register_schema(
+    name="RoverAdapter",
+    version=1,
+    schema={
+        "required": ["adapter_id", "status"],
+        "types": {"adapter_id": str, "status": dict},
+    },
+)
+AdapterRegistry.register_schema(
+    name="HabitatAdapter",
+    version=1,
+    schema={
+        "required": ["module_id", "status"],
+        "types": {"module_id": str, "status": dict},
+    },
+)