build(agent): new-agents-4#58ba63 iteration

src/flashplan_edge_optimized_cross_venue_exe/__init__.py

@@ -1,5 +1,6 @@
 """FlashPlan Edge package initializer"""
 
 from .core import LocalProblem, SharedVariables, PlanDelta, EdgePlanner, admm_step
+from .energia_bridge import EnergiaBridge, GraphContractRegistry
 
-__all__ = ["LocalProblem", "SharedVariables", "PlanDelta", "EdgePlanner", "admm_step"]
+__all__ = ["LocalProblem", "SharedVariables", "PlanDelta", "EdgePlanner", "admm_step", "EnergiaBridge", "GraphContractRegistry"]

src/flashplan_edge_optimized_cross_venue_exe/energia_bridge.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from typing import Dict, Any, Optional
+
+from .core import LocalProblem, SharedVariables, PlanDelta
+
+
+class GraphContractRegistry:
+    """
+    Lightweight in-memory registry for adapters and data schemas.
+    """
+    def __init__(self):
+        self.adapters: Dict[str, Dict[str, Any]] = {}
+
+    def register(self, name: str, schema: Dict[str, Any]) -> None:
+        self.adapters[name] = {"schema": schema}
+
+    def get(self, name: str) -> Optional[Dict[str, Any]]:
+        return self.adapters.get(name)
+
+    def all(self) -> Dict[str, Dict[str, Any]]:
+        return self.adapters
+
+
+class EnergiaBridge:
+    """
+    Lightweight bridge mapping from GridGuard-inspired primitives to a
+    canonical IR (Objects=LocalProblems, Morphisms=SharedVariables, PlanDelta),
+    with per-message metadata for replay/audit. This is intentionally tiny and
+    dependency-free to serve as a drop-in scaffold for MVP wiring.
+    """
+    def __init__(self) -> None:
+        self.registry = GraphContractRegistry()
+
+    # Canonicalization: LocalProblem -> Object
+    def to_canonical_object(self, lp: LocalProblem) -> Dict[str, Any]:
+        return {
+            "type": "LocalProblem",
+            "version": 1,
+            "object_id": lp.asset_id,
+            "attributes": {
+                "objective": lp.objective,
+                "target_metric": lp.target_metric,
+                "constraints": lp.constraints,
+            },
+        }
+
+    # Canonicalize SharedVariables as Morphisms/Signals
+    def to_canonical_morphisms(self, shared: SharedVariables) -> Dict[str, Any]:
+        return {
+            "type": "SharedVariables",
+            "version": 1,
+            "signals": shared.signals,
+            "latency_budget": shared.latency_budget,
+        }
+
+    # Delta messages
+    def delta_to_canonical(self, delta: PlanDelta) -> Dict[str, Any]:
+        return {
+            "type": "PlanDelta",
+            "version": 1,
+            "trades": delta.trades,
+            "hedge_updates": delta.hedge_updates,
+            "metadata": delta.metadata,
+        }
+
+    # Basic round-trip transforms
+    def from_canonical_object(self, obj: Dict[str, Any]) -> LocalProblem:
+        attrs = obj.get("attributes", {})
+        return LocalProblem(
+            asset_id=obj.get("object_id", ""),
+            objective=attrs.get("objective", "min"),
+            target_metric=float(attrs.get("target_metric", 0.0)),
+            constraints=attrs.get("constraints", {}),
+        )
+
+    def apply_delta(self, lp: LocalProblem, shared: SharedVariables, delta: PlanDelta) -> PlanDelta:
+        """
+        Minimal 'simulation' of applying a delta to a local problem and shared vars.
+        This is intentionally simplistic for MVP parity; real systems would apply
+        constraints, update signals, and adjust budgets here.
+        """
+        # naive: bump a signal if trades exist, to show the delta had an effect
+        if delta and delta.trades:
+            shared.signals["last_delta_trades"] = len(delta.trades)
+        return delta
+
+
+__all__ = ["EnergiaBridge", "GraphContractRegistry"]