Signal operators - Automate.ax

Import top-level functions and public types from automate.ax. Most operators also have a fluent form on Signal, KeyedSignal, or GlobalSignal. Signals compose synchronously during planning and materialize during durable execution. Transformer, predicate, key, and section callbacks must be synchronous. Put external work in actions.

Choose an operator family

Need	Family	Operators
Derive a value in the current context	Value composition	Property access, `transform`, `t`
Route a value or handle its terminal state	Routing and lifecycle	`gate`, `filter`, `partition`, `branch`, `outcome`, `onSuccess`, `onFailure`, `onClose`
Combine or admit repeated and independent occurrences	Stream coordination	`merge`, `correlate`, `collect`, `once`, `take`, `rateLimit`, `each`, `funnel`, `window`
Select or resume within one causal occurrence	Causal scheduling	`race`, `delay`, `timeout`
Limit overlapping occurrence-scoped work	Execution coordination	`serialize`, `concurrent`

merge and race answer different questions. merge forms one stream from independent roots and emits every occurrence. race selects one alternative within a shared activation boundary. Dependency and structure operators—dependentOn, withPrerequisites, scope, and group—compose with every family.

Property access

Reading a non-function property returns another signal. Chained access works across objects, arrays, and primitives:

const path = request.path // Signal<string>
const length = request.path.length // Signal<number>

The derived signal closes or fails with its source. Accessing a property after the source resolves to null or undefined fails with TypeError. Signal methods reserve their names, and function-valued payload properties aren’t projected as callable values. Use transform to call a value method such as names.join(", ").

`transform`

Creates a derived signal from one or more inputs.

// Top-level forms
const normalized = transform(request.path, (path) => path.toLowerCase())
const requestLine = transform(
  [request.method, request.path],
  (method, path) => `${method} ${path}`,
)

// Fluent forms
const fluentNormalized = request.path.transform((path) => path.toLowerCase())
const fluentLine = request.method.transform(
  request.path,
  (method, path) => `${method} ${path}`,
)
const summary = request.method.transform(
  [request.path, request.headers],
  (method, path, headers) => ({ headers, method, path }),
)

The transformer runs after every input emits and must be pure. An input failure fails the result; an input closure closes it. A thrown transformer error fails with code expression_failed and doesn’t consume an action retry. Arbitrary transforms don’t preserve keyBy or globally annotations.

`t`

Builds a Signal<string> from a tagged template containing at least one signal:

const summary = t`${request.method} request to ${request.path}`

Static interpolations add no dependency. Automate.ax converts interpolated values to text after every signal emits. Calling t without a signal interpolation throws TypeError during planning. Use transform for conditional logic, value methods, or non-string output.

`isSignal`

Narrows an unknown value to Signal:

function trim(value: string | Signal<string>) {
  return isSignal(value) ? value.transform((text) => text.trim()) : value.trim()
}

It returns true only for objects implementing the Automate.ax signal protocol. Action inputs already accept compatible signals; use this guard in helpers that accept both in-memory values and signals.

`keyBy`

Returns an immutable KeyedSignal<T> whose cross-context operations use the encoded key as their partition:

const byCustomer = event.keyBy((value) => value.customerId)

The key function must be pure and return an encoded Automate.ax value. Calling keyBy replaces an existing keyed or global annotation.

`globally`

Returns a GlobalSignal<T> whose cross-context operations share one partition:

const batch = feedback.globally().collect(10)

Calling globally replaces an existing keyed or global annotation. Use keyBy when one entity’s occurrences must not reset another entity’s timer or enter the same batch.

`timestamp`

Returns the stable platform time at which an occurrence emitted:

const receivedAt = timestamp(request)
const fluentReceivedAt = request.timestamp()

Both forms return Signal<Date>. Trigger timestamps use ingestion time, action timestamps use completion time, and coordinated signals use their persisted decision time. A pure derivation uses the most recent contributing durable boundary. Use a provider timestamp when its domain time is the required value.

`correlationId`

Returns a stable keyed Signal<string> derived from an occurrence’s durable boundary:

const interactionId = correlationId(request)
const sharedBoundaryId = correlationId(request, false)

const fluentInteractionId = request.correlationId()
const fluentSharedBoundaryId = request.correlationId(false)

perDeclaration defaults to true, so separate declarations anchored to one occurrence receive independent IDs. Pass false to derive the ID only from the durable dependency boundary. Prefer a provider’s domain ID when it already identifies the entity or conversation.

`correlate`

Matches one occurrence from every keyed stream by exact encoded key:

const match = correlate([requestById, decisionById])
const orderedMatch = correlate([requestById, decisionById], { ordered: true })
const expiringMatch = correlate([requestById, decisionById], {
  occurrenceTtl: "30d",
})
const invoiceWithCurrentPlan = correlate([
  invoiceByCustomer,
  {
    signal: planByCustomer,
    selection: "latest",
    consumption: "retain",
  },
])

const fluentMatch = requestById.correlate(decisionById)
const fluentMany = requestById.correlate([decisionById, auditById], {
  ordered: true,
})

Option	Default	Description
`ordered`	`false`	Require stream occurrences to arrive in declaration order.
`occurrenceTtl`	No expiry	Milliseconds or a compact duration string. After expiry, an unmatched occurrence can’t participate in a new match.

Each input defaults to { selection: "oldest", consumption: "consume" }. Set selection: "latest" to discard older eligible occurrences for that input. Set consumption: "retain" to reuse the selected occurrence in later matches: oldest retention behaves like a latch, while latest retention behaves like an updating register. Retained occurrences remain operational state until superseded, expired by occurrenceTtl, or the automation is removed. The child context contains every selected parent history, so downstream work can consume the original keyed signals. The returned Signal<null> exposes the match boundary. Reusing one durable signal origin in several positions throws TypeError; collect distinct occurrences first when one stream must fill several roles.

`merge`

Combines independent occurrence streams without matching or consuming them together:

const contactEvent = merge([contactCreated, contactUpdated])
const fluentEvent = contactCreated.merge(contactUpdated)
const fluentMany = contactCreated.merge([contactUpdated, contactDeleted])

Every successful input occurrence emits once in its own child context. Inputs don’t wait for or consume one another, and arrival order doesn’t suppress any occurrence. The result is the union of the input value types. Use correlate when related streams must meet by key, or race when alternatives within one causal occurrence compete to finish first.

`collect`

Consumes a fixed number of keyed or global occurrences in arrival order:

const ten = collect(feedback.globally(), 10)
const daily = collect(feedbackByCustomer, 10, { occurrenceTtl: "1d" })
const dynamic = collect(feedbackByCustomer, requestedCount, {
  inheritConflictingValuesFrom: "first",
})

const fluentTen = feedback.globally().collect(10)
const fluentDynamic = feedbackByCustomer.collect(requestedCount, {
  inheritConflictingValuesFrom: "first",
})

Option	Default	Description
`inheritConflictingValuesFrom`	`"last"`	Choose the first or last selected parent when another value is ambiguous in the merged child context. Array order doesn’t change.
`occurrenceTtl`	No expiry	Milliseconds or a compact duration string. After expiry, an unmatched occurrence no longer counts toward a collection.

count accepts a positive integer or Signal<number>. Invalid literal counts throw during planning. Signal-derived counts match only occurrences whose resolved counts are equal. The result is a partition-preserving NonEmptyArray<T>.

`once` and `take`

Admit a bounded number of occurrences from each keyed or global partition:

const firstSignup = signupsByContact.once()
const firstThreeAttempts = take(attemptsByAccount, 3)
const oncePerMonth = signupsByContact.once({ ttl: "30d" })

once(signal, options?) is take(signal, 1, options?). Without ttl, the count remains claimed permanently. With ttl, the claim starts when the first occurrence is admitted and the complete count resets when it expires. The result preserves the source partition.

`rateLimit`

Admit at most a fixed number of occurrences in each rolling interval:

const queued = requestsByProvider.rateLimit({ limit: 100, interval: "1m" })
const sampled = rateLimit(telemetry.globally(), {
  limit: 1_000,
  interval: "1m",
  overflow: "drop",
})

overflow defaults to "wait", which queues excess occurrences FIFO until rolling capacity returns. "drop" closes excess occurrences without emitting. Every key has independent capacity; globally() creates one shared limit.

`each`

Creates one durable child context per array item:

// Gather one returned signal per item.
const sent = each(recipients, (recipient) =>
  sendEmail({ to: recipient, subject: "Welcome", text: "Thanks for joining." }),
)
const fluentSent = recipients.each((recipient) =>
  sendEmail({ to: recipient, subject: "Welcome", text: "Thanks for joining." }),
)

// Return nothing when only the per-item effects matter.
each(recipients, (recipient) => {
  sendEmail({ to: recipient, subject: "Welcome", text: "Thanks for joining." })
})

Returning a signal produces Signal<TResult[]> in source-array order. A failed item fails the gathered result, a closed item closes it, and an empty source gathers to []. Returning nothing produces void and doesn’t join the item contexts.

`gate`

Preserves a value only when a separate boolean signal emits true:

const selected = gate(request, isApiRequest)
const fluentSelected = request.gate(isApiRequest)

false or a closed condition closes the result without materializing the value. A failed condition fails it. The result preserves a keyed or global annotation carried by the value.

`filter`

Preserves a value only when a synchronous predicate returns true:

const urgent = filter(request, (value) => value.priority === "urgent")
const fluentUrgent = request.filter((value) => value.priority === "urgent")

const paid = filter(
  status,
  (value): value is PaidStatus => value.type === "paid",
)
const fluentPaid = status.filter(
  (value): value is PaidStatus => value.type === "paid",
)

The boolean-predicate overload preserves the input type. The type-guard overload narrows it. A value that fails the predicate closes the result; a thrown predicate fails with code expression_failed. The result preserves a keyed or global source annotation.

`partition`

Splits one signal into complementary matching and remaining paths:

const [urgent, routine] = partition(
  request,
  (value) => value.priority === "urgent",
)
const [fluentUrgent, fluentRoutine] = request.partition(
  (value) => value.priority === "urgent",
)

const [paid, unpaid] = partition(
  status,
  (value): value is PaidStatus => value.type === "paid",
)

Exactly one output emits the original value; the other closes. A type-guard predicate narrows the matching output and excludes that type from the other output. Both outputs preserve a keyed or global source annotation.

`dependentOn`

Preserves one signal value while adding prerequisites to its downstream dependency chain:

const ready = dependentOn(request, indexed)
const readyAfterBoth = dependentOn(request, [indexed, audited])

const fluentReady = request.dependentOn(indexed)
const fluentReadyAfterBoth = request.dependentOn([indexed, audited])

The result closes or fails when a prerequisite does. It preserves a keyed or global annotation carried by the primary signal. The operator doesn’t delay the declaration that produced the primary signal; it changes only the returned signal’s dependencies.

`withPrerequisites`

Applies one signal or a nonempty tuple to every durable declaration in a synchronous section:

withPrerequisites(request, () => {
  sendEmail({ subject: "Request received", text: "Processing started." })
})

const result = withPrerequisites([request, approved], () =>
  sendEmail({ subject: "Approved", text: "Processing started." }),
)

Nested sections combine their prerequisites. A directly returned signal also inherits them; other return values pass through unchanged. Don’t declare triggers inside withPrerequisites or branch. A trigger starts an independent root context.

`branch`

Declares complementary synchronous sections selected by boolean signals:

// if / else
branch(
  urgent,
  () => sendEmail({ subject: "Urgent", text: "Escalate it." }),
  () => sendEmail({ subject: "Routine", text: "Queue it." }),
)

// if / else if / else
branch(
  urgent,
  () => urgentResult,
  needsReview,
  () => reviewResult,
  () => routineResult,
)

// Fluent receiver form
urgent.branch(
  () => urgentResult,
  () => routineResult,
)

Later conditions materialize only after every earlier condition is false. When every selected section returns a signal, branch returns the selected result. With only a true section, its result closes when the condition is false. Sections must be synchronous and can’t declare triggers.

`race`

Selects the first causally related signal to emit or fail:

const winner = race([fast, slow])
const fluentWinner = fast.race(slow)
const fluentMany = fast.race([medium, slow])

Inputs must descend from the same activation-boundary occurrence. Failure wins like emission. Closed inputs leave the race, which closes only when every input closes. The selected result creates the closest child boundary for downstream actions. Independent roots can’t race; use merge when either of several triggers should start the same flow.

`serialize`

Runs one structured work region at a time in each keyed or global partition:

const contactByUser = contactEvent.keyBy((event) => event.user)

const updated = serialize(contactByUser, () => {
  sendAudit({ event: contactEvent })
  return updateContact({ event: contactEvent })
})

const fluentUpdated = contactByUser.serialize(() =>
  updateContact({ event: contactEvent }),
)

The source signal has three roles: each emission enters the queue, its key selects the queue, and its value and provenance remain available inside the section. Use globally() for one queue across all occurrences. The section creates an isolated structured-concurrency region. Independent declarations inside it can run concurrently; declaration order doesn’t serialize them. The next occurrence in the same partition starts only after every context and unmatched coordination offer created by the current region becomes terminal. Other partitions continue independently. The callback doesn’t return a completion signal. If it returns a signal, that value remains unavailable downstream until the whole region closes. Returning nothing is valid when only the enclosed effects matter. Sections must be synchronous and can’t declare triggers.

`concurrent`

Runs up to a fixed number of structured work regions in each partition:

const result = concurrent(jobsByAccount, { limit: 5 }, () =>
  processJob({ job: jobsByAccount }),
)
const fluentResult = jobsByAccount.concurrent({ limit: 5 }, () =>
  processJob({ job: jobsByAccount }),
)

concurrent has the same section and completion semantics as serialize, but admits up to limit active regions per key. Additional occurrences wait FIFO. serialize(stream, section) is the capacity-one form.

`delay`

Resumes a value or synchronous section after a durable relative delay:

const elapsed = delay("5m") // Signal<null>
const delayedRequest = delay("5m", request)
const fluentDelayedRequest = request.delay("5m")
const delayedResult = delay("5m", () =>
  sendEmail({ subject: "Reminder", text: "Five minutes passed." }),
)

Duration accepts nonnegative milliseconds, a compact duration string, or a compatible signal. A value-carrying delay preserves the source’s keyed or global annotation. The timer-only form declares a hidden delivery trigger; scope it to its initiating occurrence with withPrerequisites when the automation has another possible root.

`timeout`

Mirrors a signal unless a durable relative deadline wins:

const result = timeout(request, "30s")
const fluentResult = request.timeout("30s")

The deadline accepts Duration. If it wins, the result fails with name: "TimeoutError" and message "Signal timed out.". Source closure leaves the deadline active. A timeout can span execution jobs but doesn’t interrupt an action handler already running. The result preserves the source’s keyed or global annotation.

`funnel`

Applies one timing and output policy to each keyed or global partition:

const firstPerMinute = funnel(statusByAccount, {
  minGap: "1m",
  triggerAt: "start",
})
const buffered = statusByAccount.funnel({
  buffer: true,
  maxBurstDuration: "10m",
  minQuietPeriod: "30s",
})

Option	Default	Description
`minQuietPeriod`	None	Wait this long after the most recent admitted occurrence before an end emission.
`maxBurstDuration`	None	End a burst this long after its first occurrence. Can’t be combined with `until`.
`until`	None	End at an absolute `Date` or `Signal<Date>`. Can’t be combined with `maxBurstDuration`.
`minGap`	None	Require this interval between emissions.
`triggerAt`	`"end"`	Emit at `"start"`, `"end"`, or `"both"`.
`select`	`"latest"`	Select `"first"` or `"latest"` for an end emission. Can’t be combined with `buffer: true`.
`buffer`	`false`	Emit every trailing occurrence in arrival order as `NonEmptyArray<T>`.
`inheritConflictingValuesFrom`	`"last"`	With `buffer: true`, choose the first or last parent when another value is ambiguous in the merged child context.

Durations accept Duration. An end emission requires minQuietPeriod, maxBurstDuration, or until. Start-only mode emits immediately and ignores the rest of the burst, so it doesn’t accept select or buffer. The result preserves the source partition; locally closed paths contribute no occurrence. Minimum-gap memory belongs to the durable partition and remains effective after older execution values leave retention.

`debounce`

Emits the latest keyed or global occurrence after its partition remains quiet:

const settled = debounce(updateByCustomer, "5m")
const fluentSettled = updateByCustomer.debounce("5m")

duration accepts Duration. Each occurrence resets only its partition’s timer. This is the selecting funnel preset { minQuietPeriod: duration }, and the result preserves the source partition.

`window`

Buffers keyed or global occurrences until a relative or absolute deadline:

const relative = window(eventsByCustomer, "5m")
const relativeFirst = window(eventsByCustomer, "5m", {
  inheritConflictingValuesFrom: "first",
})
const absolute = window(eventsByCustomer, { until: endOfDay })

const fluentRelative = eventsByCustomer.window("5m")
const fluentAbsolute = eventsByCustomer.window({ until: endOfDay })

Option	Default	Description
`until`	Required in the absolute form	A `Date` or `Signal<Date>`. An already-past deadline becomes ready immediately.
`inheritConflictingValuesFrom`	`"last"`	Choose the first or last selected parent when another value is ambiguous in the merged child context. Array order doesn’t change.

The relative form accepts Duration. The first occurrence opens the window. The partition-preserving result is a NonEmptyArray<T> in arrival order.

`outcome`

Converts every terminal state into a tagged value and always succeeds after the source terminates:

const result = outcome(request)
const fluentResult = request.outcome()

The result is Signal<SignalOutcome<T>>. Unlike value-preserving operators, outcome doesn’t retain a keyed or global annotation.

`succeeded`

Emits whether a signal terminated successfully with a value:

const didSucceed = succeeded(result)
const fluentDidSucceed = result.succeeded()

The result is true for success and false for failure or closure. The predicate itself succeeds after any terminal state.

`failed`

Emits whether a signal terminated with an automation error:

const didFail = failed(result)
const fluentDidFail = result.failed()

The result is true for failure and false for success or closure. The predicate itself succeeds after any terminal state.

`closed`

Emits whether a signal closed without a value:

const didClose = closed(result)
const fluentDidClose = result.closed()

The result is true for closure and false for success or failure. Failure isn’t closure. The predicate itself succeeds after any terminal state.

`onSuccess`

Selects a signal’s successful value and closes for failure or closure:

const value = onSuccess(result)
const fluentValue = result.onSuccess()

const notice = onSuccess(result, (selected, source) =>
  sendEmail({ subject: "Completed", text: selected.summary }),
)
const fluentNotice = result.onSuccess((selected, source) =>
  sendEmail({ subject: "Completed", text: selected.summary }),
)

The section overload applies the selected value as a prerequisite and returns the section result. The value form preserves a keyed or global source annotation.

`onFailure`

Selects a signal’s normalized AutomationError and closes for success or closure:

const failure = onFailure(result)
const fluentFailure = result.onFailure()

const notice = onFailure(result, (selected, source) =>
  sendEmail({ subject: selected.code, text: selected.message }),
)
const fluentNotice = result.onFailure((selected, source) =>
  sendEmail({ subject: selected.code, text: selected.message }),
)

The section overload applies the failure as a prerequisite and returns the section result. AutomationError contains author-safe details, not raw errors or stacks.

`onClose`

Selects closure as Signal<null> and closes for success or failure:

const closure = onClose(result)
const fluentClosure = result.onClose()

const notice = onClose(result, (selected, source) =>
  sendEmail({ subject: "Skipped", text: "The route produced no value." }),
)
const fluentNotice = result.onClose((selected, source) =>
  sendEmail({ subject: "Skipped", text: "The route produced no value." }),
)

The section overload applies the closure as a prerequisite and returns the section result.

`markSignificant`

Marks the connected run significant only when the selected signal emits, then returns the original signal:

const marked = markSignificant(notice)
const fluentMarked = notice.markSignificant()

Recent runs hides runs without an emitted marked signal by default. Show insignificant runs reveals them as dimmed rows. Closure or failure doesn’t mark the run significant.

`scope`

Gives a synchronous section its own durable hook namespace:

const result = scope(() => sendEmail({ subject: "Scoped work", text: "Done." }))

Adding or removing declarations inside the scope doesn’t renumber hooks after it. scope adds no prerequisite or presentation metadata. branch, each, and delay create internal scopes for the same isolation.

`group`

Names a synchronous declaration group and gives it an isolated hook namespace:

const result = group({ name: "Notify members", presentation: "expanded" }, () =>
  sendEmail({ subject: "Grouped work", text: "Done." }),
)

Option	Default	Description
`name`	Required	Author-facing name shown in run details.
`presentation`	`"collapsed"`	Initial treatment: `"collapsed"`, `"expanded"`, or `"hidden"`.

Like scope, group doesn’t add a prerequisite. It returns the section result.

Types and errors

| Type | Contract | | --- | --- | --- | | Signal<T> | Symbolic value that materializes during execution. | | KeyedSignal<T> | Signal<T> with an encoded partition key from keyBy. | | GlobalSignal<T> | Signal<T> with one shared cross-context partition from globally. | | NonEmptyArray<T> | [T, ...T[]]. Complete collections and windows return this type. | | Duration | Milliseconds, a compact DurationString, or a compatible signal carrying either. | | Deadline | Date | Signal<Date>. | | CorrelationOptions | Options accepted by correlate. | | CorrelationInput | A keyed signal or { signal, selection?, consumption? }. | | AdmissionOptions | Optional claim ttl accepted by once and take. | | RateLimitOptions | Rolling limit, interval, and overflow behavior. | | ConcurrencyOptions | Positive active-region limit accepted by concurrent. | | ValueInheritanceOptions | Parent-value selection accepted by collect and window. | | WindowUntilOptions | ValueInheritanceOptions & { until: Deadline }. | | GroupOptions | Options accepted by group. | | FunnelOptions | Type-checked timing, edge, selection, and buffering combinations accepted by funnel. | SignalOutcome<T> is mutually exclusive:

type SignalOutcome<T> =
  | { status: "closed" }
  | { failure: AutomationError; status: "failed" }
  | { status: "succeeded"; value: T }

AutomationError contains author-safe failure data:

interface AutomationError {
  cause?: AutomationError
  code: string
  details?: JSONValue
  message: string
  name?: string
}

code is the stable machine-readable classification. message, optional name, JSON details, and a depth-limited cause are safe to expose to automation authors. Raw errors and stacks don’t enter signal values.

​Choose an operator family

​Property access

​transform

​t

​isSignal

​keyBy

​globally

​timestamp

​correlationId

​correlate

​merge

​collect

​once and take

​rateLimit

​each

​gate

​filter

​partition

​dependentOn

​withPrerequisites

​branch

​race

​serialize

​concurrent

​delay

​timeout

​funnel

​debounce

​window

​outcome

​succeeded

​failed

​closed

​onSuccess

​onFailure

​onClose

​markSignificant

​scope

​group

​Types and errors