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
This commit is contained in:
Mohamed Boudra
2026-08-03 13:49:53 +02:00
committed by GitHub
parent adf33930b3
commit 5c7623997c
15 changed files with 837 additions and 47 deletions
+5 -2
View File
@@ -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
@@ -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" },
+1
View File
@@ -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(),
});
@@ -441,6 +441,45 @@ class TestAgentSession implements AgentSession {
async close(): Promise<void> {}
}
class McpCapableTestAgentSession extends TestAgentSession {
override readonly capabilities = {
...TEST_CAPABILITIES,
supportsMcpServers: true,
};
}
class CloseRecordingTestAgentSession extends TestAgentSession {
closed = false;
override async close(): Promise<void> {
this.closed = true;
}
}
class McpCapableTestAgentClient extends TestAgentClient {
override readonly capabilities = {
...TEST_CAPABILITIES,
supportsMcpServers: true,
};
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
this.createdConfigs.push(config);
return new McpCapableTestAgentSession(config);
}
override async resumeSession(
_handle: AgentPersistenceHandle,
config?: Partial<AgentSessionConfig>,
): Promise<AgentSession> {
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<AgentSession> {
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<AgentRunResult> {
promptStarted = true;
return super.run();
}
override async startTurn(): Promise<{ turnId: string }> {
promptStarted = true;
return super.startTurn();
}
override async close(): Promise<void> {
sessionClosed = true;
}
}
class UnsupportedMcpClient extends TestAgentClient {
override readonly capabilities = {
...TEST_CAPABILITIES,
supportsMcpServers: true,
};
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
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<AgentSession> {
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<AgentSession> {
return original;
}
override async resumeSession(): Promise<AgentSession> {
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<AgentSession> {
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<AgentSession> {
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<AgentSessionConfig> | 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);
}
}
@@ -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<void> {
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;
@@ -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);
@@ -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 };
}
@@ -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<Partial<AgentSessionConfig> | 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<AgentSessionConfig>,
): Promise<AgentSession> {
this.resumeOverrides.push(overrides);
return this.inner.resumeSession(handle, overrides);
}
fetchCatalog(...args: Parameters<AgentClient["fetchCatalog"]>) {
return this.inner.fetchCatalog(...args);
}
}
class UnsupportedMcpResumeSession extends StubAgentSession {
closed = false;
constructor() {
super({
sessionId: "unsupported-mcp-resume-session",
supportsStreaming: false,
});
}
override async close(): Promise<void> {
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<AgentSession> {
return this.replacement;
}
fetchCatalog(...args: Parameters<AgentClient["fetchCatalog"]>) {
return this.inner.fetchCatalog(...args);
}
async archiveNativeSession(): Promise<void> {
this.nativeArchiveCalls.push("archive");
}
async unarchiveNativeSession(): Promise<void> {
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 {
@@ -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();
@@ -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<string, unknown>;
env?: Record<string, string>;
mcpServers?: Record<string, McpServerConfig>;
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<string, McpServerConfig> | 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 {
@@ -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;
@@ -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<HubRelationshipHarness> {
const harness = new HubRelationshipHarness(archiveWatchFiles);
return this.launch(archiveWatchFiles, false);
}
static async startWithAgentMcp(
archiveWatchFiles: ArchiveWatchFiles = nodeArchiveWatchFiles,
): Promise<HubRelationshipHarness> {
return this.launch(archiveWatchFiles, true);
}
private static async launch(
archiveWatchFiles: ArchiveWatchFiles,
mcpEnabled: boolean,
): Promise<HubRelationshipHarness> {
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: {
+57 -12
View File
@@ -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<void> {
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<StoredAgentRecord | null>((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);
@@ -99,10 +99,14 @@ function buildHarness() {
const projectByWorkspaceId = new Map<string, ProjectPlacementPayload | null>();
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: {} });
@@ -54,6 +54,7 @@ interface Deferred<T> {
interface FakeAgentSessionOptions {
providerName: string;
config: AgentSessionConfig;
supportsMcpServers?: boolean;
sessionId?: string;
memoryMarker?: string | null;
closeSession?: () => Promise<void>;
@@ -63,6 +64,7 @@ interface FakeAgentSessionOptions {
export interface TestAgentClientOptions {
closeSession?: () => Promise<void>;
onStartTurn?: (prompt: AgentPromptInput) => void;
supportsMcpServers?: boolean;
}
function createDeferred<T>(): Deferred<T> {
@@ -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,