app/docs/code/nodes.md

# Writing node code

A Function node is a Python file. That is all it is — no base class, no
decorator, no framework import unless you want one.

```python
def process(temperature, setpoint=21.0):
    """Ask for heat when the room is below the comfort point."""
    return {"heat": temperature < setpoint}
```

## The rules

**One function called `process`.** If the file defines exactly one public
function under another name, that one is used instead. Two, and the node
refuses to load rather than guessing.

**Arguments come from ports and settings, by name.** `temperature` above is an
input port; `setpoint` is a setting typed into the node's panel. Both arrive as
keyword arguments, which is why a setting may not share a name with a port.
See [Where a node's values come from](../concepts/values.md).

**The return value is a dict keyed by output ports.** Every value is checked
against the port's declared type before it is published. A key that is not a
declared port is an error, not a silent drop — nothing leaves a node except
through a port it declared.

**Nothing else is importable from the engine.** Node code runs in a separate
process, on a separate interpreter, with none of Fluksio's own modules on its
path. What it can import is what the [Modules](../interface/operations.md)
screen installed — which can include [your own project](#your-own-code-as-a-package),
so a node need not be a self-contained file.

**A node is a pure function of its inputs.** No context object, no global
store, no handle to reach for. A running total or a debounce timer has a
specific shape — see [Keeping state in a flow](../concepts/state.md).

## Producing values over time

A node that produces values *during* its execution is a generator. Every
`yield` is a dict keyed by output port, published the instant it happens:

```python
def process(lr, steps):
    loss = 1.0
    for _ in range(steps):
        loss = train_one_step(lr)
        yield {"loss": loss}          # published now
    return {"final_loss": loss}
```

Whatever the generator `return`s at the end is the node's result — what
downstream nodes read. If you never `return`, the last thing you yield is the
result instead.

Mark the port so the flow says what it does:

```json
{"name": "loss", "dtype": "float", "stream": true}
```

Two consequences. In a [run](../concepts/runs.md), the whole series is kept as
that run's metrics — this is why there is no `log_metric()` anywhere in the
API. And **the node's timeout starts measuring silence rather than duration**:
each emission resets the deadline, so a node yielding every few seconds can run
for hours under a timeout of 300.

### `fluksio.emit`

Where a `yield` cannot reach — the value comes from inside somebody else's
callback, and they call you rather than the other way round:

```python
import fluksio


def process():
    model.fit(callbacks=[LambdaCallback(
        on_epoch_end=lambda epoch, logs: fluksio.emit(loss=logs["loss"])
    )])
    return {"weights": ...}
```

Same ports, same type checking, same publication. Prefer `yield` where you can
reach it; `emit` where you cannot.

## Bytes: artifacts

Messages are JSON, which is what lets the same value pass through Redis, the
work queue and the worker protocol unchanged. A checkpoint is not that.

```python
import fluksio


def process(dataset):
    path = fluksio.load_artifact(dataset)     # → a local path to read
    ...
    return {
        "weights": fluksio.save_artifact("model.pt", media_type="application/octet-stream"),
        "score": 0.94,
    }
```

`save_artifact` takes bytes or a path, stores them by their SHA-256 digest, and
returns a small reference — digest, size, media type, name — which is what an
`artifact`-typed port carries.

Because the address is the content's hash, a sweep whose fifty configs share
one preprocessed input stores it once, and a reference stays valid wherever the
store is reachable from — including on another machine.

## Printing

`print` works and is captured. The first 16 KB per call is kept and shown in
the flow editor's log panel and on the run's per-node record; the rest is
dropped, so a node printing in a loop cannot fill anything up.

Use it to debug. Do not use it to record results — a number worth keeping is an
output port, not a line of text.

## Errors

An exception fails that node's execution, not the flow. The message you see is
one line from the frame in *your* code, not a stack through the engine — that
is a deliberate choice about what is actionable.

The node keeps its last error visible after it recovers, so a failure that
fired an alert at 03:00 still says what it was at 09:00. It can also be
acknowledged from the canvas.

## Timeouts

`timeout` on a node is how many seconds its code may run before it is stopped.
The default is 30, and it covers the *first* call's imports, which can be much
slower than the body — a node importing torch is not being slow, it is loading.

Above 60 seconds, a live flow may deliver the same work again while the node is
still running. In a batch run, which never redelivers, it is an idle timeout
instead: silence this long is a kill.

## Running a node somewhere else

A node declares the label of the machine it needs:

```json
{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}
```

`require` (the default) waits for a worker carrying that label; `prefer` runs
locally when none is attached. A node bound to a device is compiled *on that
machine* — a node importing `torch` is correct on the GPU box and a missing
module on the engine, so checking it here would fail something that is fine.

See [Remote workers](workers.md).

## Sharing code between flows

A node's source can be promoted to the shared library from its panel, and other
flows can then use it by reference. One copy, one place to edit — and every
flow using it runs the edit, which is the point and also the caution.

Shared sources live in `_lib/` in the flow repository, so they are versioned
with everything else.

## Packages

Node code runs in a virtual environment of its own, on the installation's data
volume — deliberately separate from the one Fluksio itself runs on.

Declare what you import in [Modules](../interface/operations.md), or over the
API:

```sh
curl -X POST $FLUKSIO/modules/apply -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"requirements": "numpy>=2\npandas\n"}'
```

It is a pip manifest installed with `uv pip sync`, versioned alongside your
flows. An install takes effect immediately; nothing restarts.

### Your own code as a package

A manifest line can name a directory, so the project you already have is
installable like any other dependency:

```text
-e /home/you/my-research
numpy>=2
```

A node body then imports it, and the logic stays where it already lives — in
your repository, under your own version control, importing its own siblings:

```python
from myresearch.train import fit


def process(lr, epochs):
    return fit(lr, epochs)
```

That is the whole of it. The node is three lines, `myresearch` can be as many
modules as it likes, and nothing was copied.

!!! warning "Editable, but not live"

    `-e` means edits reach the venv without reinstalling — but a node's process
    already holds the imported module in memory. The engine's workers are
    long-lived, so a change to your code is picked up when they are retired,
    which is what **Apply** does. Pressing it after an edit is the loop.

    A [worker](workers.md) you attach yourself is the exception: it starts a
    process per call, so it reads your code fresh every run. If you are
    iterating on the code many times an hour, point one at your own
    interpreter — `fluksio-worker --python "$(which python)"` — and mark the
    node with its label.

!!! note "The path is a deployment detail"

    It is resolved on whichever machine runs the node, and the manifest is
    committed to the flow repository — so an absolute path from your laptop
    means nothing inside a container or on a GPU box. Those need their own
    install of the same project; a VCS requirement
    (`myresearch @ git+ssh://…@a1b2c3d`) travels where a path does not.

## A worked example

The repository ships a small supervised fit as a seedable demo — three nodes,
a batch flow, streaming metrics, artifacts between stages, and a GPU-labelled
node that falls back to the engine when no worker is attached. It is the
shortest complete thing to read:

```text
prepare ──dataset(artifact)──▶ train ──weights(artifact)──▶ evaluate
                                 │
                                 └── loss (streaming float) ──▶ chart
```

`make seed-demo` builds it against a running stack.

## See also

- [Where a node's values come from](../concepts/values.md)
- [Keeping state in a flow](../concepts/state.md)
- [Runs: pipelines that finish](../concepts/runs.md)
- [Node types](../reference/node-types.md) — the ones you do not have to write