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

.gitignore

@@ -0,0 +1,21 @@
+node_modules/
+.npmrc
+.env
+.env.*
+__tests__/
+coverage/
+.nyc_output/
+dist/
+build/
+.cache/
+*.log
+.DS_Store
+tmp/
+.tmp/
+__pycache__/
+*.pyc
+.venv/
+venv/
+*.egg-info/
+.pytest_cache/
+READY_TO_PUBLISH

AGENTS.md

@@ -0,0 +1,15 @@
+# NeuPlan Architecture and Guidelines
+
+- Purpose
+  - Provide a production-ready scaffold for a neuromorphic planning stack, enabling end-to-end wiring from a lightweight DSL to a neuromorphic backend.
+- Tech Stack
+  - Language: Python 3.9+
+  - Packaging: pyproject.toml with setuptools
+  - Core modules: neuplan.dsl (LocalProblem, PlanDelta, SharedVariables, to_nir), neuplan.runtime (OnboardRuntime), neuplan.backends.loihi (LoihiBackend)
+- Testing
+  - test.sh should execute: python -m build, pytest
+- Testing commands
+  - Build: python3 -m build
+  - Run tests: pytest -q
+- Contribution
+  - Create a feature branch, implement, run tests, and open a PR. Include README updates and READY_TO_PUBLISH if ready.

README.md

@@ -1,3 +1,18 @@
-# neuplan-neuromorphic-compiler-runtime-fo
+# NeuPlan MVP
 
-Problem: Space robotics fleets require autonomous planning with strict energy budgets and intermittent connectivity. Conventional CPU-based planners struggle to deliver low-latency, energy-efficient inference for onboard AGI reasoning, while external
+This repository contains a minimal, production-oriented scaffold for a neuromorphic planning
+stack intended for onboard autonomous planning in space robotics contexts. The MVP focuses on
+providing a DSL, a toy neuromorphic intermediate representation (N-IR), a backend shim, and a
+deterministic onboard runtime that can be extended to real neuromorphic hardware.
+
+Highlights
+- DSL for LocalProblem, PlanDelta, and SharedVariables
+- Toy translation to N-IR suitable for testing and integration
+- Loihi-like backend shim with deterministic latency/energy model
+- Onboard runtime to execute planning within strict budgets
+- Basic tests ensuring end-to-end flow and a minimal demo CLI
+
+Installation and testing
+- Run: ./test.sh (requires Python 3.9+ and build tools; creates a virtual environment automatically during build)
+
+This is an early-stage MVP; expect to see iterative improvements with governance, safety, and HIL features in subsequent iterations.

neuplan/__init__.py

@@ -0,0 +1,12 @@
+"""NeuPlan: toy neuromorphic planning stack scaffold.
+
+This package provides a minimal, production-oriented scaffold for a
+NeuPlan MVP: DSL types, a neuromorphic IR translator, a backend shim,
+and an onboard runtime that can be wired to actual neuromorphic hardware.
+"""
+
+from . import dsl
+from . import runtime
+from . import backends
+
+__all__ = ["dsl", "runtime", "backends"]

neuplan/backends/__init__.py

@@ -0,0 +1,3 @@
+from .loihi import LoihiBackend
+
+__all__ = ["LoihiBackend"]

neuplan/backends/loihi.py

@@ -0,0 +1,29 @@
+"""Toy Loihi-like backend for NeuPlan MVP.
+
+This is a lightweight simulator that pretends to run a neuromorphic graph on
+Loihi-like hardware. It converts a NeuPlan NIR (toy dict) into a latency/energy
+estimate and returns a simple executable plan skeleton.
+"""
+from __future__ import annotations
+
+from typing import Dict, Any, List
+
+
+class LoihiBackend:
+    def __init__(self, quantization_bits: int = 8, time_scale: float = 1.0) -> None:
+        self.quantization_bits = quantization_bits
+        self.time_scale = time_scale
+
+    def run(self, nir: Dict[str, Any], time_budget_s: float) -> Dict[str, Any]:
+        nodes: List[Dict[str, Any]] = nir.get("nodes", [])
+        # naive latency model: 5ms per node scaled by time_scale
+        latency = max(0.001, len(nodes) * 0.005 / max(1e-6, self.time_scale))
+        status = "ok" if latency <= time_budget_s else "timeout"
+        energy = max(0.01, len(nodes) * 0.02)
+        plan = {"steps": [{"node": n} for n in nodes]}
+        return {
+            "status": status,
+            "latency_s": latency,
+            "energy_j": energy,
+            "plan": plan,
+        }

neuplan/cli.py

@@ -0,0 +1,28 @@
+"""Minimal command-line interface to exercise NeuPlan MVP."""
+from __future__ import annotations
+
+import json
+from typing import List
+from neuplan.dsl import LocalProblem, PlanDelta, SharedVariables
+from neuplan.backends.loihi import LoihiBackend
+from neuplan.runtime import OnboardRuntime
+from neuplan.dsl import to_nir
+
+
+def demo():
+    # Simple demo problem: two assets with a delta depending on energy
+    lp1 = LocalProblem(asset="rover1", constraints={"max_speed": 5}, objective={"energy": 1.0})
+    lp2 = LocalProblem(asset="droneA", constraints={"max_alt": 100}, objective={"energy": 0.8})
+    delta = PlanDelta(delta_id="d1", changes={"requires": ["rover1"]})
+    shared = SharedVariables(variables={"global_time": 0})
+
+    nir = to_nir([lp1, lp2], [delta], shared)
+    backend = LoihiBackend()
+    runtime = OnboardRuntime(backend)
+    res = runtime.plan([lp1, lp2], [delta], shared, time_budget_s=2.0)
+    print("NIR:", json.dumps(nir, indent=2))
+    print("Result:", json.dumps(res, indent=2))
+
+
+if __name__ == "__main__":
+    demo()

neuplan/dsl.py

@@ -0,0 +1,74 @@
+"""NeuPlan DSL: minimal LocalProblem / PlanDelta / SharedVariables model.
+
+This is a lightweight, easily testable sketch translating planning problems
+into a toy neuromorphic intermediate representation (N-IR).
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Dict, List, Any
+import time
+
+
+@dataclass
+class LocalProblem:
+    asset: str
+    constraints: Dict[str, Any] = field(default_factory=dict)
+    objective: Dict[str, float] = field(default_factory=dict)
+
+
+@dataclass
+class PlanDelta:
+    delta_id: str
+    changes: Dict[str, Any] = field(default_factory=dict)
+    timestamp: float = field(default_factory=lambda: time.time())
+
+
+@dataclass
+class SharedVariables:
+    variables: Dict[str, Any] = field(default_factory=dict)
+
+
+def to_nir(local_problems: List[LocalProblem], deltas: List[PlanDelta], shared: SharedVariables) -> Dict[str, Any]:
+    """Translate a set of problems/deltas into a toy neuromorphic IR.
+
+    The real project would generate a graph of spiking neurons with temporal
+    dynamics encoding constraints; here we emit a deterministic, testable
+    toy representation for MVP validation and integration testing.
+    """
+    nodes: List[Dict[str, Any]] = []
+    edges: List[Dict[str, Any]] = []
+
+    # Create nodes for LocalProblems
+    for idx, lp in enumerate(local_problems):
+        n = {
+            "id": f"LP:{idx}:{lp.asset}",
+            "type": "LocalProblem",
+            "asset": lp.asset,
+            "constraints": lp.constraints,
+            "objective": lp.objective,
+        }
+        nodes.append(n)
+
+    # Create nodes for each PlanDelta
+    for d in deltas:
+        n = {
+            "id": f"DELTA:{d.delta_id}",
+            "type": "PlanDelta",
+            "delta_id": d.delta_id,
+            "changes": d.changes,
+            "timestamp": d.timestamp,
+        }
+        nodes.append(n)
+
+    # Shared variables as a single hub node if present
+    if shared and shared.variables:
+        nodes.append({"id": "SharedVariables:root", "type": "SharedVariables", "payload": shared.variables})
+
+    # Simplified edges: connect LocalProblems to Delta changes if named in constraints
+    for lp in local_problems:
+        for d in deltas:
+            if lp.asset in (d.changes.get("requires", []) or []):
+                edges.append({"src": f"LP:{local_problems.index(lp)}:{lp.asset}", "dst": f"DELTA:{d.delta_id}"})
+
+    return {"nodes": nodes, "edges": edges}

neuplan/runtime.py

@@ -0,0 +1,21 @@
+"""OnboardRuntime: minimal deterministic planner runtime scaffold."""
+from __future__ import annotations
+
+from typing import List, Dict, Any
+from time import time
+
+from .dsl import LocalProblem, PlanDelta, SharedVariables, to_nir
+
+
+class OnboardRuntime:
+    def __init__(self, backend) -> None:
+        # backend expected to implement run(nir: dict, time_budget_s: float) -> dict
+        self.backend = backend
+
+    def plan(self, local_problems: List[LocalProblem], deltas: List[PlanDelta], shared: SharedVariables, time_budget_s: float) -> Dict[str, Any]:
+        nir = to_nir(local_problems, deltas, shared)
+        result = self.backend.run(nir, time_budget_s)
+        # Attach some provenance-like fields for auditing in MVP
+        result["timestamp"] = time()
+        result["nir"] = nir
+        return result

pyproject.toml

@@ -0,0 +1,16 @@
+[build-system]
+requires = ["setuptools>=61.0", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "neuplan"
+version = "0.1.0"
+description = "Prototype NeuPlan: neuromorphic planning stack (toy)"
+readme = "README.md"
+requires-python = ">=3.9"
+
+[project.urls]
+Homepage = "https://example.com/neuplan"
+
+[tool.setuptools.packages.find]
+where = ["neuplan"]

test.sh

@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+echo "Running Python build to verify packaging metadata..."
+python3 -m build
+
+echo "Running tests..."
+pytest -q
+
+echo "Tests completed successfully."

tests/test_neuplan.py

@@ -0,0 +1,29 @@
+import json
+import sys
+import os
+# Ensure local package path is discoverable when running from the repo root
+repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+if repo_root not in sys.path:
+    sys.path.insert(0, repo_root)
+
+from neuplan.dsl import LocalProblem, PlanDelta, SharedVariables
+from neuplan.dsl import to_nir
+from neuplan.backends.loihi import LoihiBackend
+from neuplan.runtime import OnboardRuntime
+
+
+def test_nir_generation_and_runtime_plan():
+    lp1 = LocalProblem(asset="rover1", constraints={"max_speed": 5}, objective={"energy": 1.0})
+    lp2 = LocalProblem(asset="droneA", constraints={"max_alt": 100}, objective={"energy": 0.8})
+    delta = PlanDelta(delta_id="d1", changes={"requires": ["rover1"]})
+    shared = SharedVariables(variables={"global_time": 0})
+
+    nir = to_nir([lp1, lp2], [delta], shared)
+    assert isinstance(nir, dict)
+    assert "nodes" in nir and "edges" in nir
+
+    backend = LoihiBackend()
+    runtime = OnboardRuntime(backend)
+    res = runtime.plan([lp1, lp2], [delta], shared, time_budget_s=2.0)
+    assert res["status"] in {"ok", "timeout"}
+    assert "plan" in res