Proposal: messages, run-addressed and buffered (alongside signals) #894

Follow-up to #877 (item 3). You mentioned you're asking people how they use signals before deciding anything, and that you're open to a proposal in the meantime. Here's what we'd need, as input to that. We aren't asking for a decision now, and there's no PR.

Desired Behavior

A new primitive, messages, that sits alongside signals rather than replacing them. A message is addressed to one workflow run and buffered until that run consumes it.

Why signals don't fit these cases. Today sendSignal delivers only to steps that are waiting at that moment; with no waiter, the data is dropped. That's a race for the most common uses:

• A human approves while the run is still on an earlier step, or during a restart.
• A webhook callback arrives before the run reaches waitForSignal.
• A parent sends a child an instruction before the child has parked.

The signals docs' own approval and Stripe-webhook examples have this race.

Semantics:

• Addressed to a run. sendMessage({ workflowRunId, name, data, idempotencyKey? }). The backend deals only in run ids. Apps that need a business key keep their own mapping, or look the run up first.
• Buffered, FIFO, consumed exactly once. Each step.waitForMessage({ name }) takes the oldest unconsumed message with that name, and the message is never consumed twice.
  • The claim is tied to the wait's step attempt, so a crash between claiming and completing the step re-delivers the same message to the same step on replay.
  • Waits that repeat, such as one per loop iteration, use a name per iteration, for example approval:${iteration}.
• Idempotent sends. idempotencyKey is unique per run, so a double-click or a retried webhook becomes one message.
• Clear errors at the edges. A run id that doesn't exist throws. A run that has already finished rejects the message without storing it, so "you were too late" is visible to the sender. Unconsumed messages stay on a finished run, for debugging, until the run is deleted.
• Wakes only a matching wait. If the run is parked in waitForMessage for that name, it wakes immediately. Otherwise the message is only stored, so sleeping runs aren't replayed for nothing.
  • The race where the waiter checks its inbox while the sender checks for waiters is closed where sleepWorkflowRun already handles delivered signals. Its UPDATE also wakes the run when an unconsumed message matches a running message-wait, and the run row serializes both sides.
• Reruns start with an empty inbox. Completed message-waits are copied with their outputs like any other step; unread messages belonged to the old run.
• Inspectable. listMessages({ workflowRunId }) returns pending and consumed messages, with the step that consumed each one, for dashboards and debugging.
• Signals unchanged. They keep the broadcast job: tell whoever is listening right now. The docs would gain a "Signals vs messages" table, and the approval and webhook examples would move to the messages page.

Backend contract. One table and three methods, plus one clause in an existing method:

workflow_messages (namespace_id, id) PK
  workflow_run_id → workflow_runs ON DELETE CASCADE
  name, data, idempotency_key, consumed_by_step_attempt_id, consumed_at, created_at
  UNIQUE (namespace_id, workflow_run_id, idempotency_key) WHERE idempotency_key IS NOT NULL
  INDEX  (namespace_id, workflow_run_id, name, created_at, id) WHERE consumed_by_step_attempt_id IS NULL

• sendMessage: checks the run exists and hasn't finished, deduplicates on the key, wakes the run only if it's parked on a matching wait.
• consumeMessage({ workflowRunId, stepAttemptId, name }): returns the message this attempt already claimed, or claims the oldest unconsumed one (FOR UPDATE SKIP LOCKED on Postgres; the immediate transaction on SQLite).
• listMessages: paginated with the existing cursor helpers.
• sleepWorkflowRun: the park check also looks for unconsumed messages that match a running message-wait.

Example Usage

// a workflow waiting for a human
const decision = await step.waitForMessage({
  name: "approval",
  schema: z.object({ approved: z.boolean(), by: z.string() }),
  timeout: "7d",
});
if (decision === null) return { status: "expired" };

// app code: safe to call before the run reaches the wait, during a restart, or twice
const run = await ow.runWorkflow(purchase.spec, input);
await ow.sendMessage({
  workflowRunId: run.workflowRun.id,
  name: "approval",
  data: { approved: true, by: "alice" },
  idempotencyKey: `approval:${purchaseId}`,
});

// from another workflow, recorded as a step so a replay doesn't resend it
await step.sendMessage({ workflowRunId: childId, name: "go", data: {} });

How Last Light would use it

Last Light is moving its workflow engine onto OpenWorkflow (nearform/lastlight#435). Human approval gates are central: a build workflow parks for review sign-off, and approvals arrive from GitHub comments, Slack or our dashboard, sometimes days later.

Today, without messages:

• Our own approvals table is the source of truth, and the signal is only a wake-up.
• ctx.approval() checks the approvals table before parking, in case the decision arrived early.
• A reconciler sweep re-sends the signal for runs where we crashed between recording a decision and sending it.

That's three moving parts to make one delivery reliable.

Admission makes the gap wide on purpose. Last Light caps how many runs execute at once. When a parked run is approved, we record the decision immediately but release the run only when a slot frees, so a burst of approvals doesn't start a burst of agent sessions. The send therefore happens long after the decision.

With messages:

• ctx.approval(gate) becomes step.waitForMessage({ name: \approval:${gate}` }), and approving becomes one sendMessage`. The approvals table is kept for audit and the dashboard only.
• The check-before-park step and the reconciler sweep are deleted.
• When admission releases the run, sending the message is enough; whether the run has parked yet doesn't matter.
• Approving a run that was canceled or superseded in the meantime fails visibly, so we can reply on the PR instead of recording a decision nobody will act on.
• Reviewer loops wait once per iteration, so gates are named per iteration.
• listMessages feeds our run timeline, so an approval that arrived before the gate shows up as one.
• We'd implement the new table and methods in our Drizzle backend (openworkflow-drizzle) and run the testBackend cases against it.

Questions

1. Does "messages" (sendMessage / waitForMessage) work as the name, or do you have another in mind?
2. Should handle.sendMessage(...) sugar on the handle runWorkflow returns be in the first version, or added later?
3. Would you rather this land after your concurrency work, since both change the Backend contract?

If the design fits what you hear from users, we're happy to implement it, including the testBackend cases (an early send, a send between the wait and the park, a duplicate idempotency key, a finished run, a crash after the claim) and the docs.