Skip to content

Signals

Send data to a running execution, or start one for a key when none is running.

Listening for Signals

A handler listens for signals by name with ctx.onSignal():

const session = conductor.createTask(
  { name: "session" },
  { invocable: true },
  async (event, ctx) => {
    ctx.onSignal("input", async (payload, { id }) => {
      // delivery is at least once: dedupe by id
      if (await alreadyHandled(id)) return;
      await handleInput(payload);
    });

    await runUntilIdle();
  },
);
  • Signals stored before the handler started are delivered first, in the order they were sent. The worker reads them when it claims the execution, so a listener receives them as soon as it registers
  • Later signals are delivered while the handler runs. The worker reads them with its next claim, so they arrive within about one pollIntervalMs
  • Handlers run one at a time. The payload is any JSON value
  • A signal counts as delivered once its handler returns
  • Register listeners before the handler awaits anything. A signal reached while its name has no listener is skipped; if a listener for it registers later in the attempt, the execution is re-queued and the next attempt receives it
  • One listener per name and attempt. The handler re-runs from the top on every attempt, so register listeners again each time

Waiting for a Signal

An execution that has nothing to do can suspend until its next signal instead of ending, with a { signal: name } branch of ctx.waitForAny(). It can wait for other branches, such as a tool execution, at the same time:

const session = conductor.createTask(
  { name: "session" },
  { invocable: true },
  async (event, ctx) => {
    ctx.onSignal("input", (payload, { id }) => saveInput(id, payload));
    const toolId = await ctx.invoke("start-tool", { name: "run-tool" }, event.payload);

    for (let n = 0; ; n++) {
      await processSavedInputs(ctx, n);
      const winner = await ctx.waitForAny(
        `idle:${n}`,
        { input: { signal: "input" }, tool: { execution: toolId } },
        { timeout: "1h" },
      );
      if (winner.key === "timeout") return;
    }
  },
);
  • The branch wins at once when the execution holds a signal with that name whose handler has not returned, including one stored before the wait. Otherwise the next signal with that name, from conductor.signal(), ctx.sendSignal() or signalWithStart(), wakes the execution
  • The wait does not consume the signal. The next attempt delivers it to ctx.onSignal() handlers as usual; winner.signal only holds its id and name
  • Signals for other names are kept and do not wake the execution
  • A signal stays undelivered until a handler for its name returns, so register a handler for every signal branch. Without one, the signal wins every later wait at once

Sending Signals

Signal a known execution:

const stored = await conductor.signal(executionId, {
  name: "input",
  id: requestId,
  payload: { text: "hello" },
});
// false if the execution is completed, failed, cancelled or unknown

Each signal has an id. Sending an id again stores it once, so a retried send is safe.

Signal or Start

signalWithStart() makes a task execution addressable by a key. It signals the execution of the task for that key, or starts one that receives the signal first:

const executionId = await conductor.signalWithStart({
  task: "session",
  key: "slack:T1:C1:1712345678.000100",
  signal: { name: "input", id: requestId, payload },
  input: {}, // payload of the execution, used only when it starts one (default: {})
});
  • At most one non-terminal execution exists per task and key. Concurrent calls for a key start one execution
  • A completed, failed or cancelled execution does not hold its key, so the next call starts a new execution
  • The call runs in one transaction under a lock on the task and key

Sending from Tasks

await ctx.sendSignal(executionId, { name: "input", id: "report-1", payload });

// memoized by the step key like other context operations
const sessionId = await ctx.signalWithStart("start-session", {
  task: "session",
  key: threadKey,
  signal: { name: "input", id: messageId, payload },
});

ctx.signalWithStart() passes the metadata of the calling execution to an execution it starts, unless its metadata option overrides it. ctx.sendSignal() starts nothing and takes no metadata.

Delivery Guarantees

Delivery is at least once per signal id. A handler may see a signal again after a crash or a failed attempt, so make handlers idempotent by id.

No signal is lost when the handler returns:

  • Completion: when the handler returns, the attempt first delivers the signals it has already read to their listeners. Settlement then checks, in the same transaction, for signals the attempt listens for but did not deliver, which are ones stored after the worker last read them. If there are any, the execution is not completed but re-queued, and the new attempt receives them when it starts. Its result is discarded and the re-run does not count as an attempt
  • Signals with no listener: a signal whose name the attempt never listened for does not re-queue it. The execution completes and the signal is not delivered
  • Retries: a failed attempt delivers its remaining signals to the next attempt
  • Sleeping or waiting: signals for an execution that sleeps, waits for a child or waits for an event are kept and delivered when its handler next runs. A signal does not wake it, unless it waits in ctx.waitForAny() with a signal branch for its name
  • Pending executions receive their signals when they first run
  • Cancellation and permanent failure: undelivered signals are dropped. Signalling a cancelled execution returns false

When the handler returns while signal handlers are still running or signals it already read are waiting for them, the attempt waits for them before settling. timeoutMs and cancellation still apply while it waits, and a signal handler that suspends, for example with ctx.sleep(), suspends the attempt. Signals that arrive after the handler returned are left to the re-queued attempt.

A handler does not have to wait for its signals: one that registers listeners and returns right away still receives every signal stored before it started. Re-queues are not limited, because each one needs a signal stored after the previous attempt started, and the next attempt receives that signal when it starts. An execution therefore re-runs only while new signals keep arriving. To handle a stream of signals in one attempt, keep the handler running, for example until it has been idle for a while.

Errors

A signal handler that throws fails the attempt like an uncaught error in the handler, with the error Signal handler for "<name>" failed: <message>. The signal is not marked delivered, so the retry receives it again. During orchestrator.stop(), a signal handler that throws, or a signal that cannot be marked delivered, releases the attempt once the handler returns, without using up a retry.