diff --git a/docs/features/mcp-connectors.md b/docs/features/mcp-connectors.md index bb504abba..ee8c19d93 100644 --- a/docs/features/mcp-connectors.md +++ b/docs/features/mcp-connectors.md @@ -144,7 +144,7 @@ executeAiTool(...) / live editor bridge | `contentAuthorization.ts` | Before a document tool is relayed to the browser, resolves the target row through `server/auth/dataAccess.ts` with the connection's own identity and capabilities (table family plus own-vs-any ownership), so the owner's admin cookie cannot widen the delegated grant. | | `transports/http.ts` | Authenticated, Origin-validated `createMcpHandler` entry for MCP 2026-07-28 plus the stateless 2025 fallback. | | `server.ts` / `registry.ts` | Low-level SDK server, TypeBox input schemas, catalog deduplication, and capability filtering. | -| `editorBridge.ts` | Per-user, per-scope live workspace bridge. The stream carries an **idle lease** (120s, re-armed by every relayed tool request) so an active batch is never cut mid-flight; only quiet streams recycle. The workspace's reconnect loop (`useMcpWorkspaceBridge`) reopens a recycled healthy stream immediately off the stream-end network event — deliberately timer-free, because hidden webviews (backgrounded browser tabs) clamp timers to minutes while network events still fire — and a tab becoming visible short-circuits any pending retry delay. | +| `editorBridge.ts` | Per-user, per-scope live workspace bridge. Tool requests are delivered **serially** (the next is pushed only once the previous settles, matching the editor's one-at-a-time execution), so each call's 90s timeout measures the editor's actual work, not time queued behind slower calls. The stream carries an **idle lease** (120s, re-armed by every relayed tool request and never expiring while a call is pending) so an active batch is never cut mid-flight; only quiet streams recycle. The workspace's reconnect loop (`useMcpWorkspaceBridge`) reopens a recycled healthy stream immediately off the stream-end network event — deliberately timer-free, because hidden webviews (backgrounded browser tabs) clamp timers to minutes while network events still fire — and a tab becoming visible short-circuits any pending retry delay. | | `tools/publishTool.ts` | Explicit canonical full-site publish with MCP audit metadata. | | `tools/uploadMediaTool.ts` | Server-resolved image upload (`media_upload`) — inline base64 or SSRF-guarded `sourceUrl` download, through the shared media pipeline. | diff --git a/server/ai/mcp/editorBridge.ts b/server/ai/mcp/editorBridge.ts index 58a71f9c5..ebdc2bcf3 100644 --- a/server/ai/mcp/editorBridge.ts +++ b/server/ai/mcp/editorBridge.ts @@ -88,6 +88,7 @@ export function createEditorBridgeStream( let bridgeId = '' let destroyBridge = (): void => {} + let pendingCount = (): number => 0 let heartbeat: ReturnType | null = null let lease: ReturnType | null = null @@ -117,7 +118,14 @@ export function createEditorBridgeStream( const armLease = (): void => { if (closed) return if (lease) clearTimeout(lease) - lease = setTimeout(cleanup, idleLeaseMs) + lease = setTimeout(() => { + // Never recycle under a pending call: that rejected it while the + // editor was still working on it (its result then 404'd). Every + // pending call is bounded by the relay's per-call timeout, so an + // orphaned stream still drains and closes. + if (pendingCount() > 0) armLease() + else cleanup() + }, idleLeaseMs) } const emit = (event: AiStreamEvent): void => { @@ -125,17 +133,21 @@ export function createEditorBridgeStream( try { controller.enqueue(encodeStreamEvent(event)) // Tool traffic proves the stream is wanted — keep the idle lease - // from expiring under an active batch. (The relay's own per-call - // timeout is 90s, comfortably inside the lease.) + // from expiring under an active batch. armLease() } catch { cleanup() } } - const created = createBridge(emit, signal) + // Serial: the editor runs relayed tools strictly one at a time, so + // pushing a whole batch at once only started every call's timeout (and + // the idle lease) early — queued calls timed out, or the lease closed the + // stream mid-batch, while the tab was still working through earlier ones. + const created = createBridge(emit, signal, undefined, undefined, { serial: true }) bridgeId = created.bridgeId destroyBridge = created.destroy + pendingCount = created.pendingCount // Newest instance of this workspace wins. The user's other workspace // remains connected, so Site and Content may serve MCP simultaneously. diff --git a/server/ai/runtime/transport.ts b/server/ai/runtime/transport.ts index c6cebf784..284ec72d7 100644 --- a/server/ai/runtime/transport.ts +++ b/server/ai/runtime/transport.ts @@ -79,16 +79,55 @@ export function createBridge( signal?: AbortSignal, timeoutMs: number = BROWSER_TOOL_TIMEOUT_MS, onSnapshot?: (snapshot: unknown) => void, + options: { + /** + * Deliver one tool request at a time: the next queued call is emitted only + * once the previous one settles, and each call's timeout starts when it is + * emitted. For consumers that execute requests strictly in order (the MCP + * editor bridge) this makes the timeout measure the browser's actual work + * instead of time spent queued behind slower calls. + */ + serial?: boolean + } = {}, ): { bridgeId: string bridge: AiBrowserBridge destroy: () => void + /** Calls awaiting a result, queued or emitted. */ + pendingCount: () => number } { const bridgeId = nanoid() const entry: BridgeEntry = { pending: new Map(), emit, onSnapshot } activeBridges.set(bridgeId, entry) let destroyed = false + // Serial delivery: `busy` while an emitted call is outstanding; queued calls + // wait in `waiting` until it settles. + const waiting: Array<() => void> = [] + let busy = false + const acquire = (start: () => void): void => { + if (!options.serial) return start() + if (busy) { + waiting.push(start) + return + } + busy = true + start() + } + const release = (): void => { + if (!options.serial) return + busy = false + if (destroyed || signal?.aborted) { + waiting.length = 0 + return + } + const next = waiting.shift() + if (next) { + busy = true + next() + } + } + const bridge: AiBrowserBridge = { callBrowser(toolName, input) { if (destroyed) { @@ -96,6 +135,12 @@ export function createBridge( } const requestId = nanoid() return new Promise((resolve, reject) => { + if (signal?.aborted) { + reject(new Error('AI chat stream aborted before tool result arrived.')) + return + } + let timer: ReturnType | null = null + let started = false // Settle (and remove) the pending wait on timeout or client disconnect // so a non-responding browser can never hang the SDK stream or leak the // bridge entry (ISS-030). @@ -106,24 +151,28 @@ export function createBridge( live.cleanup() reject(err) } - const timer = setTimeout( - () => settle(new Error(`Browser tool "${toolName}" result timed out.`)), - timeoutMs, - ) const onAbort = () => settle(new Error('AI chat stream aborted before tool result arrived.')) + const start = () => { + started = true + timer = setTimeout( + () => settle(new Error(`Browser tool "${toolName}" result timed out.`)), + timeoutMs, + ) + emit({ type: 'toolRequest', requestId, toolName, input }) + } const cleanup = () => { - clearTimeout(timer) + if (timer) clearTimeout(timer) signal?.removeEventListener('abort', onAbort) + if (started) release() + else { + const queued = waiting.indexOf(start) + if (queued !== -1) waiting.splice(queued, 1) + } } - if (signal?.aborted) { - clearTimeout(timer) - reject(new Error('AI chat stream aborted before tool result arrived.')) - return - } signal?.addEventListener('abort', onAbort, { once: true }) entry.pending.set(requestId, { resolve, reject, cleanup }) - emit({ type: 'toolRequest', requestId, toolName, input }) + acquire(start) }) }, } @@ -149,7 +198,7 @@ export function createBridge( activeBridges.delete(bridgeId) } - return { bridgeId, bridge, destroy } + return { bridgeId, bridge, destroy, pendingCount: () => entry.pending.size } } /** diff --git a/src/__tests__/ai/bridgeTimeoutAbort.test.ts b/src/__tests__/ai/bridgeTimeoutAbort.test.ts index 338daa63a..02908d2c2 100644 --- a/src/__tests__/ai/bridgeTimeoutAbort.test.ts +++ b/src/__tests__/ai/bridgeTimeoutAbort.test.ts @@ -108,3 +108,87 @@ describe('bridge snapshot refresh', () => { destroy() }) }) + +/** + * The MCP editor runs relayed tools one at a time. Pushing a whole batch at + * once started every call's timeout at push time, so calls queued behind slow + * ones timed out before the editor even reached them. A serial bridge emits + * the next request only after the previous one settles. + */ +describe('serial bridge delivery', () => { + function serialBridge(timeoutMs = 1000, signal?: AbortSignal) { + const requests: string[] = [] + const created = createBridge( + (ev: AiStreamEvent) => { + if (ev.type === 'toolRequest') requests.push(ev.requestId) + }, + signal, + timeoutMs, + undefined, + { serial: true }, + ) + return { ...created, requests } + } + + test('emits the next request only after the previous one settles', async () => { + const { bridgeId, bridge, destroy, requests } = serialBridge() + const first = bridge.callBrowser('a', {}) + const second = bridge.callBrowser('b', {}) + expect(requests).toHaveLength(1) + + expect(resolveBridgeToolResult(bridgeId, requests[0], { ok: true, data: 1 })).toBe(true) + await expect(first).resolves.toMatchObject({ ok: true }) + expect(requests).toHaveLength(2) + + expect(resolveBridgeToolResult(bridgeId, requests[1], { ok: true, data: 2 })).toBe(true) + await expect(second).resolves.toMatchObject({ ok: true }) + destroy() + }) + + test('a queued call’s timeout starts when it is emitted, not when it is queued', async () => { + const { bridgeId, bridge, destroy, requests } = serialBridge(60) + const first = bridge.callBrowser('slow', {}) + const second = bridge.callBrowser('fast', {}) + // The first call takes most of the timeout; the second must not expire + // while it waits its turn. + await new Promise((r) => setTimeout(r, 45)) + resolveBridgeToolResult(bridgeId, requests[0], { ok: true, data: 1 }) + await first + await new Promise((r) => setTimeout(r, 30)) + expect(resolveBridgeToolResult(bridgeId, requests[1], { ok: true, data: 2 })).toBe(true) + await expect(second).resolves.toMatchObject({ ok: true }) + destroy() + }) + + test('a timed-out call releases the slot to the next one', async () => { + const { bridge, destroy, requests } = serialBridge(20) + const first = bridge.callBrowser('stuck', {}) + const second = bridge.callBrowser('next', {}) + await expect(first).rejects.toThrow(/timed out/i) + expect(requests).toHaveLength(2) + await expect(second).rejects.toThrow(/timed out/i) + destroy() + }) + + test('destroy rejects queued calls without emitting them', async () => { + const { bridge, destroy, requests } = serialBridge() + const first = bridge.callBrowser('a', {}) + const second = bridge.callBrowser('b', {}) + destroy() + await expect(first).rejects.toThrow(/ended/) + await expect(second).rejects.toThrow(/ended/) + expect(requests).toHaveLength(1) + }) + + test('abort rejects queued calls without emitting them', async () => { + const controller = new AbortController() + const { bridge, destroy, requests } = serialBridge(1000, controller.signal) + const first = bridge.callBrowser('a', {}) + const second = bridge.callBrowser('b', {}) + controller.abort() + await expect(first).rejects.toThrow(/abort/i) + await expect(second).rejects.toThrow(/abort/i) + expect(requests).toHaveLength(1) + destroy() + }) +}) diff --git a/src/__tests__/ai/editorBridgeLease.test.ts b/src/__tests__/ai/editorBridgeLease.test.ts index f9867dfeb..c6066e949 100644 --- a/src/__tests__/ai/editorBridgeLease.test.ts +++ b/src/__tests__/ai/editorBridgeLease.test.ts @@ -11,6 +11,7 @@ import { getEditorBridgeForUser, hasEditorBridge, } from '../../../server/ai/mcp/editorBridge' +import { resolveBridgeToolResult } from '../../../server/ai/runtime/transport' const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)) @@ -28,12 +29,45 @@ function drain(stream: ReadableStream): void { })() } +/** + * Play the editor: read the stream and answer each tool request `delayMs` + * after it arrives, strictly one at a time — as `useMcpWorkspaceBridge` does. + */ +function playEditor(stream: ReadableStream, delayMs: number): void { + void (async () => { + const reader = stream.pipeThrough(new TextDecoderStream()).getReader() + let bridgeId = '' + let buffer = '' + try { + for (;;) { + const { done, value } = await reader.read() + if (done) return + buffer += value + let nl: number + while ((nl = buffer.indexOf('\n')) !== -1) { + const line = buffer.slice(0, nl).trim() + buffer = buffer.slice(nl + 1) + if (!line) continue + const event = JSON.parse(line) + if (event.type === 'bridgeReady') bridgeId = event.bridgeId + if (event.type === 'toolRequest') { + await sleep(delayMs) + resolveBridgeToolResult(bridgeId, event.requestId, { ok: true, data: event.toolName }) + } + } + } + } catch { + /* stream torn down by the test */ + } + })() +} + describe('editor bridge idle lease', () => { it('keeps an active stream alive past the lease while tools flow, then recycles it when idle', async () => { const controller = new AbortController() const userId = `lease-user-${Date.now()}` const stream = createEditorBridgeStream(userId, 'site', controller.signal, { idleLeaseMs: 80 }) - drain(stream) + playEditor(stream, 5) await sleep(10) expect(hasEditorBridge(userId, 'site')).toBe(true) @@ -42,9 +76,7 @@ describe('editor bridge idle lease', () => { for (let i = 0; i < 6; i++) { const bridge = getEditorBridgeForUser(userId, 'site') expect(bridge).not.toBeNull() - // Fire-and-forget: nothing posts a result in this test, and the call - // rejects when the bridge is destroyed at the end. - bridge?.callBrowser('noop_tool', {}).catch(() => {}) + await bridge?.callBrowser('noop_tool', {}) await sleep(40) expect(hasEditorBridge(userId, 'site')).toBe(true) } @@ -70,3 +102,43 @@ describe('editor bridge idle lease', () => { controller.abort() }) }) + +describe('editor bridge batch delivery', () => { + it('survives a batch whose serial execution outlasts the idle lease', async () => { + const controller = new AbortController() + const userId = `lease-batch-${Date.now()}` + const stream = createEditorBridgeStream(userId, 'site', controller.signal, { idleLeaseMs: 80 }) + + playEditor(stream, 50) + await sleep(10) + + // Four calls pushed together take ~200ms to execute serially — well past + // the 80ms lease. Every one must still succeed. + const bridge = getEditorBridgeForUser(userId, 'site') + const results = await Promise.all( + ['a', 'b', 'c', 'd'].map((name) => bridge!.callBrowser(name, {})), + ) + expect(results.map((r) => r.ok)).toEqual([true, true, true, true]) + expect(hasEditorBridge(userId, 'site')).toBe(true) + + controller.abort() + }) + + it('does not recycle the stream while a call is still pending', async () => { + const controller = new AbortController() + const userId = `lease-pending-${Date.now()}` + const stream = createEditorBridgeStream(userId, 'site', controller.signal, { idleLeaseMs: 40 }) + drain(stream) + await sleep(10) + + // Nothing answers: the call stays pending far past the lease. Closing the + // stream now would reject it while the editor may still be working on it. + const pending = getEditorBridgeForUser(userId, 'site')!.callBrowser('slow_tool', {}) + pending.catch(() => {}) + await sleep(150) + expect(hasEditorBridge(userId, 'site')).toBe(true) + + controller.abort() + await expect(pending).rejects.toThrow() + }) +})