feat(workflows): resume durable waits through shared queue

docs/workflows.md

@@ -58,9 +58,14 @@ Due-wait discovery uses PostgreSQL's clock. Resuming a wait compares the exact p
 well as the step status. A stale wake therefore cannot resume a step whose wait was cancelled or
 rescheduled after that wake was created.
 
-The execution slice that turns due workflow waits into existing one-shot wake/queue items is separate
-from this state foundation. Keeping the state transition separate from delivery lets the existing
-`work_items` lease/idempotency mechanism remain the single durable execution queue.
+Due workflow waits are bridged into the existing `work_items` queue. The queue remains the single
+durable execution mechanism: it owns claims, leases, retries and idempotency, while the workflow store
+owns state transitions. Each queued wake names the exact persisted wait timestamp; if a wait is
+cancelled or rescheduled before delivery, the old item cannot resume the step and is finished as stale.
+
+The bridge uses a bounded periodic recovery sweep rather than busy-waiting on an external website.
+Agents can sleep while external generation is pending, and a server/app restart simply leaves the
+persisted wait to be discovered and resumed later.
 
 ## Pause, resume and cancel
 

scripts/release-preflight.ts

@@ -1476,6 +1476,8 @@ function checkDurableWorkflowState(): void {
   const schema = read("server/src/db/schema/coworker.ts");
   const store = read("server/src/workflows/store.ts");
   const migration = read("server/drizzle/0043_workflow_state.sql");
+  const wake = read("server/src/workflows/wake.ts");
+  const server = read("server/src/index.ts");
 
   for (const evidence of [
     'workflowRunStatus = pgEnum("workflow_run_status"',
@@ -1503,6 +1505,21 @@ function checkDurableWorkflowState(): void {
     }
   }
 
+  for (const evidence of [
+    "WORKFLOW_WAIT_RESUME_KIND",
+    "resumeWaitingStep(",
+    "waitUntil.toISOString()",
+    "queue.renew({",
+    "queue.release({",
+  ]) {
+    if (!wake.includes(evidence)) {
+      fail(`workflows: durable wake bridge is missing ${evidence}`);
+    }
+  }
+  if (!server.includes("sweepWorkflowWaits(workflowWake)")) {
+    fail("workflows: durable wake bridge is not running from the server");
+  }
+
   for (const evidence of [
     'CREATE TYPE "public"."workflow_run_status"',
     'CREATE TYPE "public"."workflow_step_status"',

server/src/index.ts

@@ -109,6 +109,8 @@ import { grantedSkills, grantedTools, REFUSAL_MARKER } from "./plugins/tools";
 import { createTurnRunner } from "./routines/run-turn";
 import { createRoutineRunner } from "./routines/runner";
 import { createRoutineStore } from "./routines/store";
+import { createWorkflowStore } from "./workflows/store";
+import { sweepWorkflowWaits } from "./workflows/wake";
 import { createIntentRouter } from "./routing/classify";
 import { createModelCompleter } from "./routing/model";
 import {
@@ -416,6 +418,7 @@ const pluginStore = createPluginStore({
  */
 const routineStore = createRoutineStore(database);
 useRoutineTools(routineStore);
+const workflowStore = createWorkflowStore(database);
 
 /**
  * Other Bots this person deliberately put in the same live channel as this Bot.
@@ -1282,6 +1285,38 @@ if (config.handoff.maxDepth > 0 && config.handoff.maxPerRun > 0) {
   repeatAfterEach(kick, 2_000);
 }
 
+/*
+ * Durable workflow waits.
+ *
+ * This is a recovery/execution bridge, not a second scheduler or lease system. The workflow store
+ * decides which exact persisted waits are due using Postgres' clock; the shared work_items queue
+ * owns claiming, leases, retries and idempotency. A stale wake is harmless because resumption
+ * compare-and-sets the exact wait timestamp before changing the step.
+ */
+const workflowWake = {
+  store: workflowStore,
+  queue: createWorkQueue(database),
+  owner: workOwner("workflow-wake"),
+};
+
+repeatAfterEach(async () => {
+  try {
+    const report = await sweepWorkflowWaits(workflowWake);
+    if (
+      report.queued > 0 ||
+      report.resumed.length > 0 ||
+      report.skipped.length > 0
+    ) {
+      console.info(JSON.stringify({ type: "workflow-wake", ...report }));
+    }
+  } catch (error) {
+    console.warn(
+      "[workflows] due waits could not be resumed:",
+      error instanceof Error ? error.message : error,
+    );
+  }
+}, 5_000);
+
 /*
  * And dropping the hops that are over, whether or not the capability is switched on.
  *

server/src/workflows/wake.ts

@@ -0,0 +1,191 @@
+import type { WorkflowStore } from "./store";
+import { DEFAULT_MAX_ATTEMPTS, type WorkQueue } from "../work/queue";
+
+export const WORKFLOW_WAIT_RESUME_KIND = "workflow_wait_resume";
+
+const DEFAULT_LIMIT = 50;
+const DEFAULT_LEASE_MS = 30_000;
+const DEFAULT_RETRY_DELAY_MS = 5_000;
+
+type WorkflowWakeStore = Pick<
+  WorkflowStore,
+  "dueWaitingSteps" | "resumeWaitingStep"
+>;
+
+export type WorkflowWakeOptions = {
+  store: WorkflowWakeStore;
+  queue: WorkQueue;
+  owner: string;
+  limit?: number;
+  leaseMs?: number;
+  retryDelayMs?: number;
+  maxAttempts?: number;
+};
+
+export type WorkflowWakeReport = {
+  considered: number;
+  resumed: Array<{ workflowId: string; stepKey: string }>;
+  skipped: Array<{ workflowId: string; stepKey: string; reason: string }>;
+};
+
+function wakeKey(workflowId: string, stepKey: string, waitUntil: Date): string {
+  return `${workflowId}:${stepKey}:${waitUntil.toISOString()}`;
+}
+
+export async function offerDueWorkflowWaits(
+  options: WorkflowWakeOptions,
+): Promise<{ queued: number; already: number }> {
+  const waits = await options.store.dueWaitingSteps(
+    options.limit ?? DEFAULT_LIMIT,
+  );
+  let queued = 0;
+  let already = 0;
+
+  for (const wait of waits) {
+    const outcome = await options.queue.offer({
+      kind: WORKFLOW_WAIT_RESUME_KIND,
+      key: wakeKey(wait.workflowId, wait.stepKey, wait.waitUntil),
+      payload: {
+        ownerUserId: wait.ownerUserId,
+        agentId: wait.agentId,
+        workflowId: wait.workflowId,
+        stepKey: wait.stepKey,
+        waitUntil: wait.waitUntil.toISOString(),
+      },
+    });
+    if (outcome === "queued") queued += 1;
+    if (outcome === "already") already += 1;
+  }
+
+  return { queued, already };
+}
+
+export async function dispatchClaimedWorkflowWaits(
+  options: WorkflowWakeOptions,
+): Promise<WorkflowWakeReport> {
+  const leaseMs = options.leaseMs ?? DEFAULT_LEASE_MS;
+  const claimed = await options.queue.claim({
+    kind: WORKFLOW_WAIT_RESUME_KIND,
+    owner: options.owner,
+    leaseMs,
+    limit: options.limit ?? DEFAULT_LIMIT,
+    ...(options.maxAttempts === undefined
+      ? {}
+      : { maxAttempts: options.maxAttempts }),
+  });
+
+  const report: WorkflowWakeReport = {
+    considered: claimed.length,
+    resumed: [],
+    skipped: [],
+  };
+
+  for (const item of claimed) {
+    const workflowId =
+      typeof item.payload.workflowId === "string"
+        ? item.payload.workflowId
+        : "";
+    const stepKey =
+      typeof item.payload.stepKey === "string" ? item.payload.stepKey : "";
+    const ownerUserId =
+      typeof item.payload.ownerUserId === "string"
+        ? item.payload.ownerUserId
+        : "";
+    const agentId =
+      typeof item.payload.agentId === "string" ? item.payload.agentId : "";
+    const stamp =
+      typeof item.payload.waitUntil === "string"
+        ? new Date(item.payload.waitUntil)
+        : null;
+
+    if (
+      !workflowId ||
+      !stepKey ||
+      !ownerUserId ||
+      !agentId ||
+      !stamp ||
+      Number.isNaN(stamp.getTime())
+    ) {
+      const reason = "workflow wake payload is incomplete or invalid";
+      await options.queue.finish({
+        kind: WORKFLOW_WAIT_RESUME_KIND,
+        key: item.key,
+        owner: options.owner,
+      });
+      report.skipped.push({ workflowId, stepKey, reason });
+      continue;
+    }
+
+    const renewed = await options.queue.renew({
+      kind: WORKFLOW_WAIT_RESUME_KIND,
+      key: item.key,
+      owner: options.owner,
+      leaseMs,
+    });
+    if (!renewed) {
+      report.skipped.push({
+        workflowId,
+        stepKey,
+        reason: "the lease went to another replica",
+      });
+      continue;
+    }
+
+    try {
+      await options.store.resumeWaitingStep(
+        { ownerUserId, agentId },
+        workflowId,
+        stepKey,
+        stamp,
+      );
+      await options.queue.finish({
+        kind: WORKFLOW_WAIT_RESUME_KIND,
+        key: item.key,
+        owner: options.owner,
+      });
+      report.resumed.push({ workflowId, stepKey });
+    } catch (error) {
+      const reason =
+        error instanceof Error
+          ? error.message
+          : "workflow wait could not resume";
+
+      // A changed/cancelled/stale wait is final for this exact timestamp. The
+      // store's compare-and-set refusal is what makes an old queue item harmless.
+      if (
+        /not due|changed after this wake|does not exist|active workflow/i.test(
+          reason,
+        )
+      ) {
+        await options.queue.finish({
+          kind: WORKFLOW_WAIT_RESUME_KIND,
+          key: item.key,
+          owner: options.owner,
+        });
+        report.skipped.push({ workflowId, stepKey, reason });
+        continue;
+      }
+
+      await options.queue.release({
+        kind: WORKFLOW_WAIT_RESUME_KIND,
+        key: item.key,
+        owner: options.owner,
+        delayMs: options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS,
+        reason,
+      });
+      report.skipped.push({ workflowId, stepKey, reason });
+    }
+  }
+
+  return report;
+}
+
+export async function sweepWorkflowWaits(
+  options: WorkflowWakeOptions,
+): Promise<WorkflowWakeReport & { queued: number; already: number }> {
+  const offered = await offerDueWorkflowWaits(options);
+  const dispatched = await dispatchClaimedWorkflowWaits(options);
+  return { ...offered, ...dispatched };
+}
+
+export const WORKFLOW_WAKE_MAX_ATTEMPTS = DEFAULT_MAX_ATTEMPTS;