Actions

Actions are named, durable task types. You register an action once, then invoke it on demand — Flo handles dispatch, retries, lease management, and dead-letter routing.

Action handlers run in user-hosted workers — long-lived processes (in Go, Python, JS, Zig, or any language that speaks the wire protocol) that pull tasks from Flo and report results. Flo owns the durable task queue, delivery, retries, and lease tracking; your worker owns the logic.

┌──────────────┐                   ┌───────────────────────┐
│  Client      │                   │  Flo Server (Shard)   │
│              │  action invoke    │                       │
│  flo action  │ ─────────────────▸│  ActionHandler        │
│  invoke X    │  ◀─ run_id ───────┤    └─ queue pending    │
└──────────────┘                   └──────────┬────────────┘
                                              │
┌──────────────┐   action_await               │
│  Worker      │ ◄────────────────────────────┘
│  (Go/Py/JS)  │   task_id + payload
│              │
│  handler()   │
│              │   action_complete(result)
│              │ ────────────────────────────▸ Flo
└──────────────┘

Quick Start

1. Register an action

flo action register send-email --timeout 60000 --max-retries 3

2. Invoke it

flo action invoke send-email '{"to":"alice@example.com","subject":"Welcome!"}'
# → Result: send-email-1

3. Check status

flo action status send-email-1
# → RUN ID          STATUS     CREATED
#   send-email-1    pending    2026-03-14 10:00:00

Nothing happens until a worker picks up the task — the run stays pending until a registered worker awaits it, then transitions to running and completed.

Core Concepts

Action

A named task type stored in Flo's action registry. Each action has:

Run

A single invocation of an action. Every invoke creates a run with a unique ID.

Worker

A long-running process that pulls pending tasks from Flo and executes them. Workers:

• Register with Flo, declaring which action types they handle
• Await tasks using long polling (blocking dequeue)
• Execute the handler, optionally extending the lease with touch
• Report completion or failure back to Flo

Workers can run anywhere — on the same machine, in a container, across the network. Multiple workers can handle the same action type for horizontal scaling.

How Actions Run

Actions are executed by external worker processes that pull tasks from Flo over the wire — the model for:

• Actions that call external APIs (payment gateways, email services)
• Long-running tasks (report generation, media processing)
• Logic that needs access to your infrastructure (databases, file systems)

The flow:

1. Invoke creates a run in pending status
2. A worker calls await (long poll) and receives the task
3. The worker runs the handler and calls complete or fail
4. Flo updates the run status and stores the result

Worker Lifecycle

┌──────────┐   register    ┌──────────┐   await     ┌──────────┐
│  Init    │ ──────────▸   │  Idle    │ ──────────▸ │ Execute  │
│          │               │ (polling)│             │          │
└──────────┘               └──────┬───┘             └────┬─────┘
                                  │                      │
                            heartbeat (30s)         complete / fail
                                  │                      │
                                  ▼                      ▼
                           ┌──────────┐          ┌──────────┐
                           │ Draining │          │  Idle    │
                           └──────────┘          └──────────┘

Register

The worker announces itself and the action types it handles:

flo worker register worker-1 process-order send-email

Registration includes:

• Worker ID — unique identifier (auto-generated if omitted)
• Task types — list of action names this worker can execute
• Max concurrency — how many tasks in parallel
• Machine ID — for grouping workers on the same host
• Metadata — arbitrary JSON for discovery

Await (Long Poll)

Workers block-wait for tasks:

flo worker await process-order --worker-id worker-1 --block 30000

When a matching pending task exists, Flo returns a task assignment:

Complete

Report successful completion with the result:

flo worker complete <task-id> --worker-id worker-1 --action process-order --result '{"status":"done"}'

Fail

Report failure, optionally requesting a retry:

# Retry — task goes back to pending
flo worker fail <task-id> --worker-id worker-1 --action process-order --error "Temporary failure" --retry
 
# Permanent failure — task is marked failed
flo worker fail <task-id> --worker-id worker-1 --action process-order --error "Invalid input"

Touch (Lease Extension)

For long-running tasks, extend the execution lease to prevent timeout:

flo worker touch <task-id> --worker-id worker-1 --action process-order --extend 30000

Heartbeat

Workers send periodic heartbeats (typically every 30 seconds) to report their current load and stay registered. If the server responds with a draining status, the worker should stop accepting new tasks and finish current ones.

Drain

Signal that the worker should finish current tasks but accept no new ones:

flo worker drain --worker-id worker-1

Labels

Actions can require specific worker capabilities using labels. When invoking an action with labels, only workers whose labels are a superset of the required labels will receive the task.

# Invoke with required labels
flo action invoke render-video '{"url":"..."}' --labels '{"gpu":true,"vram_gb":24}'

A worker with labels {"gpu":true, "vram_gb":24, "region":"us-east"} matches — it has all required keys with equal values. A worker with {"gpu":true, "vram_gb":16} does not match — vram_gb differs.

Label matching rules:

• All keys in required must exist in worker labels
• Values must be exactly equal (string, number, or boolean)
• Extra keys on the worker side are ignored
• Nested objects and arrays are not compared (flat values only)

Invocation Options

Idempotency

Use idempotency_key to prevent duplicate invocations:

flo action invoke charge-payment '{"order":"ORD-123"}' --idempotency-key order-123-charge
 
# Same key returns the existing run (no new execution)
flo action invoke charge-payment '{"order":"ORD-123"}' --idempotency-key order-123-charge
# → same run ID

Retry Behavior

When a worker reports failure with --retry:

1. The run status resets to pending
2. The attempt counter increments
3. The task goes back to the queue for the next available worker
4. Backoff delay is applied: retry_delay_ms × 2^(attempt-1)

When retries are exhausted (max_retries reached) or the worker fails without --retry:

• The run status is set to failed
• The error message and timestamps are recorded
• The run can still be queried via action status

Namespace Isolation

Actions and runs are scoped to namespaces. The same action name can exist independently in different namespaces:

flo action register send-email --namespace prod
flo action register send-email --namespace staging
 
# These create separate runs in separate registries
flo action invoke send-email '{}' --namespace prod
flo action invoke send-email '{}' --namespace staging

Persistence

Action registrations and run state are persisted to the Unified Append Log (UAL). On node restart:

• Registrations are replayed — all actions reappear in the registry
• Runs are replayed — pending and running tasks are restored
• Workers must re-register and resume polling

SDK: Building Workers

The SDKs provide a high-level ActionWorker that handles registration, polling, concurrency, heartbeats, and error recovery. You just write handler functions.

Handler Signature

Every action handler receives an ActionContext and returns result bytes:

type ActionHandler func(actx *ActionContext) (result []byte, err error)

ActionContext

The context object passed to every handler:

Methods:

Complete Examples

package main
 
import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
 
    flo "github.com/floruntime/flo-go"
)
 
type OrderRequest struct {
    OrderID    string  `json:"order_id"`
    CustomerID string  `json:"customer_id"`
    Amount     float64 `json:"amount"`
    Items      []Item  `json:"items"`
}
 
type Item struct {
    SKU      string `json:"sku"`
    Quantity int    `json:"quantity"`
}
 
func processOrder(actx *flo.ActionContext) ([]byte, error) {
    var req OrderRequest
    if err := actx.Into(&req); err != nil {
        return nil, fmt.Errorf("invalid input: %w", err)
    }
 
    log.Printf("Processing order %s ($%.2f)", req.OrderID, req.Amount)
 
    // For long-running tasks, extend the lease periodically
    for i, item := range req.Items {
        log.Printf("  Item %d/%d: %s", i+1, len(req.Items), item.SKU)
 
        // Extend lease every 3 items
        if (i+1) % 3 == 0 {
            actx.Touch(30000)
        }
    }
 
    return actx.Bytes(map[string]string{
        "order_id": req.OrderID,
        "status":   "processed",
    })
}
 
func sendEmail(actx *flo.ActionContext) ([]byte, error) {
    var input map[string]string
    actx.Into(&input)
 
    log.Printf("Sending email to %s", input["to"])
 
    return actx.Bytes(map[string]string{"status": "sent"})
}
 
func main() {
    client := flo.NewClient("localhost:9000",
        flo.WithNamespace("myapp"),
    )
    if err := client.Connect(); err != nil {
        log.Fatal(err)
    }
    defer client.Close()
 
    w, err := client.NewActionWorker(flo.ActionWorkerOptions{
        Concurrency:   10,
        ActionTimeout: 5 * time.Minute,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer w.Close()
 
    w.MustRegisterAction("process-order", processOrder)
    w.MustRegisterAction("send-email", sendEmail)
 
    // Graceful shutdown
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        sigCh := make(chan os.Signal, 1)
        signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
        <-sigCh
        w.Stop()
        cancel()
    }()
 
    log.Println("Worker starting...")
    w.Start(ctx)
}

Worker Configuration

flo.ActionWorkerOptions{
    WorkerID:      "my-worker",       // auto-generated if empty
    MachineID:     "host-01",         // defaults to hostname
    Concurrency:   10,                // max parallel tasks
    ActionTimeout: 5 * time.Minute,   // per-task timeout
    BlockMS:       30000,             // long-poll timeout
}

SDK: Invoking Actions

You don't need a worker to invoke actions — any client can invoke and check status:

client := flo.NewClient("localhost:9000")
client.Connect()
defer client.Close()
 
// Register
client.Action.Register("process-image", flo.ActionTypeUser, &flo.ActionRegisterOptions{
    TimeoutMS:   ptr(60000),
    MaxRetries:  ptr(3),
    Description: "Resize and optimize images",
})
 
// Invoke
result, _ := client.Action.Invoke("process-image",
    []byte(`{"url":"https://example.com/img.jpg","width":800}`),
    &flo.ActionInvokeOptions{
        Priority:       ptr(uint8(100)),
        IdempotencyKey: "img-resize-abc",
    },
)
fmt.Println("Run ID:", result.RunID)
 
// Poll for status
status, _ := client.Action.Status(result.RunID, nil)
fmt.Printf("Status: %s, Output: %s\n", status.Status, status.Output)
 
// Delete
client.Action.Delete("process-image", nil)

Workflows Integration

Actions are the building blocks that Workflows compose. A workflow step references an action with the @actions/ prefix:

steps:
  charge:
    run: "@actions/charge-payment"
    retry:
      max_attempts: 3
      backoff: exponential
    transitions:
      success: ship
      failure: flo.Failed

When a workflow invokes an action, it parks in waiting until a worker claims the task and reports completion, then resumes with the result.

Related Docs

• Workers — Low-level worker protocol details
• Workflows — Compose actions into multi-step orchestrations
• Stream Processing — Continuous data pipelines (different from actions)