From 6dd4134a2a58e148544a860808450cd1cfb10156 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 2 Nov 2026 16:37:48 +0000 Subject: [PATCH 1/4] test(server): reproduce #29440, coalesced webhook windows dispatch the prompt from when they opened Co-Authored-By: Claude Opus 5.5 --- .../ScheduledTaskService.webhook.test.ts | 34 +++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 6fb5de1ff8873..184112245bf7a 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1031,6 +1031,40 @@ it.effect.each([1, 3])( }, ); +it.effect("a coalesced window dispatches the prompt saved after it opened", () => { + const workers = new Map>(); + return withService( + ({ service, launches }) => + Effect.gen(function* () { + const schedule = { + type: "webhook", + dedupe: { coalesceKey: "{{body.id}}", coalesceWindowSeconds: 10 }, + }; + const { task } = yield* service.upsert( + yield* webhookTaskInput({ prompt: "Use the original instructions.", schedule }), + ); + const bodyText = JSON.stringify({ id: "review" }); + yield* service.triggerWebhook( + requestFor(task, { bodyText, body: new TextEncoder().encode(bodyText) }), + ); + yield* service.upsert( + yield* webhookTaskInput({ prompt: "Use the updated instructions.", schedule }), + ); + yield* TestClock.adjust("10 seconds"); + yield* workers.get("webhook-coalescing")!; + const launched = yield* Queue.take(launches); + assert.equal(launched.initialMessage?.text, "Use the updated instructions."); + const owner = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries[0]!; + const { delivery } = yield* service.getWebhookDelivery({ + id: task.id, + deliveryId: owner.id, + }); + assert.equal(delivery.renderedPrompt, "Use the updated instructions."); + }), + { workers }, + ); +}); + it.effect("recovers a persisted coalescing window after restarting the service", () => { const workers = new Map>(); return withService( From 3dbe9d1b4cc8badeb3d19161f8e5a087d80e7149 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 2 Nov 2026 16:45:28 +0000 Subject: [PATCH 2/4] fix(server): coalesced webhook windows dispatch the current prompt Co-Authored-By: Claude Opus 5.5 --- apps/server/src/scheduledTasks/ScheduledTaskService.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 3d91ffa623a6c..715ac72de7385 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2330,11 +2330,14 @@ export const layer = Layer.effect( yield* remove; continue; } + const current = yield* findTask(taskId); const prepared = yield* Effect.gen(function* () { + // The snapshot keeps the identity runTask checks; the prompt is the one saved now. const task = yield* decodePendingTask(entry.task_json); const request = yield* decodePendingRequest(entry.request_json); const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); - return { task, rendered: renderWebhookPrompt(task.prompt, request, frameId) }; + const prompt = current?.createdAt === task.createdAt ? current.prompt : task.prompt; + return { task, rendered: renderWebhookPrompt(prompt, request, frameId) }; }).pipe(Effect.exit); if (Exit.isFailure(prepared)) { if (Cause.hasInterruptsOnly(prepared.cause)) From ee9ecd218060d24c9086e57d418e41fa024f326b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 2 Nov 2026 16:56:30 +0000 Subject: [PATCH 3/4] fix(server): keep the dispatched webhook prompt through recovery A running window replays its first dispatch, so it keeps the prompt that dispatch saved into the pending row instead of re-reading the task. A task row that cannot be read falls back to the snapshot, so runTask fails the delivery as before. Co-Authored-By: Claude Opus 5.5 --- .../scheduledTasks/ScheduledTaskService.ts | 26 ++++++-- .../ScheduledTaskService.webhook.test.ts | 63 +++++++++++++++++++ .../src/scheduledTasks/webhookDedupe.ts | 2 +- 3 files changed, 84 insertions(+), 7 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 715ac72de7385..9d58382860081 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -66,6 +66,7 @@ import { acceptImmediateWebhookDedupe, releaseWebhookReplay, decodePendingTask, + encodePendingTask, decodePendingRequest, dedupeError, } from "./webhookDedupe.ts"; @@ -880,7 +881,6 @@ export const layer = Layer.effect( : `${active.id}:webhook:${webhook.deliveryId}`; const commandId = CommandId.make(`scheduled-task:${fireKey}`); const messageId = MessageId.make(`scheduled-task-message:${fireKey}`); - // A webhook prompt was rendered from the row when the request arrived. const prompt = webhook?.prompt ?? active.prompt; // A bound thread can move to another project (No project to a project) while @@ -2330,14 +2330,26 @@ export const layer = Layer.effect( yield* remove; continue; } - const current = yield* findTask(taskId); + // A running row was already sent, so recovery keeps the prompt it was sent with. + // A failed load falls back to the snapshot and runTask fails the delivery. + const current = + entry.state === "dispatching" + ? yield* findTask(taskId).pipe(Effect.orElseSucceed(() => null)) + : null; const prepared = yield* Effect.gen(function* () { + const snapshot = yield* decodePendingTask(entry.task_json); // The snapshot keeps the identity runTask checks; the prompt is the one saved now. - const task = yield* decodePendingTask(entry.task_json); + const task = + current?.createdAt === snapshot.createdAt && current.prompt !== snapshot.prompt + ? { ...snapshot, prompt: current.prompt } + : snapshot; const request = yield* decodePendingRequest(entry.request_json); const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); - const prompt = current?.createdAt === task.createdAt ? current.prompt : task.prompt; - return { task, rendered: renderWebhookPrompt(prompt, request, frameId) }; + return { + task, + taskJson: task === snapshot ? null : yield* encodePendingTask(task), + rendered: renderWebhookPrompt(task.prompt, request, frameId), + }; }).pipe(Effect.exit); if (Exit.isFailure(prepared)) { if (Cause.hasInterruptsOnly(prepared.cause)) @@ -2350,11 +2362,13 @@ export const layer = Layer.effect( yield* remove; continue; } - const { task, rendered } = prepared.value; + const { task, taskJson, rendered } = prepared.value; if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { yield* markDeliveryFailed(entry.delivery_id, PROMPT_TOO_LONG); yield* releaseFailedDelivery(entry.delivery_id); } else { + if (taskJson !== null) + yield* sql`UPDATE scheduled_task_webhook_pending SET task_json = ${taskJson} WHERE delivery_id = ${entry.delivery_id}`; yield* sql`UPDATE scheduled_task_webhook_deliveries SET outcome = 'accepted', error = NULL, rendered_prompt = ${rendered.prompt.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT)}, missing_fields_json = ${encodeMissingFieldsJson(rendered.missing)} WHERE delivery_id = ${entry.delivery_id}`; if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) return; } diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 184112245bf7a..1e5f9a379624f 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1065,6 +1065,69 @@ it.effect("a coalesced window dispatches the prompt saved after it opened", () = ); }); +it.effect("retrying a running window keeps the prompt it was dispatched with", () => { + const workers = new Map>(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const schedule = { + type: "webhook", + dedupe: { coalesceKey: "{{body}}", coalesceWindowSeconds: 10 }, + }; + const save = (prompt: string) => + Effect.flatMap(webhookTaskInput({ prompt, schedule }), service.upsert); + const { task } = yield* save("Prompt when the window opened"); + yield* service.triggerWebhook(requestFor(task)); + yield* save("Prompt at dispatch"); + yield* sql`CREATE TRIGGER fail_completion BEFORE UPDATE ON scheduled_tasks + WHEN NEW.last_run_status = 'succeeded' BEGIN SELECT RAISE(FAIL, 'completion unavailable'); END`; + yield* TestClock.adjust("10 seconds"); + yield* drainPendingDispatch(workers); + const launched = yield* Queue.take(launches); + assert.equal(launched.initialMessage?.text, "Prompt at dispatch"); + yield* save("Prompt saved after dispatch"); + yield* sql`DROP TRIGGER fail_completion`; + yield* drainPendingDispatch(workers); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`).length, 0); + const owner = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries[0]!; + const { delivery } = yield* service.getWebhookDelivery({ + id: task.id, + deliveryId: owner.id, + }); + assert.equal(delivery.renderedPrompt, "Prompt at dispatch"); + }), + { workers }, + ); +}); + +it.effect("a coalesced window fails its delivery when the task row cannot be read", () => { + const workers = new Map>(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + schedule: { + type: "webhook", + dedupe: { coalesceKey: "{{body}}", coalesceWindowSeconds: 10 }, + }, + }), + ); + yield* service.triggerWebhook(requestFor(task)); + yield* sql`UPDATE scheduled_tasks SET model_selection_json = 'not json' WHERE task_id = ${task.id}`; + yield* TestClock.adjust("10 seconds"); + yield* drainPendingDispatch(workers); + assert.equal(yield* Queue.size(launches), 0); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`).length, 0); + const [delivery] = yield* sql<{ + outcome: string; + }>`SELECT outcome FROM scheduled_task_webhook_deliveries`; + assert.equal(delivery?.outcome, "dispatch_failed"); + }), + { workers }, + ); +}); + it.effect("recovers a persisted coalescing window after restarting the service", () => { const workers = new Map>(); return withService( diff --git a/apps/server/src/scheduledTasks/webhookDedupe.ts b/apps/server/src/scheduledTasks/webhookDedupe.ts index 21d40b25ac5ed..76f04a97e6dfe 100644 --- a/apps/server/src/scheduledTasks/webhookDedupe.ts +++ b/apps/server/src/scheduledTasks/webhookDedupe.ts @@ -111,7 +111,7 @@ export const acceptImmediateWebhookDedupe = Effect.fn( }); const PendingTaskJson = Schema.fromJsonString(ScheduledTask); -const encodePendingTask = Schema.encodeEffect(PendingTaskJson); +export const encodePendingTask = Schema.encodeEffect(PendingTaskJson); export const decodePendingTask = Schema.decodeUnknownEffect(PendingTaskJson); const PendingRequestJson = Schema.fromJsonString( Schema.Struct({ From 63563e1c4c79672c442e785f74b3aef4198a2164 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:14:47 -0800 Subject: [PATCH 4/4] fix(server): a failed task read fails the coalesced delivery before dispatch Co-Authored-By: Claude Opus 5.5 --- apps/server/src/scheduledTasks/ScheduledTaskService.ts | 8 ++------ .../scheduledTasks/ScheduledTaskService.webhook.test.ts | 5 ++++- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 82ea5f5868ebe..afeb73de3df96 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2330,13 +2330,9 @@ export const layer = Layer.effect( yield* remove; continue; } - // A running row was already sent, so recovery keeps the prompt it was sent with. - // A failed load falls back to the snapshot and runTask fails the delivery. - const current = - entry.state === "dispatching" - ? yield* findTask(taskId).pipe(Effect.orElseSucceed(() => null)) - : null; const prepared = yield* Effect.gen(function* () { + // A running row was already sent, so recovery keeps the prompt it was sent with. + const current = entry.state === "dispatching" ? yield* findTask(taskId) : null; const snapshot = yield* decodePendingTask(entry.task_json); // The snapshot keeps the identity runTask checks; the prompt is the one saved now. const task = diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 91ef89130c0d5..6303b38c05ce6 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1121,8 +1121,11 @@ it.effect("a coalesced window fails its delivery when the task row cannot be rea assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`).length, 0); const [delivery] = yield* sql<{ outcome: string; - }>`SELECT outcome FROM scheduled_task_webhook_deliveries`; + error: string | null; + }>`SELECT outcome, error FROM scheduled_task_webhook_deliveries`; assert.equal(delivery?.outcome, "dispatch_failed"); + // Failing before dispatch means a later successful read cannot send the stale snapshot. + assert.equal(delivery?.error, "The pending webhook payload could not be prepared."); }), { workers }, );