From cae35961e53569cc464c7accb4b0518701277f75 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Mon, 31 Aug 2026 21:32:11 -0600 Subject: [PATCH 01/12] fix(mcp): route subscriptions to SSE owner (#1751) --- components/mcp/lifecycle.ts | 4 +- components/mcp/session.ts | 16 +- components/mcp/sessionRegistry.ts | 4 + components/mcp/subscriptionRouting.ts | 261 ++++++++++++++++++ components/mcp/transport.ts | 57 ++-- .../fixtures/custom-resources/resources.js | 10 + integrationTests/mcp/sse-listchanged.test.ts | 213 +++++++++++++- .../components/mcp/adapters/fastify.test.js | 3 + .../mcp/adapters/harperHttp.test.js | 3 + unitTests/components/mcp/lifecycle.test.js | 3 + unitTests/components/mcp/session.test.js | 13 +- .../mcp/subscriptionRouting.test.js | 199 +++++++++++++ unitTests/components/mcp/transport.test.js | 19 +- utility/hdbTerms.ts | 2 + 14 files changed, 755 insertions(+), 52 deletions(-) create mode 100644 components/mcp/subscriptionRouting.ts create mode 100644 unitTests/components/mcp/subscriptionRouting.test.js diff --git a/components/mcp/lifecycle.ts b/components/mcp/lifecycle.ts index aafcca9ef9..b64cf98c26 100644 --- a/components/mcp/lifecycle.ts +++ b/components/mcp/lifecycle.ts @@ -7,7 +7,7 @@ * server SHOULD respond with its preferred supported version so the client * can decide whether to connect on the older version or disconnect. */ -import { createSession, saveSession, type McpSessionRecord } from './session.ts'; +import { createSession, patchSession, type McpSessionRecord } from './session.ts'; import { packageJson } from '../../utility/packageUtils.js'; export const PROTOCOL_VERSION_PREFERRED = '2025-06-18'; @@ -106,6 +106,6 @@ export async function handleInitialize( export async function handleInitialized(session: McpSessionRecord): Promise { if (session.initialized) return session; const updated: McpSessionRecord = { ...session, initialized: true }; - await saveSession(updated); + await patchSession(session.id, { initialized: true }); return updated; } diff --git a/components/mcp/session.ts b/components/mcp/session.ts index b537cc34ae..46a6674e1b 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -3,7 +3,7 @@ * * Eviction is delegated to Harper's native TTL (`Table.setTTLExpiration`): * every write to a session record updates its `version`, which Harper uses - * to determine expiration. So calling `saveSession(record)` on each request + * to determine expiration. So calling `patchSession(id, changes)` on each request * gives sliding-window idle semantics for free — no custom timer, no sweep. * * Spec: when a request bears an `Mcp-Session-Id` the server doesn't @@ -63,6 +63,8 @@ export interface McpSessionRecord { * to clients that declared support. Undefined = client declared none. */ clientCapabilities?: Record; + /** Node-local hint for the worker that owns the current GET-SSE stream. */ + streamOwner?: { threadId: number; token: string }; } let _sessionTable: Table | undefined; @@ -90,6 +92,7 @@ function declareSessionTable(): Table { { name: 'logLevel' }, { name: 'subscriptions' }, { name: 'clientCapabilities' }, + { name: 'streamOwner' }, ], }); } @@ -149,12 +152,9 @@ export async function loadSession(id: string): Promise return record; } -/** - * Persist updated session state. Used to bump `lastActivity` (sliding-window - * idle reset) and to flip `initialized` after `notifications/initialized`. - */ -export async function saveSession(record: McpSessionRecord): Promise { - await (getTable() as any).put(record); +/** Incrementally update session fields without replacing concurrent changes. */ +export async function patchSession(id: string, changes: Partial>): Promise { + await (getTable() as any).patch({ id, ...changes }); } export async function deleteSession(id: string): Promise { @@ -174,6 +174,6 @@ export async function deleteSession(id: string): Promise { */ export async function touchSession(record: McpSessionRecord): Promise { const touched: McpSessionRecord = { ...record, lastActivity: Date.now() }; - await saveSession(touched); + await patchSession(record.id, { lastActivity: touched.lastActivity }); return touched; } diff --git a/components/mcp/sessionRegistry.ts b/components/mcp/sessionRegistry.ts index 9c10840930..308ed723e9 100644 --- a/components/mcp/sessionRegistry.ts +++ b/components/mcp/sessionRegistry.ts @@ -17,6 +17,7 @@ * without a graceful close). */ import { IterableEventQueue } from '../../resources/IterableEventQueue.ts'; +import { randomUUID } from 'node:crypto'; import type { McpLogLevel } from './logging.ts'; import type { AuthedUser } from './toolRegistry.ts'; import type { McpProfile } from './transport.ts'; @@ -29,6 +30,8 @@ export interface SseEvent { export interface RegisteredSession { sessionId: string; + /** Uniquely identifies this particular GET-SSE stream. */ + streamToken: string; profile: McpProfile; user: AuthedUser; queue: IterableEventQueue; @@ -98,6 +101,7 @@ export function registerSession(sessionId: string, profile: McpProfile, user: Au const queue = new IterableEventQueue(); const record: RegisteredSession = { sessionId, + streamToken: randomUUID(), profile, user, queue, diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts new file mode 100644 index 0000000000..99d73e94ef --- /dev/null +++ b/components/mcp/subscriptionRouting.ts @@ -0,0 +1,261 @@ +/** Route MCP resource-subscription commands to the worker owning the GET-SSE stream. */ +import { randomUUID } from 'node:crypto'; +import { threadId } from 'node:worker_threads'; +import { ITC_EVENT_TYPES } from '../../utility/hdbTerms.ts'; +import harperLogger from '../../utility/logging/harper_logger.ts'; +import { loadSession, patchSession, type McpSessionRecord } from './session.ts'; +import { getRegisteredSession } from './sessionRegistry.ts'; +import { addResourceSubscription, removeResourceSubscription } from './subscriptions.ts'; +import type { AuthedUser } from './toolRegistry.ts'; + +const DEFAULT_RESPONSE_TIMEOUT_MS = 2_000; +const MAX_PENDING = 100; +const MAX_PENDING_PER_SESSION = 25; + +export type SubscriptionRouteResult = 'success' | 'not-subscribable' | 'no-live-stream' | 'internal-error'; +type Operation = 'subscribe' | 'unsubscribe'; + +interface Command { + requestId: string; + originator: number; + sessionId: string; + streamToken: string; + operation: Operation; + uri: string; + user?: AuthedUser; +} + +interface Response { + requestId: string; + originator: number; + result: SubscriptionRouteResult; +} + +interface ItcBridge { + sendToThread(threadId: number, event: { type: string; message: unknown }): boolean; + onMessageByType(type: string, listener: (event: { message?: unknown }) => void): void; +} + +// manageThreads assigns this connected-port array as the package-global `threads` export. Its +// direct-send helper lives on that array, while typed listener registration is a module export. +declare const threads: { sendToThread(threadId: number, event: { type: string; message: unknown }): boolean }; + +interface Pending { + sessionId: string; + targetThreadId: number; + resolve: (result: SubscriptionRouteResult) => void; + timer: ReturnType; +} + +const pending = new Map(); +const operationChains = new Map>(); +let wired = false; +let bridgeOverride: ItcBridge | undefined; +let currentThreadId = (): number => threadId; +let responseTimeoutMs = DEFAULT_RESPONSE_TIMEOUT_MS; + +function bridge(): ItcBridge { + if (bridgeOverride) return bridgeOverride; + const { onMessageByType } = require('../../server/threads/manageThreads.js'); + return { sendToThread: threads.sendToThread.bind(threads), onMessageByType }; +} + +export function _setSubscriptionItcForTest(fake: ItcBridge | undefined): void { + bridgeOverride = fake; + wired = false; +} + +export function _setSubscriptionThreadIdForTest(value: number | undefined): void { + currentThreadId = value === undefined ? () => threadId : () => value; +} + +export function _setSubscriptionTimeoutForTest(value: number | undefined): void { + responseTimeoutMs = value ?? DEFAULT_RESPONSE_TIMEOUT_MS; +} + +export function _resetSubscriptionRoutingForTest(): void { + for (const entry of pending.values()) clearTimeout(entry.timer); + pending.clear(); + operationChains.clear(); + wired = false; + currentThreadId = () => threadId; + responseTimeoutMs = DEFAULT_RESPONSE_TIMEOUT_MS; +} + +export function _pendingSubscriptionRouteCount(): number { + return pending.size; +} + +function ensureWired(): void { + if (wired) return; + wired = true; + bridge().onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { + const command = event.message as Command; + void handleCommand(command).catch((error) => { + harperLogger.error('MCP subscription command failed', error); + sendResponse(command, 'internal-error'); + }); + }); + bridge().onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, (event) => { + const response = event.message as Response; + const entry = pending.get(response?.requestId); + if (!entry || response.originator !== entry.targetThreadId) return; + if (!['success', 'not-subscribable', 'no-live-stream', 'internal-error'].includes(response.result)) return; + clearTimeout(entry.timer); + pending.delete(response.requestId); + entry.resolve(response.result); + }); +} + +/** Attach the owner-side command listener and persist the routing hint for a new GET stream. */ +export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { + ensureWired(); + await patchSession(sessionId, { streamOwner: { threadId: currentThreadId(), token: streamToken } }); +} + +function countPendingForSession(sessionId: string): number { + let count = 0; + for (const entry of pending.values()) if (entry.sessionId === sessionId) count++; + return count; +} + +function subscriptionUser(user: AuthedUser): AuthedUser { + return { + ...(user.username ? { username: user.username } : {}), + ...(user._scopedToken ? { _scopedToken: true } : {}), + ...(user.role + ? { + role: { + ...(user.role.role ? { role: user.role.role } : {}), + ...(user.role.permission ? { permission: user.role.permission } : {}), + }, + } + : {}), + }; +} + +function routeRemote( + owner: NonNullable, + command: Omit +): Promise { + ensureWired(); + if (pending.size >= MAX_PENDING || countPendingForSession(command.sessionId) >= MAX_PENDING_PER_SESSION) { + return Promise.resolve('internal-error'); + } + const requestId = randomUUID(); + return new Promise((resolve) => { + const timer = setTimeout(() => { + pending.delete(requestId); + resolve('no-live-stream'); + }, responseTimeoutMs); + timer.unref(); + pending.set(requestId, { sessionId: command.sessionId, targetThreadId: owner.threadId, resolve, timer }); + let sent = false; + try { + sent = bridge().sendToThread(owner.threadId, { + type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, + message: { ...command, requestId, originator: currentThreadId(), streamToken: owner.token }, + }); + } catch (error) { + harperLogger.error('Unable to route MCP subscription command', error); + } + if (!sent) { + clearTimeout(timer); + pending.delete(requestId); + resolve('no-live-stream'); + } + }); +} + +function serializeSessionOperation(sessionId: string, operation: () => Promise): Promise { + const previous = operationChains.get(sessionId) ?? Promise.resolve(); + const current = previous.then(operation, operation); + const tail = current + .then( + () => undefined, + () => undefined + ) + .finally(() => { + if (operationChains.get(sessionId) === tail) operationChains.delete(sessionId); + }); + operationChains.set(sessionId, tail); + return current; +} + +async function executeLocal(command: Command): Promise { + const registered = getRegisteredSession(command.sessionId); + if (!registered || registered.streamToken !== command.streamToken) return 'no-live-stream'; + return serializeSessionOperation(command.sessionId, async () => { + if (command.operation === 'subscribe') { + if (!command.user) return 'internal-error'; + const added = await addResourceSubscription(command.sessionId, command.uri, command.user); + if (!added) return 'not-subscribable'; + try { + const session = await loadSession(command.sessionId); + if (!session) { + removeResourceSubscription(command.sessionId, command.uri); + return 'no-live-stream'; + } + if (!session.subscriptions?.includes(command.uri)) { + await patchSession(command.sessionId, { + subscriptions: [...(session.subscriptions ?? []), command.uri], + }); + } + return 'success'; + } catch (error) { + removeResourceSubscription(command.sessionId, command.uri); + throw error; + } + } + const session = await loadSession(command.sessionId); + if (session?.subscriptions?.includes(command.uri)) { + await patchSession(command.sessionId, { + subscriptions: session.subscriptions.filter((uri) => uri !== command.uri), + }); + } + removeResourceSubscription(command.sessionId, command.uri); + return 'success'; + }); +} + +async function handleCommand(command: Command): Promise { + let result: SubscriptionRouteResult; + try { + result = await executeLocal(command); + } catch (error) { + harperLogger.error('MCP subscription owner failed to execute command', error); + result = 'internal-error'; + } + sendResponse(command, result); +} + +function sendResponse(command: Command, result: SubscriptionRouteResult): void { + try { + bridge().sendToThread(command.originator, { + type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, + message: { requestId: command.requestId, originator: currentThreadId(), result } satisfies Response, + }); + } catch (error) { + harperLogger.trace(`Unable to return MCP subscription response: ${(error as Error).message}`); + } +} + +export async function routeResourceSubscription(args: { + session: McpSessionRecord; + operation: Operation; + uri: string; + user?: AuthedUser; +}): Promise { + const owner = args.session.streamOwner; + if (!owner) return 'no-live-stream'; + const command = { + sessionId: args.session.id, + operation: args.operation, + uri: args.uri, + ...(args.user ? { user: subscriptionUser(args.user) } : {}), + }; + if (owner.threadId === currentThreadId()) { + return executeLocal({ ...command, requestId: '', originator: currentThreadId(), streamToken: owner.token }); + } + return routeRemote(owner, command); +} diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index f5187ec2bb..eb742108d5 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -32,16 +32,12 @@ import { decodeCursor } from './pagination.ts'; import { seedSessionSnapshot } from './listChanged.ts'; import { tryAdmit, resolveClientIdentity } from './rateLimit.ts'; import { checkDurableQuota } from './quota.ts'; -import { deleteSession, loadSession, saveSession, touchSession, type McpSessionRecord } from './session.ts'; +import { deleteSession, loadSession, patchSession, touchSession, type McpSessionRecord } from './session.ts'; import { listResources, listResourceTemplates, readResource, completeResourceArgument } from './resources.ts'; import { ensureApplicationToolsFresh } from './tools/application.ts'; import { getPrompt, listPrompts, completePromptArgument } from './promptRegistry.ts'; -import { - addResourceSubscription, - removeResourceSubscription, - dropSessionSubscriptions, - restoreResourceSubscriptions, -} from './subscriptions.ts'; +import { dropSessionSubscriptions, restoreResourceSubscriptions } from './subscriptions.ts'; +import { claimSubscriptionOwner, routeResourceSubscription } from './subscriptionRouting.ts'; import { sendServerRequest, routeClientResponse, @@ -50,8 +46,8 @@ import { } from './serverRequests.ts'; import { registerSession, + unregisterSession, touchRegisteredSession, - getRegisteredSession, replaySince, type SseEvent, } from './sessionRegistry.ts'; @@ -370,7 +366,7 @@ async function dispatchSetLevel( // on the worker where the change fires. Cross-worker push is a separate, // subsystem-wide design item (tracked in the MCP design-doc issue). session.logLevel = level; - await saveSession(session); + await patchSession(session.id, { logLevel: level }); setSessionLogLevel(session.id, level); return jsonResponse(200, buildSuccess(messageId, {})); } @@ -403,6 +399,12 @@ async function handleGet(request: NormRequest): Promise { }; } const record = registerSession(sessionId, request.profile, effectiveUser(request)); + try { + await claimSubscriptionOwner(sessionId, record.streamToken); + } catch (error) { + unregisterSession(sessionId); + throw error; + } // Seed the live record with any previously-set logging level so a reconnect // (or a setLevel that preceded this stream) keeps delivering notifications/message. // (A fresh record's logLevel is already undefined, so a direct assign is safe.) @@ -420,7 +422,7 @@ async function handleGet(request: NormRequest): Promise { const restored = await restoreResourceSubscriptions(sessionId, session.subscriptions, effectiveUser(request)); if (restored.length !== session.subscriptions.length) { session.subscriptions = restored; - await saveSession(session); + await patchSession(session.id, { subscriptions: restored }); } } // Resumability (#3.8): on reconnect with Last-Event-ID, replay buffered frames @@ -917,26 +919,25 @@ async function dispatchResourcesSubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/subscribe requires params.uri') ); } - // Require a live GET SSE stream: that's where notifications/resources/updated is - // delivered, and its 'close' is the only teardown hook for the subscription. A - // subscription opened without a stream would leak its audit-log iterator and - // drop every update silently. - if (!getRegisteredSession(session.id)) { + const result = await routeResourceSubscription({ + session, + operation: 'subscribe', + uri, + user: effectiveUser(request), + }); + if (result === 'no-live-stream') { return jsonResponse( 200, buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'open the GET SSE stream before subscribing to resources') ); } - const ok = await addResourceSubscription(session.id, uri, effectiveUser(request)); - if (!ok) { + if (result === 'not-subscribable') { // Only row-backed application resources are subscribable; synthetic harper://* // URIs (and unknown URIs) have no change source. return jsonResponse(200, buildError(messageId, ERROR_CODES.INVALID_PARAMS, `resource is not subscribable: ${uri}`)); } - // Persist the URI on the durable record so it survives an SSE reconnect. - if (!session.subscriptions?.includes(uri)) { - session.subscriptions = [...(session.subscriptions ?? []), uri]; - await saveSession(session); + if (result === 'internal-error') { + return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource subscription failed')); } return jsonResponse(200, buildSuccess(messageId, {})); } @@ -955,10 +956,16 @@ async function dispatchResourcesUnsubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/unsubscribe requires params.uri') ); } - removeResourceSubscription(session.id, uri); - if (session.subscriptions?.includes(uri)) { - session.subscriptions = session.subscriptions.filter((u) => u !== uri); - await saveSession(session); + const result = await routeResourceSubscription({ session, operation: 'unsubscribe', uri }); + if (result === 'no-live-stream') { + // The live owner is already gone. Remove durable state locally so a later + // reconnect cannot restore the cancelled subscription. + const fresh = await loadSession(session.id); + if (fresh?.subscriptions?.includes(uri)) { + await patchSession(session.id, { subscriptions: fresh.subscriptions.filter((u) => u !== uri) }); + } + } else if (result === 'internal-error') { + return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe failed')); } return jsonResponse(200, buildSuccess(messageId, {})); } diff --git a/integrationTests/fixtures/custom-resources/resources.js b/integrationTests/fixtures/custom-resources/resources.js index 547c278df6..f40db0703a 100644 --- a/integrationTests/fixtures/custom-resources/resources.js +++ b/integrationTests/fixtures/custom-resources/resources.js @@ -1,3 +1,13 @@ +import { threadId } from 'node:worker_threads'; + +/** Test-only endpoint used to pin HTTP requests to distinct keep-alive worker connections. */ +export class WorkerIdentity extends Resource { + static loadAsInstance = false; + get() { + return { threadId }; + } +} + // WorkItem: async write-then-patch pattern (CDI RT enqueueing + AI inference result attachment) export class WorkItem extends tables.WorkItem { // Author-opt-in custom MCP surface declared on a subclass of an exported @table (#1448). diff --git a/integrationTests/mcp/sse-listchanged.test.ts b/integrationTests/mcp/sse-listchanged.test.ts index cc9d1ef48c..b97b7fb2ac 100644 --- a/integrationTests/mcp/sse-listchanged.test.ts +++ b/integrationTests/mcp/sse-listchanged.test.ts @@ -20,6 +20,7 @@ import { suite, test, before, after } from 'node:test'; import { ok, strictEqual } from 'node:assert'; import { resolve } from 'node:path'; +import { Agent, request as httpRequest } from 'node:http'; import { setupHarperWithFixture, teardownHarper, type ContextWithHarper } from '@harperfast/integration-testing'; @@ -114,6 +115,65 @@ async function mcpCall( return res.json(); } +async function requestWithAgent( + baseUrl: string, + agent: Agent, + path: string, + options: { method?: string; headers?: Record; body?: string; localPort?: number } = {} +): Promise<{ status: number; body: string }> { + return new Promise((resolveRequest, reject) => { + const request = httpRequest( + new URL(path, baseUrl), + { method: options.method ?? 'GET', headers: options.headers, agent, localPort: options.localPort }, + (response) => { + const chunks: Buffer[] = []; + response.on('data', (chunk) => chunks.push(Buffer.from(chunk))); + response.on('end', () => + resolveRequest({ status: response.statusCode ?? 0, body: Buffer.concat(chunks).toString() }) + ); + } + ); + request.on('error', reject); + if (options.body) request.end(options.body); + else request.end(); + }); +} + +async function workerIdentity(baseUrl: string, auth: string, agent: Agent, localPort: number): Promise { + const response = await requestWithAgent(baseUrl, agent, '/WorkerIdentity', { + headers: { authorization: auth }, + localPort, + }); + strictEqual(response.status, 200, `WorkerIdentity failed: ${response.status} ${response.body}`); + return JSON.parse(response.body).threadId; +} + +async function mcpCallWithAgent( + baseUrl: string, + auth: string, + agent: Agent, + session: { sessionId: string; protocolVersion: string }, + method: string, + params: Record, + id: number, + localPort: number +): Promise { + const response = await requestWithAgent(baseUrl, agent, '/mcp', { + method: 'POST', + headers: { + 'content-type': 'application/json', + 'accept': 'application/json, text/event-stream', + 'authorization': auth, + 'mcp-session-id': session.sessionId, + 'mcp-protocol-version': session.protocolVersion, + }, + body: JSON.stringify({ jsonrpc: '2.0', id, method, params }), + localPort, + }); + strictEqual(response.status, 200, `MCP ${method} failed: ${response.status} ${response.body}`); + return JSON.parse(response.body); +} + /** Upsert a WorkItem record by id through the application REST API. */ async function putWorkItem(ctx: ContextWithHarper, id: string, body: Record): Promise { const res = await fetch(new URL(`/WorkItem/${id}`, ctx.harper.httpURL), { @@ -223,10 +283,102 @@ async function openSse( }; } +/** Open SSE on the sole socket in `agent`, pinning the stream to that socket's worker. */ +async function openSseWithAgent( + baseUrl: string, + auth: string, + agent: Agent, + session: { sessionId: string; protocolVersion: string }, + localPort: number, + headerTimeoutMs = 2500 +): Promise<{ + status: number; + next: (predicate: (msg: any) => boolean, timeoutMs: number) => Promise; + close: () => void; +}> { + let request; + const response = (await Promise.race([ + new Promise((resolveResponse, reject) => { + request = httpRequest( + new URL('/mcp', baseUrl), + { + method: 'GET', + agent, + localPort, + headers: { + 'accept': 'text/event-stream', + 'authorization': auth, + 'mcp-session-id': session.sessionId, + 'mcp-protocol-version': session.protocolVersion, + }, + }, + resolveResponse + ); + request.on('error', reject); + request.end(); + }), + new Promise((_, reject) => + setTimeout(() => reject(new Error('pinned SSE headers never flushed (hung)')), headerTimeoutMs) + ), + ])) as import('node:http').IncomingMessage; + + let buffer = ''; + const parsed: any[] = []; + const waiters: Array<{ predicate: (message: any) => boolean; resolve: (message: any) => void }> = []; + response.on('data', (chunk) => { + buffer += chunk.toString(); + let index; + while ((index = buffer.indexOf('\n\n')) !== -1) { + const frame = buffer.slice(0, index); + buffer = buffer.slice(index + 2); + const dataLine = frame.split('\n').find((line) => line.startsWith('data:')); + if (!dataLine) continue; + let message; + try { + message = JSON.parse(dataLine.slice(5).trim()); + } catch { + continue; + } + parsed.push(message); + for (let i = waiters.length - 1; i >= 0; i--) { + if (waiters[i].predicate(message)) { + waiters[i].resolve(message); + waiters.splice(i, 1); + } + } + } + }); + + return { + status: response.statusCode ?? 0, + next(predicate, timeoutMs) { + const existing = parsed.find(predicate); + if (existing) return Promise.resolve(existing); + return new Promise((resolveNext) => { + const timer = setTimeout(() => resolveNext(undefined), timeoutMs); + waiters.push({ + predicate, + resolve: (message) => { + clearTimeout(timer); + resolveNext(message); + }, + }); + }); + }, + close() { + response.destroy(); + request?.destroy(); + }, + }; +} + suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => { before(async () => { await setupHarperWithFixture(ctx, FIXTURE_PATH, { - config: { mcp: { operations: { mountPath: '/mcp' }, application: { mountPath: '/mcp' } } }, + config: { + threads: { count: 4 }, + mcp: { operations: { mountPath: '/mcp' }, application: { mountPath: '/mcp' } }, + }, env: {}, }); }); @@ -318,4 +470,63 @@ suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => sse.close(); } }); + + test('N5: resource subscription routes from a sibling POST worker to the GET-SSE owner', async (t) => { + if (process.platform === 'win32') { + t.skip('Harper forces one HTTP worker on Windows'); + return; + } + const auth = adminAuth(ctx); + const session = await initialize(ctx.harper.httpURL, auth); + const firstLocalPort = 30000 + (process.pid % 10000); + const getAgent = new Agent({ keepAlive: true, maxSockets: 1 }); + const getThreadId = await workerIdentity(ctx.harper.httpURL, auth, getAgent, firstLocalPort); + const sse = await openSseWithAgent(ctx.harper.httpURL, auth, getAgent, session, firstLocalPort); + let postAgent; + let postThreadId = getThreadId; + let postLocalPort = firstLocalPort; + try { + strictEqual(sse.status, 200, 'pinned GET SSE establishes'); + for (let attempt = 0; attempt < 24 && postThreadId === getThreadId; attempt++) { + postAgent?.destroy(); + postAgent = new Agent({ keepAlive: true, maxSockets: 1 }); + postLocalPort = firstLocalPort + attempt + 1; + postThreadId = await workerIdentity(ctx.harper.httpURL, auth, postAgent, postLocalPort); + } + ok(postAgent, 'created a POST keep-alive connection'); + if (postThreadId === getThreadId) { + t.skip(`runtime exposed one application HTTP worker to all socket probes (thread ${getThreadId})`); + return; + } + + const id = `cross_worker_${Date.now().toString(36)}`; + const uri = new URL(`/WorkItem/${id}`, ctx.harper.httpURL).href; + await putWorkItem(ctx, id, { state: 'pending', payload: 'cross-worker' }); + const subscription = await mcpCallWithAgent( + ctx.harper.httpURL, + auth, + postAgent!, + session, + 'resources/subscribe', + { uri }, + 20, + postLocalPort + ); + strictEqual( + subscription.error, + undefined, + `cross-worker resources/subscribe should succeed: ${JSON.stringify(subscription.error)}` + ); + await putWorkItem(ctx, id, { state: 'done', payload: 'cross-worker' }); + const event = await sse.next( + (message) => message?.method === 'notifications/resources/updated' && message?.params?.uri === uri, + 5000 + ); + ok(event, 'the owner worker delivered resources/updated on the pinned GET stream'); + } finally { + sse.close(); + getAgent.destroy(); + postAgent?.destroy(); + } + }); }); diff --git a/unitTests/components/mcp/adapters/fastify.test.js b/unitTests/components/mcp/adapters/fastify.test.js index 4a96d0288b..fa5812272e 100644 --- a/unitTests/components/mcp/adapters/fastify.test.js +++ b/unitTests/components/mcp/adapters/fastify.test.js @@ -8,6 +8,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/adapters/harperHttp.test.js b/unitTests/components/mcp/adapters/harperHttp.test.js index c80fc25c57..c445e563b9 100644 --- a/unitTests/components/mcp/adapters/harperHttp.test.js +++ b/unitTests/components/mcp/adapters/harperHttp.test.js @@ -11,6 +11,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/lifecycle.test.js b/unitTests/components/mcp/lifecycle.test.js index 573f7f2cd0..e7c0c60eda 100644 --- a/unitTests/components/mcp/lifecycle.test.js +++ b/unitTests/components/mcp/lifecycle.test.js @@ -16,6 +16,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; diff --git a/unitTests/components/mcp/session.test.js b/unitTests/components/mcp/session.test.js index fab9989451..7a65b8d42e 100644 --- a/unitTests/components/mcp/session.test.js +++ b/unitTests/components/mcp/session.test.js @@ -2,7 +2,6 @@ const assert = require('node:assert'); const { createSession, loadSession, - saveSession, deleteSession, touchSession, _setSessionTableForTest, @@ -15,6 +14,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; @@ -67,15 +69,6 @@ describe('mcp/session', () => { }); }); - describe('saveSession', () => { - it('persists changes', async () => { - const created = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); - await saveSession({ ...created, initialized: true }); - const reloaded = await loadSession(created.id); - assert.equal(reloaded.initialized, true); - }); - }); - describe('deleteSession', () => { it('removes the record and subsequent loads return null', async () => { const created = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js new file mode 100644 index 0000000000..cf6ae74023 --- /dev/null +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -0,0 +1,199 @@ +const assert = require('node:assert'); +const { + claimSubscriptionOwner, + routeResourceSubscription, + _setSubscriptionItcForTest, + _setSubscriptionThreadIdForTest, + _setSubscriptionTimeoutForTest, + _resetSubscriptionRoutingForTest, + _pendingSubscriptionRouteCount, +} = require('#src/components/mcp/subscriptionRouting'); +const { ITC_EVENT_TYPES } = require('#src/utility/hdbTerms'); +const { createSession, loadSession, patchSession, _setSessionTableForTest } = require('#src/components/mcp/session'); +const { registerSession, _resetSessionRegistryForTest } = require('#src/components/mcp/sessionRegistry'); +const { _setSubscribeImplForTest } = require('#src/components/mcp/resources'); +const { _resetSubscriptionsForTest } = require('#src/components/mcp/subscriptions'); + +const USER = { username: 'alice', role: { permission: { super_user: true } } }; + +function fakeTable() { + const store = new Map(); + return { + async put(record) { + store.set(record.id, { ...record }); + }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, + async get(id) { + const record = store.get(id); + return record && { ...record }; + }, + async delete(id) { + store.delete(id); + }, + }; +} + +function fakeBridge(send) { + const listeners = new Map(); + return { + listeners, + onMessageByType(type, listener) { + listeners.set(type, listener); + }, + sendToThread(target, event) { + return send?.(target, event, listeners) ?? true; + }, + }; +} + +describe('mcp/subscriptionRouting', () => { + beforeEach(() => { + _setSessionTableForTest(fakeTable()); + _setSubscriptionThreadIdForTest(1); + }); + + afterEach(() => { + _resetSubscriptionsForTest(); + _resetSessionRegistryForTest(); + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); + _setSessionTableForTest(undefined); + _setSubscribeImplForTest(undefined); + }); + + async function remoteSession() { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await patchSession(session.id, { streamOwner: { threadId: 7, token: 'owner-token' } }); + return loadSession(session.id); + } + + it('accepts a correlated response only from the expected owner thread', async () => { + const bridge = fakeBridge((_target, event, listeners) => { + assert.equal(event.message.user.password, undefined, 'credentials must not cross the worker boundary'); + const requestId = event.message.requestId; + setImmediate(() => { + listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ + message: { requestId, originator: 8, result: 'not-subscribable' }, + }); + listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ + message: { requestId, originator: 7, result: 'success' }, + }); + }); + return true; + }); + _setSubscriptionItcForTest(bridge); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: { ...USER, password: 'do-not-forward' }, + }); + assert.equal(result, 'success'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('fails quickly when the persisted owner thread is unreachable', async () => { + _setSubscriptionItcForTest(fakeBridge(() => false)); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'no-live-stream'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('bounds a sent command when the owner never responds', async () => { + _setSubscriptionTimeoutForTest(5); + _setSubscriptionItcForTest(fakeBridge(() => true)); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'no-live-stream'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + + it('rejects a command whose stream token no longer owns the local registry', async () => { + let response; + const bridge = fakeBridge((_target, event) => { + if (event.type === ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE) response = event.message; + return true; + }); + _setSubscriptionItcForTest(bridge); + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + await bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r1', + originator: 1, + sessionId: session.id, + streamToken: 'stale-token', + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }, + }); + await new Promise(setImmediate); + assert.equal(response.result, 'no-live-stream'); + }); + + it('executes subscribe and unsubscribe on the owner and updates the durable URI list', async () => { + let response; + const bridge = fakeBridge((_target, event) => { + if (event.type === ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE) response = event.message; + return true; + }); + _setSubscriptionItcForTest(bridge); + _setSubscriptionThreadIdForTest(7); + _setSubscribeImplForTest(async (_path, user) => { + assert.deepEqual(user, USER); + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + const uri = 'https://app.test/Product/1'; + bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r2', + originator: 1, + sessionId: session.id, + streamToken: registered.streamToken, + operation: 'subscribe', + uri, + user: USER, + }, + }); + for (let i = 0; i < 20 && !response; i++) await new Promise(setImmediate); + assert.equal(response.result, 'success'); + assert.deepEqual((await loadSession(session.id)).subscriptions, [uri]); + + response = undefined; + bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND)({ + message: { + requestId: 'r3', + originator: 1, + sessionId: session.id, + streamToken: registered.streamToken, + operation: 'unsubscribe', + uri, + }, + }); + for (let i = 0; i < 20 && !response; i++) await new Promise(setImmediate); + assert.equal(response.result, 'success'); + assert.deepEqual((await loadSession(session.id)).subscriptions, []); + }); +}); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 2d7ede1898..40ece178c6 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -1,8 +1,9 @@ const assert = require('node:assert'); +const { threadId } = require('node:worker_threads'); const rewire = require('rewire'); const transport_mod = rewire('#src/components/mcp/transport'); const { handleMcpRequest } = transport_mod; -const { _setSessionTableForTest, createSession, loadSession, saveSession } = require('#src/components/mcp/session'); +const { _setSessionTableForTest, createSession, loadSession, patchSession } = require('#src/components/mcp/session'); const { getRegisteredSession, pushSessionFrame, @@ -47,6 +48,9 @@ function makeFakeTable() { async put(record) { store.set(record.id, { ...record }); }, + async patch(record) { + store.set(record.id, { ...store.get(record.id), ...record }); + }, async get(id) { const r = store.get(id); return r ? { ...r } : undefined; @@ -273,10 +277,9 @@ describe('mcp/transport', () => { it('does not roll back lastActivity when persisting the level (touchSession adopted)', async () => { // Force a known-old lastActivity, then setLevel: the request's touchSession - // must advance it, and the level-persisting saveSession must NOT write the + // must advance it, and the level patch must not write the // stale load-time value back (root fix — handlePost adopts the touched copy). - const stale = await loadSession(sessionId); - await saveSession({ ...stale, lastActivity: 1 }); + await patchSession(sessionId, { lastActivity: 1 }); await handleMcpRequest( makeReq({ body: jsonRpc(2, 'logging/setLevel', { level: 'info' }), @@ -1113,9 +1116,13 @@ describe('mcp/transport', () => { }); describe('resources/subscribe + resources/unsubscribe', () => { - beforeEach(() => { + beforeEach(async () => { // A live GET stream is required to subscribe — register one for the session. - registerSession(sessionId, 'application', { username: 'alice', role: { permission: { super_user: true } } }); + const registered = registerSession(sessionId, 'application', { + username: 'alice', + role: { permission: { super_user: true } }, + }); + await patchSession(sessionId, { streamOwner: { threadId, token: registered.streamToken } }); // Inject a fake change stream so dispatch doesn't need the real audit log. // `null` for the sentinel path makes the resource non-subscribable. _setSubscribeImplForTest(async (path) => diff --git a/utility/hdbTerms.ts b/utility/hdbTerms.ts index 775e6c7e3b..d27005698c 100644 --- a/utility/hdbTerms.ts +++ b/utility/hdbTerms.ts @@ -952,6 +952,8 @@ export const ITC_EVENT_TYPES = { // MCP §3.7: route a client's response to a server→client request back to the // worker awaiting it (the response POST can land on any worker). MCP_CLIENT_RESPONSE: 'mcp_client_response', + MCP_SUBSCRIPTION_COMMAND: 'mcp_subscription_command', + MCP_SUBSCRIPTION_RESPONSE: 'mcp_subscription_response', // #1736: components load per-worker, so a `server.registerOperation()` made there lands in // a worker-local OPERATION_FUNCTION_MAP the main-thread ops-API dispatcher can't see. A // worker announces each registration (OPERATION_REGISTERED) so the main thread can forward From 160dc70ef1e1efe2cee87de9733ce3652d11fb86 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Mon, 31 Aug 2026 21:50:34 -0600 Subject: [PATCH 02/12] Harden MCP subscription routing (#1751) --- components/mcp/session.ts | 10 ++++- components/mcp/subscriptionRouting.ts | 27 +++++++++---- components/mcp/toolRegistry.ts | 1 + components/mcp/transport.ts | 31 +++++++++++---- unitTests/components/mcp/session.test.js | 5 +++ .../mcp/subscriptionRouting.test.js | 38 ++++++++++++++++++- unitTests/components/mcp/transport.test.js | 29 ++++++++++++++ 7 files changed, 122 insertions(+), 19 deletions(-) diff --git a/components/mcp/session.ts b/components/mcp/session.ts index 46a6674e1b..58a8b0a37c 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -148,7 +148,15 @@ export async function createSession({ */ export async function loadSession(id: string): Promise { const record = (await (getTable() as any).get(id)) as McpSessionRecord | undefined | null; - if (!record) return null; + if ( + !record || + typeof record.protocolVersion !== 'string' || + typeof record.initialized !== 'boolean' || + typeof record.user !== 'string' || + typeof record.createdAt !== 'number' || + typeof record.lastActivity !== 'number' + ) + return null; return record; } diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index 99d73e94ef..8f503c2e3a 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -8,11 +8,11 @@ import { getRegisteredSession } from './sessionRegistry.ts'; import { addResourceSubscription, removeResourceSubscription } from './subscriptions.ts'; import type { AuthedUser } from './toolRegistry.ts'; -const DEFAULT_RESPONSE_TIMEOUT_MS = 2_000; +const DEFAULT_RESPONSE_TIMEOUT_MS = 30_000; const MAX_PENDING = 100; const MAX_PENDING_PER_SESSION = 25; -export type SubscriptionRouteResult = 'success' | 'not-subscribable' | 'no-live-stream' | 'internal-error'; +export type SubscriptionRouteResult = 'success' | 'not-subscribable' | 'no-live-stream' | 'timeout' | 'internal-error'; type Operation = 'subscribe' | 'unsubscribe'; interface Command { @@ -56,8 +56,17 @@ let responseTimeoutMs = DEFAULT_RESPONSE_TIMEOUT_MS; function bridge(): ItcBridge { if (bridgeOverride) return bridgeOverride; - const { onMessageByType } = require('../../server/threads/manageThreads.js'); - return { sendToThread: threads.sendToThread.bind(threads), onMessageByType }; + try { + const { onMessageByType } = require('../../server/threads/manageThreads.js'); + if (typeof threads !== 'undefined' && typeof threads.sendToThread === 'function') { + return { sendToThread: threads.sendToThread.bind(threads), onMessageByType }; + } + } catch (error) { + harperLogger.trace(`MCP subscription routing is unavailable: ${(error as Error).message}`); + return { sendToThread: () => false, onMessageByType: () => {} }; + } + harperLogger.trace('MCP subscription routing is unavailable: thread bridge is not initialized'); + return { sendToThread: () => false, onMessageByType: () => {} }; } export function _setSubscriptionItcForTest(fake: ItcBridge | undefined): void { @@ -100,7 +109,8 @@ function ensureWired(): void { const response = event.message as Response; const entry = pending.get(response?.requestId); if (!entry || response.originator !== entry.targetThreadId) return; - if (!['success', 'not-subscribable', 'no-live-stream', 'internal-error'].includes(response.result)) return; + if (!['success', 'not-subscribable', 'no-live-stream', 'timeout', 'internal-error'].includes(response.result)) + return; clearTimeout(entry.timer); pending.delete(response.requestId); entry.resolve(response.result); @@ -122,6 +132,7 @@ function countPendingForSession(sessionId: string): number { function subscriptionUser(user: AuthedUser): AuthedUser { return { ...(user.username ? { username: user.username } : {}), + ...(user.authExpiresAt !== undefined ? { authExpiresAt: user.authExpiresAt } : {}), ...(user._scopedToken ? { _scopedToken: true } : {}), ...(user.role ? { @@ -146,7 +157,7 @@ function routeRemote( return new Promise((resolve) => { const timer = setTimeout(() => { pending.delete(requestId); - resolve('no-live-stream'); + resolve('timeout'); }, responseTimeoutMs); timer.unref(); pending.set(requestId, { sessionId: command.sessionId, targetThreadId: owner.threadId, resolve, timer }); @@ -167,7 +178,7 @@ function routeRemote( }); } -function serializeSessionOperation(sessionId: string, operation: () => Promise): Promise { +export function withSessionSubscriptionLock(sessionId: string, operation: () => Promise): Promise { const previous = operationChains.get(sessionId) ?? Promise.resolve(); const current = previous.then(operation, operation); const tail = current @@ -185,7 +196,7 @@ function serializeSessionOperation(sessionId: string, operation: () => Promis async function executeLocal(command: Command): Promise { const registered = getRegisteredSession(command.sessionId); if (!registered || registered.streamToken !== command.streamToken) return 'no-live-stream'; - return serializeSessionOperation(command.sessionId, async () => { + return withSessionSubscriptionLock(command.sessionId, async () => { if (command.operation === 'subscribe') { if (!command.user) return 'internal-error'; const added = await addResourceSubscription(command.sessionId, command.uri, command.user); diff --git a/components/mcp/toolRegistry.ts b/components/mcp/toolRegistry.ts index fc0131ba4f..35b41c229d 100644 --- a/components/mcp/toolRegistry.ts +++ b/components/mcp/toolRegistry.ts @@ -60,6 +60,7 @@ export interface ToolDescriptor { /** Authenticated user object as Harper builds it (subset we touch). */ export interface AuthedUser { username?: string; + authExpiresAt?: number; // Attribution-only principal that must not be re-resolved against hdb_user (see refreshSessionUser). _scopedToken?: boolean; role?: { diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index eb742108d5..e81de3fbf7 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -37,7 +37,11 @@ import { listResources, listResourceTemplates, readResource, completeResourceArg import { ensureApplicationToolsFresh } from './tools/application.ts'; import { getPrompt, listPrompts, completePromptArgument } from './promptRegistry.ts'; import { dropSessionSubscriptions, restoreResourceSubscriptions } from './subscriptions.ts'; -import { claimSubscriptionOwner, routeResourceSubscription } from './subscriptionRouting.ts'; +import { + claimSubscriptionOwner, + routeResourceSubscription, + withSessionSubscriptionLock, +} from './subscriptionRouting.ts'; import { sendServerRequest, routeClientResponse, @@ -418,13 +422,13 @@ async function handleGet(request: NormRequest): Promise { record.queue.once('close', () => dropSessionSubscriptions(sessionId)); // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. - if (session.subscriptions?.length) { - const restored = await restoreResourceSubscriptions(sessionId, session.subscriptions, effectiveUser(request)); - if (restored.length !== session.subscriptions.length) { - session.subscriptions = restored; - await patchSession(session.id, { subscriptions: restored }); - } - } + await withSessionSubscriptionLock(sessionId, async () => { + const currentSession = await loadSession(sessionId); + const subscriptions = currentSession?.subscriptions; + if (!subscriptions?.length) return; + const restored = await restoreResourceSubscriptions(sessionId, subscriptions, effectiveUser(request)); + if (restored.length !== subscriptions.length) await patchSession(sessionId, { subscriptions: restored }); + }); // Resumability (#3.8): on reconnect with Last-Event-ID, replay buffered frames // the client missed (those with a higher id) before live frames flow. Re-sent // raw so their original event ids are preserved. Best-effort + per-worker: the @@ -936,6 +940,12 @@ async function dispatchResourcesSubscribe( // URIs (and unknown URIs) have no change source. return jsonResponse(200, buildError(messageId, ERROR_CODES.INVALID_PARAMS, `resource is not subscribable: ${uri}`)); } + if (result === 'timeout') { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource subscription timed out; retry the request') + ); + } if (result === 'internal-error') { return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource subscription failed')); } @@ -964,6 +974,11 @@ async function dispatchResourcesUnsubscribe( if (fresh?.subscriptions?.includes(uri)) { await patchSession(session.id, { subscriptions: fresh.subscriptions.filter((u) => u !== uri) }); } + } else if (result === 'timeout') { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe timed out; retry the request') + ); } else if (result === 'internal-error') { return jsonResponse(200, buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe failed')); } diff --git a/unitTests/components/mcp/session.test.js b/unitTests/components/mcp/session.test.js index 7a65b8d42e..bc5968de4c 100644 --- a/unitTests/components/mcp/session.test.js +++ b/unitTests/components/mcp/session.test.js @@ -67,6 +67,11 @@ describe('mcp/session', () => { const loaded = await loadSession('not-a-session'); assert.equal(loaded, null); }); + + it('returns null for a partial record left by a late patch after deletion', async () => { + fake.store.set('deleted-session', { id: 'deleted-session', lastActivity: Date.now() }); + assert.equal(await loadSession('deleted-session'), null); + }); }); describe('deleteSession', () => { diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index cf6ae74023..f72115dc5f 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -2,6 +2,7 @@ const assert = require('node:assert'); const { claimSubscriptionOwner, routeResourceSubscription, + withSessionSubscriptionLock, _setSubscriptionItcForTest, _setSubscriptionThreadIdForTest, _setSubscriptionTimeoutForTest, @@ -72,6 +73,7 @@ describe('mcp/subscriptionRouting', () => { it('accepts a correlated response only from the expected owner thread', async () => { const bridge = fakeBridge((_target, event, listeners) => { assert.equal(event.message.user.password, undefined, 'credentials must not cross the worker boundary'); + assert.equal(event.message.user.authExpiresAt, 12345); const requestId = event.message.requestId; setImmediate(() => { listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ @@ -88,7 +90,7 @@ describe('mcp/subscriptionRouting', () => { session: await remoteSession(), operation: 'subscribe', uri: 'https://app.test/Product/1', - user: { ...USER, password: 'do-not-forward' }, + user: { ...USER, authExpiresAt: 12345, password: 'do-not-forward' }, }); assert.equal(result, 'success'); assert.equal(_pendingSubscriptionRouteCount(), 0); @@ -115,7 +117,7 @@ describe('mcp/subscriptionRouting', () => { uri: 'https://app.test/Product/1', user: USER, }); - assert.equal(result, 'no-live-stream'); + assert.equal(result, 'timeout'); assert.equal(_pendingSubscriptionRouteCount(), 0); }); @@ -196,4 +198,36 @@ describe('mcp/subscriptionRouting', () => { assert.equal(response.result, 'success'); assert.deepEqual((await loadSession(session.id)).subscriptions, []); }); + + it('serializes owner commands behind reconnect restoration for the same session', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + let releaseRestore; + const restoreBlocked = new Promise((resolve) => (releaseRestore = resolve)); + const restoring = withSessionSubscriptionLock(session.id, () => restoreBlocked); + let subscribeStarted = false; + _setSubscribeImplForTest(async () => { + subscribeStarted = true; + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + const subscribing = routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/2', + user: USER, + }); + await new Promise(setImmediate); + assert.equal(subscribeStarted, false); + releaseRestore(); + await restoring; + assert.equal(await subscribing, 'success'); + assert.equal(subscribeStarted, true); + }); }); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 40ece178c6..3756382e14 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -23,6 +23,11 @@ const { _setHttpUrlPrefixForTest, _setSubscribeImplForTest, } = require('#src/components/mcp/resources'); +const { + _setSubscriptionItcForTest, + _setSubscriptionTimeoutForTest, + _resetSubscriptionRoutingForTest, +} = require('#src/components/mcp/subscriptionRouting'); function makeFakeResources(entries) { const map = new Map(); @@ -1186,6 +1191,30 @@ describe('mcp/transport', () => { assert.equal(res.jsonBody.error.code, -32602); }); + it('returns a retryable internal error when subscription routing times out', async () => { + _setSubscriptionItcForTest({ + onMessageByType() {}, + sendToThread() { + return true; + }, + }); + _setSubscriptionTimeoutForTest(5); + await patchSession(sessionId, { streamOwner: { threadId: threadId + 1, token: 'remote-owner' } }); + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(73, 'resources/subscribe', { uri: 'https://app.test:9926/Product/1' }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /timed out; retry/); + } finally { + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); + } + }); + it('unsubscribe removes the URI from the durable record', async () => { const uri = 'https://app.test:9926/Product/2'; await handleMcpRequest( From c8da87f38d546eb5551f3972e0f83e0f58f2f909 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Mon, 31 Aug 2026 21:55:23 -0600 Subject: [PATCH 03/12] Retry MCP subscription bridge wiring (#1751) --- components/mcp/sessionRegistry.ts | 1 - components/mcp/subscriptionRouting.ts | 16 +++++++++------- components/mcp/transport.ts | 7 ++----- .../components/mcp/subscriptionRouting.test.js | 14 ++++++++++++++ unitTests/components/mcp/transport.test.js | 3 +++ 5 files changed, 28 insertions(+), 13 deletions(-) diff --git a/components/mcp/sessionRegistry.ts b/components/mcp/sessionRegistry.ts index 308ed723e9..f72110e14a 100644 --- a/components/mcp/sessionRegistry.ts +++ b/components/mcp/sessionRegistry.ts @@ -30,7 +30,6 @@ export interface SseEvent { export interface RegisteredSession { sessionId: string; - /** Uniquely identifies this particular GET-SSE stream. */ streamToken: string; profile: McpProfile; user: AuthedUser; diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index 8f503c2e3a..de716cb5ae 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -32,6 +32,7 @@ interface Response { } interface ItcBridge { + available?: boolean; sendToThread(threadId: number, event: { type: string; message: unknown }): boolean; onMessageByType(type: string, listener: (event: { message?: unknown }) => void): void; } @@ -59,14 +60,14 @@ function bridge(): ItcBridge { try { const { onMessageByType } = require('../../server/threads/manageThreads.js'); if (typeof threads !== 'undefined' && typeof threads.sendToThread === 'function') { - return { sendToThread: threads.sendToThread.bind(threads), onMessageByType }; + return { available: true, sendToThread: threads.sendToThread.bind(threads), onMessageByType }; } } catch (error) { harperLogger.trace(`MCP subscription routing is unavailable: ${(error as Error).message}`); - return { sendToThread: () => false, onMessageByType: () => {} }; + return { available: false, sendToThread: () => false, onMessageByType: () => {} }; } harperLogger.trace('MCP subscription routing is unavailable: thread bridge is not initialized'); - return { sendToThread: () => false, onMessageByType: () => {} }; + return { available: false, sendToThread: () => false, onMessageByType: () => {} }; } export function _setSubscriptionItcForTest(fake: ItcBridge | undefined): void { @@ -97,15 +98,16 @@ export function _pendingSubscriptionRouteCount(): number { function ensureWired(): void { if (wired) return; - wired = true; - bridge().onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { + const itc = bridge(); + if (itc.available === false) return; + itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { const command = event.message as Command; void handleCommand(command).catch((error) => { harperLogger.error('MCP subscription command failed', error); sendResponse(command, 'internal-error'); }); }); - bridge().onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, (event) => { + itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE, (event) => { const response = event.message as Response; const entry = pending.get(response?.requestId); if (!entry || response.originator !== entry.targetThreadId) return; @@ -115,9 +117,9 @@ function ensureWired(): void { pending.delete(response.requestId); entry.resolve(response.result); }); + wired = true; } -/** Attach the owner-side command listener and persist the routing hint for a new GET stream. */ export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { ensureWired(); await patchSession(sessionId, { streamOwner: { threadId: currentThreadId(), token: streamToken } }); diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index e81de3fbf7..6f32e994cc 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -256,11 +256,8 @@ async function handlePost(request: NormRequest): Promise { return { status: 403, headers: {} }; } - // Sliding-window idle reset. Awaited (not fire-and-forget) so a concurrent - // DELETE that arrives mid-request can't be resurrected by a late put. Adopt - // the touched copy (fresh `lastActivity`) so any later save in this request - // — `handleInitialized`, `dispatchSetLevel` — persists the current activity - // time instead of rolling it back to the load-time value. + // Sliding-window idle reset. Await persistence before dispatch; loadSession + // rejects any partial row left by a concurrent DELETE/patch race. session = await touchSession(session); // A client's response to a server→client request (#3.7): route it to the diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index f72115dc5f..7c47856915 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -53,6 +53,7 @@ describe('mcp/subscriptionRouting', () => { beforeEach(() => { _setSessionTableForTest(fakeTable()); _setSubscriptionThreadIdForTest(1); + _setSubscriptionItcForTest(fakeBridge()); }); afterEach(() => { @@ -108,6 +109,19 @@ describe('mcp/subscriptionRouting', () => { assert.equal(_pendingSubscriptionRouteCount(), 0); }); + it('retries listener wiring after the thread bridge becomes available', async () => { + const bridge = fakeBridge(); + bridge.available = false; + _setSubscriptionItcForTest(bridge); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await claimSubscriptionOwner(session.id, 'first-stream'); + assert.equal(bridge.listeners.size, 0); + bridge.available = true; + await claimSubscriptionOwner(session.id, 'second-stream'); + assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND), true); + assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE), true); + }); + it('bounds a sent command when the owner never responds', async () => { _setSubscriptionTimeoutForTest(5); _setSubscriptionItcForTest(fakeBridge(() => true)); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 3756382e14..a2b24bea8a 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -101,6 +101,7 @@ describe('mcp/transport', () => { _setResourcesForTest(makeFakeResources([])); _setOpenApiGeneratorForTest(() => ({ openapi: '3.0.3', info: { title: 'fake' }, paths: {} })); _setHttpUrlPrefixForTest(''); + _setSubscriptionItcForTest({ onMessageByType() {}, sendToThread: () => true }); }); afterEach(() => { @@ -110,6 +111,8 @@ describe('mcp/transport', () => { _setResourcesForTest(undefined); _setOpenApiGeneratorForTest(undefined); _setHttpUrlPrefixForTest(undefined); + _resetSubscriptionRoutingForTest(); + _setSubscriptionItcForTest(undefined); }); describe('POST initialize', () => { From d6d163ef0c8c057ecfc3e5afcb12722a355f3468 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Mon, 31 Aug 2026 22:03:49 -0600 Subject: [PATCH 04/12] Preserve MCP subscription principal fields (#1751) --- components/mcp/subscriptionRouting.ts | 4 ++-- unitTests/components/mcp/subscriptionRouting.test.js | 10 +++++++++- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index de716cb5ae..36c0e378c4 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -133,13 +133,13 @@ function countPendingForSession(sessionId: string): number { function subscriptionUser(user: AuthedUser): AuthedUser { return { - ...(user.username ? { username: user.username } : {}), + ...(user.username !== undefined ? { username: user.username } : {}), ...(user.authExpiresAt !== undefined ? { authExpiresAt: user.authExpiresAt } : {}), ...(user._scopedToken ? { _scopedToken: true } : {}), ...(user.role ? { role: { - ...(user.role.role ? { role: user.role.role } : {}), + ...(user.role.role !== undefined ? { role: user.role.role } : {}), ...(user.role.permission ? { permission: user.role.permission } : {}), }, } diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index 7c47856915..b806c9a8d7 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -74,7 +74,9 @@ describe('mcp/subscriptionRouting', () => { it('accepts a correlated response only from the expected owner thread', async () => { const bridge = fakeBridge((_target, event, listeners) => { assert.equal(event.message.user.password, undefined, 'credentials must not cross the worker boundary'); + assert.equal(event.message.user.username, ''); assert.equal(event.message.user.authExpiresAt, 12345); + assert.equal(event.message.user.role.role, ''); const requestId = event.message.requestId; setImmediate(() => { listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE)({ @@ -91,7 +93,13 @@ describe('mcp/subscriptionRouting', () => { session: await remoteSession(), operation: 'subscribe', uri: 'https://app.test/Product/1', - user: { ...USER, authExpiresAt: 12345, password: 'do-not-forward' }, + user: { + ...USER, + username: '', + authExpiresAt: 12345, + role: { ...USER.role, role: '' }, + password: 'do-not-forward', + }, }); assert.equal(result, 'success'); assert.equal(_pendingSubscriptionRouteCount(), 0); From 6393eee5eddf4144728bd838c062a8f11856c4ba Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Mon, 31 Aug 2026 23:00:34 -0600 Subject: [PATCH 05/12] Validate MCP subscription commands (#1751) --- components/mcp/subscriptionRouting.ts | 20 ++++++++++++++++++- .../mcp/subscriptionRouting.test.js | 16 +++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index 36c0e378c4..5875dba32a 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -31,6 +31,20 @@ interface Response { result: SubscriptionRouteResult; } +function isCommand(value: unknown): value is Command { + if (!value || typeof value !== 'object') return false; + const command = value as Partial; + return ( + typeof command.requestId === 'string' && + typeof command.originator === 'number' && + typeof command.sessionId === 'string' && + typeof command.streamToken === 'string' && + (command.operation === 'subscribe' || command.operation === 'unsubscribe') && + typeof command.uri === 'string' && + (command.user === undefined || (command.user !== null && typeof command.user === 'object')) + ); +} + interface ItcBridge { available?: boolean; sendToThread(threadId: number, event: { type: string; message: unknown }): boolean; @@ -101,7 +115,11 @@ function ensureWired(): void { const itc = bridge(); if (itc.available === false) return; itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { - const command = event.message as Command; + const command = event.message; + if (!isCommand(command)) { + harperLogger.warn('Ignoring malformed MCP subscription command'); + return; + } void handleCommand(command).catch((error) => { harperLogger.error('MCP subscription command failed', error); sendResponse(command, 'internal-error'); diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index b806c9a8d7..74f91b4756 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -169,6 +169,22 @@ describe('mcp/subscriptionRouting', () => { assert.equal(response.result, 'no-live-stream'); }); + it('ignores malformed commands without attempting a response', async () => { + let sent = 0; + const bridge = fakeBridge(() => { + sent++; + return true; + }); + _setSubscriptionItcForTest(bridge); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + await claimSubscriptionOwner(session.id, 'owner-token'); + const listener = bridge.listeners.get(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND); + assert.doesNotThrow(() => listener({ message: undefined })); + assert.doesNotThrow(() => listener({ message: { requestId: 'r1', originator: 1 } })); + await new Promise(setImmediate); + assert.equal(sent, 0); + }); + it('executes subscribe and unsubscribe on the owner and updates the durable URI list', async () => { let response; const bridge = fakeBridge((_target, event) => { From bd1b3dce2eb64b3f154a75fce74e7968225bde78 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Tue, 1 Sep 2026 10:29:20 -0600 Subject: [PATCH 06/12] Fix concurrent MCP subscription updates --- components/mcp/session.ts | 37 ++++++++++++++++++ components/mcp/subscriptionRouting.ts | 20 ++++------ components/mcp/transport.ts | 30 ++++++++------ unitTests/components/mcp/session.test.js | 39 +++++++++++++++++++ .../mcp/subscriptionRouting.test.js | 19 +++++++++ unitTests/components/mcp/transport.test.js | 19 +++++++++ 6 files changed, 139 insertions(+), 25 deletions(-) diff --git a/components/mcp/session.ts b/components/mcp/session.ts index 58a8b0a37c..6565a1d9a8 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -165,6 +165,43 @@ export async function patchSession(id: string, changes: Partial { + return new Promise((resolve, reject) => { + const attempt = () => { + try { + if (store.tryLock(key, attempt)) resolve(); + } catch (error) { + reject(error); + } + }; + attempt(); + }); +} + +/** Serialize durable subscription read-modify-writes across HTTP workers. */ +export async function updateSessionSubscriptions( + id: string, + update: (subscriptions: string[]) => string[] | Promise +): Promise { + const store = getTable().primaryStore; + const key = subscriptionLockKey(id); + await acquireSessionSubscriptionLock(store, key); + try { + const session = await loadSession(id); + if (!session) return null; + const subscriptions = session.subscriptions ?? []; + const updated = await update(subscriptions); + if (updated !== subscriptions) await patchSession(id, { subscriptions: updated }); + return updated === subscriptions ? session : { ...session, subscriptions: updated }; + } finally { + store.unlock(key); + } +} + export async function deleteSession(id: string): Promise { await (getTable() as any).delete(id); // Tear down ancillary per-session in-memory state — the `tools/list` diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index 5875dba32a..a0d8c96f7a 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -3,7 +3,7 @@ import { randomUUID } from 'node:crypto'; import { threadId } from 'node:worker_threads'; import { ITC_EVENT_TYPES } from '../../utility/hdbTerms.ts'; import harperLogger from '../../utility/logging/harper_logger.ts'; -import { loadSession, patchSession, type McpSessionRecord } from './session.ts'; +import { patchSession, updateSessionSubscriptions, type McpSessionRecord } from './session.ts'; import { getRegisteredSession } from './sessionRegistry.ts'; import { addResourceSubscription, removeResourceSubscription } from './subscriptions.ts'; import type { AuthedUser } from './toolRegistry.ts'; @@ -222,28 +222,22 @@ async function executeLocal(command: Command): Promise const added = await addResourceSubscription(command.sessionId, command.uri, command.user); if (!added) return 'not-subscribable'; try { - const session = await loadSession(command.sessionId); + const session = await updateSessionSubscriptions(command.sessionId, (subscriptions) => + subscriptions.includes(command.uri) ? subscriptions : [...subscriptions, command.uri] + ); if (!session) { removeResourceSubscription(command.sessionId, command.uri); return 'no-live-stream'; } - if (!session.subscriptions?.includes(command.uri)) { - await patchSession(command.sessionId, { - subscriptions: [...(session.subscriptions ?? []), command.uri], - }); - } return 'success'; } catch (error) { removeResourceSubscription(command.sessionId, command.uri); throw error; } } - const session = await loadSession(command.sessionId); - if (session?.subscriptions?.includes(command.uri)) { - await patchSession(command.sessionId, { - subscriptions: session.subscriptions.filter((uri) => uri !== command.uri), - }); - } + await updateSessionSubscriptions(command.sessionId, (subscriptions) => + subscriptions.includes(command.uri) ? subscriptions.filter((uri) => uri !== command.uri) : subscriptions + ); removeResourceSubscription(command.sessionId, command.uri); return 'success'; }); diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index 6f32e994cc..910b9e066e 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -32,7 +32,14 @@ import { decodeCursor } from './pagination.ts'; import { seedSessionSnapshot } from './listChanged.ts'; import { tryAdmit, resolveClientIdentity } from './rateLimit.ts'; import { checkDurableQuota } from './quota.ts'; -import { deleteSession, loadSession, patchSession, touchSession, type McpSessionRecord } from './session.ts'; +import { + deleteSession, + loadSession, + patchSession, + touchSession, + updateSessionSubscriptions, + type McpSessionRecord, +} from './session.ts'; import { listResources, listResourceTemplates, readResource, completeResourceArgument } from './resources.ts'; import { ensureApplicationToolsFresh } from './tools/application.ts'; import { getPrompt, listPrompts, completePromptArgument } from './promptRegistry.ts'; @@ -419,13 +426,13 @@ async function handleGet(request: NormRequest): Promise { record.queue.once('close', () => dropSessionSubscriptions(sessionId)); // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. - await withSessionSubscriptionLock(sessionId, async () => { - const currentSession = await loadSession(sessionId); - const subscriptions = currentSession?.subscriptions; - if (!subscriptions?.length) return; - const restored = await restoreResourceSubscriptions(sessionId, subscriptions, effectiveUser(request)); - if (restored.length !== subscriptions.length) await patchSession(sessionId, { subscriptions: restored }); - }); + await withSessionSubscriptionLock(sessionId, () => + updateSessionSubscriptions(sessionId, async (subscriptions) => { + if (!subscriptions.length) return subscriptions; + const restored = await restoreResourceSubscriptions(sessionId, subscriptions, effectiveUser(request)); + return restored.length === subscriptions.length ? subscriptions : restored; + }) + ); // Resumability (#3.8): on reconnect with Last-Event-ID, replay buffered frames // the client missed (those with a higher id) before live frames flow. Re-sent // raw so their original event ids are preserved. Best-effort + per-worker: the @@ -967,10 +974,9 @@ async function dispatchResourcesUnsubscribe( if (result === 'no-live-stream') { // The live owner is already gone. Remove durable state locally so a later // reconnect cannot restore the cancelled subscription. - const fresh = await loadSession(session.id); - if (fresh?.subscriptions?.includes(uri)) { - await patchSession(session.id, { subscriptions: fresh.subscriptions.filter((u) => u !== uri) }); - } + await updateSessionSubscriptions(session.id, (subscriptions) => + subscriptions.includes(uri) ? subscriptions.filter((subscription) => subscription !== uri) : subscriptions + ); } else if (result === 'timeout') { return jsonResponse( 200, diff --git a/unitTests/components/mcp/session.test.js b/unitTests/components/mcp/session.test.js index bc5968de4c..37263675c5 100644 --- a/unitTests/components/mcp/session.test.js +++ b/unitTests/components/mcp/session.test.js @@ -4,13 +4,33 @@ const { loadSession, deleteSession, touchSession, + updateSessionSubscriptions, _setSessionTableForTest, } = require('#src/components/mcp/session'); function makeFakeTable() { const store = new Map(); + const locks = new Set(); + const waiters = new Map(); return { store, + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, async put(record) { store.set(record.id, { ...record }); }, @@ -102,4 +122,23 @@ describe('mcp/session', () => { assert.equal(touched.protocolVersion, '2025-06-18'); }); }); + + describe('updateSessionSubscriptions', () => { + it('serializes concurrent read-modify-writes for the same session', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + fake.store.get(session.id).subscriptions = ['first', 'second']; + const originalPatch = fake.patch; + fake.patch = async (record) => { + if (record.subscriptions) await new Promise((resolve) => setImmediate(resolve)); + return originalPatch.call(fake, record); + }; + + await Promise.all([ + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions.filter((uri) => uri !== 'first')), + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions.filter((uri) => uri !== 'second')), + ]); + + assert.deepEqual((await loadSession(session.id)).subscriptions, []); + }); + }); }); diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index 74f91b4756..7464c946d4 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -19,7 +19,26 @@ const USER = { username: 'alice', role: { permission: { super_user: true } } }; function fakeTable() { const store = new Map(); + const locks = new Set(); + const waiters = new Map(); return { + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, async put(record) { store.set(record.id, { ...record }); }, diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index a2b24bea8a..05f7710a01 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -48,8 +48,27 @@ function makeFakeResources(entries) { function makeFakeTable() { const store = new Map(); + const locks = new Set(); + const waiters = new Map(); return { store, + primaryStore: { + tryLock(key, callback) { + if (!locks.has(key)) { + locks.add(key); + return true; + } + const queued = waiters.get(key) ?? []; + queued.push(callback); + waiters.set(key, queued); + return false; + }, + unlock(key) { + locks.delete(key); + const callback = waiters.get(key)?.shift(); + if (callback) setImmediate(callback); + }, + }, async put(record) { store.set(record.id, { ...record }); }, From f94f58741bf541c9466efff10e6972fb2fc8220d Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 18:09:04 -0600 Subject: [PATCH 07/12] Address MCP subscription review races --- components/mcp/session.ts | 21 ++++++- components/mcp/subscriptionRouting.ts | 22 +++++-- components/mcp/transport.ts | 17 ++++- integrationTests/mcp/sse-listchanged.test.ts | 36 ++++++++--- unitTests/components/mcp/session.test.js | 22 +++++++ .../mcp/subscriptionRouting.test.js | 58 +++++++++++++++++ unitTests/components/mcp/transport.test.js | 63 +++++++++++++++++-- 7 files changed, 217 insertions(+), 22 deletions(-) diff --git a/components/mcp/session.ts b/components/mcp/session.ts index 6565a1d9a8..d0f1785e30 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -34,6 +34,7 @@ const DEFAULT_IDLE_TIMEOUT_SECONDS = 1800; * is identical either way.) */ const EVICTION_WINDOW_SECONDS = 60; +const DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS = 10_000; export interface McpSessionRecord { id: string; @@ -68,6 +69,7 @@ export interface McpSessionRecord { } let _sessionTable: Table | undefined; +let subscriptionLockTimeoutMs = DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS; /** * Lazily declare the system table. Called by `ensureSessionTable()` at @@ -114,6 +116,10 @@ export function _setSessionTableForTest(fake: Table | undefined): void { _sessionTable = fake; } +export function _setSubscriptionLockTimeoutForTest(value: number | undefined): void { + subscriptionLockTimeoutMs = value ?? DEFAULT_SUBSCRIPTION_LOCK_TIMEOUT_MS; +} + function getTable(): Table { if (!_sessionTable) throw new Error('MCP session table not initialized'); return _sessionTable; @@ -171,10 +177,23 @@ function subscriptionLockKey(id: string): string { function acquireSessionSubscriptionLock(store: Table['primaryStore'], key: string): Promise { return new Promise((resolve, reject) => { + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + reject(new Error('Timed out acquiring MCP subscription lock')); + }, subscriptionLockTimeoutMs); const attempt = () => { + if (settled) return; try { - if (store.tryLock(key, attempt)) resolve(); + if (store.tryLock(key, attempt)) { + settled = true; + clearTimeout(timer); + resolve(); + } } catch (error) { + settled = true; + clearTimeout(timer); reject(error); } }; diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index a0d8c96f7a..2d5c6051fc 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -150,6 +150,8 @@ function countPendingForSession(sessionId: string): number { } function subscriptionUser(user: AuthedUser): AuthedUser { + // Cross-worker subscribe authorization may rely only on this data-only principal shape. + // Built-in checks consume role.permission; credentials and mutable user state stay local. return { ...(user.username !== undefined ? { username: user.username } : {}), ...(user.authExpiresAt !== undefined ? { authExpiresAt: user.authExpiresAt } : {}), @@ -217,6 +219,8 @@ async function executeLocal(command: Command): Promise const registered = getRegisteredSession(command.sessionId); if (!registered || registered.streamToken !== command.streamToken) return 'no-live-stream'; return withSessionSubscriptionLock(command.sessionId, async () => { + const current = getRegisteredSession(command.sessionId); + if (!current || current.streamToken !== command.streamToken) return 'no-live-stream'; if (command.operation === 'subscribe') { if (!command.user) return 'internal-error'; const added = await addResourceSubscription(command.sessionId, command.uri, command.user); @@ -243,14 +247,17 @@ async function executeLocal(command: Command): Promise }); } -async function handleCommand(command: Command): Promise { - let result: SubscriptionRouteResult; +async function executeLocalContained(command: Command): Promise { try { - result = await executeLocal(command); + return await executeLocal(command); } catch (error) { harperLogger.error('MCP subscription owner failed to execute command', error); - result = 'internal-error'; + return 'internal-error'; } +} + +async function handleCommand(command: Command): Promise { + const result = await executeLocalContained(command); sendResponse(command, result); } @@ -280,7 +287,12 @@ export async function routeResourceSubscription(args: { ...(args.user ? { user: subscriptionUser(args.user) } : {}), }; if (owner.threadId === currentThreadId()) { - return executeLocal({ ...command, requestId: '', originator: currentThreadId(), streamToken: owner.token }); + return executeLocalContained({ + ...command, + requestId: '', + originator: currentThreadId(), + streamToken: owner.token, + }); } return routeRemote(owner, command); } diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index 910b9e066e..14083126ba 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -57,7 +57,6 @@ import { } from './serverRequests.ts'; import { registerSession, - unregisterSession, touchRegisteredSession, replaySince, type SseEvent, @@ -410,7 +409,7 @@ async function handleGet(request: NormRequest): Promise { try { await claimSubscriptionOwner(sessionId, record.streamToken); } catch (error) { - unregisterSession(sessionId); + record.queue.emit('close'); throw error; } // Seed the live record with any previously-set logging level so a reconnect @@ -970,7 +969,19 @@ async function dispatchResourcesUnsubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/unsubscribe requires params.uri') ); } - const result = await routeResourceSubscription({ session, operation: 'unsubscribe', uri }); + let result = await routeResourceSubscription({ session, operation: 'unsubscribe', uri }); + if (result === 'no-live-stream') { + const currentSession = await loadSession(session.id); + const owner = session.streamOwner; + const currentOwner = currentSession?.streamOwner; + if ( + currentSession && + currentOwner && + (!owner || currentOwner.threadId !== owner.threadId || currentOwner.token !== owner.token) + ) { + result = await routeResourceSubscription({ session: currentSession, operation: 'unsubscribe', uri }); + } + } if (result === 'no-live-stream') { // The live owner is already gone. Remove durable state locally so a later // reconnect cannot restore the cancelled subscription. diff --git a/integrationTests/mcp/sse-listchanged.test.ts b/integrationTests/mcp/sse-listchanged.test.ts index b97b7fb2ac..46caf6f0d4 100644 --- a/integrationTests/mcp/sse-listchanged.test.ts +++ b/integrationTests/mcp/sse-listchanged.test.ts @@ -478,22 +478,42 @@ suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => } const auth = adminAuth(ctx); const session = await initialize(ctx.harper.httpURL, auth); - const firstLocalPort = 30000 + (process.pid % 10000); - const getAgent = new Agent({ keepAlive: true, maxSockets: 1 }); - const getThreadId = await workerIdentity(ctx.harper.httpURL, auth, getAgent, firstLocalPort); - const sse = await openSseWithAgent(ctx.harper.httpURL, auth, getAgent, session, firstLocalPort); + const firstLocalPort = 20000 + (process.pid % 9000); + let getAgent; + let getThreadId; + let getLocalPort; + for (let attempt = 0; attempt < 25; attempt++) { + const candidateAgent = new Agent({ keepAlive: true, maxSockets: 1 }); + try { + getLocalPort = firstLocalPort + attempt; + getThreadId = await workerIdentity(ctx.harper.httpURL, auth, candidateAgent, getLocalPort); + getAgent = candidateAgent; + break; + } catch (error) { + candidateAgent.destroy(); + if ((error as NodeJS.ErrnoException).code !== 'EADDRINUSE') throw error; + } + } + ok(getAgent && getThreadId !== undefined && getLocalPort !== undefined, 'found a free pinned GET socket'); + const sse = await openSseWithAgent(ctx.harper.httpURL, auth, getAgent, session, getLocalPort); let postAgent; let postThreadId = getThreadId; - let postLocalPort = firstLocalPort; + let postLocalPort = getLocalPort; + let postConnected = false; try { strictEqual(sse.status, 200, 'pinned GET SSE establishes'); for (let attempt = 0; attempt < 24 && postThreadId === getThreadId; attempt++) { postAgent?.destroy(); postAgent = new Agent({ keepAlive: true, maxSockets: 1 }); - postLocalPort = firstLocalPort + attempt + 1; - postThreadId = await workerIdentity(ctx.harper.httpURL, auth, postAgent, postLocalPort); + postLocalPort = getLocalPort + attempt + 1; + try { + postThreadId = await workerIdentity(ctx.harper.httpURL, auth, postAgent, postLocalPort); + postConnected = true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== 'EADDRINUSE') throw error; + } } - ok(postAgent, 'created a POST keep-alive connection'); + ok(postAgent && postConnected, 'created a POST keep-alive connection on a free local port'); if (postThreadId === getThreadId) { t.skip(`runtime exposed one application HTTP worker to all socket probes (thread ${getThreadId})`); return; diff --git a/unitTests/components/mcp/session.test.js b/unitTests/components/mcp/session.test.js index 37263675c5..e072ea7c18 100644 --- a/unitTests/components/mcp/session.test.js +++ b/unitTests/components/mcp/session.test.js @@ -6,6 +6,7 @@ const { touchSession, updateSessionSubscriptions, _setSessionTableForTest, + _setSubscriptionLockTimeoutForTest, } = require('#src/components/mcp/session'); function makeFakeTable() { @@ -55,6 +56,7 @@ describe('mcp/session', () => { }); afterEach(() => { _setSessionTableForTest(undefined); + _setSubscriptionLockTimeoutForTest(undefined); }); describe('createSession', () => { @@ -124,6 +126,26 @@ describe('mcp/session', () => { }); describe('updateSessionSubscriptions', () => { + it('times out without acquiring a lock later from a stale wake callback', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const key = `mcp-subscriptions:${session.id}`; + assert.equal( + fake.primaryStore.tryLock(key, () => {}), + true + ); + _setSubscriptionLockTimeoutForTest(5); + + await assert.rejects( + updateSessionSubscriptions(session.id, (subscriptions) => subscriptions), + /Timed out acquiring MCP subscription lock/ + ); + fake.primaryStore.unlock(key); + await new Promise(setImmediate); + + await updateSessionSubscriptions(session.id, (subscriptions) => [...subscriptions, 'recovered']); + assert.deepEqual((await loadSession(session.id)).subscriptions, ['recovered']); + }); + it('serializes concurrent read-modify-writes for the same session', async () => { const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); fake.store.get(session.id).subscriptions = ['first', 'second']; diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index 7464c946d4..374f8c7f00 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -287,4 +287,62 @@ describe('mcp/subscriptionRouting', () => { assert.equal(await subscribing, 'success'); assert.equal(subscribeStarted, true); }); + + it('rechecks stream ownership after waiting for an earlier session operation', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + let releaseEarlierOperation; + const earlierOperation = withSessionSubscriptionLock( + session.id, + () => new Promise((resolve) => (releaseEarlierOperation = resolve)) + ); + let subscribeStarted = false; + _setSubscribeImplForTest(async () => { + subscribeStarted = true; + }); + const subscribing = routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/3', + user: USER, + }); + await new Promise(setImmediate); + registerSession(session.id, 'application', USER); + releaseEarlierOperation(); + await earlierOperation; + + assert.equal(await subscribing, 'no-live-stream'); + assert.equal(subscribeStarted, false); + }); + + it('normalizes local owner failures to an internal-error result', async () => { + _setSubscriptionThreadIdForTest(7); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const registered = registerSession(session.id, 'application', USER); + await claimSubscriptionOwner(session.id, registered.streamToken); + _setSubscribeImplForTest(async () => ({ + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + })); + const table = fakeTable(); + await table.put(await loadSession(session.id)); + table.patch = async () => { + throw new Error('write failed'); + }; + _setSessionTableForTest(table); + + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/4', + user: USER, + }), + 'internal-error' + ); + }); }); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 05f7710a01..f41b96f39c 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -23,11 +23,9 @@ const { _setHttpUrlPrefixForTest, _setSubscribeImplForTest, } = require('#src/components/mcp/resources'); -const { - _setSubscriptionItcForTest, - _setSubscriptionTimeoutForTest, - _resetSubscriptionRoutingForTest, -} = require('#src/components/mcp/subscriptionRouting'); +const subscriptionRouting = require('#src/components/mcp/subscriptionRouting'); +const { _setSubscriptionItcForTest, _setSubscriptionTimeoutForTest, _resetSubscriptionRoutingForTest } = + subscriptionRouting; function makeFakeResources(entries) { const map = new Map(); @@ -253,6 +251,30 @@ describe('mcp/transport', () => { assert.ok(res.sseIterable, 'SSE stream returned'); }); + it('does not close a superseding GET stream when an earlier owner claim fails', async () => { + let replacement; + const originalClaim = subscriptionRouting.claimSubscriptionOwner; + subscriptionRouting.claimSubscriptionOwner = async () => { + replacement = registerSession(sessionId, 'application', { + username: 'alice', + role: { permission: { super_user: true } }, + }); + throw new Error('claim failed'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 500); + assert.equal(getRegisteredSession(sessionId), replacement); + } finally { + subscriptionRouting.claimSubscriptionOwner = originalClaim; + } + }); + it('accepts a matching MCP-Protocol-Version and returns Method-not-found for unknown methods', async () => { const res = await handleMcpRequest( makeReq({ @@ -1256,6 +1278,37 @@ describe('mcp/transport', () => { const saved = await loadSession(sessionId); assert.ok(!(saved.subscriptions ?? []).includes(uri), 'URI dropped from the record'); }); + + it('reroutes unsubscribe when the stream owner changes during the first attempt', async () => { + const uri = 'https://app.test:9926/Product/3'; + await patchSession(sessionId, { subscriptions: [uri] }); + let calls = 0; + const originalRoute = subscriptionRouting.routeResourceSubscription; + subscriptionRouting.routeResourceSubscription = async ({ session }) => { + calls++; + if (calls === 1) { + await patchSession(sessionId, { streamOwner: { threadId: threadId + 1, token: 'new-owner' } }); + return 'no-live-stream'; + } + assert.deepEqual(session.streamOwner, { threadId: threadId + 1, token: 'new-owner' }); + await patchSession(sessionId, { subscriptions: [] }); + return 'success'; + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(76, 'resources/unsubscribe', { uri }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.deepEqual(res.jsonBody.result, {}); + assert.equal(calls, 2); + assert.deepEqual((await loadSession(sessionId)).subscriptions, []); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + } + }); }); describe('completion/complete', () => { From a3130a51fc6aab861048333677d873ee0cae6256 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 18:31:52 -0600 Subject: [PATCH 08/12] Contain MCP subscription restore failures --- components/mcp/session.ts | 1 - components/mcp/subscriptionRouting.ts | 2 +- components/mcp/transport.ts | 40 ++++++++--- unitTests/components/mcp/transport.test.js | 79 +++++++++++++++++++++- 4 files changed, 109 insertions(+), 13 deletions(-) diff --git a/components/mcp/session.ts b/components/mcp/session.ts index d0f1785e30..4d23f7715e 100644 --- a/components/mcp/session.ts +++ b/components/mcp/session.ts @@ -166,7 +166,6 @@ export async function loadSession(id: string): Promise return record; } -/** Incrementally update session fields without replacing concurrent changes. */ export async function patchSession(id: string, changes: Partial>): Promise { await (getTable() as any).patch({ id, ...changes }); } diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index 2d5c6051fc..ef1271072d 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -149,7 +149,7 @@ function countPendingForSession(sessionId: string): number { return count; } -function subscriptionUser(user: AuthedUser): AuthedUser { +export function subscriptionUser(user: AuthedUser): AuthedUser { // Cross-worker subscribe authorization may rely only on this data-only principal shape. // Built-in checks consume role.permission; credentials and mutable user state stay local. return { diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index 14083126ba..8e2d801f9d 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -47,6 +47,7 @@ import { dropSessionSubscriptions, restoreResourceSubscriptions } from './subscr import { claimSubscriptionOwner, routeResourceSubscription, + subscriptionUser, withSessionSubscriptionLock, } from './subscriptionRouting.ts'; import { @@ -57,6 +58,7 @@ import { } from './serverRequests.ts'; import { registerSession, + getRegisteredSession, touchRegisteredSession, replaySince, type SseEvent, @@ -425,13 +427,23 @@ async function handleGet(request: NormRequest): Promise { record.queue.once('close', () => dropSessionSubscriptions(sessionId)); // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. - await withSessionSubscriptionLock(sessionId, () => - updateSessionSubscriptions(sessionId, async (subscriptions) => { - if (!subscriptions.length) return subscriptions; - const restored = await restoreResourceSubscriptions(sessionId, subscriptions, effectiveUser(request)); - return restored.length === subscriptions.length ? subscriptions : restored; - }) - ); + try { + await withSessionSubscriptionLock(sessionId, () => + updateSessionSubscriptions(sessionId, async (subscriptions) => { + if (!subscriptions.length) return subscriptions; + const restored = await restoreResourceSubscriptions( + sessionId, + subscriptions, + subscriptionUser(effectiveUser(request)) + ); + return restored.length === subscriptions.length ? subscriptions : restored; + }) + ); + } catch (error) { + record.queue.emit('close'); + throw error; + } + if (getRegisteredSession(sessionId) !== record) dropSessionSubscriptions(sessionId); // Resumability (#3.8): on reconnect with Last-Event-ID, replay buffered frames // the client missed (those with a higher id) before live frames flow. Re-sent // raw so their original event ids are preserved. Best-effort + per-worker: the @@ -985,9 +997,17 @@ async function dispatchResourcesUnsubscribe( if (result === 'no-live-stream') { // The live owner is already gone. Remove durable state locally so a later // reconnect cannot restore the cancelled subscription. - await updateSessionSubscriptions(session.id, (subscriptions) => - subscriptions.includes(uri) ? subscriptions.filter((subscription) => subscription !== uri) : subscriptions - ); + try { + await updateSessionSubscriptions(session.id, (subscriptions) => + subscriptions.includes(uri) ? subscriptions.filter((subscription) => subscription !== uri) : subscriptions + ); + } catch (error) { + harperLogger.error('MCP resource unsubscribe durable update failed', error); + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe failed; retry the request') + ); + } } else if (result === 'timeout') { return jsonResponse( 200, diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index f41b96f39c..405efbf9e9 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -3,7 +3,8 @@ const { threadId } = require('node:worker_threads'); const rewire = require('rewire'); const transport_mod = rewire('#src/components/mcp/transport'); const { handleMcpRequest } = transport_mod; -const { _setSessionTableForTest, createSession, loadSession, patchSession } = require('#src/components/mcp/session'); +const sessionModule = require('#src/components/mcp/session'); +const { _setSessionTableForTest, createSession, loadSession, patchSession } = sessionModule; const { getRegisteredSession, pushSessionFrame, @@ -275,6 +276,58 @@ describe('mcp/transport', () => { } }); + it('closes the registered GET stream when subscription restoration fails', async () => { + const originalLock = subscriptionRouting.withSessionSubscriptionLock; + subscriptionRouting.withSessionSubscriptionLock = async () => { + throw new Error('restore failed'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 500); + assert.equal(getRegisteredSession(sessionId), undefined); + } finally { + subscriptionRouting.withSessionSubscriptionLock = originalLock; + } + }); + + it('restores subscriptions with the same projected principal used by subscribe', async () => { + const uri = 'https://app.test:9926/Product/restore'; + await patchSession(sessionId, { subscriptions: [uri] }); + let restoredUser; + _setSubscribeImplForTest(async (_path, user) => { + restoredUser = user; + return { + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + }; + }); + + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + userObject: { + username: 'alice', + password: 'do-not-forward', + customClaim: 'not-in-contract', + role: { permission: { super_user: true } }, + }, + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(restoredUser.username, 'alice'); + assert.equal(restoredUser.password, undefined); + assert.equal(restoredUser.customClaim, undefined); + res.sseIterable.emit('close'); + }); + it('accepts a matching MCP-Protocol-Version and returns Method-not-found for unknown methods', async () => { const res = await handleMcpRequest( makeReq({ @@ -1309,6 +1362,30 @@ describe('mcp/transport', () => { subscriptionRouting.routeResourceSubscription = originalRoute; } }); + + it('contains a durable unsubscribe failure in the request JSON-RPC response', async () => { + const originalRoute = subscriptionRouting.routeResourceSubscription; + const originalUpdate = sessionModule.updateSessionSubscriptions; + subscriptionRouting.routeResourceSubscription = async () => 'no-live-stream'; + sessionModule.updateSessionSubscriptions = async () => { + throw new Error('lock timed out'); + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(77, 'resources/unsubscribe', { uri: 'https://app.test:9926/Product/4' }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(res.jsonBody.id, 77); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /retry/); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + sessionModule.updateSessionSubscriptions = originalUpdate; + } + }); }); describe('completion/complete', () => { From dc973e5b34af94c8acdf350b3df25969c691ae2e Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 18:48:53 -0600 Subject: [PATCH 09/12] Harden MCP subscription recovery --- components/mcp/subscriptionRouting.ts | 19 ++++--- components/mcp/subscriptions.ts | 11 ++-- components/mcp/transport.ts | 56 ++++++++++++------- .../mcp/subscriptionRouting.test.js | 18 +++++- .../components/mcp/subscriptions.test.js | 9 +++ unitTests/components/mcp/transport.test.js | 28 ++++++++++ 6 files changed, 106 insertions(+), 35 deletions(-) diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index ef1271072d..c8e663eb2b 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -110,10 +110,10 @@ export function _pendingSubscriptionRouteCount(): number { return pending.size; } -function ensureWired(): void { - if (wired) return; +function ensureWired(): boolean { + if (wired) return true; const itc = bridge(); - if (itc.available === false) return; + if (itc.available === false) return false; itc.onMessageByType(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, (event) => { const command = event.message; if (!isCommand(command)) { @@ -136,10 +136,11 @@ function ensureWired(): void { entry.resolve(response.result); }); wired = true; + return true; } export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { - ensureWired(); + if (!ensureWired()) throw new Error('MCP subscription routing is unavailable'); await patchSession(sessionId, { streamOwner: { threadId: currentThreadId(), token: streamToken } }); } @@ -150,8 +151,8 @@ function countPendingForSession(sessionId: string): number { } export function subscriptionUser(user: AuthedUser): AuthedUser { - // Cross-worker subscribe authorization may rely only on this data-only principal shape. - // Built-in checks consume role.permission; credentials and mutable user state stay local. + // Preserve role permissions and scoped-token expiry for authorization rechecks; + // credentials and mutable user state must not cross the worker boundary. return { ...(user.username !== undefined ? { username: user.username } : {}), ...(user.authExpiresAt !== undefined ? { authExpiresAt: user.authExpiresAt } : {}), @@ -171,7 +172,7 @@ function routeRemote( owner: NonNullable, command: Omit ): Promise { - ensureWired(); + if (!ensureWired()) return Promise.resolve('no-live-stream'); if (pending.size >= MAX_PENDING || countPendingForSession(command.sessionId) >= MAX_PENDING_PER_SESSION) { return Promise.resolve('internal-error'); } @@ -184,18 +185,20 @@ function routeRemote( timer.unref(); pending.set(requestId, { sessionId: command.sessionId, targetThreadId: owner.threadId, resolve, timer }); let sent = false; + let sendFailed = false; try { sent = bridge().sendToThread(owner.threadId, { type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, message: { ...command, requestId, originator: currentThreadId(), streamToken: owner.token }, }); } catch (error) { + sendFailed = true; harperLogger.error('Unable to route MCP subscription command', error); } if (!sent) { clearTimeout(timer); pending.delete(requestId); - resolve('no-live-stream'); + resolve(sendFailed ? 'internal-error' : 'no-live-stream'); } }); } diff --git a/components/mcp/subscriptions.ts b/components/mcp/subscriptions.ts index 40bc22bdb2..190b651c6d 100644 --- a/components/mcp/subscriptions.ts +++ b/components/mcp/subscriptions.ts @@ -74,23 +74,24 @@ export function dropSessionSubscriptions(sessionId: string): void { /** * Re-establish subscriptions on SSE reconnect from the durable URI list. Each is * best-effort: a URI that's no longer subscribable (resource removed) is skipped. - * Returns the URIs that were successfully restored (the caller prunes the rest - * from the durable record). + * Returns the URIs that should remain durable: successful subscriptions and + * retryable failures are retained, while confirmed non-subscribable URIs are omitted. */ export async function restoreResourceSubscriptions( sessionId: string, uris: ReadonlyArray, user: AuthedUser ): Promise { - const restored: string[] = []; + const retained: string[] = []; for (const uri of uris) { try { - if (await addResourceSubscription(sessionId, uri, user)) restored.push(uri); + if (await addResourceSubscription(sessionId, uri, user)) retained.push(uri); } catch (err) { harperLogger.trace(`MCP subscription restore ${uri}: ${(err as Error).message}`); + retained.push(uri); } } - return restored; + return retained; } /** Test seam — count of live subscriptions for a session. */ diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index 8e2d801f9d..14c72d323c 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -428,17 +428,24 @@ async function handleGet(request: NormRequest): Promise { // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. try { - await withSessionSubscriptionLock(sessionId, () => - updateSessionSubscriptions(sessionId, async (subscriptions) => { - if (!subscriptions.length) return subscriptions; - const restored = await restoreResourceSubscriptions( - sessionId, - subscriptions, - subscriptionUser(effectiveUser(request)) - ); - return restored.length === subscriptions.length ? subscriptions : restored; - }) - ); + await withSessionSubscriptionLock(sessionId, async () => { + const snapshot = await updateSessionSubscriptions(sessionId, (subscriptions) => subscriptions); + if (!snapshot) throw new Error('MCP session disappeared during subscription restore'); + const attempted = snapshot.subscriptions ?? []; + if (!attempted.length) return; + const retained = await restoreResourceSubscriptions( + sessionId, + attempted, + subscriptionUser(effectiveUser(request)) + ); + if (retained.length === attempted.length) return; + const attemptedSet = new Set(attempted); + const retainedSet = new Set(retained); + await updateSessionSubscriptions(sessionId, (subscriptions) => { + const updated = subscriptions.filter((uri) => !attemptedSet.has(uri) || retainedSet.has(uri)); + return updated.length === subscriptions.length ? subscriptions : updated; + }); + }); } catch (error) { record.queue.emit('close'); throw error; @@ -981,18 +988,25 @@ async function dispatchResourcesUnsubscribe( buildError(messageId, ERROR_CODES.INVALID_PARAMS, 'resources/unsubscribe requires params.uri') ); } - let result = await routeResourceSubscription({ session, operation: 'unsubscribe', uri }); - if (result === 'no-live-stream') { + let routedSession = session; + let result: Awaited> = 'no-live-stream'; + let ownerChangedAfterFinalAttempt = false; + for (let attempt = 0; attempt < 3; attempt++) { + result = await routeResourceSubscription({ session: routedSession, operation: 'unsubscribe', uri }); + if (result !== 'no-live-stream') break; const currentSession = await loadSession(session.id); - const owner = session.streamOwner; + const owner = routedSession.streamOwner; const currentOwner = currentSession?.streamOwner; - if ( - currentSession && - currentOwner && - (!owner || currentOwner.threadId !== owner.threadId || currentOwner.token !== owner.token) - ) { - result = await routeResourceSubscription({ session: currentSession, operation: 'unsubscribe', uri }); - } + if (!currentSession || !currentOwner) break; + if (owner && currentOwner.threadId === owner.threadId && currentOwner.token === owner.token) break; + routedSession = currentSession; + ownerChangedAfterFinalAttempt = attempt === 2; + } + if (result === 'no-live-stream' && ownerChangedAfterFinalAttempt) { + return jsonResponse( + 200, + buildError(messageId, ERROR_CODES.INTERNAL_ERROR, 'resource unsubscribe owner changed; retry the request') + ); } if (result === 'no-live-stream') { // The live owner is already gone. Remove durable state locally so a later diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index 374f8c7f00..f4d655141d 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -141,7 +141,7 @@ describe('mcp/subscriptionRouting', () => { bridge.available = false; _setSubscriptionItcForTest(bridge); const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); - await claimSubscriptionOwner(session.id, 'first-stream'); + await assert.rejects(claimSubscriptionOwner(session.id, 'first-stream'), /routing is unavailable/); assert.equal(bridge.listeners.size, 0); bridge.available = true; await claimSubscriptionOwner(session.id, 'second-stream'); @@ -149,6 +149,22 @@ describe('mcp/subscriptionRouting', () => { assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE), true); }); + it('reports a send exception as an internal error rather than a missing stream', async () => { + _setSubscriptionItcForTest( + fakeBridge(() => { + throw new Error('could not clone command'); + }) + ); + const result = await routeResourceSubscription({ + session: await remoteSession(), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }); + assert.equal(result, 'internal-error'); + assert.equal(_pendingSubscriptionRouteCount(), 0); + }); + it('bounds a sent command when the owner never responds', async () => { _setSubscriptionTimeoutForTest(5); _setSubscriptionItcForTest(fakeBridge(() => true)); diff --git a/unitTests/components/mcp/subscriptions.test.js b/unitTests/components/mcp/subscriptions.test.js index 41c2dc5dc1..8f39ebca96 100644 --- a/unitTests/components/mcp/subscriptions.test.js +++ b/unitTests/components/mcp/subscriptions.test.js @@ -132,4 +132,13 @@ describe('mcp/subscriptions', () => { assert.deepEqual(restored, [URI], 'only the subscribable URI is restored'); assert.equal(_liveSubscriptionCount('s1'), 1); }); + + it('keeps a durable URI when restoration throws so a later reconnect can retry it', async () => { + _setSubscribeImplForTest(async () => { + throw new Error('authorization service unavailable'); + }); + const retained = await restoreResourceSubscriptions('s1', [URI], USER); + assert.deepEqual(retained, [URI]); + assert.equal(_liveSubscriptionCount('s1'), 0); + }); }); diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 405efbf9e9..499cfa05a7 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -1363,6 +1363,34 @@ describe('mcp/transport', () => { } }); + it('retains durable state when the unsubscribe owner keeps changing', async () => { + const uri = 'https://app.test:9926/Product/5'; + await patchSession(sessionId, { subscriptions: [uri] }); + let calls = 0; + const originalRoute = subscriptionRouting.routeResourceSubscription; + subscriptionRouting.routeResourceSubscription = async () => { + calls++; + await patchSession(sessionId, { streamOwner: { threadId: threadId + calls, token: `owner-${calls}` } }); + return 'no-live-stream'; + }; + try { + const res = await handleMcpRequest( + makeReq({ + body: jsonRpc(78, 'resources/unsubscribe', { uri }), + headers: { 'mcp-session-id': sessionId, 'mcp-protocol-version': '2025-06-18' }, + }) + ); + assert.equal(res.status, 200); + assert.equal(res.jsonBody.id, 78); + assert.equal(res.jsonBody.error.code, -32603); + assert.match(res.jsonBody.error.message, /retry/); + assert.equal(calls, 3); + assert.deepEqual((await loadSession(sessionId)).subscriptions, [uri]); + } finally { + subscriptionRouting.routeResourceSubscription = originalRoute; + } + }); + it('contains a durable unsubscribe failure in the request JSON-RPC response', async () => { const originalRoute = subscriptionRouting.routeResourceSubscription; const originalUpdate = sessionModule.updateSessionSubscriptions; From 969971ae0af2045979750cb2259247519d714091 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 19:04:27 -0600 Subject: [PATCH 10/12] Preserve local MCP subscription behavior --- components/mcp/subscriptionRouting.ts | 29 ++++++++++--- components/mcp/transport.ts | 24 +++++------ integrationTests/mcp/sse-listchanged.test.ts | 8 ++-- .../mcp/subscriptionRouting.test.js | 23 ++++++++++- unitTests/components/mcp/transport.test.js | 41 +++++++++++++++++-- 5 files changed, 99 insertions(+), 26 deletions(-) diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index c8e663eb2b..f333e8b178 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -139,9 +139,13 @@ function ensureWired(): boolean { return true; } -export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { - if (!ensureWired()) throw new Error('MCP subscription routing is unavailable'); +export async function claimSubscriptionOwner(sessionId: string, streamToken: string): Promise { + if (!ensureWired()) { + await patchSession(sessionId, { streamOwner: undefined }); + return false; + } await patchSession(sessionId, { streamOwner: { threadId: currentThreadId(), token: streamToken } }); + return true; } function countPendingForSession(sessionId: string): number { @@ -189,7 +193,13 @@ function routeRemote( try { sent = bridge().sendToThread(owner.threadId, { type: ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND, - message: { ...command, requestId, originator: currentThreadId(), streamToken: owner.token }, + message: { + ...command, + ...(command.user ? { user: subscriptionUser(command.user) } : {}), + requestId, + originator: currentThreadId(), + streamToken: owner.token, + }, }); } catch (error) { sendFailed = true; @@ -282,13 +292,22 @@ export async function routeResourceSubscription(args: { user?: AuthedUser; }): Promise { const owner = args.session.streamOwner; - if (!owner) return 'no-live-stream'; const command = { sessionId: args.session.id, operation: args.operation, uri: args.uri, - ...(args.user ? { user: subscriptionUser(args.user) } : {}), + ...(args.user ? { user: args.user } : {}), }; + if (!owner) { + const registered = getRegisteredSession(args.session.id); + if (!registered) return 'no-live-stream'; + return executeLocalContained({ + ...command, + requestId: '', + originator: currentThreadId(), + streamToken: registered.streamToken, + }); + } if (owner.threadId === currentThreadId()) { return executeLocalContained({ ...command, diff --git a/components/mcp/transport.ts b/components/mcp/transport.ts index 14c72d323c..78fa3107ad 100644 --- a/components/mcp/transport.ts +++ b/components/mcp/transport.ts @@ -47,7 +47,6 @@ import { dropSessionSubscriptions, restoreResourceSubscriptions } from './subscr import { claimSubscriptionOwner, routeResourceSubscription, - subscriptionUser, withSessionSubscriptionLock, } from './subscriptionRouting.ts'; import { @@ -264,8 +263,8 @@ async function handlePost(request: NormRequest): Promise { return { status: 403, headers: {} }; } - // Sliding-window idle reset. Await persistence before dispatch; loadSession - // rejects any partial row left by a concurrent DELETE/patch race. + // Adopt the persisted sliding-window update so later writes cannot roll + // lastActivity back; loadSession rejects partial rows from DELETE/patch races. session = await touchSession(session); // A client's response to a server→client request (#3.7): route it to the @@ -428,24 +427,25 @@ async function handleGet(request: NormRequest): Promise { // Restore durable resource subscriptions (#3.6) on (re)connect. Best-effort: // a URI that's no longer subscribable is dropped from the persisted list. try { - await withSessionSubscriptionLock(sessionId, async () => { + const sessionPresent = await withSessionSubscriptionLock(sessionId, async () => { const snapshot = await updateSessionSubscriptions(sessionId, (subscriptions) => subscriptions); - if (!snapshot) throw new Error('MCP session disappeared during subscription restore'); + if (!snapshot) return false; const attempted = snapshot.subscriptions ?? []; - if (!attempted.length) return; - const retained = await restoreResourceSubscriptions( - sessionId, - attempted, - subscriptionUser(effectiveUser(request)) - ); - if (retained.length === attempted.length) return; + if (!attempted.length) return true; + const retained = await restoreResourceSubscriptions(sessionId, attempted, effectiveUser(request)); + if (retained.length === attempted.length) return true; const attemptedSet = new Set(attempted); const retainedSet = new Set(retained); await updateSessionSubscriptions(sessionId, (subscriptions) => { const updated = subscriptions.filter((uri) => !attemptedSet.has(uri) || retainedSet.has(uri)); return updated.length === subscriptions.length ? subscriptions : updated; }); + return true; }); + if (!sessionPresent) { + record.queue.emit('close'); + return { status: 404, headers: {} }; + } } catch (error) { record.queue.emit('close'); throw error; diff --git a/integrationTests/mcp/sse-listchanged.test.ts b/integrationTests/mcp/sse-listchanged.test.ts index 46caf6f0d4..0831d1347b 100644 --- a/integrationTests/mcp/sse-listchanged.test.ts +++ b/integrationTests/mcp/sse-listchanged.test.ts @@ -514,10 +514,10 @@ suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => } } ok(postAgent && postConnected, 'created a POST keep-alive connection on a free local port'); - if (postThreadId === getThreadId) { - t.skip(`runtime exposed one application HTTP worker to all socket probes (thread ${getThreadId})`); - return; - } + ok( + postThreadId !== getThreadId, + `expected a second application HTTP worker after 24 socket probes; all used thread ${getThreadId}` + ); const id = `cross_worker_${Date.now().toString(36)}`; const uri = new URL(`/WorkItem/${id}`, ctx.harper.httpURL).href; diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index f4d655141d..dfc718dc54 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -141,14 +141,33 @@ describe('mcp/subscriptionRouting', () => { bridge.available = false; _setSubscriptionItcForTest(bridge); const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); - await assert.rejects(claimSubscriptionOwner(session.id, 'first-stream'), /routing is unavailable/); + await patchSession(session.id, { streamOwner: { threadId: 7, token: 'stale-owner' } }); + assert.equal(await claimSubscriptionOwner(session.id, 'first-stream'), false); + assert.equal((await loadSession(session.id)).streamOwner, undefined); assert.equal(bridge.listeners.size, 0); bridge.available = true; - await claimSubscriptionOwner(session.id, 'second-stream'); + assert.equal(await claimSubscriptionOwner(session.id, 'second-stream'), true); assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_COMMAND), true); assert.equal(bridge.listeners.has(ITC_EVENT_TYPES.MCP_SUBSCRIPTION_RESPONSE), true); }); + it('routes locally without publishing an owner when the thread bridge is unavailable', async () => { + const bridge = fakeBridge(); + bridge.available = false; + _setSubscriptionItcForTest(bridge); + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + const record = registerSession(session.id, 'application', USER); + assert.equal(await claimSubscriptionOwner(session.id, record.streamToken), false); + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'unsubscribe', + uri: 'https://app.test/Product/1', + }), + 'success' + ); + }); + it('reports a send exception as an internal error rather than a missing stream', async () => { _setSubscriptionItcForTest( fakeBridge(() => { diff --git a/unitTests/components/mcp/transport.test.js b/unitTests/components/mcp/transport.test.js index 499cfa05a7..fe1cc7b2c5 100644 --- a/unitTests/components/mcp/transport.test.js +++ b/unitTests/components/mcp/transport.test.js @@ -276,6 +276,41 @@ describe('mcp/transport', () => { } }); + it('serves the GET stream when cross-worker routing is unavailable', async () => { + const originalClaim = subscriptionRouting.claimSubscriptionOwner; + subscriptionRouting.claimSubscriptionOwner = async () => false; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 200); + assert.ok(res.sseIterable); + res.sseIterable.emit('close'); + } finally { + subscriptionRouting.claimSubscriptionOwner = originalClaim; + } + }); + + it('returns 404 when the durable session disappears during reconnect', async () => { + const originalUpdate = sessionModule.updateSessionSubscriptions; + sessionModule.updateSessionSubscriptions = async () => null; + try { + const res = await handleMcpRequest( + makeReq({ + method: 'GET', + headers: { 'mcp-session-id': sessionId, 'accept': 'text/event-stream' }, + }) + ); + assert.equal(res.status, 404); + assert.equal(getRegisteredSession(sessionId), undefined); + } finally { + sessionModule.updateSessionSubscriptions = originalUpdate; + } + }); + it('closes the registered GET stream when subscription restoration fails', async () => { const originalLock = subscriptionRouting.withSessionSubscriptionLock; subscriptionRouting.withSessionSubscriptionLock = async () => { @@ -295,7 +330,7 @@ describe('mcp/transport', () => { } }); - it('restores subscriptions with the same projected principal used by subscribe', async () => { + it('restores subscriptions with the full local principal', async () => { const uri = 'https://app.test:9926/Product/restore'; await patchSession(sessionId, { subscriptions: [uri] }); let restoredUser; @@ -323,8 +358,8 @@ describe('mcp/transport', () => { ); assert.equal(res.status, 200); assert.equal(restoredUser.username, 'alice'); - assert.equal(restoredUser.password, undefined); - assert.equal(restoredUser.customClaim, undefined); + assert.equal(restoredUser.password, 'do-not-forward'); + assert.equal(restoredUser.customClaim, 'not-in-contract'); res.sseIterable.emit('close'); }); From 752eca4a6711fb46c74d9fd9bb9a7d54dc91e9ba Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 19:09:14 -0600 Subject: [PATCH 11/12] Fence MCP local owner fallback --- components/mcp/subscriptionRouting.ts | 1 + .../mcp/subscriptionRouting.test.js | 26 ++++++++++++++++++- 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/components/mcp/subscriptionRouting.ts b/components/mcp/subscriptionRouting.ts index f333e8b178..7fd2e94e07 100644 --- a/components/mcp/subscriptionRouting.ts +++ b/components/mcp/subscriptionRouting.ts @@ -299,6 +299,7 @@ export async function routeResourceSubscription(args: { ...(args.user ? { user: args.user } : {}), }; if (!owner) { + if (ensureWired()) return 'no-live-stream'; const registered = getRegisteredSession(args.session.id); if (!registered) return 'no-live-stream'; return executeLocalContained({ diff --git a/unitTests/components/mcp/subscriptionRouting.test.js b/unitTests/components/mcp/subscriptionRouting.test.js index dfc718dc54..300c7dec1f 100644 --- a/unitTests/components/mcp/subscriptionRouting.test.js +++ b/unitTests/components/mcp/subscriptionRouting.test.js @@ -93,6 +93,7 @@ describe('mcp/subscriptionRouting', () => { it('accepts a correlated response only from the expected owner thread', async () => { const bridge = fakeBridge((_target, event, listeners) => { assert.equal(event.message.user.password, undefined, 'credentials must not cross the worker boundary'); + assert.equal(event.message.user.customClaim, undefined, 'custom principal state must remain local'); assert.equal(event.message.user.username, ''); assert.equal(event.message.user.authExpiresAt, 12345); assert.equal(event.message.user.role.role, ''); @@ -118,6 +119,7 @@ describe('mcp/subscriptionRouting', () => { authExpiresAt: 12345, role: { ...USER.role, role: '' }, password: 'do-not-forward', + customClaim: 'local-only', }, }); assert.equal(result, 'success'); @@ -155,17 +157,39 @@ describe('mcp/subscriptionRouting', () => { const bridge = fakeBridge(); bridge.available = false; _setSubscriptionItcForTest(bridge); + _setSubscribeImplForTest(async () => ({ + end() {}, + [Symbol.asyncIterator]() { + return { next: () => new Promise(() => {}) }; + }, + })); const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); const record = registerSession(session.id, 'application', USER); assert.equal(await claimSubscriptionOwner(session.id, record.streamToken), false); assert.equal( await routeResourceSubscription({ session: await loadSession(session.id), - operation: 'unsubscribe', + operation: 'subscribe', uri: 'https://app.test/Product/1', + user: USER, }), 'success' ); + assert.deepEqual((await loadSession(session.id)).subscriptions, ['https://app.test/Product/1']); + }); + + it('does not guess a local owner when worker routing is wired', async () => { + const session = await createSession({ user: 'alice', protocolVersion: '2025-06-18' }); + registerSession(session.id, 'application', USER); + assert.equal( + await routeResourceSubscription({ + session: await loadSession(session.id), + operation: 'subscribe', + uri: 'https://app.test/Product/1', + user: USER, + }), + 'no-live-stream' + ); }); it('reports a send exception as an internal error rather than a missing stream', async () => { From 5c60fea0e5e9b0eefd33d174b08e7de36143d832 Mon Sep 17 00:00:00 2001 From: Kyle Bernhardy Date: Fri, 4 Sep 2026 19:28:21 -0600 Subject: [PATCH 12/12] Skip cross-worker socket probe on Bun --- integrationTests/mcp/sse-listchanged.test.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/integrationTests/mcp/sse-listchanged.test.ts b/integrationTests/mcp/sse-listchanged.test.ts index 0831d1347b..4025d7b8fa 100644 --- a/integrationTests/mcp/sse-listchanged.test.ts +++ b/integrationTests/mcp/sse-listchanged.test.ts @@ -472,8 +472,12 @@ suite('MCP v1 SSE channel + list_changed delivery', (ctx: ContextWithHarper) => }); test('N5: resource subscription routes from a sibling POST worker to the GET-SSE owner', async (t) => { - if (process.platform === 'win32') { - t.skip('Harper forces one HTTP worker on Windows'); + if (process.platform === 'win32' || process.env.HARPER_RUNTIME === 'bun') { + t.skip( + process.platform === 'win32' + ? 'Harper forces one HTTP worker on Windows' + : 'Bun assigns pinned socket probes to one HTTP worker; Node CI covers cross-worker routing' + ); return; } const auth = adminAuth(ctx);