1import * as Scheduler from "../scheduling/Scheduler.ts";
2import * as NodeUtil from "node:util";
3
4import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
5import { assert, it, vi } from "@effect/vitest";
6import {
7 ProjectId,
8 ProviderInstanceId,
9 ScheduledTaskError,
10 ScheduledTaskId,
11 type ScheduledTaskUpsertInput,
12 SecretRef,
13 ThreadId,
14} from "@t3tools/contracts";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as Exit from "effect/Exit";
18import * as Layer from "effect/Layer";
19import * as Logger from "effect/Logger";
20import * as Option from "effect/Option";
21import * as SqlClient from "effect/sql/SqlClient";
22
23import * as Fiber from "effect/Fiber";
24import * as Stream from "effect/Stream";
25import * as Deferred from "effect/Deferred";
26import * as Ref from "effect/Ref";
27import * as Schema from "effect/Schema";
28import * as TestClock from "effect/testing/TestClock";
29
30import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
31import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
32import * as SecretRequests from "../secrets/SecretRequests.ts";
33import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
34import * as SqlitePersistence from "../persistence/Sqlite.ts";
35import * as ScheduledTaskService from "./ScheduledTaskService.ts";
36import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
37
38const isScheduledTaskError = Schema.is(ScheduledTaskError);
39
40const threadShell = (
41 projectId: ProjectId,
42 state: { archivedAt?: string | null; deletedAt?: string | null } = {},
43) => ({ projectId, archivedAt: null, deletedAt: null, ...state }) as never;
44
45const insertRow = (
46 sql: SqlClient.SqlClient,
47 row: {
48 id: string | null;
49 next: string | null;
50 enabled: number;
51 status: string;
52 scheduleJson?: string;
53 prompt?: string;
54 },
55 now: string,
56) =>
57 sql`INSERT INTO scheduled_tasks ${sql.insert({
58 task_id: row.id,
59 title: "task",
60 prompt: row.prompt ?? "Run task",
61 enabled: row.enabled,
62 schedule_json: row.scheduleJson ?? '{"type":"interval","everyMs":60000}',
63 project_id: "project:test",
64 thread_id: null,
65 workspace_strategy_json: '{"type":"root"}',
66 model_selection_json: '{"instanceId":"codex","model":"gpt-5"}',
67 runtime_mode: "full-access",
68 interaction_mode: "default",
69 created_by: "user",
70 creation_source: "web",
71 created_at: now,
72 updated_at: now,
73 next_run_at: row.next,
74 last_run_at: null,
75 last_run_status: row.status,
76 last_run_error: null,
77 run_count: 0,
78 })}`;
79
80it.effect("loads only due tasks and skips corrupt due rows without decoding settled tasks", () =>
81 Effect.gen(function* () {
82 const sql = yield* SqlClient.SqlClient;
83 const now = "2026-09-09T12:00:00.000Z";
84 const secret = "PROMPT-SECRET-DO-NOT-LOG";
85 const credential = "CREDENTIAL-SECRET-DO-NOT-LOG";
86 for (const row of [
87 { id: "due-now", next: now, enabled: 1, status: "never" },
88 { id: "", next: now, enabled: 1, status: "never" },
89 { id: " ", next: now, enabled: 1, status: "never" },
90 { id: null, next: now, enabled: 1, status: "never" },
91 {
92 id: "due-earlier",
93 next: "2026-09-09T11:00:00.000Z",
94 enabled: 1,
95 status: "failed",
96 },
97 { id: "due-bad-date", next: "", enabled: 1, status: "never" },
98 {
99 id: "due-corrupt",
100 next: now,
101 enabled: 1,
102 status: "never",
103 scheduleJson: "broken",
104 },
105 {
106 id: "due-cred",
107 next: now,
108 enabled: 1,
109 status: "never",
110 scheduleJson: `{"type":"interval","everyMs":"${credential}"}`,
111 },
112 { id: "disabled", next: now, enabled: 0, status: "never", scheduleJson: "broken" },
113 {
114 id: "future",
115 next: "2026-09-09T12:00:00.001Z",
116 enabled: 1,
117 status: "never",
118 scheduleJson: "broken",
119 },
120 {
121 id: "unscheduled",
122 next: null,
123 enabled: 1,
124 status: "never",
125 scheduleJson: "broken",
126 },
127 { id: "running", next: now, enabled: 1, status: "running", scheduleJson: "broken" },
128 ]) {
129 yield* insertRow(sql, { ...row, prompt: secret }, now);
130 }
131 const warnings: unknown[] = [];
132 const tasks = yield* ScheduledTaskService.listDueTasks(DateTime.makeUnsafe(now)).pipe(
133 Effect.provide(
134 Logger.layer([
135 Logger.make(({ message }) => {
136 warnings.push(message);
137 }),
138 ]),
139 ),
140 );
141 assert.deepEqual(
142 tasks.map((task) => task.id),
143 ["due-earlier", "due-now"],
144 );
145 assert.equal(warnings.length, 6);
146 const annotations = warnings.map(
147 (warning) => (warning as ReadonlyArray<unknown>)[1] as Record<string, unknown>,
148 );
149 assert.deepEqual(
150 annotations.map((annotation) => annotation.taskId),
151 ["due-bad-date", null, "", " ", "due-corrupt", "due-cred"],
152 );
153 const causes = annotations.map((annotation) => annotation.cause);
154 assert.deepEqual(
155 causes.map((cause) => (isScheduledTaskError(cause) ? cause.taskId : undefined)),
156 [undefined, undefined, undefined, undefined, "due-corrupt", "due-cred"],
157 );
158 for (const cause of causes) {
159 if (cause !== undefined) assert.isTrue(isScheduledTaskError(cause));
160 }
161 const rendered = NodeUtil.inspect(annotations, { depth: null });
162 assert.isFalse(rendered.includes(secret));
163 assert.isFalse(rendered.includes(credential));
164 const running = yield* sql<{
165 last_run_status: string;
166 }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'running'`;
167 assert.equal(running[0]?.last_run_status, "running");
168 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
169);
170
171it.effect(
172 "releases interrupted runs on startup and still executes due tasks, skipping corrupt rows",
173 () =>
174 Effect.gen(function* () {
175 const sql = yield* SqlClient.SqlClient;
176 const now = "2026-09-09T12:00:00.000Z";
177 const hugeInterval = '{"type":"interval","everyMs":9000000000000000}';
178 for (const row of [
179 { id: "stuck-valid", next: now, status: "running" },
180 { id: "stuck-corrupt", next: now, status: "running", scheduleJson: "broken" },
181 { id: "stuck-huge", next: now, status: "running", scheduleJson: hugeInterval },
182 { id: null, next: now, status: "running" },
183 { id: "due-bad-date", next: "", status: "never" },
184 { id: "due-healthy", next: now, status: "never" },
185 { id: "due-huge", next: now, status: "never", scheduleJson: hugeInterval },
186 { id: "due-second", next: now, status: "never" },
187 { id: "due-third", next: now, status: "never" },
188 ]) {
189 yield* insertRow(sql, { ...row, enabled: 1 }, now);
190 }
191
192 yield* TestClock.setTime(Date.parse(now) + 1_000);
193 const dispatched = yield* Ref.make(0);
194 const lastDispatched = yield* Deferred.make<void>();
195 const releaseLast = yield* Deferred.make<void>();
196 yield* Effect.scoped(
197 Effect.gen(function* () {
198 yield* Layer.build(
199 Layer.provideMerge(
200 ScheduledTaskService.layer,
201 Layer.mergeAll(
202 ManagedPolicy.layerTest(),
203 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
204 launch: () =>
205 Ref.updateAndGet(dispatched, (n) => n + 1).pipe(
206 Effect.andThen((n) =>
207 n === 4
208 ? Deferred.succeed(lastDispatched, undefined).pipe(
209 Effect.andThen(Deferred.await(releaseLast)),
210 )
211 : Effect.void,
212 ),
213 Effect.andThen(Effect.die(new Error("test launch failure"))),
214 ),
215 }),
216 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
217 namedProjectsRoot: "/projects",
218 isScratchProject: () => Effect.succeed(false),
219 }),
220 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
221 Layer.mock(SecretRequests.SecretRequests)({}),
222 NodeCrypto.layer,
223 Scheduler.layer,
224 ),
225 ),
226 );
227 // The poll delays before its first pass; advance the frozen clock once.
228 yield* TestClock.adjust("6 seconds");
229 yield* Deferred.await(lastDispatched);
230 const inflight = yield* sql<{
231 last_run_status: string;
232 }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'due-third'`;
233 assert.equal(inflight[0]?.last_run_status, "running");
234 yield* Deferred.succeed(releaseLast, undefined);
235 }),
236 );
237
238 const rows = yield* sql<{
239 task_id: string | null;
240 last_run_status: string;
241 last_run_error: string | null;
242 next_run_at: string | null;
243 run_count: number;
244 }>`SELECT task_id, last_run_status, last_run_error, next_run_at, run_count
245 FROM scheduled_tasks ORDER BY task_id`;
246 const byId = new Map(rows.map((row) => [row.task_id, row]));
247 assert.equal(rows.length, 9);
248 assert.equal(byId.get("stuck-valid")?.last_run_status, "failed");
249 assert.isNotNull(byId.get("stuck-valid")?.next_run_at);
250 assert.equal(byId.get("stuck-corrupt")?.last_run_status, "failed");
251 assert.equal(byId.get(null)?.last_run_status, "failed");
252 const stuckHuge = byId.get("stuck-huge");
253 assert.equal(stuckHuge?.last_run_status, "failed");
254 assert.isNull(stuckHuge?.next_run_at);
255 for (const id of ["stuck-valid", "stuck-corrupt", "stuck-huge", null]) {
256 const row = byId.get(id);
257 assert.equal(row?.last_run_error, "Run was interrupted by a server restart.");
258 assert.equal(row?.run_count, 1);
259 }
260 for (const id of ["due-healthy", "due-huge", "due-second"]) {
261 const row = byId.get(id);
262 assert.equal(row?.last_run_status, "failed");
263 assert.isTrue(row?.last_run_error?.includes("test launch failure") === true);
264 assert.equal(row?.run_count, 1);
265 }
266 assert.isNull(byId.get("due-huge")?.next_run_at);
267 assert.equal(byId.get("due-bad-date")?.last_run_status, "never");
268 assert.equal(byId.get("due-bad-date")?.run_count, 0);
269 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
270);
271
272it.effect(
273 "treats a next_run_at corrupted before the dispatch re-read as not due instead of defecting the poll",
274 () =>
275 Effect.gen(function* () {
276 const sql = yield* SqlClient.SqlClient;
277 const now = "2026-09-09T12:00:00.000Z";
278 for (const row of [
279 { id: "run-a", next: now, status: "never" },
280 { id: "run-b-victim", next: now, status: "never" },
281 { id: "run-c", next: now, status: "never" },
282 { id: "run-d", next: now, status: "never" },
283 ]) {
284 yield* insertRow(sql, { ...row, enabled: 1 }, now);
285 }
286 yield* TestClock.setTime(Date.parse(now) + 1_000);
287 const dispatched = yield* Ref.make(0);
288 const lastDispatched = yield* Deferred.make<void>();
289 const releaseLast = yield* Deferred.make<void>();
290 yield* Effect.scoped(
291 Effect.gen(function* () {
292 yield* Layer.build(
293 Layer.provideMerge(
294 ScheduledTaskService.layer,
295 Layer.mergeAll(
296 ManagedPolicy.layerTest(),
297 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
298 launch: () =>
299 Ref.updateAndGet(dispatched, (n) => n + 1).pipe(
300 Effect.andThen((n) =>
301 n === 1
302 ? sql`UPDATE scheduled_tasks
303 SET next_run_at = ''
304 WHERE task_id = 'run-b-victim'`.pipe(Effect.asVoid, Effect.orDie)
305 : n === 3
306 ? Deferred.succeed(lastDispatched, undefined).pipe(
307 Effect.andThen(Deferred.await(releaseLast)),
308 )
309 : Effect.void,
310 ),
311 Effect.andThen(Effect.die(new Error("test launch failure"))),
312 ),
313 }),
314 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
315 namedProjectsRoot: "/projects",
316 isScratchProject: () => Effect.succeed(false),
317 }),
318 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
319 Layer.mock(SecretRequests.SecretRequests)({}),
320 NodeCrypto.layer,
321 Scheduler.layer,
322 ),
323 ),
324 );
325 yield* TestClock.adjust("6 seconds");
326 yield* Deferred.await(lastDispatched);
327 assert.equal(yield* Ref.get(dispatched), 3);
328 const rows = yield* sql<{
329 task_id: string;
330 last_run_status: string;
331 next_run_at: string | null;
332 run_count: number;
333 }>`SELECT task_id, last_run_status, next_run_at, run_count
334 FROM scheduled_tasks ORDER BY task_id`;
335 const byId = new Map(rows.map((row) => [row.task_id, row]));
336 const victim = byId.get("run-b-victim");
337 assert.equal(victim?.last_run_status, "never");
338 assert.equal(victim?.run_count, 0);
339 assert.equal(victim?.next_run_at, "");
340 for (const id of ["run-a", "run-c"]) {
341 const row = byId.get(id);
342 assert.equal(row?.last_run_status, "failed");
343 assert.equal(row?.run_count, 1);
344 }
345 yield* Deferred.succeed(releaseLast, undefined);
346 }),
347 );
348 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
349);
350
351it.effect("thread cleanup notifies subscribers and a finishing run cannot recreate the task", () =>
352 Effect.gen(function* () {
353 const started = yield* Deferred.make<void>();
354 const release = yield* Deferred.make<void>();
355 const subscribed = yield* Deferred.make<void>();
356 const running = yield* Deferred.make<void>();
357 const removed = yield* Deferred.make<void>();
358 const layer = ScheduledTaskService.layer.pipe(
359 Layer.provide(
360 Layer.mergeAll(
361 ManagedPolicy.layerTest(),
362 Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
363 Layer.mock(ThreadManagementService.ThreadManagementService)({
364 sendToThread: () =>
365 Deferred.succeed(started, undefined).pipe(
366 Effect.andThen(Deferred.await(release)),
367 Effect.andThen(Effect.die("test dispatch failure")),
368 ),
369 getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))),
370 }),
371 Layer.mock(SecretRequests.SecretRequests)({}),
372 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
373 namedProjectsRoot: "/projects",
374 isScratchProject: () => Effect.succeed(false),
375 }),
376 NodeCrypto.layer,
377 Scheduler.layer,
378 ),
379 ),
380 );
381 yield* Effect.gen(function* () {
382 const service = yield* ScheduledTaskService.ScheduledTaskService;
383 const threadId = ThreadId.make("thread:deleted-during-run");
384 const { task } = yield* service.upsert({
385 id: ScheduledTaskId.make("task:deleted-during-run"),
386 title: "Check orchestration",
387 prompt: "Check progress",
388 enabled: true,
389 schedule: { type: "interval", everyMs: 60_000 },
390 projectId: ProjectId.make("project:test"),
391 threadId,
392 workspaceStrategy: { type: "root" },
393 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
394 runtimeMode: "full-access",
395 interactionMode: "default",
396 });
397 yield* service.subscribeList().pipe(
398 Stream.runForEach(({ tasks }) => {
399 const current = tasks.find((candidate) => candidate.id === task.id);
400 if (current === undefined) return Deferred.succeed(removed, undefined);
401 return current.lastRunStatus === "running"
402 ? Deferred.succeed(running, undefined)
403 : Deferred.succeed(subscribed, undefined);
404 }),
405 Effect.forkChild,
406 );
407 yield* Deferred.await(subscribed);
408 const run = yield* service.runNow({ id: task.id }).pipe(Effect.forkChild);
409 yield* Deferred.await(started);
410 yield* Deferred.await(running);
411 const refused = yield* service.runNow({ id: task.id }).pipe(Effect.flip);
412 assert.equal(refused.message, "Schedule task is already running.");
413 yield* service.deleteForThread(threadId);
414 yield* Deferred.await(removed);
415 yield* Deferred.succeed(release, undefined);
416 yield* Fiber.join(run);
417 assert.lengthOf((yield* service.list()).tasks, 0);
418 }).pipe(Effect.provide(layer));
419 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
420);
421
422it.effect.each([
423 { status: "failed", error: "Weekly usage limit reached" },
424 { status: "interrupted", error: "The run was stopped." },
425 { status: "cancelled", error: "The run was cancelled." },
426] as const)("a run that ends $status after its message was accepted records it", (outcome) =>
427 Effect.gen(function* () {
428 const projectId = ProjectId.make("project:test");
429 const ended = yield* Deferred.make<void>();
430 const sent = yield* Ref.make("");
431 const run = { id: "run:2", status: outcome.status, rootNodeId: "node:root" };
432 const layer = ScheduledTaskService.layer.pipe(
433 Layer.provide(
434 Layer.mergeAll(
435 ManagedPolicy.layerTest(),
436 Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
437 Layer.mock(ThreadManagementService.ThreadManagementService)({
438 sendToThread: (input) => Ref.set(sent, input.messageId).pipe(Effect.as({} as never)),
439 getThreadShell: () => Effect.succeed(threadShell(projectId)),
440 getProjectThreadRecords: () =>
441 Effect.map(
442 Ref.get(sent),
443 (userMessageId) =>
444 ({
445 runs: [
446 { id: "run:1", status: "completed", userMessageId: "message:earlier" },
447 { ...run, userMessageId },
448 ],
449 messages: [{ id: userMessageId, runId: "run:2" }],
450 turnItems: [
451 {
452 type: "error",
453 status: "failed",
454 runId: "run:2",
455 nodeId: "node:root",
456 failure: { message: "Weekly usage limit reached" },
457 },
458 ],
459 }) as never,
460 ),
461 waitForThread: (input) =>
462 Deferred.await(ended).pipe(
463 Effect.as({
464 threadId: input.threadId,
465 run: { ...run, id: input.runId, userMessageId: "" },
466 timedOut: false,
467 } as never),
468 ),
469 }),
470 Layer.mock(SecretRequests.SecretRequests)({}),
471 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
472 namedProjectsRoot: "/projects",
473 isScratchProject: () => Effect.succeed(false),
474 }),
475 NodeCrypto.layer,
476 Scheduler.layer,
477 ),
478 ),
479 );
480 yield* Effect.gen(function* () {
481 const service = yield* ScheduledTaskService.ScheduledTaskService;
482 const { task } = yield* service.upsert({
483 id: ScheduledTaskId.make("task:quota"),
484 title: "Quota bookkeeping",
485 prompt: "Update the notes",
486 enabled: true,
487 schedule: { type: "interval", everyMs: 60_000 },
488 projectId,
489 threadId: ThreadId.make("thread:quota"),
490 workspaceStrategy: { type: "root" },
491 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
492 runtimeMode: "full-access",
493 interactionMode: "default",
494 });
495 yield* service.runNow({ id: task.id });
496 const failed = yield* service.subscribeList().pipe(
497 Stream.map(({ tasks }) => tasks.find((candidate) => candidate.id === task.id)),
498 Stream.filter((current) => current?.lastRunStatus === "failed"),
499 Stream.runHead,
500 Effect.forkChild,
501 );
502 yield* Deferred.succeed(ended, undefined);
503 const recorded = Option.getOrUndefined(yield* Fiber.join(failed));
504 assert.equal(recorded?.lastRunError, outcome.error);
505 assert.equal(recorded?.runCount, 1);
506 }).pipe(Effect.provide(layer));
507 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
508);
509
510it.effect("a bound run follows its thread when promotion moves it mid-send", () =>
511 Effect.gen(function* () {
512 const scratch = ProjectId.make("project:scratch");
513 const promoted = ProjectId.make("project:promoted");
514 const sends = yield* Ref.make<Array<{ projectId: ProjectId; commandId: string }>>([]);
515 // Saving reads it first. Then the rebind lands, and the promotion commits between the shell
516 // read and the retry.
517 const shellProjects = yield* Ref.make([promoted, scratch, promoted]);
518 const layer = ScheduledTaskService.layer.pipe(
519 Layer.provide(
520 Layer.mergeAll(
521 ManagedPolicy.layerTest(),
522 Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
523 Layer.mock(ThreadManagementService.ThreadManagementService)({
524 sendToThread: (input) =>
525 Ref.updateAndGet(sends, (all) => [
526 ...all,
527 { projectId: input.projectId, commandId: input.commandId },
528 ]).pipe(
529 Effect.flatMap((all) =>
530 all.length < 3
531 ? Effect.fail(
532 new ThreadManagementService.ThreadManagementThreadNotFoundError({
533 projectId: input.projectId,
534 threadId: input.threadId,
535 }),
536 )
537 : Effect.succeed({} as never),
538 ),
539 ),
540 getThreadShell: () =>
541 Ref.modify(shellProjects, ([next, ...rest]) => [threadShell(next ?? promoted), rest]),
542 }),
543 Layer.mock(SecretRequests.SecretRequests)({}),
544 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
545 namedProjectsRoot: "/projects",
546 isScratchProject: () => Effect.succeed(false),
547 }),
548 NodeCrypto.layer,
549 Scheduler.layer,
550 ),
551 ),
552 );
553 yield* Effect.gen(function* () {
554 const service = yield* ScheduledTaskService.ScheduledTaskService;
555 const threadId = ThreadId.make("thread:promoted-during-run");
556 const { task } = yield* service.upsert({
557 id: ScheduledTaskId.make("task:promoted-during-run"),
558 title: "Check in",
559 prompt: "Check progress",
560 enabled: true,
561 schedule: { type: "interval", everyMs: 60_000 },
562 projectId: promoted,
563 threadId,
564 workspaceStrategy: { type: "root" },
565 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
566 runtimeMode: "full-access",
567 interactionMode: "default",
568 });
569
570 const result = yield* service.runNow({ id: task.id });
571
572 assert.equal(result.threadId, threadId);
573 const attempts = yield* Ref.get(sends);
574 assert.deepEqual(
575 attempts.map((attempt) => attempt.projectId),
576 [promoted, scratch, promoted],
577 );
578 assert.equal(new Set(attempts.map((attempt) => attempt.commandId)).size, 1);
579 }).pipe(Effect.provide(layer));
580 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
581);
582
583it.effect("a fresh-session task launches one thread, then starts each later run fresh in it", () =>
584 Effect.gen(function* () {
585 const launches = yield* Ref.make(0);
586 const sends = yield* Ref.make<Array<{ threadId: ThreadId; freshContext?: boolean }>>([]);
587 const launched = ThreadId.make("thread:fresh-session-task");
588 const layer = ScheduledTaskService.layer.pipe(
589 Layer.provide(
590 Layer.mergeAll(
591 ManagedPolicy.layerTest(),
592 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
593 launch: () =>
594 Ref.update(launches, (n) => n + 1).pipe(
595 Effect.as({ threadId: launched, resumed: false } as never),
596 ),
597 }),
598 Layer.mock(ThreadManagementService.ThreadManagementService)({
599 getThreadShell: () => Effect.succeed({ deletedAt: null } as never),
600 sendToThread: (input) =>
601 Ref.update(sends, (all) => [
602 ...all,
603 {
604 threadId: input.threadId,
605 ...(input.freshContext === undefined ? {} : { freshContext: input.freshContext }),
606 },
607 ]).pipe(Effect.as({} as never)),
608 }),
609 Layer.mock(SecretRequests.SecretRequests)({}),
610 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
611 namedProjectsRoot: "/projects",
612 isScratchProject: () => Effect.succeed(false),
613 }),
614 NodeCrypto.layer,
615 Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }),
616 ),
617 ),
618 );
619 yield* Effect.gen(function* () {
620 const service = yield* ScheduledTaskService.ScheduledTaskService;
621 const input = {
622 id: ScheduledTaskId.make("task:fresh-session"),
623 title: "Triage",
624 prompt: "Triage new issues",
625 enabled: true,
626 schedule: { type: "interval", everyMs: 60_000 },
627 projectId: ProjectId.make("project:test"),
628 workspaceStrategy: { type: "root" },
629 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
630 runtimeMode: "full-access",
631 interactionMode: "default",
632 } as const;
633 yield* service.upsert({ ...input, threadMode: "fresh_session" });
634
635 assert.equal((yield* service.runNow({ id: input.id })).threadId, launched);
636 // A form opened before the first run saves no thread; the binding stays.
637 const { task } = yield* service.upsert({
638 ...input,
639 threadId: null,
640 threadMode: "fresh_session",
641 });
642 assert.equal(task.threadId, launched);
643 assert.equal(task.threadMode, "fresh_session");
644 // A client that predates thread modes sends neither; both stay.
645 const legacy = yield* service.upsert({ ...input, threadId: null });
646 assert.equal(legacy.task.threadId, launched);
647 assert.equal(legacy.task.threadMode, "fresh_session");
648 yield* service.runNow({ id: input.id });
649
650 assert.equal(yield* Ref.get(launches), 1);
651 assert.deepEqual(yield* Ref.get(sends), [{ threadId: launched, freshContext: true }]);
652 // Its thread stays in the old project, so moving the task leaves it for a new one.
653 const moved = yield* service.upsert({
654 ...input,
655 projectId: ProjectId.make("project:moved"),
656 threadId: launched,
657 });
658 assert.equal(moved.task.threadId, null);
659 assert.equal(moved.task.threadMode, "fresh_session");
660 const unbound = yield* service.upsert({ ...input, threadMode: "new_thread" });
661 assert.equal(unbound.task.threadId, null);
662 }).pipe(Effect.provide(layer));
663 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
664);
665
666it.effect("a first run whose thread was deleted before it bound removes the task", () =>
667 Effect.gen(function* () {
668 const layer = ScheduledTaskService.layer.pipe(
669 Layer.provide(
670 Layer.mergeAll(
671 ManagedPolicy.layerTest(),
672 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
673 launch: () =>
674 Effect.succeed({
675 threadId: ThreadId.make("thread:deleted"),
676 resumed: false,
677 } as never),
678 }),
679 // Deleted before the binding landed, so its cleanup found no task to remove.
680 Layer.mock(ThreadManagementService.ThreadManagementService)({
681 getThreadShell: () =>
682 Effect.succeed({ deletedAt: DateTime.makeUnsafe("2026-11-03T00:00:00Z") } as never),
683 }),
684 Layer.mock(SecretRequests.SecretRequests)({}),
685 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
686 namedProjectsRoot: "/projects",
687 isScratchProject: () => Effect.succeed(false),
688 }),
689 NodeCrypto.layer,
690 Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }),
691 ),
692 ),
693 );
694 yield* Effect.gen(function* () {
695 const service = yield* ScheduledTaskService.ScheduledTaskService;
696 const { task } = yield* service.upsert({
697 id: ScheduledTaskId.make("task:thread-deleted-mid-launch"),
698 title: "Triage",
699 prompt: "Triage new issues",
700 enabled: true,
701 schedule: { type: "interval", everyMs: 60_000 },
702 projectId: ProjectId.make("project:test"),
703 threadMode: "continue",
704 workspaceStrategy: { type: "root" },
705 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
706 runtimeMode: "full-access",
707 interactionMode: "default",
708 });
709 yield* service.runNow({ id: task.id });
710 assert.deepEqual((yield* service.list()).tasks, []);
711 }).pipe(Effect.provide(layer));
712 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
713);
714
715it.effect.each(["recreated under the same id", "moved to another project"] as const)(
716 "a first run never binds its thread to a task %s",
717 (change) =>
718 Effect.gen(function* () {
719 const launching = yield* Deferred.make<void>();
720 const proceed = yield* Deferred.make<void>();
721 const layer = ScheduledTaskService.layer.pipe(
722 Layer.provide(
723 Layer.mergeAll(
724 ManagedPolicy.layerTest(),
725 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
726 launch: () =>
727 Deferred.succeed(launching, undefined).pipe(
728 Effect.andThen(Deferred.await(proceed)),
729 Effect.as({
730 threadId: ThreadId.make("thread:old-generation"),
731 resumed: false,
732 } as never),
733 ),
734 }),
735 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
736 Layer.mock(SecretRequests.SecretRequests)({}),
737 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
738 namedProjectsRoot: "/projects",
739 isScratchProject: () => Effect.succeed(false),
740 }),
741 NodeCrypto.layer,
742 Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }),
743 ),
744 ),
745 );
746 yield* Effect.gen(function* () {
747 const service = yield* ScheduledTaskService.ScheduledTaskService;
748 const input = {
749 id: ScheduledTaskId.make("task:recreated-mid-launch"),
750 title: "Triage",
751 prompt: "Triage new issues",
752 enabled: true,
753 schedule: { type: "interval", everyMs: 60_000 },
754 projectId: ProjectId.make("project:test"),
755 threadMode: "fresh_session",
756 workspaceStrategy: { type: "root" },
757 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
758 runtimeMode: "full-access",
759 interactionMode: "default",
760 } as const;
761 yield* service.upsert(input);
762 const run = yield* Effect.forkChild(service.runNow({ id: input.id }));
763 yield* Deferred.await(launching);
764 if (change === "recreated under the same id") {
765 yield* service.delete({ id: input.id });
766 yield* TestClock.adjust("1 second");
767 yield* service.upsert(input);
768 } else {
769 yield* service.upsert({ ...input, projectId: ProjectId.make("project:moved") });
770 }
771 yield* Deferred.succeed(proceed, undefined);
772 yield* Fiber.await(run);
773
774 const replacement = (yield* service.list()).tasks.find((task) => task.id === input.id);
775 assert.equal(replacement?.threadId, null);
776 }).pipe(Effect.provide(layer));
777 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
778);
779
780it.effect("refuses saving an enabled task above the policy ceiling", () =>
781 Effect.gen(function* () {
782 const layer = ScheduledTaskService.layer.pipe(
783 Layer.provide(
784 Layer.mergeAll(
785 ManagedPolicy.layerTest({
786 schemaVersion: 1,
787 organization: "Example Corp",
788 limits: { maxRuntimeMode: "auto-accept-edits" },
789 }),
790 Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
791 Layer.mock(ThreadManagementService.ThreadManagementService)({
792 getThreadShell: () => Effect.succeed(threadShell(ProjectId.make("project:test"))),
793 }),
794 Layer.mock(SecretRequests.SecretRequests)({}),
795 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
796 namedProjectsRoot: "/projects",
797 isScratchProject: () => Effect.succeed(false),
798 }),
799 NodeCrypto.layer,
800 Scheduler.layer,
801 ),
802 ),
803 );
804 yield* Effect.gen(function* () {
805 const service = yield* ScheduledTaskService.ScheduledTaskService;
806 const save = (enabled: boolean) =>
807 service.upsert({
808 id: ScheduledTaskId.make("task:above-ceiling"),
809 title: "Nightly",
810 prompt: "Check progress",
811 enabled,
812 schedule: { type: "interval", everyMs: 60_000 },
813 projectId: ProjectId.make("project:test"),
814 workspaceStrategy: { type: "root" },
815 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
816 runtimeMode: "full-access",
817 interactionMode: "default",
818 });
819 const refused = yield* save(true).pipe(Effect.flip);
820 assert.equal(refused.message, "Example Corp limits threads to Auto-accept edits.");
821 assert.lengthOf((yield* service.list()).tasks, 0);
822 // A disabled task does not run, so it can still be saved.
823 yield* save(false);
824 assert.lengthOf((yield* service.list()).tasks, 1);
825 // Nor can it be switched back on while its mode is above the ceiling.
826 const reenabled = yield* service
827 .setEnabled({ id: ScheduledTaskId.make("task:above-ceiling"), enabled: true })
828 .pipe(Effect.flip);
829 assert.equal(reenabled.message, "Example Corp limits threads to Auto-accept edits.");
830 // A task bound to a thread runs in that thread's mode, so its own is not checked.
831 const bound = {
832 id: ScheduledTaskId.make("task:bound"),
833 title: "Follow up",
834 prompt: "Check progress",
835 enabled: true,
836 schedule: { type: "interval", everyMs: 60_000 },
837 projectId: ProjectId.make("project:test"),
838 workspaceStrategy: { type: "root" },
839 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
840 runtimeMode: "full-access",
841 interactionMode: "default",
842 } as const;
843 yield* service.upsert({ ...bound, threadId: ThreadId.make("thread:bound") });
844 // Nor when an edit omits the thread it stays bound to.
845 yield* service.upsert(bound);
846 }).pipe(Effect.provide(layer));
847 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
848);
849
850it.effect("checks a task's destination thread only when the destination changes", () =>
851 Effect.gen(function* () {
852 const project = ProjectId.make("project:test");
853 const otherProject = ProjectId.make("project:other");
854 const destination = ThreadId.make("thread:destination");
855 const archivedAt = "2026-11-05T00:00:00.000Z";
856 const shells = yield* Ref.make<ReadonlyMap<ThreadId, never>>(
857 new Map([
858 [destination, threadShell(project)],
859 [ThreadId.make("thread:archived"), threadShell(project, { archivedAt })],
860 [ThreadId.make("thread:deleted"), threadShell(project, { deletedAt: archivedAt })],
861 [ThreadId.make("thread:other-project"), threadShell(otherProject)],
862 ]),
863 );
864 const consumed = yield* Ref.make(0);
865 const layer = ScheduledTaskService.layer.pipe(
866 Layer.provide(
867 Layer.mergeAll(
868 ManagedPolicy.layerTest(),
869 Layer.mock(ThreadLaunchService.ThreadLaunchService)({}),
870 Layer.mock(ThreadManagementService.ThreadManagementService)({
871 getThreadShell: (threadId) =>
872 Ref.get(shells).pipe(Effect.map((all) => all.get(threadId) ?? null)),
873 }),
874 Layer.mock(SecretRequests.SecretRequests)({
875 consume: () => Ref.update(consumed, (count) => count + 1).pipe(Effect.as("secret")),
876 }),
877 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
878 namedProjectsRoot: "/projects",
879 isScratchProject: () => Effect.succeed(false),
880 }),
881 NodeCrypto.layer,
882 Scheduler.layer,
883 ),
884 ),
885 );
886 yield* Effect.gen(function* () {
887 const service = yield* ScheduledTaskService.ScheduledTaskService;
888 const id = ScheduledTaskId.make("task:destination");
889 const save = (
890 threadId: ThreadId,
891 schedule: ScheduledTaskUpsertInput["schedule"] = { type: "interval", everyMs: 60_000 },
892 ) =>
893 service.upsert({
894 id,
895 title: "Follow up",
896 prompt: "Check progress",
897 enabled: true,
898 schedule,
899 projectId: project,
900 threadId,
901 workspaceStrategy: { type: "root" },
902 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
903 runtimeMode: "full-access",
904 interactionMode: "default",
905 });
906
907 for (const [threadId, message] of [
908 ["thread:missing", "That thread no longer exists."],
909 ["thread:deleted", "That thread no longer exists."],
910 ["thread:archived", "That thread is archived. Unarchive it or pick another destination."],
911 ["thread:other-project", "That thread belongs to a different project."],
912 ] as const) {
913 const refused = yield* save(ThreadId.make(threadId)).pipe(Effect.flip);
914 assert.equal(refused.message, message);
915 }
916 assert.lengthOf((yield* service.list()).tasks, 0);
917
918 // A rejected save must not burn a one-shot secret.
919 yield* save(ThreadId.make("thread:missing"), {
920 type: "webhook",
921 signature: {
922 header: "x-signature",
923 encoding: "hex",
924 prefix: "",
925 secretRef: SecretRef.make("secret-ref:00000000000000000000000000000001"),
926 },
927 }).pipe(Effect.flip);
928 assert.equal(yield* Ref.get(consumed), 0);
929
930 yield* save(destination);
931 // A task whose thread was archived later stays editable.
932 yield* Ref.update(shells, (all) =>
933 new Map(all).set(destination, threadShell(project, { archivedAt })),
934 );
935 assert.equal((yield* save(destination)).task.threadId, destination);
936 }).pipe(Effect.provide(layer));
937 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
938);
939
940it.effect("a run interrupted right after it reserves the task does not block the next run", () =>
941 Effect.gen(function* () {
942 const launches = yield* Ref.make(0);
943 const layer = ScheduledTaskService.layer.pipe(
944 Layer.provide(
945 Layer.mergeAll(
946 ManagedPolicy.layerTest(),
947 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
948 launch: () =>
949 Ref.update(launches, (n) => n + 1).pipe(
950 Effect.as({ threadId: ThreadId.make("thread:launched"), resumed: false } as never),
951 ),
952 }),
953 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
954 Layer.mock(SecretRequests.SecretRequests)({}),
955 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
956 namedProjectsRoot: "/projects",
957 isScratchProject: () => Effect.succeed(false),
958 }),
959 NodeCrypto.layer,
960 Layer.succeed(Scheduler.Scheduler, { register: () => Effect.void }),
961 ),
962 ),
963 );
964 yield* Effect.gen(function* () {
965 const service = yield* ScheduledTaskService.ScheduledTaskService;
966 const { task } = yield* service.upsert({
967 id: ScheduledTaskId.make("task:interrupted-reservation"),
968 title: "Check in",
969 prompt: "Check progress",
970 enabled: true,
971 schedule: { type: "interval", everyMs: 60_000 },
972 projectId: ProjectId.make("project:test"),
973 workspaceStrategy: { type: "root" },
974 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
975 runtimeMode: "full-access",
976 interactionMode: "default",
977 });
978
979 // The reservation is the first Set.add of the task id; interrupt the run right there.
980 const add = Set.prototype.add;
981 let interrupted = false;
982 const first = yield* Effect.gen(function* () {
983 using _reserve = vi.spyOn(Set.prototype, "add").mockImplementation(function (
984 this: Set<unknown>,
985 value: unknown,
986 ) {
987 const result = add.call(this, value);
988 if (value === task.id && !interrupted) {
989 interrupted = true;
990 Fiber.getCurrent()?.interruptUnsafe();
991 }
992 return result;
993 });
994 return yield* service
995 .runNow({ id: task.id })
996 .pipe(Effect.forkChild, Effect.flatMap(Fiber.await));
997 });
998 assert.isTrue(interrupted);
999 assert.isTrue(Exit.hasInterrupts(first));
1000
1001 const second = yield* service.runNow({ id: task.id });
1002 assert.equal(second.threadId, ThreadId.make("thread:launched"));
1003 assert.equal(yield* Ref.get(launches), 1);
1004 }).pipe(Effect.provide(layer));
1005 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
1006);