diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index bc8275d7c..be59baed1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -60,6 +60,7 @@ export type AgentManagerOptions = { export type WaitForAgentOptions = { signal?: AbortSignal; + waitForActive?: boolean; }; export type WaitForAgentResult = { @@ -611,9 +612,12 @@ export class AgentManager { }; } - if (!isAgentBusy(snapshot.lifecycle)) { + const initialStatus = snapshot.lifecycle; + const initialBusy = isAgentBusy(initialStatus); + const waitForActive = options?.waitForActive ?? false; + if (!waitForActive && !initialBusy) { return { - status: snapshot.lifecycle, + status: initialStatus, permission: null, lastMessage: this.getLastAssistantMessage(agentId) }; @@ -632,7 +636,8 @@ export class AgentManager { return; } - let currentStatus: AgentLifecycleStatus = snapshot.lifecycle; + let currentStatus: AgentLifecycleStatus = initialStatus; + let hasStarted = initialBusy; // Bug #3 Fix: Declare unsubscribe and abortHandler upfront so cleanup can reference them let unsubscribe: (() => void) | null = null; @@ -691,7 +696,11 @@ export class AgentManager { finish(pending); return; } - if (!isAgentBusy(event.agent.lifecycle)) { + if (isAgentBusy(event.agent.lifecycle)) { + hasStarted = true; + return; + } + if (!waitForActive || hasStarted) { finish(null); } } diff --git a/packages/server/src/server/agent/agent-mcp.e2e.test.ts b/packages/server/src/server/agent/agent-mcp.e2e.test.ts index 7e889a0f9..440e492c4 100644 --- a/packages/server/src/server/agent/agent-mcp.e2e.test.ts +++ b/packages/server/src/server/agent/agent-mcp.e2e.test.ts @@ -1,3 +1,4 @@ +import net from "node:net"; import os from "node:os"; import path from "node:path"; import { mkdtemp, readFile, rm } from "node:fs/promises"; @@ -5,20 +6,15 @@ import { describe, expect, test } from "vitest"; import { experimental_createMCPClient } from "ai"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; -import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; +import { createPaseoDaemon, type PaseoDaemonConfig } from "../bootstrap.js"; type McpToolResult = { structuredContent?: Record; - content?: Array< - { structuredContent?: Record } | Record - >; + content?: Array<{ structuredContent?: Record } | Record>; }; type McpClient = { - callTool: (input: { - name: string; - args?: Record; - }) => Promise; + callTool: (input: { name: string; args?: Record }) => Promise; close: () => Promise; }; @@ -26,13 +22,23 @@ type PermissionPayload = { id: string; }; -function getStructuredContent( - result: McpToolResult -): Record | null { - if ( - result.structuredContent && - typeof result.structuredContent === "object" - ) { +async function getAvailablePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.once("error", reject); + server.listen(0, () => { + const address = server.address(); + if (!address || typeof address === "string") { + server.close(() => reject(new Error("Failed to acquire port"))); + return; + } + server.close(() => resolve(address.port)); + }); + }); +} + +function getStructuredContent(result: McpToolResult): Record | null { + if (result.structuredContent && typeof result.structuredContent === "object") { return result.structuredContent; } const content = result.content?.[0]; @@ -45,10 +51,7 @@ function getStructuredContent( return null; } -async function waitForFile( - filePath: string, - timeoutMs = 30000 -): Promise { +async function waitForFile(filePath: string, timeoutMs = 30000): Promise { const startedAt = Date.now(); while (Date.now() - startedAt < timeoutMs) { try { @@ -82,109 +85,199 @@ async function waitForAgentCompletion( } } +const hasClaudeCredentials = Boolean( + process.env.CLAUDE_CODE_OAUTH_TOKEN || process.env.ANTHROPIC_API_KEY +); + describe("agent MCP end-to-end", () => { - test("creates a Claude agent and writes a file", async () => { - const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-agent-cwd-")); - const daemonHandle = await createTestPaseoDaemon(); + const runTest = hasClaudeCredentials ? test : test.skip; + runTest( + "creates a Claude agent and writes a file", + async () => { + const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-home-")); + const staticDir = await mkdtemp(path.join(os.tmpdir(), "paseo-static-")); + const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-agent-cwd-")); + const port = await getAvailablePort(); + const basicUsers = { test: "pass" }; + const [agentMcpUser, agentMcpPassword] = + Object.entries(basicUsers)[0] ?? []; + const agentMcpAuthHeader = + agentMcpUser && agentMcpPassword + ? `Basic ${Buffer.from(`${agentMcpUser}:${agentMcpPassword}`).toString("base64")}` + : undefined; + const agentMcpBearerToken = + agentMcpUser && agentMcpPassword + ? Buffer.from(`${agentMcpUser}:${agentMcpPassword}`).toString("base64") + : undefined; - const transport = new StreamableHTTPClientTransport( - new URL(`http://127.0.0.1:${daemonHandle.port}/mcp/agents`), - daemonHandle.agentMcpAuthHeader - ? { - requestInit: { - headers: { Authorization: daemonHandle.agentMcpAuthHeader }, - }, - } - : undefined - ); - const client = (await experimental_createMCPClient({ - transport, - })) as McpClient; - - let agentId: string | null = null; - - try { - const result = (await client.callTool({ - name: "create_agent", - args: { - cwd: agentCwd, - title: "MCP e2e codex smoke", - agentType: "codex", - initialMode: "read-only", - background: false, + const daemonConfig: PaseoDaemonConfig = { + port, + paseoHome, + agentMcpRoute: "/mcp/agents", + agentMcpAllowedHosts: [`127.0.0.1:${port}`, `localhost:${port}`], + auth: { + basicUsers, + agentMcpAuthHeader, + agentMcpBearerToken, + realm: "Voice Assistant", }, - })) as McpToolResult; - - const payload = getStructuredContent(result); - expect(payload).toBeTruthy(); - agentId = payload?.agentId as string | null; - expect(agentId).toBeTruthy(); - - const prompt = [ - 'Run this command exactly: ["bash", "-lc", "echo ok > mcp-smoke.txt"].', - "After the command runs, reply with done and stop.", - ].join("\n"); - - const promptResult = (await client.callTool({ - name: "send_agent_prompt", - args: { - agentId, - prompt, - sessionMode: "read-only", - background: false, - }, - })) as McpToolResult; - const promptPayload = getStructuredContent(promptResult); - const permission = promptPayload?.permission as PermissionPayload | null; - if (!permission?.id) { - const activityResult = (await client.callTool({ - name: "get_agent_activity", - args: { agentId, limit: 10 }, - })) as McpToolResult; - console.log( - "[agent-mcp.e2e] send_agent_prompt payload:", - promptPayload - ); - console.log( - "[agent-mcp.e2e] get_agent_activity payload:", - getStructuredContent(activityResult) - ); - } - expect(permission?.id).toBeTruthy(); - await client.callTool({ - name: "respond_to_permission", - args: { - agentId, - requestId: permission!.id, - response: { behavior: "allow" }, + staticDir, + mcpDebug: false, + agentClients: {}, + agentRegistryPath: path.join(paseoHome, "agents.json"), + agentControlMcp: { + url: `http://127.0.0.1:${port}/mcp/agents`, + ...(agentMcpAuthHeader + ? { headers: { Authorization: agentMcpAuthHeader } } + : {}), }, + }; + + const previousCodexSessionDir = process.env.CODEX_SESSION_DIR; + const previousCodexHome = process.env.CODEX_HOME; + const codexSessionDir = await mkdtemp( + path.join(os.tmpdir(), "codex-session-") + ); + const codexHome = await mkdtemp(path.join(os.tmpdir(), "codex-home-")); + process.env.CODEX_SESSION_DIR = codexSessionDir; + process.env.CODEX_HOME = codexHome; + + const daemon = await createPaseoDaemon(daemonConfig); + await new Promise((resolve) => { + daemon.httpServer.listen(port, () => resolve()); }); - await waitForAgentCompletion(client, agentId); - const filePath = path.join(agentCwd, "mcp-smoke.txt"); - let contents: string; + const transport = new StreamableHTTPClientTransport( + new URL(`http://127.0.0.1:${port}/mcp/agents`), + agentMcpAuthHeader + ? { requestInit: { headers: { Authorization: agentMcpAuthHeader } } } + : undefined + ); + const client = (await experimental_createMCPClient({ + transport, + })) as McpClient; + + let agentId: string | null = null; + try { - contents = await waitForFile(filePath); - } catch (error) { - const activityResult = (await client.callTool({ - name: "get_agent_activity", - args: { agentId, limit: 25 }, + const initialPrompt = [ + "You must call the tool named shell.", + "Run this command exactly: [\"bash\", \"-lc\", \"echo ok > mcp-smoke.txt\"].", + "After the tool runs, reply with done and stop.", + ].join("\n"); + + const result = (await client.callTool({ + name: "create_agent", + args: { + cwd: agentCwd, + title: "MCP e2e smoke", + agentType: "claude", + initialMode: "default", + initialPrompt, + background: false, + }, })) as McpToolResult; - const activityPayload = getStructuredContent(activityResult); - const activitySummary = activityPayload?.content; - const details = activitySummary - ? `Agent activity:\n${activitySummary}` - : "Agent activity unavailable"; - throw new Error(`${(error as Error).message}\n${details}`); + + const payload = getStructuredContent(result); + expect(payload).toBeTruthy(); + agentId = payload?.agentId as string | null; + expect(agentId).toBeTruthy(); + const createPermission = payload?.permission as PermissionPayload | null; + expect(createPermission?.id).toBeTruthy(); + await client.callTool({ + name: "respond_to_permission", + args: { + agentId, + requestId: createPermission!.id, + response: { behavior: "allow" }, + }, + }); + await waitForAgentCompletion(client, agentId); + + const filePath = path.join(agentCwd, "mcp-smoke.txt"); + let contents: string; + try { + contents = await waitForFile(filePath); + } catch (error) { + const activityResult = (await client.callTool({ + name: "get_agent_activity", + args: { agentId, limit: 25 }, + })) as McpToolResult; + const activityPayload = getStructuredContent(activityResult); + const activitySummary = activityPayload?.content; + const details = activitySummary + ? `Agent activity:\n${activitySummary}` + : "Agent activity unavailable"; + throw new Error(`${(error as Error).message}\n${details}`); + } + expect(contents.trim()).toBe("ok"); + + const prompt = [ + "You must call the tool named shell.", + "Run this command exactly: [\"bash\", \"-lc\", \"echo ok-2 > mcp-smoke-2.txt\"].", + "After the tool runs, reply with done and stop.", + ].join("\n"); + + const promptResult = (await client.callTool({ + name: "send_agent_prompt", + args: { + agentId, + prompt, + sessionMode: "default", + background: false, + }, + })) as McpToolResult; + + const promptPayload = getStructuredContent(promptResult); + const promptPermission = promptPayload?.permission as PermissionPayload | null; + expect(promptPermission?.id).toBeTruthy(); + + const waitPermissionResult = (await client.callTool({ + name: "wait_for_agent", + args: { agentId }, + })) as McpToolResult; + const waitPermissionPayload = getStructuredContent(waitPermissionResult); + const waitPermission = + waitPermissionPayload?.permission as PermissionPayload | null; + expect(waitPermission?.id).toBe(promptPermission?.id); + + await client.callTool({ + name: "respond_to_permission", + args: { + agentId, + requestId: promptPermission!.id, + response: { behavior: "allow" }, + }, + }); + + await waitForAgentCompletion(client, agentId); + + const secondFilePath = path.join(agentCwd, "mcp-smoke-2.txt"); + const secondContents = await waitForFile(secondFilePath); + expect(secondContents.trim()).toBe("ok-2"); + } finally { + if (agentId) { + await client.callTool({ name: "kill_agent", args: { agentId } }); + } + await client.close(); + await daemon.close().catch(() => undefined); + if (previousCodexSessionDir === undefined) { + delete process.env.CODEX_SESSION_DIR; + } else { + process.env.CODEX_SESSION_DIR = previousCodexSessionDir; + } + if (previousCodexHome === undefined) { + delete process.env.CODEX_HOME; + } else { + process.env.CODEX_HOME = previousCodexHome; + } + await rm(paseoHome, { recursive: true, force: true }); + await rm(staticDir, { recursive: true, force: true }); + await rm(agentCwd, { recursive: true, force: true }); + await rm(codexSessionDir, { recursive: true, force: true }); + await rm(codexHome, { recursive: true, force: true }); } - expect(contents.trim()).toBe("ok"); - } finally { - if (agentId) { - await client.callTool({ name: "kill_agent", args: { agentId } }); - } - await client.close(); - await daemonHandle.close(); - await rm(agentCwd, { recursive: true, force: true }); - } - }, 180_000); + }, + 180_000 + ); }); diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index 1c8c2e4bb..716b27520 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -62,7 +62,8 @@ function expandPath(path: string): string { async function waitForAgentWithTimeout( agentManager: AgentManager, agentId: string, - existingSignal?: AbortSignal + existingSignal?: AbortSignal, + options?: { waitForActive?: boolean } ): Promise { const timeoutSignal = AbortSignal.timeout(AGENT_WAIT_TIMEOUT_MS); const abortController = new AbortController(); @@ -99,6 +100,7 @@ async function waitForAgentWithTimeout( try { const result = await agentManager.waitForAgentEvent(agentId, { signal: abortController.signal, + waitForActive: options?.waitForActive, }); return result; } catch (error) { @@ -297,7 +299,9 @@ export async function createAgentMcpServer( // If not running in background, wait for completion if (!background) { - const result = await waitForAgentWithTimeout(agentManager, snapshot.id); + const result = await waitForAgentWithTimeout(agentManager, snapshot.id, undefined, { + waitForActive: true, + }); const responseData = { agentId: snapshot.id, @@ -509,7 +513,9 @@ export async function createAgentMcpServer( // If not running in background, wait for completion if (!background) { - const result = await waitForAgentWithTimeout(agentManager, agentId); + const result = await waitForAgentWithTimeout(agentManager, agentId, undefined, { + waitForActive: true, + }); const responseData = { success: true,