The action protocol

An action is a program that turns one input item into zero or more output items. Millstream starts it, hands it items one at a time over stdin, and reads its answers on stdout, one JSON object per line. That is the whole contract: an action needs no Millstream library, no network access to the engine and no knowledge of the store.

Transport

The engine writes to the action's stdin and reads its stdout. Nothing else goes to stdout.
stderr is the action's log. The node re-emits each line with the run, step, item, delivery and attempt.
Files go through a working directory, MILLSTREAM_WORKDIR, never through the protocol.
One JSON object per line, UTF-8, at most 8 MiB. Receivers ignore keys they do not know, so the protocol can grow without breaking actions.

A conversation

{"type":"hello","protocol":1,"action":"lab/score-listing","version":"2.1.0","config":{"minRooms":2},"vars":{"city":"Lyon"}}
{"type":"ready"}
{"type":"item","delivery":"dl_3f…","id":"it_9a…","key":"listing-42","payload":{"url":"https://example.com/42"},"attempt":1,"idempotencyKey":"7e2b…"}
{"type":"metric","delivery":"dl_3f…","name":"pages_fetched","value":12}
{"type":"emit","delivery":"dl_3f…","key":"listing-42","payload":{"score":81}}
{"type":"done","delivery":"dl_3f…"}
{"type":"shutdown"}

Message	From	Means
`hello`	engine	The action, its config already interpolated for the run, the run's variables, and lessons for agent steps
`ready`	action	Prepared; answered within 10 seconds
`item`	engine	One delivery: the item's id, key, payload, artifacts, run, attempt and idempotency key
`emit`	action	Stages an output; outputs publish only when the delivery is done
`metric`	action	A custom measurement, recorded with the attempt
`done`	action	The delivery succeeded
`error`	action	It failed, `retryable` or not
`abstain`	action	Unsure of this item: hand it to the step's fallback
`shutdown`	engine	Exit with status 0

A process handles one delivery at a time, and concurrency comes from running several. It may serve item after item (the default for code and containers, which amortises start-up) or one item each (the default for agents, so no context leaks between items).

Retries and idempotency

Outputs are staged by emit and published only on done, so a delivery that fails, crashes or times out publishes nothing, and a retry starts clean. Retryable errors, crashes and timeouts are retried with exponential backoff and jitter; a non-retryable error dead-letters the item at once.

Delivery is at least once. idempotencyKey, a hash of the step, the item and its generation, is the same on every attempt and changes on replay. An action that changes something outside Millstream declares effects: external and passes the key to the system it calls, so a retried request is applied once.

The Go SDK

The SDK reads items, decodes each payload into your Go type, calls your function and writes the answers.

package main

import "github.com/monolithiclab/millstream/sdk/go"

func main() {
    millstream.Map(func(ctx millstream.Context, in Listing) ([]Priced, error) {
        if in.SurfaceM2 == 0 {
            return nil, ctx.Abstain("no surface area")
        }
        ctx.Metric("works_estimate_eur", float64(estimate(in)), nil)
        return []Priced{price(in, ctx.Config())}, nil
    })
}

millstream.MapKeyed emits several keyed outputs per item. millstream init action <name> --lang raw writes a Python action with the standard library only, a starting point for any other language.

Conformance

millstream test action --conformance checks an action against the protocol, whatever it computes: ready in time, unknown keys tolerated, every line well framed, every item answered, several outputs keyed distinctly, silence between deliveries, a malformed config answered per item instead of crashing, and a clean exit on shutdown.

millstream test action ./actions/rentability --conformance

PASS ready: ready in 4ms
...
PASS shutdown: exited with status 0
listings/greet: conformant

Fixture tests compare an action's outputs with golden ones, and run in CI like any unit test:

millstream test action ./actions/rentability

Read the quick start Pipelines Agents