Flow Control

Flow control shapes how a workflow runs under load. Each control is an optional, flat field on the workflow definition, alongside retry. A workflow with no flow config runs unshaped - every control is opt-in and off by default.

defineWorkflow({
  name: "sync.account",
  concurrency: { limit: 5, key: "accountId" },
  handler: async (ctx) => {
    await ctx.step.run("sync", () => syncAccount(ctx.event.data.accountId));
  },
});

from duraton import ConcurrencyConfig, define_workflow
from duraton.context import StepContext


async def handle(ctx: StepContext) -> object:
    account_id = ctx.event.data["accountId"]
    return await ctx.step.run("sync", lambda: sync_account(account_id))


sync_account_wf = define_workflow(
    "sync.account",
    handle,
    concurrency=ConcurrencyConfig(limit=5, key="accountId"),
)

type Account struct {
	ID string `json:"accountId"`
}

var syncAccountWf = duraton.DefineWorkflow(duraton.WorkflowDefinition{
	Name:        "sync.account",
	Concurrency: &duraton.ConcurrencyConfig{Limit: 5, Key: "accountId"},
	Handler: func(c *duraton.Context) (any, error) {
		account, err := duraton.EventData[Account](c)
		if err != nil {
			return nil, err
		}
		return duraton.Run(c, "sync", func() (any, error) {
			return syncAccount(account.ID)
		})
	},
})

At a glance

Keys

Most controls take an optional key: a dotted path into the event data that scopes the control to a value. key: "accountId" gives each account its own independent limit; key: "user.id" reads a nested field. Keys are field paths, not expressions. An omitted key - or a missing / non-scalar field - scopes the control to the whole workflow.

Concurrency

Caps how many runs execute at once in a scope. A slot is held only while a run is actively executing, so a run that is sleeping or awaiting an event releases its slot and does not consume one. The count is taken across the shared database, so the limit is global, not per-process. Over-limit runs are not dropped - they wait and retry as slots free, preserving order.

concurrency: { limit: 5, key: "accountId" }

Project-wide ceiling

Above the per-workflow concurrency you set in code, each project has a concurrency ceiling that caps how many runs execute at once across the whole project, regardless of workflow or key. It comes from your plan, not from a workflow field (0 = unlimited). A run must clear both its per-workflow limit and the project ceiling to start; whichever is tighter applies, and an over-ceiling run waits and retries exactly like a per-workflow over-limit run.

curl "$DURATON_URL/flow-state" | jq .concurrency
# { "limit": 50, "inUse": 12 }

Throttle

Bounds how often runs start, smoothing bursts by spreading overflow into the future - one start every perMs / limit. No run is lost; excess runs begin later.

throttle: { limit: 100, perMs: 60_000, key: "customer" }

Rate limit

Same window as throttle, opposite action: instead of delaying overflow it drops it. Up to limit runs start per perMs; the rest are shed and the event response reports dropped: true. Use it for abuse protection where shedding beats queueing.

rateLimit: { limit: 1000, perMs: 60_000, key: "ip" }

Debounce

Coalesces a burst of events into a single run that fires after periodMs of quiet. Each new event slides the deadline forward and replaces the payload, so only the last event in a quiet-bounded burst runs.

debounce: { periodMs: 5_000, key: "documentId" }

Batch

Collects events into one run, flushing when the buffer hits maxSize or timeoutMs elapses, whichever comes first. The run receives the events as ctx.events; ctx.event is the first of them.

defineWorkflow({
  name: "index.documents",
  batch: { maxSize: 100, timeoutMs: 5_000, key: "index" },
  handler: async (ctx) => {
    for (const e of ctx.events ?? []) {
      await ctx.step.run(e.data.id, () => index(e.data));
    }
  },
});

from duraton import define_workflow
from duraton.context import StepContext


async def index_documents(ctx: StepContext) -> object:
    for e in ctx.events or []:
        await ctx.step.run(e.data["id"], lambda: index(e.data))
    return None


index_documents_wf = define_workflow(
    "index.documents",
    index_documents,
    extra={"batch": {"maxSize": 100, "timeoutMs": 5000, "key": "index"}},
)

Priority

Shifts a workflow's runs earlier in the shared queue by shiftMs, so they dequeue ahead of other workflows competing for the same slots.

priority: { shiftMs: 60_000 }

Singleton

Allows at most one non-terminal run per key.

singleton: { key: "accountId", mode: "skip" }

Idempotency

Suppresses a second run of this workflow for the same derived key within a time window. The first matching event starts a run; a later event whose key resolves to the same value inside the window is dropped for this workflow (the response reports deduped: true with no runId). The event is still recorded and still wakes waitForEvent waiters - only the duplicate run is suppressed.

idempotency: { key: "orderId", periodMs: 86_400_000 } // at most one run per orderId per 24h

The key is a dotted field path into the event data, resolved the same way as every other control. A structurally malformed path - empty segments, or a leading or trailing dot - is rejected when the workflow is registered.

To verify a per-key path resolves, send two events with distinct payloads and confirm two runs start. Sending the same payload twice is deduped whether or not the path resolves, so it proves nothing.

This is run-level dedupe keyed off the event payload. To dedupe a whole event regardless of which workflows it matches - the usual safety net for an at-least-once caller retrying POST /events - send a dedupeId on the event instead (see the wire protocol); a repeat of that id within 24h is dropped before any fan-out.

Budget cap

A hard ceiling on a single run's AI spend across its step.ai.* steps. At least one axis is set. The run halts before the step.ai call that would cross a set axis - the call never runs - and fails with a BudgetError. Everything before the halt stays committed, so a replay picks up a fresh run with spend reset to zero.

cap: { maxCost: 0.25, maxTokens: 40_000 } // at most $0.25 or 40k tokens per run

Enforcement is halt-before-call, evaluated on the runner against the run's committed spend, which Duraton sums from every step.ai step's metered usage and supplies on each invoke. Because each AI step commits before the next runs, the check needs no in-run bookkeeping - it reads the authoritative rollup.

The ceiling is crossed by at most the one call that reaches it: Duraton cannot know a call's cost before it runs, so the call that pushes spend to the limit completes, and the next one halts. A cap-hit run is an ordinary failed run - its terminal BudgetError shows on the run's error surface, and for an agent loop the halted turn is the loop's last (failed) iteration. There is no separate "capped" state to handle. Raising the cap (a redeploy, or an upstream override) and replaying the failed run lets it run to completion under the new ceiling.

AI budget

Where a cap bounds a single run, a budget bounds a workflow's aggregate AI spend across all its runs over a rolling window. When the workflow's spend crosses a set axis, Duraton pauses the workflow's in-flight runs at their checkpoint - holding no worker, keeping their progress - and resumes them on its own when the window rolls forward or you raise the budget.

budget: { maxTokens: 6_000, windowMs: 3_600_000, warnAtPct: 80 } // 6k tokens per hour

At least one axis and a positive windowMs are set. A budget is aggregate and window-scoped, so only Duraton - which meters every run's AI spend - can evaluate it. It is checked on a background loop: each tick it sums the scope's spend over the trailing windowMs from the same metered usage the cap reads, and when a set axis is crossed it pauses the scope's in-flight runs. Overshoot is bounded by one check interval (plus any per-run cap).

A budget-paused run is an ordinary paused run; its pausedBy records reason: "budget" so the inspector, the API, and an agent through the MCP get_run can tell it from a manual pause. It resumes automatically when the window rolls the old spend off or you raise the budget - a manual resume is only an override, and a manually paused run is never auto-resumed.

AI throttle

Where a budget sets a hard ceiling that pauses, a tokenThrottle sets a token rate that spreads - it protects your own provider quota (your BYO keys) so one busy key can't saturate it. It is the token-denominated sibling of throttle: instead of one unit per run start, Duraton debits each AI step's actual token usage into the rate bucket after the step completes, and when the bucket is drained it delays new run starts for that key into the future - the run waits in the queue, holding no worker, until the rate recovers.

tokenThrottle: { tokens: 1_000, perMs: 60_000, key: "userId" } // 1k tokens/min per user

Both tokens and perMs are positive. Because a step's tokens are known only after it runs, Duraton gates a run's start on the key's recent usage and debits the actual tokens when each AI step commits - so a key that just spent heavily has its next runs spread out, while a fresh key starts immediately. It shapes the rate of starts, not a single run: a run already in flight spends its tokens, and the debit delays that key's next starts. Pair it with a cap to also bound any one run.

Observing flow control

curl "$DURATON_URL/flow-state?workflow=index.documents"

{
  "debounce": [{ "app": "docs", "workflow": "index.documents", "pending": 3, "nextFireAt": "2026-07-01T10:00:05Z" }],
  "batch": [{ "app": "docs", "workflow": "index.documents", "buffered": 42, "oldestAt": "2026-07-01T10:00:01Z" }],
  "concurrency": { "limit": 50, "inUse": 12 }
}

The console renders the same three reads in the Workflows list and on Overview.