1// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no createHmac.
2import * as NodeCrypto from "node:crypto";
3
4import * as NodePlatformCrypto from "@effect/platform-node/NodeCrypto";
5import { assert, it } from "@effect/vitest";
6import { ScheduledTaskUpsertInput, SecretRequestError } from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import * as Deferred from "effect/Deferred";
9import * as Effect from "effect/Effect";
10import * as Fiber from "effect/Fiber";
11import * as Layer from "effect/Layer";
12import * as Metric from "effect/Metric";
13import * as Queue from "effect/Queue";
14import * as EffectScheduler from "effect/Scheduler";
15import * as Schema from "effect/Schema";
16import * as TestClock from "effect/testing/TestClock";
17
18import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
19import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
20import * as SecretRequests from "../secrets/SecretRequests.ts";
21import * as SqlitePersistence from "../persistence/Sqlite.ts";
22import * as Scheduler from "../scheduling/Scheduler.ts";
23import * as ScheduledTaskService from "./ScheduledTaskService.ts";
24
25const decodeUpsertInput = Schema.decodeUnknownEffect(ScheduledTaskUpsertInput);
26
27type LaunchInput = ThreadLaunchService.ThreadLaunchInput;
28
29const webhookTaskInput = (overrides: Record<string, unknown> = {}) =>
30 decodeUpsertInput({
31 id: "scheduled-task:hook",
32 title: "Review PRs",
33 prompt: "Review this PR: {{body.pull_request.url}}",
34 enabled: true,
35 schedule: { type: "webhook" },
36 projectId: "project-webhook",
37 workspaceStrategy: { type: "root" },
38 modelSelection: { instanceId: "codex", model: "gpt-5.4" },
39 runtimeMode: "full-access",
40 interactionMode: "default",
41 ...overrides,
42 });
43
44const pullRequestBody = new TextEncoder().encode(
45 JSON.stringify({ pull_request: { url: "https://github.com/org/repo/pull/45" } }),
46);
47
48const requestFor = (
49 task: { readonly id: string; readonly webhook?: { readonly path: string } | undefined },
50 overrides: Partial<ScheduledTaskService.WebhookTriggerRequest> = {},
51): ScheduledTaskService.WebhookTriggerRequest => ({
52 hookId: task.id,
53 token: task.webhook?.path.split("/").at(-1) ?? "",
54 method: "POST",
55 path: `/api/hooks/${task.id}`,
56 query: "",
57 headers: { "content-type": "application/json" },
58 body: pullRequestBody,
59 bodyText: new TextDecoder().decode(pullRequestBody),
60 ...overrides,
61});
62
63/**
64 * Runs `body` against a service whose launches are pushed to `launches`;
65 * `gate`, when given, holds each launch until the test releases it.
66 */
67const withService = <A, E>(
68 body: (input: {
69 readonly service: ScheduledTaskService.ScheduledTaskService["Service"];
70 readonly launches: Queue.Queue<LaunchInput>;
71 /** Secrets the user entered for an agent, by ref; consuming one removes it. */
72 readonly secretsByRef: Map<string, string>;
73 }) => Effect.Effect<A, E, never>,
74 options: { readonly gate?: Deferred.Deferred<void>; readonly relayHookBaseUrl?: string } = {},
75) =>
76 Effect.gen(function* () {
77 const launches = yield* Queue.unbounded<LaunchInput>();
78 const secretsByRef = new Map<string, string>();
79 const layerDependencies = Layer.mergeAll(
80 NodePlatformCrypto.layer,
81 Scheduler.layer,
82 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
83 launch: (input) =>
84 Queue.offer(launches, input).pipe(
85 Effect.andThen(options.gate ? Deferred.await(options.gate) : Effect.void),
86 Effect.as({ threadId: "thread-1", resumed: false } as never),
87 ),
88 }),
89 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
90 Layer.mock(SecretRequests.SecretRequests)({
91 consume: ({ ref }) => {
92 const value = secretsByRef.get(ref);
93 secretsByRef.delete(ref);
94 return value === undefined
95 ? Effect.fail(new SecretRequestError({ reason: "ref_unavailable" }))
96 : Effect.succeed(value);
97 },
98 }),
99 Layer.succeed(
100 ScheduledTaskService.ScheduledTaskWebhookOrigin,
101 Effect.succeed({ relayHookBaseUrl: options.relayHookBaseUrl ?? null }),
102 ),
103 );
104 return yield* Effect.gen(function* () {
105 const service = yield* ScheduledTaskService.ScheduledTaskService;
106 return yield* body({ service, launches, secretsByRef });
107 }).pipe(Effect.provide(ScheduledTaskService.layer.pipe(Layer.provide(layerDependencies))));
108 }).pipe(Effect.provide(SqlitePersistence.layerMemory));
109
110it.effect("dispatches exactly the rendered prompt and logs the delivery", () =>
111 withService(({ service, launches }) =>
112 Effect.gen(function* () {
113 const { task } = yield* service.upsert(yield* webhookTaskInput());
114 assert.equal(task.nextRunAt, null);
115 assert.isDefined(task.webhook);
116 assert.isTrue(task.webhook!.path.startsWith("/api/hooks/scheduled-task%3Ahook/"));
117 // Not linked to T3 Connect in tests.
118 assert.equal(task.webhook!.url, null);
119
120 const result = yield* service.triggerWebhook(requestFor(task));
121 assert.equal(result._tag, "accepted");
122 const launched = yield* Queue.take(launches);
123 assert.equal(
124 launched.initialMessage?.text,
125 "Review this PR: https://github.com/org/repo/pull/45",
126 );
127 assert.equal(
128 launched.commandId,
129 `scheduled-task:${task.id}:webhook:${result._tag === "accepted" ? result.deliveryId : ""}`,
130 );
131
132 const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id });
133 assert.equal(deliveries.length, 1);
134 assert.equal(deliveries[0]?.outcome, "accepted");
135 const { delivery } = yield* service.getWebhookDelivery({
136 id: task.id,
137 deliveryId: deliveries[0]!.id,
138 });
139 assert.equal(delivery.body, new TextDecoder().decode(pullRequestBody));
140 assert.equal(delivery.renderedPrompt, launched.initialMessage?.text);
141 }),
142 ),
143);
144
145it("builds the relay hook URL from the managed tunnel's key, never the environment id", () => {
146 const relayUrl = "https://relay.example.com/";
147 assert.equal(
148 ScheduledTaskService.relayHookBaseUrl({
149 relayUrl,
150 tunnelName: "t3coderelay-managedendpoint-dev-julius-0123456789abcdef",
151 }),
152 "https://relay.example.com/v1/hooks/0123456789abcdef",
153 );
154 for (const tunnelName of [undefined, "t3coderelay-managedendpoint", "x-0123456789ABCDEF"]) {
155 assert.isNull(ScheduledTaskService.relayHookBaseUrl({ relayUrl, tunnelName }));
156 }
157});
158
159it.effect("gives webhook tasks a relay URL when the environment has a managed tunnel", () =>
160 withService(
161 ({ service }) =>
162 Effect.gen(function* () {
163 const { task } = yield* service.upsert(yield* webhookTaskInput());
164 const token = task.webhook!.path.split("/").at(-1);
165 assert.equal(
166 task.webhook?.url,
167 `https://relay.example.com/v1/hooks/0123456789abcdef/scheduled-task%3Ahook/${token}`,
168 );
169 }),
170 { relayHookBaseUrl: "https://relay.example.com/v1/hooks/0123456789abcdef" },
171 ),
172);
173
174it.effect("answers not found for a wrong token or unknown hook without logging", () =>
175 withService(({ service }) =>
176 Effect.gen(function* () {
177 const { task } = yield* service.upsert(yield* webhookTaskInput());
178 const wrongToken = yield* service.triggerWebhook(requestFor(task, { token: "nope" }));
179 assert.equal(wrongToken._tag, "not_found");
180 const unknown = yield* service.triggerWebhook(requestFor(task, { hookId: "missing" }));
181 assert.equal(unknown._tag, "not_found");
182 assert.equal((yield* service.listWebhookDeliveries({ id: task.id })).deliveries.length, 0);
183 }),
184 ),
185);
186
187it.effect("rotating the token retires the old URL and saving keeps it", () =>
188 withService(({ service }) =>
189 Effect.gen(function* () {
190 const input = yield* webhookTaskInput();
191 const { task } = yield* service.upsert(input);
192 const saved = yield* service.upsert(yield* webhookTaskInput({ title: "Renamed" }));
193 assert.equal(saved.task.webhook?.path, task.webhook?.path);
194
195 const rotated = yield* service.rotateWebhookToken({ id: task.id });
196 assert.notEqual(rotated.task.webhook?.path, task.webhook?.path);
197 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "not_found");
198 assert.equal((yield* service.triggerWebhook(requestFor(rotated.task)))._tag, "accepted");
199 }),
200 ),
201);
202
203it.effect("checks the configured signature and keeps the secret write-only", () =>
204 withService(({ service }) =>
205 Effect.gen(function* () {
206 const { task } = yield* service.upsert(
207 yield* webhookTaskInput({
208 schedule: {
209 type: "webhook",
210 signature: {
211 header: "X-Hub-Signature-256",
212 encoding: "hex",
213 prefix: "sha256=",
214 secret: "s3cret",
215 },
216 },
217 }),
218 );
219 assert.deepEqual(task.schedule, {
220 type: "webhook",
221 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" },
222 maxDeliveryAgeMinutes: null,
223 });
224 assert.isTrue(task.webhook?.hasSecret);
225
226 const unsigned = yield* service.triggerWebhook(requestFor(task));
227 assert.equal(unsigned._tag, "rejected_signature");
228
229 const signature = `sha256=${NodeCrypto.createHmac("sha256", "s3cret").update(pullRequestBody).digest("hex")}`;
230 const signed = yield* service.triggerWebhook(
231 requestFor(task, {
232 headers: { "content-type": "application/json", "x-hub-signature-256": signature },
233 }),
234 );
235 assert.equal(signed._tag, "accepted");
236
237 // Saving without a secret keeps the stored one.
238 const resaved = yield* service.upsert(
239 yield* webhookTaskInput({
240 schedule: {
241 type: "webhook",
242 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" },
243 },
244 }),
245 );
246 assert.isTrue(resaved.task.webhook?.hasSecret);
247
248 const outcomes = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries.map(
249 (delivery) => [delivery.outcome, delivery.signatureVerified],
250 );
251 assert.deepEqual(outcomes.toSorted(), [
252 ["accepted", true],
253 ["rejected_signature", false],
254 ]);
255 }),
256 ),
257);
258
259it.effect("logs but does not run deliveries to a disabled task, and refuses run now", () =>
260 withService(({ service, launches }) =>
261 Effect.gen(function* () {
262 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
263 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "disabled");
264 assert.equal(yield* Queue.size(launches), 0);
265 const runNow = yield* service.runNow({ id: task.id }).pipe(Effect.flip);
266 assert.equal(runNow.message, "Webhook tasks run when their URL receives a request.");
267 }),
268 ),
269);
270
271it.effect("queues a burst of deliveries instead of dropping them", () =>
272 Effect.gen(function* () {
273 const gate = yield* Deferred.make<void>();
274 yield* withService(
275 ({ service, launches }) =>
276 Effect.gen(function* () {
277 const { task } = yield* service.upsert(yield* webhookTaskInput());
278 const results = yield* Effect.forEach([1, 2, 3], () =>
279 service.triggerWebhook(requestFor(task)),
280 );
281 assert.deepEqual(
282 results.map((result) => result._tag),
283 ["accepted", "accepted", "accepted"],
284 );
285 // Only the first is dispatching; the others wait their turn.
286 yield* Queue.take(launches);
287 yield* Deferred.succeed(gate, undefined);
288 const rest = yield* Effect.all([Queue.take(launches), Queue.take(launches)]);
289 assert.equal(new Set(rest.map((launch) => launch.commandId)).size, 2);
290 }),
291 { gate },
292 );
293 }),
294);
295
296it.effect("rate limits a hook past 60 deliveries a minute", () =>
297 withService(({ service }) =>
298 Effect.gen(function* () {
299 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
300 const results = yield* Effect.forEach(Array.from({ length: 61 }), () =>
301 service.triggerWebhook(requestFor(task)),
302 );
303 assert.equal(results.at(-2)?._tag, "disabled");
304 assert.equal(results.at(-1)?._tag, "rate_limited");
305 // Further rejections in the same window are counted, not logged.
306 yield* Effect.forEach([1, 2, 3], () => service.triggerWebhook(requestFor(task)));
307 const outcomes = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries.map(
308 (delivery) => delivery.outcome,
309 );
310 assert.equal(outcomes.filter((outcome) => outcome === "rate_limited").length, 1);
311 }),
312 ),
313);
314
315it.effect("a save carrying a stale token cannot undo a rotation", () =>
316 withService(({ service }) =>
317 Effect.gen(function* () {
318 const { task } = yield* service.upsert(yield* webhookTaskInput());
319 const rotated = yield* service.rotateWebhookToken({ id: task.id });
320 // The editor was opened before the rotation and saves afterwards.
321 const saved = yield* service.upsert(yield* webhookTaskInput({ title: "Edited" }));
322 assert.equal(saved.task.webhook?.path, rotated.task.webhook?.path);
323 assert.equal((yield* service.triggerWebhook(requestFor(task)))._tag, "not_found");
324 }),
325 ),
326);
327
328it.effect("keeps the newest 50 deliveries when they share a timestamp", () =>
329 withService(({ service }) =>
330 Effect.gen(function* () {
331 // Paused, so each request is logged without starting a run. The test
332 // clock is frozen, so every delivery has the same received_at.
333 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
334 const ids = yield* Effect.forEach(Array.from({ length: 55 }), (_, index) =>
335 service.triggerWebhook(requestFor(task, { query: `n=${index}` })).pipe(Effect.as(index)),
336 );
337 const { deliveries } = yield* service.listWebhookDeliveries({ id: task.id });
338 assert.equal(deliveries.length, 50);
339 const first = yield* service.getWebhookDelivery({
340 id: task.id,
341 deliveryId: deliveries[0]!.id,
342 });
343 const last = yield* service.getWebhookDelivery({
344 id: task.id,
345 deliveryId: deliveries.at(-1)!.id,
346 });
347 assert.equal(first.delivery.query, `n=${ids.at(-1)}`);
348 assert.equal(last.delivery.query, "n=5");
349 }),
350 ),
351);
352
353it.effect("keeps credential headers and query values out of the delivery log", () =>
354 withService(({ service }) =>
355 Effect.gen(function* () {
356 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
357 yield* service.triggerWebhook(
358 requestFor(task, {
359 query: "page=2&api_key=k",
360 headers: {
361 "content-type": "application/json",
362 authorization: "Bearer sender-token",
363 "x-webhook-key": "k",
364 "x-github-event": "push",
365 },
366 }),
367 );
368 const [summary] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
369 const { delivery } = yield* service.getWebhookDelivery({
370 id: task.id,
371 deliveryId: summary!.id,
372 });
373 assert.equal(delivery.query, "page=2&api_key=[redacted]");
374 assert.equal(delivery.headers.authorization, "[redacted]");
375 assert.equal(delivery.headers["x-webhook-key"], "[redacted]");
376 assert.equal(delivery.headers["x-github-event"], "push");
377 }),
378 ),
379);
380
381const queuedDeliveryCases = [
382 {
383 change: "paused",
384 reason: "The task was paused before this delivery ran.",
385 apply: (service: ScheduledTaskService.ScheduledTaskService["Service"], id: string) =>
386 service.setEnabled({ id: id as never, enabled: false }),
387 },
388 {
389 change: "switched to an interval trigger",
390 reason: "The task's trigger changed before this delivery ran.",
391 apply: (service: ScheduledTaskService.ScheduledTaskService["Service"]) =>
392 webhookTaskInput({ schedule: { type: "interval", everyMs: 3_600_000 } }).pipe(
393 Effect.flatMap(service.upsert),
394 ),
395 },
396] as const;
397
398it.effect.each(queuedDeliveryCases)(
399 "a delivery queued behind a run does not start once the task is $change",
400 ({ reason, apply }) =>
401 Effect.gen(function* () {
402 const gate = yield* Deferred.make<void>();
403 yield* withService(
404 ({ service, launches }) =>
405 Effect.gen(function* () {
406 const { task } = yield* service.upsert(yield* webhookTaskInput());
407 yield* service.triggerWebhook(requestFor(task));
408 const queued = yield* service.triggerWebhook(requestFor(task));
409 yield* Queue.take(launches);
410 yield* apply(service, task.id);
411 yield* Deferred.succeed(gate, undefined);
412 // The queued delivery is marked failed instead of launching.
413 const deliveryId = queued._tag === "accepted" ? queued.deliveryId : undefined;
414 let delivery = (yield* service.getWebhookDelivery({
415 id: task.id,
416 deliveryId: deliveryId!,
417 })).delivery;
418 while (delivery.outcome === "accepted") {
419 yield* Effect.yieldNow;
420 delivery = (yield* service.getWebhookDelivery({
421 id: task.id,
422 deliveryId: deliveryId!,
423 })).delivery;
424 }
425 assert.equal(delivery.outcome, "dispatch_failed");
426 assert.equal(delivery.error, reason);
427 assert.equal(yield* Queue.size(launches), 0);
428 }),
429 { gate },
430 );
431 }),
432);
433
434it.effect("a save without a secret keeps a secret changed after it was read", () =>
435 withService(({ service }) =>
436 Effect.gen(function* () {
437 const signature = { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" };
438 const { task } = yield* service.upsert(
439 yield* webhookTaskInput({
440 schedule: { type: "webhook", signature: { ...signature, secret: "old" } },
441 }),
442 );
443 yield* service.upsert(
444 yield* webhookTaskInput({
445 schedule: { type: "webhook", signature: { ...signature, secret: "new" } },
446 }),
447 );
448 // A form opened before the change saves without sending a secret.
449 yield* service.upsert(
450 yield* webhookTaskInput({ title: "Edited", schedule: { type: "webhook", signature } }),
451 );
452 const sign = (secret: string) =>
453 `sha256=${NodeCrypto.createHmac("sha256", secret).update(pullRequestBody).digest("hex")}`;
454 const withSignature = (secret: string) =>
455 requestFor(task, {
456 headers: { "content-type": "application/json", "x-hub-signature-256": sign(secret) },
457 });
458 assert.equal(
459 (yield* service.triggerWebhook(withSignature("old")))._tag,
460 "rejected_signature",
461 );
462 assert.equal((yield* service.triggerWebhook(withSignature("new")))._tag, "accepted");
463 }),
464 ),
465);
466
467it.effect("caps deliveries waiting behind a stuck run", () =>
468 Effect.gen(function* () {
469 const gate = yield* Deferred.make<void>();
470 yield* withService(
471 ({ service, launches }) =>
472 Effect.gen(function* () {
473 const { task } = yield* service.upsert(yield* webhookTaskInput());
474 yield* service.triggerWebhook(requestFor(task));
475 yield* Queue.take(launches);
476 // The cap counts the running delivery too: 19 more wait, the next is refused.
477 const waiting = yield* Effect.forEach(Array.from({ length: 20 }), () =>
478 service.triggerWebhook(requestFor(task)),
479 );
480 assert.equal(waiting.filter((result) => result._tag === "accepted").length, 19);
481 assert.equal(waiting.at(-1)?._tag, "rate_limited");
482 // The refused request is not logged.
483 const logged = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
484 assert.equal(logged.length, 20);
485 yield* Deferred.succeed(gate, undefined);
486 }),
487 { gate },
488 );
489 }),
490);
491
492it.effect("logs a body's first 64 KiB by bytes, not characters", () =>
493 withService(({ service }) =>
494 Effect.gen(function* () {
495 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
496 // 30 000 three-byte characters: under 64 Ki characters, over 64 KiB.
497 const text = "界".repeat(30_000);
498 const body = new TextEncoder().encode(text);
499 yield* service.triggerWebhook(requestFor(task, { body, bodyText: text }));
500 const [summary] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
501 const { delivery } = yield* service.getWebhookDelivery({
502 id: task.id,
503 deliveryId: summary!.id,
504 });
505 assert.isTrue(delivery.bodyTruncated);
506 assert.isAtMost(new TextEncoder().encode(delivery.body).byteLength, 64 * 1024 + 3);
507 }),
508 ),
509);
510
511it.effect("counts the prompt as the provider does, without surrounding whitespace", () =>
512 withService(({ service, launches }) =>
513 Effect.gen(function* () {
514 const { task } = yield* service.upsert(yield* webhookTaskInput({ prompt: "{{body.text}}" }));
515 // Over the limit as sent, within it once the padding is trimmed.
516 const text = `{"text":"${" ".repeat(1_000)}${"x".repeat(119_990)}${"\\n".repeat(1_000)}"}`;
517 const result = yield* service.triggerWebhook(
518 requestFor(task, { body: new TextEncoder().encode(text), bodyText: text }),
519 );
520 assert.equal(result._tag, "accepted");
521 yield* Queue.take(launches);
522 }),
523 ),
524);
525
526it.effect("does not start a run when the filled-in prompt is too long", () =>
527 withService(({ service, launches }) =>
528 Effect.gen(function* () {
529 const { task } = yield* service.upsert(yield* webhookTaskInput({ prompt: "{{body}}" }));
530 const text = "x".repeat(200_000);
531 const body = new TextEncoder().encode(text);
532 const result = yield* service.triggerWebhook(requestFor(task, { body, bodyText: text }));
533 assert.equal(result._tag, "accepted");
534 assert.equal(yield* Queue.size(launches), 0);
535 const { delivery } = yield* service.getWebhookDelivery({
536 id: task.id,
537 deliveryId: result._tag === "accepted" ? result.deliveryId : ("" as never),
538 });
539 assert.equal(delivery.outcome, "dispatch_failed");
540 assert.equal(delivery.error, "The filled-in prompt is too long.");
541 assert.equal(delivery.renderedPrompt?.length, 64 * 1024);
542 // The queue slot was never taken: a normal delivery still runs.
543 const ok = yield* service.triggerWebhook(requestFor(task));
544 assert.equal(ok._tag, "accepted");
545 yield* Queue.take(launches);
546 }),
547 ),
548);
549
550it.effect("a held request already delivered directly runs only once", () =>
551 withService(({ service, launches }) =>
552 Effect.gen(function* () {
553 const { task } = yield* service.upsert(yield* webhookTaskInput());
554 const direct = yield* service.triggerWebhook(
555 requestFor(task, { relayDeliveryId: "relay-1" }),
556 );
557 const replayed = yield* service.triggerWebhook(
558 requestFor(task, { relayDeliveryId: "relay-1", receivedAt: "2026-10-04T10:00:00.000Z" }),
559 );
560 assert.equal(direct._tag, "accepted");
561 // Same delivery, answered the same way, but recorded as a duplicate.
562 assert.deepEqual(replayed, { ...direct, outcome: "duplicate" } as typeof replayed);
563 yield* Queue.take(launches);
564 const logged = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
565 assert.equal(logged.length, 1);
566 }),
567 ),
568);
569
570it.effect("a held request whose sender hung up mid-request still runs", () =>
571 withService(({ service, launches }) =>
572 Effect.gen(function* () {
573 const { task } = yield* service.upsert(yield* webhookTaskInput());
574 // The relay times out and the request fiber is interrupted. A small
575 // operation budget makes the request yield often, so stepping the
576 // interrupt one yield later each time lands it at every point in the
577 // request (one takes about 40 yields), then the relay retries.
578 for (let step = 0; step < 60; step++) {
579 const request = requestFor(task, { relayDeliveryId: `hung-up-${step}` });
580 const fiber = yield* service
581 .triggerWebhook(request)
582 .pipe(Effect.provideService(EffectScheduler.MaxOpsBeforeYield, 8), Effect.forkChild);
583 for (let yields = 0; yields < step; yields++) yield* Effect.yieldNow;
584 yield* Fiber.interrupt(fiber);
585 const retried = yield* service.triggerWebhook(request);
586 assert.equal(retried._tag, "accepted");
587 const deliveryId = retried._tag === "accepted" ? retried.deliveryId : undefined;
588 // Logged and run exactly once, whether or not the first attempt got through.
589 yield* service.getWebhookDelivery({ id: task.id, deliveryId: deliveryId! });
590 const launch = yield* Queue.take(launches);
591 assert.include(launch.commandId, `hung-up-${step}`);
592 // Keep each step in a fresh rate-limit window.
593 yield* TestClock.adjust("61 seconds");
594 }
595 assert.equal(yield* Queue.size(launches), 0);
596 }),
597 ),
598);
599
600it.effect("a held request runs once even after the log has trimmed it", () =>
601 withService(({ service, launches }) =>
602 Effect.gen(function* () {
603 const { task } = yield* service.upsert(yield* webhookTaskInput());
604 const first = yield* service.triggerWebhook(requestFor(task, { relayDeliveryId: "kept" }));
605 assert.equal(first._tag, "accepted");
606 yield* Queue.take(launches);
607 // Push the original row out of the 50-row delivery log.
608 const paused = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
609 yield* Effect.forEach(Array.from({ length: 55 }), () =>
610 service.triggerWebhook(requestFor(paused.task)),
611 );
612 yield* service.upsert(yield* webhookTaskInput());
613 const replay = yield* service.triggerWebhook(requestFor(task, { relayDeliveryId: "kept" }));
614 assert.equal(replay._tag, "accepted");
615 assert.equal(yield* Queue.size(launches), 0);
616 }),
617 ),
618);
619
620it.effect("a rate-limited held request can run on a later pass", () =>
621 withService(({ service, launches }) =>
622 Effect.gen(function* () {
623 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
624 // Spend the task's 60-a-minute budget.
625 yield* Effect.forEach(Array.from({ length: 60 }), () =>
626 service.triggerWebhook(requestFor(task)),
627 );
628 const limited = yield* service.triggerWebhook(requestFor(task, { relayDeliveryId: "later" }));
629 assert.equal(limited._tag, "rate_limited");
630 yield* service.upsert(yield* webhookTaskInput());
631 yield* TestClock.adjust("61 seconds");
632 const retried = yield* service.triggerWebhook(requestFor(task, { relayDeliveryId: "later" }));
633 assert.equal(retried._tag, "accepted");
634 yield* Queue.take(launches);
635 }),
636 ),
637);
638
639it.effect("logs a held request at the time the relay received it", () =>
640 withService(({ service }) =>
641 Effect.gen(function* () {
642 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
643 const receivedAt = DateTime.formatIso(DateTime.subtract(yield* DateTime.now, { minutes: 5 }));
644 yield* service.triggerWebhook(requestFor(task, { relayDeliveryId: "relay-2", receivedAt }));
645 const [delivery] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
646 assert.equal(delivery?.receivedAt, receivedAt);
647 }),
648 ),
649);
650
651it.effect("a receive time in the future counts as now", () =>
652 withService(({ service, launches }) =>
653 Effect.gen(function* () {
654 const { task } = yield* service.upsert(
655 yield* webhookTaskInput({ schedule: { type: "webhook", maxDeliveryAgeMinutes: 30 } }),
656 );
657 const now = yield* DateTime.now;
658 const result = yield* service.triggerWebhook(
659 requestFor(task, {
660 relayDeliveryId: "future",
661 receivedAt: DateTime.formatIso(DateTime.add(now, { days: 365 })),
662 }),
663 );
664 assert.equal(result._tag, "accepted");
665 yield* Queue.take(launches);
666 const [delivery] = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries;
667 assert.equal(delivery?.receivedAt, DateTime.formatIso(now));
668 }),
669 ),
670);
671
672it.effect("skips a held request older than the task's max age", () =>
673 withService(({ service, launches }) =>
674 Effect.gen(function* () {
675 const { task } = yield* service.upsert(
676 yield* webhookTaskInput({ schedule: { type: "webhook", maxDeliveryAgeMinutes: 30 } }),
677 );
678 const now = yield* DateTime.now;
679 const old = DateTime.formatIso(DateTime.subtract(now, { minutes: 31 }));
680 const fresh = DateTime.formatIso(DateTime.subtract(now, { minutes: 5 }));
681 const tooOld = yield* service.triggerWebhook(
682 requestFor(task, { relayDeliveryId: "old", receivedAt: old }),
683 );
684 assert.equal(tooOld._tag, "expired");
685 assert.equal(yield* Queue.size(launches), 0);
686 const ok = yield* service.triggerWebhook(
687 requestFor(task, { relayDeliveryId: "fresh", receivedAt: fresh }),
688 );
689 assert.equal(ok._tag, "accepted");
690 const outcomes = (yield* service.listWebhookDeliveries({ id: task.id })).deliveries.map(
691 (delivery) => delivery.outcome,
692 );
693 assert.includeMembers(outcomes, ["expired", "accepted"]);
694 }),
695 ),
696);
697
698it.effect("runs a held request of any age when no max age is set", () =>
699 withService(({ service }) =>
700 Effect.gen(function* () {
701 const { task } = yield* service.upsert(yield* webhookTaskInput());
702 const now = yield* DateTime.now;
703 const result = yield* service.triggerWebhook(
704 requestFor(task, {
705 relayDeliveryId: "ancient",
706 receivedAt: DateTime.formatIso(DateTime.subtract(now, { hours: 23 })),
707 }),
708 );
709 assert.equal(result._tag, "accepted");
710 }),
711 ),
712);
713
714it.effect("deleting a task removes its delivery log", () =>
715 withService(({ service }) =>
716 Effect.gen(function* () {
717 const { task } = yield* service.upsert(yield* webhookTaskInput({ enabled: false }));
718 yield* service.triggerWebhook(requestFor(task));
719 yield* service.delete({ id: task.id });
720 assert.equal((yield* service.listWebhookDeliveries({ id: task.id })).deliveries.length, 0);
721 }),
722 ),
723);
724
725const githubSignature = (secret: string) =>
726 `sha256=${NodeCrypto.createHmac("sha256", secret).update(pullRequestBody).digest("hex")}`;
727
728it.effect("a signature can take the user's secret by ref, which works only once", () =>
729 withService(({ service, launches, secretsByRef }) =>
730 Effect.gen(function* () {
731 secretsByRef.set("secret-ref:00000000000000000000000000000001", "github-secret");
732 const githubSchedule = (secretRef: string) => ({
733 type: "webhook",
734 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=", secretRef },
735 });
736 const { task } = yield* service.upsert(
737 yield* webhookTaskInput({
738 schedule: githubSchedule("secret-ref:00000000000000000000000000000001"),
739 }),
740 );
741 assert.isTrue(task.webhook!.hasSecret);
742 const signed = yield* service.triggerWebhook(
743 requestFor(task, {
744 headers: {
745 "content-type": "application/json",
746 "x-hub-signature-256": githubSignature("github-secret"),
747 },
748 }),
749 );
750 assert.equal(signed._tag, "accepted");
751 yield* Queue.take(launches);
752
753 // The ref was consumed by that save.
754 const reused = yield* service
755 .upsert(
756 yield* webhookTaskInput({
757 id: "scheduled-task:other",
758 schedule: githubSchedule("secret-ref:00000000000000000000000000000001"),
759 }),
760 )
761 .pipe(Effect.flip);
762 assert.include(reused.message, "already used");
763 }),
764 ),
765);
766
767it.effect("a retried save with an already used secretRef keeps the stored secret", () =>
768 withService(({ service, launches, secretsByRef }) =>
769 Effect.gen(function* () {
770 secretsByRef.set("secret-ref:00000000000000000000000000000002", "github-secret");
771 const save = webhookTaskInput({
772 id: undefined,
773 commandId: "command:mcp:schedule-task:release-hook",
774 schedule: {
775 type: "webhook",
776 signature: {
777 header: "x-hub-signature-256",
778 encoding: "hex",
779 prefix: "sha256=",
780 secretRef: "secret-ref:00000000000000000000000000000002",
781 },
782 },
783 });
784 const first = yield* service.upsert(yield* save);
785 // The agent never saw the first result, so it sends the same call again,
786 // this time with a plain secret alongside the used ref.
787 const retried = yield* service.upsert(
788 yield* webhookTaskInput({
789 id: undefined,
790 commandId: "command:mcp:schedule-task:release-hook",
791 schedule: {
792 type: "webhook",
793 signature: {
794 header: "x-hub-signature-256",
795 encoding: "hex",
796 prefix: "sha256=",
797 secretRef: "secret-ref:00000000000000000000000000000002",
798 secret: "made-up-secret",
799 },
800 },
801 }),
802 );
803 assert.equal(retried.task.id, first.task.id);
804 const signed = yield* service.triggerWebhook(
805 requestFor(retried.task, {
806 headers: {
807 "content-type": "application/json",
808 "x-hub-signature-256": githubSignature("github-secret"),
809 },
810 }),
811 );
812 assert.equal(signed._tag, "accepted");
813 yield* Queue.take(launches);
814 }),
815 ),
816);
817
818it.effect("a signature without any secret is refused", () =>
819 withService(({ service }) =>
820 Effect.gen(function* () {
821 const failure = yield* service
822 .upsert(
823 yield* webhookTaskInput({
824 schedule: {
825 type: "webhook",
826 signature: { header: "x-hub-signature-256", encoding: "hex", prefix: "sha256=" },
827 },
828 }),
829 )
830 .pipe(Effect.flip);
831 assert.include(failure.message, "needs a signing secret");
832 }),
833 ),
834);
835
836const signatureFor = (secret: string) =>
837 `sha256=${NodeCrypto.createHmac("sha256", secret).update(pullRequestBody).digest("hex")}`;
838
839/** Count recorded by `t3_webhook_deliveries_total` for one outcome and source. */
840const deliveriesCounted = (outcome: string, source: "relay" | "direct") =>
841 Metric.snapshot.pipe(
842 Effect.map((snapshots) => {
843 const found = snapshots.find(
844 (snapshot) =>
845 snapshot.id === "t3_webhook_deliveries_total" &&
846 snapshot.attributes?.outcome === outcome &&
847 snapshot.attributes?.source === source,
848 );
849 return found?.type === "Counter" ? Number(found.state.count) : 0;
850 }),
851 );
852
853it.effect("counts each handled request by what happened to it", () =>
854 withService(({ service, launches }) =>
855 Effect.gen(function* () {
856 const { task } = yield* service.upsert(
857 yield* webhookTaskInput({
858 schedule: {
859 type: "webhook",
860 signature: {
861 header: "x-hub-signature-256",
862 encoding: "hex",
863 prefix: "sha256=",
864 secret: "github-secret",
865 },
866 },
867 }),
868 );
869 const before = {
870 accepted: yield* deliveriesCounted("accepted", "direct"),
871 rejected: yield* deliveriesCounted("rejected_signature", "direct"),
872 notFound: yield* deliveriesCounted("not_found", "direct"),
873 relayAccepted: yield* deliveriesCounted("accepted", "relay"),
874 duplicate: yield* deliveriesCounted("duplicate", "relay"),
875 };
876 const signed = {
877 "content-type": "application/json",
878 "x-hub-signature-256": signatureFor("github-secret"),
879 };
880 yield* service.triggerWebhook(requestFor(task, { headers: signed }));
881 yield* Queue.take(launches);
882 yield* service.triggerWebhook(requestFor(task));
883 yield* service.triggerWebhook(requestFor(task, { token: "wrong" }));
884 // A request the relay held, then the same request again.
885 const relayed = requestFor(task, { headers: signed, relayDeliveryId: "relay-1" });
886 yield* service.triggerWebhook(relayed);
887 yield* Queue.take(launches);
888 yield* service.triggerWebhook(relayed);
889
890 assert.equal((yield* deliveriesCounted("accepted", "direct")) - before.accepted, 1);
891 assert.equal((yield* deliveriesCounted("rejected_signature", "direct")) - before.rejected, 1);
892 assert.equal((yield* deliveriesCounted("not_found", "direct")) - before.notFound, 1);
893 assert.equal((yield* deliveriesCounted("accepted", "relay")) - before.relayAccepted, 1);
894 assert.equal((yield* deliveriesCounted("duplicate", "relay")) - before.duplicate, 1);
895 }),
896 ),
897);