From fb9afaf1e3826bb4b2d3e751e7e1227e6852bd0b Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:38:35 +0700 Subject: [PATCH 1/7] feat(workflows): bridge due waits into durable queue --- server/src/workflows/wake.ts | 192 +++++++++++++++++++++++++++++++++++ 1 file changed, 192 insertions(+) create mode 100644 server/src/workflows/wake.ts diff --git a/server/src/workflows/wake.ts b/server/src/workflows/wake.ts new file mode 100644 index 000000000..2f2f03685 --- /dev/null +++ b/server/src/workflows/wake.ts @@ -0,0 +1,192 @@ +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 { + 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 { + const offered = await offerDueWorkflowWaits(options); + const dispatched = await dispatchClaimedWorkflowWaits(options); + return { ...offered, ...dispatched }; +} + +export const WORKFLOW_WAKE_MAX_ATTEMPTS = DEFAULT_MAX_ATTEMPTS; From 36302ec3fa15ee4eb95ef6a948ab02e19ac30315 Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:38:37 +0700 Subject: [PATCH 2/7] test(workflows): cover durable wait bridge --- server/tests/workflow-wake.test.ts | 183 +++++++++++++++++++++++++++++ 1 file changed, 183 insertions(+) create mode 100644 server/tests/workflow-wake.test.ts diff --git a/server/tests/workflow-wake.test.ts b/server/tests/workflow-wake.test.ts new file mode 100644 index 000000000..5b0ab7462 --- /dev/null +++ b/server/tests/workflow-wake.test.ts @@ -0,0 +1,183 @@ +import { describe, expect, test } from "bun:test"; +import type { WorkItem, WorkQueue } from "../src/work/queue"; +import { + dispatchClaimedWorkflowWaits, + offerDueWorkflowWaits, + WORKFLOW_WAIT_RESUME_KIND, +} from "../src/workflows/wake"; + +function queueStub(overrides: Partial = {}): WorkQueue { + return { + offer: async () => "queued", + claim: async () => [], + renew: async () => true, + finish: async () => true, + release: async () => true, + purge: async () => 0, + ...overrides, + }; +} + +describe("workflow wake bridge", () => { + test("offers one idempotent item for the exact persisted wait", async () => { + const waitUntil = new Date("2026-09-20T07:30:00.000Z"); + const offered: Array> = []; + const queue = queueStub({ + offer: async (item) => { + offered.push(item as unknown as Record); + return "queued"; + }, + }); + + const result = await offerDueWorkflowWaits({ + owner: "worker-1", + queue, + store: { + dueWaitingSteps: async () => [ + { + ownerUserId: "user-1", + agentId: "bot-1", + workflowId: "workflow-1", + stepKey: "render", + waitUntil, + }, + ], + resumeWaitingStep: async () => { + throw new Error("not used"); + }, + }, + }); + + expect(result).toEqual({ queued: 1, already: 0 }); + expect(offered[0]).toMatchObject({ + kind: WORKFLOW_WAIT_RESUME_KIND, + key: "workflow-1:render:2026-09-20T07:30:00.000Z", + payload: { + ownerUserId: "user-1", + agentId: "bot-1", + workflowId: "workflow-1", + stepKey: "render", + waitUntil: "2026-09-20T07:30:00.000Z", + }, + }); + }); + + test("resumes with identity and exact timestamp from the durable item", async () => { + const item: WorkItem = { + kind: WORKFLOW_WAIT_RESUME_KIND, + key: "wake-1", + attempts: 1, + payload: { + ownerUserId: "user-1", + agentId: "bot-1", + workflowId: "workflow-1", + stepKey: "render", + waitUntil: "2026-09-20T07:30:00.000Z", + }, + }; + const calls: unknown[][] = []; + const report = await dispatchClaimedWorkflowWaits({ + owner: "worker-1", + queue: queueStub({ claim: async () => [item] }), + store: { + dueWaitingSteps: async () => [], + resumeWaitingStep: async (...args) => { + calls.push(args); + return {} as never; + }, + }, + }); + + expect(report.resumed).toEqual([ + { workflowId: "workflow-1", stepKey: "render" }, + ]); + expect(calls).toHaveLength(1); + expect(calls[0]?.[0]).toEqual({ + ownerUserId: "user-1", + agentId: "bot-1", + }); + expect((calls[0]?.[3] as Date).toISOString()).toBe( + "2026-09-20T07:30:00.000Z", + ); + }); + + test("finishes a stale exact wake instead of retrying it", async () => { + let finished = 0; + let released = 0; + const report = await dispatchClaimedWorkflowWaits({ + owner: "worker-1", + queue: queueStub({ + claim: async () => [ + { + kind: WORKFLOW_WAIT_RESUME_KIND, + key: "wake-1", + attempts: 1, + payload: { + ownerUserId: "user-1", + agentId: "bot-1", + workflowId: "workflow-1", + stepKey: "render", + waitUntil: "2026-09-20T07:30:00.000Z", + }, + }, + ], + finish: async () => { + finished += 1; + return true; + }, + release: async () => { + released += 1; + return true; + }, + }), + store: { + dueWaitingSteps: async () => [], + resumeWaitingStep: async () => { + throw new Error( + "That wait is not due, or it was changed after this wake was scheduled.", + ); + }, + }, + }); + + expect(finished).toBe(1); + expect(released).toBe(0); + expect(report.resumed).toEqual([]); + expect(report.skipped).toHaveLength(1); + }); + + test("releases transient failures with backoff", async () => { + let delayMs = 0; + await dispatchClaimedWorkflowWaits({ + owner: "worker-1", + retryDelayMs: 7_000, + queue: queueStub({ + claim: async () => [ + { + kind: WORKFLOW_WAIT_RESUME_KIND, + key: "wake-1", + attempts: 1, + payload: { + ownerUserId: "user-1", + agentId: "bot-1", + workflowId: "workflow-1", + stepKey: "render", + waitUntil: "2026-09-20T07:30:00.000Z", + }, + }, + ], + release: async (input) => { + delayMs = input.delayMs; + return true; + }, + }), + store: { + dueWaitingSteps: async () => [], + resumeWaitingStep: async () => { + throw new Error("database temporarily unavailable"); + }, + }, + }); + expect(delayMs).toBe(7_000); + }); +}); From 80d162022522fa9a18898b83f9ea6987b4e86dbe Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:39:01 +0700 Subject: [PATCH 3/7] feat(workflows): run durable wait bridge --- server/src/index.ts | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/server/src/index.ts b/server/src/index.ts index 4a3d0a1f1..ac6970c95 100644 --- a/server/src/index.ts +++ b/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. * From d09f30fe5dc8c8e61195286b08d00f580a55e8bd Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:39:05 +0700 Subject: [PATCH 4/7] docs(workflows): document durable wake execution --- docs/workflows.md | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/docs/workflows.md b/docs/workflows.md index c9670c1d4..7402b2609 100644 --- a/docs/workflows.md +++ b/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 From 240d143f140f831fe9e88cadb0966673dbe8d7e4 Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:39:22 +0700 Subject: [PATCH 5/7] test(release): pin workflow wake bridge --- scripts/release-preflight.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/scripts/release-preflight.ts b/scripts/release-preflight.ts index 9230a4ce2..2155b6458 100644 --- a/scripts/release-preflight.ts +++ b/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"', From b372c6724daf15c6ba6ecb09e9fea060344b1562 Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:40:44 +0700 Subject: [PATCH 6/7] style(workflows): apply biome formatting --- server/src/workflows/wake.ts | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/server/src/workflows/wake.ts b/server/src/workflows/wake.ts index 2f2f03685..ab5448734 100644 --- a/server/src/workflows/wake.ts +++ b/server/src/workflows/wake.ts @@ -1,8 +1,5 @@ import type { WorkflowStore } from "./store"; -import { - DEFAULT_MAX_ATTEMPTS, - type WorkQueue, -} from "../work/queue"; +import { DEFAULT_MAX_ATTEMPTS, type WorkQueue } from "../work/queue"; export const WORKFLOW_WAIT_RESUME_KIND = "workflow_wait_resume"; @@ -149,7 +146,9 @@ export async function dispatchClaimedWorkflowWaits( report.resumed.push({ workflowId, stepKey }); } catch (error) { const reason = - error instanceof Error ? error.message : "workflow wait could not resume"; + 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. From 7a43061b3cbf3b79cba1819519d1e6f9d4ac5854 Mon Sep 17 00:00:00 2001 From: duc15052006-dotcom Date: Sat, 19 Sep 2026 19:42:10 +0700 Subject: [PATCH 7/7] test(workflows): avoid unsafe optional chaining --- server/tests/workflow-wake.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/tests/workflow-wake.test.ts b/server/tests/workflow-wake.test.ts index 5b0ab7462..f6c20b7d7 100644 --- a/server/tests/workflow-wake.test.ts +++ b/server/tests/workflow-wake.test.ts @@ -96,7 +96,7 @@ describe("workflow wake bridge", () => { ownerUserId: "user-1", agentId: "bot-1", }); - expect((calls[0]?.[3] as Date).toISOString()).toBe( + expect((calls[0]?.[3] as Date | undefined)?.toISOString()).toBe( "2026-09-20T07:30:00.000Z", ); });