From 25c7d5afa96b91815507f2b3af606079ff52caa8 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 30 Jan 2026 14:50:01 +0700 Subject: [PATCH] refactor: consolidate agent wait methods and use in-memory MCP transport --- CLAUDE.md | 7 + packages/app/e2e/helpers/app.ts | 2 +- packages/app/e2e/permission-prompt.spec.ts | 4 +- packages/cli/src/commands/agent/send.ts | 6 +- packages/cli/src/commands/agent/wait.ts | 18 +- .../server/src/client/daemon-client-v2.ts | 223 +++++------------- packages/server/src/server/bootstrap.ts | 20 +- .../src/server/daemon-client-v2.e2e.test.ts | 10 +- .../daemon-e2e/agent-basics.e2e.test.ts | 2 +- .../daemon-e2e/agent-operations.e2e.test.ts | 4 +- .../daemon-restart-resume.e2e.test.ts | 4 +- .../src/server/daemon-e2e/images.e2e.test.ts | 4 +- .../daemon-e2e/permissions-claude.e2e.test.ts | 17 +- .../daemon-e2e/permissions-codex.e2e.test.ts | 22 +- .../server/daemon-e2e/persistence.e2e.test.ts | 14 +- .../server/daemon-e2e/self-id-mcp.e2e.test.ts | 32 +-- .../server/daemon-e2e/streaming.e2e.test.ts | 17 +- .../server/daemon-e2e/tool-calls.e2e.test.ts | 12 +- .../daemon-e2e/two-cycle-resume.e2e.test.ts | 6 +- .../daemon-e2e/wait-for-idle.e2e.test.ts | 20 +- .../wait-for-permission.e2e.test.ts | 123 ---------- packages/server/src/server/session.ts | 69 +----- .../server/src/server/websocket-server.ts | 9 +- .../src/server/websocket-session-bridge.ts | 15 +- 24 files changed, 191 insertions(+), 469 deletions(-) delete mode 100644 packages/server/src/server/daemon-e2e/wait-for-permission.e2e.test.ts diff --git a/CLAUDE.md b/CLAUDE.md index 61ba0a0de..a295d685b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,8 +65,15 @@ Run `npx expo-doctor` to diagnose version mismatches and native module issues. **CRITICAL: ALWAYS RUN TYPECHECK AFTER EVERY CHANGE.** +## Agent Authentication + +All agent providers (Claude, Codex, OpenCode) handle their own authentication outside of environment variables. They are authenticated without providing any extra configuration—Paseo does not manage API keys or tokens for agents. + +**Do not add auth checks to tests.** If auth fails for whatever reason, let the user know instead of patching the code or adding conditional skips. + ## NEVER DO THESE THINGS - **NEVER restart the Paseo daemon/server** - The daemon is running in Tmux and managed by the user. Restarting it disrupts active sessions, loses state, and breaks workflows. If there's a connectivity issue, investigate the cause - do not restart. - **NEVER kill or restart processes in Tmux** without explicit user permission - **NEVER assume a timeout means the service needs restarting** - Timeouts can be transient network issues, not service failures +- **NEVER add authentication checks to tests** - Agent providers handle their own auth. If tests fail due to auth issues, report it rather than adding conditional skips or env var checks diff --git a/packages/app/e2e/helpers/app.ts b/packages/app/e2e/helpers/app.ts index 962601fbd..025819f0f 100644 --- a/packages/app/e2e/helpers/app.ts +++ b/packages/app/e2e/helpers/app.ts @@ -422,7 +422,7 @@ export const denyPermission = async (page: Page) => { await denyButton.click(); }; -export async function waitForAgentIdle(page: Page, timeout = 30000) { +export async function waitForAgentFinishUI(page: Page, timeout = 30000) { const stopButton = page.getByRole('button', { name: /stop|cancel/i }); await expect(stopButton).not.toBeVisible({ timeout }); } diff --git a/packages/app/e2e/permission-prompt.spec.ts b/packages/app/e2e/permission-prompt.spec.ts index d2a83e64a..035a6d519 100644 --- a/packages/app/e2e/permission-prompt.spec.ts +++ b/packages/app/e2e/permission-prompt.spec.ts @@ -6,7 +6,7 @@ import { waitForPermissionPrompt, allowPermission, denyPermission, - waitForAgentIdle, + waitForAgentFinishUI, getToolCallCount, } from './helpers/app'; import { createTempGitRepo } from './helpers/workspace'; @@ -63,7 +63,7 @@ test.describe('permission prompts', () => { await waitForPermissionPrompt(page, 30000); await denyPermission(page); - await waitForAgentIdle(page); + await waitForAgentFinishUI(page); expect(existsSync(filePath)).toBe(false); diff --git a/packages/cli/src/commands/agent/send.ts b/packages/cli/src/commands/agent/send.ts index 55ce8b5c3..866123b81 100644 --- a/packages/cli/src/commands/agent/send.ts +++ b/packages/cli/src/commands/agent/send.ts @@ -182,8 +182,8 @@ export async function runSendCommand( } } - // Wait for agent to become idle - await client.waitForAgentIdle(agentId, 600000) // 10 minute timeout + // Wait for agent to finish + const state = await client.waitForFinish(agentId, 600000) // 10 minute timeout await client.close() @@ -192,7 +192,7 @@ export async function runSendCommand( data: { agentId, status: 'completed', - message: 'Agent completed processing the message', + message: state.status === 'error' ? 'Agent finished with error' : 'Agent completed processing the message', }, schema: agentSendSchema, } diff --git a/packages/cli/src/commands/agent/wait.ts b/packages/cli/src/commands/agent/wait.ts index 248f50be4..c0cd75ab5 100644 --- a/packages/cli/src/commands/agent/wait.ts +++ b/packages/cli/src/commands/agent/wait.ts @@ -140,35 +140,33 @@ export async function runWaitCommand( throw error } - // Wait for agent to become idle OR request permission (whichever comes first) + // Wait for agent to finish (idle, error, or permission) try { - const idlePromise = client.waitForAgentIdle(agentId, timeoutMs).then(() => ({ type: 'idle' as const })) - const permissionPromise = client - .waitForPermission(agentId, timeoutMs) - .then((request) => ({ type: 'permission' as const, request })) - - const result = await Promise.race([idlePromise, permissionPromise]) + const state = await client.waitForFinish(agentId, timeoutMs) await client.close() - if (result.type === 'permission') { + // Check if agent has pending permissions + if (state.pendingPermissions && state.pendingPermissions.length > 0) { + const permission = state.pendingPermissions[0] return { type: 'single', data: { agentId, status: 'permission', - message: `Agent is waiting for permission: ${result.request.kind}`, + message: `Agent is waiting for permission: ${permission.kind}`, }, schema: agentWaitSchema, } } + // Agent is idle or error return { type: 'single', data: { agentId, status: 'idle', - message: 'Agent is now idle', + message: state.status === 'error' ? 'Agent finished with error' : 'Agent is now idle', }, schema: agentWaitSchema, } diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client-v2.ts index bdd1dd161..73ed08ad7 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client-v2.ts @@ -1906,196 +1906,79 @@ export class DaemonClientV2 { ); } - async waitForAgentIdle( + async waitForFinish( agentId: string, timeout = 60000 ): Promise { - const current = this.agentIndex.get(agentId); - const pendingPermissionIds = new Set(); - if (current?.pendingPermissions) { - for (const request of current.pendingPermissions) { - pendingPermissionIds.add(request.id); - } - } + // We need to see the agent start (running/initializing) before we can + // consider idle/error as "finished". Otherwise we'd return the old idle + // state from before the task started. + // + // Permission requests are different - if there are pending permissions, + // the agent needs attention NOW regardless of whether we saw it start. + let sawStart = false; + let finishedState: AgentSnapshotPayload | null = null; - // Track whether we've seen a running state. Important: we only set this - // when we see running IN THE QUEUE, not based on current state. This prevents - // finding old idle states that occurred before the current run started. - let sawRunningInQueue = false; - let queuedIdle: AgentSnapshotPayload | null = null; + // Scan message queue for state changes + // We need to scan the ENTIRE queue to find the final state, not return early + for (const msg of this.messageQueue) { + if ( + msg.type === "agent_update" && + msg.payload.kind === "upsert" && + msg.payload.agent.id === agentId + ) { + const agent = msg.payload.agent; - const updatePendingPermissions = (msg: SessionOutboundMessage): void => { - if ( - msg.type === "agent_permission_request" && - msg.payload.agentId === agentId - ) { - pendingPermissionIds.add(msg.payload.request.id); - return; - } - if ( - msg.type === "agent_permission_resolved" && - msg.payload.agentId === agentId - ) { - pendingPermissionIds.delete(msg.payload.requestId); - return; - } - if (msg.type === "agent_stream" && msg.payload.agentId === agentId) { - if (msg.payload.event.type === "permission_requested") { - pendingPermissionIds.add(msg.payload.event.request.id); - } else if (msg.payload.event.type === "permission_resolved") { - pendingPermissionIds.delete(msg.payload.event.requestId); + // Track if agent has started processing a task + // Only "running" counts - "initializing" is the agent process starting up + if (agent.status === "running") { + sawStart = true; + finishedState = null; // Reset - any previous finished state was before this run + } + + // Check for finished state - save it but don't return yet + // We need to scan the whole queue to find the FINAL state + const hasPendingPermissions = (agent.pendingPermissions?.length ?? 0) > 0; + if (hasPendingPermissions) { + // Permission means agent needs attention + finishedState = agent; + } else if (sawStart && (agent.status === "idle" || agent.status === "error")) { + finishedState = agent; } } - }; - - for (const msg of this.messageQueue) { - updatePendingPermissions(msg); - if ( - msg.type !== "agent_update" || - msg.payload.kind !== "upsert" || - msg.payload.agent.id !== agentId - ) { - continue; - } - const status = msg.payload.agent.status; - const hasPendingPermissions = - (msg.payload.agent.pendingPermissions?.length ?? 0) > 0 || - pendingPermissionIds.size > 0; - if (status === "running" || hasPendingPermissions) { - sawRunningInQueue = true; - queuedIdle = null; // Reset: any previous idle was before this run - } - // Return immediately if we have pending permissions (even if still running) - if (sawRunningInQueue && hasPendingPermissions) { - return msg.payload.agent; - } - if ( - sawRunningInQueue && - (status === "idle" || status === "error") && - !hasPendingPermissions - ) { - queuedIdle = msg.payload.agent; - } - } - if (queuedIdle) { - return queuedIdle; } - // If current state is running (or has pending permissions), we need to wait - // for the next idle. If current is already idle and we didn't see running - // in the queue, this is an edge case - the agent might not have started yet - // or was already idle. Use the current state's running status to seed the waiter. - let sawRunning = - sawRunningInQueue || - current?.status === "running" || - pendingPermissionIds.size > 0; + // Only return from queue if we have a definitive finished state (idle/error) + // Don't return from queue if the latest state has pending permissions - + // the permission might be getting resolved right now, so wait for new messages + const hasPendingPermissionsInQueue = (finishedState?.pendingPermissions?.length ?? 0) > 0; + if (finishedState && !hasPendingPermissionsInQueue) { + return finishedState; + } + // Wait for agent to finish return this.waitFor( (msg) => { - updatePendingPermissions(msg); if ( msg.type === "agent_update" && msg.payload.kind === "upsert" && msg.payload.agent.id === agentId ) { - const status = msg.payload.agent.status; - const hasPendingPermissions = - (msg.payload.agent.pendingPermissions?.length ?? 0) > 0 || - pendingPermissionIds.size > 0; - if (status === "running" || hasPendingPermissions) { - sawRunning = true; - } - // Return if we have pending permissions (even if still running) - // OR if agent is idle/error with no pending permissions (after having run) - if (sawRunning && hasPendingPermissions) { - return msg.payload.agent; - } - if ( - sawRunning && - (status === "idle" || status === "error") && - !hasPendingPermissions - ) { - return msg.payload.agent; - } - } - return null; - }, - timeout, - { skipQueue: true } - ); - } + const agent = msg.payload.agent; - async waitForPermission( - agentId: string, - timeout = 30000 - ): Promise { - const snapshotPending = this.agentIndex.get(agentId)?.pendingPermissions?.[0]; - if (snapshotPending) { - return snapshotPending; - } - - let queuedRequest: AgentPermissionRequest | null = null; - const pendingById = new Map(); - for (const msg of this.messageQueue) { - if ( - msg.type === "agent_permission_request" && - msg.payload.agentId === agentId - ) { - pendingById.set(msg.payload.request.id, msg.payload.request); - queuedRequest = msg.payload.request; - continue; - } - if (msg.type === "agent_permission_resolved") { - if (msg.payload.agentId === agentId) { - pendingById.delete(msg.payload.requestId); - if (queuedRequest?.id === msg.payload.requestId) { - queuedRequest = null; + // Track if agent has started processing a task + // Only "running" counts - "initializing" is the agent process starting up + if (agent.status === "running") { + sawStart = true; } - } - continue; - } - if (msg.type === "agent_stream" && msg.payload.agentId === agentId) { - if (msg.payload.event.type === "permission_requested") { - pendingById.set( - msg.payload.event.request.id, - msg.payload.event.request - ); - queuedRequest = msg.payload.event.request; - continue; - } - if (msg.payload.event.type === "permission_resolved") { - pendingById.delete(msg.payload.event.requestId); - if (queuedRequest?.id === msg.payload.event.requestId) { - queuedRequest = null; + + // Check for finished state + const hasPendingPermissions = (agent.pendingPermissions?.length ?? 0) > 0; + if (hasPendingPermissions) { + return agent; } - } - } - } - - if (queuedRequest && pendingById.has(queuedRequest.id)) { - return queuedRequest; - } - if (pendingById.size > 0) { - let mostRecent: AgentPermissionRequest | null = null; - for (const request of pendingById.values()) { - mostRecent = request; - } - if (mostRecent) { - return mostRecent; - } - } - - return this.waitFor( - (msg) => { - if ( - msg.type === "agent_permission_request" && - msg.payload.agentId === agentId - ) { - return msg.payload.request; - } - if (msg.type === "agent_stream" && msg.payload.agentId === agentId) { - if (msg.payload.event.type === "permission_requested") { - return msg.payload.event.request; + if (sawStart && (agent.status === "idle" || agent.status === "error")) { + return agent; } } return null; diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 2c363ffdb..a7ce51867 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -4,6 +4,7 @@ import { createReadStream, unlinkSync, existsSync } from "fs"; import { stat } from "fs/promises"; import { randomUUID } from "node:crypto"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; +import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import type { Logger } from "pino"; @@ -264,6 +265,22 @@ export async function createPaseoDaemon( return transport; }; + // Create in-memory transport for Session's Agent MCP client (voice assistant tools) + const createInMemoryAgentMcpTransport = async (): Promise => { + const agentMcpServer = await createAgentMcpServer({ + agentManager, + agentStorage, + paseoHome: config.paseoHome, + logger, + }); + + const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); + + await agentMcpServer.connect(serverTransport); + + return clientTransport; + }; + const handleAgentMcpRequest: express.RequestHandler = async (req, res) => { if (config.mcpDebug) { logger.debug( @@ -491,8 +508,7 @@ export async function createPaseoDaemon( agentStorage, downloadTokenStore, config.paseoHome, - agentMcpRoute, - config.selfIdMcpSocketPath, + createInMemoryAgentMcpTransport, { allowedOrigins }, { stt: sttService, tts: ttsService }, terminalManager diff --git a/packages/server/src/server/daemon-client-v2.e2e.test.ts b/packages/server/src/server/daemon-client-v2.e2e.test.ts index adfcec316..68eca99d9 100644 --- a/packages/server/src/server/daemon-client-v2.e2e.test.ts +++ b/packages/server/src/server/daemon-client-v2.e2e.test.ts @@ -294,7 +294,7 @@ describe("daemon client v2 E2E", () => { } }); await ctx.client.sendMessage(agent.id, "Say 'hello' and nothing else"); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); unsubscribeStream(); unsubscribeRawStream(); expect(finalState.status).toBe("idle"); @@ -446,7 +446,9 @@ describe("daemon client v2 E2E", () => { ].join("\n") ); - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).toBeTruthy(); expect(permission.id).toBeTruthy(); @@ -460,7 +462,7 @@ describe("daemon client v2 E2E", () => { const permissionResolved = await permissionResolvedPromise; expect(permissionResolved.payload.requestId).toBe(permission.id); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); expect(existsSync(filePath)).toBe(true); } finally { @@ -485,7 +487,7 @@ describe("daemon client v2 E2E", () => { }); await ctx.client.sendMessage(agent.id, "Say 'hello' and nothing else"); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const snapshotPromise = waitForSignal(15000, (resolve) => { const unsubscribe = ctx.client.on("agent_stream_snapshot", (message) => { diff --git a/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts index 3c75df49f..2a68b6d3d 100644 --- a/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts @@ -46,7 +46,7 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, "Say 'hello world' and nothing else"); // Wait for the agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); // Verify agent completed without error expect(finalState.status).toBe("idle"); diff --git a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts index 3357f298f..507e3e4da 100644 --- a/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts @@ -105,7 +105,7 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, "Say 'test' and nothing else"); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); // The timestamp SHOULD have been updated (should be later than initial) @@ -328,7 +328,7 @@ describe("daemon E2E", () => { ctx.client.clearMessageQueue(); await ctx.client.sendMessage(agent.id, "Say 'hello' and nothing else"); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); // Mode should still be "read-only" after the message expect(finalState.currentModeId).toBe("read-only"); diff --git a/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts b/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts index a96b17bd8..39030e685 100644 --- a/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts @@ -143,7 +143,7 @@ describe("daemon restart and agent resume", () => { `Remember this marker string for a test: "${marker}". Just confirm you've remembered it with a short reply.` ); - const afterRemember = await currentDaemon.client.waitForAgentIdle(agent.id, 120000); + const afterRemember = await currentDaemon.client.waitForFinish(agent.id, 120000); expect(afterRemember.status).toBe("idle"); expect(afterRemember.lastError).toBeUndefined(); @@ -208,7 +208,7 @@ describe("daemon restart and agent resume", () => { "What was the marker string I asked you to remember earlier? Just reply with the exact string." ); - const afterMessage = await currentDaemon.client.waitForAgentIdle(resumedAgent.id, 120000); + const afterMessage = await currentDaemon.client.waitForFinish(resumedAgent.id, 120000); expect(afterMessage.status).toBe("idle"); expect(afterMessage.lastError).toBeUndefined(); diff --git a/packages/server/src/server/daemon-e2e/images.e2e.test.ts b/packages/server/src/server/daemon-e2e/images.e2e.test.ts index 7ba2b363b..2ed2f9640 100644 --- a/packages/server/src/server/daemon-e2e/images.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/images.e2e.test.ts @@ -64,7 +64,7 @@ describe("daemon E2E", () => { ); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); expect(finalState.lastError).toBeUndefined(); @@ -135,7 +135,7 @@ describe("daemon E2E", () => { ); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); expect(finalState.lastError).toBeUndefined(); diff --git a/packages/server/src/server/daemon-e2e/permissions-claude.e2e.test.ts b/packages/server/src/server/daemon-e2e/permissions-claude.e2e.test.ts index 65a3ee48c..59a5d3b99 100644 --- a/packages/server/src/server/daemon-e2e/permissions-claude.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/permissions-claude.e2e.test.ts @@ -18,9 +18,6 @@ function tmpCwd(): string { const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; const CODEX_TEST_REASONING_EFFORT = "low"; -const hasClaudeCredentials = - !!process.env.CLAUDE_SESSION_TOKEN || !!process.env.ANTHROPIC_API_KEY; - describe("daemon E2E", () => { let ctx: DaemonTestContext; @@ -32,7 +29,7 @@ describe("daemon E2E", () => { await ctx.cleanup(); }, 60000); - (hasClaudeCredentials ? describe : describe.skip)("permission flow: Claude", () => { + describe("permission flow: Claude", () => { // Use isolated Claude config to ensure permission prompts are triggered // (user's real config may have allow rules that auto-approve commands) let restoreClaudeConfig: () => void; @@ -81,7 +78,9 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, prompt); // Wait for permission request - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).not.toBeNull(); expect(permission.id).toBeTruthy(); expect(permission.kind).toBe("tool"); @@ -92,7 +91,7 @@ describe("daemon E2E", () => { }); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); // Verify the file was deleted @@ -152,7 +151,9 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, prompt); // Wait for permission request - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).not.toBeNull(); expect(permission.id).toBeTruthy(); @@ -163,7 +164,7 @@ describe("daemon E2E", () => { }); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); // Verify the file was NOT deleted diff --git a/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts b/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts index d091673d7..c0652ea42 100644 --- a/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts @@ -58,7 +58,9 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, prompt); // Wait for permission request - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).not.toBeNull(); expect(permission.id).toBeTruthy(); expect(permission.kind).toBe("tool"); @@ -69,7 +71,7 @@ describe("daemon E2E", () => { }); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); // Verify the file was created @@ -122,7 +124,9 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, prompt); // Wait for permission request - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).not.toBeNull(); expect(permission.id).toBeTruthy(); @@ -133,7 +137,7 @@ describe("daemon E2E", () => { }); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); // Verify the file was NOT created @@ -207,7 +211,7 @@ describe("daemon E2E", () => { ); // Wait for this to complete - await ctx.client.waitForAgentIdle(agent.id, 60000); + await ctx.client.waitForFinish(agent.id, 60000); // Verify we got an assistant message in the queue const queue = ctx.client.getMessageQueue(); @@ -295,7 +299,9 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, writePrompt); // Step 3: Wait for permission request - const permission = await ctx.client.waitForPermission(agent.id, 60000); + const permissionState = await ctx.client.waitForFinish(agent.id, 60000); + expect(permissionState.pendingPermissions?.length).toBeGreaterThan(0); + const permission = permissionState.pendingPermissions[0]; expect(permission).not.toBeNull(); expect(permission.id).toBeTruthy(); @@ -306,7 +312,7 @@ describe("daemon E2E", () => { }); // Wait for agent to complete after denial - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); // Verify file was NOT created after denial expect(existsSync(filePath)).toBe(false); @@ -353,7 +359,7 @@ describe("daemon E2E", () => { await ctx.client.sendMessage(agent.id, writePrompt2); // Wait for agent to complete - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); // Step 7: Verify file was created (mode switch worked) expect(existsSync(filePath)).toBe(true); diff --git a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts index 42ffdf3b9..4d60c35cf 100644 --- a/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/persistence.e2e.test.ts @@ -52,7 +52,7 @@ describe("daemon E2E", () => { ); // Wait for agent to complete - const afterMessage = await ctx.client.waitForAgentIdle(agent.id, 120000); + const afterMessage = await ctx.client.waitForFinish(agent.id, 120000); expect(afterMessage.status).toBe("idle"); // Get the timeline to verify we have messages @@ -118,7 +118,7 @@ describe("daemon E2E", () => { "What did I ask you to say earlier?" ); - const afterResume = await ctx.client.waitForAgentIdle( + const afterResume = await ctx.client.waitForFinish( resumedAgent.id, 120000 ); @@ -166,7 +166,7 @@ describe("daemon E2E", () => { ); // Wait for agent to complete - const afterMessage = await ctx.client.waitForAgentIdle(agent.id, 120000); + const afterMessage = await ctx.client.waitForFinish(agent.id, 120000); expect(afterMessage.status).toBe("idle"); // Verify we have timeline items before restart @@ -424,7 +424,7 @@ describe("daemon E2E", () => { "What was the number I asked you to remember earlier? Reply with just the number and nothing else." ); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); expect(finalState.lastError).toBeUndefined(); @@ -495,7 +495,7 @@ describe("daemon E2E", () => { ctx.client.clearMessageQueue(); await ctx.client.sendMessage(agent.id, "Say 'hello' and nothing else."); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 60000); + const finalState = await ctx.client.waitForFinish(agent.id, 60000); // Agent should complete (possibly with Codex warning about missing file, // but should still function) @@ -680,7 +680,7 @@ describe("daemon E2E", () => { `Remember this number: ${magicNumber}. Just confirm you've remembered it and reply with a single short sentence.` ); - const afterRemember = await ctx.client.waitForAgentIdle(agent.id, 120000); + const afterRemember = await ctx.client.waitForFinish(agent.id, 120000); expect(afterRemember.status).toBe("idle"); expect(afterRemember.lastError).toBeUndefined(); @@ -729,7 +729,7 @@ describe("daemon E2E", () => { "What was the number I asked you to remember earlier? Reply with just the number and nothing else." ); - const afterRecall = await ctx.client.waitForAgentIdle(resumedAgent.id, 120000); + const afterRecall = await ctx.client.waitForFinish(resumedAgent.id, 120000); expect(afterRecall.status).toBe("idle"); expect(afterRecall.lastError).toBeUndefined(); diff --git a/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts b/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts index 219b53d4f..75aa70c11 100644 --- a/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts @@ -33,31 +33,19 @@ describe("self-id MCP e2e", () => { "Use the set_title MCP tool to change your title to 'Updated via MCP'. Only call set_title, nothing else." ); - // Wait for agent to complete (MCP tools may auto-approve without permission) - // If a permission is requested, approve it - let finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + // Wait for permission request (default mode requires permission for MCP tools) + const state = await ctx.client.waitForFinish(agent.id, 60000); + expect(state.pendingPermissions?.length).toBeGreaterThan(0); + expect(state.pendingPermissions![0].name).toBe("mcp__paseo-self-id__set_title"); - // Check if we got blocked on permission - if (finalState.status === "running" && finalState.pendingPermissions?.length) { - const permission = finalState.pendingPermissions[0]; - await ctx.client.respondToPermission(agent.id, permission.id, { - behavior: "allow", - }); - finalState = await ctx.client.waitForAgentIdle(agent.id, 60000); - } - - // Log final state for debugging if not idle - if (finalState.status !== "idle") { - console.error( - "Agent did not reach idle state:", - JSON.stringify(finalState, null, 2) - ); - } + // Approve the permission + await ctx.client.respondToPermission(agent.id, state.pendingPermissions![0].id, { + behavior: "allow", + }); + // Wait for agent to complete + const finalState = await ctx.client.waitForFinish(agent.id, 60000); expect(finalState.status).toBe("idle"); - expect(finalState.lastError).toBeUndefined(); - - // Verify the title was changed via set_title expect(finalState.title).toBe("Updated via MCP"); }, 180000); }); diff --git a/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts b/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts index 9b7db9056..4521b337c 100644 --- a/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/streaming.e2e.test.ts @@ -17,12 +17,7 @@ function tmpCwd(): string { const CODEX_TEST_MODEL = "gpt-5.1-codex-mini"; const CODEX_TEST_REASONING_EFFORT = "low"; -const hasClaudeCredentials = - !!process.env.CLAUDE_SESSION_TOKEN || !!process.env.ANTHROPIC_API_KEY; - -const describeWithClaude = hasClaudeCredentials ? describe : describe.skip; - -describeWithClaude("daemon E2E", () => { +describe("daemon E2E", () => { let ctx: DaemonTestContext; beforeEach(async () => { @@ -57,7 +52,7 @@ describeWithClaude("daemon E2E", () => { ); // Wait for agent to complete - const finalState = await ctx.client.waitForAgentIdle(agent.id, 120000); + const finalState = await ctx.client.waitForFinish(agent.id, 120000); expect(finalState.status).toBe("idle"); expect(finalState.lastError).toBeUndefined(); @@ -179,7 +174,7 @@ describeWithClaude("daemon E2E", () => { "Remember the number 42. Just confirm you remember it." ); - let state = await ctx.client.waitForAgentIdle(agent.id, 120000); + let state = await ctx.client.waitForFinish(agent.id, 120000); expect(state.status).toBe("idle"); expect(state.lastError).toBeUndefined(); @@ -191,7 +186,7 @@ describeWithClaude("daemon E2E", () => { "Now remember the word 'elephant'. Just confirm you remember both the number and the word." ); - state = await ctx.client.waitForAgentIdle(agent.id, 120000); + state = await ctx.client.waitForFinish(agent.id, 120000); expect(state.status).toBe("idle"); expect(state.lastError).toBeUndefined(); @@ -204,7 +199,7 @@ describeWithClaude("daemon E2E", () => { "Write a complete sentence using both the number (42) and the word (elephant) you remembered. The sentence should be grammatically correct English." ); - state = await ctx.client.waitForAgentIdle(agent.id, 120000); + state = await ctx.client.waitForFinish(agent.id, 120000); expect(state.status).toBe("idle"); expect(state.lastError).toBeUndefined(); @@ -594,7 +589,7 @@ describeWithClaude("daemon E2E", () => { // Agent should go idle within 5 seconds after interrupt log(`waiting for idle...`); - const finalState = await ctx.client.waitForAgentIdle(agent.id, 10000); + const finalState = await ctx.client.waitForFinish(agent.id, 10000); const idleReceivedAt = Date.now(); log(`got idle state: ${finalState.status} (took ${idleReceivedAt - stopSentAt}ms after Stop)`); expect(finalState.status).toBe("idle"); diff --git a/packages/server/src/server/daemon-e2e/tool-calls.e2e.test.ts b/packages/server/src/server/daemon-e2e/tool-calls.e2e.test.ts index 4c15dc87d..d0d91b4a3 100644 --- a/packages/server/src/server/daemon-e2e/tool-calls.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/tool-calls.e2e.test.ts @@ -81,7 +81,7 @@ describe("daemon E2E", () => { "Read the file /etc/hosts and tell me how many lines it has. Be brief." ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); @@ -120,7 +120,7 @@ describe("daemon E2E", () => { "Run `echo hello` and tell me what it outputs. Be brief." ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); @@ -164,7 +164,7 @@ describe("daemon E2E", () => { `Edit the file ${testFile} and change "hello" to "goodbye". Be brief.` ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); @@ -202,7 +202,7 @@ describe("daemon E2E", () => { "Run `echo hello` and tell me what it outputs. Be brief." ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); @@ -248,7 +248,7 @@ describe("daemon E2E", () => { "Read the file /etc/hosts and tell me how many lines it has. Be brief." ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); @@ -292,7 +292,7 @@ describe("daemon E2E", () => { `Edit the file ${testFile} and change "hello" to "goodbye". Be brief.` ); - await ctx.client.waitForAgentIdle(agent.id, 120000); + await ctx.client.waitForFinish(agent.id, 120000); const toolCalls = extractToolCalls(ctx.client.getMessageQueue(), agent.id); expect(toolCalls.length).toBeGreaterThan(0); diff --git a/packages/server/src/server/daemon-e2e/two-cycle-resume.e2e.test.ts b/packages/server/src/server/daemon-e2e/two-cycle-resume.e2e.test.ts index 00a70b4e7..02772715c 100644 --- a/packages/server/src/server/daemon-e2e/two-cycle-resume.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/two-cycle-resume.e2e.test.ts @@ -53,7 +53,7 @@ describe("two-cycle Codex agent resume", () => { `For this test session, remember this project name: "${MARKER}". Just confirm you've noted it.` ); - const afterSecret = await ctx.client.waitForAgentIdle(agent.id, 120000); + const afterSecret = await ctx.client.waitForFinish(agent.id, 120000); expect(afterSecret.status).toBe("idle"); expect(afterSecret.lastError).toBeUndefined(); @@ -103,7 +103,7 @@ describe("two-cycle Codex agent resume", () => { "Acknowledge you still remember the project name. Just say yes or no." ); - const afterAck = await ctx.client.waitForAgentIdle(resumed1.id, 120000); + const afterAck = await ctx.client.waitForFinish(resumed1.id, 120000); expect(afterAck.status).toBe("idle"); expect(afterAck.lastError).toBeUndefined(); @@ -134,7 +134,7 @@ describe("two-cycle Codex agent resume", () => { "What was the project name I asked you to remember at the very beginning of our conversation? Reply with the exact name." ); - const afterRecall = await ctx.client.waitForAgentIdle(resumed2.id, 120000); + const afterRecall = await ctx.client.waitForFinish(resumed2.id, 120000); expect(afterRecall.status).toBe("idle"); expect(afterRecall.lastError).toBeUndefined(); diff --git a/packages/server/src/server/daemon-e2e/wait-for-idle.e2e.test.ts b/packages/server/src/server/daemon-e2e/wait-for-idle.e2e.test.ts index 796f5a902..f32253676 100644 --- a/packages/server/src/server/daemon-e2e/wait-for-idle.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/wait-for-idle.e2e.test.ts @@ -12,10 +12,10 @@ function tmpCwd(): string { } /** - * Tests for waitForAgentIdle edge cases. + * Tests for waitForFinish edge cases. * Uses haiku for speed. Allow higher timeouts in CI / congested environments. */ -describe("waitForAgentIdle edge cases", () => { +describe("waitForFinish edge cases", () => { let ctx: DaemonTestContext; beforeEach(async () => { @@ -26,7 +26,7 @@ describe("waitForAgentIdle edge cases", () => { await ctx.cleanup(); }, 30000); - test("waitForAgentIdle immediately after sendMessage", async () => { + test("waitForFinish immediately after sendMessage", async () => { const cwd = tmpCwd(); const agent = await ctx.client.createAgent({ @@ -37,9 +37,9 @@ describe("waitForAgentIdle edge cases", () => { modeId: "bypassPermissions", }); - // This was the original bug: waitForAgentIdle returned old idle states + // This was the original bug: waitForFinish returned old idle states await ctx.client.sendMessage(agent.id, "Say 'hello'"); - const state = await ctx.client.waitForAgentIdle(agent.id, 30000); + const state = await ctx.client.waitForFinish(agent.id, 30000); expect(state.status).toBe("idle"); @@ -58,13 +58,13 @@ describe("waitForAgentIdle edge cases", () => { modeId: "bypassPermissions", }); - // Send 3 messages without waiting - tests that waitForAgentIdle + // Send 3 messages without waiting - tests that waitForFinish // finds the idle AFTER the last running state await ctx.client.sendMessage(agent.id, "Say 'one'"); await ctx.client.sendMessage(agent.id, "Say 'two'"); await ctx.client.sendMessage(agent.id, "Say 'three'"); - const state = await ctx.client.waitForAgentIdle(agent.id, 30000); + const state = await ctx.client.waitForFinish(agent.id, 30000); expect(state.status).toBe("idle"); // Verify all 3 messages were recorded @@ -82,7 +82,7 @@ describe("waitForAgentIdle edge cases", () => { rmSync(cwd, { recursive: true, force: true }); }, 45000); - test("two agents: waitForAgentIdle filters by agent", async () => { + test("two agents: waitForFinish filters by agent", async () => { const cwd1 = tmpCwd(); const cwd2 = tmpCwd(); @@ -107,11 +107,11 @@ describe("waitForAgentIdle edge cases", () => { await ctx.client.sendMessage(agent2.id, "Say 'agent two'"); // Wait for each - should not be confused by the other's state - const state2 = await ctx.client.waitForAgentIdle(agent2.id, 30000); + const state2 = await ctx.client.waitForFinish(agent2.id, 30000); expect(state2.status).toBe("idle"); expect(state2.id).toBe(agent2.id); - const state1 = await ctx.client.waitForAgentIdle(agent1.id, 30000); + const state1 = await ctx.client.waitForFinish(agent1.id, 30000); expect(state1.status).toBe("idle"); expect(state1.id).toBe(agent1.id); diff --git a/packages/server/src/server/daemon-e2e/wait-for-permission.e2e.test.ts b/packages/server/src/server/daemon-e2e/wait-for-permission.e2e.test.ts deleted file mode 100644 index b8d910cd2..000000000 --- a/packages/server/src/server/daemon-e2e/wait-for-permission.e2e.test.ts +++ /dev/null @@ -1,123 +0,0 @@ -import { describe, test, expect, beforeEach, afterEach } from "vitest"; -import { mkdtempSync, rmSync } from "fs"; -import { tmpdir } from "os"; -import path from "path"; -import { - createDaemonTestContext, - type DaemonTestContext, -} from "../test-utils/index.js"; - -function tmpCwd(): string { - return mkdtempSync(path.join(tmpdir(), "wait-perm-e2e-")); -} - -/** - * Tests for wait returning on permission request. - * - * The `paseo wait` command should return when: - * 1. Agent completes (goes idle) - * 2. Agent requests permission - * - * This test verifies that waitForAgentIdle correctly returns when - * an agent requests permission, not just when it goes idle. - */ -describe("wait returns on permission request", () => { - let ctx: DaemonTestContext; - - beforeEach(async () => { - ctx = await createDaemonTestContext(); - }); - - afterEach(async () => { - await ctx.cleanup(); - }, 60000); - - describe("Claude provider", () => { - test( - "waitForAgentIdle returns when permission is requested (not just when idle)", - async () => { - const cwd = tmpCwd(); - const testFilePath = path.join(cwd, "test-wait-perm.txt"); - - // Create Claude agent with default mode (always ask for permissions) - const agent = await ctx.client.createAgent({ - provider: "claude", - model: "haiku", - cwd, - title: "Wait Permission Test", - modeId: "default", - }); - - expect(agent.id).toBeTruthy(); - expect(agent.status).toBe("idle"); - - // Clear message queue before sending prompt - ctx.client.clearMessageQueue(); - - // Send a prompt that requires file write permission - const prompt = [ - `You must use the Write tool to create a file at "${testFilePath}" with content "hello".`, - "Do not respond before attempting to write the file.", - ].join(" "); - - await ctx.client.sendMessage(agent.id, prompt); - - // CRITICAL: This is the behavior we're testing - // waitForAgentIdle should return when permission is requested, - // NOT wait until the agent is fully idle (which would timeout or - // require us to approve the permission first) - const startTime = Date.now(); - const state = await ctx.client.waitForAgentIdle(agent.id, 60000); - const waitDuration = Date.now() - startTime; - - // If wait returns because of permission request, we should have: - // 1. Agent status still "running" (not idle yet - waiting for permission) - // 2. Pending permissions in the state - // 3. Wait should return quickly (under 30 seconds, not timeout) - - // Check that we have a pending permission - const hasPendingPermission = - (state.pendingPermissions && state.pendingPermissions.length > 0); - - // Log for debugging - console.log("Wait returned after", waitDuration, "ms"); - console.log("Agent status:", state.status); - console.log("Pending permissions:", state.pendingPermissions?.length ?? 0); - - // THE ASSERTION: - // If waitForAgentIdle correctly yields on permission request, - // we should either: - // - Get a state with pending permissions (status might be "running") - // - Or the state should be from right when permission was requested - // - // If waitForAgentIdle does NOT yield on permission request, - // this test will either: - // - Timeout (60s) - // - Return only after we never approve permission and agent errors/gives up - - // This test will FAIL if waitForAgentIdle waits for full idle - // instead of returning on permission request - expect(hasPendingPermission).toBe(true); - - // Also verify we can get the permission via waitForPermission - // (This should return immediately since permission is already pending) - const permission = await ctx.client.waitForPermission(agent.id, 5000); - expect(permission).toBeTruthy(); - expect(permission.kind).toBe("tool"); - - // Clean up: deny the permission so agent can finish - await ctx.client.respondToPermission(agent.id, permission.id, { - behavior: "deny", - message: "Test complete", - }); - - // Wait for agent to finish processing the denial - await ctx.client.waitForAgentIdle(agent.id, 30000); - - await ctx.client.deleteAgent(agent.id); - rmSync(cwd, { recursive: true, force: true }); - }, - 120000 - ); - }); -}); diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 06ac314a0..38abeb9cc 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -3,7 +3,6 @@ import { readFile, mkdir, writeFile, stat } from "fs/promises"; import { exec } from "child_process"; import { promisify, inspect } from "util"; import { join, resolve, sep } from "path"; -import http from "http"; import invariant from "tiny-invariant"; import { z } from "zod"; import { streamText, stepCountIs } from "ai"; @@ -46,7 +45,9 @@ import { extractTimestamps, } from "./persistence-hooks.js"; import { experimental_createMCPClient } from "ai"; -import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; + +export type AgentMcpTransportFactory = () => Promise; import { buildProviderRegistry } from "./agent/provider-registry.js"; import { AgentManager } from "./agent/agent-manager.js"; import type { ManagedAgent } from "./agent/agent-manager.js"; @@ -290,8 +291,7 @@ export class Session { private agentTools: ToolSet | null = null; private agentManager: AgentManager; private readonly agentStorage: AgentStorage; - private readonly agentMcpRoute: string; - private readonly mcpSocketPath: string; + private readonly createAgentMcpTransport: AgentMcpTransportFactory; private readonly downloadTokenStore: DownloadTokenStore; private readonly pushTokenStore: PushTokenStore; private readonly providerRegistry: ReturnType; @@ -320,8 +320,7 @@ export class Session { paseoHome: string, agentManager: AgentManager, agentStorage: AgentStorage, - agentMcpRoute: string, - mcpSocketPath: string, + createAgentMcpTransport: AgentMcpTransportFactory, stt: OpenAISTT | null, tts: OpenAITTS | null, terminalManager: TerminalManager | null, @@ -335,8 +334,7 @@ export class Session { this.paseoHome = paseoHome; this.agentManager = agentManager; this.agentStorage = agentStorage; - this.agentMcpRoute = agentMcpRoute; - this.mcpSocketPath = mcpSocketPath; + this.createAgentMcpTransport = createAgentMcpTransport; this.terminalManager = terminalManager; this.voiceConversationStore = voiceConversationStore; this.abortController = new AbortController(); @@ -506,61 +504,12 @@ export class Session { } /** - * Initialize Agent MCP client for this session + * Initialize Agent MCP client for this session using in-memory transport */ private async initializeAgentMcp(): Promise { try { - // Create a custom fetch that uses the Unix socket - const socketFetch = async ( - input: string | URL, - init?: RequestInit - ): Promise => { - const url = new URL(input.toString()); - const path = url.pathname + url.search; - - return new Promise((resolve, reject) => { - const req = http.request( - { - socketPath: this.mcpSocketPath, - path, - method: init?.method ?? "GET", - headers: { - ...Object.fromEntries( - new Headers(init?.headers).entries() - ), - }, - }, - (res: import("http").IncomingMessage) => { - const chunks: Buffer[] = []; - res.on("data", (chunk: Buffer) => chunks.push(chunk)); - res.on("end", () => { - const body = Buffer.concat(chunks); - resolve( - new Response(body, { - status: res.statusCode ?? 500, - statusText: res.statusMessage ?? "", - headers: new Headers( - res.headers as Record - ), - }) - ); - }); - res.on("error", reject); - } - ); - req.on("error", reject); - if (init?.body) { - req.write(init.body); - } - req.end(); - }); - }; - - // Connect to the local MCP server using Unix socket - const transport = new StreamableHTTPClientTransport( - new URL(`http://localhost${this.agentMcpRoute}`), - { fetch: socketFetch as typeof fetch } - ); + // Create an in-memory transport connected to the Agent MCP server + const transport = await this.createAgentMcpTransport(); this.agentMcpClient = await experimental_createMCPClient({ transport, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 3755b45fd..58644d840 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,5 +1,6 @@ import { WebSocketServer } from "ws"; import type { Server as HTTPServer } from "http"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; @@ -10,6 +11,8 @@ import type pino from "pino"; import type { WSOutboundMessage } from "./messages.js"; import { WebSocketSessionBridge } from "./websocket-session-bridge.js"; +export type AgentMcpTransportFactory = () => Promise; + type WebSocketServerConfig = { allowedOrigins: Set; }; @@ -29,8 +32,7 @@ export class VoiceAssistantWebSocketServer { agentStorage: AgentStorage, downloadTokenStore: DownloadTokenStore, paseoHome: string, - agentMcpRoute: string, - selfIdMcpSocketPath: string, + createAgentMcpTransport: AgentMcpTransportFactory, wsConfig: WebSocketServerConfig, speech?: { stt: OpenAISTT | null; tts: OpenAITTS | null }, terminalManager?: TerminalManager | null @@ -42,8 +44,7 @@ export class VoiceAssistantWebSocketServer { agentStorage, downloadTokenStore, paseoHome, - agentMcpRoute, - selfIdMcpSocketPath, + createAgentMcpTransport, speech, terminalManager ); diff --git a/packages/server/src/server/websocket-session-bridge.ts b/packages/server/src/server/websocket-session-bridge.ts index c4a4c112a..9cb657acb 100644 --- a/packages/server/src/server/websocket-session-bridge.ts +++ b/packages/server/src/server/websocket-session-bridge.ts @@ -1,5 +1,6 @@ import type { IncomingMessage } from "http"; import type { WebSocket } from "ws"; +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; import { join } from "path"; import { WSInboundMessageSchema, @@ -19,6 +20,8 @@ import type { OpenAITTS } from "./agent/tts-openai.js"; import type { TerminalManager } from "../terminal/terminal-manager.js"; import type pino from "pino"; +export type AgentMcpTransportFactory = () => Promise; + export class WebSocketSessionBridge { private readonly logger: pino.Logger; private readonly sessions: Map = new Map(); @@ -29,8 +32,7 @@ export class WebSocketSessionBridge { private readonly paseoHome: string; private readonly pushTokenStore: PushTokenStore; private readonly pushService: PushService; - private readonly agentMcpRoute: string; - private readonly selfIdMcpSocketPath: string; + private readonly createAgentMcpTransport: AgentMcpTransportFactory; private readonly stt: OpenAISTT | null; private readonly tts: OpenAITTS | null; private readonly terminalManager: TerminalManager | null; @@ -42,8 +44,7 @@ export class WebSocketSessionBridge { agentStorage: AgentStorage, downloadTokenStore: DownloadTokenStore, paseoHome: string, - agentMcpRoute: string, - selfIdMcpSocketPath: string, + createAgentMcpTransport: AgentMcpTransportFactory, speech?: { stt: OpenAISTT | null; tts: OpenAITTS | null }, terminalManager?: TerminalManager | null ) { @@ -52,8 +53,7 @@ export class WebSocketSessionBridge { this.agentStorage = agentStorage; this.downloadTokenStore = downloadTokenStore; this.paseoHome = paseoHome; - this.agentMcpRoute = agentMcpRoute; - this.selfIdMcpSocketPath = selfIdMcpSocketPath; + this.createAgentMcpTransport = createAgentMcpTransport; this.stt = speech?.stt ?? null; this.tts = speech?.tts ?? null; this.terminalManager = terminalManager ?? null; @@ -89,8 +89,7 @@ export class WebSocketSessionBridge { this.paseoHome, this.agentManager, this.agentStorage, - this.agentMcpRoute, - this.selfIdMcpSocketPath, + this.createAgentMcpTransport, this.stt, this.tts, this.terminalManager,