wandelbots-novaflow 0.1.0

Nova on Nodes

NovaFlow

The novaflow package allows you to build distributed action services for the NovaFlow behavior tree orchestration framework. It provides a simple decorator-based API for creating microservices that communicate with the control plane via CloudEvents over NATS. It exports the ActionService class, which allows you to define multiple actions per service with automatic handling of communication with control plane via CloudEvents sent over NATS, including heartbeat management and status reporting.

Define an Action Service

Create a Python file with your action definition:

from pydantic import BaseModel
from novaflow.core import ActionService

# Define input and output models
class MyInput(BaseModel):
    message: str

class MyOutput(BaseModel):
    processed: str

# Create service
service = ActionService(name="MyService", version="1.0.0")

# Define actions with input and output validation
@service.action(
    action_name="ProcessMessage",
    input_model=MyInput,
    output_model=MyOutput
)
async def process_message(params: MyInput) -> MyOutput:
    """Process a message."""
    return MyOutput(
        processed=params.message.upper(),
    )

# Run service
if __name__ == "__main__":
    service.run()

Conditions are not actions. A gate/branch in a behavior tree is a synchronous Condition node that evaluates a predicate against the blackboard — not a service decorator. See the Condition node and the State Provider framework (../docs/state-providers.md) rather than defining a boolean-returning action.

Organizing Actions with Routers

For larger services, you can organize actions into separate modules using Router:

# In actions/message_processing.py
from pydantic import BaseModel
from novaflow.core import Router

router = Router()

class MessageInput(BaseModel):
    text: str

class MessageOutput(BaseModel):
    result: str

@router.action("ProcessText", MessageInput, MessageOutput)
async def process_text(params: MessageInput) -> dict:
    return {"result": params.text.upper()}

# In run.py
from novaflow.core import ActionService
from actions import message_processing

service = ActionService(name="MessageService")
service.include_router(message_processing.router)
service.run()

Input and Output Models

Input / Output Models: Optional Pydantic BaseModel with flat fields using only supported types. Set to None if the action requires no input or output respectively.

Models are validated at service startup, as well as on each request. Action will validate incoming data against the input model and outgoing data against the output model. If the function returns data when no output model is defined, an error will be raised. This ensures strict adherence to the defined data contracts.

[!NOTE] This is similar to how FastAPI validates request and response models. However, the current approach is less flexible as it does not support automated parsing from type hints and requires explicit model definitions.

Action Function

Every action becomes an endpoint for the service. It will automatically listen to events sent on <nats-cmd-prefix>.<ServiceName>.<ActionName> subject. So the control plane can invoke the action by sending a CloudEvent to that subject. This maps directly to the tree definition like so:

{
    "version": "3.0",
    "root": {
        "id": "node-1",
        "kind": "action",
        "namespace": "wandelbots",
        "service": "<ServiceName>",
        "operation": "<ActionName>",
        "schema_version": "1.0.0",
        "data": {
            "message": { "ref": "value", "data": "Hello World" }
        }
    }
}

(data.message is wrapped as { "ref": "value", "data": … } — the field carries either an inline value or an output reference; data matches the action's input model.)

Run the Action Service

uv run your_action.py

Communication Protocol

Services communicate with the control plane using CloudEvents over NATS:

Control Plane                    Service
     |                              |
     |-----(CloudEvent Request)---->|
     |<----(RUNNING)----------------|  t=0s   (execution started)
     |<----(RUNNING)----------------|  t=2s   (heartbeat)
     |<----(RUNNING)----------------|  t=4s   (heartbeat)
     |<----(RUNNING)----------------|  t=6s   (heartbeat)
     |                              | ... function still executing ...
     |<----(RUNNING)----------------|  t=8s   (heartbeat)
     |<----(SUCCESS)----------------|  t=9s   (completed!)
     | (return response)            |

Message Flow

1. Request: Control plane sends CloudEvent with action data to svc.cmd.{ServiceName}.{ActionName}
2. RUNNING: Service immediately responds with status and sends heartbeat updates every 2s
3. SUCCESS/FAILED: Service sends final status with result data or error message

For the complete NATS communication architecture, see ADR 017.

Development & Testing

Running tests:

uv run pytest tests/

Halt Semantics

Actions run under a one-verb halt model: when a run is halted (user- initiated stop, parent BT preemption, etc.), the SDK fires a HaltToken your handler can observe, then escalates to SIGKILL on deadline if the handler doesn't return cooperatively. There is no soft/hard mode — one cooperative grace path, one supervisor floor.

from novaflow.core.halt import HaltToken

@router.action("Move", MoveInput, MoveOutput)
async def move(input: MoveInput, halt: HaltToken) -> MoveOutput:
    # `halt.is_halted` polls cheaply; `halt.wait_for_halt(timeout=...)`
    # races against the halt event; `halt.on_halt(callback)` runs a
    # cleanup when halt fires.
    async with my_db.connect() as db:
        return await analyse(db, halt=halt)

For resources that have a context manager (async with), the language-level cleanup is the right pattern — __aexit__ runs on halt-induced CancelledError. For ordering-sensitive cleanup or legacy libraries without context managers, halt.on_halt(close_fn) registers a cleanup callback. See ADR 005 — Action Halt.

Note on physical halt. Motion actions built on the Nova SDK (via CancellableMovementController) halt with confirmed standstill — the controller surfaces "standstill confirmed" on the wire's message field. This is a property of how the handler is built; an action that bypasses Nova does not get it. The supervisor SIGKILL floor still terminates the process regardless. status="halted" is a controlled-stop signal, not a safety-rated stop — safety-rated stops remain the hardware E-stop circuit's responsibility. Conversely, when a motion action observes a hardware safety event on the controller (e.g. via Nova's stream_robot_controller_state), the run terminates as status="failed", not "halted" — the safety event is an external interruption from the stack's POV, not a halt the stack initiated. See docs/terminology.md and issue #337.

Developer CLI (nf)

The nf command is the single entry point for novaflow developer tooling. Installing this package puts nf on your PATH (declared as a [project.scripts] entry in pyproject.toml):

uv run nf                       # top-level help
uv run nf actions               # action-author commands
uv run nf actions lint          # halt-cleanup heuristic check
uv run nf actions conformance   # halt-conformance harness driver (stub)

Subcommands follow git-style nesting: nf <group> <command>. The actions group hosts every command an action author needs; future groups (nf trees …, nf services …) will follow the same pattern.

Halt-cleanup Lint (advisory)

novaflow.lint.halt is a heuristic AST scan that flags action handlers that acquire a resource (a httpx client, a db pool, a websocket, a camera handle, …) without a paired cleanup — no async with, no .close(), no halt.on_halt, not returned. Run it against your action package:

uv run nf actions lint                  # defaults to src/actions/
uv run nf actions lint path/to/handlers # override

The rule is warning-only (~70% true positive / ~30% false positive based on a real-codebase scan), exit code is always 0. It's a developer aid, not a CI gate. The conformance harness (below) is the authoritative safety check.

Halt-conformance Testing

novaflow.conformance is a halt-conformance harness that verifies your action handlers actually release their resources on halt — by running them against a real (testcontainer) backend and observing the server side, not just trusting the in-process bookkeeping.

The pattern (also see actions/builtin/balance/tests/test_halt_conformance.py):

# tests/test_halt_conformance.py
import pytest
from novaflow.conformance import HaltConformanceHarness
from novaflow.conformance.adapters.httpx import HttpxAdapter
from novaflow.conformance.harness import ConformanceCase
from novaflow.core.halt import HaltToken

from src.actions.my_action import my_handler, MyInput


@pytest.mark.asyncio
async def test_my_action_halt_conformance():
    async def handler(halt: HaltToken, fixture: dict):
        try:
            await my_handler(MyInput(base_url=fixture["base_url"]), halt=halt)
        except Exception:
            pass

    async with HttpxAdapter(drain_timeout_s=2.0) as adapter:
        harness = HaltConformanceHarness("MyAction", adapter)
        # Case 1: handler completes before halt fires (clean path).
        harness.add_case(ConformanceCase("clean", handler, halt_after_ms=2000))
        # Case 2: halt fires mid-request — handler must still release.
        # 100 ms (not 10 ms) — macOS' localhost TCP handshake plus
        # first-byte takes ~20–30 ms in pytest-asyncio's loop, so a
        # 10 ms halt sometimes fires before the request leaves the
        # client. 100 ms reliably exercises the in-flight cancel.
        harness.add_case(ConformanceCase("halt_mid", handler, halt_after_ms=100))
        result = await harness.run()
        assert result.all_passed, result.describe()

Run via:

uv run pytest tests/test_halt_conformance.py

Built-in adapters:

Adapter	Use when your action talks to…	Real backend
HttpxAdapter	HTTP APIs via httpx.AsyncClient	in-process asyncio HTTP server with TCP-level peer-close detection
SyntheticAdapter	a fake resource (testing the harness itself)	in-memory pool

For a custom backend (postgres, gRPC, MQTT, vendor-specific client), implement the small Adapter protocol — see docs/halt-conformance.md for the full guide and novaflow/src/novaflow/conformance/adapters/synthetic.py as the reference.

The harness is a dev-side quality bar, not a runtime gate — the worker pool's SIGKILL floor enforces termination regardless. A non-conforming action still runs; it just gets a "no halt-conformance attestation" badge in the operator UI.