build(agent): new-agents#a6e6ec iteration

catopt_query/contracts.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, asdict
+from typing import Dict, Any, List, Optional
+
+from .protocol import LocalProblem, CanonicalPlan
+
+
+@dataclass(frozen=True)
+class LocalProblemDSL:
+    # Minimal domain-specific language for a per-shard problem
+    id: str
+    domain: str = "default"
+    assets: List[str] = None
+    objective: str = ""
+    constraints: Dict[str, Any] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> "LocalProblemDSL":
+        if d is None:
+            d = {}
+        return LocalProblemDSL(
+            id=d.get("id", ""),
+            domain=d.get("domain", "default"),
+            assets=d.get("assets", []),
+            objective=d.get("objective", ""),
+            constraints=d.get("constraints", {}),
+        )
+
+
+@dataclass(frozen=True)
+class SharedVariablesDSL:
+    version: int
+    signals: Dict[str, float]
+    priors: Dict[str, float]
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> "SharedVariablesDSL":
+        return SharedVariablesDSL(
+            version=int(d.get("version", 0)),
+            signals=dict(d.get("signals", {})),
+            priors=dict(d.get("priors", {})),
+        )
+
+
+@dataclass(frozen=True)
+class PlanDeltaDSL:
+    delta_id: str
+    timestamp: float
+    changes: Dict[str, Any]
+    contract_id: str = ""
+
+    def to_dict(self) -> Dict[str, Any]:
+        return asdict(self)
+
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> "PlanDeltaDSL":
+        return PlanDeltaDSL(
+            delta_id=d.get("delta_id", ""),
+            timestamp=float(d.get("timestamp", 0.0)),
+            changes=dict(d.get("changes", {})),
+            contract_id=d.get("contract_id", ""),
+        )
+
+
+@dataclass(frozen=True)
+class DualVariables:
+    multipliers: Dict[str, float] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {"multipliers": self.multipliers or {}}
+
+    @staticmethod
+    def from_dict(d: Dict[str, Any]) -> "DualVariables":
+        return DualVariables(multipliers=dict(d.get("multipliers", {})))
+
+
+class GraphOfContractsRegistry:
+    """Lightweight in-memory registry for Graph-of-Contracts metadata.
+
+    This is intentionally small and dependency-free to bootstrap adapters
+    and schemas across domains. It is not a persistent store; tests can
+    extend it if needed.
+    """
+
+    _registry: Dict[str, Dict[str, Any]] = {}
+
+    @classmethod
+    def register_contract(cls, adapter_name: str, version: str, schema: Dict[str, Any]) -> None:
+        cls._registry[adapter_name] = {
+            "version": version,
+            "schema": schema,
+        }
+
+    @classmethod
+    def get_contract(cls, adapter_name: str) -> Optional[Dict[str, Any]]:
+        return cls._registry.get(adapter_name)
+
+    @classmethod
+    def all_contracts(cls) -> Dict[str, Dict[str, Any]]:
+        return dict(cls._registry)
+
+
+def to_canonical_from_dsl(local_dsl: LocalProblemDSL) -> CanonicalPlan:
+    """Translate a minimal LocalProblemDSL into the canonical CanonicalPlan.
+
+    This is a pragmatic bridge that preserves existing canonical representations
+    while enabling a DSL-driven composition workflow.
+    """
+    # Translate DSL into a LocalProblem object using only lightweight fields
+    lp = LocalProblem(
+        shard_id=local_dsl.id,
+        projection=local_dsl.assets or [],
+        predicates=[local_dsl.objective] if local_dsl.objective else [],
+        costs=0.0,
+        constraints=local_dsl.constraints or {},
+    )
+
+    return CanonicalPlan(projection=lp.projection, predicates=lp.predicates, estimated_cost=0.0)
+
+
+__all__ = [
+    "LocalProblemDSL",
+    "SharedVariablesDSL",
+    "PlanDeltaDSL",
+    "DualVariables",
+    "GraphOfContractsRegistry",
+    "to_canonical_from_dsl",
+]