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

equicompiler_algebraic_portfolio_dsl_to_/core.py

@@ -3,6 +3,9 @@ from __future__ import annotations
 
 import json
 from typing import Dict, List
+import copy
+from datetime import datetime
+import hashlib
 
 
 def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
@@ -43,6 +46,8 @@ def parse_dsl_to_ir(dsl: str) -> Dict[str, object]:
         "constraints": constraints,
         "version": "0.1",
     }
+    # Attach lightweight, deterministic attestations to support auditability.
+    ir = _attach_attestations(ir)
     return ir
 
 
@@ -50,4 +55,25 @@ def to_json(ir: Dict[str, object]) -> str:
     return json.dumps(ir, indent=2, sort_keys=True)
 
 
+def _attach_attestations(ir: Dict[str, object]) -> Dict[str, object]:
+    """Attach deterministic attestations to the IR for auditability.
+
+    This is a lightweight, deterministic placeholder for cryptographic attestations.
+    It computes a SHA-256 digest of the IR excluding the attestations field and
+    adds per-step attestations to the IR under the 'attestations' key.
+    """
+    # Work on a shallow copy to avoid mutating input in unexpected ways
+    ir_copy = copy.deepcopy(ir)
+    base_for_digest = {k: v for k, v in ir_copy.items() if k != "attestations"}
+    digest = hashlib.sha256(json.dumps(base_for_digest, sort_keys=True).encode("utf-8")).hexdigest()
+    timestamp = datetime.utcnow().isoformat() + "Z"
+
+    attestations = [
+        {"step": "parse", "timestamp": timestamp, "digest": digest, "algorithm": "SHA-256"},
+        {"step": "verify", "timestamp": timestamp, "digest": digest, "algorithm": "SHA-256"},
+    ]
+    ir_copy["attestations"] = attestations
+    return ir_copy
+
+
 __all__ = ["parse_dsl_to_ir", "to_json"]

equicompiler_algebraic_portfolio_dsl_to_/registry.py

@@ -0,0 +1,79 @@
+"""Graph of Contracts (GoC) and versioned adapters skeleton.
+
+This module provides a tiny in-process registry to model versioned adapters
+and contracts that would connect the EquiCompiler to data feeds, brokers,
+and other runtime components. It is intended as a scaffold for future
+extensions and to support the MVP where plug-and-play backends are wired in
+via stable, versioned contracts.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+import time
+
+
+@dataclass
+class VersionedContract:
+    name: str
+    version: str
+    payload: Dict[str, Any] = field(default_factory=dict)
+    timestamp: float = field(default_factory=lambda: time.time())
+
+    def as_dict(self) -> Dict[str, Any]:
+        return {
+            "name": self.name,
+            "version": self.version,
+            "payload": self.payload,
+            "timestamp": self.timestamp,
+        }
+
+
+class GraphOfContractsRegistry:
+    """In-memory registry for versioned contracts/adapters."""
+
+    def __init__(self) -> None:
+        self._contracts: Dict[str, VersionedContract] = {}
+        self._adapters: Dict[str, VersionedContract] = {}
+
+    # Contracts management
+    def register_contract(self, name: str, version: str, payload: Optional[Dict[str, Any]] = None) -> VersionedContract:
+        payload = payload or {}
+        vc = VersionedContract(name=name, version=version, payload=payload)
+        key = f"{name}:{version}"
+        self._contracts[key] = vc
+        return vc
+
+    def get_contract(self, name: str, version: Optional[str] = None) -> Optional[VersionedContract]:
+        if version is None:
+            # Return the latest by timestamp if available
+            candidates = [c for k, c in self._contracts.items() if k.startswith(f"{name}:")]
+            if not candidates:
+                return None
+            return max(candidates, key=lambda c: c.timestamp)
+        return self._contracts.get(f"{name}:{version}")
+
+    def list_contracts(self) -> List[VersionedContract]:
+        return list(self._contracts.values())
+
+    # Adapters management (GoC entries)
+    def register_adapter(self, adapter_name: str, version: str, payload: Optional[Dict[str, Any]] = None) -> VersionedContract:
+        payload = payload or {}
+        ac = VersionedContract(name=adapter_name, version=version, payload=payload)
+        key = f"adapter:{adapter_name}:{version}"
+        self._adapters[key] = ac
+        return ac
+
+    def get_adapter(self, adapter_name: str, version: Optional[str] = None) -> Optional[VersionedContract]:
+        if version is None:
+            candidates = [a for k, a in self._adapters.items() if k.startswith(f"adapter:{adapter_name}:")]
+            if not candidates:
+                return None
+            return max(candidates, key=lambda c: c.timestamp)
+        return self._adapters.get(f"adapter:{adapter_name}:{version}")
+
+    def list_adapters(self) -> List[VersionedContract]:
+        return list(self._adapters.values())
+
+
+__all__ = ["GraphOfContractsRegistry", "VersionedContract"]