Skip to content
Open
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions apps/server/src/scheduledTasks/ScheduledTaskService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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) };
Comment thread
maria-rcks marked this conversation as resolved.
Resolved
}).pipe(Effect.exit);
if (Exit.isFailure(prepared)) {
if (Cause.hasInterruptsOnly(prepared.cause))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, Effect.Effect<void>>();
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<string, Effect.Effect<void>>();
return withService(
Expand Down
3 changes: 2 additions & 1 deletion docs/user/project-settings.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
Loading