From 9e5c46b38f532d35dde51dbe8bb54aa46cda13b4 Mon Sep 17 00:00:00 2001 From: Fuksss Date: Tue, 27 Oct 2026 20:24:54 +0100 Subject: [PATCH 1/2] fix(server): coalesced webhook deliveries run the task's current prompt A coalesced delivery keeps the task prompt from the start of its window. Editing the task before dispatch runs the old instructions with current settings. Load the current task before rendering the pending request and keep the snapshot for the replacement check. Document the prompt timing. Test: dispatches and logs the current prompt after an edit during the coalescing window --- .../scheduledTasks/ScheduledTaskService.ts | 5 ++- .../ScheduledTaskService.webhook.test.ts | 42 +++++++++++++++++++ docs/user/project-settings.md | 3 +- 3 files changed, 47 insertions(+), 3 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 3d91ffa623a6c..3445afb400841 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -880,7 +880,7 @@ 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. + // Webhook prompts already include the framed request data. const prompt = webhook?.prompt ?? active.prompt; // A bound thread can move to another project (No project to a project) while @@ -2332,9 +2332,10 @@ export const layer = Layer.effect( } const prepared = yield* Effect.gen(function* () { const task = yield* decodePendingTask(entry.task_json); + const current = yield* loadTask(task.id); 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) }; + return { task, rendered: renderWebhookPrompt(current.prompt, request, frameId) }; }).pipe(Effect.exit); if (Exit.isFailure(prepared)) { if (Cause.hasInterruptsOnly(prepared.cause)) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 6fb5de1ff8873..bfe67ecacb666 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1031,6 +1031,48 @@ it.effect.each([1, 3])( }, ); +it.effect( + "dispatches and logs the current prompt after an edit during the coalescing window", + () => { + const workers = new Map>(); + return withService( + ({ service, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + prompt: "Use the original instructions.", + schedule: { + type: "webhook", + dedupe: { coalesceKey: "{{body.id}}", coalesceWindowSeconds: 10 }, + }, + }), + ); + yield* service.triggerWebhook( + requestFor(task, { + bodyText: '{"id":"review"}', + body: new TextEncoder().encode('{"id":"review"}'), + }), + ); + yield* service.upsert( + yield* webhookTaskInput({ prompt: "Use the updated instructions." }), + ); + yield* TestClock.adjust("11 seconds"); + yield* drainPendingDispatch(workers); + + const launched = yield* Queue.take(launches); + const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id }); + const { delivery } = yield* service.getWebhookDelivery({ + id: task.id, + deliveryId: deliveries[0]!.id, + }); + assert.equal(delivery.renderedPrompt, "Use the updated instructions."); + assert.equal(launched.initialMessage?.text, "Use the updated instructions."); + }), + { workers }, + ); + }, +); + it.effect("recovers a persisted coalescing window after restarting the service", () => { const workers = new Map>(); return withService( diff --git a/docs/user/project-settings.md b/docs/user/project-settings.md index ba481be9f5b00..aee2451d0d201 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -141,7 +141,8 @@ To suppress repeated deliveries, set a **Duplicate request key** template, such To combine a burst, set a **Coalesce key**, such as `{{body.id}}`, and a window of 10–3600 seconds (120 by default). One run uses the latest body, headers, and query when the window from the first delivery closes; later requests -do not extend it. Pending windows survive a server restart. Keys must contain +do not extend it. The run uses the task's prompt at dispatch time. +Pending windows survive a server restart. Keys must contain request placeholders; if any placeholder is missing, or the render is empty, that key's rule is skipped. Clear the templates to turn dedupe off. From a6b81db51c46723ac11b5730da88a56b83468397 Mon Sep 17 00:00:00 2001 From: Fuksss Date: Mon, 2 Nov 2026 21:04:11 +0100 Subject: [PATCH 2/2] fix(server): a retried coalesced delivery keeps the prompt it first ran A prompt edit after dispatch could change a running delivery's log on retry. An oversized edit could also release the replay claim for a run that already started. Save the first dispatch prompt in the pending snapshot and reuse it for running retries. Fold dispatch-time prompt edits into the coalescing table and clarify retry behavior in the guide. Tests: - a running coalesced retry keeps its first prompt and replay claim after edits - coalesces %i deliveries into one run with the latest body, headers and query - recovers a persisted coalescing window after restarting the service --- .../scheduledTasks/ScheduledTaskService.ts | 9 +- .../ScheduledTaskService.webhook.test.ts | 99 +++++++++++-------- docs/user/project-settings.md | 3 +- 3 files changed, 65 insertions(+), 46 deletions(-) diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 3445afb400841..361189cd90720 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.ts @@ -2332,10 +2332,11 @@ export const layer = Layer.effect( } const prepared = yield* Effect.gen(function* () { const task = yield* decodePendingTask(entry.task_json); - const current = yield* loadTask(task.id); + const prompt = + entry.state === "running" ? task.prompt : (yield* loadTask(task.id)).prompt; const request = yield* decodePendingRequest(entry.request_json); const frameId = Buffer.from(yield* crypto.randomBytes(6)).toString("hex"); - return { task, rendered: renderWebhookPrompt(current.prompt, request, frameId) }; + return { task, prompt, rendered: renderWebhookPrompt(prompt, request, frameId) }; }).pipe(Effect.exit); if (Exit.isFailure(prepared)) { if (Cause.hasInterruptsOnly(prepared.cause)) @@ -2348,12 +2349,14 @@ export const layer = Layer.effect( yield* remove; continue; } - const { task, rendered } = prepared.value; + const { task, prompt, 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 { 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}`; + // Retries must retain the prompt used by the first dispatch. + yield* sql`UPDATE scheduled_task_webhook_pending SET task_json = json_set(task_json, '$.prompt', ${prompt}) WHERE delivery_id = ${entry.delivery_id}`; if (!(yield* runWebhookDelivery(task, entry.delivery_id, rendered.prompt))) return; } yield* remove; diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index bfe67ecacb666..188ee3e5b665e 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -981,6 +981,11 @@ it.effect.each([1, 3])( ); if (value < count) yield* TestClock.adjust("2 seconds"); } + yield* service.upsert( + yield* webhookTaskInput({ + prompt: "Updated instructions: {{body.value}} {{headers.version}} {{query.version}}", + }), + ); const owner = (yield* service.listWebhookDeliveries({ id: task.id, acceptAnyOutcome: true, @@ -1009,6 +1014,7 @@ it.effect.each([1, 3])( (launched.initialMessage!.text.match(new RegExp(`>${count}<`, "g")) ?? []).length, 3, ); + assert.include(launched.initialMessage!.text, "Updated instructions:"); yield* workers.get("webhook-coalescing")!; assert.equal(yield* Queue.size(launches), 0); const outcomes = (yield* service.listWebhookDeliveries({ @@ -1022,6 +1028,7 @@ it.effect.each([1, 3])( })).delivery; assert.isFalse(afterDispatch.pendingWindow); assert.equal(afterDispatch.body, beforeDispatch.body); + assert.equal(afterDispatch.renderedPrompt, launched.initialMessage!.text); assert.include(afterDispatch.renderedPrompt!, `>${count}<`); assert.equal(outcomes.filter((o) => o === "accepted").length, 1); assert.equal(outcomes.filter((o) => o === "coalesced").length, count - 1); @@ -1031,48 +1038,6 @@ it.effect.each([1, 3])( }, ); -it.effect( - "dispatches and logs the current prompt after an edit during the coalescing window", - () => { - const workers = new Map>(); - return withService( - ({ service, launches }) => - Effect.gen(function* () { - const { task } = yield* service.upsert( - yield* webhookTaskInput({ - prompt: "Use the original instructions.", - schedule: { - type: "webhook", - dedupe: { coalesceKey: "{{body.id}}", coalesceWindowSeconds: 10 }, - }, - }), - ); - yield* service.triggerWebhook( - requestFor(task, { - bodyText: '{"id":"review"}', - body: new TextEncoder().encode('{"id":"review"}'), - }), - ); - yield* service.upsert( - yield* webhookTaskInput({ prompt: "Use the updated instructions." }), - ); - yield* TestClock.adjust("11 seconds"); - yield* drainPendingDispatch(workers); - - const launched = yield* Queue.take(launches); - const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id }); - const { delivery } = yield* service.getWebhookDelivery({ - id: task.id, - deliveryId: deliveries[0]!.id, - }); - assert.equal(delivery.renderedPrompt, "Use the updated instructions."); - assert.equal(launched.initialMessage?.text, "Use the updated instructions."); - }), - { workers }, - ); - }, -); - it.effect("recovers a persisted coalescing window after restarting the service", () => { const workers = new Map>(); return withService( @@ -1483,6 +1448,56 @@ it.effect("keeps replay keys when dispatch succeeds but completion bookkeeping f ), ); +it.effect("a running coalesced retry keeps its first prompt and replay claim after edits", () => { + const workers = new Map>(); + return withService( + ({ service, sql, launches }) => + Effect.gen(function* () { + const { task } = yield* service.upsert( + yield* webhookTaskInput({ + prompt: "Original instructions.", + schedule: { + type: "webhook", + dedupe: { + replayKey: "{{body}}", + coalesceKey: "{{body}}", + coalesceWindowSeconds: 10, + }, + }, + }), + ); + yield* service.triggerWebhook(requestFor(task)); + yield* service.upsert(yield* webhookTaskInput({ prompt: "First dispatch instructions." })); + 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); + assert.equal( + (yield* Queue.take(launches)).initialMessage?.text, + "First dispatch instructions.", + ); + const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id }); + const deliveryId = deliveries[0]!.id; + for (const prompt of ["Later instructions.", "x".repeat(120_001)]) { + yield* service.upsert(yield* webhookTaskInput({ prompt })); + yield* drainPendingDispatch(workers); + const { delivery } = yield* service.getWebhookDelivery({ id: task.id, deliveryId }); + assert.equal(delivery.renderedPrompt, "First dispatch instructions."); + assert.equal(delivery.error, "The run started, but completion could not be recorded."); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_replay_keys`).length, 1); + } + yield* sql`DROP TRIGGER fail_completion`; + yield* drainPendingDispatch(workers); + const retry = yield* service.triggerWebhook(requestFor(task)); + assert.equal(retry._tag === "accepted" && retry.outcome, "duplicate"); + assert.equal(yield* Queue.size(launches), 0); + assert.equal((yield* service.list()).tasks[0]!.runCount, 1); + assert.equal((yield* sql`SELECT 1 FROM scheduled_task_webhook_pending`).length, 0); + }), + { workers, completedLaunches: new Set() }, + ); +}); + it.effect.each([119_000, 120_000])( "bounds coalesced prompt logs and checks trimmed length for %i body characters", (length) => { diff --git a/docs/user/project-settings.md b/docs/user/project-settings.md index aee2451d0d201..f8c24d0133b90 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -141,7 +141,8 @@ To suppress repeated deliveries, set a **Duplicate request key** template, such To combine a burst, set a **Coalesce key**, such as `{{body.id}}`, and a window of 10–3600 seconds (120 by default). One run uses the latest body, headers, and query when the window from the first delivery closes; later requests -do not extend it. The run uses the task's prompt at dispatch time. +do not extend it. The run uses the task's prompt at its first dispatch. +A retry keeps that prompt. Pending windows survive a server restart. Keys must contain request placeholders; if any placeholder is missing, or the render is empty, that key's rule is skipped. Clear the templates to turn dedupe off.