From 44cec57aeaee90aa7cf5969c8f71344cff657a62 Mon Sep 17 00:00:00 2001 From: agent-db0ec53c058f1326 Date: Thu, 16 Apr 2026 23:11:30 +0200 Subject: [PATCH] build(agent): molt-z#db0ec5 iteration --- nova_plan/__init__.py | 2 +- nova_plan/dsl.py | 84 ++++++++++++++++++++++++++++++------------- 2 files changed, 60 insertions(+), 26 deletions(-) diff --git a/nova_plan/__init__.py b/nova_plan/__init__.py index 41cf4c8..a9c814e 100644 --- a/nova_plan/__init__.py +++ b/nova_plan/__init__.py @@ -1,3 +1,3 @@ """NovaPlan MVP package init""" -__all__ = ["planner", "contracts", "ledger", "adapters", "catopt_bridge"] +__all__ = ["planner", "contracts", "ledger", "adapters", "catopt_bridge", "dsl"] diff --git a/nova_plan/dsl.py b/nova_plan/dsl.py index 973a9d8..d885b3c 100644 --- a/nova_plan/dsl.py +++ b/nova_plan/dsl.py @@ -1,34 +1,68 @@ -"""Minimal NovaPlan DSL scaffold. +"""Minimal NovaPlan DSL scaffolding. -This is intentionally tiny: a single, ergonomic shim to create a LocalProblem -out of a high-level DSL-like description. It is designed for MVP experimentation -and to bootstrap adapters, not for production parsing. +This module provides lightweight dataclasses to model a canonical signal/plan +DSL that can be used by adapters and bridge components to represent local +problems, signals, and planning hypotheses in a structured way. + +The goal is to enable quick experimentation with interoperable representations +without pulling in heavy dependencies or reworking the core MVP logic. """ from __future__ import annotations -from typing import Dict, Callable - -from nova_plan.planner import LocalProblem +import json +from dataclasses import dataclass, asdict +from typing import List, Dict, Any -class LocalProblemDSL: - """Tiny DSL to describe a LocalProblem and convert to a LocalProblem instance.""" +@dataclass +class SignalNode: + id: str + metadata: Dict[str, Any] # arbitrary per-signal metadata (venue, ts, confidence, etc.) - def __init__(self, agent_id: str, objective_expr: str, variables: Dict[str, float], constraints: Dict[str, object] | None = None): - # Keep the input shape friendly for tests; the actual objective is not - # exercised in the MVP tests, so we keep a simple stub objective. - self.agent_id = agent_id - self.objective_expr = objective_expr - self.variables = variables - self.constraints = constraints or {} - - def to_local_problem(self) -> LocalProblem: - """Return a LocalProblem instance corresponding to this DSL description.""" - # Minimal stub objective: ignore expression, provide a deterministic function. - def _objective(local_vars: Dict[str, float], shared_vars: Dict[str, float]) -> float: - return 0.0 - - return LocalProblem(id=self.agent_id, objective=_objective, variables=self.variables, constraints=self.constraints) + def to_json(self) -> str: + return json.dumps(asdict(self)) -__all__ = ["LocalProblemDSL"] +@dataclass +class Edge: + src: str + dst: str + weight: float = 1.0 + + def to_json(self) -> str: + return json.dumps(asdict(self)) + + +@dataclass +class Scenario: + name: str + nodes: List[SignalNode] + edges: List[Edge] + + def to_json(self) -> str: + data = { + "name": self.name, + "nodes": [asdict(n) for n in self.nodes], + "edges": [asdict(e) for e in self.edges], + } + return json.dumps(data) + + +@dataclass +class HedgePlan: + plan_id: str + scenario: Scenario + metadata: Dict[str, Any] # additional plan-level metadata + timestamp: float + + def to_json(self) -> str: + data = { + "plan_id": self.plan_id, + "scenario": json.loads(self.scenario.to_json()), + "metadata": self.metadata, + "timestamp": self.timestamp, + } + return json.dumps(data) + + +__all__ = ["SignalNode", "Edge", "Scenario", "HedgePlan"]