From cd77ecd837df40d042599d8847403eb416b35b04 Mon Sep 17 00:00:00 2001 From: "paseo-bot[bot]" <266920839+paseo-bot[bot]@users.noreply.github.com> Date: Sat, 17 Oct 2026 01:10:35 +0200 Subject: [PATCH] Start idle OMP agents when they receive follow-ups (#7169) * fix(server): run slash commands sent by another agent out of band again * fix(omp): send /steer and /follow-up as prompts so an idle agent starts a run * test(omp): verify idle follow-ups and parent notifications Cover real daemon and MCP delivery, and keep the fake runtime tests independent of an installed OMP executable. Co-authored-by: Hung Le <22406703+nhthung@users.noreply.github.com> * fix(omp): report idle follow-up admission failures Use the existing prompt lifecycle for an idle command so acknowledgements and late rejections settle the child and notify its caller. Keep busy commands on the streaming prompt path, and accept one runtime prompt input object. --------- Co-authored-by: Hung Le <22406703+nhthung@users.noreply.github.com> Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com> --- .../src/server/agent/agent-manager.test.ts | 35 ++++ .../server/src/server/agent/agent-manager.ts | 9 +- .../src/server/agent/mcp-server.test.ts | 179 +++++++++++++++++- .../server/agent/providers/omp/agent.test.ts | 13 ++ .../src/server/agent/providers/omp/agent.ts | 44 ++++- .../agent/providers/omp/cli-runtime.test.ts | 24 ++- .../server/agent/providers/omp/cli-runtime.ts | 14 +- .../server/agent/providers/omp/rpc-types.ts | 1 + .../src/server/agent/providers/omp/runtime.ts | 17 +- .../providers/omp/test-utils/fake-omp.ts | 32 ++-- .../server/daemon-e2e/omp.real.e2e.test.ts | 86 ++++++++- 11 files changed, 403 insertions(+), 51 deletions(-) diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 763659e9f4..44e38a344f 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -10929,6 +10929,41 @@ test("authoritative timeline records a daemon-handled submitted prompt before it } }); +test("tryRunOutOfBand gives providers the command another agent sent, not its envelope", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-agent-out-of-band-")); + const seenPrompts: AgentPromptInput[] = []; + + class SlashCommandSession extends TestAgentSession { + override tryHandleOutOfBand(prompt: AgentPromptInput) { + seenPrompts.push(prompt); + return prompt === "/handled" ? { run: async () => {} } : null; + } + } + const client = new (class extends TestAgentClient { + override async createSession(config: AgentSessionConfig) { + return new SlashCommandSession(config); + } + })(); + const manager = new AgentManager({ clients: { codex: client }, logger }); + const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, { + workspaceId: undefined, + }); + + try { + const fromAgent = prepareAgentMessage("/handled", { kind: "agent-message", agentId: "sender" }); + expect(manager.tryRunOutOfBand(agent.id, fromAgent.prompt)).toBe(true); + expect(seenPrompts).toEqual(["/handled"]); + + // A human who pastes an envelope gets it quoted, and it stays a plain prompt. + const pasted = prepareAgentMessage(String(fromAgent.prompt)); + expect(manager.tryRunOutOfBand(agent.id, pasted.prompt)).toBe(false); + expect(seenPrompts.at(-1)).toBe(pasted.prompt); + } finally { + await manager.closeAgent(agent.id); + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("replaceAgentRun succeeds when foreground turn terminal event is never delivered", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-stale-fg-")); const storagePath = join(workdir, "agents"); diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index e09d62f6a9..ff91ef85ef 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -79,7 +79,7 @@ import { type PendingForegroundRun, } from "./agent-run-state.js"; import { invokeRewindCapability, type RewindMode } from "./rewind/rewind.js"; -import { projectAgentMessage } from "./agent-messages/index.js"; +import { parseAgentMessage, projectAgentMessage } from "./agent-messages/index.js"; import { isStaleProviderSessionError } from "./stale-provider-session-error.js"; import { stripInternalPaseoMcpServer, withRuntimePaseoMcpServer } from "./runtime-mcp-config.js"; import { resolveCreateAgentTitles } from "./create-agent-title.js"; @@ -2459,7 +2459,12 @@ export class AgentManager { */ tryRunOutOfBand(agentId: string, prompt: AgentPromptInput, options?: AgentRunOptions): boolean { const agent = this.requireSessionAgent(agentId); - const handler = agent.session.tryHandleOutOfBand?.(prompt); + // Providers parse slash commands from the raw text, so a command another agent sent + // must reach them without its sender envelope. + const agentMessage = typeof prompt === "string" ? parseAgentMessage(prompt) : null; + const handler = agent.session.tryHandleOutOfBand?.( + agentMessage?.source ? agentMessage.text : prompt, + ); if (!handler) { return false; } diff --git a/packages/server/src/server/agent/mcp-server.test.ts b/packages/server/src/server/agent/mcp-server.test.ts index de0099cecf..3a90b7f577 100644 --- a/packages/server/src/server/agent/mcp-server.test.ts +++ b/packages/server/src/server/agent/mcp-server.test.ts @@ -15,6 +15,8 @@ import { createAgentMcpServer as createAgentMcpServerWithDependencies } from "./ import { AgentManager, type ManagedAgent } from "./agent-manager.js"; import { AgentStorage, type StoredAgentRecord } from "./agent-storage.js"; import { createTestAgentClients } from "../test-utils/fake-agent-client.js"; +import { OmpAgentClient } from "./providers/omp/agent.js"; +import { FakeOmp } from "./providers/omp/test-utils/fake-omp.js"; import type { AgentClient, AgentMode, @@ -3856,8 +3858,7 @@ class HeldTurnAgentSession implements AgentSession { } tryHandleOutOfBand(prompt: AgentPromptInput) { - const text = typeof prompt === "string" ? (parseAgentMessage(prompt)?.text ?? prompt) : null; - return text === "/side-command" ? { run: async () => {} } : null; + return prompt === "/side-command" ? { run: async () => {} } : null; } async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> { @@ -4557,6 +4558,180 @@ describe("send_agent_prompt MCP tool", () => { }); }); +describe("send_agent_prompt MCP tool to an OMP child", () => { + const logger = createTestLogger(); + const existingCwd = process.cwd(); + + async function setupOmpChild() { + const workdir = await mkdtemp(join(tmpdir(), "mcp-omp-child-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const parentClient = new HeldTurnAgentClient("claude", false); + const omp = new FakeOmp(); + const agentManager = new AgentManager({ + clients: { + claude: parentClient, + omp: new OmpAgentClient({ + logger, + runtime: omp, + noTurnScheduler: { waitForSettle: async () => {} }, + runtimeSettings: { command: { mode: "replace", argv: [process.execPath] } }, + }), + }, + registry: storage, + logger, + }); + const parent = await agentManager.createAgent( + { provider: "claude", cwd: existingCwd }, + undefined, + { workspaceId: "wks_parent" }, + ); + const child = await agentManager.createAgent( + { provider: "omp", cwd: existingCwd, title: "OMP child" }, + undefined, + { workspaceId: "wks_parent" }, + ); + const server = await createAgentMcpServer({ + agentManager, + agentStorage: storage, + callerAgentId: parent.id, + providerSnapshotManager: createOpenCodeManager().manager, + logger, + }); + return { + agentManager, + storage, + workdir, + parentClient, + ompSession: omp.latestSession(), + childId: child.id, + tool: registeredTool(server, "send_agent_prompt"), + }; + } + + async function expectOneFinishNotification( + agentManager: AgentManager, + parentClient: HeldTurnAgentClient, + childId: string, + ) { + await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("idle")); + await vi.waitFor(() => { + expect(parentClient.sessions[0]!.prompts).toHaveLength(1); + expect(parentClient.sessions[0]!.prompts[0]).toContain( + `Agent ${childId} (OMP child) finished.`, + ); + }); + } + + it.each([ + { command: "follow-up", streamingBehavior: "followUp" }, + { command: "steer", streamingBehavior: "steer" }, + ] as const)( + "delivers /$command into a turn OMP started on its own, without aborting it", + async ({ command, streamingBehavior }) => { + const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } = + await setupOmpChild(); + try { + ompSession.beginTurn(); + ompSession.streamAssistantText("background job finished, working on it"); + await vi.waitFor(() => { + expect(agentManager.getAgent(childId)?.activeTurnId).toMatch(/^autonomous-/); + }); + + const response = await invokeToolWithParsedInput(tool, { + agentId: childId, + prompt: `/${command} then summarize the logs`, + }); + + expect(response.structuredContent).toMatchObject({ + success: true, + disposition: "out_of_band", + status: "running", + }); + await vi.waitFor(() => { + expect(ompSession.prompts).toEqual([ + { message: "then summarize the logs", imageCount: 0, streamingBehavior }, + ]); + }); + expect(ompSession.abortRequested).toBe(false); + expect(agentManager.getAgent(childId)?.lifecycle).toBe("running"); + + ompSession.finishTurn(); + await expectOneFinishNotification(agentManager, parentClient, childId); + } finally { + await removeAgentStateDir(agentManager, storage, workdir); + } + }, + ); + + it.each(["before ack", "after ack"])( + "reports an idle follow-up rejected %s and notifies its parent", + async (order) => { + const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } = + await setupOmpChild(); + try { + ompSession.promptAck = { requestId: "follow-up-rejected" }; + const rejection = { + type: "prompt_result", + id: "follow-up-rejected", + agentInvoked: false, + status: "error", + sessionSettled: true, + error: { message: "Provider rejected the follow-up", retryable: false }, + } as const; + const promptStarted = ompSession.nextPrompt(); + const response = invokeToolWithParsedInput(tool, { + agentId: childId, + prompt: "/follow-up check the build", + }); + await promptStarted; + if (order === "before ack") ompSession.emit(rejection); + await response; + if (order === "after ack") ompSession.emit(rejection); + await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("error")); + expect(agentManager.getAgent(childId)?.lastError).toContain( + "Provider rejected the follow-up", + ); + await vi.waitFor(() => expect(parentClient.sessions[0]!.prompts).toHaveLength(1)); + expect(parentClient.sessions[0]!.prompts[0]).toContain( + `Agent ${childId} (OMP child) errored.`, + ); + } finally { + await removeAgentStateDir(agentManager, storage, workdir); + } + }, + ); + + it("lets an idle child start a run with the follow-up", async () => { + const { agentManager, storage, workdir, parentClient, ompSession, childId, tool } = + await setupOmpChild(); + try { + const response = await invokeToolWithParsedInput(tool, { + agentId: childId, + prompt: "/follow-up check the build", + }); + + expect(response.structuredContent).toMatchObject({ + success: true, + disposition: "out_of_band", + status: "running", + }); + await vi.waitFor(() => { + expect(ompSession.prompts).toEqual([ + { message: "check the build", imageCount: 0, streamingBehavior: "followUp" }, + ]); + }); + + ompSession.beginTurn(); + ompSession.streamAssistantText("Build is green"); + await vi.waitFor(() => expect(agentManager.getAgent(childId)?.lifecycle).toBe("running")); + ompSession.finishTurn(); + await expectOneFinishNotification(agentManager, parentClient, childId); + } finally { + await removeAgentStateDir(agentManager, storage, workdir); + } + }); +}); + describe("update_agent MCP tool", () => { const logger = createTestLogger(); diff --git a/packages/server/src/server/agent/providers/omp/agent.test.ts b/packages/server/src/server/agent/providers/omp/agent.test.ts index a1dc9130a8..c9911a64b8 100644 --- a/packages/server/src/server/agent/providers/omp/agent.test.ts +++ b/packages/server/src/server/agent/providers/omp/agent.test.ts @@ -39,6 +39,19 @@ const ABORTED_TERMINAL_RESPONSE: OmpAgentMessage = { errorMessage: "Interrupted by user", }; +test("an idle follow-up is submitted as a prompt so OMP can start a turn", async () => { + const runtime = new FakeOmp(); + const client = new OmpAgentClient({ logger: createTestLogger(), runtime }); + const session = await client.createSession({ provider: "omp", cwd: process.cwd() }); + onTestFinished(() => session.close()); + const handler = session.tryHandleOutOfBand?.("/follow-up check the build"); + expect(handler).not.toBeNull(); + await handler!.run({ emit: () => {} }); + expect(runtime.latestSession().prompts).toEqual([ + { message: "check the build", imageCount: 0, streamingBehavior: "followUp" }, + ]); +}); + test("OMP ready timeout defaults to 20 seconds and RPC timeout overrides both", () => { expect(resolveOmpProviderOptions({}).runtimeOptions).toMatchObject({ readyTimeoutMs: 20_000, diff --git a/packages/server/src/server/agent/providers/omp/agent.ts b/packages/server/src/server/agent/providers/omp/agent.ts index 5413364cdb..ab82df9475 100644 --- a/packages/server/src/server/agent/providers/omp/agent.ts +++ b/packages/server/src/server/agent/providers/omp/agent.ts @@ -705,6 +705,7 @@ export class OmpAgentSession implements AgentSession { private activeAssistantMessageId: string | null = null; private activeTurnTerminalAssistantMessage: OmpAgentMessage | null = null; private activeTurnStarted = false; + private outOfBandTurnStartAnnounced = false; private activeTurnHasUserMessage = false; private activeNoTurnPromptText: string | null = null; private readonly pendingNoTurnOutputs: Array<{ turnId: string; message: string }> = []; @@ -849,6 +850,20 @@ export class OmpAgentSession implements AgentSession { } async startTurn(prompt: AgentPromptInput, options?: AgentRunOptions): Promise { + return this.startPrompt({ prompt, options }); + } + + private async startPrompt({ + prompt, + options, + streamingBehavior, + announceTurn = false, + }: { + prompt: AgentPromptInput; + options?: AgentRunOptions; + streamingBehavior?: "steer" | "followUp"; + announceTurn?: boolean; + }): Promise { if (this.activeTurnId) { throw new Error("An OMP turn is already active"); } @@ -861,17 +876,26 @@ export class OmpAgentSession implements AgentSession { this.activeAssistantMessageId = null; this.activeTurnTerminalAssistantMessage = null; this.activeTurnStarted = false; + this.outOfBandTurnStartAnnounced = false; this.activeTurnHasUserMessage = false; this.activePromptRequestId = null; this.clearNoTurnBuffers(); this.activeNoTurnPromptText = payload.text; this.usagePoller.startTurn(); + if (announceTurn) { + this.outOfBandTurnStartAnnounced = true; + this.emit({ type: "turn_started", provider: this.provider, turnId }); + } void (async () => { try { if (this.runtimeDead) await this.replaceRuntime(this.currentModeId ?? "full"); if (this.closed) throw new Error("OMP session is closed"); - const ack = await this.runtimeSession.prompt(payload.text, payload.images); + const ack = await this.runtimeSession.prompt({ + message: payload.text, + images: payload.images, + streamingBehavior, + }); this.activePromptRequestId = ack.requestId ?? null; const correlatedResult = ack.requestId ? this.pendingPromptResults.get(ack.requestId) @@ -1191,6 +1215,7 @@ export class OmpAgentSession implements AgentSession { this.activeAssistantMessageId = null; this.activeTurnTerminalAssistantMessage = null; this.activeTurnStarted = false; + this.outOfBandTurnStartAnnounced = false; this.activeTurnHasUserMessage = false; this.pendingClientMessages.length = 0; this.clearNoTurnBuffers(); @@ -1235,12 +1260,13 @@ export class OmpAgentSession implements AgentSession { if (!message) { return null; } + const streamingBehavior = commandName === "steer" ? "steer" : "followUp"; return { run: async () => { - if (commandName === "steer") { - this.runtimeSession.steer(message); + if (!this.activeTurnId && !this.activeTurnStarted) { + await this.startPrompt({ prompt: message, streamingBehavior, announceTurn: true }); } else { - this.runtimeSession.followUp(message); + await this.runtimeSession.prompt({ message, streamingBehavior }); } }, }; @@ -1804,11 +1830,11 @@ export class OmpAgentSession implements AgentSession { case "turn_start": this.activeTurnStarted = true; this.clearNoTurnBuffers(); - this.emit({ - type: "turn_started", - provider: this.provider, - turnId, - }); + if (this.outOfBandTurnStartAnnounced) { + this.outOfBandTurnStartAnnounced = false; + } else { + this.emit({ type: "turn_started", provider: this.provider, turnId }); + } return; case "message_start": this.handleMessageStart(event); diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts index 2deae45850..4c8a81fe46 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.test.ts @@ -143,6 +143,26 @@ describe("OMP CLI runtime", () => { await expect(result).rejects.toThrow("rejected"); await session.close(); }); + test("sends the prompt streamingBehavior only when one is given", async () => { + const child = createOmpChild(); + const session = await createRuntime(child, []).startSession({ cwd: "/workspace/project" }); + + for (const streamingBehavior of ["followUp", undefined] as const) { + const pending = capturePendingCommand(child, "prompt"); + const result = session.prompt({ message: "after this, update the docs", streamingBehavior }); + const command = await pending; + expect(withoutRequestId(command)).toEqual({ + type: "prompt", + message: "after this, update the docs", + ...(streamingBehavior ? { streamingBehavior } : {}), + }); + child.stdout.write( + `${JSON.stringify({ id: command.id, type: "response", command: "prompt", success: true })}\n`, + ); + await result; + } + await session.close(); + }); test("uses the configured RPC timeout and attributes the pending phase", async () => { vi.useFakeTimers(); const child = createOmpChild(); @@ -438,7 +458,7 @@ describe("OMP CLI runtime", () => { replyToCommands(child, () => undefined); const session = await createRuntime(child).startSession({ cwd: "/workspace/project" }); - await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" }); + await expect(session.prompt({ message: "hello" })).resolves.toEqual({ requestId: "req_1" }); }); test("reports OMP 18.3's late prompt rejection as a failed prompt result", async () => { @@ -455,7 +475,7 @@ describe("OMP CLI runtime", () => { const events: unknown[] = []; session.onEvent((event) => events.push(event)); - await expect(session.prompt("hello")).resolves.toEqual({ requestId: "req_1" }); + await expect(session.prompt({ message: "hello" })).resolves.toEqual({ requestId: "req_1" }); expect(events).toContainEqual({ type: "prompt_result", diff --git a/packages/server/src/server/agent/providers/omp/cli-runtime.ts b/packages/server/src/server/agent/providers/omp/cli-runtime.ts index 75bd79f5f0..5ae9d8f9c4 100644 --- a/packages/server/src/server/agent/providers/omp/cli-runtime.ts +++ b/packages/server/src/server/agent/providers/omp/cli-runtime.ts @@ -15,6 +15,7 @@ import { type OmpRuntime, type OmpRuntimeLaunch, type OmpRuntimeSession, + type OmpPromptInput, type OmpStartSessionInput, } from "./runtime.js"; import { @@ -142,14 +143,12 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { }; } - async prompt( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise { + async prompt({ message, images, streamingBehavior }: OmpPromptInput): Promise { const { id: requestId, promise } = this.process.startRequest({ type: "prompt", message, ...(images?.length ? { images } : {}), + ...(streamingBehavior ? { streamingBehavior: streamingBehavior } : {}), }); const ack = OmpPromptAckSchema.parse(await promise) ?? {}; return { requestId, ...ack }; @@ -282,13 +281,6 @@ class OmpCliRuntimeSession implements OmpRuntimeSession { await this.request({ type: "steer", message, ...(images?.length ? { images } : {}) }); } - followUp( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): void { - this.process.send({ type: "follow_up", message, ...(images?.length ? { images } : {}) }); - } - async handoff(customInstructions?: string): Promise { await this.request({ type: "handoff", diff --git a/packages/server/src/server/agent/providers/omp/rpc-types.ts b/packages/server/src/server/agent/providers/omp/rpc-types.ts index b0514c7863..733dedf571 100644 --- a/packages/server/src/server/agent/providers/omp/rpc-types.ts +++ b/packages/server/src/server/agent/providers/omp/rpc-types.ts @@ -535,6 +535,7 @@ export const OmpRpcCommandSchema = z.discriminatedUnion("type", [ type: z.literal("prompt"), message: z.string(), images: z.array(OmpImageContentSchema).optional(), + streamingBehavior: z.enum(["steer", "followUp"]).optional(), }), z.object({ ...OmpCommandBase, diff --git a/packages/server/src/server/agent/providers/omp/runtime.ts b/packages/server/src/server/agent/providers/omp/runtime.ts index 691e907139..523595ed07 100644 --- a/packages/server/src/server/agent/providers/omp/runtime.ts +++ b/packages/server/src/server/agent/providers/omp/runtime.ts @@ -42,13 +42,18 @@ export interface OmpStartSessionInput { extraArgs?: string[]; } +export interface OmpPromptInput { + message: string; + images?: Array<{ type: "image"; data: string; mimeType: string }>; + streamingBehavior?: "steer" | "followUp"; +} + export interface OmpRuntimeSession { readonly environment: Record; onEvent(callback: (event: OmpRuntimeEvent) => void): () => void; - prompt( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise; + // With streamingBehavior, a busy OMP queues the message instead of rejecting it, and an + // idle OMP starts a run with it. + prompt(input: OmpPromptInput): Promise; compact(customInstructions?: string): Promise; setAutoCompaction(enabled: boolean): Promise; abort(): Promise; @@ -71,10 +76,6 @@ export interface OmpRuntimeSession { message: string, images?: Array<{ type: "image"; data: string; mimeType: string }>, ): Promise; - followUp( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): void; handoff(customInstructions?: string): Promise; respondToExtensionUiRequest( id: string, diff --git a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts index 58688c245d..02033f8dea 100644 --- a/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts +++ b/packages/server/src/server/agent/providers/omp/test-utils/fake-omp.ts @@ -2,6 +2,7 @@ import type { OmpRuntime, OmpRuntimeLaunch, OmpRuntimeSession, + OmpPromptInput, OmpStartSessionInput, } from "../runtime.js"; import type { @@ -133,7 +134,11 @@ export class FakeOmpSession implements OmpRuntimeSession { readonly environment: Record; fastModeResult = { enabled: false, active: false }; readonly setFastModeRequests: boolean[] = []; - readonly prompts: Array<{ message: string; imageCount: number }> = []; + readonly prompts: Array<{ + message: string; + imageCount: number; + streamingBehavior?: "steer" | "followUp"; + }> = []; readonly compactRequests: Array<{ customInstructions?: string }> = []; readonly setAutoCompactionRequests: boolean[] = []; readonly subagentSubscriptionRequests: FakeOmpSubagentSubscriptionLevel[] = []; @@ -143,7 +148,6 @@ export class FakeOmpSession implements OmpRuntimeSession { readonly handoffRequests: Array<{ customInstructions?: string }> = []; readonly steerRequests: Array<{ message: string; imageCount: number }> = []; steerError: Error | null = null; - readonly followUpRequests: Array<{ message: string; imageCount: number }> = []; readonly hostToolSetRequests: OmpRpcHostToolDefinition[][] = []; readonly hostToolResults: OmpRpcHostToolResult[] = []; readonly hostToolUpdates: OmpRpcHostToolUpdate[] = []; @@ -216,14 +220,19 @@ export class FakeOmpSession implements OmpRuntimeSession { }; } - async prompt( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): Promise { - if (this.state.isStreaming || this.state.isCompacting) { + async prompt({ message, images, streamingBehavior }: OmpPromptInput): Promise { + const busy = this.state.isStreaming || this.state.isCompacting; + if (busy && !streamingBehavior) { throw new Error("Agent is already processing"); } - this.prompts.push({ message, imageCount: images?.length ?? 0 }); + this.prompts.push({ + message, + imageCount: images?.length ?? 0, + ...(streamingBehavior ? { streamingBehavior: streamingBehavior } : {}), + }); + if (busy) { + return {}; + } this.promptWaiters.shift()?.(); const heldPrompt = this.nextHeldPrompt; if (heldPrompt) { @@ -412,13 +421,6 @@ export class FakeOmpSession implements OmpRuntimeSession { if (this.steerError) throw this.steerError; } - followUp( - message: string, - images?: Array<{ type: "image"; data: string; mimeType: string }>, - ): void { - this.followUpRequests.push({ message, imageCount: images?.length ?? 0 }); - } - sendHostToolResult(result: OmpRpcHostToolResult): void { this.hostToolResults.push(result); this.hostToolResultWaiters.shift()?.(result); diff --git a/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts index ab407e31bc..247068220a 100644 --- a/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/omp.real.e2e.test.ts @@ -5,6 +5,8 @@ import path from "node:path"; import { promisify } from "node:util"; import pino from "pino"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { afterEach, beforeAll, describe, expect, test } from "vitest"; import type { AgentTimelineItem } from "../agent/agent-sdk-types.js"; @@ -131,6 +133,14 @@ async function timeline(client: DaemonClient, agentId: string): Promise entry.item); } +async function readAssistantText(harness: Harness, agentId: string): Promise { + const items = await timeline(harness.client, agentId); + return items + .filter((item) => item.type === "assistant_message") + .map((item) => item.text) + .join(""); +} + function compactionFiller(records = 1200): string { const teas = ["assam", "darjeeling", "sencha", "oolong"]; return Array.from( @@ -257,8 +267,80 @@ async function contextWindowUsedTokens(client: DaemonClient, agentId: string) { describe("daemon E2E (real OMP)", () => { test( - "a prompt OMP rejects before its agent runs fails the turn with OMP's error", + "an idle out-of-band follow-up starts a turn", async () => { + const harness = await createHarness(); + try { + const agent = await createAgent(harness, "idle-follow-up"); + await harness.client.sendMessage(agent.id, "/follow-up Reply exactly IDLE_FOLLOWUP_OK"); + await expect + .poll(() => readAssistantText(harness, agent.id), { timeout: 30_000 }) + .toContain("IDLE_FOLLOWUP_OK"); + const finish = await harness.client.waitForFinish(agent.id, TIMEOUT_MS); + expect(finish.status).toBe("idle"); + } finally { + await closeHarness(harness); + } + }, + TIMEOUT_MS, + ); + + test( + "an agent-sent idle follow-up starts a turn and notifies its parent", + async () => { + const harness = await createHarness(); + const mcp = new Client({ name: "idle-follow-up-test", version: "1.0.0" }); + try { + const parent = await createAgent(harness, "follow-up-parent"); + const child = await createAgent(harness, "follow-up-child"); + await mcp.connect( + new StreamableHTTPClientTransport( + new URL( + `http://127.0.0.1:${harness.daemon.port}/mcp/agents?callerAgentId=${parent.id}`, + ), + ), + ); + const response = await mcp.callTool({ + name: "send_agent_prompt", + arguments: { agentId: child.id, prompt: "/follow-up Reply exactly MCP_IDLE_FOLLOWUP_OK" }, + }); + expect(response.isError).not.toBe(true); + expect(response.structuredContent).toMatchObject({ + success: true, + disposition: "out_of_band", + }); + await expect + .poll(() => readAssistantText(harness, child.id), { timeout: 30_000 }) + .toContain("MCP_IDLE_FOLLOWUP_OK"); + await expect + .poll( + async () => { + const items = await timeline(harness.client, parent.id); + let notifications = 0; + for (const item of items) { + if ( + item.type === "tool_call" && + item.agentMessage?.sender.id === child.id && + item.agentMessage.event === "finished" + ) + notifications++; + } + return notifications; + }, + { timeout: 15_000 }, + ) + .toBe(1); + } finally { + await mcp.close(); + await closeHarness(harness); + } + }, + TIMEOUT_MS, + ); + + test.each(["Reply with ok.", "/follow-up Reply with ok."])( + "a rejected %s fails the turn with OMP's error", + async (prompt) => { const harness = await createHarness({ ompHome: createOfflineOmpHome() }); try { const agent = await harness.client.createAgent({ @@ -268,7 +350,7 @@ describe("daemon E2E (real OMP)", () => { model: "anthropic/claude-haiku-4-5", modeId: "full", }); - await harness.client.sendMessage(agent.id, "Reply with ok."); + await harness.client.sendMessage(agent.id, prompt); const finish = await harness.client.waitForFinish(agent.id, 60_000); expect(finish.status).toBe("error"); expect(finish.error).toContain("No API key found for anthropic");