From 4990fb65c39925c2a4239250ee6f63604e2232c7 Mon Sep 17 00:00:00 2001 From: agent-dd492b85242a98c5 Date: Sun, 19 Apr 2026 20:04:16 +0200 Subject: [PATCH] build(agent): new-agents-3#dd492b iteration --- README.md | 5 +++ nova_plan/dsl.py | 77 ++++++++++++++++++++++++++++++------------ tests/test_dsl_seed.py | 37 ++++++++++++++++++++ 3 files changed, 97 insertions(+), 22 deletions(-) create mode 100644 tests/test_dsl_seed.py diff --git a/README.md b/README.md index 6925006..14625f9 100644 --- a/README.md +++ b/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. diff --git a/nova_plan/dsl.py b/nova_plan/dsl.py index a01d111..d3492da 100644 --- a/nova_plan/dsl.py +++ b/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 -requiring callers to import planner directly. This is not a full DSL, just a light -wrapper to bootstrap tests and examples. +This module provides a tiny Python-based DSL surface to seed LocalProblem +instances and produce basic PlanDelta-like deltas, enabling adapters and tests +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): - self.id = id - self.objective = lambda vars, shared: 0.0 - self.variables: Dict[str, float] = {} - self.constraints: Dict[str, Any] = {} + """A compact DSL representation of a LocalProblem seed. - def with_variables(self, vars: Dict[str, float]) -> "LocalProblemDSL": - self.variables = dict(vars) - return self + This is deliberately simple and focuses on interoperability wiring: + - id, domain, and assets describe the problem scope + - 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": - self.objective = func - return self + id: str + domain: str + assets: List[str] + objective: str + constraints: List[str] - def build(self) -> LocalProblem: - return LocalProblem(self.id, self.objective, self.variables, self.constraints) + def to_local_problem(self) -> LocalProblem: + """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: - """Convenience factory to create a LocalProblem with given variables.""" - lp = LocalProblem(id=id, objective=lambda v, s: 0.0, variables=dict(variables or {})) - return lp +def seed_delta_for_local_problem( + lp: LocalProblem, shared_vars: Dict[str, float] | None = None, agent_id: str | None = None +) -> PlanDelta: + """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"] diff --git a/tests/test_dsl_seed.py b/tests/test_dsl_seed.py new file mode 100644 index 0000000..b849ed6 --- /dev/null +++ b/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)