fix: cap MCP tool-call concurrency via MCP_TOOL_CALL_CONCURRENCY (default 4)

.changeset/mcp-tool-call-concurrency.md

@@ -0,0 +1,6 @@
+---
+"@truefoundry/trueforge-core": patch
+"@truefoundry/trueforge": patch
+---
+
+Cap parallel MCP tool execution at 4 in-flight calls (MCP_TOOL_CALL_CONCURRENCY).

packages/trueforge-core/src/agent-session/ITurnResourceResolver.ts

@@ -66,4 +66,5 @@ export interface ITurnResourceResolver<TTurnCustom extends object = Record<strin
    * knowing what was resolved.
    */
   close(): Promise<void>;
+  readonly mcpToolCallConcurrency: number;
 }

packages/trueforge-core/src/agent-session/SessionHandle.ts

@@ -485,6 +485,7 @@ export class SessionHandle<
       capabilityState: input.previousThreadSnapshot?.capability_state ?? undefined,
       tracing: input.tracing,
       logger: input.resolver.logger,
+      mcpToolCallConcurrency: input.resolver.mcpToolCallConcurrency,
     });
   }
 
@@ -536,6 +537,7 @@ export class SessionHandle<
         capabilities,
         tracing: input.tracing,
         logger: input.resolver.logger,
+        mcpToolCallConcurrency: input.resolver.mcpToolCallConcurrency,
       });
     };
   }

packages/trueforge-core/src/agent-session/TurnResourceResolver.ts

@@ -75,6 +75,7 @@ export class TurnResourceResolver<
       mcp: (name: string) => Promise<{ url: string; headers?: RemoteMcpHeaders }>;
       mcpRequestTimeoutMs: number;
       mcpConnectTimeoutMs: number;
+      mcpToolCallConcurrency: number;
       /** One sandbox type per runtime. Omit = no sandbox support. */
       sandboxProvider?: TurnSandboxFactory | undefined;
       /**
@@ -91,6 +92,10 @@ export class TurnResourceResolver<
     return this.deps.logger;
   }
 
+  get mcpToolCallConcurrency(): number {
+    return this.deps.mcpToolCallConcurrency;
+  }
+
   /** Default: no-op tracing. Override to plug in a real tracer. */
   createTracing(): AgentTracing {
     return NOOP_AGENT_TRACING;

packages/trueforge-core/src/core/index.ts

@@ -50,6 +50,7 @@ export { openUI } from './capabilities/builtins/OpenUI';
 // MCP contracts
 export type { ApprovalDecision } from './events/schema';
 export { ClientSideTool } from './mcp/ClientSideTool';
+export { DEFAULT_MCP_TOOL_CALL_CONCURRENCY } from './mcp/executeToolCalls';
 export { isAuthRequired, toolResultResponse } from './mcp/IMCPServer';
 export type {
   AgentToolSchema,

packages/trueforge-core/src/core/mcp/executeToolCalls.ts

@@ -5,6 +5,7 @@ import type { MCPAuthRequired } from '../mcp/IMCPServer';
 import type { AgentThreadCreateSubAgent } from '../runtime/AgentThread.types';
 import { InternalEventType } from '../runtime/AgentThread.types';
 import type { SandboxInfo } from '../sandbox/Sandbox';
+import { mapWithConcurrency } from '../util/promiseUtils';
 import type { MappedMCPTool } from './convertMCPServers';
 import {
   isApprovalRequiredResponse,
@@ -14,6 +15,13 @@ import {
   toolResultResponse,
 } from './IMCPServer';
 
+/**
+ * Single default for callers and env `MCP_TOOL_CALL_CONCURRENCY`.
+ * Tool bodies sit in memory until LargeToolResponse truncates, so peak RAM is in-flight × largest body.
+ * 4 still covers typical 2–8 parallel calls without extra wait, and caps a 20+ dump instead of matching it.
+ */
+export const DEFAULT_MCP_TOOL_CALL_CONCURRENCY = 4;
+
 export interface ToolCallResult {
   message: LLMToolMessage;
   // Absent only for unknown-tool calls (LLM hallucinated a name not in toolMapping):
@@ -42,11 +50,15 @@ export async function executeToolCalls({
   toolMapping,
   threadId,
   approvalDecisions,
+  concurrency,
+  signal,
 }: {
   assistantMessage: InternalEnrichedAssistantMessage;
   toolMapping: Map<string, MappedMCPTool>;
   threadId: string;
   approvalDecisions: Map<string, ApprovalDecision>;
+  concurrency: number;
+  signal?: AbortSignal | undefined;
 }): Promise<ExecuteToolCallsResult> {
   const toolMessages: ToolCallResult[] = [];
   const initializationInfo: MCPServerInitInfo[] = [];
@@ -70,43 +82,52 @@ export async function executeToolCalls({
     };
   }
 
-  const toolCallPromises = assistantMessage.tool_calls.map(async toolCall => {
-    const toolInfo = toolMapping.get(toolCall.function.name);
-    if (!toolInfo) {
-      return {
-        toolCall,
-        toolInfo,
-        response: toolResultResponse({ text: `Tool ${toolCall.function.name} not found in tool mapping` }),
-        failure: true,
-        completedAt: new Date().toISOString(),
-      };
-    }
-
-    try {
-      const args: Record<string, unknown> = JSON.parse(toolCall.function.arguments || '{}') as Record<string, unknown>;
-      const response = await toolInfo.toolSet.callTool(
-        {
-          name: toolInfo.originalToolName,
-          arguments: args,
-        },
-        approvalDecisions.get(toolCall.id),
-      );
-      return { toolCall, toolInfo, response, failure: false, completedAt: new Date().toISOString() };
-    } catch (error) {
-      return {
-        toolCall,
-        toolInfo,
-        response: toolResultResponse({
-          text: JSON.stringify({ error: error instanceof Error ? error.message : 'Tool execution failed' }),
-          isError: true,
-        }),
-        failure: true,
-        completedAt: new Date().toISOString(),
-      };
-    }
-  });
+  // After cancel, workers stop taking new tool calls from the queue.
+  // Do not throw: calls already in flight may still finish and those results are kept.
+  // AgentThread.execute() then returns on abort so deriveState() does not see leftover open tool calls.
+  const results = await mapWithConcurrency(
+    assistantMessage.tool_calls,
+    concurrency,
+    async toolCall => {
+      const toolInfo = toolMapping.get(toolCall.function.name);
+      if (!toolInfo) {
+        return {
+          toolCall,
+          toolInfo,
+          response: toolResultResponse({ text: `Tool ${toolCall.function.name} not found in tool mapping` }),
+          failure: true,
+          completedAt: new Date().toISOString(),
+        };
+      }
 
-  const results = await Promise.all(toolCallPromises);
+      try {
+        const args: Record<string, unknown> = JSON.parse(toolCall.function.arguments || '{}') as Record<
+          string,
+          unknown
+        >;
+        const response = await toolInfo.toolSet.callTool(
+          {
+            name: toolInfo.originalToolName,
+            arguments: args,
+          },
+          approvalDecisions.get(toolCall.id),
+        );
+        return { toolCall, toolInfo, response, failure: false, completedAt: new Date().toISOString() };
+      } catch (error) {
+        return {
+          toolCall,
+          toolInfo,
+          response: toolResultResponse({
+            text: JSON.stringify({ error: error instanceof Error ? error.message : 'Tool execution failed' }),
+            isError: true,
+          }),
+          failure: true,
+          completedAt: new Date().toISOString(),
+        };
+      }
+    },
+    signal,
+  );
   for (const { toolCall, toolInfo, response, failure, completedAt } of results) {
     if (isCallToolResponseCreateSubAgent(response)) {
       createThreadEvents.push({

packages/trueforge-core/src/core/runtime/AgentThread.ts

@@ -506,6 +506,7 @@ export class AgentThread {
   private sandbox?: Sandbox | undefined;
   private readonly tracing: AgentTracing;
   private readonly logger: Logger;
+  private readonly mcpToolCallConcurrency: number;
 
   private metrics: AgentThreadMetrics = createEmptyAgentThreadMetrics();
   private tfyManagedServerNames = new Set<string>();
@@ -521,6 +522,7 @@ export class AgentThread {
   constructor(input: AgentThreadConstructorInput) {
     this.tracing = input.tracing;
     this.logger = input.logger.child({ module: 'AgentThread' });
+    this.mcpToolCallConcurrency = input.mcpToolCallConcurrency;
     this.threadId = input.threadId;
     this.definition = input.definition;
     this.context = input.context ? [...input.context] : [];
@@ -1146,6 +1148,7 @@ export class AgentThread {
 
   private async *stepToolResponse(
     toolMapping: Map<string, MappedMCPTool>,
+    signal?: AbortSignal,
   ): AsyncGenerator<AgentThreadEvent, StepOutcome, unknown> {
     const assistantMessage = lastAssistantInContext(this.context);
     if (!assistantMessage) {
@@ -1174,6 +1177,8 @@ export class AgentThread {
       toolMapping,
       threadId: this.threadId,
       approvalDecisions: decisions,
+      concurrency: this.mcpToolCallConcurrency,
+      signal,
     });
     void clientSideToolCalls;
     if (approvalRequiredToolCalls.length > 0) {
@@ -1376,7 +1381,7 @@ export class AgentThread {
             if (signal?.aborted) {
               return;
             }
-            outcome = yield* this.stepToolResponse(toolMapping);
+            outcome = yield* this.stepToolResponse(toolMapping, signal);
             break;
           }
           case 'user-input-required': {
@@ -1389,7 +1394,10 @@ export class AgentThread {
             throw new Error('unreachable');
           }
         }
-        if (outcome === 'exit') {
+        // After a step, abort must return before the next deriveState().
+        // A partial tool batch leaves open calls; tool-response-required → tool-response-required is invalid.
+        // Do not check at the top of the loop: user-input-required must still emit.
+        if (outcome === 'exit' || signal?.aborted) {
           return;
         }
       }

packages/trueforge-core/src/core/runtime/AgentThread.types.ts

@@ -176,4 +176,5 @@ export interface AgentThreadConstructorInput {
   capabilityState?: CapabilityState | undefined;
   tracing: AgentTracing;
   logger: Logger;
+  mcpToolCallConcurrency: number;
 }

packages/trueforge-core/src/core/util/promiseUtils.ts

@@ -63,3 +63,38 @@ export async function* mergeAsyncGenerators<T>(
     pending.set(idx, getNextIteration(idx));
   }
 }
+
+/** Runs `fn` over `items` with at most `concurrency` calls in flight. Results stay in input order. */
+export async function mapWithConcurrency<T, R>(
+  items: readonly T[],
+  concurrency: number,
+  fn: (item: T, index: number) => Promise<R>,
+  signal?: AbortSignal,
+): Promise<R[]> {
+  if (items.length === 0) {
+    return [];
+  }
+
+  const workerCount = Math.max(1, Math.min(concurrency, items.length));
+  const completed: { index: number; value: R }[] = [];
+  let nextIndex = 0;
+
+  const worker = async (): Promise<void> => {
+    while (nextIndex < items.length) {
+      if (signal?.aborted) {
+        return;
+      }
+      const index = nextIndex;
+      nextIndex += 1;
+      const item = items[index];
+      if (item === undefined) {
+        return;
+      }
+      completed.push({ index, value: await fn(item, index) });
+    }
+  };
+
+  await Promise.all(Array.from({ length: workerCount }, () => worker()));
+  completed.sort((a, b) => a.index - b.index);
+  return completed.map(entry => entry.value);
+}

packages/trueforge-core/tests/agent-session/capabilityState.test.ts

@@ -9,6 +9,7 @@ import { Sessions } from '../../src/agent-session/Sessions';
 import { InMemorySessionStore } from '../../src/agent-session/store/InMemorySessionStore';
 import type { AgentCapability, JsonValue } from '../../src/core/capabilities/AgentCapability';
 import type { AgentContextProcessorOutput } from '../../src/core/capabilities/AgentContextProcessor';
+import { DEFAULT_MCP_TOOL_CALL_CONCURRENCY } from '../../src/core/mcp/executeToolCalls';
 import { AgentThread } from '../../src/core/runtime/AgentThread';
 import { InternalEventType } from '../../src/core/runtime/AgentThread.types';
 import { NOOP_AGENT_TRACING } from '../../src/core/tracing/NoopAgentTracing';
@@ -281,6 +282,7 @@ describe('capability_state (tfy.plan fixture)', () => {
       capabilities: [badCapability],
       tracing: NOOP_AGENT_TRACING,
       logger: makeSilentLogger(),
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
     });
     for await (const event of thread.send([{ type: EventType.USER_MESSAGE, content: 'x' }])) {
       void event;

packages/trueforge-core/tests/agent-session/resolveAgentSpec.test.ts

@@ -3,6 +3,7 @@ import { EventType } from '../../src/agent-session/schemas/events';
 import { Sessions } from '../../src/agent-session/Sessions';
 import { InMemorySessionStore } from '../../src/agent-session/store/InMemorySessionStore';
 import { TurnResourceResolver } from '../../src/agent-session/TurnResourceResolver';
+import { DEFAULT_MCP_TOOL_CALL_CONCURRENCY } from '../../src/core/mcp/executeToolCalls';
 import { makeAgentSpec, makeMockILLM, makeSilentLogger, makeTestResolver, mintTestTurnId } from './testHelpers';
 
 describe('TurnResourceResolver.resolveAgentSpec', () => {
@@ -12,6 +13,7 @@ describe('TurnResourceResolver.resolveAgentSpec', () => {
       mcp: () => Promise.reject(new Error('unused')),
       mcpRequestTimeoutMs: 1_000,
       mcpConnectTimeoutMs: 1_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       logger: makeSilentLogger(),
     });
 
@@ -26,6 +28,7 @@ describe('TurnResourceResolver.resolveAgentDefinition', () => {
       mcp: () => Promise.reject(new Error('unused')),
       mcpRequestTimeoutMs: 1_000,
       mcpConnectTimeoutMs: 1_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       logger: makeSilentLogger(),
     });
 
@@ -62,6 +65,7 @@ describe('TurnResourceResolver.resolveAgentDefinition', () => {
       mcp: () => Promise.reject(new Error('unused')),
       mcpRequestTimeoutMs: 1_000,
       mcpConnectTimeoutMs: 1_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       logger: makeSilentLogger(),
     });
     const spec = AgentSpecSchema.parse({

packages/trueforge-core/tests/agent-session/testHelpers.ts

@@ -14,6 +14,7 @@ import type {
   RawAssistantMessageWithUsage,
 } from '../../src/core/llm/LLMTypes';
 import { getEmptyUsage } from '../../src/core/llm/LLMTypes';
+import { DEFAULT_MCP_TOOL_CALL_CONCURRENCY } from '../../src/core/mcp/executeToolCalls';
 import { getEmptyCurrentContextUsage } from '../../src/core/runtime/contextUsage';
 import type { Sandbox } from '../../src/core/sandbox/Sandbox';
 import { makeMockILLM, makeSilentLogger } from '../core/harnessMocks';
@@ -100,6 +101,7 @@ export function makeTestResolver<TTurnCustom extends object = Record<string, nev
     },
     mcpRequestTimeoutMs: 60_000,
     mcpConnectTimeoutMs: 5_000,
+    mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
     logger: makeSilentLogger(),
     ...(options?.agent !== undefined ? { agent: options.agent } : {}),
     ...(options?.sandbox
@@ -122,6 +124,7 @@ export function makeTestResolver<TTurnCustom extends object = Record<string, nev
     get logger() {
       return base.logger;
     },
+    mcpToolCallConcurrency: base.mcpToolCallConcurrency,
     createTracing: () => base.createTracing(),
     resolveAgentSpec: input => base.resolveAgentSpec(input),
     resolveSandbox: input => base.resolveSandbox(input),

packages/trueforge-core/tests/agent-session/turnStream.test.ts

@@ -3,6 +3,7 @@ import { CancellationReason } from '../../src/agent-session/schemas/turn';
 import { Sessions } from '../../src/agent-session/Sessions';
 import { InMemorySessionStore } from '../../src/agent-session/store/InMemorySessionStore';
 import { TurnResourceResolver } from '../../src/agent-session/TurnResourceResolver';
+import { DEFAULT_MCP_TOOL_CALL_CONCURRENCY } from '../../src/core/mcp/executeToolCalls';
 import { RemoteMCP } from '../../src/core/mcp/RemoteMCP';
 import { makeStubPublicSandbox } from '../core/harnessMocks';
 import {
@@ -400,6 +401,7 @@ describe('TurnHandle.stream()', () => {
       mcp: () => Promise.resolve({ url: 'http://localhost' }),
       mcpRequestTimeoutMs: 60_000,
       mcpConnectTimeoutMs: 5_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       sandboxProvider: () => Promise.resolve(sandbox),
       logger,
     });
@@ -454,6 +456,7 @@ describe('TurnResourceResolver caches', () => {
       mcp: () => Promise.resolve({ url: 'http://example.invalid' }),
       mcpRequestTimeoutMs: 60_000,
       mcpConnectTimeoutMs: 5_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       logger,
     });
     await resolver.resolveTwice();
@@ -478,6 +481,7 @@ describe('TurnResourceResolver caches', () => {
       mcp: () => Promise.reject(new Error('unused')),
       mcpRequestTimeoutMs: 1_000,
       mcpConnectTimeoutMs: 1_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       logger,
     });
     await resolver.resolveTwice();
@@ -498,6 +502,7 @@ describe('TurnResourceResolver caches', () => {
       mcp: () => Promise.resolve({ url: 'http://localhost' }),
       mcpRequestTimeoutMs: 60_000,
       mcpConnectTimeoutMs: 5_000,
+      mcpToolCallConcurrency: DEFAULT_MCP_TOOL_CALL_CONCURRENCY,
       sandboxProvider: () => {
         sandboxCreates += 1;
         return Promise.resolve(sandbox);