diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts index 6303b38c05ce6..f7de3e808067f 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); @@ -1541,6 +1548,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 ba481be9f5b00..f8c24d0133b90 100644 --- a/docs/user/project-settings.md +++ b/docs/user/project-settings.md @@ -141,7 +141,9 @@ 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 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.