diff --git a/packages/server/src/server/agent/agent-prompt.test.ts b/packages/server/src/server/agent/agent-prompt.test.ts index af291256e..89b175d13 100644 --- a/packages/server/src/server/agent/agent-prompt.test.ts +++ b/packages/server/src/server/agent/agent-prompt.test.ts @@ -53,6 +53,7 @@ interface FinishNotificationScenario { finishChildAndReadParentPrompt(): Promise; closeChildAndReadParentPrompt(): Promise; parentPrompts(): string[]; + steerAttemptCount(): number; wasParentPrompted(): boolean; } @@ -62,6 +63,7 @@ function createFinishNotificationScenario( let subscriber: ((event: AgentManagerEvent) => void) | null = null; let resolveParentPrompt: ((prompt: string) => void) | null = null; let parentPrompted = false; + let steerAttemptCount = 0; const parentPrompts: string[] = []; const childAgent: ManagedAgent = Object.create(null); @@ -96,6 +98,10 @@ function createFinishNotificationScenario( }); Reflect.set(agentManager, "tryRunOutOfBand", () => false); Reflect.set(agentManager, "hasInFlightRun", () => Boolean(options?.parentPromptError)); + Reflect.set(agentManager, "steerOrReplaceActiveTurn", async () => { + steerAttemptCount += 1; + return { status: "inactive" }; + }); Reflect.set(agentManager, "streamAgent", (_agentId: string, prompt: string) => { parentPrompted = true; parentPrompts.push(prompt); @@ -233,6 +239,9 @@ function createFinishNotificationScenario( parentPrompts() { return parentPrompts; }, + steerAttemptCount() { + return steerAttemptCount; + }, wasParentPrompted() { return parentPrompted; }, @@ -257,6 +266,7 @@ test("finish notifications tell the parent the child's last assistant message", "Agent child-agent (Child Agent) finished.\n\n\nImplemented the cleanup and all checks pass.\n", ), ); + expect(scenario.steerAttemptCount()).toBe(1); }); test("finish notifications truncate oversized child responses", async () => { diff --git a/packages/server/src/server/agent/agent-prompt.ts b/packages/server/src/server/agent/agent-prompt.ts index 020a3e91c..05650efd1 100644 --- a/packages/server/src/server/agent/agent-prompt.ts +++ b/packages/server/src/server/agent/agent-prompt.ts @@ -402,6 +402,7 @@ export function setupFinishNotification(params: SetupFinishNotificationParams): agentStorage, agentId: callerAgentId, prompt: formatSystemNotificationPrompt(body), + activeTurnBehavior: "steer", unarchive: false, logger, }); diff --git a/packages/server/src/server/schedule/service.test.ts b/packages/server/src/server/schedule/service.test.ts index 89b017159..322bffa8f 100644 --- a/packages/server/src/server/schedule/service.test.ts +++ b/packages/server/src/server/schedule/service.test.ts @@ -463,6 +463,40 @@ describe("ScheduleService", () => { ); }); + test("delivers agent-target schedules through the steer-or-interrupt path", async () => { + const manager = new AgentManager({ + logger: createTestLogger(), + clients: createTestAgentClients(), + registry: agentStorage, + }); + const agent = await manager.createAgent({ provider: "claude", cwd: tempDir }, undefined, { + workspaceId: undefined, + }); + const steerOrReplace = vi.spyOn(manager, "steerOrReplaceActiveTurn"); + const service = createScheduleService({ + paseoHome: tempDir, + logger: createTestLogger(), + agentManager: manager, + agentStorage, + providerSnapshotManager: NO_UNATTENDED_SCHEDULE_POLICY, + now: () => now, + }); + const schedule = await service.create({ + prompt: "Check scheduled work", + cadence: { type: "every", everyMs: 60_000 }, + target: { type: "agent", agentId: agent.id }, + }); + + await service.runOnce(schedule.id); + + expect(steerOrReplace).toHaveBeenCalledTimes(1); + expect(steerOrReplace.mock.calls[0]).toEqual([ + agent.id, + expect.stringContaining(`Schedule fired (id=${schedule.id}, run=`), + undefined, + ]); + }); + test("titles scheduled new agents from the schedule prompt", async () => { const manager = new AgentManager({ logger: createTestLogger(), diff --git a/packages/server/src/server/schedule/service.ts b/packages/server/src/server/schedule/service.ts index 9b740a4a5..e3e9504e4 100644 --- a/packages/server/src/server/schedule/service.ts +++ b/packages/server/src/server/schedule/service.ts @@ -7,7 +7,11 @@ import type { AgentSessionConfig } from "../agent/agent-sdk-types.js"; import type { AgentStorage } from "../agent/agent-storage.js"; import { curateAgentActivity } from "../agent/activity-curator.js"; import { ensureAgentLoaded } from "../agent/agent-loading.js"; -import { formatSystemNotificationPrompt } from "../agent/agent-prompt.js"; +import { + formatSystemNotificationPrompt, + startAgentRun, + type AgentRunController, +} from "../agent/agent-prompt.js"; import { resolveCreateAgentTitles } from "../agent/create-agent-title.js"; import { type BoundCreateAgentCommand, formatProviderModel } from "../agent/create-agent/create.js"; import type { PersistedWorkspaceRecord } from "../workspace-registry.js"; @@ -196,17 +200,24 @@ function buildRunOutput(params: { } type ScheduleAgentManager = Pick< - AgentManager, - | "createAgent" + AgentRunController, | "getAgent" - | "getRegisteredProviderIds" + | "tryRunOutOfBand" | "hasInFlightRun" - | "hydrateTimelineFromProvider" - | "resumeAgentFromPersistence" - | "runAgent" - | "waitForAgentEvent" - | "waitForAgentClose" ->; + | "replaceAgentRun" + | "steerOrReplaceActiveTurn" + | "streamAgent" +> & + Pick< + AgentManager, + | "createAgent" + | "getRegisteredProviderIds" + | "hydrateTimelineFromProvider" + | "resumeAgentFromPersistence" + | "runAgent" + | "waitForAgentEvent" + | "waitForAgentClose" + >; interface ScheduleWorkspaceCreateInput { cwd: string; @@ -843,14 +854,25 @@ export class ScheduleService { if (this.agentManager.hasInFlightRun(agent.id)) { throw new Error(`Agent ${agent.id} already has an active run`); } - const result = await this.agentManager.runAgent(agent.id, wrappedPrompt); - const timelineText = curateAgentActivity(result.timeline); + await startAgentRun(this.agentManager, agent.id, wrappedPrompt, this.logger, { + replaceRunning: true, + activeTurnBehavior: "steer", + }); + const waitResult = await this.agentManager.waitForAgentEvent(agent.id, { + waitForActive: true, + }); + if (waitResult.permission) { + throw new Error(`Scheduled agent ${agent.id} is waiting for permission`); + } + if (waitResult.status === "error") { + throw new Error(waitResult.lastMessage ?? `Scheduled agent ${agent.id} failed`); + } return { agentId: agent.id, output: buildRunOutput({ output: null, - timelineText, - finalText: result.finalText, + timelineText: "", + finalText: waitResult.lastMessage ?? "", }), }; }