diff --git a/packages/server/src/server/schedule/service.test.ts b/packages/server/src/server/schedule/service.test.ts index 224df7cd3f..612bb7e9a9 100644 --- a/packages/server/src/server/schedule/service.test.ts +++ b/packages/server/src/server/schedule/service.test.ts @@ -1,4 +1,4 @@ -import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { tmpdir } from "node:os"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; @@ -803,7 +803,7 @@ describe("ScheduleService", () => { const runStarted = new Promise((resolve) => { releaseRun = resolve; }); - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore(join(tempDir, "schedules"), createTestLogger()); const legacy = await store.create({ name: null, prompt: "finish/update race", @@ -1902,6 +1902,51 @@ describe("ScheduleService", () => { await service2.stop(); }); + test("starts with the valid schedules when the schedules directory holds files that are not schedules", async () => { + const service1 = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async () => ({ agentId: null, output: "ok" }), + }); + const created = await service1.create({ + prompt: "Still scheduled", + cadence: { type: "every", everyMs: 60_000 }, + target: { + type: "new-agent", + config: { provider: "claude", cwd: tempDir }, + }, + runOnCreate: false, + }); + await service1.stop(); + + const schedulesDir = join(tempDir, "schedules"); + await writeFile(join(schedulesDir, "notes.json"), JSON.stringify({ hello: "world" })); + const { lastRunAt: _omitted, ...withoutLastRunAt } = created; + await writeFile( + join(schedulesDir, "deadbeef.json"), + JSON.stringify({ ...withoutLastRunAt, id: "deadbeef" }), + ); + await writeFile(join(schedulesDir, "broken.json"), "{ not json"); + + const service2 = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: new AgentManager({ logger: createTestLogger() }), + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + runner: async () => ({ agentId: null, output: "ok" }), + }); + await service2.start(); + + expect((await service2.list()).map((schedule) => schedule.id)).toEqual([created.id]); + await service2.stop(); + }); + test("startup recovery archives an interrupted run workspace with an associated agent", async () => { const service1 = createScheduleService({ paseoHome: tempDir, @@ -1926,7 +1971,7 @@ describe("ScheduleService", () => { const interruptedAt = now.toISOString(); const associatedAgentId = "11111111-1111-4111-8111-111111111111"; const workspaceId = "wks_interrupted_with_agent"; - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore(join(tempDir, "schedules"), createTestLogger()); await store.update(created.id, (schedule) => ({ ...schedule, runs: [ @@ -1994,7 +2039,7 @@ describe("ScheduleService", () => { const interruptedAt = now.toISOString(); const workspaceId = "wks_interrupted_without_agent"; - const store = new ScheduleStore(join(tempDir, "schedules")); + const store = new ScheduleStore(join(tempDir, "schedules"), createTestLogger()); await store.update(created.id, (schedule) => ({ ...schedule, runs: [ diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 192b424337..53d2432f88 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -264,8 +264,8 @@ export class ScheduleService { private tickTimer: ReturnType | null = null; constructor(options: ScheduleServiceOptions) { - this.store = new ScheduleStore(join(options.paseoHome, "schedules")); this.logger = options.logger.child({ module: "schedule-service" }); + this.store = new ScheduleStore(join(options.paseoHome, "schedules"), this.logger); this.agentManager = options.agentManager; this.agentStorage = options.agentStorage; this.createAgent = options.createAgent; diff --git a/packages/server/src/server/schedule/store.test.ts b/packages/server/src/server/schedule/store.test.ts index 7e19bbfd92..5f69119d4d 100644 --- a/packages/server/src/server/schedule/store.test.ts +++ b/packages/server/src/server/schedule/store.test.ts @@ -1,7 +1,10 @@ -import { mkdtemp, rm } from "node:fs/promises"; +import { chmod, mkdtemp, rm, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { tmpdir } from "node:os"; +import { Writable } from "node:stream"; +import pino from "pino"; import { afterEach, beforeEach, describe, expect, test } from "vitest"; +import { createTestLogger } from "../../test-utils/test-logger.js"; import { ScheduleStore } from "./store.js"; describe("ScheduleStore", () => { @@ -10,7 +13,7 @@ describe("ScheduleStore", () => { beforeEach(async () => { tempDir = await mkdtemp(join(tmpdir(), "schedule-store-test-")); - store = new ScheduleStore(tempDir); + store = new ScheduleStore(tempDir, createTestLogger()); }); afterEach(async () => { @@ -40,13 +43,66 @@ describe("ScheduleStore", () => { runs: [], }); - const reloaded = new ScheduleStore(tempDir); + const reloaded = new ScheduleStore(tempDir, createTestLogger()); const listed = await reloaded.list(); expect(created.id).toHaveLength(8); expect(listed).toEqual([created]); }); + test("reports an invalid schedule file once while it stays invalid, by name, and lists the rest", async () => { + const created = await store.create({ + name: "Morning summary", + prompt: "Summarize new commits", + cadence: { type: "every", everyMs: 60_000 }, + target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } }, + status: "active", + createdAt: "2026-01-01T00:00:00.000Z", + updatedAt: "2026-01-01T00:00:00.000Z", + nextRunAt: "2026-01-01T00:01:00.000Z", + lastRunAt: null, + pausedAt: null, + expiresAt: null, + maxRuns: null, + runs: [], + }); + await writeFile(join(tempDir, "notes.json"), JSON.stringify({ hello: "world" })); + const logLines: Array<{ msg: string; filePath?: string }> = []; + const logger = pino( + { level: "error" }, + new Writable({ + write(chunk, _encoding, callback) { + logLines.push(JSON.parse(chunk.toString("utf8"))); + callback(); + }, + }), + ); + const reloaded = new ScheduleStore(tempDir, logger); + + expect(await reloaded.list()).toEqual([created]); + expect(await reloaded.list()).toEqual([created]); + await rm(join(tempDir, "notes.json")); + expect(await reloaded.list()).toEqual([created]); + await writeFile(join(tempDir, "notes.json"), "{ not json"); + expect(await reloaded.list()).toEqual([created]); + + const skipped = { + msg: "Skipping invalid schedule file", + filePath: join(tempDir, "notes.json"), + }; + expect(logLines.map(({ msg, filePath }) => ({ msg, filePath }))).toEqual([skipped, skipped]); + }); + + test.skipIf(process.platform === "win32" || process.getuid?.() === 0)( + "fails the listing when a schedule file cannot be read", + async () => { + await writeFile(join(tempDir, "unreadable.json"), "{}"); + await chmod(join(tempDir, "unreadable.json"), 0o000); + + await expect(store.list()).rejects.toMatchObject({ code: "EACCES" }); + }, + ); + test("update round-trips an updated schedule to disk", async () => { const created = await store.create({ name: "before", @@ -81,7 +137,7 @@ describe("ScheduleStore", () => { }; await store.update(created.id, () => updated); - const reloaded = await new ScheduleStore(tempDir).get(created.id); + const reloaded = await new ScheduleStore(tempDir, createTestLogger()).get(created.id); expect(reloaded).toEqual(updated); }); @@ -182,7 +238,9 @@ describe("ScheduleStore", () => { prompt: "after", runs: [{ id: "run-1" }], }); - await expect(new ScheduleStore(tempDir).get(created.id)).resolves.toMatchObject({ + await expect( + new ScheduleStore(tempDir, createTestLogger()).get(created.id), + ).resolves.toMatchObject({ prompt: "after", runs: [{ id: "run-1" }], }); @@ -211,7 +269,7 @@ describe("ScheduleStore", () => { } } - const gatedStore = new GatedListScheduleStore(tempDir); + const gatedStore = new GatedListScheduleStore(tempDir, createTestLogger()); const target = { type: "new-agent" as const, config: { provider: "claude" as const, cwd: tempDir }, diff --git a/packages/server/src/server/schedule/store.ts b/packages/server/src/server/schedule/store.ts index f3583d5f70..6feee5bc85 100644 --- a/packages/server/src/server/schedule/store.ts +++ b/packages/server/src/server/schedule/store.ts @@ -1,6 +1,7 @@ import { randomBytes } from "node:crypto"; import { mkdir, readFile, readdir, rm } from "node:fs/promises"; import { join } from "node:path"; +import type { Logger } from "pino"; import { StoredScheduleSchema, type ScheduleTarget, @@ -87,11 +88,27 @@ function matchesNameAndTarget( ); } +function parseStoredSchedule( + content: string, +): { success: true; data: StoredSchedule } | { success: false; error: unknown } { + let json: unknown; + try { + json = JSON.parse(content); + } catch (error) { + return { success: false, error }; + } + return StoredScheduleSchema.safeParse(json); +} + export class ScheduleStore { private readonly scheduleMutations = new Map>(); private readonly identityMutations = new Map>(); + private reportedInvalidFiles = new Set(); - constructor(private readonly dir: string) {} + constructor( + private readonly dir: string, + private readonly logger: Logger, + ) {} private filePath(id: string): string { return join(this.dir, `${id}.json`); @@ -101,17 +118,32 @@ export class ScheduleStore { await mkdir(this.dir, { recursive: true }); } + // The service lists schedules on every tick, so a file that is not a valid schedule is + // reported when it first appears rather than once per second. async list(): Promise { await this.ensureDir(); const entries = await readdir(this.dir, { withFileTypes: true }); - const schedules = await Promise.all( + const files = await Promise.all( entries .filter((entry) => entry.isFile() && entry.name.endsWith(".json")) .map(async (entry) => { - const content = await readFile(join(this.dir, entry.name), "utf-8"); - return StoredScheduleSchema.parse(JSON.parse(content)); + const filePath = join(this.dir, entry.name); + return { filePath, parsed: parseStoredSchedule(await readFile(filePath, "utf-8")) }; }), ); + const schedules: StoredSchedule[] = []; + const invalidFiles = new Set(); + for (const { filePath, parsed } of files) { + if (parsed.success) { + schedules.push(parsed.data); + continue; + } + invalidFiles.add(filePath); + if (!this.reportedInvalidFiles.has(filePath)) { + this.logger.error({ err: parsed.error, filePath }, "Skipping invalid schedule file"); + } + } + this.reportedInvalidFiles = invalidFiles; return schedules.sort((left, right) => left.createdAt.localeCompare(right.createdAt)); }