From 4c60a75a75a9a4289eeeb3401e8b3b35e1a737ef Mon Sep 17 00:00:00 2001 From: Wassim SAMAD Date: Wed, 11 Nov 2026 11:11:19 -0500 Subject: [PATCH] mcp: a host can pause between the calls of a run_batch A batch is one synchronous operation, so a long one holds the event loop for its whole length and the host's timers and I/O wait for it. core's batch is now a sequence that stops after each call; the synchronous operation runs it through as before. A host that gives betweenBatchCalls is awaited after each call. Only a host with a scene of its own per request should: the batch is planned on the scene as its first call found it and applied after its last. Without the hook a batch is uninterrupted. Co-Authored-By: Claude Fable 5.2 --- packages/core/src/agent-operations/batch.ts | 31 ++++++- packages/core/src/agent-operations/index.ts | 5 +- packages/mcp/src/server.ts | 3 + packages/mcp/src/tools/index.ts | 11 ++- .../mcp/src/tools/run-batch-pause.test.ts | 88 +++++++++++++++++++ packages/mcp/src/tools/shared-tools.ts | 30 ++++++- wiki/architecture/agent-surfaces.md | 2 + 7 files changed, 163 insertions(+), 7 deletions(-) create mode 100644 packages/mcp/src/tools/run-batch-pause.test.ts diff --git a/packages/core/src/agent-operations/batch.ts b/packages/core/src/agent-operations/batch.ts index 879772507f..212e47a143 100644 --- a/packages/core/src/agent-operations/batch.ts +++ b/packages/core/src/agent-operations/batch.ts @@ -5,7 +5,7 @@ import type { SceneMaterialId } from '../schema/scene-material' import { applySceneChanges, mergeSceneChanges } from './apply-changes' import { requireFurnishing } from './design-brief' import { referencedSceneMaterialIds } from './scene-materials' -import type { AgentOperation, SceneChanges } from './types' +import type { AgentOperation, AgentOperationOutcome, SceneChanges } from './types' type BatchInput = { calls: { tool: string; input?: Record }[] @@ -41,17 +41,24 @@ function prepareCall( return { operation, input: parsed.data } } +/** A batch one call at a time: each resume runs the next call, and the last returns the outcome. */ +export type BatchSteps = ( + ...args: Parameters> +) => Generator + /** * `run_batch`: shared tools in order, each on the scene the calls before it left, their changes * merged into one set — one undo step, one write on either surface. The user: the model spent a * round trip per step on what could be serialised (deleting balcony walls one by one, the same * build on every floor). Pre-check and operation refusals follow the same per-call policy. + * It stops after each call, so a host that can let other work run between a batch's calls does + * (`createBatchOperation` runs it through in one go). */ -export function createBatchOperation( +export function createBatchSteps( operations: Readonly>>, contracts: readonly Contract[], -): AgentOperation { - return (nodes, { calls, onRefusal = 'stop' }, context) => { +): BatchSteps { + return function* (nodes, { calls, onRefusal = 'stop' }, context) { let working = nodes let workingContext = context let kept: RefusalKeep['readings'] | undefined @@ -109,6 +116,7 @@ export function createBatchOperation( error: error.message, }) } + yield } const applied = reports.filter((report) => report.status === 'ok').length const merged = mergeSceneChanges(sets) @@ -135,3 +143,18 @@ export function createBatchOperation( } } } + +/** `run_batch` as one synchronous operation: every call, back to back. */ +export function createBatchOperation( + operations: Readonly>>, + contracts: readonly Contract[], +): AgentOperation { + const steps = createBatchSteps(operations, contracts) + return (nodes, input, context) => { + const batch = steps(nodes, input, context) + for (;;) { + const step = batch.next() + if (step.done) return step.value + } + } +} diff --git a/packages/core/src/agent-operations/index.ts b/packages/core/src/agent-operations/index.ts index 9288aa219e..31fb35c880 100644 --- a/packages/core/src/agent-operations/index.ts +++ b/packages/core/src/agent-operations/index.ts @@ -5,7 +5,7 @@ import { addLevel } from './add-level' import { addObjectInBatch } from './add-object' import { addWall } from './add-wall' import { viewAssets } from './asset-views' -import { createBatchOperation } from './batch' +import { createBatchOperation, createBatchSteps } from './batch' import { addCornerWindow } from './corner-window' import { createRoof } from './create-roof' import { createRoom } from './create-room' @@ -170,6 +170,9 @@ export const AGENT_OPERATIONS = { run_batch: createBatchOperation(BATCHED_OPERATIONS, AGENT_TOOL_CONTRACTS), } as const satisfies AgentOperations +/** `run_batch` one call at a time, for a host that lets other work run between a batch's calls. */ +export const RUN_BATCH_STEPS = createBatchSteps(BATCHED_OPERATIONS, AGENT_TOOL_CONTRACTS) + export { type AssetViewFacts, type AssetViewIO, diff --git a/packages/mcp/src/server.ts b/packages/mcp/src/server.ts index 5dff597bb5..a701706c43 100644 --- a/packages/mcp/src/server.ts +++ b/packages/mcp/src/server.ts @@ -70,6 +70,8 @@ export type CreatePascalMcpServerOptions = { sceneViews?: SceneViewHost /** Reads catalog items' pictures and surfaces before view_assets and paint. */ assetViews?: PlanReadingHosts['assetViews'] + /** A pause after each call of a run_batch, for a host with a scene of its own per request. */ + betweenBatchCalls?: PlanReadingHosts['betweenBatchCalls'] /** Lines the host adds to what a client reads at connect (the person's own settings). */ instructions?: string } @@ -95,6 +97,7 @@ export function createPascalMcpServer(opts: CreatePascalMcpServerOptions): McpSe readFile: opts.readFile, sceneViews: opts.sceneViews, assetViews: opts.assetViews, + betweenBatchCalls: opts.betweenBatchCalls, }) registerVisionTools(server, operations) if (opts.services) registerHostedServiceTools(server, opts.services) diff --git a/packages/mcp/src/tools/index.ts b/packages/mcp/src/tools/index.ts index 12e041be1e..2885f68c20 100644 --- a/packages/mcp/src/tools/index.ts +++ b/packages/mcp/src/tools/index.ts @@ -52,11 +52,20 @@ export type ToolHosts = PlanReadingHosts & { export function registerTools( server: McpServer, operations: SceneOperations, - { catalog, geometryScripts, objectMakers, readFile, sceneViews, assetViews }: ToolHosts = {}, + { + catalog, + geometryScripts, + objectMakers, + readFile, + sceneViews, + assetViews, + betweenBatchCalls, + }: ToolHosts = {}, ): void { registerGetScene(server, operations) registerSharedTools(server, operations, catalog, { assetViews, + betweenBatchCalls, buildObject: (input) => compileObjectOnHost(geometryScripts, operations, input), }) registerAddColumn(server, operations, geometryScripts) diff --git a/packages/mcp/src/tools/run-batch-pause.test.ts b/packages/mcp/src/tools/run-batch-pause.test.ts new file mode 100644 index 0000000000..06d7907a64 --- /dev/null +++ b/packages/mcp/src/tools/run-batch-pause.test.ts @@ -0,0 +1,88 @@ +import { describe, expect, test } from 'bun:test' +import { Client, InMemoryTransport } from '@modelcontextprotocol/client' +import { BuildingNode, LevelNode } from '@pascal-app/core/schema' +import { SceneBridge } from '../bridge/scene-bridge' +import { createPascalMcpServer } from '../server' + +type Result = { + isError?: boolean + content: { type: string; text: string }[] + structuredContent?: { + status: string + applied: number + refused: number + calls: { index: number; tool: string; status: string; code?: string }[] + } +} + +const wall = (x: number) => ({ tool: 'add_wall', input: { start: [x, 0], end: [x, 3] } }) +const missing = { tool: 'delete_node', input: { id: 'wall_that_is_not_there' } } + +/** A one-level building served by its own server; `pause` is the host's, when it gives one. */ +async function served(pause?: (bridge: SceneBridge) => Promise) { + const level = LevelNode.parse({ id: 'level_0', parentId: 'building_main', level: 0 }) + const building = BuildingNode.parse({ id: 'building_main', children: [level.id] }) + const bridge = new SceneBridge() + bridge.setScene({ [building.id]: building, [level.id]: level }, [building.id] as never) + const server = createPascalMcpServer({ + bridge, + ...(pause && { betweenBatchCalls: () => pause(bridge) }), + }) + const [serverTransport, clientTransport] = InMemoryTransport.createLinkedPair() + const client = new Client({ name: 'run-batch-pause-test', version: '0.0.0' }) + await Promise.all([server.connect(serverTransport), client.connect(clientTransport)]) + const walls = () => Object.values(bridge.getNodes()).filter((node) => node.type === 'wall').length + const batch = async (calls: unknown[], onRefusal?: 'stop' | 'skip') => + (await client.callTool({ + name: 'run_batch', + arguments: { calls, ...(onRefusal && { onRefusal }) }, + })) as Result + return { batch, bridge, walls } +} + +/** A result without the ids its calls minted. */ +const shape = (result: Result) => ({ + isError: result.isError === true, + status: result.structuredContent?.status, + applied: result.structuredContent?.applied, + refused: result.structuredContent?.refused, + calls: result.structuredContent?.calls.map(({ index, tool, status, code }) => ({ + index, + tool, + status, + code, + })), +}) + +describe("run_batch on a host that pauses between a batch's calls", () => { + test('pauses after each call, with the scene as the batch found it, then applies the batch whole', async () => { + const seen: number[] = [] + const host = await served(async () => { + // Other work of the host runs here: a timer, a request's I/O. + await new Promise((resolve) => setTimeout(resolve, 0)) + seen.push(host.walls()) + }) + const result = await host.batch([wall(0), wall(2), { tool: 'get_walls' }, wall(4)]) + expect(seen).toEqual([0, 0, 0, 0]) + expect(host.walls()).toBe(3) + expect(host.bridge.undo()).toBe(1) + expect(host.walls()).toBe(0) + expect(result.structuredContent).toMatchObject({ status: 'applied', applied: 4, refused: 0 }) + }) + + test('answers as a batch run in one go does: in order, stopped by a refusal, or past it', async () => { + const batches: [unknown[], ('stop' | 'skip')?][] = [ + [[wall(0), { tool: 'get_walls' }, wall(2)]], + [[wall(0), missing, wall(2)]], + [[wall(0), missing, wall(2)], 'skip'], + [[wall(0), { tool: 'not_a_batch_tool' }]], + ] + for (const [calls, onRefusal] of batches) { + const paused = await served(() => new Promise((resolve) => setTimeout(resolve, 0))) + const direct = await served() + const [a, b] = [await paused.batch(calls, onRefusal), await direct.batch(calls, onRefusal)] + expect(shape(a)).toEqual(shape(b)) + expect(paused.walls()).toBe(direct.walls()) + } + }) +}) diff --git a/packages/mcp/src/tools/shared-tools.ts b/packages/mcp/src/tools/shared-tools.ts index 2399417b5f..ca31329fc2 100644 --- a/packages/mcp/src/tools/shared-tools.ts +++ b/packages/mcp/src/tools/shared-tools.ts @@ -13,6 +13,7 @@ import { createAssetViewLoader, createFetchAssetViewIO, furnishingRefusal, + RUN_BATCH_STEPS, readCatalog, type SceneChanges, startOfSession, @@ -119,6 +120,13 @@ export type PlanReadingHosts = { assetViews?: AssetViewLoader /** Compiles one add_object module for a batch before the synchronous operations run. */ buildObject?: (input: Omit) => Promise + /** + * Awaited after each call of a run_batch, so the host's timers and I/O run during a long one. + * Only for a host whose scene no other call can reach meanwhile (a bridge of its own for each + * request): the batch is planned on the scene as its first call found it and applied after its + * last. Without it a batch runs uninterrupted. + */ + betweenBatchCalls?: () => Promise } const defaultAssetViews = createAssetViewLoader(createFetchAssetViewIO()) @@ -711,6 +719,19 @@ function modelResult( return output } +/** A batch one call at a time, each in the scene's scope, the host's pause after each. */ +async function runBatchInSteps( + bridge: SceneOperations, + batch: ReturnType, + pause: () => Promise, +) { + for (;;) { + const step = inScene(bridge, () => batch.next()) + if (step.done) return step.value + await pause() + } +} + export function registerSharedTools( server: McpServer, bridge: SceneOperations, @@ -806,7 +827,14 @@ export function registerOperationTool( pictures: needs.pictures, }) try { - outcome = inScene(bridge, () => tool.operation(nodes, input as never, context)) + outcome = + tool.contract.name === runBatchTool.name && hosts.betweenBatchCalls + ? await runBatchInSteps( + bridge, + RUN_BATCH_STEPS(nodes, input as never, context), + hosts.betweenBatchCalls, + ) + : inScene(bridge, () => tool.operation(nodes, input as never, context)) } catch (error) { // A refusal can ask the session to remember what it refused (a catalog piece under an open source). if (isAgentRefusal(error) && error.keep) bridge.agentSession?.keep(error.keep) diff --git a/wiki/architecture/agent-surfaces.md b/wiki/architecture/agent-surfaces.md index 47831f2032..6b9e617ce7 100644 --- a/wiki/architecture/agent-surfaces.md +++ b/wiki/architecture/agent-surfaces.md @@ -117,6 +117,8 @@ Views are planned on the site: a building's position and yaw move and turn the b **Objects in a batch.** `run_batch` runs `add_object` calls. A compile is async and the operations are sync, so the host builds every permitted add_object module first (`buildBatchObjects`, three compiles at a time, with the same compile its lone `add_object` uses) and hands the results to the batch in `AgentContext.batchObjects`; `addObjectInBatch` places each, with the receipt a lone call gives (`achieved` per object). `add_object` is in `INDEPENDENT_BATCH_TOOLS`: a module that fails (`script_failed`), a missing reason, an invalid input or a refused placement is that call's own report, whatever `onRefusal` says, and the other objects are applied. Every other tool's refusal follows `onRefusal`: `stop` (default) applies nothing, `skip` reports the refusal and applies the other calls. Unsupported tools, invalid inputs and brief refusals follow this same policy. A host that built nothing answers `object_not_built` (the chat builds through `executeRunBatch`, the MCP through `registerOperationTool`'s `buildObject` host). +**A pause between a batch's calls.** A batch is one synchronous operation: its calls run back to back, and its changes are applied in the same stretch, so nothing else on the server runs meanwhile and no other call can reach the scene between planning and applying. A long batch (several whole-scene checks on a large building) therefore holds the event loop for its whole length, and a host's timers do not fire. A host that builds a scene of its own for each request gives `betweenBatchCalls` (`createPascalMcpServer`), and the MCP then runs the batch one call at a time (`RUN_BATCH_STEPS`, each step in the scene's scope) and awaits the host after each: same order, same refusal policy, same result, one history step. A host that serves concurrent calls on one bridge leaves it out and keeps the uninterrupted batch. + **Inside a floor.** `interior: { levelId }` looks into one floor. The plan checks the level (`interior_level_invalid`, also for a level holding only roofs and ceilings; `interior_hidden` for a hidden level or ancestor, a hidden site excepted, since it hides only its ground; `interior_transform_unsupported` for a tipped building) and that a target is on it and in the picture (`interior_target_elsewhere`, also for its roofs and ceilings), frames the floor from 55° above unless a target, an eye or a camera is given, and hands the host only the level id. The editor that renders it works out the rest from its own scene with `sceneViewInterior`: for that one render the other floors and this floor's roofs and ceilings are hidden, everything on the floor (walls, openings, furniture) and its shadows are cut at 0.65 of the storey height, its walls are drawn whole whatever the user's wall mode, and the user's solo and isolation are lifted on it. Everything is put back right after the render, so the live view does not change. **Level presentation.** `presentation: { mode: 'solo', levelId }`, `{ mode: 'exploded' }` or `{ mode: 'stacked' }` captures the viewer's settled level arrangement, without changing the graph or saved viewer preferences. Solo preserves full-height walls, ceilings and roofs owned by that level, and keeps higher levels as shadow casters; it does not apply the interior cut. Exploded framing includes the shared level gap and explicitly classified retained roof bounds. Both surfaces carry this policy to the same snapshot renderer. `interior` and `presentation` cannot be combined; missing/hidden Solo levels and targets on another level are refused. The by-hand equivalent is the viewer's level mode and level selection.