From 6c8078d36a945ad64c4aef7bc92311338d77f4ba Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Tue, 1 Sep 2026 19:00:31 +0000 Subject: [PATCH] feat(runtime): MCP progress projector and warm-runtime proofs (#96) Born on the Effect render-event stream: emit notifications/progress only when the caller supplied a token, return one CallToolResult, and fail closed on epoch mismatch, restart, and a missing runtime. --- .changeset/mcp-progress-projector.md | 9 + packages/agent-bundle/src/build/entries.ts | 7 +- .../agent-bundle/src/build/entry-shell.ts | 104 ++++--- .../agent-bundle/src/build/inspect-bundler.ts | 7 +- .../agent-bundle/tests/entry-shell.test.ts | 11 + .../tests/generated-route-server.test.ts | 244 ++++++++++++++++ packages/rsc-runtime/README.md | 16 +- packages/rsc-runtime/src/dispatcher.ts | 2 + packages/rsc-runtime/src/effect/boundary.ts | 2 + packages/rsc-runtime/src/index.ts | 30 ++ packages/rsc-runtime/src/project-mcp.ts | 272 ++++++++++++++++++ packages/rsc-runtime/src/warm-runtime.ts | 94 ++++++ packages/rsc-runtime/tests/dispatcher.test.ts | 20 ++ .../rsc-runtime/tests/effect-boundary.test.ts | 11 + .../rsc-runtime/tests/mcp-projector.test.ts | 230 +++++++++++++++ .../rsc-runtime/tests/warm-runtime.test.ts | 92 ++++++ 16 files changed, 1106 insertions(+), 45 deletions(-) create mode 100644 .changeset/mcp-progress-projector.md create mode 100644 packages/rsc-runtime/src/project-mcp.ts create mode 100644 packages/rsc-runtime/src/warm-runtime.ts create mode 100644 packages/rsc-runtime/tests/mcp-projector.test.ts create mode 100644 packages/rsc-runtime/tests/warm-runtime.test.ts diff --git a/.changeset/mcp-progress-projector.md b/.changeset/mcp-progress-projector.md new file mode 100644 index 000000000..6e916e0f3 --- /dev/null +++ b/.changeset/mcp-progress-projector.md @@ -0,0 +1,9 @@ +--- +"@agent-bundle/runtime": minor +"agent-bundle": patch +--- + +Add the standards-compatible MCP progress/final projector and warm-runtime +fail-closed host. Generated tool calls emit `notifications/progress` only when +the caller supplied a token, return one `CallToolResult`, and refuse silent +rich-content drops, epoch mismatch, and a missing or restarted runtime. diff --git a/packages/agent-bundle/src/build/entries.ts b/packages/agent-bundle/src/build/entries.ts index 7d7948600..3b8116189 100644 --- a/packages/agent-bundle/src/build/entries.ts +++ b/packages/agent-bundle/src/build/entries.ts @@ -9,6 +9,7 @@ import { emitPlanEntries, resolveArtifactDestination } from './emit.ts'; import { scanEntryExports } from './entry-exports.ts'; import { generatedExecutableEntrySource, + generatedRouteArtifactEpoch, generatedRouteFlightWorkerSource, generatedRouteMcpEntrySource, generatedStdioMcpEntrySource, @@ -215,7 +216,11 @@ export const compileMcpEntries = async ( const server = servers.find((candidate) => candidate.id === entry.id); return server?.generatedRoutes === undefined ? undefined - : generatedRouteFlightWorkerSource({ routes: server.generatedRoutes, serverName: server.name }); + : generatedRouteFlightWorkerSource({ + artifactEpoch: generatedRouteArtifactEpoch(options.plugin), + routes: server.generatedRoutes, + serverName: server.name, + }); }); // Factory-exporting entries (default export) are wrapped in the framework // stdio lifecycle shell; self-connecting entries keep today's behavior byte diff --git a/packages/agent-bundle/src/build/entry-shell.ts b/packages/agent-bundle/src/build/entry-shell.ts index fe943e1e2..b37c89588 100644 --- a/packages/agent-bundle/src/build/entry-shell.ts +++ b/packages/agent-bundle/src/build/entry-shell.ts @@ -81,10 +81,16 @@ export interface GeneratedRouteMcpEntryOptions { } export interface GeneratedRouteFlightWorkerOptions { + readonly artifactEpoch: string; readonly routes: readonly CompiledAgentRoute[]; readonly serverName: string; } +export const generatedRouteArtifactEpoch = (plugin: { + readonly name: string; + readonly version: string; +}): string => `${plugin.name}@${plugin.version}`; + const routeProtocolName = (route: CompiledAgentRoute): string => route.id.slice(route.id.lastIndexOf('/') + 1); @@ -118,20 +124,30 @@ export const generatedRouteFlightWorkerSource = (options: GeneratedRouteFlightWo '// Generated routes contain only intrinsic Agent protocol elements, so no client references exist.', 'globalThis.__rspack_rsc_manifest__ ??= Object.freeze({ clientManifest: Object.freeze({}) });', "if (parentPort === null) throw new Error('Generated Flight worker requires a parent port.');", + 'process.stdout.write = process.stderr.write.bind(process.stderr);', + `const ARTIFACT_EPOCH = ${JSON.stringify(options.artifactEpoch)};`, + 'const processLifetime = { hits: 0, instanceId: crypto.randomUUID(), pid: process.pid };', 'const routes = Object.freeze({', ...routeRecords(routes), '});', 'const requests = new Map();', '', 'const render = async (message) => {', + ' if (message.artifactEpoch !== undefined && message.artifactEpoch !== ARTIFACT_EPOCH) {', + " parentPort.postMessage({ code: 'artifact-epoch-mismatch', id: message.id, message: `Runtime artifact epoch ${JSON.stringify(ARTIFACT_EPOCH)} does not match request epoch ${JSON.stringify(message.artifactEpoch)}`, receivedEpoch: message.artifactEpoch, type: 'error' });", + ' return;', + ' }', ' const route = routes[message.invocation.props.operationId];', " if (route === undefined || typeof route.module.default !== 'function') throw new TypeError('Generated MCP route must default-export an async Server Component.');", ' const controller = new AbortController();', ' requests.set(message.id, controller);', + ' processLifetime.hits += 1;', ' try {', ' const bytes = await runAgentRequest({', ' ...(message.actor === undefined ? {} : { actor: message.actor }),', - ' invocation: { kind: \'tool\', operationId: route.id, surface: route.name },', + ' invocation: { artifactEpoch: ARTIFACT_EPOCH, kind: \'tool\', operationId: route.id, surface: route.name },', + ' progress: { report: async (update) => { parentPort.postMessage({ id: message.id, type: \'progress\', update }); } },', + ' providers: { processLifetime: { hits: processLifetime.hits, instanceId: processLifetime.instanceId, pid: processLifetime.pid } },', ' ...(message.session === undefined ? {} : { session: message.session }),', ' signal: controller.signal,', ' }, async () => {', @@ -166,7 +182,10 @@ const routeRegistrations = (routes: readonly CompiledAgentRoute[]): readonly str ` ...${stableJson(config)},`, ` inputSchema: ${access}.module.inputSchema,`, ` outputSchema: ${access}.module.resultSchema,`, - ` }, async (input, context) => projectToolResult(await renderRoute(dispatcher, ${access}, input, context)));`, + ` }, async (input, context) => {`, + ` const rendered = await renderRoute(dispatcher, ${access}, input, context);`, + ' return attachMcpStructuredContent(rendered.toolResult, rendered.result);', + ' });', ].join('\n')); break; } @@ -210,52 +229,70 @@ const routeRegistrations = (routes: readonly CompiledAgentRoute[]): readonly str }; /** - * The generated MCP entry owns the final-only dispatcher and one warm Flight + * The generated MCP entry owns the stream projector and one warm Flight * worker. The worker is split only to satisfy React's react-server condition; * it is reused for every request until the MCP server closes. */ export const generatedRouteMcpEntrySource = (options: GeneratedRouteMcpEntryOptions): string => { const routes = executableMcpRoutes(options.routes); + const artifactEpoch = generatedRouteArtifactEpoch(options.plugin); return [ "import { Worker } from 'node:worker_threads';", "import { McpServer } from '@modelcontextprotocol/server';", - "import { agent, available, createAgentRenderDispatcher, runAgentRequest } from '@agent-bundle/runtime';", + "import { AgentRuntimeError, agent, attachMcpStructuredContent, available, createAgentRenderDispatcher, createWarmFlightHost, projectMcpRenderStream, runAgentRequest } from '@agent-bundle/runtime';", "import mcpApps from 'agent-bundle/mcp-apps';", ...routeImports(routes), '', + `const ARTIFACT_EPOCH = ${JSON.stringify(artifactEpoch)};`, 'const routes = Object.freeze({', ...routeRecords(routes), '});', '', + 'const workerError = (message) => {', + " if (message.code === 'artifact-epoch-mismatch') return new AgentRuntimeError('artifact-epoch-mismatch', message.message, { expectedEpoch: ARTIFACT_EPOCH, receivedEpoch: message.receivedEpoch });", + " if (message.code === 'runtime-unavailable') return new AgentRuntimeError('runtime-unavailable', message.message);", + " if (message.code === 'runtime-restarted') return new AgentRuntimeError('runtime-restarted', message.message);", + ' return new Error(message.message);', + '};', + '', 'const createWorkerHost = () => {', - ` const worker = new Worker(new URL(${JSON.stringify(`./${options.workerFile}`)}, import.meta.url));`, + ` const worker = new Worker(new URL(${JSON.stringify(`./${options.workerFile}`)}, import.meta.url), { stderr: true, stdout: true });`, + ' worker.stdout?.on(\'data\', (chunk) => process.stderr.write(chunk));', + ' worker.stderr?.on(\'data\', (chunk) => process.stderr.write(chunk));', ' const pending = new Map();', ' let sequence = 0;', + ' let exited = false;', ' const failPending = (error) => { for (const request of pending.values()) request.reject(error); pending.clear(); };', - ' worker.on(\'error\', failPending);', - ' worker.on(\'exit\', (code) => { if (code !== 0) failPending(new Error(`Generated Flight worker exited with code ${String(code)}.`)); });', + ' worker.on(\'error\', (error) => { exited = true; failPending(error); });', + ' worker.on(\'exit\', (code) => {', + ' exited = true;', + " failPending(new AgentRuntimeError(code === 0 ? 'runtime-unavailable' : 'runtime-restarted', code === 0 ? 'The MCP render runtime is unavailable' : `The MCP render runtime restarted; worker exited with code ${String(code)}.`));", + ' });', ' worker.on(\'message\', (message) => {', ' const request = pending.get(message.id);', ' if (request === undefined) return;', + " if (message.type === 'progress') { void request.progress?.report(message.update); return; }", ' pending.delete(message.id);', ' request.signal.removeEventListener(\'abort\', request.abort);', - " if (message.type === 'error') { request.reject(new Error(message.message)); return; }", + " if (message.type === 'error') { request.reject(workerError(message)); return; }", ' request.resolve(new ReadableStream({ start(controller) { controller.enqueue(message.bytes); controller.close(); } }));', ' });', - ' return Object.freeze({', + ' const host = Object.freeze({', ' close: async () => { await worker.terminate(); },', - ' execute: async ({ invocation, signal }) => {', + ' execute: async ({ artifactEpoch, invocation, progress, signal }) => {', + " if (exited) throw new AgentRuntimeError('runtime-unavailable', 'The MCP render runtime is unavailable');", ' const context = await agent();', ' const id = ++sequence;', ' return new Promise((resolve, reject) => {', " const abort = () => { worker.postMessage({ id, type: 'cancel' }); pending.delete(id); reject(new DOMException('Agent render was aborted', 'AbortError')); };", - ' pending.set(id, { abort, reject, resolve, signal });', + ' pending.set(id, { abort, progress, reject, resolve, signal });', " signal.addEventListener('abort', abort, { once: true });", ' if (signal.aborted) { abort(); return; }', - " worker.postMessage({ actor: context.actor, id, invocation, session: context.session, type: 'render' });", + ' worker.postMessage({ actor: context.actor, artifactEpoch: artifactEpoch ?? ARTIFACT_EPOCH, id, invocation, session: context.session, type: \'render\' });', ' });', ' },', ' });', + ' return createWarmFlightHost({ artifactEpoch: ARTIFACT_EPOCH, close: host.close, host });', '};', '', 'const requestIdentity = (context) => ({', @@ -263,37 +300,30 @@ export const generatedRouteMcpEntrySource = (options: GeneratedRouteMcpEntryOpti " ...(typeof context.sessionId === 'string' && context.sessionId.trim() !== '' ? { session: available({ sessionId: context.sessionId }, 'native') } : {}),", '});', '', + 'const mcpProjectorOptions = (context) => {', + ' const progressToken = context.mcpReq._meta?.progressToken;', + ' return {', + ' signal: context.mcpReq.signal,', + ' ...(progressToken === undefined ? {} : {', + ' progressToken,', + " sendProgress: (params) => context.mcpReq.notify({ method: 'notifications/progress', params }),", + ' }),', + ' };', + '};', + '', 'const renderRoute = async (dispatcher, route, input, context) => runAgentRequest({', ' ...requestIdentity(context),', - " invocation: { kind: 'tool', operationId: route.id, surface: route.name },", + " invocation: { artifactEpoch: ARTIFACT_EPOCH, kind: 'tool', operationId: route.id, surface: route.name },", ' signal: context.mcpReq.signal,', '}, async () => {', - ' const document = await dispatcher.dispatch({ invocation: { kind: \'tool\', props: { input, operationId: route.id } }, signal: context.mcpReq.signal });', - ' return { document, result: route.module.resultSchema.parse(document.value) };', + ' const projected = await projectMcpRenderStream(dispatcher.stream({', + ' artifactEpoch: ARTIFACT_EPOCH,', + " invocation: { kind: 'tool', props: { input, operationId: route.id } },", + ' signal: context.mcpReq.signal,', + ' }), mcpProjectorOptions(context));', + ' return { document: projected.document, result: route.module.resultSchema.parse(projected.document.value), toolResult: projected.result };', '});', '', - 'const appendNode = (node, content) => {', - ' switch (node.kind) {', - " case 'result': for (const child of node.children) appendNode(child, content); break;", - " case 'context':", - " case 'markdown':", - " case 'text': content.push({ text: node.text, type: 'text' }); break;", - " case 'json': content.push({ text: JSON.stringify(node.value), type: 'text' }); break;", - " case 'progress': content.push({ text: node.message ?? `Progress: ${node.completed}${node.total === undefined ? '' : `/${node.total}`}`, type: 'text' }); break;", - " case 'image': content.push({ data: node.data, mimeType: node.mimeType, type: 'image' }); break;", - " case 'audio': content.push({ data: node.data, mimeType: node.mimeType, type: 'audio' }); break;", - " case 'resource': content.push({ ...(node.mimeType === undefined ? {} : { mimeType: node.mimeType }), name: node.name, type: 'resource_link', uri: node.uri }); break;", - " case 'error': content.push({ text: `[${node.code}] ${node.message}`, type: 'text' }); break;", - " default: throw new TypeError(`Unsupported Agent Document node: ${String(node.kind)}`);", - ' }', - '};', - '', - 'const projectToolResult = ({ document, result }) => {', - ' const content = [];', - ' appendNode(document.root, content);', - ' return { content, ...(document.status === \'represented-error\' ? { isError: true } : {}), ...(result !== null && typeof result === \'object\' && !Array.isArray(result) ? { structuredContent: result } : {}) };', - '};', - '', 'const createGeneratedRouteServer = () => {', ` const server = new McpServer(${stableJson(options.plugin)});`, ' const workerHost = createWorkerHost();', diff --git a/packages/agent-bundle/src/build/inspect-bundler.ts b/packages/agent-bundle/src/build/inspect-bundler.ts index 9bae7ba99..b4670a770 100644 --- a/packages/agent-bundle/src/build/inspect-bundler.ts +++ b/packages/agent-bundle/src/build/inspect-bundler.ts @@ -3,6 +3,7 @@ import type { AgentBundleToolsConfig, NormalizedPlugin } from '../core/types.ts' import { scanEntryExports } from './entry-exports.ts'; import { generatedExecutableEntrySource, + generatedRouteArtifactEpoch, generatedRouteFlightWorkerSource, generatedRouteMcpEntrySource, generatedStdioMcpEntrySource, @@ -211,7 +212,11 @@ const mcpEntryEntries = async ( rscManifest: true, source: entry.source, sourceInputs: [], - virtualSource: generatedRouteFlightWorkerSource({ routes: generatedRoutes, serverName }), + virtualSource: generatedRouteFlightWorkerSource({ + artifactEpoch: generatedRouteArtifactEpoch({ name: model.metadata.name, version: model.metadata.version }), + routes: generatedRoutes, + serverName, + }), }, kind: 'mcp-entry', name: `${serverName}:flight`, diff --git a/packages/agent-bundle/tests/entry-shell.test.ts b/packages/agent-bundle/tests/entry-shell.test.ts index 03607bcde..51e7f9a1c 100644 --- a/packages/agent-bundle/tests/entry-shell.test.ts +++ b/packages/agent-bundle/tests/entry-shell.test.ts @@ -127,12 +127,20 @@ it('generates one final-only Flight MCP factory from filesystem routes', () => { expect(source).toContain("from '@modelcontextprotocol/server'"); expect(source).toContain("from 'agent-bundle/mcp-apps'"); expect(source).toContain('createAgentRenderDispatcher'); + expect(source).toContain('createWarmFlightHost'); + expect(source).toContain('{ stderr: true, stdout: true }'); + expect(source).toContain('projectMcpRenderStream'); + expect(source).toContain('attachMcpStructuredContent'); expect(source).toContain('runAgentRequest'); + expect(source).toContain('notifications/progress'); + expect(source).toContain('ARTIFACT_EPOCH'); + expect(source).toContain('route-fixture@1.2.3'); expect(source).toContain('server.registerTool("inspect"'); expect(source).toContain('server.registerResource("catalog", "catalog://books"'); expect(source).toContain('server.registerPrompt("curate"'); expect(source).toContain('export default createGeneratedRouteServer'); expect(source).not.toContain('lowerMcpResult'); + expect(source).not.toContain('projectToolResult'); }); @@ -143,6 +151,7 @@ it('generates the warm react-server Flight worker separately from the MCP dispat expect(typeof generate).toBe('function'); if (generate === undefined) return; const source = generate({ + artifactEpoch: 'route-fixture@1.2.3', routes: [{ config: {}, id: 'tool:curator/inspect', @@ -154,5 +163,7 @@ it('generates the warm react-server Flight worker separately from the MCP dispat expect(source).toContain("from '@agent-bundle/runtime/flight/server'"); expect(source).toContain("from 'node:worker_threads'"); expect(source).toContain('runAgentRequest'); + expect(source).toContain('processLifetime'); + expect(source).toContain('route-fixture@1.2.3'); expect(source).toContain('/project/src/mcp/curator/tools/inspect.tsx'); }); diff --git a/packages/agent-bundle/tests/generated-route-server.test.ts b/packages/agent-bundle/tests/generated-route-server.test.ts index 035d9d1de..266e240ce 100644 --- a/packages/agent-bundle/tests/generated-route-server.test.ts +++ b/packages/agent-bundle/tests/generated-route-server.test.ts @@ -132,3 +132,247 @@ it('lists and calls a generated filesystem tool through final-only Flight', { re await client.close(); } }); + +const writeGeneratedProject = async ( + root: string, + files: Readonly>, +): Promise => { + await symlink(join(process.cwd(), 'examples', 'audiobook-curator', 'node_modules'), join(root, 'node_modules'), 'dir'); + await Promise.all([ + writeProjectFile(root, 'package.json', JSON.stringify({ + dependencies: { + '@agent-bundle/runtime': 'workspace:*', + '@modelcontextprotocol/server': '2.0.0', + react: '19.2.8', + zod: '4.4.3', + }, + name: 'generated-routes-fixture', + type: 'module', + version: '1.0.0', + })), + writeProjectFile(root, 'agent-bundle.config.ts', [ + "import { defineConfig } from 'agent-bundle/config';", + "export default defineConfig({ plugin: { name: 'generated-routes-fixture', version: '1.0.0' }, targets: ['portable'] });", + '', + ].join('\n')), + ...Object.entries(files).map(([path, contents]) => writeProjectFile(root, path, contents)), + ]); +}; + +const connectGeneratedServer = async (root: string): Promise<{ + readonly client: Client; + readonly close: () => Promise; +}> => { + const output = join(root, 'artifact'); + const compiled = await build({ output, root, targets: ['portable'] }); + const server = compiled.model.mcpServers[0]; + if (server?.args?.[0] === undefined) throw new Error('expected a generated MCP entry'); + const client = new Client({ name: 'generated-route-test', version: '0.0.0' }); + const transport = new StdioClientTransport({ + args: [join(output, 'portable', server.args[0])], + command: process.execPath, + stderr: 'pipe', + }); + let diagnostics = ''; + transport.stderr?.on('data', (chunk) => { diagnostics += String(chunk); }); + try { + await client.connect(transport); + } catch (error) { + throw new Error(`Generated route server failed to connect: ${diagnostics}`, { cause: error }); + } + return { + client, + close: async () => { + await client.close(); + }, + }; +}; + +const callGeneratedTool = async (client: Client, name: string): Promise => { + try { + return await client.callTool({ arguments: {}, name }, { signal: AbortSignal.timeout(10_000) }); + } catch (error) { + return error; + } +}; + +const expectFailClosed = (outcome: unknown, message: RegExp): void => { + if (outcome instanceof Error) { + expect(`${outcome.name} ${outcome.message}`).toMatch(message); + return; + } + expect(outcome).toMatchObject({ isError: true }); + expect(JSON.stringify(outcome)).toMatch(message); +}; + +it('observes one process-lifetime provider across consecutive generated tool calls', { retry: 2, timeout: 60_000 }, async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-generated-warm-')); + roots.push(root); + await writeGeneratedProject(root, { + 'src/mcp/curator/tools/warmth.tsx': [ + "import { Agent, agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + "export const config = { annotations: { readOnlyHint: true }, description: 'Observe process lifetime.' };", + 'export const inputSchema = z.object({}).strict();', + 'export const resultSchema = z.object({ hits: z.number(), instanceId: z.string(), pid: z.number() }).strict();', + 'export default async function Warmth() {', + ' const context = await agent();', + ' const processLifetime = context.providers.processLifetime;', + " if (processLifetime === undefined || typeof processLifetime !== 'object' || processLifetime === null) {", + " throw new Error('process-lifetime provider was not installed');", + ' }', + ' const value = processLifetime as { hits: number; instanceId: string; pid: number };', + " return createElement(Agent.Result, { value }, createElement(Agent.Text, null, `hit ${String(value.hits)}`));", + '}', + '', + ].join('\n'), + }); + const session = await connectGeneratedServer(root); + try { + const first = await session.client.callTool({ arguments: {}, name: 'warmth' }, { signal: AbortSignal.timeout(10_000) }); + const second = await session.client.callTool({ arguments: {}, name: 'warmth' }, { signal: AbortSignal.timeout(10_000) }); + expect(first).toMatchObject({ + content: [{ text: 'hit 1', type: 'text' }], + structuredContent: { hits: 1 }, + }); + expect(second).toMatchObject({ + content: [{ text: 'hit 2', type: 'text' }], + structuredContent: { hits: 2 }, + }); + const firstId = (first.structuredContent as { instanceId: string }).instanceId; + const secondContent = second.structuredContent as { instanceId: string; pid: number }; + expect(secondContent.instanceId).toBe(firstId); + expect(secondContent.pid).toBe((first.structuredContent as { pid: number }).pid); + } finally { + await session.close(); + } +}); + +it('emits MCP progress notifications only when a progress token is supplied', { retry: 2, timeout: 60_000 }, async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-generated-progress-')); + roots.push(root); + await writeGeneratedProject(root, { + 'src/mcp/curator/tools/progress.tsx': [ + "import { Agent, agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + "export const config = { annotations: { readOnlyHint: true }, description: 'Report progress.' };", + 'export const inputSchema = z.object({}).strict();', + 'export const resultSchema = z.object({ ok: z.literal(true) }).strict();', + 'export default async function Progress() {', + ' const context = await agent();', + " await context.progress.report({ completed: 1, message: 'halfway', total: 2 });", + " await context.progress.report({ completed: 2, message: 'done', total: 2 });", + " return createElement(Agent.Result, { value: { ok: true } }, createElement(Agent.Text, null, 'finished'));", + '}', + '', + ].join('\n'), + }); + const session = await connectGeneratedServer(root); + const notifications: Array<{ readonly message?: string; readonly progress: number; readonly progressToken: string | number; readonly total?: number }> = []; + session.client.setNotificationHandler('notifications/progress', (notification) => { + notifications.push(notification.params); + }); + try { + await expect(session.client.callTool({ arguments: {}, name: 'progress' }, { signal: AbortSignal.timeout(10_000) })).resolves.toMatchObject({ + content: [{ text: 'finished', type: 'text' }], + structuredContent: { ok: true }, + }); + expect(notifications).toEqual([]); + await expect(session.client.callTool({ + arguments: {}, + name: 'progress', + _meta: { progressToken: 'tok-1' }, + }, { signal: AbortSignal.timeout(10_000) })).resolves.toMatchObject({ + structuredContent: { ok: true }, + }); + expect(notifications).toEqual([ + { message: 'halfway', progress: 1, progressToken: 'tok-1', total: 2 }, + { message: 'done', progress: 2, progressToken: 'tok-1', total: 2 }, + ]); + } finally { + await session.close(); + } +}); + +it('maps notifications/cancelled into the renderer AbortSignal', { retry: 2, timeout: 60_000 }, async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-generated-cancel-')); + roots.push(root); + await writeGeneratedProject(root, { + 'src/mcp/curator/tools/hang.tsx': [ + "import { Agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + "export const config = { annotations: { readOnlyHint: true }, description: 'Hang until cancelled.' };", + 'export const inputSchema = z.object({}).strict();', + 'export const resultSchema = z.object({ ok: z.literal(true) }).strict();', + 'export default async function Hang({ signal }) {', + ' await new Promise((_, reject) => {', + " const fail = () => reject(new DOMException('aborted', 'AbortError'));", + ' if (signal.aborted) { fail(); return; }', + " signal.addEventListener('abort', fail, { once: true });", + ' });', + " return createElement(Agent.Result, { value: { ok: true } }, createElement(Agent.Text, null, 'should not complete'));", + '}', + '', + ].join('\n'), + }); + const session = await connectGeneratedServer(root); + try { + const controller = new AbortController(); + const pending = session.client.callTool({ arguments: {}, name: 'hang' }, { signal: controller.signal }); + await Promise.resolve(); + controller.abort(); + await expect(pending).rejects.toThrow(/abort/i); + } finally { + await session.close(); + } +}); + +it('fails closed when the generated runtime worker restarts', { retry: 2, timeout: 60_000 }, async () => { + const root = await mkdtemp(join(tmpdir(), 'agent-bundle-generated-restart-')); + roots.push(root); + await writeGeneratedProject(root, { + 'src/mcp/curator/tools/warmth.tsx': [ + "import { Agent, agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + "export const config = { annotations: { readOnlyHint: true }, description: 'Observe process lifetime.' };", + 'export const inputSchema = z.object({}).strict();', + 'export const resultSchema = z.object({ hits: z.number(), instanceId: z.string(), pid: z.number() }).strict();', + 'export default async function Warmth() {', + ' const context = await agent();', + ' const value = context.providers.processLifetime as { hits: number; instanceId: string; pid: number };', + " return createElement(Agent.Result, { value }, createElement(Agent.Text, null, `hit ${String(value.hits)}`));", + '}', + '', + ].join('\n'), + 'src/mcp/curator/tools/halt.tsx': [ + "import { Agent } from '@agent-bundle/runtime';", + "import { createElement } from 'react';", + "import { z } from 'zod';", + "export const config = { description: 'Halt the Flight worker.' };", + 'export const inputSchema = z.object({}).strict();', + 'export const resultSchema = z.object({ ok: z.literal(true) }).strict();', + 'export default async function Halt() {', + ' process.exit(1);', + " return createElement(Agent.Result, { value: { ok: true } }, createElement(Agent.Text, null, 'halted'));", + '}', + '', + ].join('\n'), + }); + const session = await connectGeneratedServer(root); + try { + await expect(session.client.callTool({ arguments: {}, name: 'warmth' }, { signal: AbortSignal.timeout(10_000) })).resolves.toMatchObject({ + structuredContent: { hits: 1 }, + }); + const halted = await callGeneratedTool(session.client, 'halt'); + expectFailClosed(halted, /unavailable|restarted|exited/i); + const afterRestart = await callGeneratedTool(session.client, 'warmth'); + expectFailClosed(afterRestart, /unavailable|restarted|exited|connection closed/i); + expect(afterRestart).not.toMatchObject({ structuredContent: { hits: 2 } }); + } finally { + await session.close(); + } +}); diff --git a/packages/rsc-runtime/README.md b/packages/rsc-runtime/README.md index 1a8ff0d81..3bec5b76a 100644 --- a/packages/rsc-runtime/README.md +++ b/packages/rsc-runtime/README.md @@ -15,10 +15,14 @@ stream; a post-completion producer is rejected with a typed `handoff-required` outcome. Depth, node count, bytes, event rate, and elapsed time are bounded on the reconciler. -The existing lowerers remain synchronous compatibility APIs. `lowerMcpResult` -walks an MCP element tree, calling function components itself, and lowers it -into a plain `CallToolResult`; `lowerHookResult` does the same for `Hook.*`. -They remain the operative MCP path until the later projector migration. +Generated MCP tool calls project the live render-event stream through +`projectMcpRenderStream`: `notifications/progress` is emitted only when the +caller supplied a progress token, shell/replace stay internal, and the +request resolves to one final `CallToolResult`. Image, audio, and resource +blocks are capability-gated — unsupported rich content uses a declared +fallback or a typed `McpProjectionError`, never a silent drop. The existing +`lowerMcpResult` / `lowerHookResult` helpers remain synchronous compatibility +APIs for the operations-model path. ```tsx import { Mcp, lowerMcpResult } from '@agent-bundle/runtime'; @@ -41,8 +45,8 @@ synchronous compatibility APIs remain operative. The package exports `Hook`, `Mcp`, `Agent`, both lowerers, the request-store APIs, the Agent Document contracts, `createAgentRenderDispatcher`, -`decodeAgentFlightStream`, and the `@agent-bundle/runtime/flight/server` -render entry. The Flight-facing versions +`projectMcpRenderStream`, `createWarmFlightHost`, `decodeAgentFlightStream`, +and the `@agent-bundle/runtime/flight/server` render entry. The Flight-facing versions are exact compatibility pins: React/React DOM `19.2.8` and `react-server-dom-rspack` `0.1.0`; the proof example compiles them with `rsbuild-plugin-rsc` `0.1.1`. The package does not own application state, diff --git a/packages/rsc-runtime/src/dispatcher.ts b/packages/rsc-runtime/src/dispatcher.ts index f41e75314..09d53f8e9 100644 --- a/packages/rsc-runtime/src/dispatcher.ts +++ b/packages/rsc-runtime/src/dispatcher.ts @@ -11,6 +11,7 @@ import { createAgentRenderEventSession, toPublicEventStream } from './reconciler export { decodeAgentDocument } from './decode-document.js'; export interface AgentRenderDispatch { + readonly artifactEpoch?: string; readonly invocation: AgentRenderInvocation; readonly progress?: AgentProgressReporter; readonly signal: AbortSignal; @@ -98,6 +99,7 @@ export const createAgentRenderDispatcher = ( }; try { pendingFlight.current = rememberFlight(host.execute({ + ...(request.artifactEpoch === undefined ? {} : { artifactEpoch: request.artifactEpoch }), invocation: request.invocation, progress: session.progress, signal: request.signal, diff --git a/packages/rsc-runtime/src/effect/boundary.ts b/packages/rsc-runtime/src/effect/boundary.ts index b09fe566d..68a250f7e 100644 --- a/packages/rsc-runtime/src/effect/boundary.ts +++ b/packages/rsc-runtime/src/effect/boundary.ts @@ -29,7 +29,9 @@ export interface RunPromiseOptions { const TYPED_ERROR_NAMES = new Set([ 'AgentContractError', 'AgentRequestError', + 'AgentRuntimeError', 'AgentStateError', + 'McpProjectionError', ]); const interruptAs = (): Effect.Effect => diff --git a/packages/rsc-runtime/src/index.ts b/packages/rsc-runtime/src/index.ts index c03e9cb30..9764e6b02 100644 --- a/packages/rsc-runtime/src/index.ts +++ b/packages/rsc-runtime/src/index.ts @@ -48,6 +48,36 @@ export type { AgentRenderDispatcher, AgentRenderDispatcherOptions, } from './dispatcher.js'; +export { + attachMcpStructuredContent, + DEFAULT_MCP_RICH_CONTENT_CAPABILITIES, + documentToCallToolResult, + MCP_PROGRESS_MESSAGE_MAX, + McpProjectionError, + projectMcpRenderStream, + shortenMcpProgressMessage, +} from './project-mcp.js'; +export type { + McpProjectedToolResult, + McpProgressNotificationParams, + McpProgressToken, + McpProjectionErrorCode, + McpRichContentCapabilities, + McpRichContentFallback, + McpRichContentKind, + ProjectMcpRenderOptions, +} from './project-mcp.js'; +export { + AgentRuntimeError, + assertArtifactEpoch, + createWarmFlightHost, +} from './warm-runtime.js'; +export type { + AgentRuntimeErrorCode, + CreateWarmFlightHostOptions, + WarmFlightHost, + WarmRuntimeIdentity, +} from './warm-runtime.js'; export { decodeAgentFlightStream } from './reconciler.js'; export type { AgentFlightDecodeOptions } from './reconciler.js'; export { lowerHookResult } from './lower-hook.js'; diff --git a/packages/rsc-runtime/src/project-mcp.ts b/packages/rsc-runtime/src/project-mcp.ts new file mode 100644 index 000000000..b635d49fc --- /dev/null +++ b/packages/rsc-runtime/src/project-mcp.ts @@ -0,0 +1,272 @@ +import type { CallToolResult } from '@modelcontextprotocol/sdk/types.js'; +import { Effect, Stream } from 'effect'; + +import { + AgentContractError, + type AgentDocument, + type AgentDocumentNode, + type AgentRenderEvent, +} from './agent-document.js'; +import { interruptWhenAborted, runPromise, toRuntimeError } from './effect/boundary.js'; +import { snapshotJsonValue, type JsonObject, type JsonValue } from './lower-mcp.js'; + +export const MCP_PROGRESS_MESSAGE_MAX = 200; + +export type McpProgressToken = string | number; + +export interface McpProgressNotificationParams { + readonly message?: string; + readonly progress: number; + readonly progressToken: McpProgressToken; + readonly total?: number; +} + +export interface McpRichContentCapabilities { + readonly audio: boolean; + readonly image: boolean; + readonly resource: boolean; +} + +export const DEFAULT_MCP_RICH_CONTENT_CAPABILITIES: McpRichContentCapabilities = Object.freeze({ + audio: true, + image: true, + resource: true, +}); + +export type McpRichContentFallback = 'text' | 'fail'; + +export type McpProjectionErrorCode = 'unsupported-rich-content'; + +export type McpRichContentKind = 'audio' | 'image' | 'resource'; + +export class McpProjectionError extends Error { + readonly code: McpProjectionErrorCode; + readonly kind?: McpRichContentKind; + + constructor( + code: McpProjectionErrorCode, + message: string, + options?: ErrorOptions & { readonly kind?: McpRichContentKind }, + ) { + super(message, options); + this.code = code; + this.name = 'McpProjectionError'; + this.kind = options?.kind; + } +} + +export interface ProjectMcpRenderOptions { + readonly capabilities?: Partial; + readonly progressToken?: McpProgressToken; + readonly richContentFallback?: McpRichContentFallback; + readonly sendProgress?: (params: McpProgressNotificationParams) => Promise; + readonly signal?: AbortSignal; + readonly structuredContent?: unknown; +} + +export interface McpProjectedToolResult { + readonly document: AgentDocument; + readonly result: CallToolResult; +} + +const resolveCapabilities = ( + partial?: Partial, +): McpRichContentCapabilities => Object.freeze({ + ...DEFAULT_MCP_RICH_CONTENT_CAPABILITIES, + ...partial, +}); + +export const shortenMcpProgressMessage = (message: string): string => { + const trimmed = message.trim(); + if (trimmed.length <= MCP_PROGRESS_MESSAGE_MAX) return trimmed; + return `${trimmed.slice(0, MCP_PROGRESS_MESSAGE_MAX - 1)}…`; +}; + +const gatedBlock = ( + kind: McpRichContentKind, + summary: string, + fallback: McpRichContentFallback, +): CallToolResult['content'][number] => { + switch (fallback) { + case 'text': + return { text: summary, type: 'text' }; + case 'fail': + throw new McpProjectionError( + 'unsupported-rich-content', + `MCP projector cannot emit ${kind} content because the selected capability does not support it`, + { kind }, + ); + default: { + const exhaustive: never = fallback; + return exhaustive; + } + } +}; + +type McpContentBlock = CallToolResult['content'][number]; + +const appendNode = ( + node: AgentDocumentNode, + content: McpContentBlock[], + capabilities: McpRichContentCapabilities, + fallback: McpRichContentFallback, +): void => { + switch (node.kind) { + case 'result': + for (const child of node.children) appendNode(child, content, capabilities, fallback); + break; + case 'context': + case 'markdown': + case 'text': + content.push({ text: node.text, type: 'text' }); + break; + case 'json': + content.push({ text: JSON.stringify(node.value), type: 'text' }); + break; + case 'progress': + break; + case 'image': + content.push(capabilities.image + ? { data: node.data, mimeType: node.mimeType, type: 'image' } + : gatedBlock('image', `[image ${node.mimeType}]`, fallback)); + break; + case 'audio': + content.push(capabilities.audio + ? { data: node.data, mimeType: node.mimeType, type: 'audio' } + : gatedBlock('audio', `[audio ${node.mimeType}]`, fallback)); + break; + case 'resource': + content.push(capabilities.resource + ? { + ...(node.mimeType === undefined ? {} : { mimeType: node.mimeType }), + name: node.name, + type: 'resource_link', + uri: node.uri, + } + : gatedBlock('resource', `[resource ${node.name} ${node.uri}]`, fallback)); + break; + case 'error': + content.push({ text: `[${node.code}] ${node.message}`, type: 'text' }); + break; + default: { + const exhaustive: never = node; + throw new AgentContractError( + 'invalid-document', + `Unsupported Agent Document node: ${String((exhaustive as { kind?: unknown }).kind)}`, + ); + } + } +}; + +const isJsonObject = (value: JsonValue): value is JsonObject => + value !== null && typeof value === 'object' && !Array.isArray(value); + +const objectStructuredContent = (value: unknown): JsonObject | undefined => { + if (value === undefined) return undefined; + const snapshot = snapshotJsonValue(value, 'MCP structured content must be JSON-serializable'); + return isJsonObject(snapshot) ? snapshot : undefined; +}; + +export const documentToCallToolResult = ( + document: AgentDocument, + options: Pick = {}, +): CallToolResult => { + const content: McpContentBlock[] = []; + appendNode( + document.root, + content, + resolveCapabilities(options.capabilities), + options.richContentFallback ?? 'fail', + ); + const structured = objectStructuredContent(options.structuredContent ?? document.value); + return { + content, + ...(document.status === 'success' ? {} : { isError: true }), + ...(structured === undefined ? {} : { structuredContent: structured }), + }; +}; + +export const attachMcpStructuredContent = ( + result: CallToolResult, + value: unknown, +): CallToolResult => { + const structured = objectStructuredContent(value); + if (structured === undefined) return result; + return { ...result, structuredContent: structured }; +}; + +const notifyProgress = ( + event: Extract, + token: McpProgressToken, + sendProgress: (params: McpProgressNotificationParams) => Promise, +): Effect.Effect => + Effect.tryPromise({ + catch: (error) => toRuntimeError(error), + try: () => sendProgress({ + progress: event.completed, + progressToken: token, + ...(event.message === undefined ? {} : { message: shortenMcpProgressMessage(event.message) }), + ...(event.total === undefined ? {} : { total: event.total }), + }), + }); + +export const projectMcpEventStream = Effect.fnUntraced(function*( + events: Stream.Stream, + options: ProjectMcpRenderOptions = {}, +) { + let lastProgress = Number.NEGATIVE_INFINITY; + let complete: AgentDocument | undefined; + const token = options.progressToken; + const sendProgress = options.sendProgress; + yield* Stream.runForEach(events, (event) => + Effect.gen(function*() { + switch (event.type) { + case 'progress': { + if (token === undefined || sendProgress === undefined) return; + if (!(event.completed > lastProgress)) return; + lastProgress = event.completed; + yield* notifyProgress(event, token, sendProgress); + return; + } + case 'shell': + case 'replace': + case 'error': + return; + case 'complete': + complete = event.document; + return; + default: { + const exhaustive: never = event; + return exhaustive; + } + } + }), + ); + if (complete === undefined) { + return yield* Effect.fail(new AgentContractError( + 'invalid-document', + 'MCP projector requires a complete document; the stream ended without one', + )); + } + return Object.freeze({ + document: complete, + result: documentToCallToolResult(complete, options), + }); +}); + +export const projectMcpRenderStream = async ( + events: ReadableStream, + options: ProjectMcpRenderOptions = {}, +): Promise => { + const program = projectMcpEventStream( + Stream.fromReadableStream({ + evaluate: () => events, + onError: (error) => toRuntimeError(error), + }), + options, + ); + return runPromise( + options.signal === undefined ? program : interruptWhenAborted(program, options.signal), + options.signal === undefined ? undefined : { signal: options.signal }, + ); +}; diff --git a/packages/rsc-runtime/src/warm-runtime.ts b/packages/rsc-runtime/src/warm-runtime.ts new file mode 100644 index 000000000..d1813176a --- /dev/null +++ b/packages/rsc-runtime/src/warm-runtime.ts @@ -0,0 +1,94 @@ +import type { AgentFlightExecutionHost, AgentRenderDispatch } from './dispatcher.js'; + +export type AgentRuntimeErrorCode = + | 'artifact-epoch-mismatch' + | 'runtime-restarted' + | 'runtime-unavailable'; + +export class AgentRuntimeError extends Error { + readonly code: AgentRuntimeErrorCode; + readonly expectedEpoch?: string; + readonly receivedEpoch?: string; + + constructor( + code: AgentRuntimeErrorCode, + message: string, + options?: ErrorOptions & { + readonly expectedEpoch?: string; + readonly receivedEpoch?: string; + }, + ) { + super(message, options); + this.code = code; + this.name = 'AgentRuntimeError'; + this.expectedEpoch = options?.expectedEpoch; + this.receivedEpoch = options?.receivedEpoch; + } +} + +export const assertArtifactEpoch = (expected: string, received: string | undefined): void => { + if (received === undefined || received === expected) return; + throw new AgentRuntimeError( + 'artifact-epoch-mismatch', + `Runtime artifact epoch ${JSON.stringify(expected)} does not match request epoch ${JSON.stringify(received)}`, + { expectedEpoch: expected, receivedEpoch: received }, + ); +}; + +export interface WarmRuntimeIdentity { + readonly artifactEpoch: string; + readonly instanceId: string; +} + +export interface WarmFlightHost extends AgentFlightExecutionHost { + readonly close: () => Promise; + readonly identity: WarmRuntimeIdentity; + readonly markUnavailable: (code?: Exclude) => void; +} + +export interface CreateWarmFlightHostOptions { + readonly artifactEpoch: string; + readonly close?: () => Promise; + readonly host: AgentFlightExecutionHost; + readonly instanceId?: string; +} + +const unavailableError = ( + code: Exclude, +): AgentRuntimeError => { + switch (code) { + case 'runtime-restarted': + return new AgentRuntimeError( + code, + 'The MCP render runtime restarted; this process no longer serves requests', + ); + case 'runtime-unavailable': + return new AgentRuntimeError(code, 'The MCP render runtime is unavailable'); + default: { + const exhaustive: never = code; + return exhaustive; + } + } +}; + +export const createWarmFlightHost = (options: CreateWarmFlightHostOptions): WarmFlightHost => { + const identity: WarmRuntimeIdentity = Object.freeze({ + artifactEpoch: options.artifactEpoch, + instanceId: options.instanceId ?? crypto.randomUUID(), + }); + let unavailable: AgentRuntimeError | undefined; + return Object.freeze({ + identity, + markUnavailable(code: Exclude = 'runtime-unavailable') { + unavailable ??= unavailableError(code); + }, + async close() { + await options.close?.(); + }, + async execute(request: AgentRenderDispatch) { + if (unavailable !== undefined) throw unavailable; + assertArtifactEpoch(identity.artifactEpoch, request.artifactEpoch); + return options.host.execute(request); + }, + }); +}; diff --git a/packages/rsc-runtime/tests/dispatcher.test.ts b/packages/rsc-runtime/tests/dispatcher.test.ts index d7893e8e2..8c9d65ba0 100644 --- a/packages/rsc-runtime/tests/dispatcher.test.ts +++ b/packages/rsc-runtime/tests/dispatcher.test.ts @@ -100,6 +100,26 @@ describe('AgentRenderDispatcher', () => { })).rejects.toMatchObject({ name: 'AbortError' }); expect(calls).toBe(0); }); + + it('forwards artifactEpoch to the execution host', () => { + const seen: Array = []; + const dispatcher = createAgentRenderDispatcher({ + execute: async (request) => { + seen.push(request.artifactEpoch); + return new ReadableStream({ + start(controller) { + controller.close(); + }, + }); + }, + }); + dispatcher.stream({ + artifactEpoch: 'epoch-a', + invocation: { kind: 'event', props: { event: 'tool/after', payload: {} } }, + signal: new AbortController().signal, + }); + expect(seen).toEqual(['epoch-a']); + }); }); diff --git a/packages/rsc-runtime/tests/effect-boundary.test.ts b/packages/rsc-runtime/tests/effect-boundary.test.ts index 310ba9206..efec73939 100644 --- a/packages/rsc-runtime/tests/effect-boundary.test.ts +++ b/packages/rsc-runtime/tests/effect-boundary.test.ts @@ -45,6 +45,17 @@ describe('effect boundary', () => { await expect(runPromise(Effect.fail(stateError))).rejects.toBe(stateError); }); + it('rethrows AgentRuntimeError and McpProjectionError by name', async () => { + const runtime = new Error('runtime unavailable'); + runtime.name = 'AgentRuntimeError'; + const projection = new Error('unsupported image'); + projection.name = 'McpProjectionError'; + expect(isTypedRuntimeError(runtime)).toBe(true); + expect(isTypedRuntimeError(projection)).toBe(true); + await expect(runPromise(Effect.fail(runtime))).rejects.toBe(runtime); + await expect(runPromise(Effect.fail(projection))).rejects.toBe(projection); + }); + it('maps interruption to DOMException AbortError', async () => { const mapped = mapCause(Cause.interrupt(1)); expect(isAbortError(mapped)).toBe(true); diff --git a/packages/rsc-runtime/tests/mcp-projector.test.ts b/packages/rsc-runtime/tests/mcp-projector.test.ts new file mode 100644 index 000000000..f5b3d80ea --- /dev/null +++ b/packages/rsc-runtime/tests/mcp-projector.test.ts @@ -0,0 +1,230 @@ +import { describe, expect, it } from '@rstest/core'; + +import { + AgentContractError, + createAgentDocument, + McpProjectionError, + projectMcpRenderStream, + type AgentDocument, + type AgentRenderEvent, + type McpProgressNotificationParams, +} from '../src/index.js'; + +const document = (overrides: Partial = {}): AgentDocument => + createAgentDocument({ + root: { + children: [{ kind: 'text', text: 'Ready.' }], + kind: 'result', + }, + status: 'success', + value: { ok: true }, + version: 1, + ...overrides, + }); + +const eventsOf = (events: readonly AgentRenderEvent[]): ReadableStream => + new ReadableStream({ + start(controller) { + for (const event of events) controller.enqueue(event); + controller.close(); + }, + }); + +describe('projectMcpRenderStream', () => { + it('emits notifications/progress only when the caller supplied a progress token', async () => { + const sent: McpProgressNotificationParams[] = []; + const complete = document(); + const withoutToken = await projectMcpRenderStream(eventsOf([ + { completed: 1, message: 'halfway', sequence: 0, total: 2, type: 'progress' }, + { document: complete, sequence: 1, type: 'complete' }, + ]), { + sendProgress: async (params) => { + sent.push(params); + }, + }); + + expect(sent).toEqual([]); + expect(withoutToken.result).toEqual({ + content: [{ text: 'Ready.', type: 'text' }], + structuredContent: { ok: true }, + }); + + const withToken = await projectMcpRenderStream(eventsOf([ + { completed: 1, message: 'halfway', sequence: 0, total: 2, type: 'progress' }, + { completed: 2, message: 'done', sequence: 1, total: 2, type: 'progress' }, + { document: complete, sequence: 2, type: 'complete' }, + ]), { + progressToken: 'tok-1', + sendProgress: async (params) => { + sent.push(params); + }, + }); + + expect(sent).toEqual([ + { message: 'halfway', progress: 1, progressToken: 'tok-1', total: 2 }, + { message: 'done', progress: 2, progressToken: 'tok-1', total: 2 }, + ]); + expect(withToken.result.content).toEqual([{ text: 'Ready.', type: 'text' }]); + expect(withToken.document).toEqual(complete); + }); + + it('maps only monotonically increasing numeric progress and shortens the message', async () => { + const sent: McpProgressNotificationParams[] = []; + const long = `inspecting ${'x'.repeat(240)}`; + await projectMcpRenderStream(eventsOf([ + { completed: 2, message: 'second', sequence: 0, type: 'progress' }, + { completed: 2, message: 'repeat', sequence: 1, type: 'progress' }, + { completed: 1, message: 'rewind', sequence: 2, type: 'progress' }, + { completed: 4, message: long, sequence: 3, total: 10, type: 'progress' }, + { document: document(), sequence: 4, type: 'complete' }, + ]), { + progressToken: 7, + sendProgress: async (params) => { + sent.push(params); + }, + }); + + expect(sent).toHaveLength(2); + expect(sent[0]).toEqual({ message: 'second', progress: 2, progressToken: 7 }); + expect(sent[1]?.progress).toBe(4); + expect(sent[1]?.total).toBe(10); + expect(sent[1]?.message?.length).toBeLessThanOrEqual(200); + expect(sent[1]?.message?.startsWith('inspecting ')).toBe(true); + expect(sent[1]?.message?.includes('Ready.')).toBe(false); + }); + + it('buffers shell and replace internally and never encodes partial content in progress', async () => { + const sent: McpProgressNotificationParams[] = []; + const shell = document({ + root: { + children: [ + { kind: 'markdown', text: '# Partial shell' }, + { completed: 0, kind: 'progress', message: 'loading' }, + ], + kind: 'result', + }, + value: { partial: true }, + }); + const final = document({ + root: { + children: [{ kind: 'markdown', text: '# Final' }], + kind: 'result', + }, + value: { ok: true }, + }); + + const projected = await projectMcpRenderStream(eventsOf([ + { document: shell, sequence: 0, type: 'shell' }, + { boundaryId: 'b:1', document: shell, sequence: 1, type: 'replace' }, + { completed: 1, message: 'working', sequence: 2, type: 'progress' }, + { document: final, sequence: 3, type: 'complete' }, + ]), { + progressToken: 'tok', + sendProgress: async (params) => { + sent.push(params); + }, + }); + + expect(sent).toEqual([{ message: 'working', progress: 1, progressToken: 'tok' }]); + expect(JSON.stringify(sent)).not.toContain('Partial shell'); + expect(JSON.stringify(sent)).not.toContain('partial'); + expect(projected.result).toEqual({ + content: [{ text: '# Final', type: 'text' }], + structuredContent: { ok: true }, + }); + }); + + it('returns one CallToolResult with supported blocks and object-valued structured content', async () => { + const projected = await projectMcpRenderStream(eventsOf([{ + document: document({ + root: { + children: [ + { kind: 'markdown', text: '# Catalog' }, + { kind: 'context', text: 'route guidance' }, + { kind: 'json', value: { count: 1 } }, + { completed: 1, kind: 'progress', message: 'status-only' }, + { data: 'aW1hZ2U=', kind: 'image', mimeType: 'image/png' }, + { data: 'YXVkaW8=', kind: 'audio', mimeType: 'audio/wav' }, + { kind: 'resource', mimeType: 'application/json', name: 'Catalog', uri: 'catalog://root' }, + { code: 'E_NOTE', kind: 'error', message: 'represented' }, + ], + kind: 'result', + }, + status: 'represented-error', + value: { count: 1 }, + }), + sequence: 0, + type: 'complete', + }])); + + expect(projected.result).toEqual({ + content: [ + { text: '# Catalog', type: 'text' }, + { text: 'route guidance', type: 'text' }, + { text: '{"count":1}', type: 'text' }, + { data: 'aW1hZ2U=', mimeType: 'image/png', type: 'image' }, + { data: 'YXVkaW8=', mimeType: 'audio/wav', type: 'audio' }, + { mimeType: 'application/json', name: 'Catalog', type: 'resource_link', uri: 'catalog://root' }, + { text: '[E_NOTE] represented', type: 'text' }, + ], + isError: true, + structuredContent: { count: 1 }, + }); + }); + + it('fails closed or uses a declared text fallback for gated rich content', async () => { + const rich = document({ + root: { + children: [{ data: 'aW1hZ2U=', kind: 'image', mimeType: 'image/png' }], + kind: 'result', + }, + }); + + await expect(projectMcpRenderStream(eventsOf([ + { document: rich, sequence: 0, type: 'complete' }, + ]), { capabilities: { image: false } })).rejects.toBeInstanceOf(McpProjectionError); + await expect(projectMcpRenderStream(eventsOf([ + { document: rich, sequence: 0, type: 'complete' }, + ]), { capabilities: { image: false } })).rejects.toMatchObject({ + code: 'unsupported-rich-content', + kind: 'image', + }); + + await expect(projectMcpRenderStream(eventsOf([ + { document: rich, sequence: 0, type: 'complete' }, + ]), { capabilities: { image: false }, richContentFallback: 'text' })).resolves.toMatchObject({ + result: { content: [{ text: '[image image/png]', type: 'text' }] }, + }); + }); + + it('omits non-object structured content instead of fabricating an object', async () => { + const projected = await projectMcpRenderStream(eventsOf([{ + document: document({ value: ['not', 'an', 'object'] }), + sequence: 0, + type: 'complete', + }])); + expect(projected.result.structuredContent).toBeUndefined(); + expect(projected.result.content).toEqual([{ text: 'Ready.', type: 'text' }]); + }); + + it('fails when the stream ends without a complete document', async () => { + await expect(projectMcpRenderStream(eventsOf([ + { document: document(), sequence: 0, type: 'shell' }, + ]))).rejects.toBeInstanceOf(AgentContractError); + await expect(projectMcpRenderStream(eventsOf([ + { document: document(), sequence: 0, type: 'shell' }, + ]))).rejects.toMatchObject({ code: 'invalid-document' }); + }); + + it('maps an abort into the renderer AbortSignal', async () => { + const controller = new AbortController(); + const stream = new ReadableStream({ + start() { + controller.abort(); + }, + }); + await expect(projectMcpRenderStream(stream, { signal: controller.signal })).rejects.toMatchObject({ + name: 'AbortError', + }); + }); +}); diff --git a/packages/rsc-runtime/tests/warm-runtime.test.ts b/packages/rsc-runtime/tests/warm-runtime.test.ts new file mode 100644 index 000000000..4e482f5b9 --- /dev/null +++ b/packages/rsc-runtime/tests/warm-runtime.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, it } from '@rstest/core'; + +import { + AgentRuntimeError, + createWarmFlightHost, + type AgentFlightExecutionHost, + type AgentRenderDispatch, +} from '../src/index.js'; + +const invocation = { + kind: 'tool' as const, + props: { input: {}, operationId: 'warmth' }, +}; + +const dispatch = (overrides: Partial = {}): AgentRenderDispatch => ({ + invocation, + signal: new AbortController().signal, + ...overrides, +}); + +const emptyFlight = (): ReadableStream => + new ReadableStream({ + start(controller) { + controller.close(); + }, + }); + +describe('createWarmFlightHost', () => { + it('serves consecutive requests from one process-lifetime identity', async () => { + let executes = 0; + const inner: AgentFlightExecutionHost = { + execute: async () => { + executes += 1; + return emptyFlight(); + }, + }; + const host = createWarmFlightHost({ artifactEpoch: 'epoch-a', host: inner, instanceId: 'runtime-1' }); + + await host.execute(dispatch()); + await host.execute(dispatch()); + + expect(executes).toBe(2); + expect(host.identity).toEqual({ artifactEpoch: 'epoch-a', instanceId: 'runtime-1' }); + }); + + it('fails closed on artifact-epoch mismatch with a typed error', async () => { + let executes = 0; + const host = createWarmFlightHost({ + artifactEpoch: 'epoch-a', + host: { + execute: async () => { + executes += 1; + return emptyFlight(); + }, + }, + }); + + await expect(host.execute(dispatch({ artifactEpoch: 'epoch-b' }))).rejects.toBeInstanceOf(AgentRuntimeError); + await expect(host.execute(dispatch({ artifactEpoch: 'epoch-b' }))).rejects.toMatchObject({ + code: 'artifact-epoch-mismatch', + expectedEpoch: 'epoch-a', + receivedEpoch: 'epoch-b', + }); + expect(executes).toBe(0); + await expect(host.execute(dispatch({ artifactEpoch: 'epoch-a' }))).resolves.toBeDefined(); + expect(executes).toBe(1); + }); + + it('fails closed after a runtime restart and never fabricates success', async () => { + const host = createWarmFlightHost({ + artifactEpoch: 'epoch-a', + host: { execute: async () => emptyFlight() }, + }); + + await host.execute(dispatch()); + host.markUnavailable('runtime-restarted'); + + await expect(host.execute(dispatch())).rejects.toBeInstanceOf(AgentRuntimeError); + await expect(host.execute(dispatch())).rejects.toMatchObject({ code: 'runtime-restarted' }); + }); + + it('yields a typed unavailable error when the runtime is missing', async () => { + const host = createWarmFlightHost({ + artifactEpoch: 'epoch-a', + host: { execute: async () => emptyFlight() }, + }); + host.markUnavailable('runtime-unavailable'); + + await expect(host.execute(dispatch())).rejects.toBeInstanceOf(AgentRuntimeError); + await expect(host.execute(dispatch())).rejects.toMatchObject({ code: 'runtime-unavailable' }); + }); +});