Merge origin/main

This commit is contained in:
Mohamed Boudra
2025-12-24 17:21:26 +07:00
3 changed files with 230 additions and 122 deletions

View File

@@ -60,6 +60,7 @@ export type AgentManagerOptions = {
export type WaitForAgentOptions = {
signal?: AbortSignal;
waitForActive?: boolean;
};
export type WaitForAgentResult = {
@@ -611,9 +612,12 @@ export class AgentManager {
};
}
if (!isAgentBusy(snapshot.lifecycle)) {
const initialStatus = snapshot.lifecycle;
const initialBusy = isAgentBusy(initialStatus);
const waitForActive = options?.waitForActive ?? false;
if (!waitForActive && !initialBusy) {
return {
status: snapshot.lifecycle,
status: initialStatus,
permission: null,
lastMessage: this.getLastAssistantMessage(agentId)
};
@@ -632,7 +636,8 @@ export class AgentManager {
return;
}
let currentStatus: AgentLifecycleStatus = snapshot.lifecycle;
let currentStatus: AgentLifecycleStatus = initialStatus;
let hasStarted = initialBusy;
// Bug #3 Fix: Declare unsubscribe and abortHandler upfront so cleanup can reference them
let unsubscribe: (() => void) | null = null;
@@ -691,7 +696,11 @@ export class AgentManager {
finish(pending);
return;
}
if (!isAgentBusy(event.agent.lifecycle)) {
if (isAgentBusy(event.agent.lifecycle)) {
hasStarted = true;
return;
}
if (!waitForActive || hasStarted) {
finish(null);
}
}

View File

@@ -1,3 +1,4 @@
import net from "node:net";
import os from "node:os";
import path from "node:path";
import { mkdtemp, readFile, rm } from "node:fs/promises";
@@ -5,20 +6,15 @@ import { describe, expect, test } from "vitest";
import { experimental_createMCPClient } from "ai";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
import { createPaseoDaemon, type PaseoDaemonConfig } from "../bootstrap.js";
type McpToolResult = {
structuredContent?: Record<string, unknown>;
content?: Array<
{ structuredContent?: Record<string, unknown> } | Record<string, unknown>
>;
content?: Array<{ structuredContent?: Record<string, unknown> } | Record<string, unknown>>;
};
type McpClient = {
callTool: (input: {
name: string;
args?: Record<string, unknown>;
}) => Promise<unknown>;
callTool: (input: { name: string; args?: Record<string, unknown> }) => Promise<unknown>;
close: () => Promise<void>;
};
@@ -26,13 +22,23 @@ type PermissionPayload = {
id: string;
};
function getStructuredContent(
result: McpToolResult
): Record<string, unknown> | null {
if (
result.structuredContent &&
typeof result.structuredContent === "object"
) {
async function getAvailablePort(): Promise<number> {
return new Promise((resolve, reject) => {
const server = net.createServer();
server.once("error", reject);
server.listen(0, () => {
const address = server.address();
if (!address || typeof address === "string") {
server.close(() => reject(new Error("Failed to acquire port")));
return;
}
server.close(() => resolve(address.port));
});
});
}
function getStructuredContent(result: McpToolResult): Record<string, unknown> | null {
if (result.structuredContent && typeof result.structuredContent === "object") {
return result.structuredContent;
}
const content = result.content?.[0];
@@ -45,10 +51,7 @@ function getStructuredContent(
return null;
}
async function waitForFile(
filePath: string,
timeoutMs = 30000
): Promise<string> {
async function waitForFile(filePath: string, timeoutMs = 30000): Promise<string> {
const startedAt = Date.now();
while (Date.now() - startedAt < timeoutMs) {
try {
@@ -82,109 +85,199 @@ async function waitForAgentCompletion(
}
}
const hasClaudeCredentials = Boolean(
process.env.CLAUDE_CODE_OAUTH_TOKEN || process.env.ANTHROPIC_API_KEY
);
describe("agent MCP end-to-end", () => {
test("creates a Claude agent and writes a file", async () => {
const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-agent-cwd-"));
const daemonHandle = await createTestPaseoDaemon();
const runTest = hasClaudeCredentials ? test : test.skip;
runTest(
"creates a Claude agent and writes a file",
async () => {
const paseoHome = await mkdtemp(path.join(os.tmpdir(), "paseo-home-"));
const staticDir = await mkdtemp(path.join(os.tmpdir(), "paseo-static-"));
const agentCwd = await mkdtemp(path.join(os.tmpdir(), "paseo-agent-cwd-"));
const port = await getAvailablePort();
const basicUsers = { test: "pass" };
const [agentMcpUser, agentMcpPassword] =
Object.entries(basicUsers)[0] ?? [];
const agentMcpAuthHeader =
agentMcpUser && agentMcpPassword
? `Basic ${Buffer.from(`${agentMcpUser}:${agentMcpPassword}`).toString("base64")}`
: undefined;
const agentMcpBearerToken =
agentMcpUser && agentMcpPassword
? Buffer.from(`${agentMcpUser}:${agentMcpPassword}`).toString("base64")
: undefined;
const transport = new StreamableHTTPClientTransport(
new URL(`http://127.0.0.1:${daemonHandle.port}/mcp/agents`),
daemonHandle.agentMcpAuthHeader
? {
requestInit: {
headers: { Authorization: daemonHandle.agentMcpAuthHeader },
},
}
: undefined
);
const client = (await experimental_createMCPClient({
transport,
})) as McpClient;
let agentId: string | null = null;
try {
const result = (await client.callTool({
name: "create_agent",
args: {
cwd: agentCwd,
title: "MCP e2e codex smoke",
agentType: "codex",
initialMode: "read-only",
background: false,
const daemonConfig: PaseoDaemonConfig = {
port,
paseoHome,
agentMcpRoute: "/mcp/agents",
agentMcpAllowedHosts: [`127.0.0.1:${port}`, `localhost:${port}`],
auth: {
basicUsers,
agentMcpAuthHeader,
agentMcpBearerToken,
realm: "Voice Assistant",
},
})) as McpToolResult;
const payload = getStructuredContent(result);
expect(payload).toBeTruthy();
agentId = payload?.agentId as string | null;
expect(agentId).toBeTruthy();
const prompt = [
'Run this command exactly: ["bash", "-lc", "echo ok > mcp-smoke.txt"].',
"After the command runs, reply with done and stop.",
].join("\n");
const promptResult = (await client.callTool({
name: "send_agent_prompt",
args: {
agentId,
prompt,
sessionMode: "read-only",
background: false,
},
})) as McpToolResult;
const promptPayload = getStructuredContent(promptResult);
const permission = promptPayload?.permission as PermissionPayload | null;
if (!permission?.id) {
const activityResult = (await client.callTool({
name: "get_agent_activity",
args: { agentId, limit: 10 },
})) as McpToolResult;
console.log(
"[agent-mcp.e2e] send_agent_prompt payload:",
promptPayload
);
console.log(
"[agent-mcp.e2e] get_agent_activity payload:",
getStructuredContent(activityResult)
);
}
expect(permission?.id).toBeTruthy();
await client.callTool({
name: "respond_to_permission",
args: {
agentId,
requestId: permission!.id,
response: { behavior: "allow" },
staticDir,
mcpDebug: false,
agentClients: {},
agentRegistryPath: path.join(paseoHome, "agents.json"),
agentControlMcp: {
url: `http://127.0.0.1:${port}/mcp/agents`,
...(agentMcpAuthHeader
? { headers: { Authorization: agentMcpAuthHeader } }
: {}),
},
};
const previousCodexSessionDir = process.env.CODEX_SESSION_DIR;
const previousCodexHome = process.env.CODEX_HOME;
const codexSessionDir = await mkdtemp(
path.join(os.tmpdir(), "codex-session-")
);
const codexHome = await mkdtemp(path.join(os.tmpdir(), "codex-home-"));
process.env.CODEX_SESSION_DIR = codexSessionDir;
process.env.CODEX_HOME = codexHome;
const daemon = await createPaseoDaemon(daemonConfig);
await new Promise<void>((resolve) => {
daemon.httpServer.listen(port, () => resolve());
});
await waitForAgentCompletion(client, agentId);
const filePath = path.join(agentCwd, "mcp-smoke.txt");
let contents: string;
const transport = new StreamableHTTPClientTransport(
new URL(`http://127.0.0.1:${port}/mcp/agents`),
agentMcpAuthHeader
? { requestInit: { headers: { Authorization: agentMcpAuthHeader } } }
: undefined
);
const client = (await experimental_createMCPClient({
transport,
})) as McpClient;
let agentId: string | null = null;
try {
contents = await waitForFile(filePath);
} catch (error) {
const activityResult = (await client.callTool({
name: "get_agent_activity",
args: { agentId, limit: 25 },
const initialPrompt = [
"You must call the tool named shell.",
"Run this command exactly: [\"bash\", \"-lc\", \"echo ok > mcp-smoke.txt\"].",
"After the tool runs, reply with done and stop.",
].join("\n");
const result = (await client.callTool({
name: "create_agent",
args: {
cwd: agentCwd,
title: "MCP e2e smoke",
agentType: "claude",
initialMode: "default",
initialPrompt,
background: false,
},
})) as McpToolResult;
const activityPayload = getStructuredContent(activityResult);
const activitySummary = activityPayload?.content;
const details = activitySummary
? `Agent activity:\n${activitySummary}`
: "Agent activity unavailable";
throw new Error(`${(error as Error).message}\n${details}`);
const payload = getStructuredContent(result);
expect(payload).toBeTruthy();
agentId = payload?.agentId as string | null;
expect(agentId).toBeTruthy();
const createPermission = payload?.permission as PermissionPayload | null;
expect(createPermission?.id).toBeTruthy();
await client.callTool({
name: "respond_to_permission",
args: {
agentId,
requestId: createPermission!.id,
response: { behavior: "allow" },
},
});
await waitForAgentCompletion(client, agentId);
const filePath = path.join(agentCwd, "mcp-smoke.txt");
let contents: string;
try {
contents = await waitForFile(filePath);
} catch (error) {
const activityResult = (await client.callTool({
name: "get_agent_activity",
args: { agentId, limit: 25 },
})) as McpToolResult;
const activityPayload = getStructuredContent(activityResult);
const activitySummary = activityPayload?.content;
const details = activitySummary
? `Agent activity:\n${activitySummary}`
: "Agent activity unavailable";
throw new Error(`${(error as Error).message}\n${details}`);
}
expect(contents.trim()).toBe("ok");
const prompt = [
"You must call the tool named shell.",
"Run this command exactly: [\"bash\", \"-lc\", \"echo ok-2 > mcp-smoke-2.txt\"].",
"After the tool runs, reply with done and stop.",
].join("\n");
const promptResult = (await client.callTool({
name: "send_agent_prompt",
args: {
agentId,
prompt,
sessionMode: "default",
background: false,
},
})) as McpToolResult;
const promptPayload = getStructuredContent(promptResult);
const promptPermission = promptPayload?.permission as PermissionPayload | null;
expect(promptPermission?.id).toBeTruthy();
const waitPermissionResult = (await client.callTool({
name: "wait_for_agent",
args: { agentId },
})) as McpToolResult;
const waitPermissionPayload = getStructuredContent(waitPermissionResult);
const waitPermission =
waitPermissionPayload?.permission as PermissionPayload | null;
expect(waitPermission?.id).toBe(promptPermission?.id);
await client.callTool({
name: "respond_to_permission",
args: {
agentId,
requestId: promptPermission!.id,
response: { behavior: "allow" },
},
});
await waitForAgentCompletion(client, agentId);
const secondFilePath = path.join(agentCwd, "mcp-smoke-2.txt");
const secondContents = await waitForFile(secondFilePath);
expect(secondContents.trim()).toBe("ok-2");
} finally {
if (agentId) {
await client.callTool({ name: "kill_agent", args: { agentId } });
}
await client.close();
await daemon.close().catch(() => undefined);
if (previousCodexSessionDir === undefined) {
delete process.env.CODEX_SESSION_DIR;
} else {
process.env.CODEX_SESSION_DIR = previousCodexSessionDir;
}
if (previousCodexHome === undefined) {
delete process.env.CODEX_HOME;
} else {
process.env.CODEX_HOME = previousCodexHome;
}
await rm(paseoHome, { recursive: true, force: true });
await rm(staticDir, { recursive: true, force: true });
await rm(agentCwd, { recursive: true, force: true });
await rm(codexSessionDir, { recursive: true, force: true });
await rm(codexHome, { recursive: true, force: true });
}
expect(contents.trim()).toBe("ok");
} finally {
if (agentId) {
await client.callTool({ name: "kill_agent", args: { agentId } });
}
await client.close();
await daemonHandle.close();
await rm(agentCwd, { recursive: true, force: true });
}
}, 180_000);
},
180_000
);
});

View File

@@ -62,7 +62,8 @@ function expandPath(path: string): string {
async function waitForAgentWithTimeout(
agentManager: AgentManager,
agentId: string,
existingSignal?: AbortSignal
existingSignal?: AbortSignal,
options?: { waitForActive?: boolean }
): Promise<WaitForAgentResult> {
const timeoutSignal = AbortSignal.timeout(AGENT_WAIT_TIMEOUT_MS);
const abortController = new AbortController();
@@ -99,6 +100,7 @@ async function waitForAgentWithTimeout(
try {
const result = await agentManager.waitForAgentEvent(agentId, {
signal: abortController.signal,
waitForActive: options?.waitForActive,
});
return result;
} catch (error) {
@@ -297,7 +299,9 @@ export async function createAgentMcpServer(
// If not running in background, wait for completion
if (!background) {
const result = await waitForAgentWithTimeout(agentManager, snapshot.id);
const result = await waitForAgentWithTimeout(agentManager, snapshot.id, undefined, {
waitForActive: true,
});
const responseData = {
agentId: snapshot.id,
@@ -509,7 +513,9 @@ export async function createAgentMcpServer(
// If not running in background, wait for completion
if (!background) {
const result = await waitForAgentWithTimeout(agentManager, agentId);
const result = await waitForAgentWithTimeout(agentManager, agentId, undefined, {
waitForActive: true,
});
const responseData = {
success: true,