mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-23 17:52:49 +00:00
Start the daemon when the schedules directory holds a file that is not a schedule (#5301)
* Start the daemon when the schedules directory holds a file that is not a schedule One unreadable .json file in the schedules directory rejected ScheduleStore.list(), which ScheduleService.start() awaits during daemon bootstrap, so the daemon never started. The store now skips a file it cannot read as a schedule and logs its path once, the way agent storage treats an invalid agent record. Closes #4745 * Skip only schedule files with invalid content A schedule file that cannot be read now fails the listing as before, instead of being skipped as if it were not a schedule. An invalid file is reported again when it reappears after being removed or fixed. * Retrigger CI after an unrelated macOS observation-test flake --------- Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
This commit is contained in:
@@ -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<void>((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: [
|
||||
|
||||
@@ -264,8 +264,8 @@ export class ScheduleService {
|
||||
private tickTimer: ReturnType<typeof setInterval> | 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;
|
||||
|
||||
@@ -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 },
|
||||
|
||||
@@ -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<string, Promise<unknown>>();
|
||||
private readonly identityMutations = new Map<string, Promise<unknown>>();
|
||||
private reportedInvalidFiles = new Set<string>();
|
||||
|
||||
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<StoredSchedule[]> {
|
||||
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<string>();
|
||||
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));
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user