build(agent): new-agents-2#7e3bbc iteration

arbsphere/energi_bridge.py

@@ -9,7 +9,7 @@ from __future__ import annotations
 
 from typing import Dict, Any
 
-from .primitives import LocalArbProblem, SharedSignals, PlanDelta, DualVariables, PrivacyBudget, AuditLog, TimeRounds, GoCRegistry
+from .primitives import LocalArbProblem, SharedSignals, PlanDelta, DualVariables, PrivacyBudget, AuditLog, TimeRounds, GoCRegistry, DeltaTrace
 
 
 def to_ir(obj: object) -> Dict[str, Any]:
@@ -20,6 +20,8 @@ def to_ir(obj: object) -> Dict[str, Any]:
         return {"contract_type": "SharedSignals", "payload": obj.to_json()}
     if isinstance(obj, PlanDelta):
         return {"contract_type": "PlanDelta", "payload": obj.to_json()}
+    if isinstance(obj, DeltaTrace):
+        return {"contract_type": "DeltaTrace", "payload": obj.to_json()}
     if isinstance(obj, DualVariables):
         return {"contract_type": "DualVariables", "payload": obj.to_json()}
     if isinstance(obj, PrivacyBudget):
@@ -43,6 +45,8 @@ def from_ir(payload: Dict[str, Any]) -> object:
         return SharedSignals.from_json(data)
     if ct == "PlanDelta":
         return PlanDelta.from_json(data)
+    if ct == "DeltaTrace":
+        return DeltaTrace.from_json(data)
     if ct == "DualVariables":
         return DualVariables.from_json(data)
     if ct == "PrivacyBudget":

arbsphere/primitives.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import json
 from dataclasses import dataclass, field
+import hashlib
 from typing import List, Dict, Any, Optional
 
 
@@ -231,3 +232,43 @@ class GoCRegistry:
     @classmethod
     def from_json(cls, data: Dict[str, Any]) -> "GoCRegistry":
         return cls(adapters=data.get("adapters", {}))
+
+
+@dataclass
+class DeltaTrace:
+    """Deterministic replay trace for a PlanDelta per venue.
+
+    This is a lightweight hash- chained log entry capturing the delta payload
+    and its provenance. It enables auditable backtesting and partition reconciliation
+    without exposing raw delta details beyond the payload envelope.
+    """
+    delta_hash: str
+    parent_hash: str | None
+    timestamp: float
+    signer: str
+    payload: Dict[str, Any]
+
+    def to_json(self) -> Dict[str, Any]:
+        return {
+            "delta_hash": self.delta_hash,
+            "parent_hash": self.parent_hash,
+            "timestamp": self.timestamp,
+            "signer": self.signer,
+            "payload": self.payload,
+        }
+
+    @classmethod
+    def from_json(cls, data: Dict[str, Any]) -> "DeltaTrace":
+        return cls(
+            delta_hash=data["delta_hash"],
+            parent_hash=data.get("parent_hash"),
+            timestamp=data["timestamp"],
+            signer=data["signer"],
+            payload=data["payload"],
+        )
+
+    @staticmethod
+    def compute_hash(payload: Dict[str, Any]) -> str:
+        # Stable hash of the JSON payload
+        payload_bytes = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        return hashlib.sha256(payload_bytes).hexdigest()

tests/test_primitives.py

@@ -1,7 +1,7 @@
 import json
 import time
 
-from arbsphere.primitives import LocalArbProblem, SharedSignals, PlanDelta, DualVariables, PrivacyBudget, AuditLog, TimeRounds, GoCRegistry
+from arbsphere.primitives import LocalArbProblem, SharedSignals, PlanDelta, DualVariables, PrivacyBudget, AuditLog, TimeRounds, GoCRegistry, DeltaTrace
 from arbsphere.energi_bridge import to_ir, from_ir
 
 
@@ -38,3 +38,16 @@ def test_dual_variables_and_privacy_audit_roundtrip():
         restored = from_ir(ir)
         assert restored is not None
 
+
+def test_delta_trace_roundtrip():
+    # Create a simple PlanDelta and wrap it in a DeltaTrace with a hash-chain style linkage
+    pd = PlanDelta(delta=[{"action": "buy", "leg": "A-B", "size": 100}], timestamp=time.time(), author="org1")
+    payload = pd.to_json()
+    # compute delta hash for this payload as if part of a chain
+    delta_hash = DeltaTrace.compute_hash(payload)
+    dt = DeltaTrace(delta_hash=delta_hash, parent_hash=None, timestamp=time.time(), signer="org1", payload=payload)
+    ir = to_ir(dt)
+    assert ir["contract_type"] == "DeltaTrace"
+    restored = from_ir(ir)
+    assert isinstance(restored, DeltaTrace)
+    assert restored.delta_hash == dt.delta_hash