Metadata-Version: 2.4
Name: wandelbots-novaflow
Version: 0.1.0
Summary: Nova on Nodes
Requires-Python: >=3.11
Requires-Dist: faststream[nats]==0.6.7
Requires-Dist: httpx>=0.27.0
Requires-Dist: loguru==0.7.3
Requires-Dist: novaflow-types==0.1.0
Requires-Dist: pydantic-settings==2.11.0
Requires-Dist: python-dotenv==1.2.2
Requires-Dist: python-statemachine<4,>=3.2.0
Description-Content-Type: text/markdown

# 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:

```python
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`:

```python
# 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](../novaflow-schemas/python/src/novaflow/types/data/__init__.py). 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:

```json
{
    "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

```bash
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](../docs/adr/017-action-communication-protocol.md).

### Development & Testing

Running tests:

```bash
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.

```python
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](../docs/adr/005-action-halt.md).

> **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`](../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`):

```bash
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:

```bash
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`):

```python
# 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:

```bash
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](../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.
