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

README.md

@@ -74,6 +74,11 @@ Interop Demo
 - Run: python3 -m nova_plan.examples.demo_energi_bridge_demo
 - This script is non-destructive and safe to run in a test environment; it prints a couple of canonical representations to stdout for quick inspection.
 
+DSL Seeds
+- NovaPlan provides a minimal Python-based DSL (nova_plan.dsl) to seed LocalProblem definitions and generate PlanDelta-like deltas for interoperability tests. This helps adapters bootstrap without a full production DSL.
+- The seeds map LocalProblem seeds into canonical objects via the existing CatOpt bridge and contract registries.
+- See tests/test_dsl_seed.py for example usage and basic validation.
+
 Toy interoperability seeds
 - A tiny toy contract seed (toy-lp) and associated signing helpers are provided in nova_plan/toy_contracts.py to bootstrap interoperability testing between adapters.
 - A small unit test (tests/test_toy_contracts.py) exercises contract registration and signing against a simple signer.

nova_plan/dsl.py

@@ -1,36 +1,69 @@
-"""Minimal NovaPlan DSL scaffolding for MVP.
+"""Minimal DSL seeds for NovaPlan interoperability.
 
-Provides tiny, test-friendly helpers to construct LocalProblem instances without
+This module provides a tiny Python-based DSL surface to seed LocalProblem
-requiring callers to import planner directly. This is not a full DSL, just a light
+instances and produce basic PlanDelta-like deltas, enabling adapters and tests
-wrapper to bootstrap tests and examples.
+to bootstrap interoperability workflows without requiring a full production DSL.
 """
 
 from __future__ import annotations
 
-from typing import Dict, Any
+from dataclasses import dataclass
+from typing import List, Dict, Any
+
 from .planner import LocalProblem
+from .contracts import PlanDelta
 
 
+@dataclass
 class LocalProblemDSL:
-    def __init__(self, id: str):
+    """A compact DSL representation of a LocalProblem seed.
-        self.id = id
-        self.objective = lambda vars, shared: 0.0
-        self.variables: Dict[str, float] = {}
-        self.constraints: Dict[str, Any] = {}
 
-    def with_variables(self, vars: Dict[str, float]) -> "LocalProblemDSL":
+    This is deliberately simple and focuses on interoperability wiring:
-        self.variables = dict(vars)
+    - id, domain, and assets describe the problem scope
-        return self
+    - objective is a simple string key understood by the translator
+    - constraints is a list of human-readable constraints (kept as strings)
+    """
 
-    def with_objective(self, func) -> "LocalProblemDSL":
+    id: str
-        self.objective = func
+    domain: str
-        return self
+    assets: List[str]
+    objective: str
+    constraints: List[str]
 
-    def build(self) -> LocalProblem:
+    def to_local_problem(self) -> LocalProblem:
-        return LocalProblem(self.id, self.objective, self.variables, self.constraints)
+        """Translate this DSL seed into a minimal LocalProblem instance.
+
+        The objective is translated into a lightweight, deterministic callable
+        that users can override or extend in adapters during integration tests.
+        """
+        # Simple objective placeholder: sum of all local variables (initialized to 0)
+        def objective(variables: Dict[str, float], shared_vars: Dict[str, float]) -> float:
+            # Basic heuristic: minimize the sum of local variables (toy objective)
+            return sum(float(v) for v in variables.values())
+
+        # Initialize a tiny set of local decision variables for each asset
+        variables: Dict[str, float] = {f"{a}_var": 0.0 for a in self.assets}
+        # Constraints are retained for provenance; not evaluated in this MVP seed
+        constraints = {"domain": self.domain, "constraints": self.constraints}
+
+        return LocalProblem(id=self.id, objective=objective, variables=variables, constraints=constraints)
 
 
-def make_local_problem(id: str, variables: Dict[str, float] | None = None) -> LocalProblem:
+def seed_delta_for_local_problem(
-    """Convenience factory to create a LocalProblem with given variables."""
+    lp: LocalProblem, shared_vars: Dict[str, float] | None = None, agent_id: str | None = None
-    lp = LocalProblem(id=id, objective=lambda v, s: 0.0, variables=dict(variables or {}))
+) -> PlanDelta:
-    return lp
+    """Create a minimal PlanDelta reflecting the current local problem delta.
+
+    This uses the tiny delta computation used in the MVP to bootstrap federation
+    and tests. It returns a PlanDelta that adapters can sign/propagate.
+    """
+    if shared_vars is None:
+        shared_vars = {}
+    delta = {}
+    for k, v in lp.variables.items():
+        delta[k] = v - shared_vars.get(k, 0.0)
+    ts = __import__("time").time()
+    return PlanDelta(agent_id=agent_id or getattr(lp, "id", "unknown"), delta=delta, timestamp=ts)
+
+
+__all__ = ["LocalProblemDSL", "seed_delta_for_local_problem"]

tests/test_dsl_seed.py

@@ -0,0 +1,37 @@
+import pytest
+
+from nova_plan.dsl import LocalProblemDSL, seed_delta_for_local_problem
+from nova_plan.planner import LocalProblem
+from nova_plan.contracts import PlanDelta
+
+
+def test_dsl_to_local_problem():
+    dsl = LocalProblemDSL(
+        id="LP1",
+        domain="space",
+        assets=["rover1", "habitat1"],
+        objective="min-energy",
+        constraints=["deadline<=t", "power<=Pmax"],
+    )
+    lp = dsl.to_local_problem()
+    assert isinstance(lp, LocalProblem)
+    assert lp.id == "LP1"
+    assert isinstance(lp.variables, dict)
+    assert "rover1_var" in lp.variables
+
+
+def test_dsl_delta_seed_generation():
+    # Create a minimal LocalProblem seed
+    dsl = LocalProblemDSL(
+        id="LP2",
+        domain="space",
+        assets=["rover2"],
+        objective="min-energy",
+        constraints=[],
+    )
+    lp = dsl.to_local_problem()
+    # Shared vars could be empty; use delta seed helper to generate a PlanDelta
+    delta = seed_delta_for_local_problem(lp, shared_vars={}, agent_id=lp.id)
+    assert isinstance(delta, PlanDelta)
+    assert delta.agent_id == lp.id
+    assert isinstance(delta.delta, dict)