From 5c7623997c7c9093b4d0fe324249ab55c4833600 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 3 Aug 2026 13:49:53 +0200 Subject: [PATCH] feat(hub): pass execution MCP servers to agents (#2813) * feat(hub): pass execution MCP servers to agents Hub-owned execution capabilities need to reach provider sessions without becoming part of outward agent snapshots. Keep the configuration daemon-private so provider recovery continues to work. * fix(hub): protect daemon MCP namespace Hub execution MCP configuration must not replace the daemon-owned Paseo server. Validate only after owned-record lookup so idempotent replay remains stable while new launches fail before provider or durable state creation. * fix(agent): reject unsupported MCP launches * fix(agent): scope MCP support guard * fix(agent): redact MCP metadata from wire payloads * fix(agent): validate MCP support on restored sessions * fix(session): rehydrate MCP config on resume * fix(session): resume newest matching agent record * fix(session): restore archive state after failed resume --- docs/hub.md | 7 +- packages/protocol/src/messages.hub.test.ts | 28 +++ packages/protocol/src/messages.ts | 1 + .../src/server/agent/agent-manager.test.ts | 214 +++++++++++++++++- .../server/src/server/agent/agent-manager.ts | 17 ++ .../server/agent/agent-projections.test.ts | 68 +++++- .../src/server/agent/agent-projections.ts | 20 +- .../src/server/daemon-client.e2e.test.ts | 193 ++++++++++++++++ .../src/server/hub/daemon-executions.test.ts | 108 +++++++++ .../src/server/hub/daemon-executions.ts | 10 + .../src/server/hub/execution-controller.ts | 1 + .../hub/test-utils/relationship-harness.ts | 66 ++++-- packages/server/src/server/session.ts | 69 +++++- .../agent-updates-service.test.ts | 57 +++++ .../server/test-utils/fake-agent-client.ts | 25 +- 15 files changed, 837 insertions(+), 47 deletions(-) diff --git a/docs/hub.md b/docs/hub.md index 39dd76368..6f7e6a51c 100644 --- a/docs/hub.md +++ b/docs/hub.md @@ -45,8 +45,11 @@ replays the original prompt. A duplicate create returns the existing agent witho turn. Hub creates use the same agent creation path as trusted clients. They may select any existing -worktree target shape. Execution completion policy remains outside the daemon: a completed agent -turn does not imply that the Hub execution is terminal. +worktree target shape and carry optional MCP server configuration for the agent session. The daemon +keeps that configuration in its private agent record so provider sessions can recover after a +restart; neither ordinary client snapshots and updates nor Hub projections expose session +configuration. Execution completion policy remains outside the daemon: a completed agent turn does +not imply that the Hub execution is terminal. The Hub ends an execution by sending `hub.execution.control.request` with the durable execution ID and either `interrupt` or `archive`. The daemon resolves the agent from the authenticated daemon diff --git a/packages/protocol/src/messages.hub.test.ts b/packages/protocol/src/messages.hub.test.ts index b42ed4524..34387d8ec 100644 --- a/packages/protocol/src/messages.hub.test.ts +++ b/packages/protocol/src/messages.hub.test.ts @@ -80,6 +80,34 @@ describe("Hub session protocol", () => { expect(SessionInboundMessageSchema.parse(message)).toEqual(message); }); + test("accepts an optional MCP server configuration on Hub creates", () => { + const message = { + type: "hub.execution.agent.create.request", + requestId: "request-mcp", + executionId: "execution-mcp", + provider: "codex", + cwd: "/workspace", + prompt: "Implement the requested change", + mcpServers: { + hub: { + type: "http", + url: "https://hub.example/mcp/executions/execution-mcp", + headers: { Authorization: "Bearer hub-execution-bearer" }, + }, + }, + }; + + expect(SessionInboundMessageSchema.parse(message)).toEqual(message); + expect(PreviousHubAgentCreateRequestSchema.parse(message)).toEqual({ + type: "hub.execution.agent.create.request", + requestId: "request-mcp", + executionId: "execution-mcp", + provider: "codex", + cwd: "/workspace", + prompt: "Implement the requested change", + }); + }); + test.each([ undefined, { mode: "branch-off", newBranch: "hub-work", base: "main" }, diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index ed0603cee..be7579257 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -2483,6 +2483,7 @@ export const HubExecutionAgentCreateRequestSchema = z.object({ thinkingOptionId: z.string().optional(), featureValues: z.record(z.string(), z.unknown()).optional(), env: z.record(z.string(), z.string()).optional(), + mcpServers: z.record(z.string(), McpServerConfigSchema).optional(), worktree: CreateAgentWorktreeTargetSchema.optional(), }); diff --git a/packages/server/src/server/agent/agent-manager.test.ts b/packages/server/src/server/agent/agent-manager.test.ts index 50903c5d2..13c58ce0d 100644 --- a/packages/server/src/server/agent/agent-manager.test.ts +++ b/packages/server/src/server/agent/agent-manager.test.ts @@ -441,6 +441,45 @@ class TestAgentSession implements AgentSession { async close(): Promise {} } +class McpCapableTestAgentSession extends TestAgentSession { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: true, + }; +} + +class CloseRecordingTestAgentSession extends TestAgentSession { + closed = false; + + override async close(): Promise { + this.closed = true; + } +} + +class McpCapableTestAgentClient extends TestAgentClient { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: true, + }; + + override async createSession(config: AgentSessionConfig): Promise { + this.createdConfigs.push(config); + return new McpCapableTestAgentSession(config); + } + + override async resumeSession( + _handle: AgentPersistenceHandle, + config?: Partial, + ): Promise { + this.resumeOverrides.push(config); + return new McpCapableTestAgentSession({ + ...config, + provider: this.provider, + cwd: config?.cwd ?? process.cwd(), + }); + } +} + class ControlledInterruptSession extends TestAgentSession { interruptCalled = false; @@ -1846,7 +1885,7 @@ test("createAgent injects paseo MCP server only into provider launch config", as override async createSession(config: AgentSessionConfig): Promise { this.lastConfig = config; - return new TestAgentSession(config); + return new McpCapableTestAgentSession(config); } } @@ -1902,6 +1941,164 @@ test("createAgent injects paseo MCP server only into provider launch config", as }); }); +test("createAgent closes and rejects a provider session that cannot honor MCP servers", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + let sessionClosed = false; + let promptStarted = false; + + class UnsupportedMcpSession extends TestAgentSession { + override async run(): Promise { + promptStarted = true; + return super.run(); + } + + override async startTurn(): Promise<{ turnId: string }> { + promptStarted = true; + return super.startTurn(); + } + + override async close(): Promise { + sessionClosed = true; + } + } + + class UnsupportedMcpClient extends TestAgentClient { + override readonly capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: true, + }; + + override async createSession(config: AgentSessionConfig): Promise { + return new UnsupportedMcpSession(config); + } + } + + const agentId = "00000000-0000-4000-8000-000000000107"; + const manager = new AgentManager({ + clients: { codex: new UnsupportedMcpClient() }, + registry: storage, + logger, + idFactory: () => agentId, + }); + + try { + await expect( + manager.createAgent( + { + provider: "codex", + cwd: workdir, + mcpServers: { + hub: { + type: "http", + url: "http://127.0.0.1:3000/api/executions/test/mcp", + }, + }, + }, + undefined, + { workspaceId: undefined }, + ), + ).rejects.toThrow("Provider 'codex' does not support MCP servers"); + + expect(sessionClosed).toBe(true); + expect(promptStarted).toBe(false); + expect(manager.getAgent(agentId)).toBeNull(); + expect(await storage.get(agentId)).toBeNull(); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("resumeAgentFromPersistence closes and rejects a session that cannot honor external MCP", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); + const storage = new AgentStorage(join(workdir, "agents"), logger); + const replacement = new CloseRecordingTestAgentSession({ provider: "codex", cwd: workdir }); + + class UnsupportedResumeClient extends TestAgentClient { + override async resumeSession(): Promise { + return replacement; + } + } + + const agentId = "00000000-0000-4000-8000-000000000108"; + const manager = new AgentManager({ + clients: { codex: new UnsupportedResumeClient() }, + registry: storage, + logger, + }); + const handle: AgentPersistenceHandle = { + provider: "codex", + sessionId: "persist-unsupported-mcp", + metadata: { cwd: workdir }, + }; + + try { + await expect( + manager.resumeAgentFromPersistence( + handle, + { + cwd: workdir, + mcpServers: { + hub: { type: "http", url: "https://hub.test/mcp/executions/resume" }, + }, + }, + agentId, + ), + ).rejects.toThrow("Provider 'codex' does not support MCP servers"); + + expect(replacement.closed).toBe(true); + expect(manager.getAgent(agentId)).toBeNull(); + expect(await storage.get(agentId)).toBeNull(); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } +}); + +test("reloadAgentSession preserves the live session when its replacement cannot honor external MCP", async () => { + const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); + const original = new CloseRecordingTestAgentSession({ provider: "codex", cwd: workdir }); + const replacement = new CloseRecordingTestAgentSession({ provider: "codex", cwd: workdir }); + + class UnsupportedReloadClient extends TestAgentClient { + override async createSession(): Promise { + return original; + } + + override async resumeSession(): Promise { + return replacement; + } + } + + const manager = new AgentManager({ + clients: { codex: new UnsupportedReloadClient() }, + registry: new AgentStorage(join(workdir, "agents"), logger), + logger, + }); + + try { + const created = await manager.createAgent( + { provider: "codex", cwd: workdir }, + "00000000-0000-4000-8000-000000000109", + { workspaceId: undefined }, + ); + + await expect( + manager.reloadAgentSession(created.id, { + mcpServers: { + hub: { type: "http", url: "https://hub.test/mcp/executions/reload" }, + }, + }), + ).rejects.toThrow("Provider 'codex' does not support MCP servers"); + + expect(replacement.closed).toBe(true); + expect(original.closed).toBe(false); + expect(manager.getAgent(created.id)?.session).toBe(original); + expect(manager.getAgent(created.id)?.lifecycle).toBe("idle"); + } finally { + rmSync(workdir, { recursive: true, force: true }); + } +}); + test("createAgent passes native Paseo tools through launch context without internal MCP", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); @@ -1930,7 +2127,7 @@ test("createAgent passes native Paseo tools through launch context without inter ): Promise { this.lastConfig = config; this.lastLaunchContext = launchContext; - return new TestAgentSession(config); + return new McpCapableTestAgentSession(config); } } @@ -1984,7 +2181,7 @@ test("createAgent passes native Paseo tools through launch context without inter }); }); -test("createAgent injects the MCP auth token as a bearer header into the launch config", async () => { +test("createAgent allows best-effort internal MCP when the provider session reports no support", async () => { const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); @@ -2033,7 +2230,7 @@ test("resumeAgentFromPersistence replaces stored internal paseo MCP with current const workdir = mkdtempSync(join(tmpdir(), "agent-manager-test-")); const storagePath = join(workdir, "agents"); const storage = new AgentStorage(storagePath, logger); - const client = new TestAgentClient(); + const client = new McpCapableTestAgentClient(); const manager = new AgentManager({ clients: { codex: client, @@ -2127,7 +2324,7 @@ test("createAgent preserves a user-provided paseo MCP config", async () => { override async createSession(config: AgentSessionConfig): Promise { this.lastConfig = config; - return new TestAgentSession(config); + return new McpCapableTestAgentSession(config); } } @@ -2455,7 +2652,10 @@ test("resumeAgentFromPersistence keeps metadata config, applies overrides, and p class ResumeCaptureClient implements AgentClient { readonly provider = "codex" as const; - readonly capabilities = TEST_CAPABILITIES; + readonly capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: true, + }; lastResumeOverrides: Partial | undefined; lastResumeLaunchContext: AgentLaunchContext | undefined; @@ -2495,7 +2695,7 @@ test("resumeAgentFromPersistence keeps metadata config, applies overrides, and p provider: "codex", cwd: overrides?.cwd ?? metadata.cwd ?? process.cwd(), }; - return new TestAgentSession(merged); + return new McpCapableTestAgentSession(merged); } } diff --git a/packages/server/src/server/agent/agent-manager.ts b/packages/server/src/server/agent/agent-manager.ts index 773bffb84..30bd6a4a1 100644 --- a/packages/server/src/server/agent/agent-manager.ts +++ b/packages/server/src/server/agent/agent-manager.ts @@ -1089,6 +1089,7 @@ export class AgentManager { const providerLaunchConfig = this.resolveProviderLaunchConfig(launchConfig, launchContext); const createOptions = this.buildCreateSessionOptions(options); const session = await client.createSession(providerLaunchConfig, launchContext, createOptions); + await this.requireExternalMcpSupport(session, storedConfig); return this.registerSession(session, storedConfig, resolvedAgentId, { labels: options.labels, initialTitle: options.initialTitle, @@ -1171,6 +1172,7 @@ export class AgentManager { launchContext, resumeOptions, ); + await this.requireExternalMcpSupport(session, storedConfig); return this.registerSession(session, storedConfig, resolvedAgentId, { ...options, persistence: handle, @@ -1297,6 +1299,7 @@ export class AgentManager { const session = handle ? await client.resumeSession(handle, providerLaunchConfig, launchContext) : await client.createSession(providerLaunchConfig, launchContext); + await this.requireExternalMcpSupport(session, storedConfig); let handedToRegistration = false; try { @@ -2895,6 +2898,20 @@ export class AgentManager { } } + private async requireExternalMcpSupport( + session: AgentSession, + storedConfig: AgentSessionConfig, + ): Promise { + if ( + Object.keys(storedConfig.mcpServers ?? {}).length === 0 || + session.capabilities.supportsMcpServers === true + ) { + return; + } + await this.closeUnregisteredSession(session); + throw new Error(`Provider '${storedConfig.provider}' does not support MCP servers`); + } + private async initializeAgentTimelineForRegister(params: { agentId: string; now: Date; diff --git a/packages/server/src/server/agent/agent-projections.test.ts b/packages/server/src/server/agent/agent-projections.test.ts index ffa1b5026..b29e3d4ec 100644 --- a/packages/server/src/server/agent/agent-projections.test.ts +++ b/packages/server/src/server/agent/agent-projections.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from "vitest"; import { AGENT_LIFECYCLE_STATUSES } from "./agent-manager.js"; import { + buildStoredAgentPayload, toAgentPayload, toRecentProviderSessionDescriptorPayload, toStoredAgentRecord, @@ -328,7 +329,16 @@ describe("toAgentPayload", () => { provider: "codex", sessionId: "persist-99", nativeHandle: { id: "native" } as unknown, - metadata: { restored: new Date("2025-03-01T00:00:00.000Z"), empty: {} }, + metadata: { + restored: new Date("2025-03-01T00:00:00.000Z"), + empty: {}, + mcpServers: { + hub: { + type: "http", + headers: { Authorization: "Bearer projection-secret" }, + }, + }, + }, }, }); const payload = toAgentPayload(agent); @@ -342,6 +352,62 @@ describe("toAgentPayload", () => { expect(agent.persistence!.sessionId).toBe("persist-99"); }); + it("removes empty persistence metadata after projecting MCP configuration", () => { + const payload = toAgentPayload( + createManagedAgent({ + provider: "codex", + config: { provider: "codex" }, + persistence: { + provider: "codex", + sessionId: "persist-mcp-only", + metadata: { mcpServers: { hub: { type: "http", url: "https://hub.test/mcp" } } }, + }, + }), + ); + + expect(payload.persistence).toEqual({ + provider: "codex", + sessionId: "persist-mcp-only", + }); + }); + + it("strips MCP metadata from stored wire payloads while preserving private persistence", () => { + const record = toStoredAgentRecord( + createManagedAgent({ + provider: "codex", + config: { provider: "codex" }, + persistence: { + provider: "codex", + sessionId: "persist-stored", + metadata: { + conversationId: "conversation-stored", + mcpServers: { + hub: { + type: "http", + headers: { Authorization: "Bearer stored-projection-secret" }, + }, + }, + }, + }, + }), + ); + + const payload = buildStoredAgentPayload(record, ["codex"]); + + expect(record.persistence?.metadata).toEqual({ + conversationId: "conversation-stored", + mcpServers: { + hub: { + type: "http", + headers: { Authorization: "Bearer stored-projection-secret" }, + }, + }, + }); + expect(payload.persistence?.metadata).toEqual({ + conversationId: "conversation-stored", + }); + }); + it("omits lastUsage when not available", () => { const agent = createManagedAgent({ lastUsage: undefined }); const payload = toAgentPayload(agent); diff --git a/packages/server/src/server/agent/agent-projections.ts b/packages/server/src/server/agent/agent-projections.ts index d0175b7f7..26cd35679 100644 --- a/packages/server/src/server/agent/agent-projections.ts +++ b/packages/server/src/server/agent/agent-projections.ts @@ -132,7 +132,7 @@ export function toAgentPayload( availableModes: cloneAvailableModes(agent.availableModes), features: normalizeFeatures(agent.features), pendingPermissions: sanitizePendingPermissions(agent.pendingPermissions), - persistence: sanitizePersistenceHandle(agent.persistence), + persistence: projectPersistenceHandleForWire(agent.persistence), title: options?.title ?? null, labels: agent.labels, }; @@ -213,7 +213,9 @@ export function buildStoredAgentPayload( const runtimeInfo = buildStoredRuntimeInfo(record); const providerAvailable = isStoredAgentProviderAvailable(record, validProviders); - const persistence = buildStoredPersistenceHandle(record, validProviders); + const persistence = projectPersistenceHandleForWire( + buildStoredPersistenceHandle(record, validProviders), + ); return { id: record.id, @@ -365,6 +367,20 @@ function sanitizePersistenceHandle( return sanitized; } +function projectPersistenceHandleForWire( + handle: AgentPersistenceHandle | null, +): AgentPersistenceHandle | null { + const projected = sanitizePersistenceHandle(handle); + if (!projected?.metadata) { + return projected; + } + delete projected.metadata.mcpServers; + if (Object.keys(projected.metadata).length === 0) { + delete projected.metadata; + } + return projected; +} + function cloneCapabilities(capabilities: AgentCapabilityFlags): AgentCapabilityFlags { return { ...capabilities }; } diff --git a/packages/server/src/server/daemon-client.e2e.test.ts b/packages/server/src/server/daemon-client.e2e.test.ts index f05c49f4d..d36589915 100644 --- a/packages/server/src/server/daemon-client.e2e.test.ts +++ b/packages/server/src/server/daemon-client.e2e.test.ts @@ -13,6 +13,7 @@ import { DaemonClient, } from "./test-utils/index.js"; import { createTestPaseoDaemon } from "./test-utils/paseo-daemon.js"; +import { createTestAgentClients } from "./test-utils/fake-agent-client.js"; import { getFullAccessConfig, getAskModeConfig } from "./daemon-e2e/agent-configs.js"; import { parsePcm16MonoWav, wordSimilarity } from "./test-utils/dictation-e2e.js"; import type { @@ -315,6 +316,88 @@ class StubAgentClient implements AgentClient { } } +class RecordingMcpResumeClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities; + readonly resumeOverrides: Array | undefined> = []; + private readonly inner = createTestAgentClients({ supportsMcpServers: true }).codex; + + constructor() { + this.capabilities = this.inner.capabilities; + } + + isAvailable() { + return this.inner.isAvailable(); + } + + createSession(config: AgentSessionConfig) { + return this.inner.createSession(config); + } + + resumeSession( + handle: AgentPersistenceHandle, + overrides?: Partial, + ): Promise { + this.resumeOverrides.push(overrides); + return this.inner.resumeSession(handle, overrides); + } + + fetchCatalog(...args: Parameters) { + return this.inner.fetchCatalog(...args); + } +} + +class UnsupportedMcpResumeSession extends StubAgentSession { + closed = false; + + constructor() { + super({ + sessionId: "unsupported-mcp-resume-session", + supportsStreaming: false, + }); + } + + override async close(): Promise { + this.closed = true; + } +} + +class RejectingMcpResumeClient implements AgentClient { + readonly provider = "codex" as const; + readonly capabilities; + readonly nativeArchiveCalls: string[] = []; + readonly replacement = new UnsupportedMcpResumeSession(); + private readonly inner = createTestAgentClients({ supportsMcpServers: true }).codex; + + constructor() { + this.capabilities = this.inner.capabilities; + } + + isAvailable() { + return this.inner.isAvailable(); + } + + createSession(config: AgentSessionConfig) { + return this.inner.createSession(config); + } + + async resumeSession(): Promise { + return this.replacement; + } + + fetchCatalog(...args: Parameters) { + return this.inner.fetchCatalog(...args); + } + + async archiveNativeSession(): Promise { + this.nativeArchiveCalls.push("archive"); + } + + async unarchiveNativeSession(): Promise { + this.nativeArchiveCalls.push("unarchive"); + } +} + test("createAgent fails when the initial turn cannot start", async () => { const testAgent = new StubAgentClient({ sessionId: "start-turn-failure-session", @@ -860,6 +943,116 @@ test("resume_agent auto-unarchives archived agents", async () => { } }, 180000); +test("resume_agent rehydrates the newest private MCP config from a redacted persistence handle", async () => { + const cwd = tmpCwd(); + const provider = new RecordingMcpResumeClient(); + const localCtx = await createDaemonTestContext({ + agentClients: { codex: provider }, + }); + const bearerA = "durable-resume-bearer-a"; + const bearerB = "durable-resume-bearer-b"; + const externalMcpA = { + hub: { + type: "http" as const, + url: "https://hub.test/mcp/executions/durable-resume", + headers: { Authorization: `Bearer ${bearerA}` }, + }, + }; + const externalMcpB = { + hub: { + ...externalMcpA.hub, + headers: { Authorization: `Bearer ${bearerB}` }, + }, + }; + + try { + const created = await localCtx.client.createAgent({ + config: { + provider: "codex", + cwd, + mcpServers: externalMcpA, + }, + }); + const projectedHandle = created.persistence; + if (!projectedHandle) { + throw new Error("Expected a projected persistence handle"); + } + expect(projectedHandle.metadata).not.toHaveProperty("mcpServers"); + expect(JSON.stringify(projectedHandle)).not.toContain(bearerA); + + await localCtx.client.archiveAgent(created.id); + const resumedWithOverride = await localCtx.client.resumeAgent(projectedHandle, { + mcpServers: externalMcpB, + }); + const newestProjectedHandle = resumedWithOverride.persistence; + if (!newestProjectedHandle) { + throw new Error("Expected a projected persistence handle after resume"); + } + expect(provider.resumeOverrides[0]?.mcpServers?.hub).toEqual(externalMcpB.hub); + expect(JSON.stringify(resumedWithOverride)).not.toContain(bearerA); + expect(JSON.stringify(resumedWithOverride)).not.toContain(bearerB); + + await localCtx.client.archiveAgent(resumedWithOverride.id); + const resumedNewest = await localCtx.client.resumeAgent(newestProjectedHandle); + + expect(provider.resumeOverrides).toHaveLength(2); + expect(provider.resumeOverrides[1]?.mcpServers?.hub).toEqual(externalMcpB.hub); + expect(provider.resumeOverrides[1]?.mcpServers?.hub).not.toEqual(externalMcpA.hub); + expect(resumedNewest.persistence?.metadata).not.toHaveProperty("mcpServers"); + expect(JSON.stringify(resumedNewest)).not.toContain(bearerA); + expect(JSON.stringify(resumedNewest)).not.toContain(bearerB); + } finally { + await localCtx.cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } +}); + +test("resume_agent restores archive state when an MCP-capable provider returns an unsupported session", async () => { + const cwd = tmpCwd(); + const provider = new RejectingMcpResumeClient(); + const localCtx = await createDaemonTestContext({ + agentClients: { codex: provider }, + }); + + try { + const created = await localCtx.client.createAgent({ + config: { + provider: "codex", + cwd, + mcpServers: { + hub: { + type: "http", + url: "https://hub.test/mcp/executions/rejected-resume", + headers: { Authorization: "Bearer rejected-resume-secret" }, + }, + }, + }, + }); + const handle = created.persistence; + if (!handle) { + throw new Error("Expected a projected persistence handle"); + } + const archived = await localCtx.client.archiveAgent(created.id); + + await expect(localCtx.client.resumeAgent(handle)).rejects.toMatchObject({ + name: "DaemonRpcError", + code: "agent_resume_failed", + requestType: "resume_agent_request", + }); + + const archivedAgents = await localCtx.client.fetchAgents({ + filter: { includeArchived: true }, + }); + const original = archivedAgents.entries.find((entry) => entry.agent.id === created.id)?.agent; + expect(original?.archivedAt).toBe(archived.archivedAt); + expect(provider.nativeArchiveCalls).toEqual(["archive", "unarchive", "archive"]); + expect(provider.replacement.closed).toBe(true); + } finally { + await localCtx.cleanup(); + rmSync(cwd, { recursive: true, force: true }); + } +}); + test("update_agent persists unloaded title and labels across auto-unarchive", async () => { const cwd = tmpCwd(); try { diff --git a/packages/server/src/server/hub/daemon-executions.test.ts b/packages/server/src/server/hub/daemon-executions.test.ts index 0b52c89c8..0ea73e834 100644 --- a/packages/server/src/server/hub/daemon-executions.test.ts +++ b/packages/server/src/server/hub/daemon-executions.test.ts @@ -27,6 +27,114 @@ test("sequential replay after reconstruction keeps one durable owned agent", asy expect(reconstructed.durableAgentCount).toBe(1); }); +test("Hub MCP configuration reaches the provider alongside Paseo MCP without entering snapshots", async () => { + const hub = await HubRelationshipHarness.startWithAgentMcp(); + await hub.beginConnect().result; + hub.connectLatestSocket(); + relationship = hub; + const bearer = "hub-execution-bearer"; + hub.beginOwnedCreate("mcp-create", "mcp-execution", { + mcpServers: { + hub: { + type: "http", + url: "https://hub.test/mcp/executions/mcp-execution", + headers: { Authorization: `Bearer ${bearer}` }, + }, + }, + }); + + const response = await hub.ownedCreateResult("mcp-create"); + + expect(hub.latestProviderCreateConfig()?.mcpServers).toMatchObject({ + paseo: { type: "http" }, + hub: { + type: "http", + url: "https://hub.test/mcp/executions/mcp-execution", + headers: { Authorization: `Bearer ${bearer}` }, + }, + }); + expect(response).toMatchObject({ + type: "hub.execution.agent.create.response", + payload: { success: true, agent: { provider: "codex" } }, + }); + expect(response.payload.agent).not.toHaveProperty("config"); + expect(response.payload.agent).not.toHaveProperty("mcpServers"); + expect(response.payload.agent.persistence?.metadata).toEqual({ + conversationId: response.payload.agent.persistence?.sessionId, + cwd: response.payload.agent.cwd, + }); + expect(JSON.stringify(response.payload.agent)).not.toContain(bearer); + + const update = hub.hubMessages().find((message) => message.type === "hub.execution.agent.update"); + expect(update).toMatchObject({ + type: "hub.execution.agent.update", + payload: { agent: { provider: "codex" } }, + }); + if (!update || update.type !== "hub.execution.agent.update") { + throw new Error("Expected a Hub execution agent update"); + } + expect(update.payload.agent).not.toHaveProperty("config"); + expect(update.payload.agent).not.toHaveProperty("mcpServers"); + expect(update.payload.agent.persistence?.metadata).toEqual({ + conversationId: update.payload.agent.persistence?.sessionId, + cwd: update.payload.agent.cwd, + }); + expect(JSON.stringify(update.payload.agent)).not.toContain(bearer); +}); + +test("new Hub executions cannot override the daemon-owned Paseo MCP server", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("reserved-mcp-create", "reserved-mcp-execution", { + mcpServers: { + paseo: { type: "http", url: "https://hub.test/replace-paseo" }, + }, + }); + + const response = await hub.ownedCreateResult("reserved-mcp-create"); + + expect(response).toMatchObject({ + type: "hub.execution.agent.create.response", + payload: { + success: false, + executionId: "reserved-mcp-execution", + agentId: null, + agent: null, + }, + }); + expect(hub.providerCreations()).toBe(0); + expect(hub.activeOwnedAgentIds()).toEqual([]); + expect(await hub.durableOwnedAgentIds()).toEqual([]); +}); + +test("reserved Paseo MCP input does not invalidate replay of an owned execution", async () => { + const hub = await launchRelationship(); + hub.beginOwnedCreate("original-create", "replayed-execution"); + const original = await hub.ownedCreateResult("original-create"); + expect(original).toMatchObject({ + type: "hub.execution.agent.create.response", + payload: { success: true, executionId: "replayed-execution" }, + }); + const providerCreations = hub.providerCreations(); + + hub.beginOwnedCreate("replay-create", "replayed-execution", { + mcpServers: { + paseo: { type: "http", url: "https://hub.test/replace-paseo" }, + }, + }); + const replay = await hub.ownedCreateResult("replay-create"); + + expect(replay).toMatchObject({ + type: "hub.execution.agent.create.response", + payload: { + success: true, + executionId: "replayed-execution", + agentId: original.payload.agentId, + }, + }); + expect(hub.providerCreations()).toBe(providerCreations); + expect(await hub.durableOwnedAgentIds()).toEqual([original.payload.agentId]); +}); + test("removing a daemon-owned agent removes its execution association", async () => { const hub = await launchRelationship(); const created = await hub.createOwnedConcurrently(); diff --git a/packages/server/src/server/hub/daemon-executions.ts b/packages/server/src/server/hub/daemon-executions.ts index 5ee0f3ca7..e8b267dfb 100644 --- a/packages/server/src/server/hub/daemon-executions.ts +++ b/packages/server/src/server/hub/daemon-executions.ts @@ -6,6 +6,7 @@ import type { } from "@getpaseo/protocol/messages"; import type { AgentManager, AgentManagerEvent, ManagedAgent } from "../agent/agent-manager.js"; +import type { McpServerConfig } from "../agent/agent-sdk-types.js"; import type { AgentStorage, StoredAgentRecord } from "../agent/agent-storage.js"; import type { BoundCreateAgentCommand } from "../agent/create-agent/create.js"; import type { CreatePaseoWorktreeWorkflowResult } from "../worktree-session.js"; @@ -25,6 +26,7 @@ export interface HubExecutionAgentCreateInput { thinkingOptionId?: string; featureValues?: Record; env?: Record; + mcpServers?: Record; worktree?: CreateAgentWorktreeTarget; } @@ -175,6 +177,7 @@ export class DaemonExecutions implements HubExecutionAgents { return this.resolveRecord(existing); } this.requireAuthority(authorityGeneration); + requireHubMcpNamespace(input.mcpServers); let createdWorktree: CreatePaseoWorktreeWorkflowResult | null = null; let createdAgentId: string | null = null; @@ -192,6 +195,7 @@ export class DaemonExecutions implements HubExecutionAgents { thinking: input.thinkingOptionId, features: input.featureValues, env: input.env, + ...(input.mcpServers ? { config: { mcpServers: input.mcpServers } } : {}), worktree: toCreateAgentWorktree(input.worktree), background: true, notifyOnFinish: false, @@ -348,6 +352,12 @@ export class DaemonExecutions implements HubExecutionAgents { } } +function requireHubMcpNamespace(mcpServers: Record | undefined): void { + if (mcpServers && Object.hasOwn(mcpServers, "paseo")) { + throw new Error('Hub execution MCP server name "paseo" is reserved by the daemon'); + } +} + function ownedCreatedWorktree( worktree: CreatePaseoWorktreeWorkflowResult | null, ): CreatePaseoWorktreeWorkflowResult | null { diff --git a/packages/server/src/server/hub/execution-controller.ts b/packages/server/src/server/hub/execution-controller.ts index 0c1b40e5d..67905303c 100644 --- a/packages/server/src/server/hub/execution-controller.ts +++ b/packages/server/src/server/hub/execution-controller.ts @@ -102,6 +102,7 @@ export class HubExecutionController { thinkingOptionId: message.thinkingOptionId, featureValues: message.featureValues, env: message.env, + mcpServers: message.mcpServers, worktree: message.worktree, }); if (this.closed) return; diff --git a/packages/server/src/server/hub/test-utils/relationship-harness.ts b/packages/server/src/server/hub/test-utils/relationship-harness.ts index 28da3eb9c..ab0b9c3a9 100644 --- a/packages/server/src/server/hub/test-utils/relationship-harness.ts +++ b/packages/server/src/server/hub/test-utils/relationship-harness.ts @@ -463,29 +463,48 @@ export class HubRelationshipHarness { private failNextSessionClose = false; private observedEnrollments = 0; private observedSockets = 0; - private readonly codex = new ControlledAgentClient( - createTestAgentClients({ - closeSession: async () => { - if (!this.failNextSessionClose) return; - this.failNextSessionClose = false; - throw new Error("Requested provider session close failure"); - }, - onStartTurn: (prompt) => { - const promptText = typeof prompt === "string" ? prompt : JSON.stringify(prompt); - if (this.promptsToFail.delete(promptText)) { - throw new Error("Requested provider prompt startup failure"); - } - this.providerPrompts.push(prompt); - }, - }).codex, - ); + private readonly codex: ControlledAgentClient; - private constructor(private readonly archiveWatchFiles: ArchiveWatchFiles) {} + private constructor( + private readonly archiveWatchFiles: ArchiveWatchFiles, + private readonly mcpEnabled: boolean, + ) { + this.codex = new ControlledAgentClient( + createTestAgentClients({ + supportsMcpServers: mcpEnabled, + closeSession: async () => { + if (!this.failNextSessionClose) return; + this.failNextSessionClose = false; + throw new Error("Requested provider session close failure"); + }, + onStartTurn: (prompt) => { + const promptText = typeof prompt === "string" ? prompt : JSON.stringify(prompt); + if (this.promptsToFail.delete(promptText)) { + throw new Error("Requested provider prompt startup failure"); + } + this.providerPrompts.push(prompt); + }, + }).codex, + ); + } static async start( archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles, ): Promise { - const harness = new HubRelationshipHarness(archiveWatchFiles); + return this.launch(archiveWatchFiles, false); + } + + static async startWithAgentMcp( + archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles, + ): Promise { + return this.launch(archiveWatchFiles, true); + } + + private static async launch( + archiveWatchFiles: ArchiveWatchFiles, + mcpEnabled: boolean, + ): Promise { + const harness = new HubRelationshipHarness(archiveWatchFiles, mcpEnabled); await harness.createHome(); await harness.startDaemon(); return harness; @@ -675,6 +694,7 @@ export class HubRelationshipHarness { worktree?: CreateAgentWorktreeTarget; prompt?: string; modeId?: string; + mcpServers?: AgentSessionConfig["mcpServers"]; } = {}, ): void { const { prompt = "Create through the Hub", ...requestOptions } = options; @@ -826,6 +846,14 @@ export class HubRelationshipHarness { ); } + latestProviderCreateConfig(): AgentSessionConfig | null { + return this.codex.createdConfigs.at(-1) ?? null; + } + + hubMessages(): SessionOutboundMessage[] { + return this.remote.sockets.flatMap(({ socket }) => socket.sent); + } + latestCreatedCwd(): string | null { return this.codex.createdConfigs.at(-1)?.cwd ?? null; } @@ -1285,7 +1313,7 @@ export class HubRelationshipHarness { paseoHome: this.paseoHome, corsAllowedOrigins: [], hostnames: true, - mcpEnabled: false, + mcpEnabled: this.mcpEnabled, staticDir, mcpDebug: false, agentClients: { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 609d718c8..6d95d32a1 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -33,7 +33,11 @@ import { CursorError } from "./pagination/cursor.js"; import { SortablePager, type SortSpec } from "./pagination/sortable-pager.js"; import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js"; import type { TurnDetectionProvider } from "./speech/turn-detection-provider.js"; -import { isStoredAgentProviderAvailable, toAgentPersistenceHandle } from "./persistence-hooks.js"; +import { + buildConfigOverrides, + isStoredAgentProviderAvailable, + toAgentPersistenceHandle, +} from "./persistence-hooks.js"; import { ensureAgentLoaded, ensureUnarchivedAgentLoaded } from "./agent/agent-loading.js"; import { formatSystemNotificationPrompt, @@ -87,7 +91,11 @@ import { setAgentModeCommand, updateAgentCommand, } from "./agent/lifecycle-command.js"; -import { buildStoredAgentPayload, toAgentPayload } from "./agent/agent-projections.js"; +import { + buildStoredAgentPayload, + resolveStoredAgentPayloadUpdatedAt, + toAgentPayload, +} from "./agent/agent-projections.js"; import { appendTimelineItemIfAgentKnown, emitLiveTimelineItemIfAgentKnown, @@ -2535,17 +2543,43 @@ export class Session { }); } - private async unarchiveAgentByHandle(handle: AgentPersistenceHandle): Promise { + private async unarchiveAgentByHandle(handle: AgentPersistenceHandle): Promise<{ + record: StoredAgentRecord; + didUnarchive: boolean; + originalArchivedAt: string | null; + } | null> { const records = await this.agentStorage.list(); - const matched = records.find( - (record) => - record.persistence?.provider === handle.provider && - record.persistence?.sessionId === handle.sessionId, - ); + const matched = records + .filter( + (record) => + record.persistence?.provider === handle.provider && + record.persistence?.sessionId === handle.sessionId, + ) + .reduce((latest, candidate) => { + if (!latest) { + return candidate; + } + const updatedDelta = + Date.parse(resolveStoredAgentPayloadUpdatedAt(candidate)) - + Date.parse(resolveStoredAgentPayloadUpdatedAt(latest)); + if (updatedDelta !== 0) { + return updatedDelta > 0 ? candidate : latest; + } + return Date.parse(candidate.createdAt) > Date.parse(latest.createdAt) ? candidate : latest; + }, null); if (!matched) { - return; + return null; } - await unarchiveAgentState(this.agentStorage, this.agentManager, matched.id); + const didUnarchive = await unarchiveAgentState( + this.agentStorage, + this.agentManager, + matched.id, + ); + return { + record: matched, + didUnarchive, + originalArchivedAt: matched.archivedAt ?? null, + }; } private async handleUpdateAgentRequest( @@ -3304,8 +3338,19 @@ export class Session { `Resuming agent ${handle.sessionId} (${handle.provider})`, ); try { - await this.unarchiveAgentByHandle(handle); - const snapshot = await this.agentManager.resumeAgentFromPersistence(handle, overrides); + const matched = await this.unarchiveAgentByHandle(handle); + const effectiveOverrides = matched + ? { ...buildConfigOverrides(matched.record), ...overrides } + : overrides; + let snapshot: ManagedAgent; + try { + snapshot = await this.agentManager.resumeAgentFromPersistence(handle, effectiveOverrides); + } catch (error) { + if (matched?.didUnarchive && matched.originalArchivedAt) { + await this.agentManager.archiveSnapshot(matched.record.id, matched.originalArchivedAt); + } + throw error; + } await unarchiveAgentState(this.agentStorage, this.agentManager, snapshot.id); await this.agentManager.hydrateTimelineFromProvider(snapshot.id); await this.agentUpdates.forwardLiveAgent(snapshot); diff --git a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts index 01aba3f1f..29da3106f 100644 --- a/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts +++ b/packages/server/src/server/session/agent-updates/agent-updates-service.test.ts @@ -99,10 +99,14 @@ function buildHarness() { const projectByWorkspaceId = new Map(); let providerVisible: (provider: string) => boolean = () => true; let buildAgentPayloadError: Error | null = null; + let enrichProjectedPayload = false; const service = createAgentUpdatesService({ emit: (message) => emitted.push(message), enrichAgentPayload: async (payload) => { + if (enrichProjectedPayload) { + return payload; + } const queuedPayload = queuedPayloadBuilds.shift(); if (queuedPayload) { return queuedPayload; @@ -152,6 +156,9 @@ function buildHarness() { setProviderVisible(fn: (provider: string) => boolean) { providerVisible = fn; }, + useProjectedPayload() { + enrichProjectedPayload = true; + }, failBuildAgentPayload(error: Error) { buildAgentPayloadError = error; }, @@ -329,6 +336,56 @@ describe("forwardLiveAgent", () => { expect(h.workspaceUpdates).toEqual(["ws-1"]); }); + test("never projects MCP credentials into ordinary client updates", async () => { + const h = buildHarness(); + h.service.beginSubscription({ subscriptionId: "sub", filter: {} }); + h.service.flushBootstrapped("sub"); + h.register(makeAgentPayload({ id: "a", workspaceId: "ws-1" })); + h.useProjectedPayload(); + const agent = h.managed("a"); + agent.config.mcpServers = { + hub: { + type: "http", + url: "https://hub.test/mcp/executions/execution-1", + headers: { Authorization: "Bearer execution-secret" }, + }, + }; + agent.persistence = { + provider: "codex", + sessionId: "session-a", + metadata: { + conversationId: "conversation-a", + mcpServers: agent.config.mcpServers, + }, + }; + expect(agent.config).toMatchObject({ + provider: "codex", + mcpServers: { + hub: { + type: "http", + url: "https://hub.test/mcp/executions/execution-1", + headers: { Authorization: "Bearer execution-secret" }, + }, + }, + }); + + await h.service.forwardLiveAgent(agent); + + const update = h.agentUpdates()[0]; + expect(update).toEqual({ + kind: "upsert", + agent: expect.objectContaining({ id: "a" }), + project: makeProject(), + }); + if (update?.kind !== "upsert") { + throw new Error("Expected an agent upsert"); + } + expect(update.agent).not.toHaveProperty("config"); + expect(update.agent).not.toHaveProperty("mcpServers"); + expect(update.agent.persistence?.metadata).toEqual({ conversationId: "conversation-a" }); + expect(JSON.stringify(update.agent)).not.toContain("execution-secret"); + }); + test("emits a remove when the agent's workspace resolves to no project", async () => { const h = buildHarness(); h.service.beginSubscription({ subscriptionId: "sub", filter: {} }); diff --git a/packages/server/src/server/test-utils/fake-agent-client.ts b/packages/server/src/server/test-utils/fake-agent-client.ts index e025634b3..ea2589c0b 100644 --- a/packages/server/src/server/test-utils/fake-agent-client.ts +++ b/packages/server/src/server/test-utils/fake-agent-client.ts @@ -54,6 +54,7 @@ interface Deferred { interface FakeAgentSessionOptions { providerName: string; config: AgentSessionConfig; + supportsMcpServers?: boolean; sessionId?: string; memoryMarker?: string | null; closeSession?: () => Promise; @@ -63,6 +64,7 @@ interface FakeAgentSessionOptions { export interface TestAgentClientOptions { closeSession?: () => Promise; onStartTurn?: (prompt: AgentPromptInput) => void; + supportsMcpServers?: boolean; } function createDeferred(): Deferred { @@ -317,7 +319,7 @@ function buildLargeTimelineItem(input: { } class FakeAgentSession implements AgentSession { - readonly capabilities = TEST_CAPABILITIES; + readonly capabilities: AgentCapabilityFlags; readonly id: string; private readonly providerName: string; private readonly config: AgentSessionConfig; @@ -334,6 +336,10 @@ class FakeAgentSession implements AgentSession { private readonly onStartTurn: ((prompt: AgentPromptInput) => void) | undefined; constructor(options: FakeAgentSessionOptions) { + this.capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: options.supportsMcpServers === true, + }; this.providerName = options.providerName; this.config = options.config; this.id = options.sessionId ?? randomUUID(); @@ -851,10 +857,14 @@ class FakeAgentSession implements AgentSession { } describePersistence(): AgentPersistenceHandle | null { + const metadata = { + ...(this.memoryMarker ? { marker: this.memoryMarker } : {}), + ...(this.config.mcpServers ? { mcpServers: this.config.mcpServers } : {}), + }; return buildPersistence( this.providerName, this.id, - this.memoryMarker ? { marker: this.memoryMarker } : undefined, + Object.keys(metadata).length > 0 ? metadata : undefined, ); } @@ -1171,11 +1181,16 @@ class FakeAgentSession implements AgentSession { } class FakeAgentClient implements AgentClient { - readonly capabilities = TEST_CAPABILITIES; + readonly capabilities: AgentCapabilityFlags; constructor( public readonly provider: string, private readonly options: TestAgentClientOptions, - ) {} + ) { + this.capabilities = { + ...TEST_CAPABILITIES, + supportsMcpServers: options.supportsMcpServers === true, + }; + } async createSession( config: AgentSessionConfig, @@ -1184,6 +1199,7 @@ class FakeAgentClient implements AgentClient { return new FakeAgentSession({ providerName: this.provider, config: { ...config }, + supportsMcpServers: this.options.supportsMcpServers, closeSession: this.options.closeSession, onStartTurn: this.options.onStartTurn, }); @@ -1206,6 +1222,7 @@ class FakeAgentClient implements AgentClient { return new FakeAgentSession({ providerName: this.provider, config: cfg, + supportsMcpServers: this.options.supportsMcpServers, sessionId: handle.sessionId, memoryMarker: typeof marker === "string" ? marker : null, closeSession: this.options.closeSession,