From ff8ab1f3211cbf0d4bb90e4362c218a8320a57eb Mon Sep 17 00:00:00 2001 From: agent-3856f9f6ac3cbf34 Date: Wed, 15 Apr 2026 01:43:58 +0200 Subject: [PATCH] build(agent): molt-a#3856f9 iteration --- README.md | 6 ++++ nova_plan/contracts.py | 12 ++++++++ nova_plan/dsl.py | 62 ++++++++++++++++++++++++++++++++++++++++++ nova_plan/planner.py | 23 ++++++++++++++++ 4 files changed, 103 insertions(+) create mode 100644 nova_plan/dsl.py diff --git a/README.md b/README.md index 2501628..045ca03 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,12 @@ How to run tests - Run: `./test.sh` - This will execute unit tests and verify packaging with `python -m build`. +MVP extensions (planned, small scope) +- Add a minimal DSL (nova_plan.dsl) and a delta-synchronization helper (delta_sync) to scaffold federated optimization flows. +- Introduce a lightweight ContractRegistry to enable versioning and interoperability of data contracts. +- Provide a tiny DSL-to-LocalProblem translator to bootstrap CatOpt-style integration with adapters. +- Expand ledger anchoring with a deterministic reconciliation log for outages. + Directory layout - nova_plan/ Core MVP implementation (planner, contracts, ledger, adapters) - tests/ Unit tests for core workflow and contracts diff --git a/nova_plan/contracts.py b/nova_plan/contracts.py index 854bb5d..ca68809 100644 --- a/nova_plan/contracts.py +++ b/nova_plan/contracts.py @@ -47,3 +47,15 @@ def serialize(obj: object) -> str: if hasattr(obj, "__dict__"): return json.dumps(obj.__dict__) return json.dumps(obj) + +# Lightweight contract registry for versioning and interoperability +class ContractRegistry: + _registry: Dict[str, int] = {} + + @classmethod + def register(cls, name: str, version: int) -> None: + cls._registry[name] = int(version) + + @classmethod + def version_of(cls, name: str, default: int | None = None) -> int | None: + return cls._registry.get(name, default) diff --git a/nova_plan/dsl.py b/nova_plan/dsl.py new file mode 100644 index 0000000..2d62561 --- /dev/null +++ b/nova_plan/dsl.py @@ -0,0 +1,62 @@ +"""Minimal NovaPlan DSL scaffolding. + +This module provides a tiny domain-specific language (DSL) abstraction +to express a local planning problem in a serialized form, which can be +translated into the existing LocalProblem used by the MVP tests. + +The DSL is intentionally lightweight and deterministic to keep tests fast +and comprehensible. It is not meant to be a full-featured language yet. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Dict, Any, Callable, Optional + +import time +from nova_plan.planner import LocalProblem +from nova_plan.contracts import PlanDelta + + +@dataclass +class LocalProblemDSL: + """A minimal serializable description of a local problem. + + This captures a simple objective expressed as a Python expression string + that will be evaluated in a safe environment at runtime. The expression + should reference `vars` for local variables and `shared` for shared ones, + for example: "(vars['x'] - shared['x'])**2". + """ + + agent_id: str + objective_expr: str + variables: Dict[str, float] + constraints: Optional[Dict[str, Any]] = None + + def to_local_problem(self) -> LocalProblem: + """Convert to a LocalProblem instance. + + The evaluation environment is intentionally restricted to avoid + executing arbitrary code. Support for richer DSLs can be added later. + """ + + def _objective(local_vars: Dict[str, float], shared: Dict[str, float]) -> float: + # Provide a lightweight evaluation context. + ctx = { + "vars": local_vars, + "shared": shared, + } + # Intentionally restrict builtins to avoid dangerous calls. + safe_builtins = {"min": min, "max": max, "abs": abs} + return float(eval(self.objective_expr, {"__builtins__": safe_builtins}, ctx)) + + return LocalProblem( + id=self.agent_id, + objective=_objective, + variables=self.variables.copy(), + constraints=(self.constraints or {}), + ) + + def to_plan_delta(self, delta: Dict[str, float], timestamp: Optional[float] = None) -> PlanDelta: + """Create a PlanDelta from this DSL with a given delta map.""" + import time as _time + return PlanDelta(agent_id=self.agent_id, delta=delta, timestamp=timestamp or _time.time()) diff --git a/nova_plan/planner.py b/nova_plan/planner.py index 9dad103..81129d8 100644 --- a/nova_plan/planner.py +++ b/nova_plan/planner.py @@ -6,6 +6,8 @@ is intentionally small and deterministic to enable unit tests and demonstrations """ from __future__ import annotations from typing import Dict, Any +import time +from nova_plan.contracts import PlanDelta class LocalProblem: """A tiny local optimization problem. @@ -39,3 +41,24 @@ def simple_admm_step(local: LocalProblem, shared_vars: Dict[str, float], rho: fl new_vars[k] = v + rho * (s - v) local.variables = new_vars return new_vars + +def delta_sync(local: LocalProblem, shared_vars: Dict[str, float], agent_id: str | None = None, rho: float = 1.0) -> PlanDelta: + """Create a PlanDelta representing the current local delta with shared_vars. + + This is a lightweight helper to bootstrap delta exchange between agents + in an ADMM-like loop. It mirrors the minimal API used by tests and MVP + demos, returning a serializable PlanDelta. + """ + delta = {} + for k, v in local.variables.items(): + s = shared_vars.get(k, 0.0) + delta[k] = v - s + # Optionally, we could apply a small adjustment here using rho if desired. + local_val = local.variables.get(k, 0.0) + shared_val = shared_vars.get(k, 0.0) + # Apply a tiny step toward the shared value to reflect a one-pass update. + local.variables[k] = local_val + rho * (shared_val - local_val) + + # Build PlanDelta for governance/traceability + ts = time.time() + return PlanDelta(agent_id=agent_id or getattr(local, "id", "unknown"), delta=delta, timestamp=ts)