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");