1import {
2 CommandId,
3 MessageId,
4 PROVIDER_SEND_TURN_MAX_INPUT_CHARS,
5 ScheduledTask,
6 ScheduledTaskError,
7 ScheduledTaskId,
8 ThreadId,
9 ScheduledTaskWebhookDeliveryId,
10 type ScheduledTaskDeleteInput,
11 type ScheduledTaskDeleteResult,
12 type ScheduledTaskGetWebhookDeliveryInput,
13 type ScheduledTaskGetWebhookDeliveryResult,
14 type ScheduledTaskListWebhookDeliveriesInput,
15 type ScheduledTaskListWebhookDeliveriesResult,
16 type ScheduledTaskRotateWebhookTokenInput,
17 type ScheduledTaskWebhookDeliveryOutcome,
18 type ScheduledTaskListResult,
19 type ScheduledTaskMutationResult,
20 type ScheduledTaskRunNowInput,
21 type ScheduledTaskRunNowResult,
22 type ScheduledTaskSetEnabledInput,
23 type ScheduledTaskUpsertInput,
24} from "@t3tools/contracts";
25import * as Cause from "effect/Cause";
26import * as Context from "effect/Context";
27import * as Crypto from "effect/Crypto";
28import * as Data from "effect/Data";
29import * as DateTime from "effect/DateTime";
30import * as Effect from "effect/Effect";
31import * as Duration from "effect/Duration";
32import * as Metric from "effect/Metric";
33import * as Layer from "effect/Layer";
34import * as Option from "effect/Option";
35import * as PubSub from "effect/PubSub";
36import * as Ref from "effect/Ref";
37import * as Result from "effect/Result";
38import * as Schema from "effect/Schema";
39import * as Semaphore from "effect/Semaphore";
40import * as Stream from "effect/Stream";
41import * as SqlClient from "effect/sql/SqlClient";
42
43import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
44import * as Metrics from "../observability/Metrics.ts";
45import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
46import * as SecretRequests from "../secrets/SecretRequests.ts";
47import * as Scheduler from "../scheduling/Scheduler.ts";
48import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts";
49import {
50 redactHeaders,
51 redactQuery,
52 renderWebhookPrompt,
53 type WebhookRequest,
54} from "./webhookTemplate.ts";
55import { constantTimeEquals, verifyWebhookSignature } from "./webhookVerification.ts";
56
57/** Path prefix of the environment route that receives webhook requests. */
58export const WEBHOOK_ROUTE_PREFIX = "/api/hooks";
59/** Deliveries kept per task; older ones are pruned on insert. */
60const WEBHOOK_DELIVERY_RETENTION = 50;
61/** Body text kept in the delivery log. Larger bodies are cut and flagged. */
62const WEBHOOK_DELIVERY_LOG_BODY_LIMIT = 64 * 1024;
63/** Rendered prompt text kept in the delivery log. */
64const WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT = 64 * 1024;
65/** Deliveries one task may hold at once, running or waiting their turn. */
66const WEBHOOK_MAX_QUEUED_PER_TASK = 20;
67/** Accepted deliveries per task per minute, enforced here as well as on the relay because the tunnel hostname is public too. */
68const WEBHOOK_RATE_LIMIT_PER_MINUTE = 60;
69
70/**
71 * Where a webhook task's public URL points: `${relayHookBaseUrl}/${taskId}/${token}`.
72 * Null when the environment has no managed tunnel on T3 Connect; clients then show the path.
73 */
74interface WebhookOrigin {
75 readonly relayHookBaseUrl: string | null;
76}
77
78const ENDPOINT_KEY = /^[0-9a-f]{16}$/;
79
80/**
81 * The relay's hook URL prefix for this environment. The relay finds the
82 * environment by its managed tunnel's key (the tunnel name's last segment),
83 * so the URL never reveals the environment id.
84 */
85export function relayHookBaseUrl(input: {
86 readonly relayUrl: string;
87 readonly tunnelName: string | undefined;
88}): string | null {
89 const endpointKey = input.tunnelName?.split("-").at(-1);
90 const relayUrl = input.relayUrl.replace(/\/+$/, "");
91 if (relayUrl === "" || endpointKey === undefined || !ENDPOINT_KEY.test(endpointKey)) {
92 return null;
93 }
94 return `${relayUrl}/v1/hooks/${endpointKey}`;
95}
96
97export class ScheduledTaskWebhookOrigin extends Context.Reference<Effect.Effect<WebhookOrigin>>(
98 "t3/scheduledTasks/ScheduledTaskWebhookOrigin",
99 {
100 defaultValue: () => Effect.succeed({ relayHookBaseUrl: null }),
101 },
102) {}
103
104/** A queued webhook delivery that no longer applies to its task; `reason` is shown in the delivery log. */
105class WebhookDeliverySkipped extends Data.TaggedError("WebhookDeliverySkipped")<{
106 readonly reason: string;
107}> {}
108
109interface RateWindow {
110 readonly accepted: ReadonlyArray<number>;
111 /** Whether a rejection was already logged in this window. */
112 readonly rejectedLogged: boolean;
113}
114
115export interface WebhookTriggerRequest extends WebhookRequest {
116 readonly hookId: string;
117 readonly token: string;
118 readonly body: Uint8Array;
119 /** Set by T3 Connect; the same id is never dispatched twice. */
120 readonly relayDeliveryId?: string;
121 /** When the relay received a held request; defaults to now. */
122 readonly receivedAt?: string;
123}
124
125/** What the HTTP route should answer. `not_found` covers unknown hooks and wrong tokens alike. */
126/**
127 * What happened to one webhook request, as recorded in metrics and spans.
128 * Sent back to the relay as `x-t3-hook-outcome`; never request contents.
129 */
130export type WebhookDeliveryOutcome =
131 | "accepted"
132 | "duplicate"
133 | "prompt_too_long"
134 | "queue_full"
135 | "rate_limited"
136 | "disabled"
137 | "rejected_signature"
138 | "expired"
139 | "not_found"
140 | "error";
141
142export type WebhookTriggerResult =
143 | {
144 readonly _tag: "accepted";
145 readonly deliveryId: ScheduledTaskWebhookDeliveryId;
146 /** A 202 covers more than a started run; this says which. */
147 readonly outcome: "accepted" | "duplicate" | "prompt_too_long";
148 }
149 | { readonly _tag: "not_found" }
150 | { readonly _tag: "rejected_signature" }
151 | { readonly _tag: "disabled" }
152 | {
153 readonly _tag: "rate_limited";
154 /** Too many requests to this hook, or too many runs already waiting. */
155 readonly outcome: "rate_limited" | "queue_full";
156 }
157 | { readonly _tag: "expired" };
158
159const decodeTask = Schema.decodeUnknownEffect(ScheduledTask);
160const decodeTaskId = Schema.decodeUnknownOption(ScheduledTaskId);
161const decodeScheduleJson = Schema.decodeUnknownEffect(
162 Schema.fromJsonString(ScheduledTask.fields.schedule),
163);
164const decodeWorkspaceStrategyJson = Schema.decodeUnknownEffect(
165 Schema.fromJsonString(ScheduledTask.fields.workspaceStrategy),
166);
167const decodeModelSelectionJson = Schema.decodeUnknownEffect(
168 Schema.fromJsonString(ScheduledTask.fields.modelSelection),
169);
170const HeadersJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.String));
171const MissingFieldsJson = Schema.fromJsonString(Schema.Array(Schema.String));
172const decodeHeadersJson = Schema.decodeUnknownOption(HeadersJson);
173const decodeMissingFieldsJson = Schema.decodeUnknownOption(MissingFieldsJson);
174const encodeHeadersJson = Schema.encodeSync(HeadersJson);
175const encodeMissingFieldsJson = Schema.encodeSync(MissingFieldsJson);
176
177interface ScheduledTaskRow {
178 readonly task_id: string;
179 readonly title: string;
180 readonly prompt: string;
181 readonly enabled: number;
182 readonly schedule_json: string;
183 readonly project_id: string;
184 readonly thread_id: string | null;
185 readonly workspace_strategy_json: string;
186 readonly model_selection_json: string;
187 readonly runtime_mode: string;
188 readonly interaction_mode: string;
189 readonly created_by: string;
190 readonly creation_source: string;
191 readonly created_at: string;
192 readonly updated_at: string;
193 readonly next_run_at: string | null;
194 readonly last_run_at: string | null;
195 readonly last_run_status: string;
196 readonly last_run_error: string | null;
197 readonly run_count: number;
198 readonly webhook_token: string | null;
199 readonly webhook_secret: string | null;
200}
201
202interface WebhookDeliveryRow {
203 readonly delivery_id: string;
204 readonly task_id: string;
205 readonly received_at: string;
206 readonly method: string;
207 readonly query: string;
208 readonly headers_json: string;
209 readonly body: string;
210 readonly body_bytes: number;
211 readonly body_truncated: number;
212 readonly outcome: string;
213 readonly signature_verified: number;
214 readonly missing_fields_json: string;
215 readonly rendered_prompt: string | null;
216 readonly error: string | null;
217}
218
219export class ScheduledTaskService extends Context.Service<
220 ScheduledTaskService,
221 {
222 readonly list: () => Effect.Effect<ScheduledTaskListResult, ScheduledTaskError>;
223 /** Emits the full task list on subscribe and again after every change (CRUD, run transitions, reschedules). */
224 readonly subscribeList: () => Stream.Stream<ScheduledTaskListResult, ScheduledTaskError>;
225 readonly upsert: (
226 input: ScheduledTaskUpsertInput,
227 ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>;
228 /** Partial update flipping only the enabled flag; never touches other fields. */
229 readonly setEnabled: (
230 input: ScheduledTaskSetEnabledInput,
231 ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>;
232 readonly delete: (
233 input: ScheduledTaskDeleteInput,
234 ) => Effect.Effect<ScheduledTaskDeleteResult, ScheduledTaskError>;
235 readonly runNow: (
236 input: ScheduledTaskRunNowInput,
237 ) => Effect.Effect<ScheduledTaskRunNowResult, ScheduledTaskError>;
238 /** Issues a new URL token for a webhook task; the old URL stops working at once. */
239 readonly rotateWebhookToken: (
240 input: ScheduledTaskRotateWebhookTokenInput,
241 ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>;
242 readonly listWebhookDeliveries: (
243 input: ScheduledTaskListWebhookDeliveriesInput,
244 ) => Effect.Effect<ScheduledTaskListWebhookDeliveriesResult, ScheduledTaskError>;
245 readonly getWebhookDelivery: (
246 input: ScheduledTaskGetWebhookDeliveryInput,
247 ) => Effect.Effect<ScheduledTaskGetWebhookDeliveryResult, ScheduledTaskError>;
248 /**
249 * Verifies, logs and dispatches one webhook request. Returns as soon as the
250 * delivery is logged; the run itself continues in the background so senders
251 * with short timeouts get their answer immediately.
252 */
253 readonly triggerWebhook: (
254 request: WebhookTriggerRequest,
255 ) => Effect.Effect<WebhookTriggerResult, ScheduledTaskError>;
256 }
257>()("t3/scheduledTasks/ScheduledTaskService") {}
258
259function taskError(message: string, input?: { taskId?: ScheduledTaskId; cause?: unknown }) {
260 return new ScheduledTaskError({
261 message,
262 ...(input?.taskId === undefined ? {} : { taskId: input.taskId }),
263 ...(input?.cause === undefined ? {} : { cause: input.cause }),
264 });
265}
266
267function iso(value: DateTime.DateTime): string {
268 return DateTime.formatIso(DateTime.toUtc(value));
269}
270
271const localNow = DateTime.withCurrentZoneLocal(DateTime.nowInCurrentZone);
272
273function nextRunAt(
274 task: Pick<ScheduledTask, "enabled" | "schedule">,
275 from: DateTime.DateTime,
276): string | null {
277 if (!task.enabled) return null;
278 // A stored interval can decode yet overflow the representable DateTime
279 // range; an unrepresentable occurrence means the task has no next run.
280 try {
281 const next = nextScheduledRunAt(task.schedule, from);
282 return next !== null && Number.isFinite(DateTime.toEpochMillis(next)) ? iso(next) : null;
283 } catch {
284 return null;
285 }
286}
287
288function errorMessage(error: unknown): string {
289 if (Cause.isCause(error)) return Cause.pretty(error);
290 if (error instanceof Error) return error.message;
291 return String(error);
292}
293
294function webhookPath(taskId: string, token: string): string {
295 return `${WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(taskId)}/${token}`;
296}
297
298function webhookEndpoint(
299 row: Pick<ScheduledTaskRow, "task_id" | "webhook_token" | "webhook_secret">,
300 origin: WebhookOrigin | null,
301): ScheduledTask["webhook"] {
302 if (row.webhook_token === null) return undefined;
303 const base = origin?.relayHookBaseUrl ?? null;
304 return {
305 path: webhookPath(row.task_id, row.webhook_token),
306 url: base === null ? null : `${base}/${encodeURIComponent(row.task_id)}/${row.webhook_token}`,
307 hasSecret: row.webhook_secret !== null,
308 };
309}
310
311const decodeRow = (row: ScheduledTaskRow, origin: WebhookOrigin | null = null) =>
312 Effect.gen(function* () {
313 const schedule = yield* decodeScheduleJson(row.schedule_json);
314 const webhook = schedule.type === "webhook" ? webhookEndpoint(row, origin) : undefined;
315 const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json);
316 const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json);
317 return yield* decodeTask({
318 // The stored id decodes through the task schema so a corrupt value fails
319 // as a typed parse error, not a `ScheduledTaskId.make` defect.
320 id: row.task_id,
321 title: row.title,
322 prompt: row.prompt,
323 enabled: row.enabled === 1,
324 schedule,
325 projectId: row.project_id,
326 threadId: row.thread_id,
327 workspaceStrategy,
328 modelSelection,
329 runtimeMode: row.runtime_mode,
330 interactionMode: row.interaction_mode,
331 createdBy: row.created_by,
332 creationSource: row.creation_source,
333 createdAt: row.created_at,
334 updatedAt: row.updated_at,
335 nextRunAt: row.next_run_at,
336 lastRunAt: row.last_run_at,
337 lastRunStatus: row.last_run_status,
338 lastRunError: row.last_run_error,
339 runCount: row.run_count,
340 ...(webhook === undefined ? {} : { webhook }),
341 });
342 }).pipe(
343 Effect.mapError((cause) => {
344 // The typed diagnostic can only carry an id that itself decodes; a
345 // corrupt stored id is omitted rather than re-thrown as a defect.
346 const taskId = decodeTaskId(row.task_id);
347 return taskError("Could not decode schedule task row.", {
348 ...(Option.isSome(taskId) ? { taskId: taskId.value } : {}),
349 cause,
350 });
351 }),
352 );
353
354/** Select poll candidates before decoding their schedules or other JSON payloads. */
355export const listDueTasks = Effect.fn("ScheduledTaskService.listDueTasks")(function* (
356 now: DateTime.DateTime,
357) {
358 const sql = yield* SqlClient.SqlClient;
359 const rows = yield* sql<ScheduledTaskRow>`
360 SELECT * FROM scheduled_tasks
361 WHERE enabled = 1 AND next_run_at IS NOT NULL
362 AND next_run_at <= ${iso(now)} AND last_run_status <> 'running'
363 ORDER BY next_run_at ASC, task_id ASC
364 `;
365 const tasks: ScheduledTask[] = [];
366 for (const row of rows) {
367 const decoded = yield* Effect.result(decodeRow(row));
368 if (Result.isSuccess(decoded)) {
369 const task = decoded.success;
370 // next_run_at is a freeform string at the schema level; a stored value
371 // that cannot parse as a DateTime would defect the poll below, so the
372 // row is skipped here like any other corrupt row.
373 if (task.nextRunAt === null || Option.isSome(DateTime.make(task.nextRunAt))) {
374 tasks.push(task);
375 } else {
376 yield* Effect.logWarning("Skipping schedule task row with invalid next_run_at", {
377 taskId: row.task_id,
378 });
379 }
380 } else {
381 yield* Effect.logWarning("Skipping undecodable schedule task row", {
382 taskId: row.task_id,
383 cause: decoded.failure,
384 });
385 }
386 }
387 return tasks;
388});
389
390export const layer = Layer.effect(
391 ScheduledTaskService,
392 Effect.gen(function* () {
393 const sql = yield* SqlClient.SqlClient;
394 const crypto = yield* Crypto.Crypto;
395 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;
396 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
397 const secretRequests = yield* SecretRequests.SecretRequests;
398 const scheduler = yield* Scheduler.Scheduler;
399 const readWebhookOrigin = yield* ScheduledTaskWebhookOrigin;
400 // Webhook deliveries for one task dispatch in arrival order rather than
401 // being dropped while an earlier delivery is still dispatching.
402 const webhookPermits = yield* Ref.make<ReadonlyMap<ScheduledTaskId, Semaphore.Semaphore>>(
403 new Map(),
404 );
405 // Keyed by task id and creation time, so deliveries of a deleted task that
406 // finish late release their own count, never a recreated task's.
407 const webhookQueued = yield* Ref.make<ReadonlyMap<string, number>>(new Map());
408 const webhookRateWindows = yield* Ref.make<ReadonlyMap<ScheduledTaskId, RateWindow>>(new Map());
409 const activeRuns = yield* Ref.make<ReadonlySet<ScheduledTaskId>>(new Set());
410 // Sliding(1) coalesces the dirty-signal: every notification triggers a
411 // full list() re-emit anyway, so a slow subscriber only ever needs the
412 // latest signal — an unbounded backlog would just grow memory.
413 const changesPubSub = yield* PubSub.sliding<void>(1);
414 const notifyChanged = PubSub.publish(changesPubSub, undefined).pipe(Effect.asVoid);
415
416 const selectAllRows = () => sql<ScheduledTaskRow>`
417 SELECT
418 task_id,
419 title,
420 prompt,
421 enabled,
422 schedule_json,
423 project_id,
424 thread_id,
425 workspace_strategy_json,
426 model_selection_json,
427 runtime_mode,
428 interaction_mode,
429 created_by,
430 creation_source,
431 created_at,
432 updated_at,
433 next_run_at,
434 last_run_at,
435 last_run_status,
436 last_run_error,
437 run_count,
438 webhook_token,
439 webhook_secret
440 FROM scheduled_tasks
441 ORDER BY updated_at DESC, task_id ASC
442 `;
443
444 // Strict decode for the API surface: a corrupt row is a visible error.
445 const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () {
446 const rows = yield* selectAllRows();
447 const origin = yield* readWebhookOrigin;
448 return yield* Effect.forEach(rows, (row) => decodeRow(row, origin), { concurrency: 1 });
449 });
450
451 const getRows = (id: ScheduledTaskId) => sql<ScheduledTaskRow>`
452 SELECT
453 task_id,
454 title,
455 prompt,
456 enabled,
457 schedule_json,
458 project_id,
459 thread_id,
460 workspace_strategy_json,
461 model_selection_json,
462 runtime_mode,
463 interaction_mode,
464 created_by,
465 creation_source,
466 created_at,
467 updated_at,
468 next_run_at,
469 last_run_at,
470 last_run_status,
471 last_run_error,
472 run_count,
473 webhook_token,
474 webhook_secret
475 FROM scheduled_tasks
476 WHERE task_id = ${id}
477 `;
478
479 /** Load a task, returning `null` when it does not exist; real load/decode failures propagate. */
480 const findTask = Effect.fn("ScheduledTaskService.findTask")(function* (id: ScheduledTaskId) {
481 const rows = yield* getRows(id).pipe(
482 Effect.mapError((cause) =>
483 taskError("Could not load schedule task.", { taskId: id, cause }),
484 ),
485 );
486 const row = rows[0];
487 if (row === undefined) return null;
488 return yield* decodeRow(row, yield* readWebhookOrigin);
489 });
490
491 const findWebhookCredentials = (id: ScheduledTaskId) =>
492 getRows(id).pipe(
493 Effect.map((rows) =>
494 rows[0] === undefined
495 ? null
496 : { token: rows[0].webhook_token, secret: rows[0].webhook_secret },
497 ),
498 Effect.mapError((cause) =>
499 taskError("Could not load schedule task.", { taskId: id, cause }),
500 ),
501 );
502
503 const newWebhookToken = crypto.randomBytes(32).pipe(
504 Effect.map((bytes) => Buffer.from(bytes).toString("base64url")),
505 Effect.mapError((cause) => taskError("Could not generate webhook token.", { cause })),
506 );
507
508 const loadTask = Effect.fn("ScheduledTaskService.loadTask")(function* (id: ScheduledTaskId) {
509 const task = yield* findTask(id);
510 if (task === null) {
511 return yield* taskError("Schedule task not found.", { taskId: id });
512 }
513 return task;
514 });
515
516 // Run-state columns (last_run_*, run_count) are intentionally absent from
517 // the conflict clause: they are owned by the run transitions below, and a
518 // concurrent settings save must not overwrite an in-flight increment.
519 // Check existence in the write itself so an edit cannot undo a deletion
520 // that landed after upsert loaded the previous task.
521 const saveTask = (
522 task: ScheduledTask,
523 requireExisting: boolean,
524 webhook: {
525 readonly token: string | null;
526 readonly secret: string | null;
527 /** False when the save carried no new secret, so a concurrent change survives. */
528 readonly secretChanged: boolean;
529 },
530 ) =>
531 sql<{ task_id: string }>`
532 INSERT INTO scheduled_tasks (
533 task_id,
534 title,
535 prompt,
536 enabled,
537 schedule_json,
538 project_id,
539 thread_id,
540 workspace_strategy_json,
541 model_selection_json,
542 runtime_mode,
543 interaction_mode,
544 created_by,
545 creation_source,
546 created_at,
547 updated_at,
548 next_run_at,
549 last_run_at,
550 last_run_status,
551 last_run_error,
552 run_count,
553 webhook_token,
554 webhook_secret
555 )
556 SELECT
557 ${task.id},
558 ${task.title},
559 ${task.prompt},
560 ${task.enabled ? 1 : 0},
561 ${JSON.stringify(task.schedule)},
562 ${task.projectId},
563 ${task.threadId},
564 ${JSON.stringify(task.workspaceStrategy)},
565 ${JSON.stringify(task.modelSelection)},
566 ${task.runtimeMode},
567 ${task.interactionMode},
568 ${task.createdBy},
569 ${task.creationSource},
570 ${task.createdAt},
571 ${task.updatedAt},
572 ${task.nextRunAt},
573 ${task.lastRunAt},
574 ${task.lastRunStatus},
575 ${task.lastRunError},
576 ${task.runCount},
577 ${webhook.token},
578 ${webhook.secret}
579 WHERE ${requireExisting ? 0 : 1} = 1
580 OR EXISTS (SELECT 1 FROM scheduled_tasks WHERE task_id = ${task.id})
581 ON CONFLICT (task_id)
582 DO UPDATE SET
583 title = excluded.title,
584 prompt = excluded.prompt,
585 enabled = excluded.enabled,
586 schedule_json = excluded.schedule_json,
587 project_id = excluded.project_id,
588 thread_id = excluded.thread_id,
589 workspace_strategy_json = excluded.workspace_strategy_json,
590 model_selection_json = excluded.model_selection_json,
591 runtime_mode = excluded.runtime_mode,
592 interaction_mode = excluded.interaction_mode,
593 creation_source = excluded.creation_source,
594 updated_at = excluded.updated_at,
595 next_run_at = excluded.next_run_at,
596 -- Only rotate changes a live token, so a save racing a rotation
597 -- cannot bring the old URL back.
598 webhook_token = CASE
599 WHEN excluded.webhook_token IS NULL THEN NULL
600 ELSE COALESCE(scheduled_tasks.webhook_token, excluded.webhook_token)
601 END,
602 webhook_secret = CASE
603 WHEN ${webhook.secretChanged ? 1 : 0} = 1 THEN excluded.webhook_secret
604 ELSE scheduled_tasks.webhook_secret
605 END
606 RETURNING task_id
607 `.pipe(
608 Effect.mapError((cause) =>
609 taskError("Could not save schedule task.", { taskId: task.id, cause }),
610 ),
611 Effect.flatMap((rows) =>
612 rows.length > 0
613 ? Effect.void
614 : taskError("Schedule task not found.", { taskId: task.id }),
615 ),
616 );
617
618 const deleteRow = (id: ScheduledTaskId) =>
619 sql
620 .withTransaction(
621 sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${id}`.pipe(
622 Effect.andThen(sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`),
623 ),
624 )
625 .pipe(
626 Effect.mapError((cause) =>
627 taskError("Could not delete schedule task.", { taskId: id, cause }),
628 ),
629 );
630
631 // Run-state transitions use targeted UPDATEs (never the full-row upsert) so
632 // a completing run cannot resurrect a deleted task or clobber concurrent
633 // edits to the task definition.
634 const markRunning = (id: ScheduledTaskId, startedAtIso: string) =>
635 sql<{ task_id: string }>`
636 UPDATE scheduled_tasks
637 SET updated_at = ${startedAtIso},
638 last_run_at = ${startedAtIso},
639 last_run_status = 'running',
640 last_run_error = NULL
641 WHERE task_id = ${id}
642 RETURNING task_id
643 `.pipe(
644 Effect.mapError((cause) =>
645 taskError("Could not mark schedule task as running.", { taskId: id, cause }),
646 ),
647 // A task deleted after the re-read must not be dispatched from the stale snapshot.
648 Effect.flatMap((rows) =>
649 rows.length > 0 ? Effect.void : taskError("Schedule task not found.", { taskId: id }),
650 ),
651 );
652
653 const markCompleted = (input: {
654 readonly id: ScheduledTaskId;
655 readonly completedAtIso: string;
656 readonly nextRunAtIso: string | null;
657 readonly status: "succeeded" | "failed";
658 readonly error: string | null;
659 readonly startedAtIso: string;
660 }) =>
661 sql`
662 UPDATE scheduled_tasks
663 SET updated_at = ${input.completedAtIso},
664 next_run_at = ${input.nextRunAtIso},
665 last_run_status = ${input.status},
666 last_run_error = ${input.error},
667 run_count = run_count + 1
668 WHERE task_id = ${input.id}
669 AND last_run_status = 'running'
670 AND last_run_at = ${input.startedAtIso}
671 `.pipe(
672 Effect.mapError((cause) =>
673 taskError("Could not record schedule task run.", { taskId: input.id, cause }),
674 ),
675 );
676
677 // Best-effort escape hatch: if anything fails between markRunning and
678 // markCompleted, write a full terminal record so runDueTasks neither skips
679 // the task forever (it filters out 'running' rows) nor re-fires it
680 // immediately: the dispatch may already have gone out, so next_run_at must
681 // advance and run_count must count the attempt.
682 const releaseStuckRun = (task: ScheduledTask, message: string) =>
683 Effect.gen(function* () {
684 const now = yield* localNow;
685 // Compute the next occurrence from the current row so a schedule
686 // edited while the run was in flight is honoured; fall back to the
687 // run's snapshot only if the re-read itself fails.
688 const reread = yield* Effect.result(findTask(task.id));
689 if (Result.isSuccess(reread) && reread.success === null) return; // deleted — nothing to release
690 const source = Result.isSuccess(reread) && reread.success !== null ? reread.success : task;
691 yield* sql`
692 UPDATE scheduled_tasks
693 SET last_run_status = 'failed',
694 last_run_error = ${message},
695 next_run_at = ${nextRunAt(source, now)},
696 updated_at = ${iso(now)},
697 run_count = run_count + 1
698 WHERE task_id = ${task.id} AND last_run_status = 'running'
699 `;
700 yield* notifyChanged;
701 }).pipe(
702 Effect.catch((cause) =>
703 Effect.logWarning("Could not release stuck schedule task run", {
704 taskId: task.id,
705 cause,
706 }),
707 ),
708 );
709
710 const runTask = Effect.fn("ScheduledTaskService.runTask")(function* (
711 task: ScheduledTask,
712 trigger: "scheduled" | "manual" | "webhook",
713 webhook?: { readonly deliveryId: string; readonly prompt: string },
714 ) {
715 const reserved = yield* Ref.modify(activeRuns, (active) => {
716 if (active.has(task.id)) return [false, active] as const;
717 const next = new Set(active);
718 next.add(task.id);
719 return [true, next] as const;
720 });
721 if (!reserved) {
722 if (trigger !== "scheduled") {
723 return yield* taskError("Schedule task is already running.", { taskId: task.id });
724 }
725 return task;
726 }
727
728 return yield* Effect.gen(function* () {
729 const startedAt = yield* localNow;
730 const startedAtIso = iso(startedAt);
731
732 // The in-memory snapshot may be stale: re-read before touching run
733 // state. The task may have been deleted, paused, or postponed since
734 // the poll loaded it — none of those may fire.
735 const active = yield* findTask(task.id);
736 if (active === null) {
737 if (webhook !== undefined) {
738 return yield* new WebhookDeliverySkipped({
739 reason: "The task was deleted before this delivery ran.",
740 });
741 }
742 // A manual run on a just-deleted task must fail loudly, not report
743 // a successful run that never dispatched.
744 if (trigger !== "scheduled") {
745 return yield* taskError("Schedule task not found.", { taskId: task.id });
746 }
747 return task;
748 }
749 // A next_run_at corrupted between the poll read and this re-read must
750 // not defect the poll; an unparseable value is treated as not due.
751 const parsedNextRunAt =
752 active.nextRunAt === null ? Option.none() : DateTime.make(active.nextRunAt);
753 if (
754 trigger === "scheduled" &&
755 (!active.enabled ||
756 Option.isNone(parsedNextRunAt) ||
757 DateTime.toEpochMillis(parsedNextRunAt.value) > DateTime.toEpochMillis(startedAt))
758 ) {
759 return active;
760 }
761 // A queued delivery must not run a task that was paused, deleted and
762 // recreated under the same id, or switched to another trigger while
763 // it waited for its turn.
764 if (webhook !== undefined) {
765 const reason =
766 active.createdAt !== task.createdAt
767 ? "The task was replaced before this delivery ran."
768 : active.schedule.type !== "webhook"
769 ? "The task's trigger changed before this delivery ran."
770 : !active.enabled
771 ? "The task was paused before this delivery ran."
772 : null;
773 if (reason !== null) return yield* new WebhookDeliverySkipped({ reason });
774 }
775
776 yield* markRunning(active.id, startedAtIso);
777 yield* notifyChanged;
778
779 // A webhook run is keyed by its delivery so the same delivery can
780 // never dispatch twice.
781 const fireKey =
782 webhook === undefined
783 ? `${active.id}:${DateTime.toEpochMillis(startedAt)}:${trigger}`
784 : `${active.id}:webhook:${webhook.deliveryId}`;
785 const commandId = CommandId.make(`scheduled-task:${fireKey}`);
786 const messageId = MessageId.make(`scheduled-task-message:${fireKey}`);
787 // Dispatch from the fresh row so prompt/model/binding edits made
788 // after the poll read are honoured. A webhook prompt was rendered
789 // from the row when the request arrived.
790 const prompt = webhook?.prompt ?? active.prompt;
791
792 // Effect.exit (not Effect.result) so defects and interruptions in the
793 // dispatch are also captured and recorded as a failed run instead of
794 // aborting before markCompleted.
795 const result =
796 active.threadId === null
797 ? yield* Effect.exit(
798 threadLaunch.launch({
799 commandId,
800 projectId: active.projectId,
801 title: active.title,
802 modelSelection: active.modelSelection,
803 runtimeMode: active.runtimeMode,
804 interactionMode: active.interactionMode,
805 workspaceStrategy: active.workspaceStrategy,
806 initialMessage: {
807 messageId,
808 scheduledTaskId: active.id,
809 text: prompt,
810 attachments: [],
811 },
812 createdBy: active.createdBy,
813 creationSource: active.creationSource,
814 }),
815 )
816 : yield* Effect.exit(
817 threadManagement.sendToThread({
818 projectId: active.projectId,
819 commandId,
820 threadId: ThreadId.make(active.threadId),
821 messageId,
822 scheduledTaskId: active.id,
823 text: prompt,
824 attachments: [],
825 modelSelection: active.modelSelection,
826 // Scheduled prompts must not interrupt tools in the bound thread.
827 mode: "queue",
828 createdBy: active.createdBy,
829 creationSource: active.creationSource,
830 }),
831 );
832
833 const completedAt = yield* localNow;
834 const runSucceeded = result._tag === "Success";
835 const lastRunStatus = runSucceeded ? ("succeeded" as const) : ("failed" as const);
836 const lastRunError = runSucceeded ? null : errorMessage(result.cause);
837 // Re-read the task so the next run is computed from the schedule as it
838 // is *now* (the user may have edited or deleted it while we ran).
839 const current = yield* findTask(task.id);
840 const scheduleSource = current ?? task;
841 const completed: ScheduledTask = {
842 ...scheduleSource,
843 updatedAt: iso(completedAt),
844 lastRunAt: startedAtIso,
845 nextRunAt: nextRunAt(scheduleSource, completedAt),
846 lastRunStatus,
847 lastRunError,
848 runCount: scheduleSource.runCount + 1,
849 };
850 if (current !== null) {
851 // startedAtIso in the guard ensures this writes only to the row this
852 // run marked as running — a task deleted mid-run and recreated with
853 // the same id (idempotent commandId replay) must not be stamped.
854 yield* markCompleted({
855 id: task.id,
856 completedAtIso: completed.updatedAt,
857 nextRunAtIso: completed.nextRunAt,
858 status: lastRunStatus,
859 error: lastRunError,
860 startedAtIso,
861 });
862 yield* notifyChanged;
863 }
864 return completed;
865 }).pipe(
866 Effect.onError((cause) => releaseStuckRun(task, errorMessage(cause))),
867 Effect.ensuring(
868 Ref.update(activeRuns, (active) => {
869 const next = new Set(active);
870 next.delete(task.id);
871 return next;
872 }),
873 ),
874 );
875 });
876
877 // A due fixed-time run that is long past its slot (server was off or
878 // asleep) is skipped and re-aimed at its next occurrence, not fired late.
879 const rescheduleMissedRun = Effect.fn("ScheduledTaskService.rescheduleMissedRun")(function* (
880 task: ScheduledTask,
881 now: DateTime.DateTime,
882 ) {
883 const next = nextRunAt(task, now);
884 yield* Effect.logInfo("Skipping missed schedule task run", {
885 taskId: task.id,
886 missedRunAt: task.nextRunAt,
887 rescheduledTo: next,
888 });
889 yield* sql`
890 UPDATE scheduled_tasks
891 SET next_run_at = ${next},
892 updated_at = ${iso(now)}
893 WHERE task_id = ${task.id}
894 `.pipe(
895 Effect.mapError((cause) =>
896 taskError("Could not reschedule missed schedule task run.", { taskId: task.id, cause }),
897 ),
898 );
899 yield* notifyChanged;
900 });
901
902 const runDueTasks = Effect.fn("ScheduledTaskService.runDueTasks")(function* () {
903 const now = yield* localNow;
904 const tasks = yield* listDueTasks(now).pipe(
905 Effect.mapError((cause) => taskError("Could not list schedule tasks.", { cause })),
906 );
907 const due = tasks.flatMap((task) =>
908 task.nextRunAt === null ? [] : [{ task, dueAt: DateTime.makeUnsafe(task.nextRunAt) }],
909 );
910 yield* Effect.forEach(
911 due,
912 ({ task, dueAt }) =>
913 Effect.suspend(() =>
914 isMissedFixedTimeRun(task.schedule, dueAt, now)
915 ? rescheduleMissedRun(task, now)
916 : runTask(task, "scheduled"),
917 ).pipe(
918 Effect.catch((cause) =>
919 Effect.logWarning("Scheduled task run failed", { taskId: task.id, cause }),
920 ),
921 ),
922 { concurrency: 1, discard: true },
923 );
924 });
925
926 // Recover from a crash or hard shutdown mid-run: rows stuck in 'running'
927 // would otherwise be skipped by the due-task filter forever. The dispatch
928 // may already have gone out before the crash, so next_run_at must advance
929 // and run_count must count the attempt — otherwise the first poll after
930 // every restart re-fires the interrupted task (same rationale as
931 // releaseStuckRun). Schedules are JSON, so this is per-row Effect work
932 // rather than a single UPDATE.
933 yield* Effect.gen(function* () {
934 const rows = yield* selectAllRows();
935 const stuck = rows.filter((row) => row.last_run_status === "running");
936 if (stuck.length === 0) return;
937 const now = yield* localNow;
938 yield* Effect.forEach(
939 stuck,
940 (row) =>
941 Effect.gen(function* () {
942 const decoded = yield* Effect.result(decodeRow(row));
943 if (Result.isSuccess(decoded)) {
944 yield* sql`
945 UPDATE scheduled_tasks
946 SET last_run_status = 'failed',
947 last_run_error = 'Run was interrupted by a server restart.',
948 next_run_at = ${nextRunAt(decoded.success, now)},
949 updated_at = ${iso(now)},
950 run_count = run_count + 1
951 WHERE task_id IS ${row.task_id} AND last_run_status = 'running'
952 `;
953 return;
954 }
955 // The schedule cannot be decoded, so the next occurrence cannot
956 // be computed — still release the row so it is not stuck in
957 // 'running' (the lenient poller skips it, so it cannot re-fire).
958 yield* Effect.logWarning(
959 "Recovering undecodable schedule task row without rescheduling",
960 { taskId: row.task_id, cause: decoded.failure },
961 );
962 yield* sql`
963 UPDATE scheduled_tasks
964 SET last_run_status = 'failed',
965 last_run_error = 'Run was interrupted by a server restart.',
966 updated_at = ${iso(now)},
967 run_count = run_count + 1
968 WHERE task_id IS ${row.task_id} AND last_run_status = 'running'
969 `;
970 }),
971 { concurrency: 1, discard: true },
972 );
973 yield* notifyChanged;
974 }).pipe(
975 Effect.catch((cause) =>
976 Effect.logWarning("Could not reset interrupted schedule task runs", { cause }),
977 ),
978 );
979
980 yield* scheduler.register("scheduled-tasks", runDueTasks());
981
982 const list: ScheduledTaskService["Service"]["list"] = () =>
983 listRows().pipe(
984 Effect.map((tasks) => ({ tasks })),
985 Effect.mapError((cause) => taskError("Could not list schedule tasks.", { cause })),
986 );
987
988 const subscribeList: ScheduledTaskService["Service"]["subscribeList"] = () =>
989 Stream.unwrap(
990 Effect.gen(function* () {
991 // Subscribe before taking the snapshot so a change landing between
992 // the two is buffered by the subscription rather than dropped.
993 const subscription = yield* PubSub.subscribe(changesPubSub);
994 return Stream.concat(
995 Stream.fromEffect(list()),
996 Stream.fromSubscription(subscription).pipe(Stream.mapEffect(() => list())),
997 );
998 }),
999 );
1000
1001 const upsert: ScheduledTaskService["Service"]["upsert"] = (input) =>
1002 Effect.gen(function* () {
1003 const now = yield* localNow;
1004 const uuid =
1005 input.commandId === undefined
1006 ? yield* crypto
1007 .randomUUIDv4()
1008 .pipe(
1009 Effect.mapError((cause) =>
1010 taskError("Could not generate schedule task id.", { cause }),
1011 ),
1012 )
1013 : null;
1014 const id =
1015 input.id ??
1016 ScheduledTaskId.make(
1017 input.commandId ? `scheduled-task:${input.commandId}` : `scheduled-task:${uuid}`,
1018 );
1019 // Look up by the *resolved* id so idempotent creates (commandId replays)
1020 // keep their run history, and so real load failures propagate instead
1021 // of silently resetting an existing row.
1022 const existingTask = yield* findTask(id);
1023 // Keep the existing next_run_at when the schedule itself is untouched:
1024 // editing a title or prompt must not postpone (or resurrect) a due
1025 // run — only schedule/enabled changes restart the clock.
1026 const schedule: ScheduledTask["schedule"] =
1027 input.schedule.type === "webhook"
1028 ? {
1029 type: "webhook",
1030 signature:
1031 input.schedule.signature == null
1032 ? null
1033 : {
1034 header: input.schedule.signature.header.toLowerCase(),
1035 encoding: input.schedule.signature.encoding,
1036 prefix: input.schedule.signature.prefix,
1037 },
1038 maxDeliveryAgeMinutes: input.schedule.maxDeliveryAgeMinutes ?? null,
1039 }
1040 : input.schedule;
1041 const webhook =
1042 input.schedule.type === "webhook"
1043 ? yield* Effect.gen(function* () {
1044 const existing = existingTask === null ? null : yield* findWebhookCredentials(id);
1045 // Saving a webhook task keeps its URL; only rotate changes it.
1046 const token = existing?.token ?? (yield* newWebhookToken);
1047 const signature =
1048 input.schedule.type === "webhook" ? input.schedule.signature : null;
1049 // A secretRef is a value the user entered for an agent; this
1050 // save consumes it, so it cannot be used again. A replay of a
1051 // save that already used it keeps the secret that save stored.
1052 const replay = input.commandId !== undefined && existing?.secret != null;
1053 const fromRef =
1054 signature?.secretRef === undefined
1055 ? undefined
1056 : yield* secretRequests
1057 .consume({ ref: signature.secretRef, projectId: input.projectId })
1058 .pipe(
1059 Effect.catch((error) =>
1060 replay
1061 ? Effect.succeed(undefined)
1062 : Effect.fail(taskError(error.message, { taskId: id })),
1063 ),
1064 );
1065 // A ref, when given, is the only source: a plain secret sent
1066 // alongside it must not replace what a replayed save stored.
1067 const provided = signature?.secretRef === undefined ? signature?.secret : fromRef;
1068 const secret = signature == null ? null : (provided ?? existing?.secret ?? null);
1069 const secretChanged =
1070 signature == null || provided !== undefined || existing === null;
1071 if (signature != null && secret === null) {
1072 return yield* taskError("A webhook signature check needs a signing secret.", {
1073 taskId: id,
1074 });
1075 }
1076 return { token, secret, secretChanged };
1077 })
1078 : { token: null, secret: null, secretChanged: true };
1079 const scheduleUnchanged =
1080 existingTask !== null &&
1081 existingTask.enabled === input.enabled &&
1082 isSameSchedule(existingTask.schedule, schedule);
1083 const task: ScheduledTask = {
1084 id,
1085 title: input.title,
1086 prompt: input.prompt,
1087 enabled: input.enabled,
1088 schedule,
1089 projectId: input.projectId,
1090 threadId: input.threadId ?? null,
1091 workspaceStrategy: input.workspaceStrategy,
1092 modelSelection: input.modelSelection,
1093 runtimeMode: input.runtimeMode,
1094 interactionMode: input.interactionMode,
1095 createdBy: existingTask?.createdBy ?? input.createdBy ?? "user",
1096 creationSource: input.creationSource ?? "web",
1097 createdAt: existingTask?.createdAt ?? iso(now),
1098 updatedAt: iso(now),
1099 nextRunAt: scheduleUnchanged
1100 ? existingTask.nextRunAt
1101 : nextRunAt({ enabled: input.enabled, schedule }, now),
1102 lastRunAt: existingTask?.lastRunAt ?? null,
1103 lastRunStatus: existingTask?.lastRunStatus ?? "never",
1104 lastRunError: existingTask?.lastRunError ?? null,
1105 runCount: existingTask?.runCount ?? 0,
1106 };
1107 yield* saveTask(task, input.requireExisting === true, webhook);
1108 yield* notifyChanged;
1109 return { task: (yield* findTask(id)) ?? task };
1110 });
1111
1112 const setEnabled: ScheduledTaskService["Service"]["setEnabled"] = (input) =>
1113 Effect.gen(function* () {
1114 const existing = yield* loadTask(input.id);
1115 if (existing.enabled === input.enabled) return { task: existing };
1116 const now = yield* localNow;
1117 const next = nextRunAt({ enabled: input.enabled, schedule: existing.schedule }, now);
1118 // RETURNING so a task deleted between the load and this UPDATE is a
1119 // visible not-found error, not a false success.
1120 const updated = yield* sql<{ task_id: string }>`
1121 UPDATE scheduled_tasks
1122 SET enabled = ${input.enabled ? 1 : 0},
1123 next_run_at = ${next},
1124 updated_at = ${iso(now)}
1125 WHERE task_id = ${input.id}
1126 RETURNING task_id
1127 `.pipe(
1128 Effect.mapError((cause) =>
1129 taskError("Could not update schedule task.", { taskId: input.id, cause }),
1130 ),
1131 );
1132 if (updated.length === 0) {
1133 return yield* taskError("Schedule task not found.", { taskId: input.id });
1134 }
1135 yield* notifyChanged;
1136 return {
1137 task: { ...existing, enabled: input.enabled, nextRunAt: next, updatedAt: iso(now) },
1138 };
1139 });
1140
1141 const deleteTask: ScheduledTaskService["Service"]["delete"] = (input) =>
1142 deleteRow(input.id).pipe(
1143 Effect.andThen(
1144 Effect.all([
1145 Ref.update(webhookRateWindows, (windows) => {
1146 const next = new Map(windows);
1147 next.delete(input.id);
1148 return next;
1149 }),
1150 Ref.update(webhookPermits, (permits) => {
1151 const next = new Map(permits);
1152 next.delete(input.id);
1153 return next;
1154 }),
1155 ]),
1156 ),
1157 Effect.andThen(notifyChanged),
1158 Effect.as({ id: input.id }),
1159 );
1160
1161 const runNow: ScheduledTaskService["Service"]["runNow"] = (input: ScheduledTaskRunNowInput) =>
1162 Effect.gen(function* () {
1163 const task = yield* loadTask(input.id);
1164 if (task.schedule.type === "webhook") {
1165 // There is no request to render the prompt from.
1166 return yield* taskError("Webhook tasks run when their URL receives a request.", {
1167 taskId: input.id,
1168 });
1169 }
1170 const next = yield* runTask(task, "manual").pipe(
1171 Effect.mapError((cause) =>
1172 taskError("Could not run schedule task.", { taskId: input.id, cause }),
1173 ),
1174 );
1175 return { task: next };
1176 });
1177
1178 const rotateWebhookToken: ScheduledTaskService["Service"]["rotateWebhookToken"] = (input) =>
1179 Effect.gen(function* () {
1180 const task = yield* loadTask(input.id);
1181 if (task.schedule.type !== "webhook") {
1182 return yield* taskError("Only webhook tasks have a URL token.", { taskId: input.id });
1183 }
1184 const token = yield* newWebhookToken;
1185 const now = yield* localNow;
1186 // Matching created_at keeps a rotation from landing on a task deleted
1187 // and recreated under the same id since it was loaded.
1188 const updated = yield* sql<{ task_id: string }>`
1189 UPDATE scheduled_tasks
1190 SET webhook_token = ${token}, updated_at = ${iso(now)}
1191 WHERE task_id = ${input.id} AND created_at = ${task.createdAt}
1192 RETURNING task_id
1193 `.pipe(
1194 Effect.mapError((cause) =>
1195 taskError("Could not rotate webhook token.", { taskId: input.id, cause }),
1196 ),
1197 );
1198 if (updated.length === 0) {
1199 return yield* taskError("Schedule task was deleted or replaced.", { taskId: input.id });
1200 }
1201 yield* notifyChanged;
1202 return { task: yield* loadTask(input.id) };
1203 });
1204
1205 const deliveryHeaders = (row: WebhookDeliveryRow): Readonly<Record<string, string>> =>
1206 Option.getOrElse(decodeHeadersJson(row.headers_json), () => ({}));
1207 const decodeDeliverySummary = (row: WebhookDeliveryRow) => ({
1208 id: ScheduledTaskWebhookDeliveryId.make(row.delivery_id),
1209 taskId: ScheduledTaskId.make(row.task_id),
1210 receivedAt: row.received_at,
1211 method: row.method,
1212 contentType: deliveryHeaders(row)["content-type"] ?? null,
1213 bodyBytes: row.body_bytes,
1214 outcome: row.outcome as ScheduledTaskWebhookDeliveryOutcome,
1215 signatureVerified: row.signature_verified === 1,
1216 missingFields: Option.getOrElse(decodeMissingFieldsJson(row.missing_fields_json), () => []),
1217 error: row.error,
1218 });
1219
1220 const listWebhookDeliveries: ScheduledTaskService["Service"]["listWebhookDeliveries"] = (
1221 input,
1222 ) =>
1223 sql<WebhookDeliveryRow>`
1224 SELECT * FROM scheduled_task_webhook_deliveries
1225 WHERE task_id = ${input.id}
1226 ORDER BY received_at DESC, rowid DESC
1227 `.pipe(
1228 Effect.map((rows) => ({ deliveries: rows.map(decodeDeliverySummary) })),
1229 Effect.mapError((cause) =>
1230 taskError("Could not list webhook deliveries.", { taskId: input.id, cause }),
1231 ),
1232 );
1233
1234 const getWebhookDelivery: ScheduledTaskService["Service"]["getWebhookDelivery"] = (input) =>
1235 Effect.gen(function* () {
1236 const rows = yield* sql<WebhookDeliveryRow>`
1237 SELECT * FROM scheduled_task_webhook_deliveries
1238 WHERE task_id = ${input.id} AND delivery_id = ${input.deliveryId}
1239 `.pipe(
1240 Effect.mapError((cause) =>
1241 taskError("Could not load webhook delivery.", { taskId: input.id, cause }),
1242 ),
1243 );
1244 const row = rows[0];
1245 if (row === undefined) {
1246 return yield* taskError("Webhook delivery not found.", { taskId: input.id });
1247 }
1248 return {
1249 delivery: {
1250 ...decodeDeliverySummary(row),
1251 query: row.query,
1252 headers: deliveryHeaders(row),
1253 body: row.body,
1254 bodyTruncated: row.body_truncated === 1,
1255 renderedPrompt: row.rendered_prompt,
1256 },
1257 };
1258 });
1259
1260 const recordDelivery = (input: {
1261 readonly id: string;
1262 readonly taskId: ScheduledTaskId;
1263 readonly receivedAt: string;
1264 readonly request: WebhookTriggerRequest;
1265 readonly outcome: ScheduledTaskWebhookDeliveryOutcome;
1266 readonly signatureVerified: boolean;
1267 readonly missing: ReadonlyArray<string>;
1268 readonly renderedPrompt: string | null;
1269 readonly error?: string;
1270 }) => {
1271 const truncated = input.request.body.byteLength > WEBHOOK_DELIVERY_LOG_BODY_LIMIT;
1272 const loggedBody = truncated
1273 ? new TextDecoder().decode(input.request.body.subarray(0, WEBHOOK_DELIVERY_LOG_BODY_LIMIT))
1274 : input.request.bodyText;
1275 return sql
1276 .withTransaction(
1277 Effect.gen(function* () {
1278 // Conditional on the task existing, so a delivery racing a delete
1279 // cannot leave rows that a recreated task with the same id would show.
1280 yield* sql`
1281 INSERT INTO scheduled_task_webhook_deliveries (
1282 delivery_id, task_id, received_at, method, query, headers_json, body,
1283 body_bytes, body_truncated, outcome, signature_verified,
1284 missing_fields_json, rendered_prompt, error
1285 )
1286 SELECT
1287 ${input.id}, ${input.taskId}, ${input.receivedAt}, ${input.request.method},
1288 ${redactQuery(input.request.query)},
1289 ${encodeHeadersJson(redactHeaders(input.request.headers))},
1290 ${loggedBody},
1291 ${input.request.body.byteLength}, ${truncated ? 1 : 0}, ${input.outcome},
1292 ${input.signatureVerified ? 1 : 0}, ${encodeMissingFieldsJson(input.missing)},
1293 ${input.renderedPrompt?.slice(0, WEBHOOK_DELIVERY_LOG_PROMPT_LIMIT) ?? null},
1294 ${input.error ?? null}
1295 WHERE EXISTS (SELECT 1 FROM scheduled_tasks WHERE task_id = ${input.taskId})
1296 -- A held request retried after a rate limit reuses its relay
1297 -- delivery id; the newer attempt replaces the logged one.
1298 ON CONFLICT (delivery_id) DO UPDATE SET
1299 outcome = excluded.outcome,
1300 signature_verified = excluded.signature_verified,
1301 missing_fields_json = excluded.missing_fields_json,
1302 rendered_prompt = excluded.rendered_prompt,
1303 error = excluded.error
1304 `;
1305 yield* sql`
1306 DELETE FROM scheduled_task_webhook_deliveries
1307 WHERE task_id = ${input.taskId}
1308 AND delivery_id NOT IN (
1309 SELECT delivery_id FROM scheduled_task_webhook_deliveries
1310 WHERE task_id = ${input.taskId}
1311 -- rowid breaks timestamp ties in arrival order; delivery ids are random.
1312 ORDER BY received_at DESC, rowid DESC
1313 LIMIT ${WEBHOOK_DELIVERY_RETENTION}
1314 )
1315 `;
1316 }),
1317 )
1318 .pipe(
1319 Effect.mapError((cause) =>
1320 taskError("Could not record webhook delivery.", { taskId: input.taskId, cause }),
1321 ),
1322 );
1323 };
1324
1325 const markDeliveryFailed = (deliveryId: string, message: string) =>
1326 sql`
1327 UPDATE scheduled_task_webhook_deliveries
1328 SET outcome = 'dispatch_failed', error = ${message}
1329 WHERE delivery_id = ${deliveryId}
1330 `.pipe(Effect.ignore);
1331
1332 /**
1333 * Sliding one-minute window counting every request with a valid token,
1334 * including ones the signature check later rejects.
1335 */
1336 const takeRateSlot = (id: ScheduledTaskId, nowMs: number) =>
1337 Ref.modify(
1338 webhookRateWindows,
1339 (
1340 windows,
1341 ): readonly [
1342 "allowed" | "first_rejected" | "rejected",
1343 ReadonlyMap<ScheduledTaskId, RateWindow>,
1344 ] => {
1345 const current = windows.get(id);
1346 const recent = (current?.accepted ?? []).filter((at) => nowMs - at < 60_000);
1347 if (recent.length >= WEBHOOK_RATE_LIMIT_PER_MINUTE) {
1348 const first = current?.rejectedLogged !== true;
1349 return [
1350 first ? "first_rejected" : "rejected",
1351 new Map(windows).set(id, { accepted: recent, rejectedLogged: true }),
1352 ];
1353 }
1354 return [
1355 "allowed",
1356 new Map(windows).set(id, { accepted: [...recent, nowMs], rejectedLogged: false }),
1357 ];
1358 },
1359 );
1360
1361 const webhookPermit = (id: ScheduledTaskId) =>
1362 Effect.gen(function* () {
1363 const existing = (yield* Ref.get(webhookPermits)).get(id);
1364 if (existing !== undefined) return existing;
1365 const created = yield* Semaphore.make(1);
1366 return yield* Ref.modify(webhookPermits, (permits) => {
1367 const raced = permits.get(id);
1368 return raced === undefined
1369 ? [created, new Map(permits).set(id, created)]
1370 : [raced, permits];
1371 });
1372 });
1373
1374 // Detached from the request so the HTTP response does not wait for the
1375 // run; scoped to the service so shutdown interrupts it.
1376 const serviceScope = yield* Effect.scope;
1377
1378 /**
1379 * Records one handled request on the span and in metrics. `outcome` is
1380 * finer than the result tag: it separates a full queue and an oversized
1381 * prompt from the rest, which is what an operator needs to act on.
1382 */
1383 const observeDelivery = (
1384 request: WebhookTriggerRequest,
1385 outcome: WebhookDeliveryOutcome,
1386 receivedAt: DateTime.DateTime | undefined,
1387 now: DateTime.DateTime | undefined,
1388 ) =>
1389 Effect.gen(function* () {
1390 const source = request.relayDeliveryId === undefined ? "direct" : "relay";
1391 yield* Effect.annotateCurrentSpan({
1392 "scheduled_task.webhook.outcome": outcome,
1393 "scheduled_task.webhook.source": source,
1394 });
1395 yield* Metrics.increment(Metrics.webhookDeliveriesTotal, { outcome, source });
1396 if (request.receivedAt !== undefined && receivedAt !== undefined && now !== undefined) {
1397 const heldMs = Math.max(
1398 0,
1399 DateTime.toEpochMillis(now) - DateTime.toEpochMillis(receivedAt),
1400 );
1401 yield* Effect.annotateCurrentSpan({ "scheduled_task.webhook.held_ms": heldMs });
1402 yield* Metric.update(Metrics.webhookHeldDelay, Duration.millis(heldMs));
1403 }
1404 });
1405
1406 const triggerWebhook: ScheduledTaskService["Service"]["triggerWebhook"] = (request) =>
1407 triggerWebhookUnobserved(request).pipe(
1408 Effect.tapError(() => observeDelivery(request, "error", undefined, undefined)),
1409 Metrics.withMetrics({ timer: Metrics.webhookDeliveryDuration }),
1410 Effect.withSpan("ScheduledTaskService.triggerWebhook", {
1411 attributes: {
1412 "scheduled_task.webhook.method": request.method,
1413 "scheduled_task.webhook.body_bytes": request.body.byteLength,
1414 "scheduled_task.webhook.source":
1415 request.relayDeliveryId === undefined ? "direct" : "relay",
1416 },
1417 }),
1418 );
1419
1420 const triggerWebhookUnobserved = (request: WebhookTriggerRequest) =>
1421 Effect.gen(function* () {
1422 const taskId = decodeTaskId(request.hookId);
1423 const notFound = observeDelivery(request, "not_found", undefined, undefined).pipe(
1424 Effect.as({ _tag: "not_found" as const }),
1425 );
1426 if (Option.isNone(taskId)) return yield* notFound;
1427 const rows = yield* getRows(taskId.value).pipe(
1428 Effect.mapError((cause) =>
1429 taskError("Could not load schedule task.", { taskId: taskId.value, cause }),
1430 ),
1431 );
1432 const row = rows[0];
1433 // A wrong token is indistinguishable from an unknown hook, so the URL
1434 // does not reveal which hooks exist.
1435 if (
1436 row === undefined ||
1437 row.webhook_token === null ||
1438 !constantTimeEquals(request.token, row.webhook_token)
1439 ) {
1440 return yield* notFound;
1441 }
1442 const task = yield* decodeRow(row);
1443 yield* Effect.annotateCurrentSpan({ "scheduled_task.id": task.id });
1444 const schedule = task.schedule;
1445 if (schedule.type !== "webhook") return yield* notFound;
1446
1447 const now = yield* localNow;
1448 // Anyone can reach the tunnel directly and set the relay's header, so
1449 // a receive time is never later than now: a future one would pin the
1450 // delivery in the log and slip past the task's max age.
1451 const receivedAt =
1452 request.receivedAt === undefined
1453 ? now
1454 : DateTime.min(
1455 Option.getOrElse(DateTime.make(request.receivedAt), () => now),
1456 now,
1457 );
1458 const observe = (outcome: WebhookDeliveryOutcome) =>
1459 observeDelivery(request, outcome, receivedAt, now);
1460 const deliveryId = ScheduledTaskWebhookDeliveryId.make(
1461 request.relayDeliveryId === undefined
1462 ? `delivery:${yield* crypto
1463 .randomUUIDv4()
1464 .pipe(
1465 Effect.mapError((cause) =>
1466 taskError("Could not generate delivery id.", { cause }),
1467 ),
1468 )}`
1469 : `delivery:relay:${request.relayDeliveryId}`,
1470 );
1471 // A held request may already have reached this environment directly
1472 // before a timeout; it runs once. Claimed in its own table, kept longer
1473 // than the relay holds a request, because the delivery log is trimmed.
1474 // A rate-limited delivery stays held on the relay, so it must give up
1475 // its claim or the next pass would treat it as already delivered.
1476 const releaseClaim = <A>(result: A) =>
1477 request.relayDeliveryId === undefined
1478 ? Effect.succeed(result)
1479 : sql`
1480 DELETE FROM scheduled_task_webhook_relay_deliveries
1481 WHERE relay_delivery_id = ${request.relayDeliveryId}
1482 `.pipe(Effect.ignore, Effect.as(result));
1483 // From the claim until the run is forked nothing may interrupt: a
1484 // request dropped in between (the relay hangs up after its timeout)
1485 // would leave a claimed delivery that never runs, or a queue slot
1486 // that is never released. Everything in here is local and quick.
1487 return yield* Effect.uninterruptible(
1488 Effect.gen(function* () {
1489 if (request.relayDeliveryId !== undefined) {
1490 const claimed = yield* sql<{ relay_delivery_id: string }>`
1491 INSERT INTO scheduled_task_webhook_relay_deliveries
1492 (relay_delivery_id, task_id, seen_at)
1493 VALUES (${request.relayDeliveryId}, ${task.id}, ${iso(now)})
1494 ON CONFLICT (relay_delivery_id) DO NOTHING
1495 RETURNING relay_delivery_id
1496 `.pipe(
1497 Effect.mapError((cause) =>
1498 taskError("Could not record webhook delivery.", { taskId: task.id, cause }),
1499 ),
1500 );
1501 if (claimed.length === 0) {
1502 // A held request this environment already ran: accepted, not run twice.
1503 yield* observe("duplicate");
1504 return { _tag: "accepted" as const, deliveryId, outcome: "duplicate" as const };
1505 }
1506 // Older claims can no longer be replayed by the relay.
1507 yield* sql`
1508 DELETE FROM scheduled_task_webhook_relay_deliveries
1509 WHERE seen_at < ${iso(DateTime.subtract(now, { hours: 48 }))}
1510 `.pipe(Effect.ignore);
1511 }
1512 const log = (
1513 outcome: ScheduledTaskWebhookDeliveryOutcome,
1514 details: {
1515 readonly signatureVerified?: boolean;
1516 readonly missing?: ReadonlyArray<string>;
1517 readonly renderedPrompt?: string;
1518 readonly error?: string;
1519 } = {},
1520 ) =>
1521 recordDelivery({
1522 id: deliveryId,
1523 taskId: task.id,
1524 receivedAt: iso(receivedAt),
1525 request,
1526 outcome,
1527 signatureVerified: details.signatureVerified ?? false,
1528 missing: details.missing ?? [],
1529 renderedPrompt: details.renderedPrompt ?? null,
1530 ...(details.error === undefined ? {} : { error: details.error }),
1531 });
1532
1533 // Only the first rejected request in a window is logged, so a flood
1534 // cannot write rows or push the real deliveries out of the log.
1535 const slot = yield* takeRateSlot(task.id, DateTime.toEpochMillis(now));
1536 if (slot !== "allowed") {
1537 if (slot === "first_rejected") yield* log("rate_limited");
1538 yield* observe("rate_limited");
1539 return yield* releaseClaim({
1540 _tag: "rate_limited" as const,
1541 outcome: "rate_limited" as const,
1542 });
1543 }
1544 if (!task.enabled) {
1545 yield* log("disabled");
1546 yield* observe("disabled");
1547 return { _tag: "disabled" as const };
1548 }
1549 const signature = schedule.signature;
1550 if (signature !== null) {
1551 const verified =
1552 row.webhook_secret !== null &&
1553 verifyWebhookSignature({
1554 signature,
1555 secret: row.webhook_secret,
1556 headers: request.headers,
1557 body: request.body,
1558 });
1559 if (!verified) {
1560 yield* log("rejected_signature");
1561 yield* observe("rejected_signature");
1562 return { _tag: "rejected_signature" as const };
1563 }
1564 }
1565 const maxAgeMinutes = schedule.maxDeliveryAgeMinutes ?? null;
1566 if (
1567 maxAgeMinutes !== null &&
1568 DateTime.toEpochMillis(now) - DateTime.toEpochMillis(receivedAt) >
1569 maxAgeMinutes * 60_000
1570 ) {
1571 yield* log("expired", { signatureVerified: signature !== null });
1572 yield* observe("expired");
1573 return { _tag: "expired" as const };
1574 }
1575
1576 const rendered = renderWebhookPrompt(task.prompt, request);
1577 // A provider refuses a turn this long, so it is not started. The
1578 // delivery is not retryable, so the claim is kept. Providers trim
1579 // the prompt before checking, so whitespace around it is free.
1580 if (rendered.prompt.trim().length > PROVIDER_SEND_TURN_MAX_INPUT_CHARS) {
1581 yield* log("dispatch_failed", {
1582 signatureVerified: signature !== null,
1583 missing: rendered.missing,
1584 renderedPrompt: rendered.prompt,
1585 error: "The filled-in prompt is too long.",
1586 });
1587 yield* observe("prompt_too_long");
1588 return { _tag: "accepted" as const, deliveryId, outcome: "prompt_too_long" as const };
1589 }
1590 // Bound the deliveries one task holds, so steady traffic to a stuck
1591 // task cannot pile up parked fibers. A refused request is not logged,
1592 // so it cannot push real deliveries out of the log.
1593 const queueKey = `${task.id}\u0000${task.createdAt}`;
1594 const queued = yield* Ref.modify(webhookQueued, (counts) => {
1595 const count = counts.get(queueKey) ?? 0;
1596 return count >= WEBHOOK_MAX_QUEUED_PER_TASK
1597 ? ([false, counts] as const)
1598 : ([true, new Map(counts).set(queueKey, count + 1)] as const);
1599 });
1600 if (!queued) {
1601 // The task is busy with WEBHOOK_MAX_QUEUED_PER_TASK deliveries already.
1602 yield* observe("queue_full");
1603 return yield* releaseClaim({
1604 _tag: "rate_limited" as const,
1605 outcome: "queue_full" as const,
1606 });
1607 }
1608 // Entries leave the map when their count reaches zero, so a deleted
1609 // task's key does not linger once its last delivery finishes.
1610 const release = Ref.update(webhookQueued, (counts) => {
1611 const next = new Map(counts);
1612 const count = (next.get(queueKey) ?? 1) - 1;
1613 if (count <= 0) next.delete(queueKey);
1614 else next.set(queueKey, count);
1615 return next;
1616 });
1617 yield* log("accepted", {
1618 signatureVerified: signature !== null,
1619 missing: rendered.missing,
1620 renderedPrompt: rendered.prompt,
1621 }).pipe(Effect.onError(() => release));
1622 yield* observe("accepted");
1623 const permit = yield* webhookPermit(task.id);
1624 const runOutcome = (outcome: "started" | "skipped" | "failed") =>
1625 Effect.all([
1626 Effect.annotateCurrentSpan({ "scheduled_task.webhook.run_outcome": outcome }),
1627 Metrics.increment(Metrics.webhookRunsTotal, { outcome }),
1628 ]);
1629 yield* runTask(task, "webhook", { deliveryId, prompt: rendered.prompt }).pipe(
1630 Effect.flatMap((completed) =>
1631 completed.lastRunStatus === "failed"
1632 ? runOutcome("failed").pipe(
1633 Effect.andThen(markDeliveryFailed(deliveryId, "The run failed to start.")),
1634 )
1635 : runOutcome("started"),
1636 ),
1637 Effect.catchTags({
1638 WebhookDeliverySkipped: (skipped) =>
1639 runOutcome("skipped").pipe(
1640 Effect.andThen(markDeliveryFailed(deliveryId, skipped.reason)),
1641 ),
1642 }),
1643 // The log is readable over RPC, so it gets a fixed reason; the
1644 // cause, which can carry request data, stays in the server log.
1645 Effect.catchCause((cause) =>
1646 Effect.logWarning("Webhook dispatch failed", { taskId: task.id, cause }).pipe(
1647 Effect.andThen(runOutcome("failed")),
1648 Effect.andThen(markDeliveryFailed(deliveryId, "The run failed to start.")),
1649 ),
1650 ),
1651 permit.withPermits(1),
1652 // Its own trace: the request that triggered it has already been answered.
1653 Effect.withSpan("ScheduledTaskService.runWebhookDelivery", {
1654 root: true,
1655 attributes: {
1656 "scheduled_task.id": task.id,
1657 "scheduled_task.webhook.delivery_id": deliveryId,
1658 },
1659 }),
1660 Effect.ensuring(release),
1661 Effect.forkIn(serviceScope),
1662 );
1663 return { _tag: "accepted" as const, deliveryId, outcome: "accepted" as const };
1664 }),
1665 );
1666 });
1667
1668 return ScheduledTaskService.of({
1669 list,
1670 subscribeList,
1671 upsert,
1672 setEnabled,
1673 delete: deleteTask,
1674 runNow,
1675 rotateWebhookToken,
1676 listWebhookDeliveries,
1677 getWebhookDelivery,
1678 triggerWebhook,
1679 });
1680 }),
1681);