diff --git a/apps/server/src/scheduledTasks/ScheduledTaskService.ts b/apps/server/src/scheduledTasks/ScheduledTaskService.ts index 4dab48a05b2a5..afeb73de3df96 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 @@ -2331,10 +2331,21 @@ export const layer = Layer.effect( continue; } const prepared = yield* Effect.gen(function* () { - const task = yield* decodePendingTask(entry.task_json); + // 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 = + 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* random.randomBytes(6)).toString("hex"); - return { task, rendered: renderWebhookPrompt(task.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)) @@ -2347,11 +2358,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 bfafd3893e170..6303b38c05ce6 100644 --- a/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts +++ b/apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts @@ -1031,6 +1031,106 @@ 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("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; + 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 }, + ); +}); + 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({