Fix evented wflow nested loop (#16312)

## Description

<!-- Provide a brief description of the changes in this PR -->

Fix evented wflow nested loop

## Related Issue(s)

<!-- Link to the issue(s) this PR addresses, using hashtag notation:
Fixes #123 -->

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Code refactoring
- [ ] Performance improvement
- [x] Test update

## Checklist

- [ ] I have made corresponding changes to the documentation (if
applicable)
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have addressed all Coderabbit comments on this PR


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## ELI5 (Explain Like I'm 5)

This PR fixes a bug where a workflow run started inside a loop (a
"nested workflow") didn't let its parent loop resume or finish
correctly. It detects when a finished child run belongs to a loop step
in the parent and routes the completion into the loop handler so the
parent can continue; tests ensure nested do-until and do-while loops
work.

## Changes Made

### Bug Fix: Nested Loop Handling in Evented Workflows
- processWorkflowEnd now inspects
parentWorkflow.stepGraph[parentWorkflow.executionPath[0]] and, if that
step is a loop, calls processWorkflowLoop with the parent workflow's
richer execution metadata so the parent loop resumes/handles the nested
run correctly. If the parent step is not a loop it falls back to
publishing a workflow.step.end for the parent (including parentContext
and nestedRunId).
- Added cleanupRun to remove abort controllers, parent-child
relationship entries, and runFormats when a workflow completes;
cancelRunAndChildren recursively aborts child runs.
- parentChildRelationships map added to track child runId → parent runId
for nested workflows and used during cancellation/cleanup.

### Type/API Updates
- Expanded ParentWorkflow public type
(packages/core/src/workflows/evented/workflow-event-processor/index.ts)
to include execution/resume fields required by nested handling:
- Added: stepGraph: StepFlowEntry[], activeSteps: Record<string,
boolean>, resumeSteps: string[], resumeData: any, input: any
  - Added optional: parentContext?: { workflowId: string; input: any }
- Retained existing fields (workflowId, runId, executionPath, resume,
stepResults, stepId, parentWorkflow?)

### Tests
- workflows/_test-utils/src/domains/loops.ts: expanded test utilities to
add nested loop coverage:
- New nested dountil and nested dowhile workflow registrations (inner
increment workflows used as loop bodies) and corresponding mocks.
- Tests exercise nested-until and nested-while scenarios and include
other loop cases (immediate-exit, non-nested loops).

### Release/Packaging
- .changeset/rare-bikes-wear.md: new changeset marking a patch for
@mastra/core describing "Fix nested loops in evented workflow".

## Files Touched (high level)
- packages/core/src/workflows/evented/workflow-event-processor/index.ts
— core logic: processWorkflowEnd branching, ParentWorkflow type
expansion, abort controller & parent-child tracking and cleanup.
- workflows/_test-utils/src/domains/loops.ts — added nested loop
workflow definitions and test mocks.
- .changeset/rare-bikes-wear.md — changeset entry.

## Impact
- Fixes nested-loop completion/resume behavior for evented workflows and
adds tests to prevent regressions.
- Low-risk patch scope but touches core event-processing logic;
reviewers should focus on correctness of processWorkflowEnd →
processWorkflowLoop interaction, cleanup/cancellation semantics, and the
ParentWorkflow shape used across event payloads.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: taofeeq-deru

.changeset/rare-bikes-wear.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fix nested loops in evented workflow

packages/core/src/workflows/evented/workflow-event-processor/index.ts

@@ -59,6 +59,15 @@ export type ParentWorkflow = {
   stepResults: Record<string, StepResult<any, any, any, any>>;
   parentWorkflow?: ParentWorkflow;
   stepId: string;
+  stepGraph: StepFlowEntry[];
+  activeSteps: Record<string, boolean>;
+  resumeSteps: string[];
+  resumeData: any;
+  input: any;
+  parentContext?: {
+    workflowId: string;
+    input: any;
+  };
 };
 
 export class WorkflowEventProcessor extends EventProcessor {
@@ -361,27 +370,55 @@ export class WorkflowEventProcessor extends EventProcessor {
 
     // handle nested workflow
     if (parentWorkflow) {
-      await this.mastra.pubsub.publish('workflows', {
-        type: 'workflow.step.end',
-        runId: parentWorkflow.runId, // Use parent's runId for event routing
-        data: {
-          workflowId: parentWorkflow.workflowId,
-          runId: parentWorkflow.runId,
-          executionPath: parentWorkflow.executionPath,
-          resumeSteps,
-          stepResults: parentWorkflow.stepResults,
-          prevResult,
-          resumeData,
-          activeSteps,
-          parentWorkflow: parentWorkflow.parentWorkflow,
-          parentContext: parentWorkflow,
-          requestContext,
-          timeTravel,
-          perStep,
-          state: finalState,
-          nestedRunId: runId, // Pass nested workflow's runId for step retrieval
-        },
-      });
+      // get the step from the parent workflow and process it if it's a loop
+      const step = parentWorkflow.stepGraph[parentWorkflow.executionPath[0]!];
+      if (step?.type === 'loop') {
+        // pick workflow information from parentWorkflow as the workflow end being processed here is actually a step in the parentWorkflow
+        await processWorkflowLoop(
+          {
+            workflow: parentWorkflow as unknown as Workflow,
+            workflowId: parentWorkflow.workflowId,
+            prevResult,
+            runId: parentWorkflow.runId,
+            executionPath: parentWorkflow.executionPath,
+            stepResults: parentWorkflow.stepResults,
+            activeSteps: parentWorkflow.activeSteps,
+            resumeSteps: parentWorkflow.resumeSteps,
+            resumeData: parentWorkflow.resumeData,
+            parentWorkflow: parentWorkflow.parentWorkflow,
+            requestContext,
+            retryCount: 0,
+          },
+          {
+            pubsub: this.mastra.pubsub,
+            stepExecutor: this.stepExecutor,
+            step,
+            stepResult: prevResult,
+          },
+        );
+      } else {
+        await this.mastra.pubsub.publish('workflows', {
+          type: 'workflow.step.end',
+          runId: parentWorkflow.runId, // Use parent's runId for event routing
+          data: {
+            workflowId: parentWorkflow.workflowId,
+            runId: parentWorkflow.runId,
+            executionPath: parentWorkflow.executionPath,
+            resumeSteps,
+            stepResults: parentWorkflow.stepResults,
+            prevResult,
+            resumeData,
+            activeSteps,
+            parentWorkflow: parentWorkflow.parentWorkflow,
+            parentContext: parentWorkflow,
+            requestContext,
+            timeTravel,
+            perStep,
+            state: finalState,
+            nestedRunId: runId, // Pass nested workflow's runId for step retrieval
+          },
+        });
+      }
     }
 
     await this.mastra.pubsub.publish('workflows-finish', {
@@ -846,11 +883,14 @@ export class WorkflowEventProcessor extends EventProcessor {
               stepId: step.step.id,
               workflowId,
               runId,
+              stepGraph,
               executionPath,
               resumeSteps,
               stepResults,
               input: prevResult,
               parentWorkflow,
+              activeSteps,
+              resumeData,
             },
             executionPath: nestedExecutionPath as any,
             runId: nestedRunId,
@@ -909,11 +949,14 @@ export class WorkflowEventProcessor extends EventProcessor {
               stepId: step.step.id,
               workflowId,
               runId,
+              stepGraph,
               executionPath,
               resumeSteps,
               stepResults,
               input: prevResult,
               parentWorkflow,
+              activeSteps,
+              resumeData,
             },
             executionPath: snapshot?.suspendedPaths?.[nestedSteps[0]!] as any,
             runId: nestedRunId,
@@ -961,12 +1004,15 @@ export class WorkflowEventProcessor extends EventProcessor {
               stepId: step.step.id,
               workflowId,
               runId,
+              stepGraph,
               executionPath,
               resumeSteps,
               stepResults,
               timeTravel,
               input: prevResult,
               parentWorkflow,
+              activeSteps,
+              resumeData,
             },
             executionPath: timeTravelParams.executionPath,
             runId: randomUUID(),
@@ -990,12 +1036,15 @@ export class WorkflowEventProcessor extends EventProcessor {
             parentWorkflow: {
               stepId: step.step.id,
               workflowId,
+              stepGraph,
               runId,
               executionPath,
               resumeSteps,
               stepResults,
               input: prevResult,
               parentWorkflow,
+              activeSteps,
+              resumeData,
             },
             executionPath: [0],
             runId: randomUUID(),