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 } from "@effect/vitest";
6import { ScheduledTaskError } from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Logger from "effect/Logger";
11import * as SqlClient from "effect/sql/SqlClient";
12
13import * as Deferred from "effect/Deferred";
14import * as Ref from "effect/Ref";
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 ScheduledTaskService from "./ScheduledTaskService.ts";
23
24const isScheduledTaskError = Schema.is(ScheduledTaskError);
25
26const insertRow = (
27 sql: SqlClient.SqlClient,
28 row: {
29 id: string | null;
30 next: string | null;
31 enabled: number;
32 status: string;
33 scheduleJson?: string;
34 prompt?: string;
35 },
36 now: string,
37) =>
38 sql`INSERT INTO scheduled_tasks ${sql.insert({
39 task_id: row.id,
40 title: "task",
41 prompt: row.prompt ?? "Run task",
42 enabled: row.enabled,
43 schedule_json: row.scheduleJson ?? '{"type":"interval","everyMs":60000}',
44 project_id: "project:test",
45 thread_id: null,
46 workspace_strategy_json: '{"type":"root"}',
47 model_selection_json: '{"instanceId":"codex","model":"gpt-5"}',
48 runtime_mode: "full-access",
49 interaction_mode: "default",
50 created_by: "user",
51 creation_source: "web",
52 created_at: now,
53 updated_at: now,
54 next_run_at: row.next,
55 last_run_at: null,
56 last_run_status: row.status,
57 last_run_error: null,
58 run_count: 0,
59 })}`;
60
61it.effect("loads only due tasks and skips corrupt due rows without decoding settled tasks", () =>
62 Effect.gen(function* () {
63 const sql = yield* SqlClient.SqlClient;
64 const now = "2026-09-09T12:00:00.000Z";
65 const secret = "PROMPT-SECRET-DO-NOT-LOG";
66 const credential = "CREDENTIAL-SECRET-DO-NOT-LOG";
67 for (const row of [
68 { id: "due-now", next: now, enabled: 1, status: "never" },
69 // Ids that cannot decode as ScheduledTaskId — empty, whitespace-only,
70 // and NULL (SQLite's TEXT primary key admits it).
71 { id: "", next: now, enabled: 1, status: "never" },
72 { id: " ", next: now, enabled: 1, status: "never" },
73 { id: null, next: now, enabled: 1, status: "never" },
74 {
75 id: "due-earlier",
76 next: "2026-09-09T11:00:00.000Z",
77 enabled: 1,
78 status: "failed",
79 },
80 // An unparseable next_run_at sorts before every real timestamp, so it
81 // passes the due filter and would defect the poll if not skipped.
82 { id: "due-bad-date", next: "", enabled: 1, status: "never" },
83 {
84 id: "due-corrupt",
85 next: now,
86 enabled: 1,
87 status: "never",
88 scheduleJson: "broken",
89 },
90 {
91 id: "due-cred",
92 next: now,
93 enabled: 1,
94 status: "never",
95 scheduleJson: `{"type":"interval","everyMs":"${credential}"}`,
96 },
97 { id: "disabled", next: now, enabled: 0, status: "never", scheduleJson: "broken" },
98 {
99 id: "future",
100 next: "2026-09-09T12:00:00.001Z",
101 enabled: 1,
102 status: "never",
103 scheduleJson: "broken",
104 },
105 {
106 id: "unscheduled",
107 next: null,
108 enabled: 1,
109 status: "never",
110 scheduleJson: "broken",
111 },
112 { id: "running", next: now, enabled: 1, status: "running", scheduleJson: "broken" },
113 ]) {
114 yield* insertRow(sql, { ...row, prompt: secret }, now);
115 }
116 const warnings: unknown[] = [];
117 const tasks = yield* ScheduledTaskService.listDueTasks(DateTime.makeUnsafe(now)).pipe(
118 Effect.provide(
119 Logger.layer([
120 Logger.make(({ message }) => {
121 warnings.push(message);
122 }),
123 ]),
124 ),
125 );
126 assert.deepEqual(
127 tasks.map((task) => task.id),
128 ["due-earlier", "due-now"],
129 );
130 // Corrupt rows sort between the healthy due rows; each is skipped with a
131 // warning instead of stopping the poll.
132 assert.equal(warnings.length, 6);
133 const annotations = warnings.map(
134 (warning) => (warning as ReadonlyArray<unknown>)[1] as Record<string, unknown>,
135 );
136 assert.deepEqual(
137 annotations.map((annotation) => annotation.taskId),
138 ["due-bad-date", null, "", " ", "due-corrupt", "due-cred"],
139 );
140 // The typed diagnostic carries a taskId only when the stored id itself
141 // decodes; the corrupt-id rows are still skipped, not fatal.
142 const causes = annotations.map((annotation) => annotation.cause);
143 assert.deepEqual(
144 causes.map((cause) => (isScheduledTaskError(cause) ? cause.taskId : undefined)),
145 [undefined, undefined, undefined, undefined, "due-corrupt", "due-cred"],
146 );
147 for (const cause of causes) {
148 if (cause !== undefined) assert.isTrue(isScheduledTaskError(cause));
149 }
150 const rendered = NodeUtil.inspect(annotations, { depth: null });
151 assert.isFalse(rendered.includes(secret));
152 assert.isFalse(rendered.includes(credential));
153 const running = yield* sql<{
154 last_run_status: string;
155 }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'running'`;
156 assert.equal(running[0]?.last_run_status, "running");
157 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
158);
159
160it.effect(
161 "releases interrupted runs on startup and still executes due tasks, skipping corrupt rows",
162 () =>
163 Effect.gen(function* () {
164 const sql = yield* SqlClient.SqlClient;
165 const now = "2026-09-09T12:00:00.000Z";
166 // An interval that decodes but overflows the representable DateTime
167 // range must not defect recovery or the poll.
168 const hugeInterval = '{"type":"interval","everyMs":9000000000000000}';
169 for (const row of [
170 { id: "stuck-valid", next: now, status: "running" },
171 { id: "stuck-corrupt", next: now, status: "running", scheduleJson: "broken" },
172 { id: "stuck-huge", next: now, status: "running", scheduleJson: hugeInterval },
173 { id: null, next: now, status: "running" },
174 // A due row with an unparseable next_run_at must not defect the poll
175 // before the healthy due tasks run.
176 { id: "due-bad-date", next: "", status: "never" },
177 { id: "due-healthy", next: now, status: "never" },
178 { id: "due-huge", next: now, status: "never", scheduleJson: hugeInterval },
179 { id: "due-second", next: now, status: "never" },
180 { id: "due-third", next: now, status: "never" },
181 ]) {
182 yield* insertRow(sql, { ...row, enabled: 1 }, now);
183 }
184
185 // it.effect freezes the clock; pin it just past the seeded due time so
186 // the poll sees the rows as due.
187 yield* TestClock.setTime(Date.parse(now) + 1_000);
188 // Runs are dispatched serially, so the fourth dispatch can only happen
189 // after the first three runs' completions were recorded — a drain
190 // receipt for the poll fiber, never a sleep. The fourth launch is held
191 // open so its in-flight state is observed deterministically.
192 const dispatched = yield* Ref.make(0);
193 const lastDispatched = yield* Deferred.make<void>();
194 const releaseLast = yield* Deferred.make<void>();
195 yield* Effect.scoped(
196 Effect.gen(function* () {
197 yield* Layer.build(
198 Layer.provideMerge(
199 ScheduledTaskService.layer,
200 Layer.mergeAll(
201 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
202 launch: () =>
203 Ref.updateAndGet(dispatched, (n) => n + 1).pipe(
204 Effect.andThen((n) =>
205 n === 4
206 ? Deferred.succeed(lastDispatched, undefined).pipe(
207 Effect.andThen(Deferred.await(releaseLast)),
208 )
209 : Effect.void,
210 ),
211 Effect.andThen(Effect.die(new Error("test launch failure"))),
212 ),
213 }),
214 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
215 Layer.mock(SecretRequests.SecretRequests)({}),
216 NodeCrypto.layer,
217 Scheduler.layer,
218 ),
219 ),
220 );
221 // The poll loop delays before its first pass, so the frozen test
222 // clock must advance once to trigger it.
223 yield* TestClock.adjust("6 seconds");
224 yield* Deferred.await(lastDispatched);
225 // The fourth dispatch implies the first three completions landed;
226 // markRunning commits before dispatch, so due-third is 'running'.
227 const inflight = yield* sql<{
228 last_run_status: string;
229 }>`SELECT last_run_status FROM scheduled_tasks WHERE task_id = 'due-third'`;
230 assert.equal(inflight[0]?.last_run_status, "running");
231 yield* Deferred.succeed(releaseLast, undefined);
232 }),
233 );
234
235 const rows = yield* sql<{
236 task_id: string | null;
237 last_run_status: string;
238 last_run_error: string | null;
239 next_run_at: string | null;
240 run_count: number;
241 }>`SELECT task_id, last_run_status, last_run_error, next_run_at, run_count
242 FROM scheduled_tasks ORDER BY task_id`;
243 const byId = new Map(rows.map((row) => [row.task_id, row]));
244 assert.equal(rows.length, 9);
245 // The decodable stuck run is rescheduled for its next occurrence; the
246 // undecodable and unrepresentable rows are released without a next run.
247 // A NULL id must still match its own row — `= NULL` never does.
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 // The healthy due tasks were dispatched and recorded failed runs; the
261 // corrupt due row was skipped without running.
262 for (const id of ["due-healthy", "due-huge", "due-second"]) {
263 const row = byId.get(id);
264 assert.equal(row?.last_run_status, "failed");
265 assert.isTrue(row?.last_run_error?.includes("test launch failure") === true);
266 assert.equal(row?.run_count, 1);
267 }
268 assert.isNull(byId.get("due-huge")?.next_run_at);
269 assert.equal(byId.get("due-bad-date")?.last_run_status, "never");
270 assert.equal(byId.get("due-bad-date")?.run_count, 0);
271 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
272);
273
274it.effect(
275 "treats a next_run_at corrupted before the dispatch re-read as not due instead of defecting the poll",
276 () =>
277 Effect.gen(function* () {
278 const sql = yield* SqlClient.SqlClient;
279 const now = "2026-09-09T12:00:00.000Z";
280 // Equal next_run_at means dispatch order follows task_id order.
281 for (const row of [
282 { id: "run-a", next: now, status: "never" },
283 { id: "run-b-victim", next: now, status: "never" },
284 { id: "run-c", next: now, status: "never" },
285 { id: "run-d", next: now, status: "never" },
286 ]) {
287 yield* insertRow(sql, { ...row, enabled: 1 }, now);
288 }
289 yield* TestClock.setTime(Date.parse(now) + 1_000);
290 const dispatched = yield* Ref.make(0);
291 const lastDispatched = yield* Deferred.make<void>();
292 const releaseLast = yield* Deferred.make<void>();
293 yield* Effect.scoped(
294 Effect.gen(function* () {
295 yield* Layer.build(
296 Layer.provideMerge(
297 ScheduledTaskService.layer,
298 Layer.mergeAll(
299 Layer.mock(ThreadLaunchService.ThreadLaunchService)({
300 launch: () =>
301 Ref.updateAndGet(dispatched, (n) => n + 1).pipe(
302 // A concurrent writer (the CLI shares the SQLite file)
303 // corrupts the next row between the poll read and its
304 // dispatch re-read. Serial dispatch makes the
305 // interleaving deterministic.
306 Effect.andThen((n) =>
307 n === 1
308 ? sql`UPDATE scheduled_tasks
309 SET next_run_at = ''
310 WHERE task_id = 'run-b-victim'`.pipe(Effect.asVoid, Effect.orDie)
311 : n === 3
312 ? Deferred.succeed(lastDispatched, undefined).pipe(
313 Effect.andThen(Deferred.await(releaseLast)),
314 )
315 : Effect.void,
316 ),
317 Effect.andThen(Effect.die(new Error("test launch failure"))),
318 ),
319 }),
320 Layer.mock(ThreadManagementService.ThreadManagementService)({}),
321 Layer.mock(SecretRequests.SecretRequests)({}),
322 NodeCrypto.layer,
323 Scheduler.layer,
324 ),
325 ),
326 );
327 yield* TestClock.adjust("6 seconds");
328 // The third dispatch (run-d) can only happen after the victim's
329 // re-read was handled — a receipt that the poll fiber survived.
330 yield* Deferred.await(lastDispatched);
331 assert.equal(yield* Ref.get(dispatched), 3);
332 const rows = yield* sql<{
333 task_id: string;
334 last_run_status: string;
335 next_run_at: string | null;
336 run_count: number;
337 }>`SELECT task_id, last_run_status, next_run_at, run_count
338 FROM scheduled_tasks ORDER BY task_id`;
339 const byId = new Map(rows.map((row) => [row.task_id, row]));
340 const victim = byId.get("run-b-victim");
341 assert.equal(victim?.last_run_status, "never");
342 assert.equal(victim?.run_count, 0);
343 assert.equal(victim?.next_run_at, "");
344 for (const id of ["run-a", "run-c"]) {
345 const row = byId.get(id);
346 assert.equal(row?.last_run_status, "failed");
347 assert.equal(row?.run_count, 1);
348 }
349 yield* Deferred.succeed(releaseLast, undefined);
350 }),
351 );
352 }).pipe(Effect.provide(SqlitePersistence.layerMemory)),
353);