authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 15:14:51-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 15:14:51-07:00
log7dfb86a32be6ec064dd8988e1a3b44cc046ff958
tree6211f3b521b89240903dbf1d7f1082dea147c39f
parent54b6b667f340c03132e9fc2cb2c3d657571442a2
signature Signed by PGP key B5690EEEBB952194

feat(server): scheduled tasks can run on a webhook (#15085)


32 files changed, 2079 insertions(+), 53 deletions(-)

apps/mobile/src/features/settings/SettingsScheduledTasksRouteScreen.tsx+4-1
...@@ -87,6 +87,7 @@ const DAYS = [...@@ -87,6 +87,7 @@ const DAYS = [
87] as const;87] as const;
8888
89function describeSchedule(task: ScheduledTask): string {89function describeSchedule(task: ScheduledTask): string {
90 if (task.schedule.type === "webhook") return "On webhook";
90 if (task.schedule.type === "interval") return formatScheduledTaskInterval(task.schedule.everyMs);91 if (task.schedule.type === "interval") return formatScheduledTaskInterval(task.schedule.everyMs);
91 const days = task.schedule.weekdays?.length ? repeatLabel(task.schedule.weekdays) : "Every day";92 const days = task.schedule.weekdays?.length ? repeatLabel(task.schedule.weekdays) : "Every day";
92 return `${days} at ${formatTime(task.schedule.timeOfDay)}`;93 return `${days} at ${formatTime(task.schedule.timeOfDay)}`;
...@@ -1024,6 +1025,8 @@ function EnvironmentTasks({...@@ -1024,6 +1025,8 @@ function EnvironmentTasks({
1024 <Pressable1025 <Pressable
1025 accessibilityRole="button"1026 accessibilityRole="button"
1026 accessibilityLabel={`Edit ${task.title}`}1027 accessibilityLabel={`Edit ${task.title}`}
1028 // Webhook tasks are not editable here yet; saving would drop their URL.
1029 disabled={task.schedule.type === "webhook"}
1027 onPress={() => {1030 onPress={() => {
1028 onEdit(task);1031 onEdit(task);
1029 }}1032 }}
...@@ -1048,7 +1051,7 @@ function EnvironmentTasks({...@@ -1048,7 +1051,7 @@ function EnvironmentTasks({
1048 </Pressable>1051 </Pressable>
1049 <ControlPillMenu1052 <ControlPillMenu
1050 actions={[1053 actions={[
1051 { id: "edit", title: "Edit" },1054 ...(task.schedule.type === "webhook" ? [] : [{ id: "edit", title: "Edit" }]),
1052 { id: "toggle", title: task.enabled ? "Pause" : "Resume" },1055 { id: "toggle", title: task.enabled ? "Pause" : "Resume" },
1053 { id: "run", title: "Run now" },1056 { id: "run", title: "Run now" },
1054 { id: "delete", title: "Delete", attributes: { destructive: true } },1057 { id: "delete", title: "Delete", attributes: { destructive: true } },
apps/mobile/src/features/settings/scheduledTaskDraft.ts+2
...@@ -51,6 +51,8 @@ export const DEFAULT_SCHEDULE: ScheduleDraft = {...@@ -51,6 +51,8 @@ export const DEFAULT_SCHEDULE: ScheduleDraft = {
51};51};
5252
53export function scheduleDraftForTask(task: Pick<ScheduledTask, "schedule">): ScheduleDraft {53export function scheduleDraftForTask(task: Pick<ScheduledTask, "schedule">): ScheduleDraft {
54 // Webhook tasks cannot be edited here yet; show them as the default schedule.
55 if (task.schedule.type === "webhook") return DEFAULT_SCHEDULE;
54 return task.schedule.type === "fixed_time"56 return task.schedule.type === "fixed_time"
55 ? {57 ? {
56 ...DEFAULT_SCHEDULE,58 ...DEFAULT_SCHEDULE,
apps/server/src/auth/RpcAuthorization.test.ts+10
...@@ -38,6 +38,16 @@ describe("RPC authorization scopes", () => {...@@ -38,6 +38,16 @@ describe("RPC authorization scopes", () => {
38 );38 );
39 });39 });
4040
41 it("keeps webhook delivery logs, which hold request bodies, behind operate scope", () => {
42 for (const method of [
43 WS_METHODS.scheduledTasksListWebhookDeliveries,
44 WS_METHODS.scheduledTasksGetWebhookDelivery,
45 WS_METHODS.scheduledTasksRotateWebhookToken,
46 ]) {
47 expect(requiredScopeForRpcMethod(method)).toBe(AuthOrchestrationOperateScope);
48 }
49 });
50
41 it("allows relay status reads without granting relay installation access", () => {51 it("allows relay status reads without granting relay installation access", () => {
42 expect(requiredScopeForRpcMethod(WS_METHODS.cloudGetRelayClientStatus)).toBe(52 expect(requiredScopeForRpcMethod(WS_METHODS.cloudGetRelayClientStatus)).toBe(
43 AuthRelayReadScope,53 AuthRelayReadScope,
apps/server/src/auth/RpcAuthorization.ts+4
...@@ -95,6 +95,10 @@ export const RPC_REQUIRED_SCOPES = {...@@ -95,6 +95,10 @@ export const RPC_REQUIRED_SCOPES = {
95 [WS_METHODS.scheduledTasksSetEnabled]: AuthOrchestrationOperateScope,95 [WS_METHODS.scheduledTasksSetEnabled]: AuthOrchestrationOperateScope,
96 [WS_METHODS.scheduledTasksDelete]: AuthOrchestrationOperateScope,96 [WS_METHODS.scheduledTasksDelete]: AuthOrchestrationOperateScope,
97 [WS_METHODS.scheduledTasksRunNow]: AuthOrchestrationOperateScope,97 [WS_METHODS.scheduledTasksRunNow]: AuthOrchestrationOperateScope,
98 [WS_METHODS.scheduledTasksRotateWebhookToken]: AuthOrchestrationOperateScope,
99 // Delivery logs hold request bodies, so they need the same scope as the URL.
100 [WS_METHODS.scheduledTasksListWebhookDeliveries]: AuthOrchestrationOperateScope,
101 [WS_METHODS.scheduledTasksGetWebhookDelivery]: AuthOrchestrationOperateScope,
98 [WS_METHODS.cloudGetRelayClientStatus]: AuthRelayReadScope,102 [WS_METHODS.cloudGetRelayClientStatus]: AuthRelayReadScope,
99 [WS_METHODS.cloudInstallRelayClient]: AuthRelayWriteScope,103 [WS_METHODS.cloudInstallRelayClient]: AuthRelayWriteScope,
100 [WS_METHODS.pullRequestsList]: AuthOrchestrationReadScope,104 [WS_METHODS.pullRequestsList]: AuthOrchestrationReadScope,
apps/server/src/http.ts+4-3
...@@ -46,6 +46,7 @@ import {...@@ -46,6 +46,7 @@ import {
46 failEnvironmentInternal,46 failEnvironmentInternal,
47} from "./auth/http.ts";47} from "./auth/http.ts";
48import * as ServerEnvironment from "./environment/ServerEnvironment.ts";48import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
49import { WEBHOOK_ROUTE_PREFIX } from "./scheduledTasks/ScheduledTaskService.ts";
49import { browserApiCorsAllowedHeaders, browserApiCorsAllowedMethods } from "./httpCors.ts";50import { browserApiCorsAllowedHeaders, browserApiCorsAllowedMethods } from "./httpCors.ts";
5051
51const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces";52const OTLP_TRACES_PROXY_PATH = "/api/observability/v1/traces";
...@@ -381,9 +382,9 @@ const UNTRACED_REQUEST_PATHS: ReadonlySet<string> = new Set([OTLP_TRACES_PROXY_P...@@ -381,9 +382,9 @@ const UNTRACED_REQUEST_PATHS: ReadonlySet<string> = new Set([OTLP_TRACES_PROXY_P
381// ignored, as in routing.382// ignored, as in routing.
382const untracedRequestsLayer = Layer.succeed(HttpMiddleware.TracerDisabledWhen)((request) => {383const untracedRequestsLayer = Layer.succeed(HttpMiddleware.TracerDisabledWhen)((request) => {
383 const queryIndex = request.url.indexOf("?");384 const queryIndex = request.url.indexOf("?");
384 return UNTRACED_REQUEST_PATHS.has(385 const path = queryIndex === -1 ? request.url : request.url.slice(0, queryIndex);
385 queryIndex === -1 ? request.url : request.url.slice(0, queryIndex),386 // Webhook URLs carry their secret token in the path, so they never reach a trace.
386 );387 return UNTRACED_REQUEST_PATHS.has(path) || path.startsWith(`${WEBHOOK_ROUTE_PREFIX}/`);
387});388});
388389
389export const withUntracedRequests = Layer.provide(untracedRequestsLayer);390export const withUntracedRequests = Layer.provide(untracedRequestsLayer);
apps/server/src/mcp/OrchestratorMcpService.ts+1
...@@ -212,6 +212,7 @@ function scheduledTaskSummary(task: ScheduledTask): OrchestratorMcpScheduledTask...@@ -212,6 +212,7 @@ function scheduledTaskSummary(task: ScheduledTask): OrchestratorMcpScheduledTask
212 schedule: task.schedule,212 schedule: task.schedule,
213 nextRunAt: task.nextRunAt,213 nextRunAt: task.nextRunAt,
214 lastRunStatus: task.lastRunStatus,214 lastRunStatus: task.lastRunStatus,
215 ...(task.webhook === undefined ? {} : { webhookUrl: task.webhook.url ?? task.webhook.path }),
215 };216 };
216}217}
217218
apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts+10-1
...@@ -443,7 +443,8 @@ function scheduledTaskFromUpsert(input: ScheduledTaskUpsertInput): ScheduledTask...@@ -443,7 +443,8 @@ function scheduledTaskFromUpsert(input: ScheduledTaskUpsertInput): ScheduledTask
443 title: input.title,443 title: input.title,
444 prompt: input.prompt,444 prompt: input.prompt,
445 enabled: input.enabled,445 enabled: input.enabled,
446 schedule: input.schedule,446 schedule:
447 input.schedule.type === "webhook" ? { type: "webhook", signature: null } : input.schedule,
447 projectId: input.projectId,448 projectId: input.projectId,
448 threadId: input.threadId ?? null,449 threadId: input.threadId ?? null,
449 workspaceStrategy: input.workspaceStrategy,450 workspaceStrategy: input.workspaceStrategy,
...@@ -471,6 +472,10 @@ const unusedScheduledTaskStubLayer = Layer.succeed(...@@ -471,6 +472,10 @@ const unusedScheduledTaskStubLayer = Layer.succeed(
471 setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"),472 setEnabled: () => Effect.die("ScheduledTaskService.setEnabled is unused in this test"),
472 delete: () => Effect.die("ScheduledTaskService.delete is unused in this test"),473 delete: () => Effect.die("ScheduledTaskService.delete is unused in this test"),
473 runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),474 runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),
475 rotateWebhookToken: () => Effect.die("unused in this test"),
476 listWebhookDeliveries: () => Effect.die("unused in this test"),
477 getWebhookDelivery: () => Effect.die("unused in this test"),
478 triggerWebhook: () => Effect.die("unused in this test"),
474 }),479 }),
475);480);
476481
...@@ -628,6 +633,10 @@ describe("orchestrator MCP toolkit", () => {...@@ -628,6 +633,10 @@ describe("orchestrator MCP toolkit", () => {
628 all.filter((candidate) => candidate.id !== input.id),633 all.filter((candidate) => candidate.id !== input.id),
629 ).pipe(Effect.as({ id: input.id })),634 ).pipe(Effect.as({ id: input.id })),
630 runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),635 runNow: () => Effect.die("ScheduledTaskService.runNow is unused in this test"),
636 rotateWebhookToken: () => Effect.die("unused in this test"),
637 listWebhookDeliveries: () => Effect.die("unused in this test"),
638 getWebhookDelivery: () => Effect.die("unused in this test"),
639 triggerWebhook: () => Effect.die("unused in this test"),
631 }),640 }),
632 );641 );
633 const testLayer = Layer.merge(642 const testLayer = Layer.merge(
apps/server/src/mcp/toolkits/orchestrator/tools.ts+1-1
...@@ -97,7 +97,7 @@ const TaskCancelTool = Tool.make("task_cancel", {...@@ -97,7 +97,7 @@ const TaskCancelTool = Tool.make("task_cancel", {
9797
98export const ScheduleTaskTool = Tool.make("schedule_task", {98export const ScheduleTaskTool = Tool.make("schedule_task", {
99 description:99 description:
100 "Create persistent recurring work in the app scheduler, which runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text: {type:'interval', everyMs:3600000} means hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} means weekday mornings. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true); use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread. Report the returned schedule and nextRunAt after success.",100 "Create persistent recurring work in the app scheduler, which runs even when no turn is active. Pass schedule as a STRUCTURED OBJECT, never JSON text: {type:'interval', everyMs:3600000} means hourly; {type:'fixed_time', timeOfDay:'09:00', weekdays:[1,2,3,4,5]} means weekday mornings; {type:'webhook'} runs on each request to a generated URL (returned as webhookUrl), and its prompt may use {{body.path}}, {{headers.name}}, {{query.name}}, {{body}} or {{request}} placeholders, which are the only request data the run sees. Omit projectId for this thread's project. In this thread's project, runs post into THIS thread by default (bindToCurrentThread=true); use false only when the user wants a fresh top-level thread per run. Elsewhere each run launches a fresh thread. Provider, model, and runtime settings inherit from this thread, or from the project default when there is no calling thread. Report the returned schedule and nextRunAt after success.",
101 parameters: OrchestratorMcpScheduleTaskInput,101 parameters: OrchestratorMcpScheduleTaskInput,
102 success: OrchestratorMcpScheduleTaskResult,102 success: OrchestratorMcpScheduleTaskResult,
103 failure: OrchestratorMcpFailure,103 failure: OrchestratorMcpFailure,
apps/server/src/persistence/Migrations.ts+2
...@@ -70,6 +70,7 @@ import Migration0053 from "./Migrations/053_PullRequestFilesViewed.ts";...@@ -70,6 +70,7 @@ import Migration0053 from "./Migrations/053_PullRequestFilesViewed.ts";
70import Migration0054 from "./Migrations/054_ProjectionThreadsAutoSettleDisabledAt.ts";70import Migration0054 from "./Migrations/054_ProjectionThreadsAutoSettleDisabledAt.ts";
71import Migration0055 from "./Migrations/055_OrchestrationV2.ts";71import Migration0055 from "./Migrations/055_OrchestrationV2.ts";
72import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts";72import Migration0056 from "./Migrations/056_RemoveRedundantProjectionIndexes.ts";
73import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts";
7374
74/**75/**
75 * Migration loader with all migrations defined inline.76 * Migration loader with all migrations defined inline.
...@@ -140,6 +141,7 @@ export const migrationEntries = [...@@ -140,6 +141,7 @@ export const migrationEntries = [
140 // Preserve this migration's schema. Future V2 schema changes need new migrations.141 // Preserve this migration's schema. Future V2 schema changes need new migrations.
141 [55, "OrchestrationV2", Migration0055],142 [55, "OrchestrationV2", Migration0055],
142 [56, "RemoveRedundantProjectionIndexes", Migration0056],143 [56, "RemoveRedundantProjectionIndexes", Migration0056],
144 [57, "ScheduledTaskWebhooks", Migration0057],
143] as const;145] as const;
144146
145export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);147export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);
apps/server/src/persistence/Migrations/055_OrchestrationV2.test.ts+3-1
...@@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -13,7 +13,7 @@ layer("055_OrchestrationV2", (it) => {
13 Effect.sync(() => {13 Effect.sync(() => {
14 assert.deepStrictEqual(14 assert.deepStrictEqual(
15 migrationEntries.map(([id]) => id),15 migrationEntries.map(([id]) => id),
16 Array.from({ length: 56 }, (_, index) => index + 1),16 Array.from({ length: 57 }, (_, index) => index + 1),
17 );17 );
18 }),18 }),
19 );19 );
...@@ -28,6 +28,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -28,6 +28,7 @@ layer("055_OrchestrationV2", (it) => {
28 [54, "ProjectionThreadsAutoSettleDisabledAt"],28 [54, "ProjectionThreadsAutoSettleDisabledAt"],
29 [55, "OrchestrationV2"],29 [55, "OrchestrationV2"],
30 [56, "RemoveRedundantProjectionIndexes"],30 [56, "RemoveRedundantProjectionIndexes"],
31 [57, "ScheduledTaskWebhooks"],
31 ]);32 ]);
32 assert.deepStrictEqual(yield* runMigrations(), []);33 assert.deepStrictEqual(yield* runMigrations(), []);
3334
...@@ -50,6 +51,7 @@ layer("055_OrchestrationV2", (it) => {...@@ -50,6 +51,7 @@ layer("055_OrchestrationV2", (it) => {
50 { migration_id: 54, name: "ProjectionThreadsAutoSettleDisabledAt" },51 { migration_id: 54, name: "ProjectionThreadsAutoSettleDisabledAt" },
51 { migration_id: 55, name: "OrchestrationV2" },52 { migration_id: 55, name: "OrchestrationV2" },
52 { migration_id: 56, name: "RemoveRedundantProjectionIndexes" },53 { migration_id: 56, name: "RemoveRedundantProjectionIndexes" },
54 { migration_id: 57, name: "ScheduledTaskWebhooks" },
53 ]);55 ]);
5456
55 const tables = yield* sql<{ readonly name: string }>`57 const tables = yield* sql<{ readonly name: string }>`
apps/server/src/persistence/Migrations/057_ScheduledTaskWebhooks.test.ts created+51
...@@ -0,0 +1,51 @@
1import { assert, it } from "@effect/vitest";
2import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as SqlClient from "effect/sql/SqlClient";
6
7import { runMigrations } from "../Migrations.ts";
8
9const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layer({ filename: ":memory:" })));
10
11layer("057_ScheduledTaskWebhooks", (it) => {
12 it.effect("keeps existing scheduled tasks and adds webhook storage", () =>
13 Effect.gen(function* () {
14 const sql = yield* SqlClient.SqlClient;
15 yield* runMigrations({ toMigrationInclusive: 56 });
16 yield* sql`INSERT INTO scheduled_tasks ${sql.insert({
17 task_id: "existing",
18 title: "task",
19 prompt: "Run",
20 enabled: 1,
21 schedule_json: '{"type":"interval","everyMs":60000}',
22 project_id: "project",
23 thread_id: null,
24 workspace_strategy_json: '{"type":"root"}',
25 model_selection_json: '{"instanceId":"codex","model":"gpt-5"}',
26 runtime_mode: "full-access",
27 interaction_mode: "default",
28 created_by: "user",
29 creation_source: "web",
30 created_at: "2026-10-01T00:00:00.000Z",
31 updated_at: "2026-10-01T00:00:00.000Z",
32 next_run_at: null,
33 last_run_at: null,
34 last_run_status: "never",
35 last_run_error: null,
36 run_count: 0,
37 })}`;
38 yield* runMigrations({ toMigrationInclusive: 57 });
39
40 const rows = yield* sql<{
41 task_id: string;
42 webhook_token: string | null;
43 }>`SELECT task_id, webhook_token FROM scheduled_tasks`;
44 assert.deepEqual(rows, [{ task_id: "existing", webhook_token: null }]);
45 const deliveries = yield* sql<{
46 count: number;
47 }>`SELECT COUNT(*) AS count FROM scheduled_task_webhook_deliveries`;
48 assert.equal(deliveries[0]?.count, 0);
49 }),
50 );
51});
apps/server/src/persistence/Migrations/057_ScheduledTaskWebhooks.ts created+34
...@@ -0,0 +1,34 @@
1import * as Effect from "effect/Effect";
2import * as SqlClient from "effect/sql/SqlClient";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 // Kept out of schedule_json so they never decode into the read model.
8 yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN webhook_token TEXT`;
9 yield* sql`ALTER TABLE scheduled_tasks ADD COLUMN webhook_secret TEXT`;
10
11 yield* sql`
12 CREATE TABLE IF NOT EXISTS scheduled_task_webhook_deliveries (
13 delivery_id TEXT PRIMARY KEY,
14 task_id TEXT NOT NULL,
15 received_at TEXT NOT NULL,
16 method TEXT NOT NULL,
17 query TEXT NOT NULL,
18 headers_json TEXT NOT NULL,
19 body TEXT NOT NULL,
20 body_bytes INTEGER NOT NULL,
21 body_truncated INTEGER NOT NULL,
22 outcome TEXT NOT NULL,
23 signature_verified INTEGER NOT NULL,
24 missing_fields_json TEXT NOT NULL,
25 rendered_prompt TEXT,
26 error TEXT
27 )
28 `;
29
30 yield* sql`
31 CREATE INDEX IF NOT EXISTS idx_scheduled_task_webhook_deliveries_task
32 ON scheduled_task_webhook_deliveries(task_id, received_at)
33 `;
34});
apps/server/src/persistence/reconcileV2PreviewMigration.test.ts+2
...@@ -37,6 +37,7 @@ describe("V2 preview upgrade", () => {...@@ -37,6 +37,7 @@ describe("V2 preview upgrade", () => {
37 [53, "PullRequestFilesViewed"],37 [53, "PullRequestFilesViewed"],
38 [54, "ProjectionThreadsAutoSettleDisabledAt"],38 [54, "ProjectionThreadsAutoSettleDisabledAt"],
39 [56, "RemoveRedundantProjectionIndexes"],39 [56, "RemoveRedundantProjectionIndexes"],
40 [57, "ScheduledTaskWebhooks"],
40 ]);41 ]);
41 assert.deepStrictEqual(yield* runMigrations(), []);42 assert.deepStrictEqual(yield* runMigrations(), []);
42 assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);43 assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
...@@ -116,6 +117,7 @@ describe("V2 preview upgrade", () => {...@@ -116,6 +117,7 @@ describe("V2 preview upgrade", () => {
116 [53, "PullRequestFilesViewed"],117 [53, "PullRequestFilesViewed"],
117 [54, "ProjectionThreadsAutoSettleDisabledAt"],118 [54, "ProjectionThreadsAutoSettleDisabledAt"],
118 [56, "RemoveRedundantProjectionIndexes"],119 [56, "RemoveRedundantProjectionIndexes"],
120 [57, "ScheduledTaskWebhooks"],
119 ]);121 ]);
120 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),122 }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
121 );123 );
apps/server/src/provider/T3OrchestrationInstructions.ts+1-1
...@@ -9,7 +9,7 @@ The \`t3-code\` MCP server provides app-owned orchestration. Treat these concept...@@ -9,7 +9,7 @@ The \`t3-code\` MCP server provides app-owned orchestration. Treat these concept
9- A delegated task/subagent is child work owned by the current thread. Use \`orchestrator_capabilities\` to discover the current provider/model IDs from the same live catalog as the composer, including configured custom models. Do not treat a native tool's model list as the full list of available subagent models. Prefer native subagent tools for same-provider work only when they support the chosen model. Use \`delegate_task\` with that provider instance and model when native tools cannot, including for same-provider work. Also use \`delegate_task\` for cross-provider or explicitly T3-owned child tasks. Retain each returned \`taskId\`, and use \`task_status\` or \`task_cancel\` to manage it. The returned \`childThreadId\` is backing storage for the subagent, not the target for starting another delegated review round.9- A delegated task/subagent is child work owned by the current thread. Use \`orchestrator_capabilities\` to discover the current provider/model IDs from the same live catalog as the composer, including configured custom models. Do not treat a native tool's model list as the full list of available subagent models. Prefer native subagent tools for same-provider work only when they support the chosen model. Use \`delegate_task\` with that provider instance and model when native tools cannot, including for same-provider work. Also use \`delegate_task\` for cross-provider or explicitly T3-owned child tasks. Retain each returned \`taskId\`, and use \`task_status\` or \`task_cancel\` to manage it. The returned \`childThreadId\` is backing storage for the subagent, not the target for starting another delegated review round.
10- \`t3_thread_launch\` and \`create_threads\` create ordinary top-level T3 conversations. Use them only when the user explicitly asks for separate/new/top-level threads or conversations. Never use them merely because the user said "subagent" or requested parallel delegated work.10- \`t3_thread_launch\` and \`create_threads\` create ordinary top-level T3 conversations. Use them only when the user explicitly asks for separate/new/top-level threads or conversations. Never use them merely because the user said "subagent" or requested parallel delegated work.
11- For every T3 delegated review round, call \`delegate_task\` again. Include the original brief, prior findings, responses, and unresolved objections in each new task prompt. Track each round by its own \`taskId\`. Use a distinct \`clientRequestId\` per round, stable across retries of that round. Do not use \`t3_thread_send\` on \`childThreadId\` to continue a delegated review.11- For every T3 delegated review round, call \`delegate_task\` again. Include the original brief, prior findings, responses, and unresolved objections in each new task prompt. Track each round by its own \`taskId\`. Use a distinct \`clientRequestId\` per round, stable across retries of that round. Do not use \`t3_thread_send\` on \`childThreadId\` to continue a delegated review.
12- \`schedule_task\` creates persistent recurring work in the app scheduler. Pass \`schedule\` as a structured object, never as JSON text: \`{"type":"interval","everyMs":3600000}\` for an interval, or \`{"type":"fixed_time","timeOfDay":"09:00","weekdays":[1,2,3,4,5]}\` for a wall-clock schedule. By default runs return to the current thread; set \`bindToCurrentThread=false\` only when the user wants a fresh thread for every run. After scheduling, report the returned cadence and next run time.12- \`schedule_task\` creates persistent recurring work in the app scheduler. Pass \`schedule\` as a structured object, never as JSON text: \`{"type":"interval","everyMs":3600000}\` for an interval, or \`{"type":"fixed_time","timeOfDay":"09:00","weekdays":[1,2,3,4,5]}\` for a wall-clock schedule, or \`{"type":"webhook"}\` to run on each request to the returned \`webhookUrl\` (the prompt may use \`{{body.path}}\`-style placeholders). By default runs return to the current thread; set \`bindToCurrentThread=false\` only when the user wants a fresh thread for every run. After scheduling, report the returned cadence and next run time.
1313
14### Choose the workspace before starting a new thread14### Choose the workspace before starting a new thread
1515
apps/server/src/scheduledTasks/Schedule.ts+4
...@@ -13,6 +13,7 @@ export function nextScheduledRunAt(...@@ -13,6 +13,7 @@ export function nextScheduledRunAt(
13 schedule: ScheduledTaskSchedule,13 schedule: ScheduledTaskSchedule,
14 from: DateTime.DateTime,14 from: DateTime.DateTime,
15): DateTime.DateTime | null {15): DateTime.DateTime | null {
16 if (schedule.type === "webhook") return null;
16 if (schedule.type === "interval") {17 if (schedule.type === "interval") {
17 // Persisted rows created before the one-minute floor remain readable, but18 // Persisted rows created before the one-minute floor remain readable, but
18 // they must not retain their old high-frequency execution rate.19 // they must not retain their old high-frequency execution rate.
...@@ -55,6 +56,8 @@ export function isSameSchedule(a: ScheduledTaskSchedule, b: ScheduledTaskSchedul...@@ -55,6 +56,8 @@ export function isSameSchedule(a: ScheduledTaskSchedule, b: ScheduledTaskSchedul
55 if (a.type === "interval") {56 if (a.type === "interval") {
56 return b.type === "interval" && a.everyMs === b.everyMs;57 return b.type === "interval" && a.everyMs === b.everyMs;
57 }58 }
59 // Webhook tasks never have a next run, whatever their signature settings.
60 if (a.type === "webhook") return b.type === "webhook";
58 if (b.type !== "fixed_time") return false;61 if (b.type !== "fixed_time") return false;
59 // The contract accepts padded and unpadded hours ("9:00" and "09:00"), so62 // The contract accepts padded and unpadded hours ("9:00" and "09:00"), so
60 // compare the parsed time — string equality would treat a format-only edit63 // compare the parsed time — string equality would treat a format-only edit
...@@ -93,6 +96,7 @@ export function isMissedFixedTimeRun(...@@ -93,6 +96,7 @@ export function isMissedFixedTimeRun(
93}96}
9497
95function describeSchedule(schedule: ScheduledTaskSchedule): string {98function describeSchedule(schedule: ScheduledTaskSchedule): string {
99 if (schedule.type === "webhook") return "On webhook";
96 if (schedule.type === "interval") {100 if (schedule.type === "interval") {
97 const minutes = schedule.everyMs / MINUTE_MS;101 const minutes = schedule.everyMs / MINUTE_MS;
98 if (Number.isInteger(minutes)) {102 if (Number.isInteger(minutes)) {
apps/server/src/scheduledTasks/ScheduledTaskService.ts+634-30
...@@ -5,8 +5,15 @@ import {...@@ -5,8 +5,15 @@ import {
5 ScheduledTaskError,5 ScheduledTaskError,
6 ScheduledTaskId,6 ScheduledTaskId,
7 ThreadId,7 ThreadId,
8 ScheduledTaskWebhookDeliveryId,
8 type ScheduledTaskDeleteInput,9 type ScheduledTaskDeleteInput,
9 type ScheduledTaskDeleteResult,10 type ScheduledTaskDeleteResult,
11 type ScheduledTaskGetWebhookDeliveryInput,
12 type ScheduledTaskGetWebhookDeliveryResult,
13 type ScheduledTaskListWebhookDeliveriesInput,
14 type ScheduledTaskListWebhookDeliveriesResult,
15 type ScheduledTaskRotateWebhookTokenInput,
16 type ScheduledTaskWebhookDeliveryOutcome,
10 type ScheduledTaskListResult,17 type ScheduledTaskListResult,
11 type ScheduledTaskMutationResult,18 type ScheduledTaskMutationResult,
12 type ScheduledTaskRunNowInput,19 type ScheduledTaskRunNowInput,
...@@ -17,6 +24,7 @@ import {...@@ -17,6 +24,7 @@ import {
17import * as Cause from "effect/Cause";24import * as Cause from "effect/Cause";
18import * as Context from "effect/Context";25import * as Context from "effect/Context";
19import * as Crypto from "effect/Crypto";26import * as Crypto from "effect/Crypto";
27import * as Data from "effect/Data";
20import * as DateTime from "effect/DateTime";28import * as DateTime from "effect/DateTime";
21import * as Effect from "effect/Effect";29import * as Effect from "effect/Effect";
22import * as Layer from "effect/Layer";30import * as Layer from "effect/Layer";
...@@ -25,6 +33,7 @@ import * as PubSub from "effect/PubSub";...@@ -25,6 +33,7 @@ import * as PubSub from "effect/PubSub";
25import * as Ref from "effect/Ref";33import * as Ref from "effect/Ref";
26import * as Result from "effect/Result";34import * as Result from "effect/Result";
27import * as Schema from "effect/Schema";35import * as Schema from "effect/Schema";
36import * as Semaphore from "effect/Semaphore";
28import * as Stream from "effect/Stream";37import * as Stream from "effect/Stream";
29import * as SqlClient from "effect/sql/SqlClient";38import * as SqlClient from "effect/sql/SqlClient";
3039
...@@ -32,6 +41,57 @@ import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts...@@ -32,6 +41,57 @@ import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts
32import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";41import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
33import * as Scheduler from "../scheduling/Scheduler.ts";42import * as Scheduler from "../scheduling/Scheduler.ts";
34import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts";43import { isMissedFixedTimeRun, isSameSchedule, nextScheduledRunAt } from "./Schedule.ts";
44import { renderWebhookPrompt, type WebhookRequest } from "./webhookTemplate.ts";
45import { constantTimeEquals, verifyWebhookSignature } from "./webhookVerification.ts";
46
47/** Path prefix of the environment route that receives webhook requests. */
48export const WEBHOOK_ROUTE_PREFIX = "/api/hooks";
49/** Deliveries kept per task; older ones are pruned on insert. */
50const WEBHOOK_DELIVERY_RETENTION = 50;
51/** Body text kept in the delivery log. Larger bodies are cut and flagged. */
52const WEBHOOK_DELIVERY_LOG_BODY_LIMIT = 64 * 1024;
53/** Deliveries one task may hold at once, running or waiting their turn. */
54const WEBHOOK_MAX_QUEUED_PER_TASK = 20;
55/** Accepted deliveries per task per minute, enforced here as well as on the relay because the tunnel hostname is public too. */
56const WEBHOOK_RATE_LIMIT_PER_MINUTE = 60;
57
58/** Where a webhook task's public URL points. `relayUrl` is null when the environment is not linked to T3 Connect. */
59interface WebhookOrigin {
60 readonly environmentId: string;
61 readonly relayUrl: string | null;
62}
63
64export class ScheduledTaskWebhookOrigin extends Context.Reference<Effect.Effect<WebhookOrigin>>(
65 "t3/scheduledTasks/ScheduledTaskWebhookOrigin",
66 {
67 defaultValue: () => Effect.succeed({ environmentId: "local", relayUrl: null }),
68 },
69) {}
70
71/** A queued webhook delivery that no longer applies to its task; `reason` is shown in the delivery log. */
72class WebhookDeliverySkipped extends Data.TaggedError("WebhookDeliverySkipped")<{
73 readonly reason: string;
74}> {}
75
76interface RateWindow {
77 readonly accepted: ReadonlyArray<number>;
78 /** Whether a rejection was already logged in this window. */
79 readonly rejectedLogged: boolean;
80}
81
82export interface WebhookTriggerRequest extends WebhookRequest {
83 readonly hookId: string;
84 readonly token: string;
85 readonly body: Uint8Array;
86}
87
88/** What the HTTP route should answer. `not_found` covers unknown hooks and wrong tokens alike. */
89export type WebhookTriggerResult =
90 | { readonly _tag: "accepted"; readonly deliveryId: ScheduledTaskWebhookDeliveryId }
91 | { readonly _tag: "not_found" }
92 | { readonly _tag: "rejected_signature" }
93 | { readonly _tag: "disabled" }
94 | { readonly _tag: "rate_limited" };
3595
36const decodeTask = Schema.decodeUnknownEffect(ScheduledTask);96const decodeTask = Schema.decodeUnknownEffect(ScheduledTask);
37const decodeTaskId = Schema.decodeUnknownOption(ScheduledTaskId);97const decodeTaskId = Schema.decodeUnknownOption(ScheduledTaskId);
...@@ -44,6 +104,12 @@ const decodeWorkspaceStrategyJson = Schema.decodeUnknownEffect(...@@ -44,6 +104,12 @@ const decodeWorkspaceStrategyJson = Schema.decodeUnknownEffect(
44const decodeModelSelectionJson = Schema.decodeUnknownEffect(104const decodeModelSelectionJson = Schema.decodeUnknownEffect(
45 Schema.fromJsonString(ScheduledTask.fields.modelSelection),105 Schema.fromJsonString(ScheduledTask.fields.modelSelection),
46);106);
107const HeadersJson = Schema.fromJsonString(Schema.Record(Schema.String, Schema.String));
108const MissingFieldsJson = Schema.fromJsonString(Schema.Array(Schema.String));
109const decodeHeadersJson = Schema.decodeUnknownOption(HeadersJson);
110const decodeMissingFieldsJson = Schema.decodeUnknownOption(MissingFieldsJson);
111const encodeHeadersJson = Schema.encodeSync(HeadersJson);
112const encodeMissingFieldsJson = Schema.encodeSync(MissingFieldsJson);
47113
48interface ScheduledTaskRow {114interface ScheduledTaskRow {
49 readonly task_id: string;115 readonly task_id: string;
...@@ -66,6 +132,25 @@ interface ScheduledTaskRow {...@@ -66,6 +132,25 @@ interface ScheduledTaskRow {
66 readonly last_run_status: string;132 readonly last_run_status: string;
67 readonly last_run_error: string | null;133 readonly last_run_error: string | null;
68 readonly run_count: number;134 readonly run_count: number;
135 readonly webhook_token: string | null;
136 readonly webhook_secret: string | null;
137}
138
139interface WebhookDeliveryRow {
140 readonly delivery_id: string;
141 readonly task_id: string;
142 readonly received_at: string;
143 readonly method: string;
144 readonly query: string;
145 readonly headers_json: string;
146 readonly body: string;
147 readonly body_bytes: number;
148 readonly body_truncated: number;
149 readonly outcome: string;
150 readonly signature_verified: number;
151 readonly missing_fields_json: string;
152 readonly rendered_prompt: string | null;
153 readonly error: string | null;
69}154}
70155
71export class ScheduledTaskService extends Context.Service<156export class ScheduledTaskService extends Context.Service<
...@@ -87,6 +172,24 @@ export class ScheduledTaskService extends Context.Service<...@@ -87,6 +172,24 @@ export class ScheduledTaskService extends Context.Service<
87 readonly runNow: (172 readonly runNow: (
88 input: ScheduledTaskRunNowInput,173 input: ScheduledTaskRunNowInput,
89 ) => Effect.Effect<ScheduledTaskRunNowResult, ScheduledTaskError>;174 ) => Effect.Effect<ScheduledTaskRunNowResult, ScheduledTaskError>;
175 /** Issues a new URL token for a webhook task; the old URL stops working at once. */
176 readonly rotateWebhookToken: (
177 input: ScheduledTaskRotateWebhookTokenInput,
178 ) => Effect.Effect<ScheduledTaskMutationResult, ScheduledTaskError>;
179 readonly listWebhookDeliveries: (
180 input: ScheduledTaskListWebhookDeliveriesInput,
181 ) => Effect.Effect<ScheduledTaskListWebhookDeliveriesResult, ScheduledTaskError>;
182 readonly getWebhookDelivery: (
183 input: ScheduledTaskGetWebhookDeliveryInput,
184 ) => Effect.Effect<ScheduledTaskGetWebhookDeliveryResult, ScheduledTaskError>;
185 /**
186 * Verifies, logs and dispatches one webhook request. Returns as soon as the
187 * delivery is logged; the run itself continues in the background so senders
188 * with short timeouts get their answer immediately.
189 */
190 readonly triggerWebhook: (
191 request: WebhookTriggerRequest,
192 ) => Effect.Effect<WebhookTriggerResult, ScheduledTaskError>;
90 }193 }
91>()("t3/scheduledTasks/ScheduledTaskService") {}194>()("t3/scheduledTasks/ScheduledTaskService") {}
92195
...@@ -125,9 +228,43 @@ function errorMessage(error: unknown): string {...@@ -125,9 +228,43 @@ function errorMessage(error: unknown): string {
125 return String(error);228 return String(error);
126}229}
127230
128const decodeRow = (row: ScheduledTaskRow) =>231/** Headers kept out of the delivery log because they commonly carry credentials. */
232const REDACTED_HEADER =
233 /^(authorization|proxy-authorization|cookie|set-cookie)$|token|secret|signature|key|password|auth/i;
234
235function redactHeaders(headers: Readonly<Record<string, string>>): Record<string, string> {
236 return Object.fromEntries(
237 Object.entries(headers).map(([name, value]) => [
238 name,
239 REDACTED_HEADER.test(name) ? "[redacted]" : value,
240 ]),
241 );
242}
243
244function webhookPath(taskId: string, token: string): string {
245 return `${WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(taskId)}/${token}`;
246}
247
248function webhookEndpoint(
249 row: Pick<ScheduledTaskRow, "task_id" | "webhook_token" | "webhook_secret">,
250 origin: WebhookOrigin | null,
251): ScheduledTask["webhook"] {
252 if (row.webhook_token === null) return undefined;
253 const relayUrl = origin?.relayUrl?.replace(/\/+$/, "") ?? null;
254 return {
255 path: webhookPath(row.task_id, row.webhook_token),
256 url:
257 relayUrl === null || origin === null
258 ? null
259 : `${relayUrl}/v1/hooks/${encodeURIComponent(origin.environmentId)}/${encodeURIComponent(row.task_id)}/${row.webhook_token}`,
260 hasSecret: row.webhook_secret !== null,
261 };
262}
263
264const decodeRow = (row: ScheduledTaskRow, origin: WebhookOrigin | null = null) =>
129 Effect.gen(function* () {265 Effect.gen(function* () {
130 const schedule = yield* decodeScheduleJson(row.schedule_json);266 const schedule = yield* decodeScheduleJson(row.schedule_json);
267 const webhook = schedule.type === "webhook" ? webhookEndpoint(row, origin) : undefined;
131 const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json);268 const workspaceStrategy = yield* decodeWorkspaceStrategyJson(row.workspace_strategy_json);
132 const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json);269 const modelSelection = yield* decodeModelSelectionJson(row.model_selection_json);
133 return yield* decodeTask({270 return yield* decodeTask({
...@@ -153,6 +290,7 @@ const decodeRow = (row: ScheduledTaskRow) =>...@@ -153,6 +290,7 @@ const decodeRow = (row: ScheduledTaskRow) =>
153 lastRunStatus: row.last_run_status,290 lastRunStatus: row.last_run_status,
154 lastRunError: row.last_run_error,291 lastRunError: row.last_run_error,
155 runCount: row.run_count,292 runCount: row.run_count,
293 ...(webhook === undefined ? {} : { webhook }),
156 });294 });
157 }).pipe(295 }).pipe(
158 Effect.mapError((cause) => {296 Effect.mapError((cause) => {
...@@ -210,6 +348,16 @@ export const layer = Layer.effect(...@@ -210,6 +348,16 @@ export const layer = Layer.effect(
210 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;348 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;
211 const threadManagement = yield* ThreadManagementService.ThreadManagementService;349 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
212 const scheduler = yield* Scheduler.Scheduler;350 const scheduler = yield* Scheduler.Scheduler;
351 const readWebhookOrigin = yield* ScheduledTaskWebhookOrigin;
352 // Webhook deliveries for one task dispatch in arrival order rather than
353 // being dropped while an earlier delivery is still dispatching.
354 const webhookPermits = yield* Ref.make<ReadonlyMap<ScheduledTaskId, Semaphore.Semaphore>>(
355 new Map(),
356 );
357 // Keyed by task id and creation time, so deliveries of a deleted task that
358 // finish late release their own count, never a recreated task's.
359 const webhookQueued = yield* Ref.make<ReadonlyMap<string, number>>(new Map());
360 const webhookRateWindows = yield* Ref.make<ReadonlyMap<ScheduledTaskId, RateWindow>>(new Map());
213 const activeRuns = yield* Ref.make<ReadonlySet<ScheduledTaskId>>(new Set());361 const activeRuns = yield* Ref.make<ReadonlySet<ScheduledTaskId>>(new Set());
214 // Sliding(1) coalesces the dirty-signal: every notification triggers a362 // Sliding(1) coalesces the dirty-signal: every notification triggers a
215 // full list() re-emit anyway, so a slow subscriber only ever needs the363 // full list() re-emit anyway, so a slow subscriber only ever needs the
...@@ -238,7 +386,9 @@ export const layer = Layer.effect(...@@ -238,7 +386,9 @@ export const layer = Layer.effect(
238 last_run_at,386 last_run_at,
239 last_run_status,387 last_run_status,
240 last_run_error,388 last_run_error,
241 run_count389 run_count,
390 webhook_token,
391 webhook_secret
242 FROM scheduled_tasks392 FROM scheduled_tasks
243 ORDER BY updated_at DESC, task_id ASC393 ORDER BY updated_at DESC, task_id ASC
244 `;394 `;
...@@ -246,7 +396,8 @@ export const layer = Layer.effect(...@@ -246,7 +396,8 @@ export const layer = Layer.effect(
246 // Strict decode for the API surface: a corrupt row is a visible error.396 // Strict decode for the API surface: a corrupt row is a visible error.
247 const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () {397 const listRows = Effect.fn("ScheduledTaskService.listRows")(function* () {
248 const rows = yield* selectAllRows();398 const rows = yield* selectAllRows();
249 return yield* Effect.forEach(rows, decodeRow, { concurrency: 1 });399 const origin = yield* readWebhookOrigin;
400 return yield* Effect.forEach(rows, (row) => decodeRow(row, origin), { concurrency: 1 });
250 });401 });
251402
252 const getRows = (id: ScheduledTaskId) => sql<ScheduledTaskRow>`403 const getRows = (id: ScheduledTaskId) => sql<ScheduledTaskRow>`
...@@ -270,7 +421,9 @@ export const layer = Layer.effect(...@@ -270,7 +421,9 @@ export const layer = Layer.effect(
270 last_run_at,421 last_run_at,
271 last_run_status,422 last_run_status,
272 last_run_error,423 last_run_error,
273 run_count424 run_count,
425 webhook_token,
426 webhook_secret
274 FROM scheduled_tasks427 FROM scheduled_tasks
275 WHERE task_id = ${id}428 WHERE task_id = ${id}
276 `;429 `;
...@@ -284,9 +437,26 @@ export const layer = Layer.effect(...@@ -284,9 +437,26 @@ export const layer = Layer.effect(
284 );437 );
285 const row = rows[0];438 const row = rows[0];
286 if (row === undefined) return null;439 if (row === undefined) return null;
287 return yield* decodeRow(row);440 return yield* decodeRow(row, yield* readWebhookOrigin);
288 });441 });
289442
443 const findWebhookCredentials = (id: ScheduledTaskId) =>
444 getRows(id).pipe(
445 Effect.map((rows) =>
446 rows[0] === undefined
447 ? null
448 : { token: rows[0].webhook_token, secret: rows[0].webhook_secret },
449 ),
450 Effect.mapError((cause) =>
451 taskError("Could not load schedule task.", { taskId: id, cause }),
452 ),
453 );
454
455 const newWebhookToken = crypto.randomBytes(32).pipe(
456 Effect.map((bytes) => Buffer.from(bytes).toString("base64url")),
457 Effect.mapError((cause) => taskError("Could not generate webhook token.", { cause })),
458 );
459
290 const loadTask = Effect.fn("ScheduledTaskService.loadTask")(function* (id: ScheduledTaskId) {460 const loadTask = Effect.fn("ScheduledTaskService.loadTask")(function* (id: ScheduledTaskId) {
291 const task = yield* findTask(id);461 const task = yield* findTask(id);
292 if (task === null) {462 if (task === null) {
...@@ -300,7 +470,16 @@ export const layer = Layer.effect(...@@ -300,7 +470,16 @@ export const layer = Layer.effect(
300 // concurrent settings save must not overwrite an in-flight increment.470 // concurrent settings save must not overwrite an in-flight increment.
301 // Check existence in the write itself so an edit cannot undo a deletion471 // Check existence in the write itself so an edit cannot undo a deletion
302 // that landed after upsert loaded the previous task.472 // that landed after upsert loaded the previous task.
303 const saveTask = (task: ScheduledTask, requireExisting: boolean) =>473 const saveTask = (
474 task: ScheduledTask,
475 requireExisting: boolean,
476 webhook: {
477 readonly token: string | null;
478 readonly secret: string | null;
479 /** False when the save carried no new secret, so a concurrent change survives. */
480 readonly secretChanged: boolean;
481 },
482 ) =>
304 sql<{ task_id: string }>`483 sql<{ task_id: string }>`
305 INSERT INTO scheduled_tasks (484 INSERT INTO scheduled_tasks (
306 task_id,485 task_id,
...@@ -322,7 +501,9 @@ export const layer = Layer.effect(...@@ -322,7 +501,9 @@ export const layer = Layer.effect(
322 last_run_at,501 last_run_at,
323 last_run_status,502 last_run_status,
324 last_run_error,503 last_run_error,
325 run_count504 run_count,
505 webhook_token,
506 webhook_secret
326 )507 )
327 SELECT508 SELECT
328 ${task.id},509 ${task.id},
...@@ -344,7 +525,9 @@ export const layer = Layer.effect(...@@ -344,7 +525,9 @@ export const layer = Layer.effect(
344 ${task.lastRunAt},525 ${task.lastRunAt},
345 ${task.lastRunStatus},526 ${task.lastRunStatus},
346 ${task.lastRunError},527 ${task.lastRunError},
347 ${task.runCount}528 ${task.runCount},
529 ${webhook.token},
530 ${webhook.secret}
348 WHERE ${requireExisting ? 0 : 1} = 1531 WHERE ${requireExisting ? 0 : 1} = 1
349 OR EXISTS (SELECT 1 FROM scheduled_tasks WHERE task_id = ${task.id})532 OR EXISTS (SELECT 1 FROM scheduled_tasks WHERE task_id = ${task.id})
350 ON CONFLICT (task_id)533 ON CONFLICT (task_id)
...@@ -361,7 +544,17 @@ export const layer = Layer.effect(...@@ -361,7 +544,17 @@ export const layer = Layer.effect(
361 interaction_mode = excluded.interaction_mode,544 interaction_mode = excluded.interaction_mode,
362 creation_source = excluded.creation_source,545 creation_source = excluded.creation_source,
363 updated_at = excluded.updated_at,546 updated_at = excluded.updated_at,
364 next_run_at = excluded.next_run_at547 next_run_at = excluded.next_run_at,
548 -- Only rotate changes a live token, so a save racing a rotation
549 -- cannot bring the old URL back.
550 webhook_token = CASE
551 WHEN excluded.webhook_token IS NULL THEN NULL
552 ELSE COALESCE(scheduled_tasks.webhook_token, excluded.webhook_token)
553 END,
554 webhook_secret = CASE
555 WHEN ${webhook.secretChanged ? 1 : 0} = 1 THEN excluded.webhook_secret
556 ELSE scheduled_tasks.webhook_secret
557 END
365 RETURNING task_id558 RETURNING task_id
366 `.pipe(559 `.pipe(
367 Effect.mapError((cause) =>560 Effect.mapError((cause) =>
...@@ -375,27 +568,38 @@ export const layer = Layer.effect(...@@ -375,27 +568,38 @@ export const layer = Layer.effect(
375 );568 );
376569
377 const deleteRow = (id: ScheduledTaskId) =>570 const deleteRow = (id: ScheduledTaskId) =>
378 sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`.pipe(571 sql
379 Effect.mapError((cause) =>572 .withTransaction(
380 taskError("Could not delete schedule task.", { taskId: id, cause }),573 sql`DELETE FROM scheduled_task_webhook_deliveries WHERE task_id = ${id}`.pipe(
381 ),574 Effect.andThen(sql`DELETE FROM scheduled_tasks WHERE task_id = ${id}`),
382 );575 ),
576 )
577 .pipe(
578 Effect.mapError((cause) =>
579 taskError("Could not delete schedule task.", { taskId: id, cause }),
580 ),
581 );
383582
384 // Run-state transitions use targeted UPDATEs (never the full-row upsert) so583 // Run-state transitions use targeted UPDATEs (never the full-row upsert) so
385 // a completing run cannot resurrect a deleted task or clobber concurrent584 // a completing run cannot resurrect a deleted task or clobber concurrent
386 // edits to the task definition.585 // edits to the task definition.
387 const markRunning = (id: ScheduledTaskId, startedAtIso: string) =>586 const markRunning = (id: ScheduledTaskId, startedAtIso: string) =>
388 sql`587 sql<{ task_id: string }>`
389 UPDATE scheduled_tasks588 UPDATE scheduled_tasks
390 SET updated_at = ${startedAtIso},589 SET updated_at = ${startedAtIso},
391 last_run_at = ${startedAtIso},590 last_run_at = ${startedAtIso},
392 last_run_status = 'running',591 last_run_status = 'running',
393 last_run_error = NULL592 last_run_error = NULL
394 WHERE task_id = ${id}593 WHERE task_id = ${id}
594 RETURNING task_id
395 `.pipe(595 `.pipe(
396 Effect.mapError((cause) =>596 Effect.mapError((cause) =>
397 taskError("Could not mark schedule task as running.", { taskId: id, cause }),597 taskError("Could not mark schedule task as running.", { taskId: id, cause }),
398 ),598 ),
599 // A task deleted after the re-read must not be dispatched from the stale snapshot.
600 Effect.flatMap((rows) =>
601 rows.length > 0 ? Effect.void : taskError("Schedule task not found.", { taskId: id }),
602 ),
399 );603 );
400604
401 const markCompleted = (input: {605 const markCompleted = (input: {
...@@ -457,7 +661,8 @@ export const layer = Layer.effect(...@@ -457,7 +661,8 @@ export const layer = Layer.effect(
457661
458 const runTask = Effect.fn("ScheduledTaskService.runTask")(function* (662 const runTask = Effect.fn("ScheduledTaskService.runTask")(function* (
459 task: ScheduledTask,663 task: ScheduledTask,
460 trigger: "scheduled" | "manual",664 trigger: "scheduled" | "manual" | "webhook",
665 webhook?: { readonly deliveryId: string; readonly prompt: string },
461 ) {666 ) {
462 const reserved = yield* Ref.modify(activeRuns, (active) => {667 const reserved = yield* Ref.modify(activeRuns, (active) => {
463 if (active.has(task.id)) return [false, active] as const;668 if (active.has(task.id)) return [false, active] as const;
...@@ -466,7 +671,7 @@ export const layer = Layer.effect(...@@ -466,7 +671,7 @@ export const layer = Layer.effect(
466 return [true, next] as const;671 return [true, next] as const;
467 });672 });
468 if (!reserved) {673 if (!reserved) {
469 if (trigger === "manual") {674 if (trigger !== "scheduled") {
470 return yield* taskError("Schedule task is already running.", { taskId: task.id });675 return yield* taskError("Schedule task is already running.", { taskId: task.id });
471 }676 }
472 return task;677 return task;
...@@ -481,9 +686,14 @@ export const layer = Layer.effect(...@@ -481,9 +686,14 @@ export const layer = Layer.effect(
481 // the poll loaded it — none of those may fire.686 // the poll loaded it — none of those may fire.
482 const active = yield* findTask(task.id);687 const active = yield* findTask(task.id);
483 if (active === null) {688 if (active === null) {
689 if (webhook !== undefined) {
690 return yield* new WebhookDeliverySkipped({
691 reason: "The task was deleted before this delivery ran.",
692 });
693 }
484 // A manual run on a just-deleted task must fail loudly, not report694 // A manual run on a just-deleted task must fail loudly, not report
485 // a successful run that never dispatched.695 // a successful run that never dispatched.
486 if (trigger === "manual") {696 if (trigger !== "scheduled") {
487 return yield* taskError("Schedule task not found.", { taskId: task.id });697 return yield* taskError("Schedule task not found.", { taskId: task.id });
488 }698 }
489 return task;699 return task;
...@@ -500,16 +710,36 @@ export const layer = Layer.effect(...@@ -500,16 +710,36 @@ export const layer = Layer.effect(
500 ) {710 ) {
501 return active;711 return active;
502 }712 }
713 // A queued delivery must not run a task that was paused, deleted and
714 // recreated under the same id, or switched to another trigger while
715 // it waited for its turn.
716 if (webhook !== undefined) {
717 const reason =
718 active.createdAt !== task.createdAt
719 ? "The task was replaced before this delivery ran."
720 : active.schedule.type !== "webhook"
721 ? "The task's trigger changed before this delivery ran."
722 : !active.enabled
723 ? "The task was paused before this delivery ran."
724 : null;
725 if (reason !== null) return yield* new WebhookDeliverySkipped({ reason });
726 }
503727
504 yield* markRunning(active.id, startedAtIso);728 yield* markRunning(active.id, startedAtIso);
505 yield* notifyChanged;729 yield* notifyChanged;
506730
507 const fireKey = `${active.id}:${DateTime.toEpochMillis(startedAt)}:${trigger}`;731 // A webhook run is keyed by its delivery so the same delivery can
732 // never dispatch twice.
733 const fireKey =
734 webhook === undefined
735 ? `${active.id}:${DateTime.toEpochMillis(startedAt)}:${trigger}`
736 : `${active.id}:webhook:${webhook.deliveryId}`;
508 const commandId = CommandId.make(`scheduled-task:${fireKey}`);737 const commandId = CommandId.make(`scheduled-task:${fireKey}`);
509 const messageId = MessageId.make(`scheduled-task-message:${fireKey}`);738 const messageId = MessageId.make(`scheduled-task-message:${fireKey}`);
510 // Dispatch from the fresh row so prompt/model/binding edits made739 // Dispatch from the fresh row so prompt/model/binding edits made
511 // after the poll read are honoured.740 // after the poll read are honoured. A webhook prompt was rendered
512 const prompt = active.prompt;741 // from the row when the request arrived.
742 const prompt = webhook?.prompt ?? active.prompt;
513743
514 // Effect.exit (not Effect.result) so defects and interruptions in the744 // Effect.exit (not Effect.result) so defects and interruptions in the
515 // dispatch are also captured and recorded as a failed run instead of745 // dispatch are also captured and recorded as a failed run instead of
...@@ -632,9 +862,10 @@ export const layer = Layer.effect(...@@ -632,9 +862,10 @@ export const layer = Layer.effect(
632 yield* Effect.forEach(862 yield* Effect.forEach(
633 due,863 due,
634 ({ task, dueAt }) =>864 ({ task, dueAt }) =>
635 (isMissedFixedTimeRun(task.schedule, dueAt, now)865 Effect.suspend(() =>
636 ? rescheduleMissedRun(task, now)866 isMissedFixedTimeRun(task.schedule, dueAt, now)
637 : runTask(task, "scheduled")867 ? rescheduleMissedRun(task, now)
868 : runTask(task, "scheduled"),
638 ).pipe(869 ).pipe(
639 Effect.catch((cause) =>870 Effect.catch((cause) =>
640 Effect.logWarning("Scheduled task run failed", { taskId: task.id, cause }),871 Effect.logWarning("Scheduled task run failed", { taskId: task.id, cause }),
...@@ -742,16 +973,50 @@ export const layer = Layer.effect(...@@ -742,16 +973,50 @@ export const layer = Layer.effect(
742 // Keep the existing next_run_at when the schedule itself is untouched:973 // Keep the existing next_run_at when the schedule itself is untouched:
743 // editing a title or prompt must not postpone (or resurrect) a due974 // editing a title or prompt must not postpone (or resurrect) a due
744 // run — only schedule/enabled changes restart the clock.975 // run — only schedule/enabled changes restart the clock.
976 const schedule: ScheduledTask["schedule"] =
977 input.schedule.type === "webhook"
978 ? {
979 type: "webhook",
980 signature:
981 input.schedule.signature == null
982 ? null
983 : {
984 header: input.schedule.signature.header.toLowerCase(),
985 encoding: input.schedule.signature.encoding,
986 prefix: input.schedule.signature.prefix,
987 },
988 }
989 : input.schedule;
990 const webhook =
991 input.schedule.type === "webhook"
992 ? yield* Effect.gen(function* () {
993 const existing = existingTask === null ? null : yield* findWebhookCredentials(id);
994 // Saving a webhook task keeps its URL; only rotate changes it.
995 const token = existing?.token ?? (yield* newWebhookToken);
996 const signature =
997 input.schedule.type === "webhook" ? input.schedule.signature : null;
998 const secret =
999 signature == null ? null : (signature.secret ?? existing?.secret ?? null);
1000 const secretChanged =
1001 signature == null || signature.secret !== undefined || existing === null;
1002 if (signature != null && secret === null) {
1003 return yield* taskError("A webhook signature check needs a signing secret.", {
1004 taskId: id,
1005 });
1006 }
1007 return { token, secret, secretChanged };
1008 })
1009 : { token: null, secret: null, secretChanged: true };
745 const scheduleUnchanged =1010 const scheduleUnchanged =
746 existingTask !== null &&1011 existingTask !== null &&
747 existingTask.enabled === input.enabled &&1012 existingTask.enabled === input.enabled &&
748 isSameSchedule(existingTask.schedule, input.schedule);1013 isSameSchedule(existingTask.schedule, schedule);
749 const task: ScheduledTask = {1014 const task: ScheduledTask = {
750 id,1015 id,
751 title: input.title,1016 title: input.title,
752 prompt: input.prompt,1017 prompt: input.prompt,
753 enabled: input.enabled,1018 enabled: input.enabled,
754 schedule: input.schedule,1019 schedule,
755 projectId: input.projectId,1020 projectId: input.projectId,
756 threadId: input.threadId ?? null,1021 threadId: input.threadId ?? null,
757 workspaceStrategy: input.workspaceStrategy,1022 workspaceStrategy: input.workspaceStrategy,
...@@ -764,15 +1029,15 @@ export const layer = Layer.effect(...@@ -764,15 +1029,15 @@ export const layer = Layer.effect(
764 updatedAt: iso(now),1029 updatedAt: iso(now),
765 nextRunAt: scheduleUnchanged1030 nextRunAt: scheduleUnchanged
766 ? existingTask.nextRunAt1031 ? existingTask.nextRunAt
767 : nextRunAt({ enabled: input.enabled, schedule: input.schedule }, now),1032 : nextRunAt({ enabled: input.enabled, schedule }, now),
768 lastRunAt: existingTask?.lastRunAt ?? null,1033 lastRunAt: existingTask?.lastRunAt ?? null,
769 lastRunStatus: existingTask?.lastRunStatus ?? "never",1034 lastRunStatus: existingTask?.lastRunStatus ?? "never",
770 lastRunError: existingTask?.lastRunError ?? null,1035 lastRunError: existingTask?.lastRunError ?? null,
771 runCount: existingTask?.runCount ?? 0,1036 runCount: existingTask?.runCount ?? 0,
772 };1037 };
773 yield* saveTask(task, input.requireExisting === true);1038 yield* saveTask(task, input.requireExisting === true, webhook);
774 yield* notifyChanged;1039 yield* notifyChanged;
775 return { task };1040 return { task: (yield* findTask(id)) ?? task };
776 });1041 });
7771042
778 const setEnabled: ScheduledTaskService["Service"]["setEnabled"] = (input) =>1043 const setEnabled: ScheduledTaskService["Service"]["setEnabled"] = (input) =>
...@@ -805,11 +1070,34 @@ export const layer = Layer.effect(...@@ -805,11 +1070,34 @@ export const layer = Layer.effect(
805 });1070 });
8061071
807 const deleteTask: ScheduledTaskService["Service"]["delete"] = (input) =>1072 const deleteTask: ScheduledTaskService["Service"]["delete"] = (input) =>
808 deleteRow(input.id).pipe(Effect.andThen(notifyChanged), Effect.as({ id: input.id }));1073 deleteRow(input.id).pipe(
1074 Effect.andThen(
1075 Effect.all([
1076 Ref.update(webhookRateWindows, (windows) => {
1077 const next = new Map(windows);
1078 next.delete(input.id);
1079 return next;
1080 }),
1081 Ref.update(webhookPermits, (permits) => {
1082 const next = new Map(permits);
1083 next.delete(input.id);
1084 return next;
1085 }),
1086 ]),
1087 ),
1088 Effect.andThen(notifyChanged),
1089 Effect.as({ id: input.id }),
1090 );
8091091
810 const runNow: ScheduledTaskService["Service"]["runNow"] = (input: ScheduledTaskRunNowInput) =>1092 const runNow: ScheduledTaskService["Service"]["runNow"] = (input: ScheduledTaskRunNowInput) =>
811 Effect.gen(function* () {1093 Effect.gen(function* () {
812 const task = yield* loadTask(input.id);1094 const task = yield* loadTask(input.id);
1095 if (task.schedule.type === "webhook") {
1096 // There is no request to render the prompt from.
1097 return yield* taskError("Webhook tasks run when their URL receives a request.", {
1098 taskId: input.id,
1099 });
1100 }
813 const next = yield* runTask(task, "manual").pipe(1101 const next = yield* runTask(task, "manual").pipe(
814 Effect.mapError((cause) =>1102 Effect.mapError((cause) =>
815 taskError("Could not run schedule task.", { taskId: input.id, cause }),1103 taskError("Could not run schedule task.", { taskId: input.id, cause }),
...@@ -818,6 +1106,318 @@ export const layer = Layer.effect(...@@ -818,6 +1106,318 @@ export const layer = Layer.effect(
818 return { task: next };1106 return { task: next };
819 });1107 });
8201108
1109 const rotateWebhookToken: ScheduledTaskService["Service"]["rotateWebhookToken"] = (input) =>
1110 Effect.gen(function* () {
1111 const task = yield* loadTask(input.id);
1112 if (task.schedule.type !== "webhook") {
1113 return yield* taskError("Only webhook tasks have a URL token.", { taskId: input.id });
1114 }
1115 const token = yield* newWebhookToken;
1116 const now = yield* localNow;
1117 // Matching created_at keeps a rotation from landing on a task deleted
1118 // and recreated under the same id since it was loaded.
1119 const updated = yield* sql<{ task_id: string }>`
1120 UPDATE scheduled_tasks
1121 SET webhook_token = ${token}, updated_at = ${iso(now)}
1122 WHERE task_id = ${input.id} AND created_at = ${task.createdAt}
1123 RETURNING task_id
1124 `.pipe(
1125 Effect.mapError((cause) =>
1126 taskError("Could not rotate webhook token.", { taskId: input.id, cause }),
1127 ),
1128 );
1129 if (updated.length === 0) {
1130 return yield* taskError("Schedule task was deleted or replaced.", { taskId: input.id });
1131 }
1132 yield* notifyChanged;
1133 return { task: yield* loadTask(input.id) };
1134 });
1135
1136 const deliveryHeaders = (row: WebhookDeliveryRow): Readonly<Record<string, string>> =>
1137 Option.getOrElse(decodeHeadersJson(row.headers_json), () => ({}));
1138 const decodeDeliverySummary = (row: WebhookDeliveryRow) => ({
1139 id: ScheduledTaskWebhookDeliveryId.make(row.delivery_id),
1140 taskId: ScheduledTaskId.make(row.task_id),
1141 receivedAt: row.received_at,
1142 method: row.method,
1143 contentType: deliveryHeaders(row)["content-type"] ?? null,
1144 bodyBytes: row.body_bytes,
1145 outcome: row.outcome as ScheduledTaskWebhookDeliveryOutcome,
1146 signatureVerified: row.signature_verified === 1,
1147 missingFields: Option.getOrElse(decodeMissingFieldsJson(row.missing_fields_json), () => []),
1148 error: row.error,
1149 });
1150
1151 const listWebhookDeliveries: ScheduledTaskService["Service"]["listWebhookDeliveries"] = (
1152 input,
1153 ) =>
1154 sql<WebhookDeliveryRow>`
1155 SELECT * FROM scheduled_task_webhook_deliveries
1156 WHERE task_id = ${input.id}
1157 ORDER BY received_at DESC, rowid DESC
1158 `.pipe(
1159 Effect.map((rows) => ({ deliveries: rows.map(decodeDeliverySummary) })),
1160 Effect.mapError((cause) =>
1161 taskError("Could not list webhook deliveries.", { taskId: input.id, cause }),
1162 ),
1163 );
1164
1165 const getWebhookDelivery: ScheduledTaskService["Service"]["getWebhookDelivery"] = (input) =>
1166 Effect.gen(function* () {
1167 const rows = yield* sql<WebhookDeliveryRow>`
1168 SELECT * FROM scheduled_task_webhook_deliveries
1169 WHERE task_id = ${input.id} AND delivery_id = ${input.deliveryId}
1170 `.pipe(
1171 Effect.mapError((cause) =>
1172 taskError("Could not load webhook delivery.", { taskId: input.id, cause }),
1173 ),
1174 );
1175 const row = rows[0];
1176 if (row === undefined) {
1177 return yield* taskError("Webhook delivery not found.", { taskId: input.id });
1178 }
1179 return {
1180 delivery: {
1181 ...decodeDeliverySummary(row),
1182 query: row.query,
1183 headers: deliveryHeaders(row),
1184 body: row.body,
1185 bodyTruncated: row.body_truncated === 1,
1186 renderedPrompt: row.rendered_prompt,
1187 },
1188 };
1189 });
1190
1191 const recordDelivery = (input: {
1192 readonly id: string;
1193 readonly taskId: ScheduledTaskId;
1194 readonly receivedAt: string;
1195 readonly request: WebhookTriggerRequest;
1196 readonly outcome: ScheduledTaskWebhookDeliveryOutcome;
1197 readonly signatureVerified: boolean;
1198 readonly missing: ReadonlyArray<string>;
1199 readonly renderedPrompt: string | null;
1200 }) => {
1201 const truncated = input.request.body.byteLength > WEBHOOK_DELIVERY_LOG_BODY_LIMIT;
1202 const loggedBody = truncated
1203 ? new TextDecoder().decode(input.request.body.subarray(0, WEBHOOK_DELIVERY_LOG_BODY_LIMIT))
1204 : input.request.bodyText;
1205 return sql
1206 .withTransaction(
1207 Effect.gen(function* () {
1208 // Conditional on the task existing, so a delivery racing a delete
1209 // cannot leave rows that a recreated task with the same id would show.
1210 yield* sql`
1211 INSERT INTO scheduled_task_webhook_deliveries (
1212 delivery_id, task_id, received_at, method, query, headers_json, body,
1213 body_bytes, body_truncated, outcome, signature_verified,
1214 missing_fields_json, rendered_prompt, error
1215 )
1216 SELECT
1217 ${input.id}, ${input.taskId}, ${input.receivedAt}, ${input.request.method},
1218 ${input.request.query}, ${encodeHeadersJson(redactHeaders(input.request.headers))},
1219 ${loggedBody},
1220 ${input.request.body.byteLength}, ${truncated ? 1 : 0}, ${input.outcome},
1221 ${input.signatureVerified ? 1 : 0}, ${encodeMissingFieldsJson(input.missing)},
1222 ${input.renderedPrompt}, NULL
1223 WHERE EXISTS (SELECT 1 FROM scheduled_tasks WHERE task_id = ${input.taskId})
1224 `;
1225 yield* sql`
1226 DELETE FROM scheduled_task_webhook_deliveries
1227 WHERE task_id = ${input.taskId}
1228 AND delivery_id NOT IN (
1229 SELECT delivery_id FROM scheduled_task_webhook_deliveries
1230 WHERE task_id = ${input.taskId}
1231 -- rowid breaks timestamp ties in arrival order; delivery ids are random.
1232 ORDER BY received_at DESC, rowid DESC
1233 LIMIT ${WEBHOOK_DELIVERY_RETENTION}
1234 )
1235 `;
1236 }),
1237 )
1238 .pipe(
1239 Effect.mapError((cause) =>
1240 taskError("Could not record webhook delivery.", { taskId: input.taskId, cause }),
1241 ),
1242 );
1243 };
1244
1245 const markDeliveryFailed = (deliveryId: string, message: string) =>
1246 sql`
1247 UPDATE scheduled_task_webhook_deliveries
1248 SET outcome = 'dispatch_failed', error = ${message}
1249 WHERE delivery_id = ${deliveryId}
1250 `.pipe(Effect.ignore);
1251
1252 /**
1253 * Sliding one-minute window counting every request with a valid token,
1254 * including ones the signature check later rejects.
1255 */
1256 const takeRateSlot = (id: ScheduledTaskId, nowMs: number) =>
1257 Ref.modify(
1258 webhookRateWindows,
1259 (
1260 windows,
1261 ): readonly [
1262 "allowed" | "first_rejected" | "rejected",
1263 ReadonlyMap<ScheduledTaskId, RateWindow>,
1264 ] => {
1265 const current = windows.get(id);
1266 const recent = (current?.accepted ?? []).filter((at) => nowMs - at < 60_000);
1267 if (recent.length >= WEBHOOK_RATE_LIMIT_PER_MINUTE) {
1268 const first = current?.rejectedLogged !== true;
1269 return [
1270 first ? "first_rejected" : "rejected",
1271 new Map(windows).set(id, { accepted: recent, rejectedLogged: true }),
1272 ];
1273 }
1274 return [
1275 "allowed",
1276 new Map(windows).set(id, { accepted: [...recent, nowMs], rejectedLogged: false }),
1277 ];
1278 },
1279 );
1280
1281 const webhookPermit = (id: ScheduledTaskId) =>
1282 Effect.gen(function* () {
1283 const existing = (yield* Ref.get(webhookPermits)).get(id);
1284 if (existing !== undefined) return existing;
1285 const created = yield* Semaphore.make(1);
1286 return yield* Ref.modify(webhookPermits, (permits) => {
1287 const raced = permits.get(id);
1288 return raced === undefined
1289 ? [created, new Map(permits).set(id, created)]
1290 : [raced, permits];
1291 });
1292 });
1293
1294 // Detached from the request so the HTTP response does not wait for the
1295 // run; scoped to the service so shutdown interrupts it.
1296 const serviceScope = yield* Effect.scope;
1297
1298 const triggerWebhook: ScheduledTaskService["Service"]["triggerWebhook"] = (request) =>
1299 Effect.gen(function* () {
1300 const taskId = decodeTaskId(request.hookId);
1301 if (Option.isNone(taskId)) return { _tag: "not_found" as const };
1302 const rows = yield* getRows(taskId.value).pipe(
1303 Effect.mapError((cause) =>
1304 taskError("Could not load schedule task.", { taskId: taskId.value, cause }),
1305 ),
1306 );
1307 const row = rows[0];
1308 // A wrong token is indistinguishable from an unknown hook, so the URL
1309 // does not reveal which hooks exist.
1310 if (
1311 row === undefined ||
1312 row.webhook_token === null ||
1313 !constantTimeEquals(request.token, row.webhook_token)
1314 ) {
1315 return { _tag: "not_found" as const };
1316 }
1317 const task = yield* decodeRow(row);
1318 if (task.schedule.type !== "webhook") return { _tag: "not_found" as const };
1319
1320 const receivedAt = yield* localNow;
1321 const deliveryUuid = yield* crypto.randomUUIDv4.pipe(
1322 Effect.mapError((cause) => taskError("Could not generate delivery id.", { cause })),
1323 );
1324 const deliveryId = ScheduledTaskWebhookDeliveryId.make(`delivery:${deliveryUuid}`);
1325 const log = (
1326 outcome: ScheduledTaskWebhookDeliveryOutcome,
1327 details: {
1328 readonly signatureVerified?: boolean;
1329 readonly missing?: ReadonlyArray<string>;
1330 readonly renderedPrompt?: string;
1331 } = {},
1332 ) =>
1333 recordDelivery({
1334 id: deliveryId,
1335 taskId: task.id,
1336 receivedAt: iso(receivedAt),
1337 request,
1338 outcome,
1339 signatureVerified: details.signatureVerified ?? false,
1340 missing: details.missing ?? [],
1341 renderedPrompt: details.renderedPrompt ?? null,
1342 });
1343
1344 // Only the first rejected request in a window is logged, so a flood
1345 // cannot write rows or push the real deliveries out of the log.
1346 const slot = yield* takeRateSlot(task.id, DateTime.toEpochMillis(receivedAt));
1347 if (slot !== "allowed") {
1348 if (slot === "first_rejected") yield* log("rate_limited");
1349 return { _tag: "rate_limited" as const };
1350 }
1351 if (!task.enabled) {
1352 yield* log("disabled");
1353 return { _tag: "disabled" as const };
1354 }
1355 const signature = task.schedule.signature;
1356 if (signature !== null) {
1357 const verified =
1358 row.webhook_secret !== null &&
1359 verifyWebhookSignature({
1360 signature,
1361 secret: row.webhook_secret,
1362 headers: request.headers,
1363 body: request.body,
1364 });
1365 if (!verified) {
1366 yield* log("rejected_signature");
1367 return { _tag: "rejected_signature" as const };
1368 }
1369 }
1370
1371 const rendered = renderWebhookPrompt(task.prompt, request);
1372 // Bound the deliveries one task holds, so steady traffic to a stuck
1373 // task cannot pile up parked fibers. A refused request is not logged,
1374 // so it cannot push real deliveries out of the log.
1375 const queueKey = `${task.id}\u0000${task.createdAt}`;
1376 const queued = yield* Ref.modify(webhookQueued, (counts) => {
1377 const count = counts.get(queueKey) ?? 0;
1378 return count >= WEBHOOK_MAX_QUEUED_PER_TASK
1379 ? ([false, counts] as const)
1380 : ([true, new Map(counts).set(queueKey, count + 1)] as const);
1381 });
1382 if (!queued) return { _tag: "rate_limited" as const };
1383 // Entries leave the map when their count reaches zero, so a deleted
1384 // task's key does not linger once its last delivery finishes.
1385 const release = Ref.update(webhookQueued, (counts) => {
1386 const next = new Map(counts);
1387 const count = (next.get(queueKey) ?? 1) - 1;
1388 if (count <= 0) next.delete(queueKey);
1389 else next.set(queueKey, count);
1390 return next;
1391 });
1392 yield* log("accepted", {
1393 signatureVerified: signature !== null,
1394 missing: rendered.missing,
1395 renderedPrompt: rendered.prompt,
1396 }).pipe(Effect.onError(() => release));
1397 const permit = yield* webhookPermit(task.id);
1398 yield* runTask(task, "webhook", { deliveryId, prompt: rendered.prompt }).pipe(
1399 Effect.flatMap((completed) =>
1400 completed.lastRunStatus === "failed"
1401 ? markDeliveryFailed(deliveryId, "The run failed to start.")
1402 : Effect.void,
1403 ),
1404 Effect.catchTag("WebhookDeliverySkipped", (skipped) =>
1405 markDeliveryFailed(deliveryId, skipped.reason),
1406 ),
1407 // The log is readable over RPC, so it gets a fixed reason; the
1408 // cause, which can carry request data, stays in the server log.
1409 Effect.catchCause((cause) =>
1410 Effect.logWarning("Webhook dispatch failed", { taskId: task.id, cause }).pipe(
1411 Effect.andThen(markDeliveryFailed(deliveryId, "The run failed to start.")),
1412 ),
1413 ),
1414 permit.withPermits(1),
1415 Effect.ensuring(release),
1416 Effect.forkIn(serviceScope),
1417 );
1418 return { _tag: "accepted" as const, deliveryId };
1419 });
1420
821 return ScheduledTaskService.of({1421 return ScheduledTaskService.of({
822 list,1422 list,
823 subscribeList,1423 subscribeList,
...@@ -825,6 +1425,10 @@ export const layer = Layer.effect(...@@ -825,6 +1425,10 @@ export const layer = Layer.effect(
825 setEnabled,1425 setEnabled,
826 delete: deleteTask,1426 delete: deleteTask,
827 runNow,1427 runNow,
1428 rotateWebhookToken,
1429 listWebhookDeliveries,
1430 getWebhookDelivery,
1431 triggerWebhook,
828 });1432 });
829 }),1433 }),
830);1434);
apps/server/src/scheduledTasks/ScheduledTaskService.webhook.test.ts created+466
...@@ -0,0 +1,466 @@
1import * as NodeCrypto from "node:crypto";
2
3import * as NodePlatformCrypto from "@effect/platform-node/NodeCrypto";
4import { assert, it } from "@effect/vitest";
5import { ScheduledTaskUpsertInput } from "@t3tools/contracts";
6import * as Deferred from "effect/Deferred";
7import * as Effect from "effect/Effect";
8import * as Layer from "effect/Layer";
9import * as Queue from "effect/Queue";
10import * as Schema from "effect/Schema";
11import * as TestClock from "effect/testing/TestClock";
12
13import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
14import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
15import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
16import * as Scheduler from "../scheduling/Scheduler.ts";
17import * as ScheduledTaskService from "./ScheduledTaskService.ts";
18
19const decodeUpsertInput = Schema.decodeUnknownEffect(ScheduledTaskUpsertInput);
20
21type LaunchInput = ThreadLaunchService.ThreadLaunchInput;
22
23const webhookTaskInput = (overrides: Record<string, unknown> = {}) =>
24 decodeUpsertInput({
25 id: "scheduled-task:hook",
26 title: "Review PRs",
27 prompt: "Review this PR: {{body.pull_request.url}}",
28 enabled: true,
29 schedule: { type: "webhook" },
30 projectId: "project-webhook",
31 workspaceStrategy: { type: "root" },
32 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
33 runtimeMode: "full-access",
34 interactionMode: "default",
35 ...overrides,
36 });
37
38const pullRequestBody = new TextEncoder().encode(
39 JSON.stringify({ pull_request: { url: "https://github.com/org/repo/pull/45" } }),
40);
41
42const requestFor = (
43 task: { readonly id: string; readonly webhook?: { readonly path: string } | undefined },
44 overrides: Partial<ScheduledTaskService.WebhookTriggerRequest> = {},
45): ScheduledTaskService.WebhookTriggerRequest => ({
46 hookId: task.id,
47 token: task.webhook?.path.split("/").at(-1) ?? "",
48 method: "POST",
49 path: `/api/hooks/${task.id}`,
50 query: "",
51 headers: { "content-type": "application/json" },
52 body: pullRequestBody,
53 bodyText: new TextDecoder().decode(pullRequestBody),
54 ...overrides,
55});
56
57/**
58 * Runs `body` against a service whose launches are pushed to `launches`;
59 * `gate`, when given, holds each launch until the test releases it.
60 */
61const withService = <A, E>(
62 body: (input: {
63 readonly service: ScheduledTaskService.ScheduledTaskService["Service"];
64 readonly launches: Queue.Queue<LaunchInput>;
65 }) => Effect.Effect<A, E, never>,
66 options: { readonly gate?: Deferred.Deferred<void> } = {},
67) =>
68 Effect.gen(function* () {
69 const launches = yield* Queue.unbounded<LaunchInput>();
70 const dependencies = Layer.mergeAll(
71 NodePlatformCrypto.layer,
72 Scheduler.layer,
73 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
74 launch: (input) =>
75 Queue.offer(launches, input).pipe(
76 Effect.andThen(options.gate ? Deferred.await(options.gate) : Effect.void),
77 Effect.as({ threadId: "thread-1", resumed: false } as never),
78 ),
79 }),
80 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
81 );
82 return yield* Effect.gen(function* () {
83 const service = yield* ScheduledTaskService.ScheduledTaskService;
84 return yield* body({ service, launches });
85 }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(dependencies))));
86 }).pipe(Effect.provide(SqlitePersistenceMemory));
87
88it.effect("dispatches exactly the rendered prompt and logs the delivery", () =>
89 withService(({ service, launches }) =>
90 Effect.gen(function* () {
91 const { task } = yield* service.upsert(yield* webhookTaskInput());
92 assert.equal(task.nextRunAt, null);
93 assert.isDefined(task.webhook);
94 assert.isTrue(task.webhook!.path.startsWith("/api/hooks/scheduled-task%3Ahook/"));
95 // Not linked to T3 Connect in tests.
96 assert.equal(task.webhook!.url, null);
97
98 const result = yield* service.triggerWebhook(requestFor(task));
99 assert.equal(result._tag, "accepted");
100 const launched = yield* Queue.take(launches);
101 assert.equal(
102 launched.initialMessage?.text,
103 "Review this PR: https://github.com/org/repo/pull/45",
104 );
105 assert.equal(
106 launched.commandId,
107 `scheduled-task:${task.id}:webhook:${result._tag === "accepted" ? result.deliveryId : ""}`,
108 );
109
110 const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id });
111 assert.equal(deliveries.length, 1);
112 assert.equal(deliveries[0]?.outcome, "accepted");
113 const { delivery } = yield* service.getWebhookDelivery({
114 id: task.id,
115 deliveryId: deliveries[0]!.id,
116 });
117 assert.equal(delivery.body, new TextDecoder().decode(pullRequestBody));
118 assert.equal(delivery.renderedPrompt, launched.initialMessage?.text);
119 }),
120 ),
121);
122
123it.effect("answers not found for a wrong token or unknown hook without logging", () =>
124 withService(({ service }) =>
125 Effect.gen(function* () {
126 const { task } = yield* service.upsert(yield* webhookTaskInput());
127 const wrongToken = yield* service.triggerWebhook(requestFor(task, { token: "nope" }));
128 assert.equal(wrongToken._tag, "not_found");
129 const unknown = yield* service.triggerWebhook(requestFor(task, { hookId: "missing" }));
130 assert.equal(unknown._tag, "not_found");
131 assert.equal((yield* service.listWebhookDeliveries({ id: task.id })).deliveries.length, 0);
132 }),
133 ),
134);
135
136it.effect("rotating the token retires the old URL and saving keeps it", () =>
137 withService(({ service }) =>
138 Effect.gen(function* () {
139 const input = yield* webhookTaskInput();
140 const { task } = yield* service.upsert(input);
141 const saved = yield* service.upsert(yield* webhookTaskInput({ title: "Renamed" }));
142 assert.equal(saved.task.webhook?.path, task.webhook?.path);
143
144 const rotated = yield* service.rotateWebhookToken({ id: task.id });
145 assert.notEqual(rotated.task.webhook?.path, task.webhook?.path);
146 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "not_found");
147 assert.equal((yield* service.triggerWebhook(requestFor(rotated.task)))._tag, "accepted");
148 }),
149 ),
150);
151
152it.effect("checks the configured signature and keeps the secret write-only", () =>
153 withService(({ service }) =>
154 Effect.gen(function* () {
155 const { task } = yield* service.upsert(
156 yield* webhookTaskInput({
157 schedule: {
158 type: "webhook",
159 signature: {
160 header: "X-Hub-Signature-256",
161 encoding: "hex",
162 prefix: "sha256=",
163 secret: "s3cret",
164 },
165 },
166 }),
167 );
168 assert.deepEqual(task.schedule, {
169 type: "webhook",
170 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" },
171 });
172 assert.isTrue(task.webhook?.hasSecret);
173
174 const unsigned = yield* service.triggerWebhook(requestFor(task));
175 assert.equal(unsigned._tag, "rejected_signature");
176
177 const signature = `sha256=${NodeCrypto.createHmac("sha256", "s3cret").update(pullRequestBody).digest("hex")}`;
178 const signed = yield* service.triggerWebhook(
179 requestFor(task, {
180 headers: { "content-type": "application/json", "x-hub-signature-256": signature },
181 }),
182 );
183 assert.equal(signed._tag, "accepted");
184
185 // Saving without a secret keeps the stored one.
186 const resaved = yield* service.upsert(
187 yield* webhookTaskInput({
188 schedule: {
189 type: "webhook",
190 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" },
191 },
192 }),
193 );
194 assert.isTrue(resaved.task.webhook?.hasSecret);
195
196 const outcomes = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries.map(
197 (delivery) => [delivery.outcome, delivery.signatureVerified],
198 );
199 assert.deepEqual(outcomes.toSorted(), [
200 ["accepted", true],
201 ["rejected_signature", false],
202 ]);
203 }),
204 ),
205);
206
207it.effect("logs but does not run deliveries to a disabled task, and refuses run now", () =>
208 withService(({ service, launches }) =>
209 Effect.gen(function* () {
210 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
211 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "disabled");
212 assert.equal(yield* Queue.size(launches), 0);
213 const runNow = yield* service.runNow({ id: task.id }).pipe(Effect.flip);
214 assert.equal(runNow.message, "Webhook tasks run when their URL receives a request.");
215 }),
216 ),
217);
218
219it.effect("queues a burst of deliveries instead of dropping them", () =>
220 Effect.gen(function* () {
221 const gate = yield* Deferred.make<void>();
222 yield* withService(
223 ({ service, launches }) =>
224 Effect.gen(function* () {
225 const { task } = yield* service.upsert(yield* webhookTaskInput());
226 const results = yield* Effect.forEach([1, 2, 3], () =>
227 service.triggerWebhook(requestFor(task)),
228 );
229 assert.deepEqual(
230 results.map((result) => result._tag),
231 ["accepted", "accepted", "accepted"],
232 );
233 // Only the first is dispatching; the others wait their turn.
234 yield* Queue.take(launches);
235 yield* Deferred.succeed(gate, undefined);
236 const rest = yield* Effect.all([Queue.take(launches), Queue.take(launches)]);
237 assert.equal(new Set(rest.map((launch) => launch.commandId)).size, 2);
238 }),
239 { gate },
240 );
241 }),
242);
243
244it.effect("rate limits a hook past 60 deliveries a minute", () =>
245 withService(({ service }) =>
246 Effect.gen(function* () {
247 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
248 const results = yield* Effect.forEach(Array.from({ length: 61 }), () =>
249 service.triggerWebhook(requestFor(task)),
250 );
251 assert.equal(results.at(-2)?._tag, "disabled");
252 assert.equal(results.at(-1)?._tag, "rate_limited");
253 // Further rejections in the same window are counted, not logged.
254 yield* Effect.forEach([1, 2, 3], () => service.triggerWebhook(requestFor(task)));
255 const outcomes = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries.map(
256 (delivery) => delivery.outcome,
257 );
258 assert.equal(outcomes.filter((outcome) => outcome === "rate_limited").length, 1);
259 }),
260 ),
261);
262
263it.effect("a save carrying a stale token cannot undo a rotation", () =>
264 withService(({ service }) =>
265 Effect.gen(function* () {
266 const { task } = yield* service.upsert(yield* webhookTaskInput());
267 const rotated = yield* service.rotateWebhookToken({ id: task.id });
268 // The editor was opened before the rotation and saves afterwards.
269 const saved = yield* service.upsert(yield* webhookTaskInput({ title: "Edited" }));
270 assert.equal(saved.task.webhook?.path, rotated.task.webhook?.path);
271 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "not_found");
272 }),
273 ),
274);
275
276it.effect("keeps the newest 50 deliveries when they share a timestamp", () =>
277 withService(({ service }) =>
278 Effect.gen(function* () {
279 // Paused, so each request is logged without starting a run. The test
280 // clock is frozen, so every delivery has the same received_at.
281 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
282 const ids = yield* Effect.forEach(Array.from({ length: 55 }), (_, index) =>
283 service.triggerWebhook(requestFor(task, { query: `n=${index}` })).pipe(Effect.as(index)),
284 );
285 const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id });
286 assert.equal(deliveries.length, 50);
287 const first = yield* service.getWebhookDelivery({
288 id: task.id,
289 deliveryId: deliveries[0]!.id,
290 });
291 const last = yield* service.getWebhookDelivery({
292 id: task.id,
293 deliveryId: deliveries.at(-1)!.id,
294 });
295 assert.equal(first.delivery.query, `n=${ids.at(-1)}`);
296 assert.equal(last.delivery.query, "n=5");
297 }),
298 ),
299);
300
301it.effect("keeps credential headers out of the delivery log", () =>
302 withService(({ service }) =>
303 Effect.gen(function* () {
304 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
305 yield* service.triggerWebhook(
306 requestFor(task, {
307 headers: {
308 "content-type": "application/json",
309 authorization: "Bearer sender-token",
310 "x-webhook-key": "k",
311 "x-github-event": "push",
312 },
313 }),
314 );
315 const [summary] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
316 const { delivery } = yield* service.getWebhookDelivery({
317 id: task.id,
318 deliveryId: summary!.id,
319 });
320 assert.equal(delivery.headers.authorization, "[redacted]");
321 assert.equal(delivery.headers["x-webhook-key"], "[redacted]");
322 assert.equal(delivery.headers["x-github-event"], "push");
323 }),
324 ),
325);
326
327const queuedDeliveryCases = [
328 {
329 change: "paused",
330 reason: "The task was paused before this delivery ran.",
331 apply: (service: ScheduledTaskService.ScheduledTaskService["Service"], id: string) =>
332 service.setEnabled({ id: id as never, enabled: false }),
333 },
334 {
335 change: "switched to an interval trigger",
336 reason: "The task's trigger changed before this delivery ran.",
337 apply: (service: ScheduledTaskService.ScheduledTaskService["Service"]) =>
338 webhookTaskInput({ schedule: { type: "interval", everyMs: 3_600_000 } }).pipe(
339 Effect.flatMap(service.upsert),
340 ),
341 },
342] as const;
343
344it.effect.each(queuedDeliveryCases)(
345 "a delivery queued behind a run does not start once the task is $change",
346 ({ reason, apply }) =>
347 Effect.gen(function* () {
348 const gate = yield* Deferred.make<void>();
349 yield* withService(
350 ({ service, launches }) =>
351 Effect.gen(function* () {
352 const { task } = yield* service.upsert(yield* webhookTaskInput());
353 yield* service.triggerWebhook(requestFor(task));
354 const queued = yield* service.triggerWebhook(requestFor(task));
355 yield* Queue.take(launches);
356 yield* apply(service, task.id);
357 yield* Deferred.succeed(gate, undefined);
358 // The queued delivery is marked failed instead of launching.
359 const deliveryId = queued._tag === "accepted" ? queued.deliveryId : undefined;
360 let delivery = (yield* service.getWebhookDelivery({
361 id: task.id,
362 deliveryId: deliveryId!,
363 })).delivery;
364 while (delivery.outcome === "accepted") {
365 yield* Effect.yieldNow;
366 delivery = (yield* service.getWebhookDelivery({
367 id: task.id,
368 deliveryId: deliveryId!,
369 })).delivery;
370 }
371 assert.equal(delivery.outcome, "dispatch_failed");
372 assert.equal(delivery.error, reason);
373 assert.equal(yield* Queue.size(launches), 0);
374 }),
375 { gate },
376 );
377 }),
378);
379
380it.effect("a save without a secret keeps a secret changed after it was read", () =>
381 withService(({ service }) =>
382 Effect.gen(function* () {
383 const signature = { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" };
384 const { task } = yield* service.upsert(
385 yield* webhookTaskInput({
386 schedule: { type: "webhook", signature: { ...signature, secret: "old" } },
387 }),
388 );
389 yield* service.upsert(
390 yield* webhookTaskInput({
391 schedule: { type: "webhook", signature: { ...signature, secret: "new" } },
392 }),
393 );
394 // A form opened before the change saves without sending a secret.
395 yield* service.upsert(
396 yield* webhookTaskInput({ title: "Edited", schedule: { type: "webhook", signature } }),
397 );
398 const sign = (secret: string) =>
399 `sha256=${NodeCrypto.createHmac("sha256", secret).update(pullRequestBody).digest("hex")}`;
400 const withSignature = (secret: string) =>
401 requestFor(task, {
402 headers: { "content-type": "application/json", "x-hub-signature-256": sign(secret) },
403 });
404 assert.equal(
405 (yield* service.triggerWebhook(withSignature("old")))._tag,
406 "rejected_signature",
407 );
408 assert.equal((yield* service.triggerWebhook(withSignature("new")))._tag, "accepted");
409 }),
410 ),
411);
412
413it.effect("caps deliveries waiting behind a stuck run", () =>
414 Effect.gen(function* () {
415 const gate = yield* Deferred.make<void>();
416 yield* withService(
417 ({ service, launches }) =>
418 Effect.gen(function* () {
419 const { task } = yield* service.upsert(yield* webhookTaskInput());
420 yield* service.triggerWebhook(requestFor(task));
421 yield* Queue.take(launches);
422 // The cap counts the running delivery too: 19 more wait, the next is refused.
423 const waiting = yield* Effect.forEach(Array.from({ length: 20 }), () =>
424 service.triggerWebhook(requestFor(task)),
425 );
426 assert.equal(waiting.filter((result) => result._tag === "accepted").length, 19);
427 assert.equal(waiting.at(-1)?._tag, "rate_limited");
428 // The refused request is not logged.
429 const logged = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
430 assert.equal(logged.length, 20);
431 yield* Deferred.succeed(gate, undefined);
432 }),
433 { gate },
434 );
435 }),
436);
437
438it.effect("logs a body's first 64 KiB by bytes, not characters", () =>
439 withService(({ service }) =>
440 Effect.gen(function* () {
441 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
442 // 30 000 three-byte characters: under 64 Ki characters, over 64 KiB.
443 const text = "界".repeat(30_000);
444 const body = new TextEncoder().encode(text);
445 yield* service.triggerWebhook(requestFor(task, { body, bodyText: text }));
446 const [summary] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
447 const { delivery } = yield* service.getWebhookDelivery({
448 id: task.id,
449 deliveryId: summary!.id,
450 });
451 assert.isTrue(delivery.bodyTruncated);
452 assert.isAtMost(new TextEncoder().encode(delivery.body).byteLength, 64 * 1024 + 3);
453 }),
454 ),
455);
456
457it.effect("deleting a task removes its delivery log", () =>
458 withService(({ service }) =>
459 Effect.gen(function* () {
460 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
461 yield* service.triggerWebhook(requestFor(task));
462 yield* service.delete({ id: task.id });
463 assert.equal((yield* service.listWebhookDeliveries({ id: task.id })).deliveries.length, 0);
464 }),
465 ),
466);
apps/server/src/scheduledTasks/webhookRoute.test.ts created+147
...@@ -0,0 +1,147 @@
1import { describe, expect, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import * as Etag from "effect/http/Etag";
6import * as HttpPlatform from "effect/http/HttpPlatform";
7import * as HttpRouter from "effect/http/HttpRouter";
8import * as HttpApi from "effect/http-api/HttpApi";
9import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
10
11import {
12 EnvironmentHttpApi,
13 ScheduledTaskWebhookDeliveryId,
14 ScheduledTaskError,
15} from "@t3tools/contracts";
16import {
17 ScheduledTaskService,
18 type WebhookTriggerRequest,
19 type WebhookTriggerResult,
20} from "./ScheduledTaskService.ts";
21import { WEBHOOK_MAX_BODY_BYTES, webhookHttpApiLayer } from "./webhookRoute.ts";
22
23class WebhookTestApi extends HttpApi.make("environment").add(EnvironmentHttpApi.groups.webhooks) {}
24
25const handlerFor = (
26 trigger: (
27 request: WebhookTriggerRequest,
28 ) => Effect.Effect<WebhookTriggerResult, ScheduledTaskError>,
29) =>
30 HttpRouter.toWebHandler(
31 HttpApiBuilder.layer(WebhookTestApi).pipe(
32 Layer.provide(webhookHttpApiLayer),
33 Layer.provide(Layer.mock(ScheduledTaskService)({ triggerWebhook: trigger })),
34 Layer.provide(
35 HttpPlatform.layer.pipe(
36 Layer.provideMerge(NodeServices.layer),
37 Layer.provideMerge(Etag.layerWeak),
38 ),
39 ),
40 Layer.provide(NodeServices.layer),
41 ),
42 { disableLogger: true },
43 );
44
45const post = (
46 path: string,
47 body: string | Uint8Array<ArrayBuffer>,
48 headers: Record<string, string> = {},
49) => new Request(`http://env.local${path}`, { method: "POST", body, headers });
50
51describe("webhook route", () => {
52 it("passes the raw request to the service and answers 202 with the delivery id", async () => {
53 let received: WebhookTriggerRequest | undefined;
54 const { handler, dispose } = handlerFor((request) => {
55 received = request;
56 return Effect.succeed({
57 _tag: "accepted",
58 deliveryId: ScheduledTaskWebhookDeliveryId.make("delivery:1"),
59 });
60 });
61 try {
62 const response = await handler(
63 post("/api/hooks/scheduled-task%3Ahook/tok?x=1", '{"a":1}', {
64 "Content-Type": "application/json",
65 "X-GitHub-Event": "push",
66 }),
67 );
68 expect(response.status).toBe(202);
69 expect(await response.json()).toEqual({ deliveryId: "delivery:1" });
70 expect(received?.hookId).toBe("scheduled-task:hook");
71 expect(received?.token).toBe("tok");
72 expect(received?.query).toBe("x=1");
73 expect(received?.headers["x-github-event"]).toBe("push");
74 expect(received?.bodyText).toBe('{"a":1}');
75 } finally {
76 await dispose();
77 }
78 });
79
80 it("maps service outcomes to status codes", async () => {
81 const cases: ReadonlyArray<[WebhookTriggerResult["_tag"], number]> = [
82 ["not_found", 404],
83 ["rejected_signature", 401],
84 ["disabled", 409],
85 ["rate_limited", 429],
86 ];
87 for (const [tag, status] of cases) {
88 const { handler, dispose } = handlerFor(() =>
89 Effect.succeed({ _tag: tag } as WebhookTriggerResult),
90 );
91 try {
92 expect((await handler(post("/api/hooks/id/tok", "{}"))).status).toBe(status);
93 } finally {
94 await dispose();
95 }
96 }
97 });
98
99 it("rejects oversized bodies and malformed paths before reaching the service", async () => {
100 let calls = 0;
101 const { handler, dispose } = handlerFor(() => {
102 calls += 1;
103 return Effect.succeed({ _tag: "not_found" });
104 });
105 try {
106 const big = new Uint8Array(WEBHOOK_MAX_BODY_BYTES + 1);
107 expect((await handler(post("/api/hooks/id/tok", big))).status).toBe(413);
108 expect((await handler(post("/api/hooks/id", "{}"))).status).toBe(404);
109 expect((await handler(post("/api/hooks/id/tok/extra", "{}"))).status).toBe(404);
110 expect((await handler(post("/api/hooks/%E0/tok", "{}"))).status).toBe(404);
111 // No content-length: the reader cap must still apply.
112 const chunked = new ReadableStream<Uint8Array>({
113 start(controller) {
114 for (let sent = 0; sent <= WEBHOOK_MAX_BODY_BYTES; sent += 64 * 1024) {
115 controller.enqueue(new Uint8Array(64 * 1024));
116 }
117 controller.close();
118 },
119 });
120 const streamed = await handler(
121 new Request("http://env.local/api/hooks/id/tok", {
122 method: "POST",
123 body: chunked,
124 // Node's fetch needs duplex for streamed bodies.
125 duplex: "half",
126 }),
127 );
128 expect(streamed.status).toBe(413);
129 expect(calls).toBe(0);
130 } finally {
131 await dispose();
132 }
133 });
134
135 it("hides service failures behind a 500", async () => {
136 const { handler, dispose } = handlerFor(() =>
137 Effect.fail(new ScheduledTaskError({ message: "database locked" })),
138 );
139 try {
140 const response = await handler(post("/api/hooks/id/tok", "{}"));
141 expect(response.status).toBe(500);
142 expect(await response.text()).not.toContain("database");
143 } finally {
144 await dispose();
145 }
146 });
147});
apps/server/src/scheduledTasks/webhookRoute.ts created+106
...@@ -0,0 +1,106 @@
1import { EnvironmentHttpApi } from "@t3tools/contracts";
2import * as ByteSize from "effect/ByteSize";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as HttpIncomingMessage from "effect/http/HttpIncomingMessage";
6import type * as HttpServerRequest from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8import * as HttpApiBuilder from "effect/http-api/HttpApiBuilder";
9
10import * as ScheduledTaskService from "./ScheduledTaskService.ts";
11
12/** Largest request body a webhook accepts. The relay enforces the same cap. */
13export const WEBHOOK_MAX_BODY_BYTES = 1024 * 1024;
14
15const json = (status: number, body: Record<string, string>) =>
16 HttpServerResponse.jsonUnsafe(body, { status });
17
18/**
19 * Handles `/api/hooks/:hookId/:token` for every accepted method. The endpoint
20 * is raw so the signature is checked over the exact body bytes; the service
21 * checks the token and signature. It is reachable directly, over the managed
22 * tunnel, or through the relay's stable `/v1/hooks/...` URL.
23 */
24const handleWebhook =
25 (scheduledTasks: ScheduledTaskService.ScheduledTaskService["Service"]) =>
26 ({
27 params,
28 request,
29 }: {
30 readonly params: { readonly hookId: string; readonly token: string };
31 readonly request: HttpServerRequest.HttpServerRequest;
32 }) =>
33 Effect.gen(function* () {
34 const contentLength = Number(request.headers["content-length"] ?? "0");
35 if (!Number.isFinite(contentLength) || contentLength > WEBHOOK_MAX_BODY_BYTES) {
36 return json(413, { error: "body_too_large" });
37 }
38 // Chunked requests carry no content-length, so the reader itself is capped.
39 const body = yield* request.arrayBuffer.pipe(
40 Effect.map((buffer) => new Uint8Array(buffer)),
41 Effect.provideService(
42 HttpIncomingMessage.MaxBodySize,
43 ByteSize.bytes(WEBHOOK_MAX_BODY_BYTES),
44 ),
45 Effect.option,
46 );
47 if (Option.isNone(body)) return json(413, { error: "body_too_large_or_unreadable" });
48 if (body.value.byteLength > WEBHOOK_MAX_BODY_BYTES) {
49 return json(413, { error: "body_too_large" });
50 }
51
52 const headers: Record<string, string> = {};
53 for (const [name, value] of Object.entries(request.headers)) {
54 if (typeof value === "string") headers[name.toLowerCase()] = value;
55 }
56 const queryIndex = request.url.indexOf("?");
57
58 const result = yield* scheduledTasks
59 .triggerWebhook({
60 hookId: params.hookId,
61 token: params.token,
62 method: request.method,
63 path: `${ScheduledTaskService.WEBHOOK_ROUTE_PREFIX}/${encodeURIComponent(params.hookId)}`,
64 query: queryIndex === -1 ? "" : request.url.slice(queryIndex + 1),
65 headers,
66 body: body.value,
67 bodyText: new TextDecoder().decode(body.value),
68 })
69 .pipe(
70 Effect.catch((cause) =>
71 Effect.logWarning("Webhook delivery failed").pipe(
72 Effect.annotateLogs({ hookId: params.hookId }),
73 Effect.andThen(Effect.logDebug("Webhook delivery failure cause", { cause })),
74 Effect.as({ _tag: "error" as const }),
75 ),
76 ),
77 );
78
79 switch (result._tag) {
80 case "accepted":
81 return json(202, { deliveryId: result.deliveryId });
82 case "not_found":
83 return json(404, { error: "hook_not_found" });
84 case "rejected_signature":
85 return json(401, { error: "invalid_signature" });
86 case "disabled":
87 return json(409, { error: "hook_disabled" });
88 case "rate_limited":
89 return json(429, { error: "rate_limited" });
90 case "error":
91 return json(500, { error: "internal_error" });
92 }
93 });
94
95export const webhookHttpApiLayer = HttpApiBuilder.group(
96 EnvironmentHttpApi,
97 "webhooks",
98 Effect.fnUntraced(function* (handlers) {
99 const handler = handleWebhook(yield* ScheduledTaskService.ScheduledTaskService);
100 return handlers
101 .handleRaw("webhookPost", handler)
102 .handleRaw("webhookPut", handler)
103 .handleRaw("webhookPatch", handler)
104 .handleRaw("webhookGet", handler);
105 }),
106);
apps/server/src/scheduledTasks/webhookTemplate.test.ts created+79
...@@ -0,0 +1,79 @@
1import { assert, describe, it } from "@effect/vitest";
2
3import { renderWebhookPrompt, type WebhookRequest } from "./webhookTemplate.ts";
4
5const githubPullRequest: WebhookRequest = {
6 method: "POST",
7 path: "/api/hooks/task-1",
8 query: "source=github",
9 headers: { "content-type": "application/json", "x-github-event": "pull_request" },
10 bodyText: JSON.stringify({
11 action: "opened",
12 pull_request: { url: "https://github.com/org/repo/pull/45", number: 45, labels: ["bug"] },
13 }),
14};
15
16describe("renderWebhookPrompt", () => {
17 it("sends only what the template names", () => {
18 const rendered = renderWebhookPrompt(
19 "Review this PR: {{body.pull_request.url}}",
20 githubPullRequest,
21 );
22 assert.equal(rendered.prompt, "Review this PR: https://github.com/org/repo/pull/45");
23 assert.deepEqual(rendered.missing, []);
24 });
25
26 it("reads headers case-insensitively, query parameters and array indexes", () => {
27 const rendered = renderWebhookPrompt(
28 "{{ headers.X-GitHub-Event }} #{{body.pull_request.number}} {{body.pull_request.labels.0}} {{query.source}}",
29 githubPullRequest,
30 );
31 assert.equal(rendered.prompt, "pull_request #45 bug github");
32 });
33
34 it("renders objects as JSON and the raw body and request on request", () => {
35 const rendered = renderWebhookPrompt(
36 "{{body.pull_request.labels}}|{{body}}",
37 githubPullRequest,
38 );
39 assert.equal(rendered.prompt, `[\n "bug"\n]|${githubPullRequest.bodyText}`);
40 const request = renderWebhookPrompt("{{request}}", githubPullRequest).prompt;
41 assert.isTrue(request.startsWith("POST /api/hooks/task-1?source=github\n"));
42 assert.include(request, "x-github-event: pull_request");
43 assert.isTrue(request.endsWith(githubPullRequest.bodyText));
44 });
45
46 it("renders missing fields empty and reports them once", () => {
47 const rendered = renderWebhookPrompt(
48 "a={{body.nope}} b={{body.nope}} c={{headers.x-missing}} d={{unknown}}",
49 githubPullRequest,
50 );
51 assert.equal(rendered.prompt, "a= b= c= d=");
52 assert.deepEqual(rendered.missing, ["body.nope", "headers.x-missing", "unknown"]);
53 });
54
55 it("addresses form-encoded bodies and leaves other bodies to {{body}}", () => {
56 const form = renderWebhookPrompt("{{body.text}} by {{body.user_name}}", {
57 ...githubPullRequest,
58 headers: { "content-type": "application/x-www-form-urlencoded" },
59 bodyText: "text=deploy+prod&user_name=alice",
60 });
61 assert.equal(form.prompt, "deploy prod by alice");
62
63 const plain = renderWebhookPrompt("{{body.field}}{{body}}", {
64 ...githubPullRequest,
65 headers: { "content-type": "text/plain" },
66 bodyText: "build failed",
67 });
68 assert.equal(plain.prompt, "build failed");
69 assert.deepEqual(plain.missing, ["body.field"]);
70 });
71
72 it("does not resolve inherited object properties", () => {
73 const rendered = renderWebhookPrompt(
74 "{{body.constructor}}{{body.__proto__}}",
75 githubPullRequest,
76 );
77 assert.equal(rendered.prompt, "");
78 });
79});
apps/server/src/scheduledTasks/webhookTemplate.ts created+114
...@@ -0,0 +1,114 @@
1/**
2 * Renders a webhook task's prompt from the request that triggered it. The
3 * rendered prompt is the only thing the agent receives, so a user who wants
4 * the whole request writes `{{request}}` or `{{body}}` themselves.
5 *
6 * The syntax is deliberately just `{{path}}` lookups — no conditionals,
7 * defaults, filters or escaping — so it never grows into a template language:
8 *
9 * - `{{body.a.b.0}}` a field of a JSON or form-encoded body
10 * - `{{headers.name}}` a request header (case-insensitive)
11 * - `{{query.name}}` a URL query parameter
12 * - `{{body}}` the raw body text
13 * - `{{request}}` method, path, query, headers and body
14 *
15 * Strings and numbers render as text, objects and arrays as JSON. A path with
16 * no value renders empty and is reported in `missing`.
17 */
18
19export interface WebhookRequest {
20 readonly method: string;
21 readonly path: string;
22 /** Raw query string without the leading `?`. */
23 readonly query: string;
24 /** Lowercased header names. */
25 readonly headers: Readonly<Record<string, string>>;
26 readonly bodyText: string;
27}
28
29export interface RenderedWebhookPrompt {
30 readonly prompt: string;
31 readonly missing: ReadonlyArray<string>;
32}
33
34const PLACEHOLDER = /\{\{\s*([^{}]*?)\s*\}\}/g;
35
36function parseBody(request: WebhookRequest): unknown {
37 const contentType = (request.headers["content-type"] ?? "").toLowerCase();
38 if (contentType.includes("application/x-www-form-urlencoded")) {
39 return Object.fromEntries(new URLSearchParams(request.bodyText));
40 }
41 // Senders are inconsistent about content types, so any body that parses
42 // as JSON is addressable.
43 try {
44 return JSON.parse(request.bodyText) as unknown;
45 } catch {
46 return undefined;
47 }
48}
49
50function lookup(root: unknown, segments: ReadonlyArray<string>): unknown {
51 let current = root;
52 for (const segment of segments) {
53 if (current === null || typeof current !== "object") return undefined;
54 if (!Object.hasOwn(current, segment)) return undefined;
55 current = (current as Record<string, unknown>)[segment];
56 }
57 return current;
58}
59
60function stringify(value: unknown): string {
61 if (typeof value === "string") return value;
62 if (typeof value === "number" || typeof value === "boolean") return String(value);
63 return JSON.stringify(value, null, 2);
64}
65
66function formatWebhookRequest(request: WebhookRequest): string {
67 const headerLines = Object.entries(request.headers).map(([name, value]) => `${name}: ${value}`);
68 return [
69 `${request.method} ${request.path}${request.query ? `?${request.query}` : ""}`,
70 ...headerLines,
71 "",
72 request.bodyText,
73 ].join("\n");
74}
75
76export function renderWebhookPrompt(
77 template: string,
78 request: WebhookRequest,
79): RenderedWebhookPrompt {
80 const missing: string[] = [];
81 let body: { parsed: unknown } | undefined;
82 const parsedBody = () => (body ??= { parsed: parseBody(request) }).parsed;
83
84 const resolve = (expression: string): unknown => {
85 const [root, ...segments] = expression.split(".");
86 switch (root) {
87 case "request":
88 return segments.length === 0 ? formatWebhookRequest(request) : undefined;
89 case "body":
90 return segments.length === 0 ? request.bodyText : lookup(parsedBody(), segments);
91 case "headers":
92 return segments.length === 0
93 ? request.headers
94 : request.headers[segments.join(".").toLowerCase()];
95 case "query": {
96 const params = new URLSearchParams(request.query);
97 if (segments.length === 0) return Object.fromEntries(params);
98 return params.get(segments.join(".")) ?? undefined;
99 }
100 default:
101 return undefined;
102 }
103 };
104
105 const prompt = template.replace(PLACEHOLDER, (_match, expression: string) => {
106 const value = resolve(expression);
107 if (value === undefined || value === null) {
108 if (!missing.includes(expression)) missing.push(expression);
109 return "";
110 }
111 return stringify(value);
112 });
113 return { prompt, missing };
114}
apps/server/src/scheduledTasks/webhookVerification.test.ts created+59
...@@ -0,0 +1,59 @@
1import * as NodeCrypto from "node:crypto";
2
3import { assert, describe, it } from "@effect/vitest";
4
5import { verifyWebhookSignature } from "./webhookVerification.ts";
6
7const body = new TextEncoder().encode('{"action":"opened"}');
8const secret = "shared-secret";
9const hmac = () => NodeCrypto.createHmac("sha256", secret).update(body);
10
11describe("verifyWebhookSignature", () => {
12 const github = { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" } as const;
13
14 it("accepts a GitHub-style hex signature with prefix", () => {
15 const headers = { "x-hub-signature-256": `sha256=${hmac().digest("hex")}` };
16 assert.isTrue(verifyWebhookSignature({ signature: github, secret, headers, body }));
17 });
18
19 it("rejects a tampered body, a wrong secret, a missing header and a missing prefix", () => {
20 const valid = `sha256=${hmac().digest("hex")}`;
21 const tampered = new TextEncoder().encode('{"action":"closed"}');
22 assert.isFalse(
23 verifyWebhookSignature({
24 signature: github,
25 secret,
26 headers: { "x-hub-signature-256": valid },
27 body: tampered,
28 }),
29 );
30 assert.isFalse(
31 verifyWebhookSignature({
32 signature: github,
33 secret: "other",
34 headers: { "x-hub-signature-256": valid },
35 body,
36 }),
37 );
38 assert.isFalse(verifyWebhookSignature({ signature: github, secret, headers: {}, body }));
39 assert.isFalse(
40 verifyWebhookSignature({
41 signature: github,
42 secret,
43 headers: { "x-hub-signature-256": hmac().digest("hex") },
44 body,
45 }),
46 );
47 });
48
49 it("accepts a base64 signature without prefix", () => {
50 assert.isTrue(
51 verifyWebhookSignature({
52 signature: { header: "X-Signature", encoding: "base64", prefix: "" },
53 secret,
54 headers: { "x-signature": hmac().digest("base64") },
55 body,
56 }),
57 );
58 });
59});
apps/server/src/scheduledTasks/webhookVerification.ts created+36
...@@ -0,0 +1,36 @@
1import * as NodeCrypto from "node:crypto";
2
3import type { ScheduledTaskWebhookSignature } from "@t3tools/contracts";
4
5/** Constant-time string comparison that does not leak length through timing. */
6export function constantTimeEquals(a: string, b: string): boolean {
7 const digestA = NodeCrypto.createHash("sha256").update(a).digest();
8 const digestB = NodeCrypto.createHash("sha256").update(b).digest();
9 return NodeCrypto.timingSafeEqual(digestA, digestB);
10}
11
12/**
13 * Checks an HMAC-SHA256 signature over the raw body bytes, as GitHub, Linear,
14 * Shopify and most other signing senders do. The header value is
15 * `<prefix><digest>`, with the digest hex- or base64-encoded.
16 */
17export function verifyWebhookSignature(input: {
18 readonly signature: ScheduledTaskWebhookSignature;
19 readonly secret: string;
20 readonly headers: Readonly<Record<string, string>>;
21 readonly body: Uint8Array;
22}): boolean {
23 const received = input.headers[input.signature.header.toLowerCase()];
24 if (received === undefined) return false;
25 const value = received.trim();
26 const prefix = input.signature.prefix;
27 if (prefix !== "" && !value.toLowerCase().startsWith(prefix.toLowerCase())) return false;
28 const digest = NodeCrypto.createHmac("sha256", input.secret).update(input.body).digest();
29 const expected =
30 input.signature.encoding === "hex" ? digest.toString("hex") : digest.toString("base64");
31 const candidate = value.slice(prefix.length);
32 return constantTimeEquals(
33 input.signature.encoding === "hex" ? candidate.toLowerCase() : candidate,
34 expected,
35 );
36}
apps/server/src/server.ts+30
...@@ -15,6 +15,7 @@ import * as Cause from "effect/Cause";...@@ -15,6 +15,7 @@ import * as Cause from "effect/Cause";
15import * as Duration from "effect/Duration";15import * as Duration from "effect/Duration";
16import * as Deferred from "effect/Deferred";16import * as Deferred from "effect/Deferred";
17import * as Effect from "effect/Effect";17import * as Effect from "effect/Effect";
18import * as Option from "effect/Option";
18import * as Layer from "effect/Layer";19import * as Layer from "effect/Layer";
19import * as Stream from "effect/Stream";20import * as Stream from "effect/Stream";
20import * as Schedule from "effect/Schedule";21import * as Schedule from "effect/Schedule";
...@@ -117,6 +118,9 @@ import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts";...@@ -117,6 +118,9 @@ import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts";
117import { authHttpApiLayer, environmentAuthenticatedAuthLayer } from "./auth/http.ts";118import { authHttpApiLayer, environmentAuthenticatedAuthLayer } from "./auth/http.ts";
118import * as ReplayMarkers from "./auth/replayMarkers.ts";119import * as ReplayMarkers from "./auth/replayMarkers.ts";
119import * as ServerSecretStore from "./auth/ServerSecretStore.ts";120import * as ServerSecretStore from "./auth/ServerSecretStore.ts";
121import { webhookHttpApiLayer } from "./scheduledTasks/webhookRoute.ts";
122import { ScheduledTaskWebhookOrigin } from "./scheduledTasks/ScheduledTaskService.ts";
123import { CLOUD_ENDPOINT_RUNTIME_CONFIG, RELAY_URL_SECRET } from "./cloud/config.ts";
120import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";124import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
121import {125import {
122 connectHttpApiLayer,126 connectHttpApiLayer,
...@@ -444,7 +448,32 @@ const CloudManagedEndpointRuntimeLive = Layer.mergeAll(...@@ -444,7 +448,32 @@ const CloudManagedEndpointRuntimeLive = Layer.mergeAll(
444 ),448 ),
445);449);
446450
451// Webhook URLs go through the relay only when the managed tunnel it forwards
452// to is configured; otherwise clients show the environment-relative path.
453const ScheduledTaskWebhookOriginLive = Layer.effect(
454 ScheduledTaskWebhookOrigin,
455 Effect.map(
456 Effect.all([ServerEnvironment.ServerEnvironment, ServerSecretStore.ServerSecretStore]),
457 ([environment, secrets]) =>
458 // The reference holds an effect so each read sees the current link state.
459 Effect.gen(function* () {
460 const [relayUrl, tunnelConfig] = yield* Effect.all([
461 secrets.get(RELAY_URL_SECRET),
462 secrets.get(CLOUD_ENDPOINT_RUNTIME_CONFIG),
463 ]).pipe(Effect.orElseSucceed(() => [Option.none(), Option.none()] as const));
464 return {
465 environmentId: yield* environment.getEnvironmentId,
466 relayUrl:
467 Option.isSome(relayUrl) && Option.isSome(tunnelConfig)
468 ? new TextDecoder().decode(relayUrl.value) || null
469 : null,
470 };
471 }),
472 ),
473);
474
447const OrchestrationV2RuntimeLayerLive = OrchestrationV2ProductionLayerLive.pipe(475const OrchestrationV2RuntimeLayerLive = OrchestrationV2ProductionLayerLive.pipe(
476 Layer.provide(ScheduledTaskWebhookOriginLive),
448 Layer.provide(ProviderEventIngestor.analyticsLive),477 Layer.provide(ProviderEventIngestor.analyticsLive),
449 Layer.provide(CheckpointStoreLayerLive),478 Layer.provide(CheckpointStoreLayerLive),
450 Layer.provide(GitWorkflowLayerLive),479 Layer.provide(GitWorkflowLayerLive),
...@@ -643,6 +672,7 @@ const makeRoutesLayer = Layer.mergeAll(...@@ -643,6 +672,7 @@ const makeRoutesLayer = Layer.mergeAll(
643 Layer.provide(pullRequestHttpApiLayer),672 Layer.provide(pullRequestHttpApiLayer),
644 Layer.provide(projectHttpApiLayer),673 Layer.provide(projectHttpApiLayer),
645 Layer.provide(serverEnvironmentHttpApiLayer),674 Layer.provide(serverEnvironmentHttpApiLayer),
675 Layer.provide(webhookHttpApiLayer),
646 Layer.provide(environmentAuthenticatedAuthLayer),676 Layer.provide(environmentAuthenticatedAuthLayer),
647 ),677 ),
648 otlpTracesProxyRouteLayer,678 otlpTracesProxyRouteLayer,
apps/server/src/ws.ts+38-6
...@@ -27,7 +27,9 @@ import {...@@ -27,7 +27,9 @@ import {
27 CommandId,27 CommandId,
28 AuthAccessStreamError,28 AuthAccessStreamError,
29 type AuthAccessStreamEvent,29 type AuthAccessStreamEvent,
30 AuthOrchestrationOperateScope,
30 type AuthEnvironmentScope,31 type AuthEnvironmentScope,
32 type ScheduledTaskListResult,
31 AuthSessionId,33 AuthSessionId,
32 ClientConnectionMethod,34 ClientConnectionMethod,
33 ClientDeviceType,35 ClientDeviceType,
...@@ -1315,6 +1317,14 @@ const makeWsRpcLayer = (...@@ -1315,6 +1317,14 @@ const makeWsRpcLayer = (
1315 const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor;1317 const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor;
1316 const resourceTelemetry = yield* ResourceTelemetry.ResourceTelemetry;1318 const resourceTelemetry = yield* ResourceTelemetry.ResourceTelemetry;
1317 const relayClient = yield* RelayClient.RelayClient;1319 const relayClient = yield* RelayClient.RelayClient;
1320 // A webhook URL starts agent runs, so only sessions that may operate
1321 // see it; read-only sessions still see the task itself.
1322 const withVisibleWebhookUrls = (result: ScheduledTaskListResult): ScheduledTaskListResult =>
1323 currentSession.scopes.includes(AuthOrchestrationOperateScope)
1324 ? result
1325 : {
1326 tasks: result.tasks.map(({ webhook: _webhook, ...task }) => task),
1327 };
1318 // RpcScopeAuthorization checks each RPC's declared scope before its handler1328 // RpcScopeAuthorization checks each RPC's declared scope before its handler
1319 // runs. This covers the one RPC whose scope depends on its input.1329 // runs. This covers the one RPC whose scope depends on its input.
1320 const authorizeEffect = <A, E, R>(1330 const authorizeEffect = <A, E, R>(
...@@ -2054,13 +2064,17 @@ const makeWsRpcLayer = (...@@ -2054,13 +2064,17 @@ const makeWsRpcLayer = (
2054 },2064 },
2055 ),2065 ),
2056 [WS_METHODS.scheduledTasksList]: (_input) =>2066 [WS_METHODS.scheduledTasksList]: (_input) =>
2057 observeRpcEffect(WS_METHODS.scheduledTasksList, scheduledTasks.list(), {2067 observeRpcEffect(
2058 "rpc.aggregate": "scheduledTasks",2068 WS_METHODS.scheduledTasksList,
2059 }),2069 scheduledTasks.list().pipe(Effect.map(withVisibleWebhookUrls)),
2070 { "rpc.aggregate": "scheduledTasks" },
2071 ),
2060 [WS_METHODS.scheduledTasksSubscribe]: (_input) =>2072 [WS_METHODS.scheduledTasksSubscribe]: (_input) =>
2061 observeRpcStream(WS_METHODS.scheduledTasksSubscribe, scheduledTasks.subscribeList(), {2073 observeRpcStream(
2062 "rpc.aggregate": "scheduledTasks",2074 WS_METHODS.scheduledTasksSubscribe,
2063 }),2075 scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)),
2076 { "rpc.aggregate": "scheduledTasks" },
2077 ),
2064 [WS_METHODS.scheduledTasksUpsert]: (input) =>2078 [WS_METHODS.scheduledTasksUpsert]: (input) =>
2065 observeRpcEffect(WS_METHODS.scheduledTasksUpsert, scheduledTasks.upsert(input), {2079 observeRpcEffect(WS_METHODS.scheduledTasksUpsert, scheduledTasks.upsert(input), {
2066 "rpc.aggregate": "scheduledTasks",2080 "rpc.aggregate": "scheduledTasks",
...@@ -2080,6 +2094,24 @@ const makeWsRpcLayer = (...@@ -2080,6 +2094,24 @@ const makeWsRpcLayer = (
2080 "rpc.aggregate": "scheduledTasks",2094 "rpc.aggregate": "scheduledTasks",
2081 "scheduled_task.id": input.id,2095 "scheduled_task.id": input.id,
2082 }),2096 }),
2097 [WS_METHODS.scheduledTasksRotateWebhookToken]: (input) =>
2098 observeRpcEffect(
2099 WS_METHODS.scheduledTasksRotateWebhookToken,
2100 scheduledTasks.rotateWebhookToken(input),
2101 { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id },
2102 ),
2103 [WS_METHODS.scheduledTasksListWebhookDeliveries]: (input) =>
2104 observeRpcEffect(
2105 WS_METHODS.scheduledTasksListWebhookDeliveries,
2106 scheduledTasks.listWebhookDeliveries(input),
2107 { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id },
2108 ),
2109 [WS_METHODS.scheduledTasksGetWebhookDelivery]: (input) =>
2110 observeRpcEffect(
2111 WS_METHODS.scheduledTasksGetWebhookDelivery,
2112 scheduledTasks.getWebhookDelivery(input),
2113 { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id },
2114 ),
2083 [WS_METHODS.serverProbe]: (_input) =>2115 [WS_METHODS.serverProbe]: (_input) =>
2084 observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), {2116 observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), {
2085 "rpc.aggregate": "server",2117 "rpc.aggregate": "server",
apps/web/src/components/chat/ThreadAutomationsPanel.tsx+6-1
...@@ -194,7 +194,12 @@ export function ThreadAutomationsPanel(props: {...@@ -194,7 +194,12 @@ export function ThreadAutomationsPanel(props: {
194 variant="ghost"194 variant="ghost"
195 part="icon"195 part="icon"
196 aria-label={`Run ${task.title} now`}196 aria-label={`Run ${task.title} now`}
197 disabled={busyTaskId !== null || task.lastRunStatus === "running"}197 // A webhook task runs from its URL; there is no request to run it with.
198 disabled={
199 busyTaskId !== null ||
200 task.lastRunStatus === "running" ||
201 task.schedule.type === "webhook"
202 }
198 onClick={() => void runNow(task)}203 onClick={() => void runNow(task)}
199 >204 >
200 <PlayIcon className="size-3.5" />205 <PlayIcon className="size-3.5" />
apps/web/src/components/settings/ScheduledTasksSettings.tsx+11-3
...@@ -162,6 +162,7 @@ function scheduleFromDraft(draft: DraftState): ScheduledTaskSchedule {...@@ -162,6 +162,7 @@ function scheduleFromDraft(draft: DraftState): ScheduledTaskSchedule {
162}162}
163163
164export function scheduleLabel(schedule: ScheduledTaskSchedule): string {164export function scheduleLabel(schedule: ScheduledTaskSchedule): string {
165 if (schedule.type === "webhook") return "On webhook";
165 if (schedule.type === "interval") {166 if (schedule.type === "interval") {
166 const minutes = schedule.everyMs / 60_000;167 const minutes = schedule.everyMs / 60_000;
167 return Number.isInteger(minutes)168 return Number.isInteger(minutes)
...@@ -312,7 +313,7 @@ function ScheduledTaskEnvironmentSection({...@@ -312,7 +313,7 @@ function ScheduledTaskEnvironmentSection({
312 const linkedTask = tasks?.find((task) => task.id === taskId);313 const linkedTask = tasks?.find((task) => task.id === taskId);
313 const openedLink = useRef(false);314 const openedLink = useRef(false);
314 useEffect(() => {315 useEffect(() => {
315 if (!openedLink.current && linkedTask) {316 if (!openedLink.current && linkedTask && linkedTask.schedule.type !== "webhook") {
316 openedLink.current = true;317 openedLink.current = true;
317 onEdit(environment.environmentId, linkedTask);318 onEdit(environment.environmentId, linkedTask);
318 }319 }
...@@ -346,6 +347,12 @@ function ScheduledTaskEnvironmentSection({...@@ -346,6 +347,12 @@ function ScheduledTaskEnvironmentSection({
346 description="This task no longer exists or is outside the selected project scope."347 description="This task no longer exists or is outside the selected project scope."
347 role="status"348 role="status"
348 />349 />
350 ) : linkedTask?.schedule.type === "webhook" ? (
351 <SettingsRow
352 title="Webhook tasks can't be edited here yet"
353 description="Update T3 Code to edit this task."
354 role="status"
355 />
349 ) : null}356 ) : null}
350 {tasks.length === 0 ? (357 {tasks.length === 0 ? (
351 <SettingsRow358 <SettingsRow
...@@ -449,11 +456,12 @@ function ScheduledTaskRow({...@@ -449,11 +456,12 @@ function ScheduledTaskRow({
449 <MoreHorizontalIcon className="size-4" />456 <MoreHorizontalIcon className="size-4" />
450 </MenuTrigger>457 </MenuTrigger>
451 <MenuPopup align="end">458 <MenuPopup align="end">
452 <MenuItem onClick={onEdit}>459 {/* Webhook tasks are not editable here yet; saving would drop their URL. */}
460 <MenuItem onClick={onEdit} disabled={task.schedule.type === "webhook"}>
453 <PencilIcon />461 <PencilIcon />
454 Edit462 Edit
455 </MenuItem>463 </MenuItem>
456 <MenuItem onClick={() => void act("run")}>464 <MenuItem onClick={() => void act("run")} disabled={task.schedule.type === "webhook"}>
457 <PlayIcon />465 <PlayIcon />
458 Run now466 Run now
459 </MenuItem>467 </MenuItem>
packages/client-runtime/src/state/server.ts+14
...@@ -1285,6 +1285,20 @@ export function createServerEnvironmentAtoms<R, E>(...@@ -1285,6 +1285,20 @@ export function createServerEnvironmentAtoms<R, E>(
1285 label: "environment-data:server:scheduled-task:run-now",1285 label: "environment-data:server:scheduled-task:run-now",
1286 tag: WS_METHODS.scheduledTasksRunNow,1286 tag: WS_METHODS.scheduledTasksRunNow,
1287 }),1287 }),
1288 rotateScheduledTaskWebhookToken: createEnvironmentRpcCommand(runtime, {
1289 label: "environment-data:server:scheduled-task:rotate-webhook-token",
1290 tag: WS_METHODS.scheduledTasksRotateWebhookToken,
1291 scheduler: configScheduler,
1292 concurrency: configConcurrency,
1293 }),
1294 listScheduledTaskWebhookDeliveries: createEnvironmentRpcCommand(runtime, {
1295 label: "environment-data:server:scheduled-task:list-webhook-deliveries",
1296 tag: WS_METHODS.scheduledTasksListWebhookDeliveries,
1297 }),
1298 getScheduledTaskWebhookDelivery: createEnvironmentRpcCommand(runtime, {
1299 label: "environment-data:server:scheduled-task:get-webhook-delivery",
1300 tag: WS_METHODS.scheduledTasksGetWebhookDelivery,
1301 }),
1288 refreshUsageRates: createEnvironmentRpcCommand(runtime, {1302 refreshUsageRates: createEnvironmentRpcCommand(runtime, {
1289 label: "environment-data:server:refresh-usage-rates",1303 label: "environment-data:server:refresh-usage-rates",
1290 tag: WS_METHODS.serverRefreshUsageRates,1304 tag: WS_METHODS.serverRefreshUsageRates,
packages/contracts/src/environmentHttp.ts+27-1
...@@ -5,6 +5,7 @@ import * as HttpApi from "effect/http-api/HttpApi";...@@ -5,6 +5,7 @@ import * as HttpApi from "effect/http-api/HttpApi";
5import * as HttpApiEndpoint from "effect/http-api/HttpApiEndpoint";5import * as HttpApiEndpoint from "effect/http-api/HttpApiEndpoint";
6import * as HttpApiGroup from "effect/http-api/HttpApiGroup";6import * as HttpApiGroup from "effect/http-api/HttpApiGroup";
7import * as HttpApiMiddleware from "effect/http-api/HttpApiMiddleware";7import * as HttpApiMiddleware from "effect/http-api/HttpApiMiddleware";
8import * as HttpApiSchema from "effect/http-api/HttpApiSchema";
8import * as HttpServerRespondable from "effect/http/HttpServerRespondable";9import * as HttpServerRespondable from "effect/http/HttpServerRespondable";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";10import * as HttpServerResponse from "effect/http/HttpServerResponse";
1011
...@@ -651,10 +652,35 @@ class EnvironmentConnectHttpApi extends HttpApiGroup.make("connect")...@@ -651,10 +652,35 @@ class EnvironmentConnectHttpApi extends HttpApiGroup.make("connect")
651 }),652 }),
652 ) {}653 ) {}
653654
655/**
656 * Public entry point for webhook tasks. Unauthenticated by design: the token
657 * in the path, and an optional body signature, are the credential. The handler
658 * reads the raw body itself so a signature is checked over the exact bytes.
659 */
660const WebhookParams = Schema.Struct({
661 hookId: TrimmedNonEmptyString,
662 token: TrimmedNonEmptyString,
663});
664const WebhookAccepted = Schema.Struct({ deliveryId: TrimmedNonEmptyString }).pipe(
665 HttpApiSchema.status(202),
666);
667const webhookEndpoint = {
668 params: WebhookParams,
669 success: WebhookAccepted,
670} as const;
671const WEBHOOK_PATH = "/api/hooks/:hookId/:token";
672
673class EnvironmentWebhooksHttpApi extends HttpApiGroup.make("webhooks")
674 .add(HttpApiEndpoint.post("webhookPost", WEBHOOK_PATH, webhookEndpoint))
675 .add(HttpApiEndpoint.put("webhookPut", WEBHOOK_PATH, webhookEndpoint))
676 .add(HttpApiEndpoint.patch("webhookPatch", WEBHOOK_PATH, webhookEndpoint))
677 .add(HttpApiEndpoint.get("webhookGet", WEBHOOK_PATH, webhookEndpoint)) {}
678
654export class EnvironmentHttpApi extends HttpApi.make("environment")679export class EnvironmentHttpApi extends HttpApi.make("environment")
655 .add(EnvironmentMetadataHttpApi)680 .add(EnvironmentMetadataHttpApi)
656 .add(EnvironmentAuthHttpApi)681 .add(EnvironmentAuthHttpApi)
657 .add(EnvironmentOrchestrationHttpApi)682 .add(EnvironmentOrchestrationHttpApi)
658 .add(EnvironmentPullRequestsHttpApi)683 .add(EnvironmentPullRequestsHttpApi)
659 .add(EnvironmentProjectsHttpApi)684 .add(EnvironmentProjectsHttpApi)
660 .add(EnvironmentConnectHttpApi) {}685 .add(EnvironmentConnectHttpApi)
686 .add(EnvironmentWebhooksHttpApi) {}
packages/contracts/src/orchestratorMcp.ts+3-1
...@@ -64,7 +64,7 @@ const OrchestratorMcpSchedule = Schema.Union([...@@ -64,7 +64,7 @@ const OrchestratorMcpSchedule = Schema.Union([
64 OrchestratorMcpScheduleFromJsonString,64 OrchestratorMcpScheduleFromJsonString,
65]).annotate({65]).annotate({
66 description:66 description:
67 "Recurring schedule object: {type:'interval', everyMs} or {type:'fixed_time', timeOfDay, weekdays?}. Never stringify it unless the provider requires the compatibility form.",67 "Trigger object: {type:'interval', everyMs}, {type:'fixed_time', timeOfDay, weekdays?}, or {type:'webhook'} to run on each request to a generated URL. Never stringify it unless the provider requires the compatibility form.",
68});68});
6969
70/**70/**
...@@ -541,6 +541,8 @@ export const OrchestratorMcpScheduledTask = Schema.Struct({...@@ -541,6 +541,8 @@ export const OrchestratorMcpScheduledTask = Schema.Struct({
541 schedule: ScheduledTaskSchedule,541 schedule: ScheduledTaskSchedule,
542 nextRunAt: Schema.NullOr(IsoDateTime),542 nextRunAt: Schema.NullOr(IsoDateTime),
543 lastRunStatus: ScheduledTaskRunStatus,543 lastRunStatus: ScheduledTaskRunStatus,
544 /** For webhook tasks: the public T3 Connect URL, or the environment-relative path when the environment is not linked. */
545 webhookUrl: Schema.optional(Schema.String),
544});546});
545export type OrchestratorMcpScheduledTask = typeof OrchestratorMcpScheduledTask.Type;547export type OrchestratorMcpScheduledTask = typeof OrchestratorMcpScheduledTask.Type;
546548
packages/contracts/src/rpc.ts+38
...@@ -310,6 +310,11 @@ import {...@@ -310,6 +310,11 @@ import {
310 ScheduledTaskListInput,310 ScheduledTaskListInput,
311 ScheduledTaskListResult,311 ScheduledTaskListResult,
312 ScheduledTaskRunNowInput,312 ScheduledTaskRunNowInput,
313 ScheduledTaskRotateWebhookTokenInput,
314 ScheduledTaskListWebhookDeliveriesInput,
315 ScheduledTaskListWebhookDeliveriesResult,
316 ScheduledTaskGetWebhookDeliveryInput,
317 ScheduledTaskGetWebhookDeliveryResult,
313 ScheduledTaskRunNowResult,318 ScheduledTaskRunNowResult,
314 ScheduledTaskSetEnabledInput,319 ScheduledTaskSetEnabledInput,
315 ScheduledTaskUpsertInput,320 ScheduledTaskUpsertInput,
...@@ -474,6 +479,9 @@ export const WS_METHODS = {...@@ -474,6 +479,9 @@ export const WS_METHODS = {
474 scheduledTasksSetEnabled: "scheduledTasks.setEnabled",479 scheduledTasksSetEnabled: "scheduledTasks.setEnabled",
475 scheduledTasksDelete: "scheduledTasks.delete",480 scheduledTasksDelete: "scheduledTasks.delete",
476 scheduledTasksRunNow: "scheduledTasks.runNow",481 scheduledTasksRunNow: "scheduledTasks.runNow",
482 scheduledTasksRotateWebhookToken: "scheduledTasks.rotateWebhookToken",
483 scheduledTasksListWebhookDeliveries: "scheduledTasks.listWebhookDeliveries",
484 scheduledTasksGetWebhookDelivery: "scheduledTasks.getWebhookDelivery",
477485
478 // Cloud environment methods486 // Cloud environment methods
479 cloudGetRelayClientStatus: "cloud.getRelayClientStatus",487 cloudGetRelayClientStatus: "cloud.getRelayClientStatus",
...@@ -1668,6 +1676,33 @@ const WsScheduledTasksRunNowRpc = Rpc.make(WS_METHODS.scheduledTasksRunNow, {...@@ -1668,6 +1676,33 @@ const WsScheduledTasksRunNowRpc = Rpc.make(WS_METHODS.scheduledTasksRunNow, {
1668 error: Schema.Union([ScheduledTaskError, EnvironmentAuthorizationError]),1676 error: Schema.Union([ScheduledTaskError, EnvironmentAuthorizationError]),
1669});1677});
16701678
1679const WsScheduledTasksRotateWebhookTokenRpc = Rpc.make(
1680 WS_METHODS.scheduledTasksRotateWebhookToken,
1681 {
1682 payload: ScheduledTaskRotateWebhookTokenInput,
1683 success: ScheduledTaskMutationResult,
1684 error: Schema.Union([ScheduledTaskError, EnvironmentAuthorizationError]),
1685 },
1686);
1687
1688const WsScheduledTasksListWebhookDeliveriesRpc = Rpc.make(
1689 WS_METHODS.scheduledTasksListWebhookDeliveries,
1690 {
1691 payload: ScheduledTaskListWebhookDeliveriesInput,
1692 success: ScheduledTaskListWebhookDeliveriesResult,
1693 error: Schema.Union([ScheduledTaskError, EnvironmentAuthorizationError]),
1694 },
1695);
1696
1697const WsScheduledTasksGetWebhookDeliveryRpc = Rpc.make(
1698 WS_METHODS.scheduledTasksGetWebhookDelivery,
1699 {
1700 payload: ScheduledTaskGetWebhookDeliveryInput,
1701 success: ScheduledTaskGetWebhookDeliveryResult,
1702 error: Schema.Union([ScheduledTaskError, EnvironmentAuthorizationError]),
1703 },
1704);
1705
1671const WsSubscribeAuthAccessRpc = Rpc.make(WS_METHODS.subscribeAuthAccess, {1706const WsSubscribeAuthAccessRpc = Rpc.make(WS_METHODS.subscribeAuthAccess, {
1672 payload: Schema.Struct({}),1707 payload: Schema.Struct({}),
1673 success: AuthAccessStreamEvent,1708 success: AuthAccessStreamEvent,
...@@ -1753,6 +1788,9 @@ export const WsRpcGroup = RpcGroup.make(...@@ -1753,6 +1788,9 @@ export const WsRpcGroup = RpcGroup.make(
1753 WsScheduledTasksSetEnabledRpc,1788 WsScheduledTasksSetEnabledRpc,
1754 WsScheduledTasksDeleteRpc,1789 WsScheduledTasksDeleteRpc,
1755 WsScheduledTasksRunNowRpc,1790 WsScheduledTasksRunNowRpc,
1791 WsScheduledTasksRotateWebhookTokenRpc,
1792 WsScheduledTasksListWebhookDeliveriesRpc,
1793 WsScheduledTasksGetWebhookDeliveryRpc,
1756 WsServerReportClientActivityRpc,1794 WsServerReportClientActivityRpc,
1757 WsServerReportHostPowerStateRpc,1795 WsServerReportHostPowerStateRpc,
1758 WsServerGetBackgroundPolicyRpc,1796 WsServerGetBackgroundPolicyRpc,
packages/contracts/src/scheduledTask.ts+138-3
...@@ -2,6 +2,7 @@ import * as Schema from "effect/Schema";...@@ -2,6 +2,7 @@ import * as Schema from "effect/Schema";
22
3import {3import {
4 CommandId,4 CommandId,
5 ForwardCompatibleArray,
5 IsoDateTime,6 IsoDateTime,
6 ProjectId,7 ProjectId,
7 ScheduledTaskId,8 ScheduledTaskId,
...@@ -54,6 +55,55 @@ const ScheduledTaskFixedTimeSchedule = Schema.Struct({...@@ -54,6 +55,55 @@ const ScheduledTaskFixedTimeSchedule = Schema.Struct({
54 description: "Run at a fixed local wall-clock time on selected weekdays.",55 description: "Run at a fixed local wall-clock time on selected weekdays.",
55});56});
5657
58const ScheduledTaskWebhookSignatureFields = {
59 header: TrimmedNonEmptyString.annotate({
60 description: "Request header carrying the signature, such as x-hub-signature-256.",
61 }),
62 encoding: Schema.Literals(["hex", "base64"]).annotate({
63 description: "How the HMAC-SHA256 digest is encoded in the header.",
64 }),
65 prefix: Schema.String.annotate({
66 description: "Text before the digest in the header value, such as 'sha256='. Empty for none.",
67 }),
68};
69
70/** HMAC-SHA256 over the raw request body. The secret is never part of the read model. */
71export const ScheduledTaskWebhookSignature = Schema.Struct(
72 ScheduledTaskWebhookSignatureFields,
73).annotate({ description: "Optional HMAC-SHA256 signature check over the raw request body." });
74export type ScheduledTaskWebhookSignature = typeof ScheduledTaskWebhookSignature.Type;
75
76const ScheduledTaskWebhookSchedule = Schema.Struct({
77 type: Schema.Literal("webhook").annotate({
78 description: "Run when the task's webhook URL receives a request.",
79 }),
80 signature: Schema.NullOr(ScheduledTaskWebhookSignature),
81}).annotate({
82 description:
83 "Run on each request to the task's webhook URL. The prompt may use {{body.path}}, {{headers.name}}, {{query.name}}, {{body}} and {{request}} placeholders.",
84});
85
86const ScheduledTaskUpsertWebhookSchedule = Schema.Struct({
87 type: Schema.Literal("webhook").annotate({
88 description: "Run when the task's webhook URL receives a request.",
89 }),
90 signature: Schema.optional(
91 Schema.NullOr(
92 Schema.Struct({
93 ...ScheduledTaskWebhookSignatureFields,
94 secret: Schema.optional(TrimmedNonEmptyString).annotate({
95 description: "Shared signing secret. Omit to keep the stored secret.",
96 }),
97 }),
98 ),
99 ).annotate({
100 description: "Signature check; omit or null to accept requests by URL token only.",
101 }),
102}).annotate({
103 description:
104 "Run on each request to the task's webhook URL. The prompt may use {{body.path}}, {{headers.name}}, {{query.name}}, {{body}} and {{request}} placeholders.",
105});
106
57/**107/**
58 * Read model for persisted schedules. Keep accepting legacy sub-minute rows so108 * Read model for persisted schedules. Keep accepting legacy sub-minute rows so
59 * users can list, disable, edit, or delete them after the write minimum changes.109 * users can list, disable, edit, or delete them after the write minimum changes.
...@@ -61,9 +111,10 @@ const ScheduledTaskFixedTimeSchedule = Schema.Struct({...@@ -61,9 +111,10 @@ const ScheduledTaskFixedTimeSchedule = Schema.Struct({
61export const ScheduledTaskSchedule = Schema.Union([111export const ScheduledTaskSchedule = Schema.Union([
62 ScheduledTaskIntervalSchedule,112 ScheduledTaskIntervalSchedule,
63 ScheduledTaskFixedTimeSchedule,113 ScheduledTaskFixedTimeSchedule,
114 ScheduledTaskWebhookSchedule,
64]).annotate({115]).annotate({
65 description:116 description:
66 "Structured recurring schedule. Pass an object with type 'interval' or 'fixed_time'.",117 "Structured trigger. Pass an object with type 'interval', 'fixed_time' or 'webhook'.",
67});118});
68export type ScheduledTaskSchedule = typeof ScheduledTaskSchedule.Type;119export type ScheduledTaskSchedule = typeof ScheduledTaskSchedule.Type;
69120
...@@ -82,14 +133,25 @@ export const ScheduledTaskUpsertSchedule = Schema.Union([...@@ -82,14 +133,25 @@ export const ScheduledTaskUpsertSchedule = Schema.Union([
82 description: "Run repeatedly after a fixed number of milliseconds.",133 description: "Run repeatedly after a fixed number of milliseconds.",
83 }),134 }),
84 ScheduledTaskFixedTimeSchedule,135 ScheduledTaskFixedTimeSchedule,
136 ScheduledTaskUpsertWebhookSchedule,
85]).annotate({137]).annotate({
86 description: "Writable recurring schedule. Pass an object with type 'interval' or 'fixed_time'.",138 description: "Writable trigger. Pass an object with type 'interval', 'fixed_time' or 'webhook'.",
87});139});
88export type ScheduledTaskUpsertSchedule = typeof ScheduledTaskUpsertSchedule.Type;140export type ScheduledTaskUpsertSchedule = typeof ScheduledTaskUpsertSchedule.Type;
89141
90export const ScheduledTaskRunStatus = Schema.Literals(["never", "running", "succeeded", "failed"]);142export const ScheduledTaskRunStatus = Schema.Literals(["never", "running", "succeeded", "failed"]);
91export type ScheduledTaskRunStatus = typeof ScheduledTaskRunStatus.Type;143export type ScheduledTaskRunStatus = typeof ScheduledTaskRunStatus.Type;
92144
145/** Where a webhook task receives requests. Present only on webhook tasks. */
146export const ScheduledTaskWebhookEndpoint = Schema.Struct({
147 /** Environment-relative path including the secret token; works on any origin that reaches the environment. */
148 path: TrimmedNonEmptyString,
149 /** Public T3 Connect URL, or null when the environment is not linked to T3 Connect. */
150 url: Schema.NullOr(TrimmedNonEmptyString),
151 hasSecret: Schema.Boolean,
152});
153export type ScheduledTaskWebhookEndpoint = typeof ScheduledTaskWebhookEndpoint.Type;
154
93export const ScheduledTask = Schema.Struct({155export const ScheduledTask = Schema.Struct({
94 id: ScheduledTaskId,156 id: ScheduledTaskId,
95 title: TrimmedNonEmptyString,157 title: TrimmedNonEmptyString,
...@@ -111,6 +173,7 @@ export const ScheduledTask = Schema.Struct({...@@ -111,6 +173,7 @@ export const ScheduledTask = Schema.Struct({
111 lastRunStatus: ScheduledTaskRunStatus,173 lastRunStatus: ScheduledTaskRunStatus,
112 lastRunError: Schema.NullOr(Schema.String),174 lastRunError: Schema.NullOr(Schema.String),
113 runCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)),175 runCount: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)),
176 webhook: Schema.optional(ScheduledTaskWebhookEndpoint),
114});177});
115export type ScheduledTask = typeof ScheduledTask.Type;178export type ScheduledTask = typeof ScheduledTask.Type;
116179
...@@ -118,7 +181,9 @@ export const ScheduledTaskListInput = Schema.Struct({});...@@ -118,7 +181,9 @@ export const ScheduledTaskListInput = Schema.Struct({});
118export type ScheduledTaskListInput = typeof ScheduledTaskListInput.Type;181export type ScheduledTaskListInput = typeof ScheduledTaskListInput.Type;
119182
120export const ScheduledTaskListResult = Schema.Struct({183export const ScheduledTaskListResult = Schema.Struct({
121 tasks: Schema.Array(ScheduledTask),184 // Trigger types grow over time; a client must not lose the whole list over
185 // one task it cannot decode.
186 tasks: ForwardCompatibleArray(ScheduledTask),
122});187});
123export type ScheduledTaskListResult = typeof ScheduledTaskListResult.Type;188export type ScheduledTaskListResult = typeof ScheduledTaskListResult.Type;
124189
...@@ -160,6 +225,76 @@ export const ScheduledTaskRunNowInput = Schema.Struct({...@@ -160,6 +225,76 @@ export const ScheduledTaskRunNowInput = Schema.Struct({
160});225});
161export type ScheduledTaskRunNowInput = typeof ScheduledTaskRunNowInput.Type;226export type ScheduledTaskRunNowInput = typeof ScheduledTaskRunNowInput.Type;
162227
228export const ScheduledTaskRotateWebhookTokenInput = Schema.Struct({
229 id: ScheduledTaskId,
230});
231export type ScheduledTaskRotateWebhookTokenInput = typeof ScheduledTaskRotateWebhookTokenInput.Type;
232
233export const ScheduledTaskWebhookDeliveryId = TrimmedNonEmptyString.pipe(
234 Schema.brand("ScheduledTaskWebhookDeliveryId"),
235);
236export type ScheduledTaskWebhookDeliveryId = typeof ScheduledTaskWebhookDeliveryId.Type;
237
238export const ScheduledTaskWebhookDeliveryOutcome = Schema.Literals([
239 "accepted",
240 "dispatch_failed",
241 "rejected_signature",
242 "disabled",
243 "rate_limited",
244]);
245export type ScheduledTaskWebhookDeliveryOutcome = typeof ScheduledTaskWebhookDeliveryOutcome.Type;
246
247export const ScheduledTaskWebhookDeliverySummary = Schema.Struct({
248 id: ScheduledTaskWebhookDeliveryId,
249 taskId: ScheduledTaskId,
250 receivedAt: IsoDateTime,
251 method: TrimmedNonEmptyString,
252 contentType: Schema.NullOr(Schema.String),
253 bodyBytes: Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)),
254 outcome: ScheduledTaskWebhookDeliveryOutcome,
255 /** True when a configured signature matched; false when none was configured. */
256 signatureVerified: Schema.Boolean,
257 /** Template placeholders that had no value in this request and rendered empty. */
258 missingFields: Schema.Array(Schema.String),
259 error: Schema.NullOr(Schema.String),
260});
261export type ScheduledTaskWebhookDeliverySummary = typeof ScheduledTaskWebhookDeliverySummary.Type;
262
263export const ScheduledTaskWebhookDelivery = Schema.Struct({
264 ...ScheduledTaskWebhookDeliverySummary.fields,
265 query: Schema.String,
266 headers: Schema.Record(Schema.String, Schema.String),
267 /** Body as UTF-8 text, cut at the log limit; see bodyTruncated. */
268 body: Schema.String,
269 bodyTruncated: Schema.Boolean,
270 renderedPrompt: Schema.NullOr(Schema.String),
271});
272export type ScheduledTaskWebhookDelivery = typeof ScheduledTaskWebhookDelivery.Type;
273
274export const ScheduledTaskListWebhookDeliveriesInput = Schema.Struct({
275 id: ScheduledTaskId,
276});
277export type ScheduledTaskListWebhookDeliveriesInput =
278 typeof ScheduledTaskListWebhookDeliveriesInput.Type;
279
280export const ScheduledTaskListWebhookDeliveriesResult = Schema.Struct({
281 deliveries: Schema.Array(ScheduledTaskWebhookDeliverySummary),
282});
283export type ScheduledTaskListWebhookDeliveriesResult =
284 typeof ScheduledTaskListWebhookDeliveriesResult.Type;
285
286export const ScheduledTaskGetWebhookDeliveryInput = Schema.Struct({
287 id: ScheduledTaskId,
288 deliveryId: ScheduledTaskWebhookDeliveryId,
289});
290export type ScheduledTaskGetWebhookDeliveryInput = typeof ScheduledTaskGetWebhookDeliveryInput.Type;
291
292export const ScheduledTaskGetWebhookDeliveryResult = Schema.Struct({
293 delivery: ScheduledTaskWebhookDelivery,
294});
295export type ScheduledTaskGetWebhookDeliveryResult =
296 typeof ScheduledTaskGetWebhookDeliveryResult.Type;
297
163export const ScheduledTaskMutationResult = Schema.Struct({298export const ScheduledTaskMutationResult = Schema.Struct({
164 task: ScheduledTask,299 task: ScheduledTask,
165});300});