From 3c1026d451e8d7f3eadd63046c5682242607d8d8 Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Mon, 27 Jul 2026 16:45:06 +0300 Subject: [PATCH 1/6] fix(auth): forward the OAuth timeout through listTools and the daemon `mcporter auth` runs the interactive OAuth browser flow inside the SDK's `tools/list` request, but `listTools` never forwarded a per-request timeout. The SDK's 60s DEFAULT_REQUEST_TIMEOUT_MSEC killed the request long before the 300s OAuth code wait could finish, so slow providers could never be authorized -- every `mcporter auth` died with `MCP error -32001: Request timed out`. Mirror the existing `callTool` timeout forwarding: - add `timeoutMs` to `ListToolsOptions` and pass `{ timeout, resetTimeoutOnProgress, maxTotalTimeout }` to `client.listTools`, with `raceWithTimeout` as an outer guard; - have the `auth` path pass `MCPORTER_OAUTH_TIMEOUT_MS` (default 300s) so the request lives at least as long as the OAuth code wait; - carry `timeoutMs` across the daemon protocol so keep-alive servers get the same deadline, and report a phase that blows it as `operation_timeout` so the keep-alive wrapper stops treating it as a dead server worth restarting; - version the daemon protocol so a daemon started before this change is replaced instead of silently dropping the new field. --- src/cli.ts | 5 +- src/cli/auth-command.ts | 14 ++++- src/daemon/client.ts | 7 ++- src/daemon/host.ts | 58 ++++++++++++++++---- src/daemon/protocol.ts | 5 ++ src/daemon/runtime-wrapper.ts | 5 ++ src/runtime.ts | 21 ++++++- tests/cli-auth.test.ts | 18 ++++-- tests/cli-oauth-timeout-flag.test.ts | 41 ++++++++++++++ tests/daemon-client-config-stale.test.ts | 44 +++++++++++++++ tests/daemon-client-lifecycle.test.ts | 4 ++ tests/daemon-client-timeout.test.ts | 44 ++++++++++++++- tests/daemon-client.test.ts | 3 + tests/daemon-host.test.ts | 31 ++++++++++- tests/keep-alive-runtime.test.ts | 25 ++++++++- tests/runtime-listtools-timeout.test.ts | 70 ++++++++++++++++++++++++ 16 files changed, 366 insertions(+), 29 deletions(-) create mode 100644 tests/runtime-listtools-timeout.test.ts diff --git a/src/cli.ts b/src/cli.ts index b9765a02..1c0039e3 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -325,7 +325,7 @@ export async function runCli(argv: string[]): Promise { return; } const { handleAuth: importedHandleAuth } = await import('./cli/auth-command.js'); - await importedHandleAuth(runtime, resolvedArgs); + await importedHandleAuth(runtime, resolvedArgs, { oauthTimeoutMs: runtimeOptionsWithPath.oauthTimeoutMs }); return; } @@ -441,7 +441,7 @@ async function invokeAuthCommand(runtimeOptions: RuntimeOptions, args: string[]) ]); const runtime = await createRuntime(runtimeOptions); try { - await importedHandleAuth(runtime, args); + await importedHandleAuth(runtime, args, { oauthTimeoutMs: runtimeOptions.oauthTimeoutMs }); } finally { await runtime.close().catch(() => {}); } @@ -630,6 +630,7 @@ function createDaemonOnlyRuntime(daemonClient: import('./daemon/client.js').Daem autoAuthorize: options?.autoAuthorize, allowCachedAuth: options?.allowCachedAuth, disableOAuth: options?.disableOAuth, + timeoutMs: options?.timeoutMs, })) as Awaited>, callTool: (server, toolName, options) => daemonClient.callTool({ diff --git a/src/cli/auth-command.ts b/src/cli/auth-command.ts index c44503b4..67bcf903 100644 --- a/src/cli/auth-command.ts +++ b/src/cli/auth-command.ts @@ -4,7 +4,7 @@ import type { OAuthAuthorizationRequest, OAuthSessionOptions } from '../oauth.js import { analyzeConnectionError } from '../error-classifier.js'; import { clearOAuthCaches } from '../oauth-persistence.js'; import type { createRuntime } from '../runtime.js'; -import { isOAuthFlowError } from '../runtime/oauth.js'; +import { isOAuthFlowError, resolveOAuthTimeoutFromEnv } from '../runtime/oauth.js'; import type { EphemeralServerSpec } from './adhoc-server.js'; import { extractEphemeralServerFlags } from './ephemeral-flags.js'; import { persistPreparedEphemeralServer, prepareEphemeralServerTarget } from './ephemeral-target.js'; @@ -17,12 +17,17 @@ type Runtime = Awaited>; type BrowserSuppression = 'default' | 'no-browser'; +export interface AuthCommandOptions { + readonly oauthTimeoutMs?: number; +} + const TRUE_VALUES = new Set(['1', 'true', 'yes']); const FALSE_VALUES = new Set(['0', 'false', 'no']); -export async function handleAuth(runtime: Runtime, args: string[]): Promise { +export async function handleAuth(runtime: Runtime, args: string[], options: AuthCommandOptions = {}): Promise { const browserSuppression = consumeBrowserSuppression(args, process.env); const noBrowser = browserSuppression === 'no-browser'; + const oauthTimeoutMs = options.oauthTimeoutMs ?? resolveOAuthTimeoutFromEnv(); let authorizationOutputEmitted = false; const markAuthorizationOutputEmitted = () => { authorizationOutputEmitted = true; @@ -85,6 +90,11 @@ export async function handleAuth(runtime: Runtime, args: string[]): Promise runtime.listTools(target, { autoAuthorize: true, + // The interactive OAuth browser flow runs inside this listTools + // request. Let it ride for as long as we're willing to wait for the + // authorization code (MCPORTER_OAUTH_TIMEOUT_MS, default 300s) instead + // of being killed by the SDK's 60s request cap. + timeoutMs: oauthTimeoutMs, ...(noBrowser ? { oauthSessionOptions: buildNoBrowserOAuthOptions(format, markAuthorizationOutputEmitted), diff --git a/src/daemon/client.ts b/src/daemon/client.ts index 642e5d05..168c6a40 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -5,6 +5,7 @@ import path from 'node:path'; import { listConfigLayerPaths } from '../config/path-discovery.js'; import { withFileLock } from '../fs-json.js'; import { getDaemonMetadataPath, getDaemonSocketPath } from './paths.js'; +import { DAEMON_PROTOCOL_VERSION } from './protocol.js'; import type { CallToolParams, CloseServerParams, @@ -34,6 +35,7 @@ export interface DaemonPaths { interface DaemonMetadata { readonly pid: number; + readonly protocolVersion?: number; readonly socketPath: string; readonly configPath: string; readonly configMtimeMs?: number | null; @@ -69,7 +71,7 @@ export class DaemonClient { } async listTools(params: ListToolsParams): Promise { - return this.invoke('listTools', params); + return this.invoke('listTools', params, params.timeoutMs); } async listResources(params: ListResourcesParams): Promise { @@ -232,6 +234,9 @@ export class DaemonClient { if (!metadata) { return 'missing'; } + if (metadata.protocolVersion !== DAEMON_PROTOCOL_VERSION) { + return 'stale'; + } const currentLayers = normalizeLayers(await collectConfigLayers(this.options)); const metadataLayers = normalizeLayers( metadata.configLayers ?? [{ path: metadata.configPath, mtimeMs: metadata.configMtimeMs ?? null }] diff --git a/src/daemon/host.ts b/src/daemon/host.ts index 57c6e03c..1cfc4f6a 100644 --- a/src/daemon/host.ts +++ b/src/daemon/host.ts @@ -1,3 +1,4 @@ +import { ErrorCode } from '@modelcontextprotocol/sdk/types.js'; import { randomUUID } from 'node:crypto'; import fs from 'node:fs/promises'; import net from 'node:net'; @@ -6,6 +7,7 @@ import { loadDaemonConfig, type ServerDefinition } from '../config.js'; import { readJsonFile, withFileLock, writeJsonFile } from '../fs-json.js'; import { isKeepAliveServer } from '../lifecycle.js'; import { createRuntime, type Runtime } from '../runtime.js'; +import { OAuthTimeoutError } from '../runtime/oauth.js'; import { collectConfigLayers, statConfigMtime } from './config-layers.js'; import { hashDaemonDefinitions } from './definition-hash.js'; import { @@ -16,15 +18,17 @@ import { logEvent, shouldLogServer, } from './log-context.js'; -import type { - CallToolParams, - CloseServerParams, - DaemonRequest, - DaemonResponse, - ListResourcesParams, - ListToolsParams, - ReadResourceParams, - StatusResult, +import { + DAEMON_OPERATION_TIMEOUT_CODE, + DAEMON_PROTOCOL_VERSION, + type CallToolParams, + type CloseServerParams, + type DaemonRequest, + type DaemonResponse, + type ListResourcesParams, + type ListToolsParams, + type ReadResourceParams, + type StatusResult, } from './protocol.js'; import { buildErrorResponse, @@ -163,6 +167,7 @@ export async function runDaemonHost(options: DaemonHostOptions): Promise { configPath: options.configPath, configLayers, socketPath: options.socketPath, + protocolVersion: DAEMON_PROTOCOL_VERSION, startedAt, logPath: options.logPath ?? null, configMtimeMs, @@ -213,6 +218,7 @@ export async function runDaemonHost(options: DaemonHostOptions): Promise { }); await writeJsonFile(options.metadataPath, { pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath: options.socketPath, configPath: options.configPath, configLayers, @@ -268,6 +274,7 @@ function metadataFromStatus( fallbackConfigLayers: Array<{ path: string; mtimeMs: number | null }> ): { pid: number; + protocolVersion: number; socketPath: string; configPath: string; configLayers?: StatusResult['configLayers']; @@ -278,6 +285,7 @@ function metadataFromStatus( } { return { pid: status.pid, + protocolVersion: status.protocolVersion, socketPath: status.socketPath, configPath: status.configPath, configLayers: status.configLayers && status.configLayers.length > 0 ? status.configLayers : fallbackConfigLayers, @@ -295,6 +303,9 @@ function daemonConfigMatches( currentConfigMtimeMs: number | null, currentDefinitionHash: string ): boolean { + if (live.protocolVersion !== DAEMON_PROTOCOL_VERSION) { + return false; + } if (live.definitionHash !== currentDefinitionHash) { return false; } @@ -483,6 +494,7 @@ async function handleSocketRequest( configLayers: Array<{ path: string; mtimeMs: number | null }>; configMtimeMs: number | null; socketPath: string; + protocolVersion: number; startedAt: number; logPath: string | null; definitionHash?: string; @@ -516,6 +528,18 @@ function normalizeDaemonDisableOAuth(value: boolean | undefined): boolean { return value === true; } +function daemonRuntimeErrorCode(error: unknown): string { + const errorCode = error && typeof error === 'object' ? (error as { code?: unknown }).code : undefined; + if ( + error instanceof OAuthTimeoutError || + errorCode === ErrorCode.RequestTimeout || + (error instanceof Error && error.message === 'Timeout') + ) { + return DAEMON_OPERATION_TIMEOUT_CODE; + } + return 'runtime_error'; +} + async function processRequest( rawPayload: string, runtime: Runtime, @@ -526,6 +550,7 @@ async function processRequest( configLayers: Array<{ path: string; mtimeMs: number | null }>; configMtimeMs: number | null; socketPath: string; + protocolVersion: number; startedAt: number; logPath: string | null; definitionHash?: string; @@ -596,6 +621,7 @@ async function processRequest( autoAuthorize: resolveDaemonListToolsAutoAuthorize(params, definition), allowCachedAuth: params.allowCachedAuth ?? true, disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), + timeoutMs: params.timeoutMs, }); markActivity(params.server, activity); if (loggable) { @@ -689,6 +715,7 @@ async function processRequest( case 'status': { const result: StatusResult = { pid: process.pid, + protocolVersion: metadata.protocolVersion, startedAt: metadata.startedAt, configPath: metadata.configPath, configLayers: metadata.configLayers, @@ -722,7 +749,7 @@ async function processRequest( } } catch (error) { return { - response: buildErrorResponse(id, 'runtime_error', error), + response: buildErrorResponse(id, daemonRuntimeErrorCode(error), error), shouldShutdown: false, }; } @@ -748,6 +775,7 @@ export async function __testProcessRequest( configLayers: Array<{ path: string; mtimeMs: number | null }>; configMtimeMs: number | null; socketPath: string; + protocolVersion?: number; startedAt: number; logPath: string | null; definitionHash?: string; @@ -755,5 +783,13 @@ export async function __testProcessRequest( logContext: LogContext, preParsedRequest?: DaemonRequest ): Promise<{ response: DaemonResponse; shouldShutdown: boolean }> { - return await processRequest(rawPayload, runtime, managedServers, activity, metadata, logContext, preParsedRequest); + return await processRequest( + rawPayload, + runtime, + managedServers, + activity, + { ...metadata, protocolVersion: metadata.protocolVersion ?? DAEMON_PROTOCOL_VERSION }, + logContext, + preParsedRequest + ); } diff --git a/src/daemon/protocol.ts b/src/daemon/protocol.ts index 925fc46a..cbb48275 100644 --- a/src/daemon/protocol.ts +++ b/src/daemon/protocol.ts @@ -1,3 +1,6 @@ +export const DAEMON_PROTOCOL_VERSION = 2; +export const DAEMON_OPERATION_TIMEOUT_CODE = 'operation_timeout'; + export type DaemonRequestMethod = | 'callTool' | 'listTools' @@ -37,6 +40,7 @@ export interface ListToolsParams { readonly autoAuthorize?: boolean; readonly allowCachedAuth?: boolean; readonly disableOAuth?: boolean; + readonly timeoutMs?: number; } export interface ListResourcesParams { @@ -59,6 +63,7 @@ export interface CloseServerParams { export interface StatusResult { readonly pid: number; + readonly protocolVersion: number; readonly startedAt: number; readonly configPath: string; readonly configMtimeMs?: number | null; diff --git a/src/daemon/runtime-wrapper.ts b/src/daemon/runtime-wrapper.ts index a84a8514..a1f5240b 100644 --- a/src/daemon/runtime-wrapper.ts +++ b/src/daemon/runtime-wrapper.ts @@ -10,6 +10,7 @@ import type { Runtime, } from '../runtime.js'; import type { DaemonClient } from './client.js'; +import { DAEMON_OPERATION_TIMEOUT_CODE } from './protocol.js'; interface KeepAliveRuntimeOptions { readonly daemonClient: DaemonClient | null; @@ -69,6 +70,7 @@ class KeepAliveRuntime implements Runtime { autoAuthorize: options?.autoAuthorize, allowCachedAuth: options?.allowCachedAuth ?? true, disableOAuth: options?.disableOAuth, + timeoutMs: options?.timeoutMs, }) )) as Awaited>; } @@ -178,6 +180,9 @@ function shouldRestartDaemonServer(error: unknown): boolean { if (!error) { return false; } + if (typeof error === 'object' && (error as { code?: unknown }).code === DAEMON_OPERATION_TIMEOUT_CODE) { + return false; + } if (error instanceof McpError) { return !NON_FATAL_CODES.has(error.code); } diff --git a/src/runtime.ts b/src/runtime.ts index eb482e3d..32974cd7 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -61,6 +61,13 @@ export interface ListToolsOptions { * headless callers that need cached-token-only behavior. */ readonly disableOAuth?: boolean; + /** + * Per-request timeout (ms) forwarded to the MCP client so listings don't hit + * the SDK's default 60s cap. Required for `auth`, where the interactive OAuth + * browser flow runs inside the `tools/list` request and routinely exceeds 60s. + * Mirrors {@link CallOptions.timeoutMs}. + */ + readonly timeoutMs?: number; } export type ListResourcesOptions = Partial & { @@ -77,6 +84,7 @@ export interface ReadResourceOptions { export interface ConnectOptions { readonly maxOAuthAttempts?: number; + readonly oauthTimeoutMs?: number; readonly skipCache?: boolean; readonly allowCachedAuth?: boolean; readonly oauthSessionOptions?: OAuthSessionOptions; @@ -253,19 +261,28 @@ class McpRuntime implements Runtime { true ); const useLegacyNoAuthorize = !autoAuthorize && disableOAuth !== true; + const timeoutMs = normalizeTimeout(options.timeoutMs); const context = await this.connect(server, { maxOAuthAttempts: useLegacyNoAuthorize ? 0 : undefined, skipCache: useLegacyNoAuthorize, allowCachedAuth, oauthSessionOptions: options.oauthSessionOptions, disableOAuth, + oauthTimeoutMs: timeoutMs, }); let closeError: unknown; const tools: ServerToolInfo[] = []; try { let cursor: string | undefined; do { - const response = await context.client.listTools(cursor ? { cursor } : undefined); + // Forward the requested timeout to the MCP client so listings -- and the + // interactive OAuth flow that listTools can trigger during `auth` -- don't + // hit the SDK's default 60s cap. Mirrors the callTool timeout forwarding. + const listPromise = context.client.listTools( + cursor ? { cursor } : undefined, + timeoutMs ? { timeout: timeoutMs, resetTimeoutOnProgress: true, maxTotalTimeout: timeoutMs } : undefined + ); + const response = timeoutMs ? await raceWithTimeout(listPromise, timeoutMs) : await listPromise; tools.push( ...(response.tools ?? []).map((tool) => ({ name: tool.name, @@ -580,7 +597,7 @@ class McpRuntime implements Runtime { let connectionDefinition = definition; let contextPromise = createClientContext(definition, this.logger, this.clientInfo, { maxOAuthAttempts: options.maxOAuthAttempts, - oauthTimeoutMs: this.oauthTimeoutMs ?? OAUTH_CODE_TIMEOUT_MS, + oauthTimeoutMs: options.oauthTimeoutMs ?? this.oauthTimeoutMs ?? OAUTH_CODE_TIMEOUT_MS, onDefinitionPromoted: (promoted) => { if ( this.serverGeneration(normalized) === generation && diff --git a/tests/cli-auth.test.ts b/tests/cli-auth.test.ts index a74ebab7..e8f9074c 100644 --- a/tests/cli-auth.test.ts +++ b/tests/cli-auth.test.ts @@ -32,18 +32,24 @@ describe('mcporter auth ad-hoc support', () => { const { handleAuth } = await cliModulePromise; const { runtime, listTools } = createRuntimeDouble(); - await handleAuth(runtime, ['--http-url', 'https://mcp.deepwiki.com/sse']); + await handleAuth(runtime, ['--http-url', 'https://mcp.deepwiki.com/sse'], { oauthTimeoutMs: 300_000 }); - expect(listTools).toHaveBeenCalledWith('mcp-deepwiki-com-sse', { autoAuthorize: true }); + expect(listTools).toHaveBeenCalledWith('mcp-deepwiki-com-sse', { + autoAuthorize: true, + timeoutMs: 300_000, + }); }); it('accepts bare URLs as the auth target', async () => { const { handleAuth } = await cliModulePromise; const { runtime, listTools } = createRuntimeDouble(); - await handleAuth(runtime, ['https://mcp.supabase.com/mcp']); + await handleAuth(runtime, ['https://mcp.supabase.com/mcp'], { oauthTimeoutMs: 300_000 }); - expect(listTools).toHaveBeenCalledWith('mcp-supabase-com-mcp', { autoAuthorize: true }); + expect(listTools).toHaveBeenCalledWith('mcp-supabase-com-mcp', { + autoAuthorize: true, + timeoutMs: 300_000, + }); }); it('reuses configured servers when auth target is a URL', async () => { @@ -62,9 +68,9 @@ describe('mcporter auth ad-hoc support', () => { getDefinition: () => definition, } as unknown as Awaited>; - await handleAuth(runtime, ['https://mcp.vercel.com']); + await handleAuth(runtime, ['https://mcp.vercel.com'], { oauthTimeoutMs: 300_000 }); - expect(listTools).toHaveBeenCalledWith('vercel', { autoAuthorize: true }); + expect(listTools).toHaveBeenCalledWith('vercel', { autoAuthorize: true, timeoutMs: 300_000 }); expect(registerDefinition).not.toHaveBeenCalled(); }); diff --git a/tests/cli-oauth-timeout-flag.test.ts b/tests/cli-oauth-timeout-flag.test.ts index 634bbb2a..f95067ab 100644 --- a/tests/cli-oauth-timeout-flag.test.ts +++ b/tests/cli-oauth-timeout-flag.test.ts @@ -52,6 +52,47 @@ describe('mcporter --oauth-timeout flag', () => { createRuntimeSpy.mockRestore(); }); + it('uses the override for the auth listTools request timeout', async () => { + const definition = { + name: 'fake', + description: 'Fake HTTP server', + command: { kind: 'http' as const, url: new URL('https://example.com/mcp') }, + }; + const listToolsSpy = vi.fn(async () => []); + const runtimeStub: Runtime = { + listServers: vi.fn(() => [definition.name]), + getDefinitions: vi.fn(() => [definition]), + getDefinition: vi.fn(() => definition), + registerDefinition: vi.fn(), + listTools: listToolsSpy, + callTool: vi.fn(), + listResources: vi.fn(), + readResource: vi.fn(), + connect: vi.fn(), + close: vi.fn(async () => {}), + }; + const runtimeModule = await import('../src/runtime.js'); + const createRuntimeSpy = vi.spyOn(runtimeModule, 'createRuntime').mockResolvedValue(runtimeStub); + const { runCli } = await import('../src/cli.js'); + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}); + const previousNoForce = process.env.MCPORTER_NO_FORCE_EXIT; + process.env.MCPORTER_NO_FORCE_EXIT = '1'; + + try { + await runCli(['--oauth-timeout', '600000', 'auth', 'fake']); + } finally { + process.env.MCPORTER_NO_FORCE_EXIT = previousNoForce; + } + + expect(listToolsSpy).toHaveBeenCalledWith( + 'fake', + expect.objectContaining({ autoAuthorize: true, timeoutMs: 600_000 }) + ); + + logSpy.mockRestore(); + createRuntimeSpy.mockRestore(); + }); + it('rejects malformed --oauth-timeout values', async () => { const { runCli } = await import('../src/cli.js'); const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}); diff --git a/tests/daemon-client-config-stale.test.ts b/tests/daemon-client-config-stale.test.ts index 0dc06d00..b471ae41 100644 --- a/tests/daemon-client-config-stale.test.ts +++ b/tests/daemon-client-config-stale.test.ts @@ -2,6 +2,7 @@ import { EventEmitter } from 'node:events'; import fs from 'node:fs/promises'; import path from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { DAEMON_PROTOCOL_VERSION } from '../src/daemon/protocol.js'; import { makeShortTempDir } from './fixtures/test-helpers.js'; const sentMethods: string[] = []; @@ -16,6 +17,9 @@ class MockSocket extends EventEmitter { write(data: string, cb?: (err?: Error | null) => void): boolean { const payload = JSON.parse(data.toString()); sentMethods.push(payload.method); + if (payload.method === 'stop') { + activeStatusPid = findNonRunningPid(); + } const response = buildResponse(payload.method, payload.id); queueMicrotask(() => { this.emit('data', JSON.stringify(response)); @@ -41,6 +45,7 @@ function buildResponse(method: string, id: string) { ok: true, result: { pid: activeStatusPid, + protocolVersion: DAEMON_PROTOCOL_VERSION, startedAt: Date.now(), configPath: activeConfigPath, configMtimeMs: activeConfigMtime, @@ -93,6 +98,7 @@ describe('DaemonClient config freshness', () => { JSON.stringify( { pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath: options.socketPath, configPath: options.configPath, startedAt: Date.now(), @@ -225,6 +231,7 @@ describe('DaemonClient config freshness', () => { JSON.stringify( { pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath, configPath, startedAt: Date.now() - 10_000, @@ -245,6 +252,43 @@ describe('DaemonClient config freshness', () => { expect(sentMethods).not.toContain('stop'); expect(launchDaemonDetached).not.toHaveBeenCalled(); }); + + it('restarts a daemon whose metadata predates the current protocol', async () => { + const tmpDir = await makeShortTempDir('daemon-protocol-stale'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + + const configPath = path.join(tmpDir, 'config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const stat = await fs.stat(configPath); + const { metadataPath, socketPath } = resolveDaemonPaths(configPath); + activeConfigPath = configPath; + activeSocketPath = socketPath; + activeConfigMtime = stat.mtimeMs; + activeStatusPid = process.pid; + activeLayers = [{ path: configPath, mtimeMs: stat.mtimeMs }]; + + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: process.pid, + socketPath, + configPath, + startedAt: Date.now() - 10_000, + logPath: null, + configMtimeMs: stat.mtimeMs, + configLayers: activeLayers, + }), + 'utf8' + ); + + const client = new DaemonClient({ configPath, configExplicit: true, rootDir: tmpDir }); + await client.listTools({ server: 'playwright' }); + + expect(sentMethods).toContain('stop'); + expect(launchDaemonDetached).toHaveBeenCalledTimes(1); + expect(sentMethods).toContain('listTools'); + }); }); function findNonRunningPid(): number { diff --git a/tests/daemon-client-lifecycle.test.ts b/tests/daemon-client-lifecycle.test.ts index 81cf2806..f1049d79 100644 --- a/tests/daemon-client-lifecycle.test.ts +++ b/tests/daemon-client-lifecycle.test.ts @@ -2,6 +2,7 @@ import fs from 'node:fs/promises'; import net from 'node:net'; import path from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { DAEMON_PROTOCOL_VERSION } from '../src/daemon/protocol.js'; import { makeShortTempDir } from './fixtures/test-helpers.js'; const launchDaemonDetached = vi.hoisted(() => vi.fn()); @@ -68,6 +69,7 @@ describe('DaemonClient lifecycle reconciliation', () => { metadataPath, JSON.stringify({ pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath, configPath, configLayers: [{ path: configPath, mtimeMs: (await fs.stat(configPath)).mtimeMs }], @@ -147,6 +149,7 @@ async function startMockDaemon( options.metadataPath, JSON.stringify({ pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath: options.socketPath, configPath: options.configPath, configLayers: [{ path: options.configPath, mtimeMs: stat.mtimeMs }], @@ -189,6 +192,7 @@ async function startStatusServer( request.method === 'status' ? { pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, startedAt: Date.now(), configPath, socketPath, diff --git a/tests/daemon-client-timeout.test.ts b/tests/daemon-client-timeout.test.ts index f34d6a09..850757c7 100644 --- a/tests/daemon-client-timeout.test.ts +++ b/tests/daemon-client-timeout.test.ts @@ -2,15 +2,20 @@ import { EventEmitter } from 'node:events'; import fs from 'node:fs/promises'; import path from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { DAEMON_PROTOCOL_VERSION } from '../src/daemon/protocol.js'; import { makeShortTempDir } from './fixtures/test-helpers.js'; const timeoutRecords: Array<{ method: string; timeout: number }> = []; class MockSocket extends EventEmitter { currentTimeout = 0; + private timeoutHandle?: NodeJS.Timeout; - setTimeout(ms: number): this { + setTimeout(ms: number, callback?: () => void): this { this.currentTimeout = ms; + if (enforceSocketTimeout && callback) { + this.timeoutHandle = setTimeout(callback, ms); + } return this; } @@ -19,6 +24,9 @@ class MockSocket extends EventEmitter { timeoutRecords.push({ method: payload.method, timeout: this.currentTimeout }); const response = buildResponse(payload.method, payload.id); setTimeout(() => { + if (this.timeoutHandle) { + clearTimeout(this.timeoutHandle); + } this.emit('data', JSON.stringify(response)); this.emit('end'); }, responseDelayMs); @@ -31,12 +39,19 @@ class MockSocket extends EventEmitter { return this; } - destroy(): this { + destroy(error?: Error): this { + if (this.timeoutHandle) { + clearTimeout(this.timeoutHandle); + } + if (error) { + queueMicrotask(() => this.emit('error', error)); + } return this; } } let responseDelayMs = 5; +let enforceSocketTimeout = false; let activeConfigPath = path.resolve('mcporter.config.json'); let activeSocketPath = ''; const createConnection = vi.fn(() => { @@ -67,6 +82,7 @@ function buildResponse(method: string, id: string) { ok: true, result: { pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, startedAt: Date.now(), configPath: activeConfigPath, socketPath: activeSocketPath, @@ -85,6 +101,7 @@ describe('DaemonClient timeouts', () => { beforeEach(async () => { timeoutRecords.length = 0; responseDelayMs = 5; + enforceSocketTimeout = false; previousDaemonTimeout = process.env.MCPORTER_DAEMON_TIMEOUT_MS; previousDaemonDir = process.env.MCPORTER_DAEMON_DIR; tmpDaemonDir = await makeShortTempDir('daemon-timeout'); @@ -142,6 +159,28 @@ describe('DaemonClient timeouts', () => { expect(callRecord?.timeout).toBe(12_345); }); + it('honors per-listTools timeout overrides', async () => { + const configPath = 'mcporter.config.json'; + await writeFreshMetadata(configPath); + const client = new DaemonClient({ configPath, configExplicit: true }); + await client.listTools({ server: 'foo', timeoutMs: 300_000 }); + const statusRecord = timeoutRecords.find((entry) => entry.method === 'status'); + const listRecord = timeoutRecords.find((entry) => entry.method === 'listTools'); + // Both the status probe and the listTools socket use the caller deadline + // verbatim. + expect(statusRecord?.timeout).toBe(300_000); + expect(listRecord?.timeout).toBe(300_000); + }); + + it('leaves callTool operation socket budget unchanged', async () => { + const configPath = 'mcporter.config.json'; + await writeFreshMetadata(configPath); + const client = new DaemonClient({ configPath, configExplicit: true }); + await client.callTool({ server: 'foo', tool: 'bar', timeoutMs: 12_345 }); + const callRecord = timeoutRecords.find((entry) => entry.method === 'callTool'); + expect(callRecord?.timeout).toBe(12_345); + }); + it('clamps daemon status preflight timeout for tiny per-call timeouts', async () => { const configPath = 'mcporter.config.json'; await writeFreshMetadata(configPath); @@ -163,6 +202,7 @@ async function writeFreshMetadata(configPath: string): Promise { paths.metadataPath, JSON.stringify({ pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath: paths.socketPath, configPath, configLayers: [{ path: activeConfigPath, mtimeMs: null }], diff --git a/tests/daemon-client.test.ts b/tests/daemon-client.test.ts index 2c5fc77c..85fc4c5a 100644 --- a/tests/daemon-client.test.ts +++ b/tests/daemon-client.test.ts @@ -4,6 +4,7 @@ import net from 'node:net'; import path from 'node:path'; import { describe, expect, it, vi } from 'vitest'; import { DaemonClient, resolveDaemonPaths } from '../src/daemon/client.js'; +import { DAEMON_PROTOCOL_VERSION } from '../src/daemon/protocol.js'; import { makeShortTempDir } from './fixtures/test-helpers.js'; describe('daemon client', () => { @@ -107,6 +108,7 @@ describe('daemon client', () => { metadataPath, JSON.stringify({ pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, socketPath, configPath, configLayers: [{ path: configPath, mtimeMs: configStats.mtimeMs }], @@ -125,6 +127,7 @@ describe('daemon client', () => { request.method === 'status' ? { pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, startedAt: Date.now(), configPath, configLayers: [{ path: configPath, mtimeMs: configStats.mtimeMs }], diff --git a/tests/daemon-host.test.ts b/tests/daemon-host.test.ts index 5e3a100b..af31f402 100644 --- a/tests/daemon-host.test.ts +++ b/tests/daemon-host.test.ts @@ -13,6 +13,7 @@ import { } from '../src/daemon/host.js'; import type { DaemonRequest } from '../src/daemon/protocol.js'; import type { Runtime } from '../src/runtime.js'; +import { OAuthTimeoutError } from '../src/runtime/oauth.js'; describe('daemon host request handling', () => { const metadata = { @@ -60,7 +61,7 @@ describe('daemon host request handling', () => { await __testProcessRequest('', runtime as unknown as Runtime, managedServers, new Map(), metadata, logContext, { id: 'list', method: 'listTools', - params: { server: 'oauth', includeSchema: true }, + params: { server: 'oauth', includeSchema: true, timeoutMs: 300_000 }, }); expect(runtime.listTools).toHaveBeenCalledWith('oauth', { @@ -68,6 +69,7 @@ describe('daemon host request handling', () => { autoAuthorize: undefined, allowCachedAuth: true, disableOAuth: false, + timeoutMs: 300_000, }); }); @@ -154,6 +156,28 @@ describe('daemon host request handling', () => { disableOAuth: false, }); }); + + it('marks OAuth listTools timeouts as non-retryable operation errors', async () => { + const runtime = createRuntimeDouble(); + vi.mocked(runtime.listTools).mockRejectedValue(new OAuthTimeoutError('oauth', 5_000)); + + const result = await __testProcessRequest( + '', + runtime as unknown as Runtime, + createManagedServers(), + new Map(), + metadata, + logContext, + { + id: 'list-timeout', + method: 'listTools', + params: { server: 'oauth', timeoutMs: 5_000 }, + } + ); + + expect(result.response.ok).toBe(false); + expect(result.response.error?.code).toBe('operation_timeout'); + }); }); const describeUnixSocket = process.platform === 'win32' ? describe.skip : describe; @@ -290,7 +314,10 @@ describe('daemon artifact cleanup', () => { }); }); -function createRuntimeDouble(): Pick { +function createRuntimeDouble(): { + callTool: ReturnType; + listTools: ReturnType; +} { return { callTool: vi.fn().mockResolvedValue({ ok: true }), listTools: vi.fn().mockResolvedValue([]), diff --git a/tests/keep-alive-runtime.test.ts b/tests/keep-alive-runtime.test.ts index d922c061..79b7389e 100644 --- a/tests/keep-alive-runtime.test.ts +++ b/tests/keep-alive-runtime.test.ts @@ -106,13 +106,14 @@ describe('createKeepAliveRuntime', () => { disableOAuth: undefined, }); - await keepAliveRuntime.listTools('alpha', { includeSchema: true }); + await keepAliveRuntime.listTools('alpha', { includeSchema: true, timeoutMs: 5_000 }); expect(daemon.listTools).toHaveBeenCalledWith({ server: 'alpha', includeSchema: true, autoAuthorize: undefined, allowCachedAuth: true, disableOAuth: undefined, + timeoutMs: 5_000, }); await keepAliveRuntime.listTools('alpha', { allowCachedAuth: false }); @@ -282,4 +283,26 @@ describe('createKeepAliveRuntime', () => { expect(daemon.callTool).toHaveBeenCalledTimes(1); expect(daemon.closeServer).not.toHaveBeenCalled(); }); + + it('does not restart or retry daemon listTools operation timeouts', async () => { + const runtime = new FakeRuntime(definitions); + const timeoutError = Object.assign(new Error("OAuth authorization for 'alpha' timed out after 5s; aborting."), { + code: 'operation_timeout', + }); + const daemon = { + callTool: vi.fn(), + closeServer: vi.fn().mockResolvedValue(undefined), + listTools: vi.fn().mockRejectedValue(timeoutError), + listResources: vi.fn(), + readResource: vi.fn(), + }; + const keepAliveRuntime = createKeepAliveRuntime(runtime as unknown as Runtime, { + daemonClient: daemon as never, + keepAliveServers: new Set(['alpha']), + }); + + await expect(keepAliveRuntime.listTools('alpha', { timeoutMs: 5_000 })).rejects.toThrow('timed out'); + expect(daemon.listTools).toHaveBeenCalledTimes(1); + expect(daemon.closeServer).not.toHaveBeenCalled(); + }); }); diff --git a/tests/runtime-listtools-timeout.test.ts b/tests/runtime-listtools-timeout.test.ts new file mode 100644 index 00000000..227c2319 --- /dev/null +++ b/tests/runtime-listtools-timeout.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it, vi } from 'vitest'; +import type { ServerDefinition } from '../src/config-schema.js'; +import { createRuntime } from '../src/runtime.js'; + +type TestRuntime = Awaited>; +type ClientContext = Awaited>; + +function fakeListToolsContext(listTools: ReturnType): ClientContext { + return { + client: { + listTools, + close: vi.fn().mockResolvedValue(undefined), + }, + transport: { close: vi.fn().mockResolvedValue(undefined) }, + definition: { + name: 'alpha', + description: 'test server', + command: { kind: 'http', url: new URL('https://alpha.example.com') }, + source: { kind: 'local', path: '/tmp' }, + }, + oauthSession: undefined, + } as unknown as ClientContext; +} + +describe('runtime.listTools request timeout forwarding', () => { + const definitions: ServerDefinition[] = [ + { + name: 'alpha', + description: 'test server', + command: { kind: 'http', url: new URL('https://alpha.example.com') }, + source: { kind: 'local', path: '/tmp' }, + }, + ]; + + it('forwards timeoutMs to the MCP client so listings (and OAuth during auth) avoid the 60s SDK cap', async () => { + const runtime = await createRuntime({ servers: definitions }); + const listTools = vi.fn().mockResolvedValue({ tools: [] }); + const connect = vi.spyOn(runtime, 'connect').mockResolvedValue(fakeListToolsContext(listTools)); + + await runtime.listTools('alpha', { timeoutMs: 5_000 }); + + expect(connect).toHaveBeenCalledWith('alpha', expect.objectContaining({ oauthTimeoutMs: 5_000 })); + expect(listTools).toHaveBeenCalledWith(undefined, { + timeout: 5_000, + resetTimeoutOnProgress: true, + maxTotalTimeout: 5_000, + }); + }); + + it('omits SDK timeout options when timeoutMs is not provided (preserves prior behavior)', async () => { + const runtime = await createRuntime({ servers: definitions }); + const listTools = vi.fn().mockResolvedValue({ tools: [] }); + vi.spyOn(runtime, 'connect').mockResolvedValue(fakeListToolsContext(listTools)); + + await runtime.listTools('alpha'); + + expect(listTools).toHaveBeenCalledWith(undefined, undefined); + }); + + it('normalizes invalid timeoutMs before OAuth connection setup', async () => { + const runtime = await createRuntime({ servers: definitions }); + const listTools = vi.fn().mockResolvedValue({ tools: [] }); + const connect = vi.spyOn(runtime, 'connect').mockResolvedValue(fakeListToolsContext(listTools)); + + await runtime.listTools('alpha', { timeoutMs: 0 }); + + expect(connect).toHaveBeenCalledWith('alpha', expect.objectContaining({ oauthTimeoutMs: undefined })); + expect(listTools).toHaveBeenCalledWith(undefined, undefined); + }); +}); From 0e37f7e383af6d742d53c1037012ae79c84f0807 Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Mon, 27 Jul 2026 16:49:39 +0300 Subject: [PATCH 2/6] fix(daemon): keep multi-phase requests alive with progress frames Sizing the daemon socket deadline for a fixed number of phases cannot work. `McpRuntime.listTools()` issues one `tools/list` request per cursor page and gives each its own `timeoutMs`, so an OAuth wait followed by two or more slow pages outlives any constant multiple of the caller deadline. The socket then expires mid-flight, the client restarts the daemon and resends the listing, and the user is walked through a second OAuth flow. Stop predicting the phase count and observe the daemon instead. While a request is in flight the host emits newline-delimited progress frames, and the client treats each frame as proof of life and restarts its socket deadline. The deadline becomes a liveness budget rather than an operation budget: a request survives however many phases it needs, while a daemon that goes silent is still torn down and restarted exactly as before. Liveness alone would be too weak, so every operation stays provably bounded: - the first frame goes out immediately and the cadence is derived from the caller's own deadline, so a deadline shorter than the default interval cannot expire before the daemon has proved it is alive; - operations with a fixed phase count -- at most one connect plus one MCP request -- carry an absolute ceiling equal to the sum of the deadlines the daemon already applies to those phases, so a server that accepts a request and never answers yields a non-retryable `operation_timeout` rather than an indefinite wait; - `listTools` keeps its per-page deadline and gains a repeated-cursor guard, so the one operation whose phase count is data-dependent cannot page forever. Both peers decode frames incrementally, so the daemon's own status probe and stop handshake stay readable, and a response with no trailing newline still parses. Regression coverage drives the real client transport and the real host framing over a socket: an OAuth wait followed by three delayed `tools/list` pages resolves with one `listTools` request, no replay, and no daemon relaunch. Sibling cases cover a deadline below the default frame interval, a daemon that goes silent, and a server that repeats a pagination cursor. --- src/daemon/client.ts | 85 +++++--- src/daemon/host.ts | 199 +++++++++++++---- src/daemon/protocol.ts | 93 ++++++++ src/runtime.ts | 10 + tests/daemon-client-timeout.test.ts | 62 +++++- tests/daemon-listtools-progress.test.ts | 273 ++++++++++++++++++++++++ tests/runtime-listtools-timeout.test.ts | 14 ++ 7 files changed, 652 insertions(+), 84 deletions(-) create mode 100644 tests/daemon-listtools-progress.test.ts diff --git a/src/daemon/client.ts b/src/daemon/client.ts index 168c6a40..5747b972 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -5,7 +5,12 @@ import path from 'node:path'; import { listConfigLayerPaths } from '../config/path-discovery.js'; import { withFileLock } from '../fs-json.js'; import { getDaemonMetadataPath, getDaemonSocketPath } from './paths.js'; -import { DAEMON_PROTOCOL_VERSION } from './protocol.js'; +import { + DAEMON_PROTOCOL_VERSION, + DaemonFrameDecoder, + isDaemonProgressFrame, + resolveProgressInterval, +} from './protocol.js'; import type { CallToolParams, CloseServerParams, @@ -255,15 +260,26 @@ export class DaemonClient { } private async sendRequest(method: DaemonRequestMethod, params: unknown, timeoutOverrideMs?: number): Promise { + // This is an *idle* deadline, not an operation deadline: the daemon keeps + // resetting it with progress frames for as long as it is working, so a + // request survives any number of sequential phases (an OAuth code wait plus + // however many `tools/list` pages a server returns). The daemon still owns + // the operation deadline and answers with `operation_timeout` when a phase + // exceeds it; the socket only gives up when the daemon goes silent. + const idleTimeoutMs = resolveDaemonTimeout(timeoutOverrideMs); const request: DaemonRequest = { id: randomUUID(), method, params, + // Tell the daemon how often this deadline needs proof of life, so a short + // deadline cannot expire before the first frame arrives. + progressIntervalMs: resolveProgressInterval(idleTimeoutMs), }; const payload = JSON.stringify(request); - const timeoutMs = resolveDaemonTimeout(timeoutOverrideMs); - const response = await new Promise((resolve, reject) => { + const parsed = await new Promise>((resolve, reject) => { const socket = net.createConnection(this.socketPath); + const decoder = new DaemonFrameDecoder(); + let response: DaemonResponse | undefined; let settled = false; const finishReject = (error: Error): void => { if (settled) { @@ -272,23 +288,31 @@ export class DaemonClient { settled = true; reject(error); }; - const finishResolve = (value: string): void => { + const finishResolve = (value: DaemonResponse): void => { if (settled) { return; } settled = true; resolve(value); }; - socket.setTimeout(timeoutMs, () => { - // If the daemon doesn't answer in time we treat it as a transport error, destroy the socket, - // and let invoke() restart the daemon so hung keep-alive servers get a fresh start. - socket.destroy( - Object.assign(new Error('Daemon request timed out.'), { - code: 'ETIMEDOUT', - }) - ); + socket.setTimeout(idleTimeoutMs); + socket.on('timeout', () => { + // The daemon stopped proving it is alive. Treat that as a transport error, destroy the + // socket, and let invoke() restart the daemon so hung keep-alive servers get a fresh start. + socket.destroy(transportError('Daemon request timed out.', 'ETIMEDOUT')); }); - let buffer = ''; + const consume = (frames: ReturnType): void => { + for (const frame of frames) { + if (isDaemonProgressFrame(frame)) { + // Proof of life from the daemon: restart the idle deadline. + socket.setTimeout(idleTimeoutMs); + continue; + } + response = frame as DaemonResponse; + // The answer is in hand; stop policing the socket while it closes. + socket.setTimeout(0); + } + }; socket.on('connect', () => { socket.write(payload, (error) => { if (error) { @@ -298,27 +322,24 @@ export class DaemonClient { }); }); socket.on('data', (chunk) => { - buffer += chunk.toString(); + consume(decoder.push(chunk.toString())); + }); + socket.on('end', () => { + consume(decoder.flush()); + if (response) { + finishResolve(response); + return; + } + finishReject( + decoder.malformed + ? transportError('Failed to parse daemon response.', 'ECONNRESET') + : transportError('Empty daemon response.', 'ECONNRESET') + ); }); - socket.on('end', () => finishResolve(buffer)); socket.on('error', (error) => { finishReject(error as Error); }); }); - const trimmed = response.trim(); - if (!trimmed) { - const error = new Error('Empty daemon response.'); - (error as NodeJS.ErrnoException).code = 'ECONNRESET'; - throw error; - } - let parsed: DaemonResponse; - try { - parsed = JSON.parse(trimmed) as DaemonResponse; - } catch { - const parseError = new Error('Failed to parse daemon response.'); - (parseError as NodeJS.ErrnoException).code = 'ECONNRESET'; - throw parseError; - } if (!parsed.ok) { const error = new Error(parsed.error?.message ?? 'Daemon error'); (error as NodeJS.ErrnoException).code = parsed.error?.code; @@ -333,6 +354,12 @@ function deriveConfigKey(configPath: string): string { return crypto.createHash('sha1').update(absolute).digest('hex').slice(0, 12); } +function transportError(message: string, code: string): Error { + const error = new Error(message); + (error as NodeJS.ErrnoException).code = code; + return error; +} + function isTransportError(error: unknown): boolean { if (!error || typeof error !== 'object') { return false; diff --git a/src/daemon/host.ts b/src/daemon/host.ts index 1cfc4f6a..efb7faa4 100644 --- a/src/daemon/host.ts +++ b/src/daemon/host.ts @@ -1,3 +1,4 @@ +import { DEFAULT_REQUEST_TIMEOUT_MSEC } from '@modelcontextprotocol/sdk/shared/protocol.js'; import { ErrorCode } from '@modelcontextprotocol/sdk/types.js'; import { randomUUID } from 'node:crypto'; import fs from 'node:fs/promises'; @@ -7,7 +8,8 @@ import { loadDaemonConfig, type ServerDefinition } from '../config.js'; import { readJsonFile, withFileLock, writeJsonFile } from '../fs-json.js'; import { isKeepAliveServer } from '../lifecycle.js'; import { createRuntime, type Runtime } from '../runtime.js'; -import { OAuthTimeoutError } from '../runtime/oauth.js'; +import { OAuthTimeoutError, resolveOAuthTimeoutFromEnv } from '../runtime/oauth.js'; +import { raceWithTimeout } from '../runtime/utils.js'; import { collectConfigLayers, statConfigMtime } from './config-layers.js'; import { hashDaemonDefinitions } from './definition-hash.js'; import { @@ -20,9 +22,14 @@ import { } from './log-context.js'; import { DAEMON_OPERATION_TIMEOUT_CODE, + DAEMON_PROGRESS_INTERVAL_MS, DAEMON_PROTOCOL_VERSION, + DaemonFrameDecoder, + encodeDaemonFrame, + isDaemonProgressFrame, type CallToolParams, type CloseServerParams, + type DaemonFrame, type DaemonRequest, type DaemonResponse, type ListResourcesParams, @@ -362,7 +369,8 @@ async function sendDaemonStop(socketPath: string): Promise { params: {}, }; const socket = net.createConnection(socketPath); - let buffer = ''; + const decoder = new DaemonFrameDecoder(); + let response: DaemonResponse | undefined; let settled = false; const finish = (result: boolean): void => { if (settled) { @@ -378,15 +386,11 @@ async function sendDaemonStop(socketPath: string): Promise { socket.write(JSON.stringify(request)); }); socket.on('data', (chunk) => { - buffer += chunk.toString(); + response = lastResponseFrame(decoder.push(chunk.toString())) ?? response; }); socket.once('end', () => { - try { - const response = JSON.parse(buffer.trim()) as DaemonResponse; - finish(response.ok); - } catch { - finish(false); - } + response = lastResponseFrame(decoder.flush()) ?? response; + finish(response?.ok === true); }); socket.once('error', () => finish(false)); }); @@ -398,6 +402,16 @@ function delay(ms: number): Promise { }); } +function lastResponseFrame(frames: DaemonFrame[]): DaemonResponse | undefined { + let response: DaemonResponse | undefined; + for (const frame of frames) { + if (!isDaemonProgressFrame(frame)) { + response = frame as DaemonResponse; + } + } + return response; +} + function isProcessAlive(pid: number): boolean { if (!Number.isInteger(pid) || pid <= 0) { return false; @@ -413,7 +427,7 @@ function isProcessAlive(pid: number): boolean { async function probeDaemonStatus(socketPath: string): Promise { return await new Promise((resolve) => { const probe = net.createConnection(socketPath); - let buffer = ''; + const decoder = new DaemonFrameDecoder(); let settled = false; const finish = (status: StatusResult | null): void => { if (settled) { @@ -424,26 +438,21 @@ async function probeDaemonStatus(socketPath: string): Promise { - try { - const response = JSON.parse(buffer.trim()) as DaemonResponse; - return response.ok && response.result ? response.result : null; - } catch { - return null; - } + const statusFrom = (frames: DaemonFrame[]): StatusResult | null => { + const response = lastResponseFrame(frames); + return response?.ok && response.result ? response.result : null; }; probe.setTimeout(DAEMON_PROBE_TIMEOUT_MS, () => finish(null)); probe.once('connect', () => { probe.write(JSON.stringify({ id: randomUUID(), method: 'status', params: {} } satisfies DaemonRequest)); }); probe.on('data', (chunk) => { - buffer += chunk.toString(); - const status = parse(); + const status = statusFrom(decoder.push(chunk.toString())); if (status) { finish(status); } }); - probe.once('end', () => finish(parse())); + probe.once('end', () => finish(statusFrom(decoder.flush()))); probe.once('error', () => finish(null)); }); } @@ -503,16 +512,27 @@ async function handleSocketRequest( shutdown: () => Promise, preParsedRequest?: DaemonRequest ): Promise { - const { response, shouldShutdown } = await processRequest( - rawPayload, - runtime, - managedServers, - activity, - metadata, - logContext, - preParsedRequest + const stopProgress = startProgressFrames( + socket, + preParsedRequest?.id ?? 'unknown', + preParsedRequest?.progressIntervalMs ); - socket.write(JSON.stringify(response), () => { + let response: DaemonResponse; + let shouldShutdown: boolean; + try { + ({ response, shouldShutdown } = await processRequest( + rawPayload, + runtime, + managedServers, + activity, + metadata, + logContext, + preParsedRequest + )); + } finally { + stopProgress(); + } + socket.write(encodeDaemonFrame(response), () => { socket.end(() => { if (shouldShutdown) { void shutdown(); @@ -521,6 +541,34 @@ async function handleSocketRequest( }); } +/** + * Emits progress frames until the request settles. The client resets its idle + * deadline on each frame, so a long multi-phase operation -- an OAuth code wait + * plus any number of paginated `tools/list` pages -- never expires the socket + * mid-flight and never triggers a duplicate attempt on retry. Silence still + * means a wedged daemon. + * + * The first frame goes out immediately: the caller armed its deadline before the + * request reached us, so waiting a full interval would burn part of a short + * budget on dispatch latency alone. + */ +function startProgressFrames(socket: net.Socket, id: string, requestedIntervalMs?: number): () => void { + const intervalMs = + requestedIntervalMs && requestedIntervalMs > 0 + ? Math.min(requestedIntervalMs, DAEMON_PROGRESS_INTERVAL_MS) + : DAEMON_PROGRESS_INTERVAL_MS; + const emit = (): void => { + if (socket.destroyed || socket.writableEnded) { + return; + } + socket.write(encodeDaemonFrame({ type: 'progress', id })); + }; + emit(); + const timer = setInterval(emit, intervalMs); + timer.unref(); + return () => clearInterval(timer); +} + function normalizeDaemonDisableOAuth(value: boolean | undefined): boolean { // Daemon messages are independent requests. Omission means the caller did // not request OAuth suppression, so a previous --no-oauth pooled transport @@ -528,6 +576,28 @@ function normalizeDaemonDisableOAuth(value: boolean | undefined): boolean { return value === true; } +/** + * Absolute ceiling for a daemon operation that runs a *fixed* number of phases: + * at most one `connect()` -- which may include an interactive OAuth code wait -- + * plus one MCP request. The ceiling is the sum of the deadlines the daemon + * already applies to those phases, not a guess about how long work should take, + * so it can only fire once every real deadline has been blown. + * + * It exists so progress frames can never keep a caller waiting on a server that + * accepts a request and never answers: the caller gets a non-retryable + * `operation_timeout` instead of an indefinite wait or a daemon restart. + * + * `listTools` is deliberately excluded -- its page count is data-dependent, so it + * is bounded per page instead, plus a repeated-cursor guard in the runtime. + */ +function operationCeilingMs(requestTimeoutMs?: number): number { + const requestPhase = + requestTimeoutMs && Number.isFinite(requestTimeoutMs) && requestTimeoutMs > 0 + ? requestTimeoutMs + : DEFAULT_REQUEST_TIMEOUT_MSEC; + return resolveOAuthTimeoutFromEnv() + requestPhase; +} + function daemonRuntimeErrorCode(error: unknown): string { const errorCode = error && typeof error === 'object' ? (error as { code?: unknown }).code : undefined; if ( @@ -589,11 +659,14 @@ async function processRequest( logEvent(logContext, `callTool start server=${params.server} tool=${params.tool}`); } try { - const result = await runtime.callTool(params.server, params.tool, { - args: params.args ?? {}, - timeoutMs: params.timeoutMs, - disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), - }); + const result = await raceWithTimeout( + runtime.callTool(params.server, params.tool, { + args: params.args ?? {}, + timeoutMs: params.timeoutMs, + disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), + }), + operationCeilingMs(params.timeoutMs) + ); markActivity(params.server, activity); if (loggable) { logEvent(logContext, `callTool success server=${params.server} tool=${params.tool}`); @@ -644,11 +717,14 @@ async function processRequest( logEvent(logContext, `listResources start server=${params.server}`); } try { - const result = await runtime.listResources(params.server, { - ...params.params, - allowCachedAuth: params.allowCachedAuth, - disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), - }); + const result = await raceWithTimeout( + runtime.listResources(params.server, { + ...params.params, + allowCachedAuth: params.allowCachedAuth, + disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), + }), + operationCeilingMs() + ); markActivity(params.server, activity); if (loggable) { logEvent(logContext, `listResources success server=${params.server}`); @@ -670,10 +746,13 @@ async function processRequest( logEvent(logContext, `readResource start server=${params.server} uri=${params.uri}`); } try { - const result = await runtime.readResource(params.server, params.uri, { - allowCachedAuth: params.allowCachedAuth, - disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), - }); + const result = await raceWithTimeout( + runtime.readResource(params.server, params.uri, { + allowCachedAuth: params.allowCachedAuth, + disableOAuth: normalizeDaemonDisableOAuth(params.disableOAuth), + }), + operationCeilingMs() + ); markActivity(params.server, activity); if (loggable) { logEvent(logContext, `readResource success server=${params.server}`); @@ -695,7 +774,7 @@ async function processRequest( logEvent(logContext, `closeServer start server=${params.server}`); } try { - await runtime.close(params.server); + await raceWithTimeout(runtime.close(params.server), operationCeilingMs()); activity.set(params.server, { connected: false }); if (loggable) { logEvent(logContext, `closeServer success server=${params.server}`); @@ -793,3 +872,35 @@ export async function __testProcessRequest( preParsedRequest ); } + +/** + * Serves a single request on `socket` through the real host path -- progress + * frames included -- so transport tests exercise the shipping framing instead of + * a hand-rolled stand-in. + */ +export async function __testHandleSocketRequest( + socket: net.Socket, + request: DaemonRequest, + runtime: Runtime, + managedServers: Map, + metadata: { + configPath: string; + configLayers: Array<{ path: string; mtimeMs: number | null }>; + configMtimeMs: number | null; + socketPath: string; + startedAt: number; + logPath: string | null; + } +): Promise { + await handleSocketRequest( + JSON.stringify(request), + socket, + runtime, + managedServers, + new Map(), + { ...metadata, protocolVersion: DAEMON_PROTOCOL_VERSION }, + createLogContext({ enabled: false, logAllServers: false, servers: new Set() }), + async () => {}, + request + ); +} diff --git a/src/daemon/protocol.ts b/src/daemon/protocol.ts index cbb48275..df25a291 100644 --- a/src/daemon/protocol.ts +++ b/src/daemon/protocol.ts @@ -1,6 +1,30 @@ export const DAEMON_PROTOCOL_VERSION = 2; export const DAEMON_OPERATION_TIMEOUT_CODE = 'operation_timeout'; +// While a request is in flight the daemon emits progress frames on the same +// socket. The client treats every frame as proof of life and restarts its idle +// deadline, so a request stays alive for as many phases as it needs -- an OAuth +// code wait plus any number of paginated `tools/list` pages -- without the +// client having to predict how many phases there will be. +export const DAEMON_PROGRESS_INTERVAL_MS = 250; +/** + * Picks a heartbeat interval that fits inside the caller's idle deadline. A + * fixed interval would outlive a short deadline and expire the socket before the + * next frame arrived -- the same restart-and-replay failure the frames exist to + * prevent -- so the cadence always tracks the deadline rather than a constant. + * + * There is no lower clamp on purpose: a floor would be exactly the constant that + * overshoots the deadlines it is supposed to protect. The result is strictly + * below the caller's deadline for every deadline above 1ms, and 1ms deadlines + * are unachievable at any cadence. + */ +export function resolveProgressInterval(idleTimeoutMs: number): number { + if (!Number.isFinite(idleTimeoutMs) || idleTimeoutMs <= 0) { + return DAEMON_PROGRESS_INTERVAL_MS; + } + return Math.min(DAEMON_PROGRESS_INTERVAL_MS, Math.max(1, Math.floor(idleTimeoutMs / 3))); +} + export type DaemonRequestMethod = | 'callTool' | 'listTools' @@ -14,6 +38,9 @@ export interface DaemonRequest { @@ -26,6 +53,72 @@ export interface DaemonResponse { }; } +export interface DaemonProgressFrame { + readonly type: 'progress'; + readonly id: string; +} + +export type DaemonFrame = DaemonProgressFrame | DaemonResponse; + +export function isDaemonProgressFrame(frame: DaemonFrame): frame is DaemonProgressFrame { + return (frame as DaemonProgressFrame).type === 'progress'; +} + +// Frames are newline-delimited JSON. `JSON.stringify` never emits a raw newline, +// so a single line always holds exactly one frame. +export function encodeDaemonFrame(frame: DaemonFrame): string { + return `${JSON.stringify(frame)}\n`; +} + +/** + * Incrementally splits a daemon socket stream into frames. Lines that fail to + * parse are reported through `malformed` so callers can decide whether an + * unreadable stream is fatal. + */ +export class DaemonFrameDecoder { + private buffer = ''; + private sawMalformedLine = false; + + push(chunk: string): DaemonFrame[] { + this.buffer += chunk; + const frames: DaemonFrame[] = []; + let newlineIndex = this.buffer.indexOf('\n'); + while (newlineIndex !== -1) { + const line = this.buffer.slice(0, newlineIndex); + this.buffer = this.buffer.slice(newlineIndex + 1); + this.collect(line, frames); + newlineIndex = this.buffer.indexOf('\n'); + } + return frames; + } + + // Parses whatever is left once the stream ends. Daemons before the framed + // protocol wrote a bare response with no trailing newline. + flush(): DaemonFrame[] { + const remainder = this.buffer; + this.buffer = ''; + const frames: DaemonFrame[] = []; + this.collect(remainder, frames); + return frames; + } + + get malformed(): boolean { + return this.sawMalformedLine; + } + + private collect(line: string, frames: DaemonFrame[]): void { + const trimmed = line.trim(); + if (trimmed.length === 0) { + return; + } + try { + frames.push(JSON.parse(trimmed) as DaemonFrame); + } catch { + this.sawMalformedLine = true; + } + } +} + export interface CallToolParams { readonly server: string; readonly tool: string; diff --git a/src/runtime.ts b/src/runtime.ts index 32974cd7..e427010c 100644 --- a/src/runtime.ts +++ b/src/runtime.ts @@ -274,6 +274,10 @@ class McpRuntime implements Runtime { const tools: ServerToolInfo[] = []; try { let cursor: string | undefined; + // A server that repeats a cursor never terminates the walk. Every other + // phase here is bounded (the OAuth wait and each page carry a deadline), + // so this is the one way `listTools` could run forever. + const seenCursors = new Set(); do { // Forward the requested timeout to the MCP client so listings -- and the // interactive OAuth flow that listTools can trigger during `auth` -- don't @@ -292,6 +296,12 @@ class McpRuntime implements Runtime { })) ); cursor = response.nextCursor ?? undefined; + if (cursor !== undefined) { + if (seenCursors.has(cursor)) { + throw new Error(`Server '${server}' repeated tools/list cursor '${cursor}'; refusing to page forever.`); + } + seenCursors.add(cursor); + } } while (cursor); } catch (error) { // Keep-alive STDIO transports often die when Chrome closes; drop the cached client diff --git a/tests/daemon-client-timeout.test.ts b/tests/daemon-client-timeout.test.ts index 850757c7..1af4a36a 100644 --- a/tests/daemon-client-timeout.test.ts +++ b/tests/daemon-client-timeout.test.ts @@ -10,11 +10,18 @@ const timeoutRecords: Array<{ method: string; timeout: number }> = []; class MockSocket extends EventEmitter { currentTimeout = 0; private timeoutHandle?: NodeJS.Timeout; + private progressHandle?: NodeJS.Timeout; - setTimeout(ms: number, callback?: () => void): this { + // The client arms this as an idle deadline and re-arms it on every frame, so + // the mock has to clear the previous timer instead of stacking them. + setTimeout(ms: number): this { this.currentTimeout = ms; - if (enforceSocketTimeout && callback) { - this.timeoutHandle = setTimeout(callback, ms); + if (this.timeoutHandle) { + clearTimeout(this.timeoutHandle); + this.timeoutHandle = undefined; + } + if (enforceSocketTimeout && ms > 0) { + this.timeoutHandle = setTimeout(() => this.emit('timeout'), ms); } return this; } @@ -23,10 +30,13 @@ class MockSocket extends EventEmitter { const payload = JSON.parse(data.toString()); timeoutRecords.push({ method: payload.method, timeout: this.currentTimeout }); const response = buildResponse(payload.method, payload.id); + if (progressIntervalMs > 0) { + this.progressHandle = setInterval(() => { + this.emit('data', `${JSON.stringify({ type: 'progress', id: payload.id })}\n`); + }, progressIntervalMs); + } setTimeout(() => { - if (this.timeoutHandle) { - clearTimeout(this.timeoutHandle); - } + this.clearTimers(); this.emit('data', JSON.stringify(response)); this.emit('end'); }, responseDelayMs); @@ -40,18 +50,28 @@ class MockSocket extends EventEmitter { } destroy(error?: Error): this { - if (this.timeoutHandle) { - clearTimeout(this.timeoutHandle); - } + this.clearTimers(); if (error) { queueMicrotask(() => this.emit('error', error)); } return this; } + + private clearTimers(): void { + if (this.timeoutHandle) { + clearTimeout(this.timeoutHandle); + this.timeoutHandle = undefined; + } + if (this.progressHandle) { + clearInterval(this.progressHandle); + this.progressHandle = undefined; + } + } } let responseDelayMs = 5; let enforceSocketTimeout = false; +let progressIntervalMs = 0; let activeConfigPath = path.resolve('mcporter.config.json'); let activeSocketPath = ''; const createConnection = vi.fn(() => { @@ -102,6 +122,7 @@ describe('DaemonClient timeouts', () => { timeoutRecords.length = 0; responseDelayMs = 5; enforceSocketTimeout = false; + progressIntervalMs = 0; previousDaemonTimeout = process.env.MCPORTER_DAEMON_TIMEOUT_MS; previousDaemonDir = process.env.MCPORTER_DAEMON_DIR; tmpDaemonDir = await makeShortTempDir('daemon-timeout'); @@ -166,12 +187,31 @@ describe('DaemonClient timeouts', () => { await client.listTools({ server: 'foo', timeoutMs: 300_000 }); const statusRecord = timeoutRecords.find((entry) => entry.method === 'status'); const listRecord = timeoutRecords.find((entry) => entry.method === 'listTools'); - // Both the status probe and the listTools socket use the caller deadline - // verbatim. + // The socket deadline is the caller deadline verbatim: it measures silence + // from the daemon, not total operation time, so it never has to be scaled by + // a guessed phase count. expect(statusRecord?.timeout).toBe(300_000); expect(listRecord?.timeout).toBe(300_000); }); + it('keeps the listTools socket alive while the daemon reports progress', async () => { + // Progress frames arrive every 20ms for 600ms, so the response lands four + // times past the point where a 150ms deadline measured against total elapsed + // time would have fired. The gap between frame and deadline is wide enough to + // survive coarse timer granularity (Windows resolves timers to ~15.6ms). + responseDelayMs = 600; + progressIntervalMs = 20; + enforceSocketTimeout = true; + const configPath = 'mcporter.config.json'; + await writeFreshMetadata(configPath); + const client = new DaemonClient({ configPath, configExplicit: true }); + + await expect(client.listTools({ server: 'foo', timeoutMs: 150 })).resolves.toEqual({ ok: true }); + + const listRecord = timeoutRecords.find((entry) => entry.method === 'listTools'); + expect(listRecord?.timeout).toBe(150); + }); + it('leaves callTool operation socket budget unchanged', async () => { const configPath = 'mcporter.config.json'; await writeFreshMetadata(configPath); diff --git a/tests/daemon-listtools-progress.test.ts b/tests/daemon-listtools-progress.test.ts new file mode 100644 index 00000000..3ee765e5 --- /dev/null +++ b/tests/daemon-listtools-progress.test.ts @@ -0,0 +1,273 @@ +import fs from 'node:fs/promises'; +import net from 'node:net'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import type { ServerDefinition } from '../src/config.js'; +import { DaemonClient, resolveDaemonPaths } from '../src/daemon/client.js'; +import { collectConfigLayers } from '../src/daemon/config-layers.js'; +import { __testHandleSocketRequest } from '../src/daemon/host.js'; +import { + DAEMON_PROGRESS_INTERVAL_MS, + DAEMON_PROTOCOL_VERSION, + resolveProgressInterval, + type DaemonRequest, +} from '../src/daemon/protocol.js'; +import type { Runtime, ServerToolInfo } from '../src/runtime.js'; +import { makeShortTempDir } from './fixtures/test-helpers.js'; + +const { launchDaemonDetached } = vi.hoisted(() => ({ launchDaemonDetached: vi.fn() })); + +vi.mock('../src/daemon/launch.js', () => ({ launchDaemonDetached })); + +// Every phase runs longer than the client's socket deadline, so the request only +// survives if the daemon's progress frames keep resetting that deadline. +const CLIENT_TIMEOUT_MS = 400; +const PHASE_MS = 300; + +interface DaemonMetadataStub { + readonly configPath: string; + readonly configLayers: Array<{ path: string; mtimeMs: number | null }>; + readonly configMtimeMs: number | null; + readonly socketPath: string; + readonly startedAt: number; + readonly logPath: string | null; +} + +interface ServedDaemon { + readonly configPath: string; + readonly requests: DaemonRequest[]; + close: () => Promise; +} + +describe('progress cadence', () => { + it('stays strictly inside every caller deadline it can', () => { + // The cadence exists to refresh the deadline before it expires, so any + // interval at or above the deadline defeats the mechanism. 1ms is the one + // deadline integer milliseconds cannot beat -- and it was already + // unachievable before progress frames existed. + for (const deadlineMs of [2, 3, 5, 12, 24, 25, 74, 75, 300, 30_000, 300_000]) { + expect(resolveProgressInterval(deadlineMs)).toBeLessThan(deadlineMs); + } + expect(resolveProgressInterval(1)).toBe(1); + }); + + it('never exceeds the default cadence for long deadlines', () => { + expect(resolveProgressInterval(300_000)).toBe(DAEMON_PROGRESS_INTERVAL_MS); + expect(resolveProgressInterval(0)).toBe(DAEMON_PROGRESS_INTERVAL_MS); + expect(resolveProgressInterval(Number.NaN)).toBe(DAEMON_PROGRESS_INTERVAL_MS); + }); +}); + +describe('daemon listTools progress frames', () => { + let tmpDir: string; + let previousDaemonDir: string | undefined; + let daemon: ServedDaemon | undefined; + + beforeEach(async () => { + launchDaemonDetached.mockClear(); + previousDaemonDir = process.env.MCPORTER_DAEMON_DIR; + tmpDir = await makeShortTempDir('daemon-progress'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + }); + + afterEach(async () => { + await daemon?.close(); + daemon = undefined; + if (previousDaemonDir === undefined) { + delete process.env.MCPORTER_DAEMON_DIR; + } else { + process.env.MCPORTER_DAEMON_DIR = previousDaemonDir; + } + await fs.rm(tmpDir, { recursive: true, force: true }); + }); + + it('survives an OAuth wait followed by several paginated tools/list pages', async () => { + const pages: ServerToolInfo[][] = [[{ name: 'alpha' }, { name: 'beta' }], [{ name: 'gamma' }], [{ name: 'delta' }]]; + const listTools = vi.fn(async (): Promise => { + // Phase 1: the interactive OAuth code wait inside connect(). + await delay(PHASE_MS); + const tools: ServerToolInfo[] = []; + // Phases 2..n: one SDK tools/list request per cursor page. + for (const page of pages) { + await delay(PHASE_MS); + tools.push(...page); + } + return tools; + }); + daemon = await serveDaemon(tmpDir, listTools); + const client = new DaemonClient({ configPath: daemon.configPath, configExplicit: true, rootDir: tmpDir }); + + const startedAt = Date.now(); + const tools = await client.listTools({ server: 'oauth', timeoutMs: CLIENT_TIMEOUT_MS }); + const elapsedMs = Date.now() - startedAt; + + expect(tools).toEqual(pages.flat()); + // The operation outlived its socket deadline several times over -- the case a + // budget sized for a fixed number of phases cannot express. + expect(elapsedMs).toBeGreaterThan(CLIENT_TIMEOUT_MS); + // No replay: the daemon ran the OAuth flow exactly once. + expect(listTools).toHaveBeenCalledTimes(1); + expect(daemon.requests.filter((request) => request.method === 'listTools')).toHaveLength(1); + // No restart: a transport timeout would have relaunched the daemon. + expect(launchDaemonDetached).not.toHaveBeenCalled(); + }); + + it('reaches a deadline shorter than the default progress interval', async () => { + // A caller deadline below DAEMON_PROGRESS_INTERVAL_MS would expire before the + // daemon's first heartbeat if the cadence were fixed, restarting and replaying + // the very request the frames exist to protect. + const shortTimeoutMs = 120; + const listTools = vi.fn(async (): Promise => { + await delay(shortTimeoutMs * 3); + return [{ name: 'alpha' }]; + }); + daemon = await serveDaemon(tmpDir, listTools); + const client = new DaemonClient({ configPath: daemon.configPath, configExplicit: true, rootDir: tmpDir }); + + const tools = await client.listTools({ server: 'oauth', timeoutMs: shortTimeoutMs }); + + expect(tools).toEqual([{ name: 'alpha' }]); + expect(listTools).toHaveBeenCalledTimes(1); + expect(launchDaemonDetached).not.toHaveBeenCalled(); + }); + + it('still trips the socket deadline when the daemon stops sending progress frames', async () => { + // A wedged daemon accepts the request and then goes quiet. Silence is the one + // signal the client still treats as a dead transport. + daemon = await serveDaemon(tmpDir, undefined); + const client = new DaemonClient({ configPath: daemon.configPath, configExplicit: true, rootDir: tmpDir }); + const sendRequest = ( + client as unknown as { + sendRequest: (method: string, params: unknown, timeoutMs?: number) => Promise; + } + ).sendRequest.bind(client); + + await expect(sendRequest('listTools', { server: 'oauth' }, CLIENT_TIMEOUT_MS)).rejects.toMatchObject({ + code: 'ETIMEDOUT', + }); + }); +}); + +/** + * Serves one daemon socket backed by the real host request path, so the test + * exercises the shipping progress framing rather than a stand-in. Passing no + * `listTools` leaves operations unanswered to model a wedged daemon. + */ +async function serveDaemon(tmpDir: string, listTools: Runtime['listTools'] | undefined): Promise { + const configPath = path.join(tmpDir, 'mcporter.config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const { socketPath, metadataPath } = resolveDaemonPaths(configPath); + // On Windows the socket is a named pipe, so only the metadata directory is a + // real path worth creating -- and there is no stale socket file to remove. + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await removeSocketFile(socketPath); + + const metadata: DaemonMetadataStub = { + configPath, + configLayers: await collectConfigLayers({ configPath, rootDir: tmpDir }), + configMtimeMs: null, + socketPath, + startedAt: Date.now(), + logPath: null, + }; + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, + socketPath, + configPath, + configLayers: metadata.configLayers, + startedAt: metadata.startedAt, + }), + 'utf8' + ); + + const requests: DaemonRequest[] = []; + const runtime = { listTools } as unknown as Runtime; + const managedServers = new Map([ + [ + 'oauth', + { name: 'oauth', command: { kind: 'http', url: new URL('https://example.test/mcp') } } as ServerDefinition, + ], + ]); + + const open = new Set(); + const server = net.createServer({ allowHalfOpen: true }, (socket) => { + socket.setEncoding('utf8'); + open.add(socket); + socket.on('close', () => open.delete(socket)); + let buffer = ''; + let handled = false; + socket.on('error', () => socket.destroy()); + socket.on('data', (chunk) => { + buffer += chunk; + if (handled) { + return; + } + let request: DaemonRequest; + try { + request = JSON.parse(buffer.trim()) as DaemonRequest; + } catch { + return; + } + handled = true; + requests.push(request); + if (request.method === 'status') { + // The client's liveness probe must keep working in both scenarios. + socket.write(`${JSON.stringify({ id: request.id, ok: true, result: buildStatus(metadata) })}\n`, () => + socket.end() + ); + return; + } + if (!listTools) { + return; + } + void __testHandleSocketRequest(socket, request, runtime, managedServers, metadata); + }); + }); + await new Promise((resolve, reject) => { + server.once('error', reject); + server.listen(socketPath, () => { + server.off('error', reject); + resolve(); + }); + }); + + return { + configPath, + requests, + close: async () => { + for (const socket of open) { + socket.destroy(); + } + await new Promise((resolve) => server.close(() => resolve())); + await removeSocketFile(socketPath); + }, + }; +} + +function buildStatus(metadata: DaemonMetadataStub) { + return { + pid: process.pid, + protocolVersion: DAEMON_PROTOCOL_VERSION, + startedAt: metadata.startedAt, + configPath: metadata.configPath, + configLayers: metadata.configLayers, + socketPath: metadata.socketPath, + servers: [], + }; +} + +async function removeSocketFile(socketPath: string): Promise { + if (process.platform === 'win32') { + return; + } + await fs.unlink(socketPath).catch(() => {}); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => { + setTimeout(resolve, ms); + }); +} diff --git a/tests/runtime-listtools-timeout.test.ts b/tests/runtime-listtools-timeout.test.ts index 227c2319..616d43c5 100644 --- a/tests/runtime-listtools-timeout.test.ts +++ b/tests/runtime-listtools-timeout.test.ts @@ -57,6 +57,20 @@ describe('runtime.listTools request timeout forwarding', () => { expect(listTools).toHaveBeenCalledWith(undefined, undefined); }); + it('walks every cursor page but refuses to page forever on a repeated cursor', async () => { + const runtime = await createRuntime({ servers: definitions }); + const listTools = vi + .fn() + .mockResolvedValueOnce({ tools: [{ name: 'a' }], nextCursor: 'page-2' }) + .mockResolvedValueOnce({ tools: [{ name: 'b' }], nextCursor: 'page-3' }) + .mockResolvedValue({ tools: [{ name: 'c' }], nextCursor: 'page-2' }); + vi.spyOn(runtime, 'connect').mockResolvedValue(fakeListToolsContext(listTools)); + + await expect(runtime.listTools('alpha')).rejects.toThrow(/repeated tools\/list cursor 'page-2'/); + // Three pages walked, then the cycle is caught instead of spinning. + expect(listTools).toHaveBeenCalledTimes(3); + }); + it('normalizes invalid timeoutMs before OAuth connection setup', async () => { const runtime = await createRuntime({ servers: definitions }); const listTools = vi.fn().mockResolvedValue({ tools: [] }); From 7911b389a8fd4604ab808da647fd1c4379521571 Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Fri, 31 Jul 2026 17:46:35 +0300 Subject: [PATCH 3/6] fix(daemon): drain busy shared daemons before protocol replacement The previous code's protocol-version check classified any pre-v2 daemon as stale and called stop() on it from ensureDaemon(). A second client upgrading mid-request -- OAuth code wait or a delayed cursor page -- would race ahead of the first client's in-flight call, kill the daemon, and force the very replay the liveness mechanism was added to prevent. The fix makes the replacement an idle-drain transition, not a preempt: * host: the status response now reports the live in-flight count, sampled at response time so a long-running request still shows up as in flight. * protocol: StatusResult.activeRequests is optional so a v1 daemon that cannot advertise it does not break old clients. * client: before restartDaemon() issues stop(), it polls the live daemon up to MCPORTER_DAEMON_DRAIN_TIMEOUT_MS (60s default) for activeRequests to read zero. Pre-v2 daemons are treated as permanently busy, so the drain timeout is the only knob for them too. If the daemon is still busy past the timeout, the replacement is abandoned and the caller fails loud rather than killing the peer. Drain timeout is env-var overridable so the test suite can exercise the refusal branch without waiting the full minute. --- src/daemon/client.ts | 73 ++++++++++++++++++++++++++++++++++++++++++ src/daemon/host.ts | 28 ++++++++++++++-- src/daemon/protocol.ts | 9 ++++++ 3 files changed, 108 insertions(+), 2 deletions(-) diff --git a/src/daemon/client.ts b/src/daemon/client.ts index 5747b972..e2b9ca0c 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -31,6 +31,27 @@ export interface DaemonClientOptions { const DEFAULT_DAEMON_TIMEOUT_MS = 30_000; const MIN_DAEMON_STATUS_TIMEOUT_MS = 1_000; +// How long a client is willing to wait for a shared daemon it has decided to +// replace to finish its in-flight work before the replacement is abandoned. +// Capped so an upgraded client never blocks the calling command indefinitely +// waiting on another process; the cap is also the only knob for v1 daemons, +// which do not advertise `activeRequests` and so look busy by definition. +// Overridable through `MCPORTER_DAEMON_DRAIN_TIMEOUT_MS` for tests that need +// to exercise the refusal branch without waiting the full default. +const DEFAULT_DAEMON_DRAIN_TIMEOUT_MS = 60_000; +const DAEMON_DRAIN_POLL_MS = 100; + +function resolveDrainTimeoutMs(): number { + const raw = process.env.MCPORTER_DAEMON_DRAIN_TIMEOUT_MS; + if (!raw) { + return DEFAULT_DAEMON_DRAIN_TIMEOUT_MS; + } + const parsed = Number.parseInt(raw, 10); + if (!Number.isFinite(parsed) || parsed <= 0) { + return DEFAULT_DAEMON_DRAIN_TIMEOUT_MS; + } + return parsed; +} export interface DaemonPaths { readonly key: string; @@ -149,12 +170,64 @@ export class DaemonClient { if (options.reason === 'stale-config' && currentStatus && (await this.checkConfigState()) === 'fresh') { return; } + // Codex / ClawSweeper P1: an upgraded client must not stop a daemon that + // another client is currently waiting on. Wait for the in-flight count to + // reach zero before issuing `stop`; bail out (do not replace) if the + // daemon is still busy past the drain timeout, so the busy client keeps + // its daemon and the caller fails loud rather than silently replaying. + if (currentStatus && currentStatus.pid === options.expectedPid) { + const drainTimeoutMs = resolveDrainTimeoutMs(); + const drained = await this.waitForDaemonIdle(currentStatus, drainTimeoutMs); + if (!drained) { + throw new Error( + `Shared daemon pid=${currentStatus.pid} is still busy after ${drainTimeoutMs}ms; refusing to replace it.` + ); + } + } await this.stop().catch(() => {}); await this.waitForStopped(); await this.launchDaemonAndWait(); }); } + /** + * Polls the live daemon's status until its `activeRequests` counter reads + * zero. Resolves `true` once the daemon is idle, `false` if the timeout + * expires, and `false` if the daemon disappears mid-wait so the caller + * falls through to the normal replacement path. + * + * A daemon that does not advertise `activeRequests` (pre-v2) is treated as + * permanently busy -- the only safe read of "is this daemon still serving + * a v1 request?" is to wait long enough for that request to finish, which + * is exactly the drain we are trying to guarantee. + */ + private async waitForDaemonIdle(seedStatus: StatusResult, timeoutMs: number): Promise { + const deadline = Date.now() + timeoutMs; + let status: StatusResult | null = seedStatus; + while (Date.now() < deadline) { + if (!status) { + return true; + } + const active = status.activeRequests; + if (active === undefined) { + // Pre-v2 daemon: keep waiting. The drain timeout bounds the wait, so a + // stuck v1 daemon eventually falls through to the refusal branch and + // is left running rather than killed mid-request. + await delay(DAEMON_DRAIN_POLL_MS); + } else if (active <= 0) { + return true; + } else { + await delay(DAEMON_DRAIN_POLL_MS); + } + status = await this.readVerifiedStatus(); + if (status && status.pid !== seedStatus.pid) { + // Another client already replaced it; nothing to drain. + return true; + } + } + return false; + } + private async startDaemon(options: { preflightTimeoutMs?: number } = {}): Promise { await this.startingWithLock(async () => { if (await this.isResponsive(options.preflightTimeoutMs)) { diff --git a/src/daemon/host.ts b/src/daemon/host.ts index efb7faa4..dd8d8e03 100644 --- a/src/daemon/host.ts +++ b/src/daemon/host.ts @@ -136,6 +136,11 @@ export async function runDaemonHost(options: DaemonHostOptions): Promise { } })(); }, daemonIdleWatcherInterval(daemonConfig.idleTimeoutMs)); + // Snapshotted by the `status` response so a client that decides the daemon is + // stale can wait for the in-flight count to drop to zero before stopping it. + // Reading the counter at response time -- not at request acceptance -- means + // the snapshot reflects work the host has actually committed to. + const readActiveDaemonRequests = (): number => activeDaemonRequests; idleWatcher.unref(); logEvent(logContext, 'Daemon host started.'); @@ -179,6 +184,7 @@ export async function runDaemonHost(options: DaemonHostOptions): Promise { logPath: options.logPath ?? null, configMtimeMs, definitionHash, + readActiveDaemonRequests, }, logContext, shutdown, @@ -507,6 +513,7 @@ async function handleSocketRequest( startedAt: number; logPath: string | null; definitionHash?: string; + readActiveDaemonRequests: () => number; }, logContext: LogContext, shutdown: () => Promise, @@ -624,6 +631,7 @@ async function processRequest( startedAt: number; logPath: string | null; definitionHash?: string; + readActiveDaemonRequests: () => number; }, logContext: LogContext, preParsedRequest?: DaemonRequest @@ -802,6 +810,12 @@ async function processRequest( definitionHash: metadata.definitionHash, socketPath: metadata.socketPath, logPath: metadata.logPath ?? undefined, + // Read at response time so a status probe after a long OAuth wait + // reports the in-flight count as the client will see it, not the + // count at request acceptance. Clients coordinate replacement on + // this number; understating it would re-open the + // mid-request-stop regression. + activeRequests: metadata.readActiveDaemonRequests(), servers: Array.from(managedServers.values()).map((def) => { const entry = activity.get(def.name); return { @@ -858,6 +872,7 @@ export async function __testProcessRequest( startedAt: number; logPath: string | null; definitionHash?: string; + readActiveDaemonRequests?: () => number; }, logContext: LogContext, preParsedRequest?: DaemonRequest @@ -867,7 +882,11 @@ export async function __testProcessRequest( runtime, managedServers, activity, - { ...metadata, protocolVersion: metadata.protocolVersion ?? DAEMON_PROTOCOL_VERSION }, + { + ...metadata, + protocolVersion: metadata.protocolVersion ?? DAEMON_PROTOCOL_VERSION, + readActiveDaemonRequests: metadata.readActiveDaemonRequests ?? ((): number => 0), + }, logContext, preParsedRequest ); @@ -890,6 +909,7 @@ export async function __testHandleSocketRequest( socketPath: string; startedAt: number; logPath: string | null; + readActiveDaemonRequests?: () => number; } ): Promise { await handleSocketRequest( @@ -898,7 +918,11 @@ export async function __testHandleSocketRequest( runtime, managedServers, new Map(), - { ...metadata, protocolVersion: DAEMON_PROTOCOL_VERSION }, + { + ...metadata, + protocolVersion: DAEMON_PROTOCOL_VERSION, + readActiveDaemonRequests: metadata.readActiveDaemonRequests ?? ((): number => 0), + }, createLogContext({ enabled: false, logAllServers: false, servers: new Set() }), async () => {}, request diff --git a/src/daemon/protocol.ts b/src/daemon/protocol.ts index df25a291..6d9a1618 100644 --- a/src/daemon/protocol.ts +++ b/src/daemon/protocol.ts @@ -172,4 +172,13 @@ export interface StatusResult { readonly connected: boolean; readonly lastUsedAt?: number; }>; + /** + * Number of in-flight daemon requests when the status was answered. Absent on + * pre-v2 daemons; clients that need to coordinate a replacement must treat + * `undefined` as "unknown, assume busy" and wait before stopping the daemon. + * Without this signal an upgraded client can stop a live v1 daemon mid-OAuth + * or mid-paginated `tools/list` and force the very replay this protocol was + * introduced to prevent. + */ + readonly activeRequests?: number; } From 4e3977cb8fbd4b8117cda7362568163e0ceb1451 Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Fri, 31 Jul 2026 17:46:42 +0300 Subject: [PATCH 4/6] test(daemon): prove busy-daemon drain and operation ceiling Regression coverage for the Codex/ClawSweeper review round: * daemon-client-config-stale: two new tests pin the drain behavior. The first asserts a busy shared daemon is left alone until its in-flight count flips to zero, then replaced; the second asserts a daemon that never drains is left running rather than killed mid-request. * daemon-host: three new tests pin the operation ceiling for callTool, listResources, and readResource against a wedged runtime, using MCPORTER_OAUTH_TIMEOUT_MS=50 and fake timers so the OAuth-timeout-plus-60s-DEFAULT_REQUEST_TIMEOUT_MSEC ceiling fires in milliseconds rather than the production minute-plus. * daemon-listtools-progress: one new test exercises a 30ms caller deadline -- well under the 250ms default progress cadence -- to prove the first progress frame beats the deadline. The existing 120ms test already exercised this implicitly; the 30ms case makes the P2 fix visible in the test name. --- tests/daemon-client-config-stale.test.ts | 139 +++++++++++++++++++++++ tests/daemon-host.test.ts | 115 +++++++++++++++++++ tests/daemon-listtools-progress.test.ts | 21 ++++ 3 files changed, 275 insertions(+) diff --git a/tests/daemon-client-config-stale.test.ts b/tests/daemon-client-config-stale.test.ts index b471ae41..d8e1e1b4 100644 --- a/tests/daemon-client-config-stale.test.ts +++ b/tests/daemon-client-config-stale.test.ts @@ -8,6 +8,11 @@ import { makeShortTempDir } from './fixtures/test-helpers.js'; const sentMethods: string[] = []; const launchDaemonDetached = vi.hoisted(() => vi.fn()); let createConnection: ReturnType; +// When true, the first `status` poll flips the fake daemon's `activeRequests` +// to zero. This models a shared daemon whose long request finishes while the +// upgrading client is still draining -- the very window the new wait-for-idle +// logic was introduced to cover. +let drainAfterFirstStatusPoll = false; class MockSocket extends EventEmitter { setTimeout(): this { @@ -20,6 +25,13 @@ class MockSocket extends EventEmitter { if (payload.method === 'stop') { activeStatusPid = findNonRunningPid(); } + if (payload.method === 'status' && drainAfterFirstStatusPoll && activeRequests > 0) { + // Simulate a shared daemon whose in-flight request finishes after the + // upgrading client has noticed the staleness but before the drain + // timeout. The next status poll must see `activeRequests: 0` so the + // replacement is allowed to proceed. + activeRequests = 0; + } const response = buildResponse(payload.method, payload.id); queueMicrotask(() => { this.emit('data', JSON.stringify(response)); @@ -52,6 +64,10 @@ function buildResponse(method: string, id: string) { configLayers: activeLayers, socketPath: activeSocketPath, servers: [], + // Default to idle so the existing config-stale tests still hit the + // `stop` path quickly; the two-client busy-daemon tests below flip + // this on to assert the new drain behavior. + activeRequests, }, }; } @@ -68,6 +84,11 @@ let activeStatusPid = process.pid; let activeSocketPath: string; let previousDaemonDir: string | undefined; let activeLayers: Array<{ path: string; mtimeMs: number | null }> = []; +// `activeRequests` is whatever the fake daemon wants to advertise to the next +// status poll. Tests that need to exercise the busy-daemon drain path bump it +// above zero; the default keeps the existing config-stale tests on the fast +// path. +let activeRequests = 0; vi.mock('node:net', () => { createConnection = vi.fn(() => { @@ -89,6 +110,8 @@ describe('DaemonClient config freshness', () => { sentMethods.length = 0; previousDaemonDir = process.env.MCPORTER_DAEMON_DIR; activeLayers = []; + activeRequests = 0; + drainAfterFirstStatusPoll = false; launchDaemonDetached.mockClear(); launchDaemonDetached.mockImplementation( (options: { metadataPath: string; socketPath: string; configPath: string }) => { @@ -289,6 +312,122 @@ describe('DaemonClient config freshness', () => { expect(launchDaemonDetached).toHaveBeenCalledTimes(1); expect(sentMethods).toContain('listTools'); }); + + // Codex / ClawSweeper P1: an upgraded client must not stop a shared daemon + // that another caller is still waiting on. The next two tests prove both + // halves of the new wait-for-idle path against a status response that + // honestly reports the in-flight count. + it('waits for a busy shared daemon to drain before sending stop', async () => { + const tmpDir = await makeShortTempDir('daemon-busy-drain'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + + const configPath = path.join(tmpDir, 'config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const stat = await fs.stat(configPath); + // Use the live test pid so `isProcessRunning` returns true and the + // client's `readVerifiedStatus` actually consults the fake daemon -- + // otherwise the drain check is bypassed and the upgrade races to a + // stop regardless of in-flight work. + const livePid = process.pid; + const { metadataPath, socketPath } = resolveDaemonPaths(configPath); + activeConfigPath = configPath; + activeSocketPath = socketPath; + activeConfigMtime = stat.mtimeMs; + activeStatusPid = livePid; + activeLayers = [{ path: configPath, mtimeMs: stat.mtimeMs }]; + // Another client is still waiting on a long request on the v1 daemon. + activeRequests = 1; + // The in-flight request finishes on the next status poll, mid-drain. + drainAfterFirstStatusPoll = true; + + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: livePid, + protocolVersion: DAEMON_PROTOCOL_VERSION - 1, + socketPath, + configPath, + startedAt: Date.now() - 10_000, + logPath: null, + configMtimeMs: stat.mtimeMs, + configLayers: activeLayers, + }), + 'utf8' + ); + + const client = new DaemonClient({ configPath, configExplicit: true, rootDir: tmpDir }); + await client.listTools({ server: 'playwright' }); + + // The upgrading client must observe `activeRequests` flip to 0 *before* + // it issues `stop`. Anything else would have killed the busy peer's + // in-flight OAuth or cursor page. + const stopIndex = sentMethods.indexOf('stop'); + const statusCount = sentMethods.filter((method) => method === 'status').length; + expect(stopIndex).toBeGreaterThan(0); + expect(statusCount).toBeGreaterThanOrEqual(2); + // The last `status` poll seen by the client before `stop` already saw + // the drained counter. + expect(activeRequests).toBe(0); + expect(launchDaemonDetached).toHaveBeenCalledTimes(1); + expect(sentMethods).toContain('listTools'); + }); + + it('refuses to replace a daemon that never drains before the timeout', async () => { + const tmpDir = await makeShortTempDir('daemon-busy-stuck'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + // Shrink the drain cap so the test does not block for the production + // 60s default. The env var is the same knob the client reads at runtime. + const previousDrainTimeout = process.env.MCPORTER_DAEMON_DRAIN_TIMEOUT_MS; + process.env.MCPORTER_DAEMON_DRAIN_TIMEOUT_MS = '500'; + + try { + const configPath = path.join(tmpDir, 'config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const stat = await fs.stat(configPath); + const livePid = process.pid; + const { metadataPath, socketPath } = resolveDaemonPaths(configPath); + activeConfigPath = configPath; + activeSocketPath = socketPath; + activeConfigMtime = stat.mtimeMs; + activeStatusPid = livePid; + activeLayers = [{ path: configPath, mtimeMs: stat.mtimeMs }]; + // The peer is wedged and never returns. The client must refuse to stop + // the daemon -- the alternative is killing the peer's request and + // triggering the very replay this PR exists to remove. + activeRequests = 1; + + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: livePid, + protocolVersion: DAEMON_PROTOCOL_VERSION - 1, // stale: triggers the drain + socketPath, + configPath, + startedAt: Date.now() - 10_000, + logPath: null, + configMtimeMs: stat.mtimeMs, + configLayers: activeLayers, + }), + 'utf8' + ); + + const client = new DaemonClient({ configPath, configExplicit: true, rootDir: tmpDir }); + await expect(client.listTools({ server: 'playwright' })).rejects.toThrow( + /still busy after 500ms; refusing to replace it/ + ); + expect(launchDaemonDetached).not.toHaveBeenCalled(); + // The busy daemon was not stopped: the peer's request is left intact. + expect(sentMethods).not.toContain('stop'); + } finally { + if (previousDrainTimeout === undefined) { + delete process.env.MCPORTER_DAEMON_DRAIN_TIMEOUT_MS; + } else { + process.env.MCPORTER_DAEMON_DRAIN_TIMEOUT_MS = previousDrainTimeout; + } + } + }); }); function findNonRunningPid(): number { diff --git a/tests/daemon-host.test.ts b/tests/daemon-host.test.ts index af31f402..b597d2b3 100644 --- a/tests/daemon-host.test.ts +++ b/tests/daemon-host.test.ts @@ -3,6 +3,7 @@ import fs from 'node:fs/promises'; import net from 'node:net'; import os from 'node:os'; import path from 'node:path'; +import { DEFAULT_REQUEST_TIMEOUT_MSEC } from '@modelcontextprotocol/sdk/shared/protocol.js'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import type { ServerDefinition } from '../src/config.js'; import { @@ -180,6 +181,118 @@ describe('daemon host request handling', () => { }); }); +describe('operation ceiling for fixed-phase daemon methods', () => { + const metadata = { + configPath: '/tmp/config.json', + configLayers: [], + configMtimeMs: Date.now(), + socketPath: '/tmp/socket', + startedAt: Date.now(), + logPath: null, + }; + const logContext = { enabled: false, logAllServers: false, servers: new Set() }; + const managedServers = createManagedServers(); + + let previousOAuthTimeout: string | undefined; + + beforeEach(() => { + previousOAuthTimeout = process.env.MCPORTER_OAUTH_TIMEOUT_MS; + }); + + afterEach(() => { + if (previousOAuthTimeout === undefined) { + delete process.env.MCPORTER_OAUTH_TIMEOUT_MS; + } else { + process.env.MCPORTER_OAUTH_TIMEOUT_MS = previousOAuthTimeout; + } + }); + + it('fires the operation ceiling for callTool when the SDK never answers', async () => { + // Codex P1: a keep-alive server that accepts callTool but never responds used to + // ride progress frames forever. The ceiling is OAuth timeout + the caller + // deadline; the SDK's own timeout is the floor either way, so a wedged call + // becomes a non-retryable `operation_timeout` instead of an indefinite wait. + process.env.MCPORTER_OAUTH_TIMEOUT_MS = '50'; + vi.useFakeTimers(); + try { + const runtime = { + callTool: vi.fn().mockImplementation(neverResolving), + listTools: vi.fn(), + listResources: vi.fn(), + readResource: vi.fn(), + close: vi.fn().mockResolvedValue(undefined), + } as unknown as Runtime; + + const promise = __testProcessRequest('', runtime, managedServers, new Map(), metadata, logContext, { + id: 'call', + method: 'callTool', + params: { server: 'oauth', tool: 'ping', timeoutMs: 5_000 }, + }); + const expectation = expect(promise).resolves.toMatchObject({ + response: { ok: false, error: { code: 'operation_timeout' } }, + }); + await vi.advanceTimersByTimeAsync(DEFAULT_REQUEST_TIMEOUT_MSEC + 5_000 + 50); + await expectation; + } finally { + vi.useRealTimers(); + } + }); + + it('fires the operation ceiling for listResources when the SDK never answers', async () => { + process.env.MCPORTER_OAUTH_TIMEOUT_MS = '50'; + vi.useFakeTimers(); + try { + const runtime = { + callTool: vi.fn(), + listTools: vi.fn(), + listResources: vi.fn().mockImplementation(neverResolving), + readResource: vi.fn(), + close: vi.fn().mockResolvedValue(undefined), + } as unknown as Runtime; + + const promise = __testProcessRequest('', runtime, managedServers, new Map(), metadata, logContext, { + id: 'resources', + method: 'listResources', + params: { server: 'oauth' }, + }); + const expectation = expect(promise).resolves.toMatchObject({ + response: { ok: false, error: { code: 'operation_timeout' } }, + }); + await vi.advanceTimersByTimeAsync(DEFAULT_REQUEST_TIMEOUT_MSEC + 50); + await expectation; + } finally { + vi.useRealTimers(); + } + }); + + it('fires the operation ceiling for readResource when the SDK never answers', async () => { + process.env.MCPORTER_OAUTH_TIMEOUT_MS = '50'; + vi.useFakeTimers(); + try { + const runtime = { + callTool: vi.fn(), + listTools: vi.fn(), + listResources: vi.fn(), + readResource: vi.fn().mockImplementation(neverResolving), + close: vi.fn().mockResolvedValue(undefined), + } as unknown as Runtime; + + const promise = __testProcessRequest('', runtime, managedServers, new Map(), metadata, logContext, { + id: 'resource', + method: 'readResource', + params: { server: 'oauth', uri: 'file://missing' }, + }); + const expectation = expect(promise).resolves.toMatchObject({ + response: { ok: false, error: { code: 'operation_timeout' } }, + }); + await vi.advanceTimersByTimeAsync(DEFAULT_REQUEST_TIMEOUT_MSEC + 50); + await expectation; + } finally { + vi.useRealTimers(); + } + }); +}); + const describeUnixSocket = process.platform === 'win32' ? describe.skip : describe; describeUnixSocket('isDaemonResponding', () => { @@ -324,6 +437,8 @@ function createRuntimeDouble(): { }; } +const neverResolving = (): Promise => new Promise(() => {}); + function createManagedServers(): Map { return new Map([ [ diff --git a/tests/daemon-listtools-progress.test.ts b/tests/daemon-listtools-progress.test.ts index 3ee765e5..8b9f4911 100644 --- a/tests/daemon-listtools-progress.test.ts +++ b/tests/daemon-listtools-progress.test.ts @@ -131,6 +131,27 @@ describe('daemon listTools progress frames', () => { expect(launchDaemonDetached).not.toHaveBeenCalled(); }); + it('emits the first progress frame before any cadence interval can elapse', async () => { + // Codex P2: the host used to wait a full cadence before the first frame, which + // would have expired a sub-250ms deadline before any proof of life arrived. + // The first frame is now written synchronously on dispatch, so a deadline + // smaller than the default cadence still keeps the request alive. + const shortTimeoutMs = 30; + const listTools = vi.fn(async (): Promise => { + // Run well past the deadline so the test would deadlock without an + // immediate first frame -- a stalled socket cannot be rescued by cadence. + await delay(shortTimeoutMs * 5); + return [{ name: 'alpha' }]; + }); + daemon = await serveDaemon(tmpDir, listTools); + const client = new DaemonClient({ configPath: daemon.configPath, configExplicit: true, rootDir: tmpDir }); + + await client.listTools({ server: 'oauth', timeoutMs: shortTimeoutMs }); + + expect(listTools).toHaveBeenCalledTimes(1); + expect(launchDaemonDetached).not.toHaveBeenCalled(); + }); + it('still trips the socket deadline when the daemon stops sending progress frames', async () => { // A wedged daemon accepts the request and then goes quiet. Silence is the one // signal the client still treats as a dead transport. From 74b69539d094a53b8fef1aaeeb8f0c1959af4a8e Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Fri, 31 Jul 2026 18:13:22 +0300 Subject: [PATCH 5/6] fix(daemon): do not stop a peer's fresh daemon after the drain The earlier drain landed a contract bug: it only checked whether the original daemon had gone idle, then immediately fell through to stop(). A second client that won the race to replace the busy daemon during the drain would see its fresh daemon killed by the upgrading client the moment the drain returned. Three changes close the gap: * client: add `probeLiveStatus` that sends a raw status probe without the pid-match filter `readVerifiedStatus` applies. `readVerifiedStatus` collapses pid mismatches into null, which would short-circuit the drain check and re-introduce the very regression the drain exists to prevent. * client: `waitForDaemonIdle` now returns both `drained` and `replacedByPeer`. The pid is tracked across polls so a swap mid-wait is visible at the moment the drain resolves. * client: `restartDaemon` uses `probeLiveStatus` and bails out on `replacedByPeer` instead of issuing `stop`. A third regression test pins the new path: a busy daemon that another client replaces mid-drain must not receive stop, and the upgrading client must not relaunch a daemon of its own. --- src/daemon/client.ts | 87 +++++++++++++++++------ tests/daemon-client-config-stale.test.ts | 88 ++++++++++++++++++++++-- 2 files changed, 147 insertions(+), 28 deletions(-) diff --git a/src/daemon/client.ts b/src/daemon/client.ts index e2b9ca0c..26aa300b 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -158,16 +158,20 @@ export class DaemonClient { private async restartDaemon(options: { reason?: 'stale-config'; expectedPid?: number } = {}): Promise { await this.startingWithLock(async () => { - const currentStatus = await this.readVerifiedStatus(); + // Probe the daemon without the pid-match filter so the drain check + // below can see whether the daemon is actually busy. `readVerifiedStatus` + // would silently return null on a pid mismatch and the busy-daemon + // regression would slip past us. + const liveStatus = await this.probeLiveStatus(); if ( - currentStatus && + liveStatus && options.expectedPid !== undefined && - currentStatus.pid !== options.expectedPid && + liveStatus.pid !== options.expectedPid && (await this.checkConfigState()) === 'fresh' ) { return; } - if (options.reason === 'stale-config' && currentStatus && (await this.checkConfigState()) === 'fresh') { + if (options.reason === 'stale-config' && liveStatus && (await this.checkConfigState()) === 'fresh') { return; } // Codex / ClawSweeper P1: an upgraded client must not stop a daemon that @@ -175,14 +179,20 @@ export class DaemonClient { // reach zero before issuing `stop`; bail out (do not replace) if the // daemon is still busy past the drain timeout, so the busy client keeps // its daemon and the caller fails loud rather than silently replaying. - if (currentStatus && currentStatus.pid === options.expectedPid) { + if (liveStatus && options.expectedPid !== undefined && liveStatus.pid === options.expectedPid) { const drainTimeoutMs = resolveDrainTimeoutMs(); - const drained = await this.waitForDaemonIdle(currentStatus, drainTimeoutMs); - if (!drained) { + const drainResult = await this.waitForDaemonIdle(liveStatus, drainTimeoutMs); + if (!drainResult.drained) { throw new Error( - `Shared daemon pid=${currentStatus.pid} is still busy after ${drainTimeoutMs}ms; refusing to replace it.` + `Shared daemon pid=${liveStatus.pid} is still busy after ${drainTimeoutMs}ms; refusing to replace it.` ); } + // If another client replaced the daemon during the drain, do not stop + // their fresh daemon -- that is exactly the regression the drain was + // added to prevent. + if (drainResult.replacedByPeer) { + return; + } } await this.stop().catch(() => {}); await this.waitForStopped(); @@ -190,42 +200,75 @@ export class DaemonClient { }); } + /** + * Sends a raw `status` probe without the pid-match filter that + * `readVerifiedStatus` applies. Needed by callers that want to decide + * between acting on the current daemon, a peer's fresh daemon, and a + * missing daemon -- the verified variant collapses all three into null and + * loses the signal. + */ + private async probeLiveStatus(timeoutMs?: number): Promise { + const metadata = await readDaemonMetadata(this.metadataPath); + if (!metadata || metadata.socketPath !== this.socketPath || !isProcessRunning(metadata.pid)) { + return null; + } + try { + return (await this.sendRequest('status', {}, timeoutMs)) as StatusResult; + } catch (error) { + if (isTransportError(error)) { + return null; + } + throw error; + } + } + /** * Polls the live daemon's status until its `activeRequests` counter reads - * zero. Resolves `true` once the daemon is idle, `false` if the timeout - * expires, and `false` if the daemon disappears mid-wait so the caller - * falls through to the normal replacement path. + * zero. Returns both whether the drain condition was met AND whether the + * daemon got swapped out by another client mid-wait -- those are different + * outcomes that the caller needs to distinguish (a replaced daemon is + * "drained" but must not be stopped). + * + * `drained` is `true` once the daemon is idle, OR the daemon is gone (status + * probe fails), OR the daemon was replaced by another client (different + * pid). `drained` is `false` only if the timeout expires with the original + * daemon still advertising in-flight work. `replacedByPeer` is `true` only + * when a fresh daemon answered a status poll while we were waiting. * * A daemon that does not advertise `activeRequests` (pre-v2) is treated as * permanently busy -- the only safe read of "is this daemon still serving * a v1 request?" is to wait long enough for that request to finish, which * is exactly the drain we are trying to guarantee. */ - private async waitForDaemonIdle(seedStatus: StatusResult, timeoutMs: number): Promise { + private async waitForDaemonIdle( + seedStatus: StatusResult, + timeoutMs: number + ): Promise<{ drained: boolean; replacedByPeer: boolean }> { const deadline = Date.now() + timeoutMs; let status: StatusResult | null = seedStatus; + let lastSeenPid: number | null = seedStatus.pid; while (Date.now() < deadline) { if (!status) { - return true; + return { drained: true, replacedByPeer: false }; } const active = status.activeRequests; if (active === undefined) { - // Pre-v2 daemon: keep waiting. The drain timeout bounds the wait, so a - // stuck v1 daemon eventually falls through to the refusal branch and - // is left running rather than killed mid-request. await delay(DAEMON_DRAIN_POLL_MS); } else if (active <= 0) { - return true; + return { drained: true, replacedByPeer: lastSeenPid !== seedStatus.pid }; } else { await delay(DAEMON_DRAIN_POLL_MS); } - status = await this.readVerifiedStatus(); - if (status && status.pid !== seedStatus.pid) { - // Another client already replaced it; nothing to drain. - return true; + try { + status = await this.sendRequest('status', {}); + } catch { + return { drained: true, replacedByPeer: false }; + } + if (status) { + lastSeenPid = status.pid; } } - return false; + return { drained: false, replacedByPeer: lastSeenPid !== seedStatus.pid }; } private async startDaemon(options: { preflightTimeoutMs?: number } = {}): Promise { diff --git a/tests/daemon-client-config-stale.test.ts b/tests/daemon-client-config-stale.test.ts index d8e1e1b4..7065158c 100644 --- a/tests/daemon-client-config-stale.test.ts +++ b/tests/daemon-client-config-stale.test.ts @@ -13,6 +13,11 @@ let createConnection: ReturnType; // upgrading client is still draining -- the very window the new wait-for-idle // logic was introduced to cover. let drainAfterFirstStatusPoll = false; +// When set, the first `status` poll after the drain has resolved swaps the +// fake daemon's reported pid to a fresh one. Models another client winning +// the race to replace the busy daemon during the drain window. +let replacementPidAfterDrain: number | null = null; +let statusPollCount = 0; class MockSocket extends EventEmitter { setTimeout(): this { @@ -25,12 +30,30 @@ class MockSocket extends EventEmitter { if (payload.method === 'stop') { activeStatusPid = findNonRunningPid(); } - if (payload.method === 'status' && drainAfterFirstStatusPoll && activeRequests > 0) { - // Simulate a shared daemon whose in-flight request finishes after the - // upgrading client has noticed the staleness but before the drain - // timeout. The next status poll must see `activeRequests: 0` so the - // replacement is allowed to proceed. - activeRequests = 0; + if (payload.method === 'status') { + statusPollCount += 1; + // Defer the drain to the SECOND status poll so the first poll still + // reports the busy daemon -- otherwise probeLiveStatus would see + // `activeRequests: 0` up front and the drain check would be skipped. + if (drainAfterFirstStatusPoll && activeRequests > 0 && statusPollCount >= 2) { + // Simulate a shared daemon whose in-flight request finishes after the + // upgrading client has noticed the staleness but before the drain + // timeout. The next status poll must see `activeRequests: 0` so the + // replacement is allowed to proceed. + activeRequests = 0; + } + // Simulate another client replacing the daemon during the drain. The + // swap happens on the same poll the drain resolves so the upgrading + // client observes the busy daemon, then the replacement -- the + // sequence the drain check was added to handle. + if ( + replacementPidAfterDrain !== null && + activeRequests === 0 && + statusPollCount >= 2 && + activeStatusPid !== replacementPidAfterDrain + ) { + activeStatusPid = replacementPidAfterDrain; + } } const response = buildResponse(payload.method, payload.id); queueMicrotask(() => { @@ -112,6 +135,8 @@ describe('DaemonClient config freshness', () => { activeLayers = []; activeRequests = 0; drainAfterFirstStatusPoll = false; + replacementPidAfterDrain = null; + statusPollCount = 0; launchDaemonDetached.mockClear(); launchDaemonDetached.mockImplementation( (options: { metadataPath: string; socketPath: string; configPath: string }) => { @@ -373,6 +398,57 @@ describe('DaemonClient config freshness', () => { expect(sentMethods).toContain('listTools'); }); + it('does not stop a peer fresh daemon if it replaces the busy one mid-drain', async () => { + const tmpDir = await makeShortTempDir('daemon-busy-replaced'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + + const configPath = path.join(tmpDir, 'config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const stat = await fs.stat(configPath); + const originalPid = process.pid; + const replacementPid = findNonRunningPid(); + const { metadataPath, socketPath } = resolveDaemonPaths(configPath); + activeConfigPath = configPath; + activeSocketPath = socketPath; + activeConfigMtime = stat.mtimeMs; + activeStatusPid = originalPid; + activeLayers = [{ path: configPath, mtimeMs: stat.mtimeMs }]; + // The busy peer drains to idle on the first poll, then another client + // races in and replaces the daemon before our post-drain re-check runs. + activeRequests = 1; + drainAfterFirstStatusPoll = true; + replacementPidAfterDrain = replacementPid; + + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: originalPid, + protocolVersion: DAEMON_PROTOCOL_VERSION - 1, + socketPath, + configPath, + startedAt: Date.now() - 10_000, + logPath: null, + configMtimeMs: stat.mtimeMs, + configLayers: activeLayers, + }), + 'utf8' + ); + + const client = new DaemonClient({ configPath, configExplicit: true, rootDir: tmpDir }); + await client.listTools({ server: 'playwright' }); + + // The peer's fresh daemon must not have received `stop` -- only the + // request that the upgrading client actually wanted to make. Killing the + // replacement would be the same regression the drain was added to stop. + expect(sentMethods).not.toContain('stop'); + // The replacing pid appeared in the status poll, proving the upgrade + // window was actually exercised. + expect(activeStatusPid).toBe(replacementPid); + expect(launchDaemonDetached).not.toHaveBeenCalled(); + expect(sentMethods).toContain('listTools'); + }); + it('refuses to replace a daemon that never drains before the timeout', async () => { const tmpDir = await makeShortTempDir('daemon-busy-stuck'); process.env.MCPORTER_DAEMON_DIR = tmpDir; From fb1901d325fa7ae4c84438b1b791e2b8d49b3846 Mon Sep 17 00:00:00 2001 From: Umut Keltek Date: Fri, 31 Jul 2026 18:55:08 +0300 Subject: [PATCH 6/6] fix(daemon): do not stop a peer daemon that was already replaced The drain handles the case where a peer swaps in a fresh daemon *while* the upgrading client is waiting. A subtler race happens *before* the drain even starts: a peer has already replaced the busy daemon, but the on-disk metadata still names the old PID, so the previous early-return check (which also required a fresh config) did not fire and the code fell through to stop() against the peer's fresh daemon. Make the pid-mismatch short-circuit in restartDaemon unconditional: if the live daemon's pid does not match the expected one, do not stop it, whether or not the metadata still reads stale. The peer owns the replacement; the upgrading client just uses whatever daemon is live. This subsumes the pid-mismatch + fresh-config fast path, which is removed rather than stacked on. The transport-error path is unaffected because expectedPid is undefined there and the guard is skipped, so the client still self-heals when no daemon is responding. A new regression test pins the scenario: a peer wins the swap before the drain starts, the upgrading client must not call stop, and the peer's daemon must keep serving. --- src/daemon/client.ts | 15 ++++--- tests/daemon-client-config-stale.test.ts | 55 ++++++++++++++++++++++++ 2 files changed, 63 insertions(+), 7 deletions(-) diff --git a/src/daemon/client.ts b/src/daemon/client.ts index 26aa300b..2a3072c8 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -163,15 +163,16 @@ export class DaemonClient { // would silently return null on a pid mismatch and the busy-daemon // regression would slip past us. const liveStatus = await this.probeLiveStatus(); - if ( - liveStatus && - options.expectedPid !== undefined && - liveStatus.pid !== options.expectedPid && - (await this.checkConfigState()) === 'fresh' - ) { + if (options.reason === 'stale-config' && liveStatus && (await this.checkConfigState()) === 'fresh') { return; } - if (options.reason === 'stale-config' && liveStatus && (await this.checkConfigState()) === 'fresh') { + // If the daemon has already been replaced by another client (PID + // mismatch before the drain even starts), the replacement is theirs to + // own -- stopping it would kill their in-flight request. The config + // check above is not enough: the peer may not have rewritten the + // metadata yet, so `checkConfigState` can still report `stale` and the + // condition would fall through to `stop()`. + if (liveStatus && options.expectedPid !== undefined && liveStatus.pid !== options.expectedPid) { return; } // Codex / ClawSweeper P1: an upgraded client must not stop a daemon that diff --git a/tests/daemon-client-config-stale.test.ts b/tests/daemon-client-config-stale.test.ts index 7065158c..677e9582 100644 --- a/tests/daemon-client-config-stale.test.ts +++ b/tests/daemon-client-config-stale.test.ts @@ -398,6 +398,61 @@ describe('DaemonClient config freshness', () => { expect(sentMethods).toContain('listTools'); }); + it('does not stop a peer fresh daemon if it replaced the busy one before the drain starts', async () => { + // ClawSweeper P1, second race: the metadata says the old (busy) daemon is + // running, but a peer has *already* replaced it by the time the upgrading + // client enters restartDaemon. The on-disk metadata is stale, so the + // config-freshness check alone does not stop the upgrading client from + // issuing `stop` against the peer's fresh daemon. The fix is an explicit + // pid-mismatch short-circuit: never stop a daemon that is not the one + // the metadata named. + const tmpDir = await makeShortTempDir('daemon-already-replaced'); + process.env.MCPORTER_DAEMON_DIR = tmpDir; + + const configPath = path.join(tmpDir, 'config.json'); + await fs.writeFile(configPath, JSON.stringify({ mcpServers: {} }), 'utf8'); + const stat = await fs.stat(configPath); + const originalPid = process.pid; + const replacementPid = findNonRunningPid(); + const { metadataPath, socketPath } = resolveDaemonPaths(configPath); + activeConfigPath = configPath; + activeSocketPath = socketPath; + activeConfigMtime = stat.mtimeMs; + // The peer won the race: the live daemon is already the fresh one. The + // metadata still names the original (busy) PID because the upgrading + // client has not yet observed the swap. + activeStatusPid = replacementPid; + activeLayers = [{ path: configPath, mtimeMs: stat.mtimeMs }]; + + await fs.mkdir(path.dirname(metadataPath), { recursive: true }); + await fs.writeFile( + metadataPath, + JSON.stringify({ + pid: originalPid, + protocolVersion: DAEMON_PROTOCOL_VERSION - 1, + socketPath, + configPath, + startedAt: Date.now() - 10_000, + logPath: null, + configMtimeMs: stat.mtimeMs, + configLayers: activeLayers, + }), + 'utf8' + ); + + const client = new DaemonClient({ configPath, configExplicit: true, rootDir: tmpDir }); + await client.listTools({ server: 'playwright' }); + + // The peer's fresh daemon must not have received `stop` -- that is the + // regression the pid-mismatch short-circuit exists to prevent. The + // upgrading client is now using the peer's daemon; if its config does + // not match the upgrading client's, that is the peer's problem to + // resolve, not the upgrading client's to escalate by killing them. + expect(sentMethods).not.toContain('stop'); + expect(launchDaemonDetached).not.toHaveBeenCalled(); + expect(sentMethods).toContain('listTools'); + }); + it('does not stop a peer fresh daemon if it replaces the busy one mid-drain', async () => { const tmpDir = await makeShortTempDir('daemon-busy-replaced'); process.env.MCPORTER_DAEMON_DIR = tmpDir;