build(agent): new-agents-3#dd492b iteration

src/idea15_edgemind_verifiable_onboard/__init__.py

@@ -2,6 +2,7 @@
 
 from .planner import Action, PlanRequest, PlanResult, EdgeMindPlanner  # re-export for convenience
 from .contracts import SafetyContract  # contractual primitive
+from .energi_bridge import EnergiBridge, LocalProblem, SharedSignals, PlanDelta, AuditLog, AdapterContract
 
 __all__ = [
     "Action",
@@ -9,4 +10,10 @@ __all__ = [
     "PlanResult",
     "EdgeMindPlanner",
     "SafetyContract",
+    "EnergiBridge",
+    "LocalProblem",
+    "SharedSignals",
+    "PlanDelta",
+    "AuditLog",
+    "AdapterContract",
 ]

src/idea15_edgemind_verifiable_onboard/energi_bridge.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Any
+
+# Canonical seed types (toy DSL seeds) for cross-vendor interoperability
+
+@dataclass
+class LocalProblem:
+    asset_id: str
+    tasks: List[str]
+    budgets: Dict[str, float]
+
+
+@dataclass
+class SharedSignals:
+    version: int
+    data: Dict[str, Any]
+
+
+@dataclass
+class PlanDelta:
+    timestamp: float
+    delta_actions: List[str]
+    safety_tags: List[str]
+
+
+@dataclass
+class AuditLog:
+    entries: List[str]
+
+
+@dataclass
+class AdapterContract:
+    name: str
+    version: str
+
+
+class EnergiBridge:
+    """Tiny bridge that maps EdgeMind primitives to a canonical representation.
+
+    This is intentionally lightweight and toy-like: it acts as a seed for a
+    more complete CatOpt-inspired IR that adapters can exchange.
+    """
+
+    def __init__(self) -> None:
+        # In a real system, this would be a persistent registry; keep it simple here
+        self.adapters: Dict[str, AdapterContract] = {}
+
+    # Canonical mapping from a LocalProblem to a dictionary representation
+    def map_local_problem(self, lp: LocalProblem) -> Dict[str, Any]:
+        return {
+            "canonical": "LocalProblem",
+            "asset_id": lp.asset_id,
+            "tasks": list(lp.tasks),
+            "budgets": dict(lp.budgets),
+        }
+
+    def map_plan_delta(self, delta: PlanDelta) -> Dict[str, Any]:
+        return {
+            "canonical": "PlanDelta",
+            "timestamp": delta.timestamp,
+            "delta_actions": list(delta.delta_actions),
+            "safety_tags": list(delta.safety_tags),
+        }
+
+    def register_adapter(self, adapter: AdapterContract) -> None:
+        self.adapters[adapter.name] = adapter
+
+    def get_adapter(self, name: str) -> AdapterContract | None:
+        return self.adapters.get(name)
+
+
+__all__ = [
+    "LocalProblem",
+    "SharedSignals",
+    "PlanDelta",
+    "AuditLog",
+    "AdapterContract",
+    "EnergiBridge",
+]