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

README.md

@@ -12,7 +12,14 @@ Core MVP outline
 - Lightweight transport (TLS-based) and a tiny ADMM-lite solver per asset
 - Graph-of-Contracts registry for schemas and per-message metadata to support audits and replay protection
 
-What’s included in this patch
+-What’s included in this patch
+- Added a minimal CatOpt bridge module: src/cosmosmesh_privacy_preserving_federated/catopt_bridge.py (already present, extended with examples)
+- Added DSL sketch: src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py
+- Added toy adapters: src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py and src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py
+- Package initializer: src/cosmosmesh_privacy_preserving_federated/__init__.py
+- Lightweight unit tests: tests/test_catopt_bridge.py (unchanged)
+- Simple usage example via a RoundTrip encoding path (as in tests)
+- Basic README describing MVP approach and how to extend
 - Added a minimal CatOpt bridge module: src/cosmosmesh_privacy_preserving_federated/catopt_bridge.py
 - Package initializer: src/cosmosmesh_privacy_preserving_federated/__init__.py
 - Lightweight unit test: tests/test_catopt_bridge.py

src/cosmosmesh_privacy_preserving_federated/adapters/habitat_life_support.py

@@ -0,0 +1,36 @@
+"""Toy habitat life-support adapter for CosmosMesh CatOpt bridge.
+
+Demonstrates exposing a LocalProblem and handling commands in a simple
+offline-first setting.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, PlanDelta
+
+
+class HabitatLifeSupportAdapter:
+    def __init__(self, habitat_id: str = "habitat-1") -> None:
+        self.habitat_id = habitat_id
+        self._state: Dict[str, Any] = {"life_support_on": True, "battery": 100.0}
+
+    def readState(self) -> Dict[str, Any]:
+        return {"habitat_id": self.habitat_id, "state": self._state}
+
+    def exposeLocalProblemData(self) -> LocalProblem:
+        lp = LocalProblem(
+            problem_id=f"habitat-{self.habitat_id}-lp",
+            version=1,
+            variables={"cooling": 22.0, "humidity": 40.0},
+            objective=1.0,
+            constraints=None,
+        )
+        return lp
+
+    def applyCommand(self, delta: PlanDelta) -> None:
+        # Simple: interpret delta changes into internal state flags if present
+        changes = delta.changes if isinstance(delta, PlanDelta) else {}
+        if isinstance(changes, dict):
+            for k, v in changes.items():
+                self._state[k] = v

src/cosmosmesh_privacy_preserving_federated/adapters/rover_planner.py

@@ -1,34 +1,39 @@
-from __future__ import annotations
+"""Toy rover planner adapter for CosmosMesh CatOpt bridge.
-from typing import Any, Dict
 
-from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, SharedVariables, DualVariables
+This adapter implements a minimal interface compatible with the CatOpt bridge
+and demonstrates how a device-specific model could be translated into a LocalProblem
+and how to expose local data as SharedVariables.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict, List
+
+from cosmosmesh_privacy_preserving_federated.catopt_bridge import LocalProblem, SharedVariable, PlanDelta, CatOptBridge
 
 
 class RoverPlannerAdapter:
-    """Minimal rover planner adapter scaffold.
+    def __init__(self, planner_id: str = "rover-1") -> None:
+        self.planner_id = planner_id
+        # Use a loosely-typed state dict to handle diverse state shapes across adapters
+        self._state: Dict[str, Any] = {
+            "tasks": ["survey", "sample"]
+        }
 
-    Exposes a tiny interface used by the MVP to demonstrate problem translation
+    def readState(self) -> Dict[str, Any]:
-    and command application. This is intentionally lightweight.
+        # Return a small snapshot of rover state as a dict
-    """
+        return {"planner_id": self.planner_id, "state": self._state}
 
-    def __init__(self) -> None:
+    def exposeLocalProblemData(self) -> LocalProblem:
-        pass
+        # Expose a simple LocalProblem for the rover to solve
-
-    def read_state(self) -> Dict[str, Any]:
-        # Placeholder rover state; in a real system this would query sensors.
-        return {"battery": 78, "position": [0.0, 0.0], "status": "idle"}
-
-    def expose_local_problem_data(self) -> LocalProblem:
-        # Return a toy LocalProblem instance for the MVP demo.
         lp = LocalProblem(
-            id="lp_rover_1",
+            problem_id=f"rover-{self.planner_id}-lp",
-            objective="minimize_energy",
-            variables={"task": "survey"},
-            constraints=["battery>=20"],
             version=1,
+            variables={"task_weight": 1.0, "battery": 90.0},
+            objective=0.0,
+            constraints=[{"battery_min": 20.0}],
         )
         return lp
 
-    def apply_command(self, command: Dict[str, Any]) -> Dict[str, Any]:
+    def applyCommand(self, command: Dict[str, Any]) -> None:
-        # Accept a command and return an acknowledgment. This is a stub.
+        # Apply a command to the rover planner (no-op in this toy example)
-        return {"ack": True, "command": command}
+        self._state["last_command"] = command

src/cosmosmesh_privacy_preserving_federated/dsl_sketch.py

@@ -1,49 +1,62 @@
-"""DSL sketch for CosmosMesh MVP primitives.
+"""Minimal DSL sketch for LocalProblem/SharedVariables/PlanDelta.
 
-The goal here is to provide lightweight dataclass-like shells that describe the
+This module provides a lightweight, Pythonic DSL to describe a per-agent
-contract between agents in a way that's friendly to MVP wiring and testing. These
+LocalProblem and associated signals (SharedVariables / PlanDelta). It is
-types are intentionally simple and purely for internal prototyping.
+intended for prototyping, tests, and documentation, not for production use.
 """
+
 from __future__ import annotations
 
-from dataclasses import dataclass, asdict
 from typing import Any, Dict, List, Optional
+from .catopt_bridge import LocalProblem, SharedVariable, PlanDelta, ContractRegistry, CatOptBridge
 
 
-@dataclass
 class LocalProblemDSL:
-    agent_id: str
+    """Fluent DSL for building a LocalProblem dataclass."""
-    objective: str
-    variables: List[str]
-    constraints: List[str]
 
-    def dict(self) -> Dict[str, Any]:
+    def __init__(self, problem_id: str, version: int = 1) -> None:
-        return asdict(self)
+        self.problem_id = problem_id
+        self.version = version
+        self.variables: Dict[str, Any] = {}
+        self.objective: Optional[Any] = None
+        self.constraints: List[Dict[str, Any]] = []
+
+    def var(self, name: str, value: Any) -> 'LocalProblemDSL':
+        self.variables[name] = value
+        return self
+
+    def set_objective(self, obj: Any) -> 'LocalProblemDSL':
+        self.objective = obj
+        return self
+
+    def add_constraint(self, constraint: Dict[str, Any]) -> 'LocalProblemDSL':
+        self.constraints.append(constraint)
+        return self
+
+    def build(self) -> LocalProblem:
+        return LocalProblem(
+            problem_id=self.problem_id,
+            version=self.version,
+            variables=self.variables,
+            objective=self.objective if self.objective is not None else 0,
+            constraints=self.constraints or None,
+        )
 
 
-@dataclass
 class SharedVariablesDSL:
-    signals: Dict[str, float]
+    """DSL helper to create SharedVariable instances."""
-    version: str = "0.1"
 
-    def dict(self) -> Dict[str, Any]:
+    @staticmethod
-        return asdict(self)
+    def sv(name: str, value: Any, version: int = 1) -> SharedVariable:
+        return SharedVariable(name, value, version)
 
 
-@dataclass
-class DualVariablesDSL:
-    lambdas: Dict[str, float]
-    version: str = "0.1"
-
-    def dict(self) -> Dict[str, Any]:
-        return asdict(self)
-
-
-@dataclass
 class PlanDeltaDSL:
-    delta_id: str
+    """DSL helper to create PlanDelta instances."""
-    changes: Dict[str, Any]
-    timestamp: Optional[str] = None
 
-    def dict(self) -> Dict[str, Any]:
+    @staticmethod
-        return asdict(self)
+    def delta(delta_id: str, changes: Dict[str, Any], timestamp: Optional[float] = None) -> PlanDelta:
+        return PlanDelta(delta_id=delta_id, changes=changes, timestamp=timestamp)
+
+
+__all__ = ["LocalProblemDSL", "SharedVariablesDSL", "PlanDeltaDSL"]