feat(conductor): middleware that runs on every attempt

[psteinroe]

Adds Conductor.create({ middleware: [...] }): functions that run around every attempt of a task handler, including each resume after ctx.sleep(), ctx.waitForEvent() and ctx.invoke(). The static context can't rebuild live objects such as clients or thread handles from execution data on each start and resume. Middleware can.

Conductor.create({
  sql,
  context: { db },
  metadata: z.object({ tenant: z.string() }),
  middleware: [
    async ({ execution, ctx }, next) =>
      als.run({ executionId: execution.id }, async () =>
        next({ ...ctx, tenant: await load(execution.metadata?.tenant) }),
      ),
  ],
});

Typing

Keys passed to next are inferred into the handler ctx type (a reverse-mapped tuple over the middleware array, like Inngest's transformInput). Middleware runs in array order and sees earlier additions at runtime. Its own ctx parameter is typed with the base context only. next returns an opaque MiddlewareResult, so a middleware must return what next returned. Middleware, MiddlewareExecution and MiddlewareResult are exported.

execution, metadata and resumed

execution exposes:

• id, task_key, queue, attempt and parent_execution_id.
• metadata: the same value as ctx.metadata from feat(executions): typed, propagated execution metadata #72, typed by the Conductor's metadata schema and undefined when the execution has none.
• resumed: true when earlier attempts left durable progress. That means a persisted step (including sleeps, received events and child results), a pending event wait (execution_wait subscription), or a pending child invocation (waiting_on_execution_id).

These values come from the claim query, which now also returns attempts and parent_execution_id. Computing resumed adds two indexed exists probes per claimed row. The root id is not included because no column stores it.

Abort safety

Hang-ups don't reject: abortAndHangup aborts and returns a promise that never settles. So that try/finally cleanup still runs, next() resolves when the attempt suspends, is cancelled or times out, and rejects only when the handler throws. The worker races the whole chain against the abort promise, and that race always settles first. Whatever middleware returns, throws or swallows after an abort is therefore ignored. Middleware errors fail the attempt and it is retried, the same as handler errors.

Scope

Middleware is Conductor-level only. Batch tasks skip it, the same way they already receive no custom context: their context has no steps and covers many executions. The internal maintenance and event-dispatch tasks skip it too.

Verification

• tests/integration/middleware.test.ts:
  • middleware runs once per attempt across sleep, event and child resumes, with the expected resumed, parent and propagated metadata values
  • AsyncLocalStorage and finally wrapping works on suspension and on completion
  • a middleware error fails attempt 1, and attempt 2 succeeds
  • a middleware that catches and swallows errors does not break suspension
  • batch, maintenance and event-dispatch tasks skip middleware
• tests/unit/middleware-types.test.ts: chained additions, metadata typed by the schema, middleware declared separately, no middleware, and that middleware must return what next returned
• just lint, bun run typecheck and the full bun test pass (409 tests)

Follow-ups / out of scope

• Attempt counter after ctx.invoke (out of scope): resuming after ctx.invoke increments attempts. updated_invoke_parents doesn't decrement it, while released and event-wait suspensions do. The parent's execution.attempt reads 2 after one child resume, and every child invocation uses up one of the parent's maxAttempts. This looks like a bug and is not fixed here.