fix(core): resume parallel sub-agent approvals in any order (#19450)

Author: roaminro

.changeset/puny-roses-fetch.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fixed parallel sub-agent approvals so they can be handled in any order. listSuspendedRuns() now returns each pending sub-agent call, and approving one resumes that specific call instead of using another call’s suspended state.

packages/core/src/agent/__tests__/parallel-delegation-approval.test.ts

@@ -186,6 +186,11 @@ function buildSupervisorAgent(storage: InMemoryStore = new InMemoryStore()) {
   return mastra.getAgent('supervisor');
 }
 
+// NOTE: default engine only. The evented engine (MASTRA_EVENTED_EXECUTION) has a
+// separate pre-existing run-lifecycle bug with parallel delegations: resuming the
+// first suspended iteration completes the outer run and deletes its snapshot,
+// permanently orphaning the sibling. Its foreach aggregation does not persist
+// per-iteration suspend payloads, so the pending check cannot see the parked sibling.
 describe('parallel sub-agent delegation (suspend/resume)', () => {
   it('emits two distinct approval requests, one per order', async () => {
     processedOrders.length = 0;
@@ -214,17 +219,14 @@ describe('parallel sub-agent delegation (suspend/resume)', () => {
       memory: { resource: 'rep_approval', thread: 'thread-wrong-target' },
     });
 
-    const approvals = await collectApprovals(stream);
-    const [{ toolCalls }] = (await supervisor.listSuspendedRuns()).runs;
-    const suspendedToolCallIds = new Set(toolCalls.map(toolCall => toolCall.toolCallId));
-    const phantomApproval = approvals.find(approval => !suspendedToolCallIds.has(approval.toolCallId));
+    await collectApprovals(stream);
+    const bogusToolCallId = 'sup-tc-nonexistent';
 
-    expect(phantomApproval).toBeDefined();
     let resumeError: unknown;
     try {
       const resumed = await supervisor.approveToolCall({
         runId: stream.runId,
-        toolCallId: phantomApproval!.toolCallId,
+        toolCallId: bogusToolCallId,
       });
       for await (const _chunk of resumed.fullStream) {
         // Drain the stream so unintended tool execution cannot leak into later tests.
@@ -235,12 +237,62 @@ describe('parallel sub-agent delegation (suspend/resume)', () => {
 
     expect(resumeError).toMatchObject({ id: 'AGENT_RESUME_TOOL_CALL_NOT_SUSPENDED' });
     await expect(
-      supervisor.resumeGenerate({ approved: true }, { runId: stream.runId, toolCallId: phantomApproval!.toolCallId }),
+      supervisor.resumeGenerate({ approved: true }, { runId: stream.runId, toolCallId: bogusToolCallId }),
     ).rejects.toMatchObject({ id: 'AGENT_RESUME_TOOL_CALL_NOT_SUSPENDED' });
     await expect(
       supervisor.resumeGenerate({ approved: true }, { runId: stream.runId, toolCallId: '' }),
     ).rejects.toMatchObject({ id: 'AGENT_RESUME_TOOL_CALL_NOT_SUSPENDED' });
     expect(processedOrders).toEqual([]);
+  }, 30_000);
+
+  it('surfaces BOTH suspended delegations in listSuspendedRuns', async () => {
+    processedOrders.length = 0;
+    const supervisor = buildSupervisorAgent();
+
+    const stream = await supervisor.stream('Process both orders in parallel.', {
+      maxSteps: 6,
+      memory: { resource: 'rep_approval', thread: 'thread-surface' },
+    });
+
+    const approvals = await collectApprovals(stream);
+    expect(approvals.length).toBe(2);
+
+    const [{ toolCalls }] = (await supervisor.listSuspendedRuns()).runs;
+    const suspendedToolCallIds = toolCalls.map(toolCall => toolCall.toolCallId).sort();
+    expect(suspendedToolCallIds).toEqual(['sup-tc-A', 'sup-tc-B']);
+    for (const toolCall of toolCalls) {
+      expect(toolCall.toolName).toBe('agent-subAgent');
+      expect(toolCall.requiresApproval).toBe(true);
+    }
+  });
+
+  it('approving the delegations OUT OF ORDER (B first) processes both orders correctly', async () => {
+    processedOrders.length = 0;
+    const supervisor = buildSupervisorAgent();
+
+    const stream = await supervisor.stream('Process both orders in parallel.', {
+      maxSteps: 6,
+      memory: { resource: 'rep_approval', thread: 'thread-out-of-order' },
+    });
+
+    const approvals = await collectApprovals(stream);
+    const runId = stream.runId;
+    expect(approvals.length).toBe(2);
+
+    // Approve the SECOND emitted card first — the field failure ("approve the
+    // bottom card") that previously resumed the wrong delegation.
+    const outOfOrder = [...approvals].reverse();
+    const resumeErrors: string[] = [];
+    for (const a of outOfOrder) {
+      const resumed = await supervisor.approveToolCall({ runId, toolCallId: a.toolCallId });
+      for await (const chunk of resumed.fullStream) {
+        if (chunk.type === 'tool-error') resumeErrors.push(JSON.stringify((chunk as any).payload ?? chunk));
+      }
+    }
+
+    expect(resumeErrors).toEqual([]);
+    // Approval order must map to execution order: B was approved first.
+    expect(processedOrders).toEqual([ORDER_B, ORDER_A]);
   });
 
   it('approving both parallel delegations one at a time processes BOTH orders', async () => {

packages/core/src/agent/agent.ts

@@ -6310,12 +6310,8 @@ export class Agent<
 
   #getSuspendedToolCalls(existingSnapshot: WorkflowRunState | null | undefined): AgentRunToolCall[] {
     const toolCalls: AgentRunToolCall[] = [];
-    for (const key in existingSnapshot?.context) {
-      const step = existingSnapshot?.context[key];
-      if (step?.status !== 'suspended') continue;
-      const payload = step.suspendPayload;
-      if (!payload) continue;
 
+    const collectFromPayload = (payload: Record<string, any>, stepKey: string) => {
       if (payload.requireToolApproval) {
         toolCalls.push({
           toolCallId: payload.requireToolApproval.toolCallId,
@@ -6325,17 +6321,55 @@ export class Agent<
         });
       } else if (payload.toolCallSuspended || payload.toolName || payload.toolCallId) {
         toolCalls.push({
-          toolCallId: payload.toolCallId ?? this.#findResumeLabelForStep(existingSnapshot, key),
+          toolCallId: payload.toolCallId ?? this.#findResumeLabelForStep(existingSnapshot, stepKey),
           toolName: payload.toolName,
           requiresApproval: false,
           suspendPayload: payload.toolCallSuspended,
         });
       }
+    };
+
+    for (const key in existingSnapshot?.context) {
+      const step = existingSnapshot?.context[key];
+      if (step?.status !== 'suspended') continue;
+      const payload = step.suspendPayload;
+      if (!payload) continue;
+
+      // A foreach step (e.g. parallel tool calls in the agentic loop) can park several
+      // iterations at once, but its step-level suspendPayload only carries the first
+      // suspended iteration. The full set lives in `__workflow_meta.foreachOutput`,
+      // where each suspended entry keeps its own per-iteration payload — surface every
+      // one of them so all pending tool calls are discoverable and resumable.
+      const suspendedIterations = this.#getSuspendedForeachIterations(payload);
+      if (suspendedIterations.length > 0) {
+        for (const iteration of suspendedIterations) {
+          collectFromPayload(iteration.suspendPayload, key);
+        }
+      } else {
+        collectFromPayload(payload, key);
+      }
     }
 
     return toolCalls;
   }
 
+  #getSuspendedForeachIterations(
+    payload: Record<string, any>,
+  ): { status: 'suspended'; suspendPayload: Record<string, any> }[] {
+    // The default engine persists foreach aggregation as an array; the evented
+    // engine persists it as an object keyed by iteration index.
+    const foreachOutput = payload.__workflow_meta?.foreachOutput;
+    const entries = Array.isArray(foreachOutput)
+      ? foreachOutput
+      : foreachOutput && typeof foreachOutput === 'object'
+        ? Object.values(foreachOutput)
+        : [];
+    return entries.filter(
+      (entry: any): entry is { status: 'suspended'; suspendPayload: Record<string, any> } =>
+        entry?.status === 'suspended' && !!entry.suspendPayload,
+    );
+  }
+
   async #validateSuspendedToolCallTarget({
     snapshot,
     toolCallId,
@@ -6397,9 +6431,19 @@ export class Agent<
 
   #getSuspendedToolInfo(
     existingSnapshot: WorkflowRunState | null | undefined,
+    targetToolCallId?: string,
   ): { toolCallId?: string; toolName?: string } | undefined {
-    const [first] = this.#getSuspendedToolCalls(existingSnapshot);
-    return first ? { toolCallId: first.toolCallId, toolName: first.toolName } : undefined;
+    const suspendedToolCalls = this.#getSuspendedToolCalls(existingSnapshot);
+    // Several tool calls can be parked at once. Never label an explicitly targeted
+    // resume with a sibling if this snapshot predates the target's persistence.
+    const info =
+      targetToolCallId !== undefined
+        ? (suspendedToolCalls.find(toolCall => toolCall.toolCallId === targetToolCallId) ?? {
+            toolCallId: targetToolCallId,
+            toolName: undefined,
+          })
+        : suspendedToolCalls[0];
+    return info ? { toolCallId: info.toolCallId, toolName: info.toolName } : undefined;
   }
 
   #getResumeSpanInput(resumeData: unknown, suspendedToolInfo?: { toolCallId?: string; toolName?: string }): unknown {
@@ -6690,7 +6734,9 @@ export class Agent<
     // For resumed runs, surface resumeData as the span input and link the resumed
     // span back to the original suspended trace. Mirrors Workflow.resume tracing.
     const isResume = !!resumeContext;
-    const suspendedToolInfo = isResume ? this.#getSuspendedToolInfo(resumeContext?.snapshot) : undefined;
+    const suspendedToolInfo = isResume
+      ? this.#getSuspendedToolInfo(resumeContext?.snapshot, options.toolCallId)
+      : undefined;
     const persistedTracingContext = isResume
       ? (resumeContext?.snapshot?.tracingContext as
           | { traceId?: string; spanId?: string; parentSpanId?: string }

packages/core/src/workflows/evented/step-executor.ts

@@ -120,6 +120,17 @@ export class StepExecutor extends MastraBase {
     let suspendDataToUse =
       params.stepResults[step.id]?.status === 'suspended' ? params.stepResults[step.id]?.suspendPayload : undefined;
 
+    // A suspended foreach step's step-level suspendPayload only carries the FIRST suspended
+    // iteration's payload. When resuming a specific iteration, use that iteration's own payload
+    // from `__workflow_meta.foreachOutput` so parallel suspensions don't read a sibling's data
+    // (e.g. another tool call's suspended run id).
+    if (suspendDataToUse && typeof params.foreachIdx === 'number') {
+      const iterationResult = suspendDataToUse.__workflow_meta?.foreachOutput?.[params.foreachIdx];
+      if (iterationResult?.status === 'suspended' && iterationResult.suspendPayload) {
+        suspendDataToUse = iterationResult.suspendPayload;
+      }
+    }
+
     // Filter out internal workflow metadata before exposing to step code
     if (suspendDataToUse && '__workflow_meta' in suspendDataToUse) {
       const { __workflow_meta, ...userSuspendData } = suspendDataToUse;

packages/core/src/workflows/handlers/step.ts

@@ -135,6 +135,18 @@ export async function executeStep(
   let suspendDataToUse =
     stepResults[step.id]?.status === 'suspended' ? stepResults[step.id]?.suspendPayload : undefined;
 
+  // A suspended foreach step's step-level suspendPayload only carries the FIRST suspended
+  // iteration's payload. When resuming a specific iteration, use that iteration's own payload
+  // from `__workflow_meta.foreachOutput` so parallel suspensions don't read a sibling's data
+  // (e.g. another tool call's suspended run id).
+  const foreachIndex = executionContext.foreachIndex;
+  if (suspendDataToUse && foreachIndex !== undefined) {
+    const iterationResult = suspendDataToUse.__workflow_meta?.foreachOutput?.[foreachIndex];
+    if (iterationResult?.status === 'suspended' && iterationResult.suspendPayload) {
+      suspendDataToUse = iterationResult.suspendPayload;
+    }
+  }
+
   // Filter out internal workflow metadata before exposing to step code
   if (suspendDataToUse && '__workflow_meta' in suspendDataToUse) {
     const { __workflow_meta, ...userSuspendData } = suspendDataToUse;

packages/core/src/workflows/workflow.ts

@@ -2660,10 +2660,29 @@ export class Workflow<
       for (const [stepName, stepResult] of suspendedSteps) {
         // @ts-expect-error - context type mismatch
         const suspendPath: string[] = [stepName, ...(stepResult?.suspendPayload?.__workflow_meta?.path ?? [])];
+        const nestedMeta = (stepResult as any)?.suspendPayload?.__workflow_meta ?? {};
+        // Keep the nested workflow metadata (foreachIndex, foreachOutput, resumeLabels) when
+        // propagating a suspension to the parent — mirrors the evented engine — so the parent
+        // snapshot is self-describing about EVERY parked iteration, not just the first one.
+        // Only runId and path change as we propagate up. Per-iteration `__streamState` blobs
+        // are stripped from the propagated copies: they can be large and resume reads them
+        // from the nested run's own snapshot, so the parent only needs the identifying fields.
+        const propagatedForeachOutput = Array.isArray(nestedMeta.foreachOutput)
+          ? nestedMeta.foreachOutput.map((entry: any) => {
+              if (entry?.status !== 'suspended' || !entry.suspendPayload) return entry;
+              const { __streamState: _streamState, ...suspendPayload } = entry.suspendPayload;
+              return { ...entry, suspendPayload };
+            })
+          : undefined;
         await suspend(
           {
             ...(stepResult as any)?.suspendPayload,
-            __workflow_meta: { runId: run.runId, path: suspendPath },
+            __workflow_meta: {
+              ...nestedMeta,
+              ...(propagatedForeachOutput ? { foreachOutput: propagatedForeachOutput } : {}),
+              runId: run.runId,
+              path: suspendPath,
+            },
           },
           {
             resumeLabel: Object.keys(res.resumeLabels ?? {}),