Polish schedules list and editing (#1860)

* feat(schedules): polish schedules list and editing

Adds host filtering, status grouping, and target resolution to the
schedules screen. Extracts a shared HostFilter from sessions. Improves
schedule editing with project targets and model hydration. Handles
missing targets server-side instead of retrying forever.

* test(mcp): update schedule mocks for createOrReplace

* fix(schedules): only show host name when multiple hosts exist
This commit is contained in:
Mohamed Boudra
2026-07-02 12:05:35 +02:00
committed by GitHub
parent 2263469342
commit 6513b56571
21 changed files with 1981 additions and 551 deletions
@@ -2803,12 +2803,14 @@ describe("create_schedule MCP tool", () => {
it("requires provider for schedules", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn(async (input: CreateScheduleInput) => createStoredSchedule(input));
const createOrReplace = vi.fn(async (input: CreateScheduleInput) =>
createStoredSchedule(input),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -2820,17 +2822,19 @@ describe("create_schedule MCP tool", () => {
name: "Default schedule",
}),
).rejects.toThrow("provider is required when target is new-agent");
expect(create).not.toHaveBeenCalled();
expect(createOrReplace).not.toHaveBeenCalled();
});
it("keeps create_schedule provider overrides compatible with provider and provider/model forms", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn(async (input: CreateScheduleInput) => createStoredSchedule(input));
const createOrReplace = vi.fn(async (input: CreateScheduleInput) =>
createStoredSchedule(input),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -2846,7 +2850,7 @@ describe("create_schedule MCP tool", () => {
provider: "codex/gpt-5.4",
});
expect(create).toHaveBeenNthCalledWith(
expect(createOrReplace).toHaveBeenNthCalledWith(
1,
expect.objectContaining({
target: {
@@ -2858,7 +2862,7 @@ describe("create_schedule MCP tool", () => {
},
}),
);
expect(create).toHaveBeenNthCalledWith(
expect(createOrReplace).toHaveBeenNthCalledWith(
2,
expect.objectContaining({
target: {
@@ -2888,12 +2892,14 @@ describe("create_schedule MCP tool", () => {
featureValues: { auto_accept: true },
},
} as ManagedAgent);
const create = vi.fn(async (input: CreateScheduleInput) => createStoredSchedule(input));
const createOrReplace = vi.fn(async (input: CreateScheduleInput) =>
createStoredSchedule(input),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
callerAgentId: "parent-agent",
logger,
});
@@ -2914,14 +2920,14 @@ describe("create_schedule MCP tool", () => {
it("passes timezone through cron create_schedule input", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn(async (scheduleInput: CreateScheduleInput) =>
const createOrReplace = vi.fn(async (scheduleInput: CreateScheduleInput) =>
createStoredSchedule(scheduleInput),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -2933,7 +2939,7 @@ describe("create_schedule MCP tool", () => {
provider: "codex",
});
expect(create).toHaveBeenCalledWith(
expect(createOrReplace).toHaveBeenCalledWith(
expect.objectContaining({
cadence: {
type: "cron",
@@ -2946,12 +2952,12 @@ describe("create_schedule MCP tool", () => {
it("rejects removed create_schedule every input", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn();
const createOrReplace = vi.fn();
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -2963,17 +2969,17 @@ describe("create_schedule MCP tool", () => {
});
expect(parsed.success).toBe(false);
expect(create).not.toHaveBeenCalled();
expect(createOrReplace).not.toHaveBeenCalled();
});
it("rejects create_schedule without cron", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn();
const createOrReplace = vi.fn();
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -2985,17 +2991,17 @@ describe("create_schedule MCP tool", () => {
}),
).rejects.toThrow(/cron/);
expect(create).not.toHaveBeenCalled();
expect(createOrReplace).not.toHaveBeenCalled();
});
it.each(["", " "])("rejects create_schedule blank timezone %#", async (timezone) => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn();
const createOrReplace = vi.fn();
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_schedule");
@@ -3009,7 +3015,7 @@ describe("create_schedule MCP tool", () => {
}),
).rejects.toThrow();
expect(create).not.toHaveBeenCalled();
expect(createOrReplace).not.toHaveBeenCalled();
});
});
@@ -3027,12 +3033,14 @@ describe("create_heartbeat MCP tool", () => {
availableModes: [],
config: { title: "Parent agent" },
} as ManagedAgent);
const create = vi.fn(async (input: CreateScheduleInput) => createStoredSchedule(input));
const createOrReplace = vi.fn(async (input: CreateScheduleInput) =>
createStoredSchedule(input),
);
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
callerAgentId: "parent-agent",
logger,
});
@@ -3045,7 +3053,7 @@ describe("create_heartbeat MCP tool", () => {
name: "status heartbeat",
});
expect(create).toHaveBeenCalledWith(
expect(createOrReplace).toHaveBeenCalledWith(
expect.objectContaining({
prompt: "check status",
cadence: {
@@ -3061,12 +3069,12 @@ describe("create_heartbeat MCP tool", () => {
it("requires an agent-scoped session", async () => {
const { agentManager, agentStorage } = createTestDeps();
const create = vi.fn();
const createOrReplace = vi.fn();
const server = await createAgentMcpServer({
agentManager,
agentStorage,
providerSnapshotManager: createOpenCodeManager().manager,
scheduleService: { create } as unknown as ScheduleService,
scheduleService: { createOrReplace } as unknown as ScheduleService,
logger,
});
const tool = registeredTool(server, "create_heartbeat");
@@ -3078,7 +3086,7 @@ describe("create_heartbeat MCP tool", () => {
}),
).rejects.toThrow("create_heartbeat requires an agent-scoped session");
expect(create).not.toHaveBeenCalled();
expect(createOrReplace).not.toHaveBeenCalled();
});
});
@@ -2009,7 +2009,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
}
const expiresAt = buildScheduleExpiry(expiresIn);
const schedule = await scheduleService.create({
const schedule = await scheduleService.createOrReplace({
prompt: prompt.trim(),
cadence: buildCronScheduleCadence({
cron,
@@ -2058,7 +2058,7 @@ export function createPaseoToolCatalog(options: PaseoToolHostDependencies): Pase
resolveCallerAgent();
const expiresAt = buildScheduleExpiry(expiresIn);
const schedule = await scheduleService.create({
const schedule = await scheduleService.createOrReplace({
prompt: prompt.trim(),
cadence: buildCronScheduleCadence({
cron,
+2 -2
View File
@@ -818,9 +818,9 @@ export async function createPaseoDaemon(
await scheduleService.start();
agentManager.setAgentArchivedCallback(async (agentId) => {
try {
await scheduleService.deleteForAgent(agentId);
await scheduleService.completeForAgent(agentId);
} catch (error) {
logger.warn({ err: error, agentId }, "Failed to delete schedules for archived agent");
logger.warn({ err: error, agentId }, "Failed to complete schedules for archived agent");
}
});
logger.info({ elapsed: elapsed() }, "Schedule service initialized");
@@ -22,7 +22,7 @@ import type {
import { createTestAgentClients } from "../test-utils/fake-agent-client.js";
import { createTestLogger } from "../../test-utils/test-logger.js";
import type { ProviderSnapshotManager } from "../agent/provider-snapshot-manager.js";
import { ScheduleService } from "./service.js";
import { ScheduleService, ScheduleTargetGoneError } from "./service.js";
import type { ScheduleExecutionResult, StoredSchedule } from "@getpaseo/protocol/schedule/types";
interface ScheduleServiceInternals {
@@ -45,6 +45,36 @@ const NO_UNATTENDED_SCHEDULE_POLICY: Pick<ProviderSnapshotManager, "resolveCreat
},
};
function buildAgentRecord(params: {
id: string;
cwd: string;
iso: string;
archivedAt?: string | null;
}) {
return {
id: params.id,
provider: "claude",
cwd: params.cwd,
createdAt: params.iso,
updatedAt: params.iso,
lastActivityAt: params.iso,
lastUserMessageAt: null,
title: params.id,
labels: {},
lastStatus: "closed" as const,
lastModeId: "default",
config: { modeId: "default" },
runtimeInfo: null,
features: [],
persistence: null,
requiresAttention: false,
attentionReason: null,
attentionTimestamp: null,
internal: false,
archivedAt: params.archivedAt ?? null,
};
}
describe("ScheduleService", () => {
let tempDir: string;
let agentStorage: AgentStorage;
@@ -1266,7 +1296,7 @@ describe("ScheduleService", () => {
await expect(service.runOnce(created.id)).rejects.toThrow("already completed");
});
test("deleteForAgent removes only schedules targeting that agent", async () => {
test("completeForAgent completes only schedules targeting that agent", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
@@ -1299,12 +1329,577 @@ describe("ScheduleService", () => {
},
});
const deleted = await service.deleteForAgent(targetAgentId);
expect(deleted).toBe(1);
now = new Date("2026-01-01T00:05:00.000Z");
const completed = await service.completeForAgent(targetAgentId);
expect(completed).toBe(1);
const remaining = await service.list();
const remainingIds = remaining.map((schedule) => schedule.id).sort();
expect(remainingIds).toEqual([otherTargeted.id, newAgentSchedule.id].sort());
expect(remainingIds).not.toContain(targeted.id);
expect(remaining.map((schedule) => schedule.id).sort()).toEqual(
[targeted.id, otherTargeted.id, newAgentSchedule.id].sort(),
);
const doomed = await service.inspect(targeted.id);
expect(doomed.status).toBe("completed");
expect(doomed.nextRunAt).toBeNull();
expect(doomed.updatedAt).toBe("2026-01-01T00:05:00.000Z");
expect((await service.inspect(otherTargeted.id)).status).toBe("active");
expect((await service.inspect(newAgentSchedule.id)).status).toBe("active");
});
test("startup sweep completes agent-target schedules whose agent is gone", async () => {
const missingAgentId = "44444444-4444-4444-8444-444444444444";
const archivedAgentId = "55555555-5555-4555-8555-555555555555";
const liveAgentId = "66666666-6666-4666-8666-666666666666";
await agentStorage.upsert(
buildAgentRecord({
id: archivedAgentId,
cwd: tempDir,
iso: now.toISOString(),
archivedAt: "2026-01-01T00:00:30.000Z",
}),
);
await agentStorage.upsert(
buildAgentRecord({ id: liveAgentId, cwd: tempDir, iso: now.toISOString() }),
);
const service1 = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const missing = await service1.create({
prompt: "ping missing",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: missingAgentId },
});
const archived = await service1.create({
prompt: "ping archived",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: archivedAgentId },
});
const live = await service1.create({
prompt: "ping live",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: liveAgentId },
});
const newAgent = await service1.create({
prompt: "spawn fresh",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } },
});
const pausedMissing = await service1.create({
prompt: "paused ping missing",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: "a1a1a1a1-a1a1-4a1a-8a1a-a1a1a1a1a1a1" },
});
await service1.pause(pausedMissing.id);
const pausedLive = await service1.create({
prompt: "paused ping live",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: liveAgentId },
});
await service1.pause(pausedLive.id);
now = new Date("2026-01-01T00:10:00.000Z");
const service2 = new ScheduleService({
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();
await service2.stop();
expect((await service2.inspect(missing.id)).status).toBe("completed");
expect((await service2.inspect(missing.id)).nextRunAt).toBeNull();
expect((await service2.inspect(archived.id)).status).toBe("completed");
expect((await service2.inspect(live.id)).status).toBe("active");
expect((await service2.inspect(newAgent.id)).status).toBe("active");
// Paused schedules are swept too when their agent is gone, but survive when it lives.
expect((await service2.inspect(pausedMissing.id)).status).toBe("completed");
expect((await service2.inspect(pausedLive.id)).status).toBe("paused");
});
test("completes the schedule when a scheduled run reports the target is gone", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => {
throw new ScheduleTargetGoneError("Agent 77777777-7777-4777-8777-777777777777 is archived");
},
});
const created = await service.create({
prompt: "ping gone target",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: "77777777-7777-4777-8777-777777777777" },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs).toHaveLength(1);
expect(inspected.runs[0]?.status).toBe("failed");
expect(inspected.runs[0]?.error).toBe("Agent 77777777-7777-4777-8777-777777777777 is archived");
});
test("does not resurrect nextRunAt when the schedule completes during an in-flight run", async () => {
const agentId = "ffffffff-ffff-4fff-8fff-ffffffffffff";
let service!: ScheduleService;
service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => {
// Simulate the agent being archived mid-run: the archive callback
// completes the schedule before this run finishes.
await service.completeForAgent(agentId);
return { agentId: null, output: "ok" };
},
});
const created = await service.create({
prompt: "ping",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs[0]?.status).toBe("succeeded");
});
test("keeps the schedule active when a run fails for a transient reason", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => {
throw new Error("network blip");
},
});
const created = await service.create({
prompt: "ping flaky target",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: "88888888-8888-4888-8888-888888888888" },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("active");
expect(inspected.nextRunAt).toBe("2026-01-01T00:02:00.000Z");
expect(inspected.runs[0]?.status).toBe("failed");
expect(inspected.runs[0]?.error).toBe("network blip");
});
test("completes the schedule when a scheduled run targets an archived agent", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const archivedAgentId = "99999999-9999-4999-8999-999999999999";
await agentStorage.upsert(
buildAgentRecord({
id: archivedAgentId,
cwd: tempDir,
iso: now.toISOString(),
archivedAt: "2026-01-01T00:00:30.000Z",
}),
);
const created = await service.create({
prompt: "ping archived target",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: archivedAgentId },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs[0]?.status).toBe("failed");
expect(inspected.runs[0]?.error).toContain("is archived");
});
test("completes the schedule when a scheduled run targets a missing agent", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const created = await service.create({
prompt: "ping missing target",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs[0]?.status).toBe("failed");
});
test("completes the schedule when a new-agent run's cwd no longer exists", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const created = await service.create({
prompt: "spawn in a deleted dir",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: join(tempDir, "deleted-worktree"),
approvalPolicy: "never",
},
},
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("completed");
expect(inspected.nextRunAt).toBeNull();
expect(inspected.runs[0]?.status).toBe("failed");
expect(inspected.runs[0]?.error).toContain("no longer exists");
});
test("keeps the schedule active when a real run fails for a non-gone reason", async () => {
// No providers registered: the agent exists and is live, but loading it fails
// with a plain error (not ScheduleTargetGoneError), so the schedule must retry.
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const agentId = "12121212-1212-4121-8121-121212121212";
await agentStorage.upsert(
buildAgentRecord({ id: agentId, cwd: tempDir, iso: now.toISOString() }),
);
const created = await service.create({
prompt: "ping live but unavailable",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
const inspected = await service.inspect(created.id);
expect(inspected.status).toBe("active");
expect(inspected.nextRunAt).toBe("2026-01-01T00:02:00.000Z");
expect(inspected.runs[0]?.status).toBe("failed");
expect(inspected.runs[0]?.error).toContain("unavailable provider");
});
test("runOnce completes the schedule when the target is gone", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
});
const created = await service.create({
prompt: "manual ping gone target",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: "13131313-1313-4131-8131-131313131313" },
});
const after = await service.runOnce(created.id);
expect(after.status).toBe("completed");
expect(after.nextRunAt).toBeNull();
expect(after.runs).toHaveLength(1);
expect(after.runs[0]?.status).toBe("failed");
});
test("createOrReplace updates the matching schedule in place instead of duplicating", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const agentId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
const first = await service.createOrReplace({
name: "babysit-pr PR 1112",
prompt: "watch the build",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
expect((await service.inspect(first.id)).runs).toHaveLength(1);
await service.pause(first.id);
now = new Date("2026-01-01T00:02:00.000Z");
const second = await service.createOrReplace({
name: "babysit-pr PR 1112",
prompt: "watch the build v2",
cadence: { type: "cron", expression: "30 9 * * *" },
target: { type: "agent", agentId },
});
expect(second.id).toBe(first.id);
expect(second.status).toBe("active");
expect(second.prompt).toBe("watch the build v2");
expect(second.cadence).toEqual({ type: "cron", expression: "30 9 * * *" });
expect(second.nextRunAt).toBe("2026-01-01T09:30:00.000Z");
expect(second.createdAt).toBe(first.createdAt);
expect(second.runs).toHaveLength(1);
expect(await service.list()).toHaveLength(1);
});
test("createOrReplace creates a sibling when name, target, or completion differ", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const agentA = "cccccccc-cccc-4ccc-8ccc-cccccccccccc";
const agentB = "dddddddd-dddd-4ddd-8ddd-dddddddddddd";
await service.createOrReplace({
name: "dup",
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: agentA },
});
await service.createOrReplace({
name: "other",
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: agentA },
});
await service.createOrReplace({
name: "dup",
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: agentB },
});
const done = await service.createOrReplace({
name: "done",
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: agentA },
maxRuns: 1,
});
now = new Date("2026-01-01T00:01:00.000Z");
await service.tick();
expect((await service.inspect(done.id)).status).toBe("completed");
const redone = await service.createOrReplace({
name: "done",
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId: agentA },
});
expect(redone.id).not.toBe(done.id);
expect(await service.list()).toHaveLength(5);
});
test("createOrReplace never dedups anonymous schedules", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const agentId = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee";
await service.createOrReplace({
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
});
await service.createOrReplace({
prompt: "p",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
});
expect(await service.list()).toHaveLength(2);
});
test("createOrReplace matches new-agent targets by config", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const first = await service.createOrReplace({
name: "nightly",
prompt: "audit",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } },
});
const second = await service.createOrReplace({
name: "nightly",
prompt: "audit v2",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "new-agent", config: { provider: "claude", cwd: tempDir } },
});
expect(second.id).toBe(first.id);
expect(await service.list()).toHaveLength(1);
const third = await service.createOrReplace({
name: "nightly",
prompt: "audit elsewhere",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "new-agent", config: { provider: "claude", cwd: join(tempDir, "sub") } },
});
expect(third.id).not.toBe(first.id);
expect(await service.list()).toHaveLength(2);
});
test("createOrReplace dedups new-agent targets regardless of config key order", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
// The stored config is round-tripped through the Zod schema (schema key
// order); this incoming literal deliberately uses a different key order.
const first = await service.createOrReplace({
name: "nightly",
prompt: "audit",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
networkAccess: true,
title: "nightly job",
approvalPolicy: "never",
},
},
});
const second = await service.createOrReplace({
name: "nightly",
prompt: "audit v2",
cadence: { type: "every", everyMs: 60_000 },
target: {
type: "new-agent",
config: {
provider: "claude",
cwd: tempDir,
networkAccess: true,
title: "nightly job",
approvalPolicy: "never",
},
},
});
expect(second.id).toBe(first.id);
expect(await service.list()).toHaveLength(1);
});
test("completeForAgent skips schedules that are already completed", async () => {
const service = new ScheduleService({
paseoHome: tempDir,
logger: createTestLogger(),
agentManager: new AgentManager({ logger: createTestLogger() }),
agentStorage,
providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY,
now: () => now,
runner: async () => ({ agentId: null, output: "ok" }),
});
const agentId = "33333333-3333-4333-8333-333333333333";
await service.create({
prompt: "already done",
cadence: { type: "every", everyMs: 60_000 },
target: { type: "agent", agentId },
maxRuns: 1,
});
now = new Date("2026-01-01T00:01:00.000Z");
expect(await service.completeForAgent(agentId)).toBe(1);
expect(await service.completeForAgent(agentId)).toBe(0);
});
});
+134 -10
View File
@@ -1,4 +1,5 @@
import { randomUUID } from "node:crypto";
import { stat } from "node:fs/promises";
import { join } from "node:path";
import type { Logger } from "pino";
import { AgentManager } from "../agent/agent-manager.js";
@@ -27,6 +28,16 @@ import type {
const SCHEDULE_TICK_INTERVAL_MS = 1000;
// A run failed because its target no longer exists: the agent was deleted or
// archived, or a new-agent cwd was removed. These are permanent, so the schedule
// is completed instead of retried until it burns down to its expiry.
export class ScheduleTargetGoneError extends Error {
constructor(message: string) {
super(message);
this.name = "ScheduleTargetGoneError";
}
}
function trimOptionalName(value: string | null | undefined): string | null {
if (typeof value !== "string") {
return null;
@@ -112,6 +123,35 @@ function shouldCompleteSchedule(schedule: StoredSchedule, now: Date): boolean {
return countCompletedRuns(schedule) >= schedule.maxRuns;
}
// Sort object keys recursively so two structurally-equal configs serialize
// identically regardless of key order. Stored configs come back from disk in Zod
// schema order while incoming configs keep their construction order, so a plain
// JSON.stringify comparison would wrongly treat identical targets as different.
function canonicalize(value: unknown): unknown {
if (Array.isArray(value)) {
return value.map(canonicalize);
}
if (value && typeof value === "object") {
const source = value as Record<string, unknown>;
return Object.fromEntries(
Object.keys(source)
.sort()
.map((key) => [key, canonicalize(source[key])]),
);
}
return value;
}
function scheduleTargetsEqual(a: ScheduleTarget, b: ScheduleTarget): boolean {
if (a.type === "agent" && b.type === "agent") {
return a.agentId === b.agentId;
}
if (a.type === "new-agent" && b.type === "new-agent") {
return JSON.stringify(canonicalize(a.config)) === JSON.stringify(canonicalize(b.config));
}
return false;
}
function completeSchedule(schedule: StoredSchedule, now: Date): StoredSchedule {
return {
...schedule,
@@ -177,6 +217,7 @@ export class ScheduleService {
async start(): Promise<void> {
await this.recoverInterruptedRuns();
await this.sweepOrphanedSchedules();
if (this.tickTimer) {
return;
}
@@ -220,6 +261,44 @@ export class ScheduleService {
return schedule;
}
// Idempotent create for the MCP write path: repeating a create with the same
// name and target (e.g. babysit-pr re-registering its heartbeat) refreshes the
// existing non-completed schedule in place instead of minting a duplicate.
async createOrReplace(input: CreateScheduleInput): Promise<StoredSchedule> {
const name = trimOptionalName(input.name);
if (name !== null) {
const existing = (await this.store.list()).find(
(schedule) =>
schedule.status !== "completed" &&
trimOptionalName(schedule.name) === name &&
scheduleTargetsEqual(schedule.target, input.target),
);
if (existing) {
const now = this.now();
const prompt = normalizePrompt(input.prompt);
validateScheduleCadence(input.cadence);
const runOnCreate = input.runOnCreate ?? input.cadence.type === "every";
const nextRunAt = runOnCreate ? now : computeNextRunAt(input.cadence, now);
const replaced: StoredSchedule = {
...existing,
name,
prompt,
cadence: input.cadence,
target: input.target,
status: "active",
pausedAt: null,
nextRunAt: nextRunAt.toISOString(),
expiresAt: input.expiresAt ?? null,
maxRuns: normalizeMaxRuns(input.maxRuns),
updatedAt: now.toISOString(),
};
await this.store.put(replaced);
return replaced;
}
}
return this.create(input);
}
async list(): Promise<StoredSchedule[]> {
return this.store.list();
}
@@ -321,26 +400,30 @@ export class ScheduleService {
await this.store.delete(id);
}
async deleteForAgent(agentId: string): Promise<number> {
async completeForAgent(agentId: string): Promise<number> {
const now = this.now();
const schedules = await this.store.list();
const matches = schedules.filter(
(schedule) => schedule.target.type === "agent" && schedule.target.agentId === agentId,
(schedule) =>
schedule.target.type === "agent" &&
schedule.target.agentId === agentId &&
schedule.status !== "completed",
);
const results = await Promise.allSettled(
matches.map((schedule) => this.store.delete(schedule.id)),
matches.map((schedule) => this.store.put(completeSchedule(schedule, now))),
);
let deleted = 0;
let completed = 0;
for (const [index, result] of results.entries()) {
if (result.status === "fulfilled") {
deleted += 1;
completed += 1;
} else {
this.logger.warn(
{ err: result.reason, scheduleId: matches[index].id, agentId },
"Failed to delete schedule for archived agent; continuing",
"Failed to complete schedule for archived agent; continuing",
);
}
}
return deleted;
return completed;
}
async runOnce(id: string): Promise<StoredSchedule> {
@@ -420,6 +503,26 @@ export class ScheduleService {
);
}
// Orphaned agent-target schedules (agent deleted while the daemon was down, or
// archived before completeForAgent existed) can never fire successfully. Complete
// them on startup so they stop ticking and surface as ended in the UI.
private async sweepOrphanedSchedules(): Promise<void> {
const now = this.now();
const schedules = await this.store.list();
await Promise.all(
schedules.map(async (schedule) => {
if (schedule.target.type !== "agent" || schedule.status === "completed") {
return;
}
const record = await this.agentStorage.get(schedule.target.agentId);
if (record && !record.archivedAt) {
return;
}
await this.store.put(completeSchedule(schedule, now));
}),
);
}
private async runSchedule(
schedule: StoredSchedule,
now: Date,
@@ -454,6 +557,7 @@ export class ScheduleService {
agentId: result.agentId,
output: result.output,
error: null,
targetGone: false,
manual,
});
} catch (error) {
@@ -464,6 +568,7 @@ export class ScheduleService {
agentId: null,
output: null,
error: error instanceof Error ? error.message : String(error),
targetGone: error instanceof ScheduleTargetGoneError,
manual,
});
} finally {
@@ -478,6 +583,7 @@ export class ScheduleService {
agentId: string | null;
output: string | null;
error: string | null;
targetGone: boolean;
manual: boolean;
}): Promise<void> {
const schedule = await this.inspect(params.scheduleId);
@@ -501,7 +607,14 @@ export class ScheduleService {
updatedAt: now.toISOString(),
};
if (params.manual) {
if (params.targetGone) {
// The target is permanently gone; retrying only burns the schedule down to
// its expiry, so complete it now regardless of manual/scheduled origin.
updated = completeSchedule(updated, now);
} else if (updated.status === "completed") {
// Completed concurrently (e.g. the target agent was archived mid-run);
// record the run outcome but leave the schedule terminal — don't advance.
} else if (params.manual) {
// Manual one-shot runs do not advance the cadence or recompute completion.
} else if (shouldCompleteSchedule(updated, now)) {
updated = completeSchedule(updated, now);
@@ -532,8 +645,11 @@ export class ScheduleService {
if (schedule.target.type === "agent") {
const wrappedPrompt = formatSystemNotificationPrompt(buildScheduleFireBody(schedule, runId));
const record = await this.agentStorage.get(schedule.target.agentId);
if (record?.archivedAt) {
throw new Error(`Agent ${schedule.target.agentId} is archived`);
if (!record) {
throw new ScheduleTargetGoneError(`Agent ${schedule.target.agentId} no longer exists`);
}
if (record.archivedAt) {
throw new ScheduleTargetGoneError(`Agent ${schedule.target.agentId} is archived`);
}
const agent = await ensureAgentLoaded(schedule.target.agentId, {
@@ -557,6 +673,14 @@ export class ScheduleService {
}
const targetConfig = schedule.target.config;
try {
await stat(targetConfig.cwd);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "ENOENT") {
throw new ScheduleTargetGoneError(`Working directory ${targetConfig.cwd} no longer exists`);
}
throw error;
}
const resolvedUnattendedConfig = targetConfig.modeId
? { modeId: targetConfig.modeId, featureValues: targetConfig.featureValues }
: await this.resolveProviderCreateConfig({