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

src/idea15_edgemind_verifiable_onboard/energi_bridge.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Optional
 
 # Canonical seed types (toy DSL seeds) for cross-vendor interoperability
 
@@ -30,6 +30,18 @@ class AuditLog:
     entries: List[str]
 
 
+@dataclass
+class DualVariables:
+    # Minimal representation of coupling signals used in joint optimization
+    signals: Dict[str, float]
+
+
+@dataclass
+class GovernanceLog:
+    # Lightweight governance/audit metadata for replayability and accountability
+    entries: List[str]
+
+
 @dataclass
 class AdapterContract:
     name: str
@@ -92,12 +104,28 @@ class EnergiBridge:
             "entries": list(log.entries),
         }
 
+    def map_dual_variables(self, dv: "DualVariables") -> Dict[str, Any]:
+        """Map DualVariables to a canonical representation."""
+        return {
+            "canonical": "DualVariables",
+            "signals": dict(dv.signals),
+        }
+
+    def map_governance(self, gv: "GovernanceLog") -> Dict[str, Any]:
+        """Map GovernanceLog to a canonical representation."""
+        return {
+            "canonical": "GovernanceLog",
+            "entries": list(gv.entries),
+        }
+
     def export_envelope(
         self,
-        lp: LocalProblem | None = None,
+        lp: Optional[LocalProblem] = None,
-        delta: PlanDelta | None = None,
+        delta: Optional[PlanDelta] = None,
-        signals: SharedSignals | None = None,
+        signals: Optional[SharedSignals] = None,
-        log: AuditLog | None = None,
+        log: Optional[AuditLog] = None,
+        dv: Optional["DualVariables"] = None,
+        gv: Optional["GovernanceLog"] = None,
     ) -> Dict[str, Any]:
         """Export a combined canonical envelope of provided pieces.
 
@@ -113,6 +141,10 @@ class EnergiBridge:
             payload["SharedSignals"] = self.map_shared_signals(signals)
         if log is not None:
             payload["AuditLog"] = self.map_audit_log(log)
+        if dv is not None:
+            payload["DualVariables"] = self.map_dual_variables(dv)
+        if gv is not None:
+            payload["GovernanceLog"] = self.map_governance(gv)
         return payload
 
 
@@ -123,4 +155,6 @@ __all__ = [
     "AuditLog",
     "AdapterContract",
     "EnergiBridge",
+    "DualVariables",
+    "GovernanceLog",
 ]

tests/test_energi_bridge_ext.py

@@ -0,0 +1,41 @@
+import os
+import sys
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "src")))
+
+from idea15_edgemind_verifiable_onboard.energi_bridge import (
+    EnergiBridge,
+    LocalProblem,
+    DualVariables,
+    GovernanceLog,
+    PlanDelta,
+)
+
+
+def test_map_dual_variables_basic():
+    eb = EnergiBridge()
+    dv = DualVariables(signals={"shadow_cost": 0.5, "price": 1.2})
+    canon = eb.map_dual_variables(dv)
+    assert canon["canonical"] == "DualVariables"
+    assert canon["signals"]["shadow_cost"] == 0.5
+    assert canon["signals"]["price"] == 1.2
+
+
+def test_map_governance_basic():
+    eb = EnergiBridge()
+    gv = GovernanceLog(entries=["gen-rule-1", "audit-2"])
+    canon = eb.map_governance(gv)
+    assert canon["canonical"] == "GovernanceLog"
+    assert canon["entries"] == ["gen-rule-1", "audit-2"]
+
+
+def test_envelope_with_new_components():
+    eb = EnergiBridge()
+    lp = LocalProblem(asset_id="robot-03", tasks=["navigate"], budgets={"energy": 10.0})
+    delta = PlanDelta(timestamp=99.9, delta_actions=["turn"], safety_tags=["ok"])
+    dv = DualVariables(signals={"shadow_cost": 0.1})
+    gv = GovernanceLog(entries=["init"])
+    envelope = eb.export_envelope(lp=lp, delta=delta, signals=None, log=None, dv=dv, gv=gv)
+    assert envelope["canonical"] == "Envelope"
+    assert envelope["LocalProblem"]["asset_id"] == "robot-03"
+    assert envelope["DualVariables"]["signals"]["shadow_cost"] == 0.1
+    assert envelope["GovernanceLog"]["entries"] == ["init"]