jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: consolidate agent wait methods and use in-memory MCP transport

Browse Source
This commit is contained in:
Mohamed Boudra
2026-01-30 14:50:01 +07:00
parent 0618dc6b0d
commit 25c7d5afa9
24 changed files with 191 additions and 469 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/client/daemon-client-v2.ts
+53 -170
View File
@@ -1906,196 +1906,79 @@ export class DaemonClientV2 {
);
}
async waitForAgentIdle(
async waitForFinish(
agentId: string,
timeout = 60000
): Promise<AgentSnapshotPayload> {
const current = this.agentIndex.get(agentId);
const pendingPermissionIds = new Set<string>();
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<AgentPermissionRequest> {
const snapshotPending = this.agentIndex.get(agentId)?.pendingPermissions?.[0];
if (snapshotPending) {
return snapshotPending;
}
let queuedRequest: AgentPermissionRequest | null = null;
const pendingById = new Map<string, AgentPermissionRequest>();
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;
packages/server/src/server/bootstrap.ts
+18 -2
View File
@@ -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<InMemoryTransport> => {
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
packages/server/src/server/daemon-client-v2.e2e.test.ts
+6 -4
View File
@@ -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) => {
packages/server/src/server/daemon-e2e/agent-basics.e2e.test.ts
+1 -1
View File
@@ -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");
packages/server/src/server/daemon-e2e/agent-operations.e2e.test.ts
+2 -2
View File
@@ -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");
packages/server/src/server/daemon-e2e/daemon-restart-resume.e2e.test.ts
+2 -2
View File
@@ -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();
packages/server/src/server/daemon-e2e/images.e2e.test.ts
+2 -2
View File
@@ -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();
packages/server/src/server/daemon-e2e/permissions-claude.e2e.test.ts
+9 -8
View File
@@ -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
packages/server/src/server/daemon-e2e/permissions-codex.e2e.test.ts
+14 -8
View File
@@ -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);
packages/server/src/server/daemon-e2e/persistence.e2e.test.ts
+7 -7
View File
@@ -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();
packages/server/src/server/daemon-e2e/self-id-mcp.e2e.test.ts
+10 -22
View File
@@ -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);
});
packages/server/src/server/daemon-e2e/streaming.e2e.test.ts
+6 -11
View File
@@ -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");
packages/server/src/server/daemon-e2e/tool-calls.e2e.test.ts
+6 -6
View File
@@ -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);
packages/server/src/server/daemon-e2e/two-cycle-resume.e2e.test.ts
+3 -3
View File
@@ -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();
packages/server/src/server/daemon-e2e/wait-for-idle.e2e.test.ts
+10 -10
View File
@@ -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);
packages/server/src/server/daemon-e2e/wait-for-permission.e2e.test.ts
-123
View File
@@ -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
);
});
});
packages/server/src/server/session.ts
+9 -60
View File
@@ -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<Transport>;
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<typeof buildProviderRegistry>;
@@ -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<void> {
try {
// Create a custom fetch that uses the Unix socket
const socketFetch = async (
input: string | URL,
init?: RequestInit
): Promise<Response> => {
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<string, string>
),
})
);
});
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,
packages/server/src/server/websocket-server.ts
+5 -4
View File
@@ -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<Transport>;
type WebSocketServerConfig = {
allowedOrigins: Set<string>;
};
@@ -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
);
packages/server/src/server/websocket-session-bridge.ts
+7 -8
View File
@@ -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<Transport>;
export class WebSocketSessionBridge {
private readonly logger: pino.Logger;
private readonly sessions: Map<WebSocket, Session> = 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,