build(agent): molt-z#db0ec5 iteration

src/cosmosmesh_privacy_preserving_federated/__init__.py

@@ -1,9 +1,16 @@
 """CosmosMesh Privacy-Preserving Federated package.
 
 Exposes MVP scaffolds for bridging CosmosMesh primitives to a CatOpt-style
-representation via the catopt_bridge module.
+representation via the catopt_bridge module and a lightweight EnergiBridge
+for canonical interoperability with a CatOpt-like IR.
 """
 
 from .catopt_bridge import CatOptBridge, ContractRegistry, LocalProblem
-
+try:
+    from .energi_bridge import EnergiBridge, LocalProblemEP  # type: ignore
+    __all__ = ["CatOptBridge", "ContractRegistry", "LocalProblem", "EnergiBridge", "LocalProblemEP"]
+except Exception:
+    # EnergiBridge not yet available; avoid import-time failure for MVPs that
+    # import this package before energibridge module is added in a follow-up
+    # patch.
     __all__ = ["CatOptBridge", "ContractRegistry", "LocalProblem"]

src/cosmosmesh_privacy_preserving_federated/energi_bridge.py

@@ -0,0 +1,112 @@
+"""EnergiBridge: Canonical bridge mapping CosmosMesh → CatOpt-like IR (prototype).
+
+This module provides a minimal, strongly-typed bridge to translate CosmosMesh
+primitives into a vendor-agnostic, CatOpt-inspired intermediate representation
+suitable for plug-in adapters across domains (energy, robotics, space).
+
+The implementation is intentionally compact and dependency-free, focusing on a
+clean data model and a convenience encoder to a simple IR dictionary that tests
+and adapters can consume.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional
+
+
+# Lightweight per-asset local problem representation for EnergiBridge
+@dataclass
+class LocalProblemEP:
+    problem_id: str
+    assets: List[str]
+    objective: str
+    constraints: List[str]
+    data_contracts: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "problem_id": self.problem_id,
+            "assets": self.assets,
+            "objective": self.objective,
+            "constraints": self.constraints,
+            "data_contracts": self.data_contracts or {},
+        }
+
+
+@dataclass
+class SharedVariableEP:
+    channel: str
+    version: int
+    payload: Dict[str, Any]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "channel": self.channel,
+            "version": self.version,
+            "payload": self.payload,
+        }
+
+
+@dataclass
+class DualVariableEP:
+    channel: str
+    version: int
+    payload: Dict[str, Any]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "channel": self.channel,
+            "version": self.version,
+            "payload": self.payload,
+        }
+
+
+@dataclass
+class PlanDeltaEP:
+    delta_id: str
+    changes: Dict[str, Any]
+    timestamp: Optional[float] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        payload = {
+            "delta_id": self.delta_id,
+            "changes": self.changes,
+            "timestamp": self.timestamp,
+        }
+        return payload
+
+
+@dataclass
+class EnergiBridge:
+    """Encoder that builds a canonical, CatOpt-like IR payload from CosmosMesh
+    primitives.
+    """
+
+    def to_ir(self,
+              lp: LocalProblemEP,
+              shared: List[SharedVariableEP],
+              duals: List[DualVariableEP],
+              plan_deltas: Optional[List[PlanDeltaEP]] = None) -> Dict[str, Any]:
+        ir: Dict[str, Any] = {
+            "Version": "0.1",
+            "Objects": {
+                "LocalProblem": lp.to_dict(),
+            },
+        }
+
+        morphisms: List[Dict[str, Any]] = []
+        for s in shared:
+            morphisms.append({"Morphisms": {"SharedVariable": s.to_dict()}})
+        for d in duals:
+            morphisms.append({"Morphisms": {"DualVariable": d.to_dict()}})
+        if morphisms:
+            ir["Morphisms"] = morphisms
+
+        if plan_deltas:
+            ir.setdefault("Objects", {})["PlanDeltas"] = [pd.to_dict() for pd in plan_deltas]
+
+        return ir
+
+
+__all__ = ["LocalProblemEP", "SharedVariableEP", "DualVariableEP", "PlanDeltaEP", "EnergiBridge"]

tests/test_energi_bridge.py

@@ -0,0 +1,36 @@
+import pytest
+from cosmosmesh_privacy_preserving_federated.energi_bridge import (
+    LocalProblemEP,
+    SharedVariableEP,
+    DualVariableEP,
+    PlanDeltaEP,
+    EnergiBridge,
+)
+
+
+def test_energi_bridge_basic_encoding():
+    lp = LocalProblemEP(
+        problem_id="lp-001",
+        assets=["rover-1", "drone-A"],
+        objective="maximize_task_completion",
+        constraints=["energy<=1000", "time_window=08:00-12:00"],
+        data_contracts={"Telemetry": {"version": 1}},
+    )
+
+    sv1 = SharedVariableEP(channel="energy_budget", version=1, payload={"value": 42})
+    dv1 = DualVariableEP(channel="lag_energy", version=1, payload={"value": 0.5})
+
+    delta = PlanDeltaEP(delta_id="d-1", changes={"rover-1": {"move_to": "A"}}, timestamp=123456.0)
+
+    bridge = EnergiBridge()
+    ir = bridge.to_ir(lp, [sv1], [dv1], [delta])
+
+    # Basic shape checks
+    assert ir["Version"] == "0.1"
+    assert "Objects" in ir
+    assert ir["Objects"]["LocalProblem"]["problem_id"] == "lp-001"
+    assert "Morphisms" in ir
+    # We expect at least one SharedVariable and one DualVariable in the Morphisms list
+    morphs = ir["Morphisms"]
+    assert any("SharedVariable" in m.get("Morphisms", {}) for m in morphs) or any("DualVariable" in m.get("Morphisms", {}) for m in morphs)
+    assert "PlanDeltas" in ir["Objects"]