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..5747b972 100644 --- a/src/daemon/client.ts +++ b/src/daemon/client.ts @@ -5,6 +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, + DaemonFrameDecoder, + isDaemonProgressFrame, + resolveProgressInterval, +} from './protocol.js'; import type { CallToolParams, CloseServerParams, @@ -34,6 +40,7 @@ export interface DaemonPaths { interface DaemonMetadata { readonly pid: number; + readonly protocolVersion?: number; readonly socketPath: string; readonly configPath: string; readonly configMtimeMs?: number | null; @@ -69,7 +76,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 +239,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 }] @@ -250,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) { @@ -267,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) { @@ -293,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; @@ -328,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 57c6e03c..efb7faa4 100644 --- a/src/daemon/host.ts +++ b/src/daemon/host.ts @@ -1,3 +1,5 @@ +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'; import net from 'node:net'; @@ -6,6 +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, 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 { @@ -16,15 +20,22 @@ import { logEvent, shouldLogServer, } from './log-context.js'; -import type { - CallToolParams, - CloseServerParams, - DaemonRequest, - DaemonResponse, - ListResourcesParams, - ListToolsParams, - ReadResourceParams, - StatusResult, +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, + type ListToolsParams, + type ReadResourceParams, + type StatusResult, } from './protocol.js'; import { buildErrorResponse, @@ -163,6 +174,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 +225,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 +281,7 @@ function metadataFromStatus( fallbackConfigLayers: Array<{ path: string; mtimeMs: number | null }> ): { pid: number; + protocolVersion: number; socketPath: string; configPath: string; configLayers?: StatusResult['configLayers']; @@ -278,6 +292,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 +310,9 @@ function daemonConfigMatches( currentConfigMtimeMs: number | null, currentDefinitionHash: string ): boolean { + if (live.protocolVersion !== DAEMON_PROTOCOL_VERSION) { + return false; + } if (live.definitionHash !== currentDefinitionHash) { return false; } @@ -351,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) { @@ -367,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)); }); @@ -387,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; @@ -402,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) { @@ -413,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)); }); } @@ -483,6 +503,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; @@ -491,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(); @@ -509,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 @@ -516,6 +576,40 @@ 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 ( + 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 +620,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; @@ -564,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}`); @@ -596,6 +694,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) { @@ -618,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}`); @@ -644,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}`); @@ -669,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}`); @@ -689,6 +794,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 +828,7 @@ async function processRequest( } } catch (error) { return { - response: buildErrorResponse(id, 'runtime_error', error), + response: buildErrorResponse(id, daemonRuntimeErrorCode(error), error), shouldShutdown: false, }; } @@ -748,6 +854,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 +862,45 @@ 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 + ); +} + +/** + * 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 925fc46a..df25a291 100644 --- a/src/daemon/protocol.ts +++ b/src/daemon/protocol.ts @@ -1,3 +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' @@ -11,6 +38,9 @@ export interface DaemonRequest { @@ -23,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; @@ -37,6 +133,7 @@ export interface ListToolsParams { readonly autoAuthorize?: boolean; readonly allowCachedAuth?: boolean; readonly disableOAuth?: boolean; + readonly timeoutMs?: number; } export interface ListResourcesParams { @@ -59,6 +156,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..e427010c 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,32 @@ 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; + // 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 { - 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, @@ -275,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 @@ -580,7 +607,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..1af4a36a 100644 --- a/tests/daemon-client-timeout.test.ts +++ b/tests/daemon-client-timeout.test.ts @@ -2,15 +2,27 @@ 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; + private progressHandle?: NodeJS.Timeout; + // 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 (this.timeoutHandle) { + clearTimeout(this.timeoutHandle); + this.timeoutHandle = undefined; + } + if (enforceSocketTimeout && ms > 0) { + this.timeoutHandle = setTimeout(() => this.emit('timeout'), ms); + } return this; } @@ -18,7 +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(() => { + this.clearTimers(); this.emit('data', JSON.stringify(response)); this.emit('end'); }, responseDelayMs); @@ -31,12 +49,29 @@ class MockSocket extends EventEmitter { return this; } - destroy(): this { + destroy(error?: Error): this { + 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(() => { @@ -67,6 +102,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 +121,8 @@ describe('DaemonClient timeouts', () => { beforeEach(async () => { 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'); @@ -142,6 +180,47 @@ 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'); + // 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); + 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 +242,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/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/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..616d43c5 --- /dev/null +++ b/tests/runtime-listtools-timeout.test.ts @@ -0,0 +1,84 @@ +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('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: [] }); + 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); + }); +});