Writing Tasks

For developers writing the code that runs inside a workflow. It covers the workflow.Flow carrier a task receives: reading and writing state, the control signals, and handling failure. The graph that decides which task runs is Building Graphs.

A task is a function your host’s ExecuteTask runs. It receives a *workflow.Flow — the carrier holding the step’s state and control signals — does its work, writes outputs back onto the flow, and returns. This guide covers the Flow API a task uses.

func charge(ctx context.Context, f *workflow.Flow) error {
    amount := f.GetFloat("amount")
    if amount <= 0 {
        return errors.New("nothing to charge")
    }
    receipt, err := billing.Charge(ctx, amount)
    if err != nil {
        return err // an ordinary failure: routes to onError or fails the flow
    }
    f.SetString("receipt", receipt)
    return nil
}

The engine populates the flow’s state from the step’s input before the call, records whatever you change as the step’s changes, and persists it. Returning a non-nil error fails the step (unless an onError transition matches; see Building Graphs).

Reading State

The flow arrives with the merged input state of the step. Read it with typed accessors:

name   := f.GetString("name")
count  := f.GetInt("count")
amount := f.GetFloat("amount")
ok     := f.GetBool("approved")
ttl    := f.GetDuration("ttl")
tags   := f.GetStrings("tags")

if f.Has("coupon") { ... }               // presence check

var order Order
ok, err := f.Get("order", &order)        // decode a field into a struct; ok is false if it is absent
if err != nil { ... }

f.Snapshot() returns a read-only copy of the entire current state. f.Parse(&v) decodes the whole state into a struct.

An absent field reads as the zero value, so an optional field needs no guard. A field that is present but holds the wrong type panics — reading a string as an int is a bug in the workflow, and handing back a 0 the task never wrote is how a GetInt("retryAfter") over a 1.5 becomes a zero-delay hot loop against a downstream. The panic is not a crash: the engine catches it at the task-call boundary and fails the step like any returned error, routing to onError if the graph has one. Use f.Get(key, &v), which returns the mismatch as an error, to handle a mistyped field instead of failing on it.

Writing State

Mutations are recorded as the step’s output delta:

f.SetString("status", "charged")
f.SetInt("attempts", n)
f.SetFloat("balance", b)
f.SetBool("approved", true)
f.SetStrings("errors", msgs)
f.Set("order", order)         // any JSON-marshalable value

f.Del("coupon", "scratch") // remove fields
f.Clear()                     // remove everything (Del on every field)

A field name is an ASCII letter or underscore, followed by ASCII letters, digits, underscores or hyphens (Field names). Every setter and Del panic on any other name, and so does Clear if the state holds one. The panic is the task’s error: it takes the graph’s onError transition, or fails the step. A bad name is a typo in the task’s code, and an error the task ignored would drop the write without a word. A host reaching tasks over a transport is held to the same rule: the step fails if the task’s result writes such a name. In that JSON form a written name ending with ! records a field the task replaced rather than added to, and one ending with + the elements a list gained since the step’s input (a task sees one only when its step runs more than once).

Binary Data Must Be Base64-Encoded

Known limitation: a NUL character (U+0000) in a string cannot be stored. It is valid UTF-8 and marshals to legal JSON, but PostgreSQL’s JSONB rejects it outright — so it works on SQLite and kills the flow on the recommended production database. Dwarf does not detect or reject this, so it is your responsibility: raw bytes belong in state base64-encoded.

f.SetString("payload", string(rawBytes))                        // FAILS on Postgres if rawBytes holds a 0x00
f.SetString("payload", base64.StdEncoding.EncodeToString(raw))  // correct

Nothing else about strings is constrained — tabs, newlines, other control characters, and emoji all round-trip fine. Invalid UTF-8 in a Go string is replaced with U+FFFD when it is stored. Bytes that arrive already encoded as JSON - a json.RawMessage passed to Set, or the state a remote host returns for a task - are kept as given, so they must be valid UTF-8: PostgreSQL rejects invalid UTF-8 just as it rejects a NUL.

Large Integers: Exact in Storage, Rounded by Untyped Reads

An integer beyond 2^53 (about 9.0e15) is stored exactly — it round-trips digit-for-digit into the next step’s state, into the final state, and through Fork and Continue. Reading one does not damage it either. What matters is how you read it back:

f.SetInt("orderID", 1234567890123456789)
f.GetInt("orderID")                     // exact — reads straight into an int

var order struct{ ID int64 `json:"orderID"` }
f.Get("orderID", &order.ID)             // exact — a typed target

var loose any
f.Get("orderID", &loose)                // ROUNDED — untyped lands in a float64: 1234567890123456768

Typed integer accessors are exact. GetInt and a Get into an integer-typed target decode straight into that type, so an integer is exact across the type’s whole range (GetInt panics on one beyond an int). GetFloat returns a float64, which holds an integer exactly only up to 2^53. Untyped reads round, because JSON’s number type decodes into float64 when there is nothing narrower to decode into: Get into an any, and Parse into a map.

A when expression on a transition is exact too: condition evaluation compares a >2^53 id as an integer, not through a float64.

g.AddTransitionWhen("A", "B", "orderID == 1234567890123456789")  // exact at any magnitude

So: use typed accessors, and carry the value as a string only if it must survive an untyped reader (Get into an any, Parse into a map) — the same reason APIs that mint 64-bit ids publish an id_str alongside them. A time.Duration is nanoseconds, so a duration past ~104 days is in the same territory for those readers.

Contributions, Not Totals, on a Reduced Field

A field the graph wires to a reducer (append, add, union, merge, …) takes contributions. A write combines with what the field already holds, wherever the task runs — in a straight line, in a fan-out branch, or at a fan-in — and a read returns the combined value. So write only your contribution, never the value you read:

// "messages" is wired to ReducerAppend.
f.Set("messages", []any{reply})   // correct: adds the reply
// f.Set("messages", append(f.GetStrings("messages"), reply))   // WRONG: adds the whole history again

Two writes in one task both count, and a read in between sees the first. A nil contributes nothing, so a task that collected nothing into a nil slice can write it safely.

To replace or clear a reduced field, say so. Overwrite replaces the field, and Del (which is Overwrite with nil) clears it; later writes in the same task add to the new value:

history := f.GetStrings("messages")
f.Overwrite("messages", history[max(0, len(history)-20):])   // keep the last 20
f.Set("messages", []any{reply})                              // then add the reply

Neither is allowed on a reduced field inside a fan-out branch: the branches run in parallel, and a reset in one has no defined meaning for the others. Overwrite returns an error there, and a Del fails the step - as does Clear, which deletes every field. Reset the field before the fan-out or after its fan-in.

A task that runs more than once keeps every run’s writes. A step re-runs its task after Retry, after a Resume of an Interrupt, and when a Subgraph returns, and each run’s contributions add to the previous runs’. That lets a task record one entry per attempt. It also means a contribution made before Interrupt or Subgraph is made twice, because the task runs from the top again:

yield, err := f.Interrupt(map[string]any{"draft": draft}, &resume)
if err != nil || yield {
    return err
}
f.Set("messages", []any{draft})   // after the approval: written once

Call Retry before a contribution that must not repeat, for the same reason. A field with no reducer is unaffected: the last write wins.

See Concepts → Reducer.

Control Signals

Beyond reading and writing state, a task can steer the engine. These are methods on the flow; after calling one, the task should return as instructed.

Retry

func (f *Flow) Retry(initialDelay time.Duration, delayMultiplier float64, maxIntervalDelay time.Duration, giveUpAfter time.Duration) bool

Retry schedules this task to run again after a backoff delay. It returns true while the backoff delays of every attempt so far, the next one included, add up to at most giveUpAfter; return nil then, and the task runs again after the delay. It returns false once they would exceed it; return your error then. Retry has no condition of its own: your code decides what is worth retrying. The typical use is shown in Handling Transient Failures.

The details:

• Delay. The delay before attempt N is min(initialDelay × delayMultiplier^N, maxIntervalDelay). A zero initialDelay retries immediately, a zero maxIntervalDelay leaves the delay uncapped, and a multiplier of 1.0 keeps it constant.
• Invalid arguments. initialDelay and maxIntervalDelay must not be negative, and delayMultiplier must be a finite number that is not negative; otherwise Retry returns false.
• Limit. giveUpAfter bounds the total backoff, not a count and not the clock: it counts only the delays Retry asks for, so a step held up behind higher-priority work does not lose its retries. A Sleep is your own wait and is not counted; to bound your sleeps too, keep your own budget in a state field and gate Retry on it. A giveUpAfter of zero or less retries without limit.
• Counting attempts. A zero initialDelay never adds to the total, so bound immediate retries by count: check f.Attempt() yourself (with giveUpAfter zero to bound by count alone). Attempt() also counts the retries your host armed without reaching the task (a worker that was down, a refused or dropped connection), so bound retries of a transient failure with giveUpAfter rather than Attempt().
• Outputs. On retry, the task’s prior output is merged back into its input, so the next attempt sees what the last one wrote.

Sleep

Delay the next step:

f.Sleep(30 * time.Minute) // the next step won't dispatch for 30 minutes
return nil

The engine sets the next step’s earliest-run time, and the step is picked up within a scheduling cycle of it coming due (typically tens of milliseconds, at most about a second) — durable across restarts, no goroutine held open.

Goto

Override routing: skip normal transition evaluation and follow the goto transition to a target (it must be registered with AddTransitionGoto):

if needsReview {
    f.Goto("ManualReview")
}
return nil

This is how you build loops and computed branching.

Interrupt (Human-in-the-Loop)

Park the flow to await external input, then receive that input when the flow is resumed. Interrupt follows a two-call pattern within the same task body:

func approve(ctx context.Context, f *workflow.Flow) error {
    var resume map[string]any
    yield, err := f.Interrupt(map[string]any{
        "question": "Approve refund of $" + f.GetString("amount") + "?",
    }, &resume)
    if err != nil {
        return err
    }
    if yield {
        return nil // first call: flow is now parked; return immediately
    }
    // re-entry after Resume: 'resume' holds the caller's data
    f.SetBool("approved", resume["approved"] == true)
    return nil
}

On the first call the flow goes to interrupted, the payload is surfaced to whoever is awaiting it (Await returns, as it does for any stop). When your application calls eng.Resume(ctx, flowKey, data), the task re-runs and Interrupt returns (false, nil) with the caller’s data unmarshaled into your &resume pointer. The resume data is delivered through that pointer — it is not merged into state. The whole exchange is drawn in Parking: Interrupt and Subgraph. See Driving Flows → Resume.

Subgraph

Call another workflow as a child and get its result back. Like Interrupt, it’s a two-call, park-and-resume pattern — and semantically a function call: only the explicit in crosses into the child, only the child’s final state (out) crosses back.

func enrich(ctx context.Context, f *workflow.Flow) error {
    var out struct {
        Profile map[string]any `json:"profile"`
    } // or: var out map[string]any
    yield, err := f.Subgraph("enrichment.workflow", map[string]any{
        "id": f.GetString("customerID"),
    }, &out)
    if err != nil {
        return err
    }
    if yield {
        return nil // child launched; this step parks until it completes
    }
    f.Set("profile", out.Profile) // adopt the fields you want
    return nil
}

The result is unmarshaled into the trailing out pointer (a *struct for type safety, a *map[string]any for dynamic access, or nil to ignore it). Pass nil input for “no arguments,” or f.Snapshot() to forward the parent’s whole state. See Fan-Out & Subgraphs → Subgraphs.

err carries the child’s error when it fails, and the reason when it could not be started (a 503 when its graph could not be loaded, a 404 when it does not exist, a 400 when it is invalid or in holds an invalid field name). See When the child fails, or cannot start.

To invoke a single unit of work as an isolated child, declare a one-task workflow and Subgraph it. A task only ever runs as part of a graph; it is not an independently invocable unit.

Baggage (Host Context)

The opaque, host-defined value set in FlowOptions.Baggage at create time rides on the dispatch context of every task. Read it with workflow.BaggageFrom(ctx) — it’s how a host carries identity/claims, tenant, or locale to task code without the engine interpreting it:

func charge(ctx context.Context, f *workflow.Flow) error {
    claims := workflow.BaggageFrom(ctx) // a workflow.State; empty, and safe to read, when none was set
    if actor := claims.GetString("actor"); actor != "" {
        token := mintToken(actor) // e.g. act as the original caller
        ...
    }
    ...
}

Baggage is set once at Create (as any JSON object — a struct or map), frozen on the flow, inherited by subgraphs and Continue, and delivered back as a workflow.State, read through the same getters as flow state, so its numbers are as exact as flow state’s. See Driving Flows → Create.

Handling Transient Failures

The engine never inspects an error’s status code or text. Any error you return is terminal for that attempt: the engine routes it via the graph’s onError transition if one exists, otherwise it fails the step.

To back off on a transient failure (a rate limit, a downstream that’s momentarily unavailable), detect it yourself and arm flow.Retry — the task owns the decision because it owns the resource identity:

err := callDownstream(ctx)
switch {
case isRateLimited(err), isUnavailable(err): // e.g. HTTP 429 / 503
    if f.Retry(100*time.Millisecond, 2.0, 10*time.Second, time.Hour) {
        return nil // re-run after backoff, bounded by its total
    }
    return err // horizon exceeded: onError or fail
default:
    return err // ordinary failure: onError or fail
}

The arguments and the time limit are explained under Retry.

Keep Payload Data Out of Error Text

An error message is not a private log line. Whatever you return is stored on the flow and the step, and it comes back from the two readers that expose no state payloads at all:

• ListOpen and ListClosed return each flow’s error and terminate reason on the summary.
• History returns each step’s error, alongside metadata — it deliberately omits state and changes, but not the error.
• Query.Search substring-matches the error text, so flows can be found by what their errors contain.

An operator console built on the lists and History is normally the low-privilege surface — it shows what happened without showing what was in it. Error text is the one channel that crosses that line, so anything you interpolate is visible to everyone who can list flows, which is usually a wider group than those allowed to read workflow data.

Name the failure; don’t quote the payload.

// Leaks: the card number and the email are now listable and searchable.
return fmt.Errorf("declined card %s for %s: %w", cardNumber, customerEmail, err)

// Fine: the failure is identifiable and the sensitive values stay in state.
return fmt.Errorf("charge declined for customer %s: %w", customerID, err)

Reference data by an internal identifier you can look up, rather than by its value. Be careful with %w and %v on downstream errors too — an API client’s error often quotes the request body it sent.

This costs you nothing in diagnosability, and it makes triage better. Errors that name a class (charge declined, upstream 503) group across flows, which is what a search during an incident is actually for. Errors that embed unique payload values match one flow each, so they are simultaneously the least useful to search and the most sensitive to expose.

One related surface: an error’s structured properties ride into the onErr state field for handler routing, so they land in state like any other field rather than in the error text. Treat them with the same care you would give any state value.

Idempotency

Execution is at-least-once: your task may run more than once for the same step. The engine guarantees that the flow’s persisted state reflects exactly one execution — a late worker’s writes are fenced off — but it cannot fence your side effects, because it does not own your downstream. Charging a card, sending an email, or posting to an API is yours to make safe.

This is not an edge case to design around later. It happens whenever:

• a replica dies, or is killed before it finishes shutting down, and the step’s claim expires (see Crash Recovery);
• a task overruns its time budget and another replica claims the step while the first attempt is still running;
• the database is unreachable long enough that the step’s outcome cannot be recorded;
• a database is restored from backup, re-dispatching everything that was in flight.

Two of those produce concurrent attempts, not merely sequential ones. A slow task whose claim expires keeps running while its replacement starts, so a dedupe that reads before it writes can be raced by its own retry. Make the check and the effect atomic.

StepKey() Is Your Idempotency Key

func chargeCard(ctx context.Context, f *workflow.Flow) error {
    return payments.Charge(ctx, payments.Request{
        Amount:         f.GetInt("amountCents"),
        Customer:       f.GetString("customerID"),
        IdempotencyKey: f.StepKey(),   // stable across every re-execution of this step
    })
}

f.StepKey() is {shard}-{hour}-{flowID}-{stepID}-{token} and is stable across every re-execution of one step: a re-run after a crash, a flow.Retry, or a retry after a database blip all reuse the same step and hand back the same key. It is also unique per step, so two steps of one flow never collide.

The other identifiers are the wrong choice, each for its own reason:

A loop is not a retry. A branch that revisits a task via flow.Goto creates a new step each time around, so each iteration gets its own StepKey and is correctly treated as separate work. Likewise Fork clones steps with new keys — a fork is a deliberate re-run and its side effects fire again.

If the Downstream Has No Idempotency Key

Make the effect naturally repeatable, or record the effect atomically in a store you control:

// A unique constraint on step_key turns "have I done this?" into one atomic write.
_, err := db.ExecContext(ctx,
    `INSERT INTO sent_emails (step_key, recipient) VALUES ($1, $2) ON CONFLICT DO NOTHING`,
    f.StepKey(), recipient)
if err != nil {
    return err
}
// ... then send. A duplicate attempt inserts nothing and can skip.

Prefer shapes that are repeatable by construction: PUT over POST, “set the balance to X” over “add X”, upserts over inserts. An effect that is naturally idempotent needs no key at all.

Give an unavoidably non-idempotent effect its own task. A step’s outcome is durably recorded before its successor runs, so isolating the effect in a single-purpose task narrows the window in which a re-run can duplicate it, and keeps the surrounding logic free to be retried without restriction.

One Thing You Cannot Rely On

A task’s flow.Set writes are discarded when it returns an error. Both failure paths agree: the onError handler receives the step’s input state plus the error, never anything the failing run wrote. What the step’s earlier runs wrote - before a Retry, or before an Interrupt or Subgraph call the task was dispatched again from - is kept, and the handler sees it. So within one run you cannot record “I already did the side effect” by writing it into state and then returning an error — the record is dropped. Use an external store, as above.

Timestamps

f.CreatedAt() and f.UpdatedAt() are populated on every dispatch. Use CreatedAt to implement a workflow-level deadline in author space, where the flow can still act on it — the engine’s only limit, SetFlowExpiration, terminates without running a task:

if time.Since(f.CreatedAt()) > 24*time.Hour {
    return errors.New("workflow exceeded its 24h budget")
}

Next: Building a Host.