fix(core): share processor state across step lifecycle chunks during streaming (#17370)

Co-authored-by: Mastra Code (anthropic/claude-opus-4-8) <noreply@mastra.ai>

Author: abhiaiyer91

.changeset/lifecycle-chunk-processor-state.md

@@ -0,0 +1,7 @@
+---
+'@mastra/core': patch
+---
+
+Fixed output processor state continuity for step lifecycle chunks during streaming. Lifecycle chunks (e.g. `step-finish`) routed through the stream `outputWriter` now share the same per-processor state map as the main model-output path, so state set in `processOutputStream` while handling content chunks is visible when the lifecycle chunk is processed (and in `processOutputResult`). Previously these chunks were handled with an isolated, empty state.
+
+Note: as part of routing lifecycle chunks through output processors, an aborted stream now surfaces content the model produced before the abort (e.g. a `text-start` chunk and partial `text`) instead of dropping it. Consumers that assumed an aborted result always had empty text may now observe partial output.

observability/mastra/src/__snapshots__/agent-structured-output-trace-generate.json

@@ -294,11 +294,11 @@
               "environment": "test"
             },
             "input": {
-              "totalChunks": 5,
+              "totalChunks": 6,
               "accumulatedText": "Mock V2 generate response"
             },
             "output": {
-              "totalChunks": 5,
+              "totalChunks": 6,
               "accumulatedText": "Mock V2 generate response"
             }
           }

observability/mastra/src/__snapshots__/agent-structured-output-trace.json

@@ -292,11 +292,11 @@
               "environment": "test"
             },
             "input": {
-              "totalChunks": 3,
+              "totalChunks": 4,
               "accumulatedText": "Mock V2 stream response"
             },
             "output": {
-              "totalChunks": 3,
+              "totalChunks": 4,
               "accumulatedText": "Mock V2 stream response"
             }
           }

packages/core/src/agent/agent-stream-processor.test.ts

@@ -217,6 +217,46 @@ describe('Processor state persistence across processOutputStream and processOutp
     expect(stateInOutputResult?.streamProcessed).toBe(true);
     expect(stateInOutputResult?.chunks).toEqual(['Hello ', 'World']);
   });
+
+  it('exposes prior processOutputStream state to the step-finish lifecycle chunk', async () => {
+    // Regression for #16687: lifecycle chunks (step-start/step-finish) are routed
+    // through output processors via a separate ProcessorRunner. That runner must share
+    // the same processorStates map as the main model-output path, otherwise the
+    // step-finish chunk is handled with a fresh, empty state and loses continuity.
+    const stateByChunkType: Record<string, Record<string, unknown>> = {};
+
+    class LifecycleStateProcessor implements Processor {
+      readonly id = 'lifecycle-state-processor';
+      readonly name = 'Lifecycle State Processor';
+
+      async processOutputStream({ part, state }: any) {
+        if (part.type === 'text-delta') {
+          state.sawText = true;
+        }
+        // Snapshot the state the processor sees for each chunk type
+        stateByChunkType[part.type] = { ...state };
+        return part;
+      }
+    }
+
+    const agent = new Agent({
+      id: 'lifecycle-agent',
+      name: 'lifecycle-agent',
+      instructions: 'Test agent',
+      model: mockModel as any,
+      outputProcessors: [new LifecycleStateProcessor()],
+    });
+
+    const stream = await agent.stream('test message');
+    for await (const _chunk of stream.textStream) {
+      // consume
+    }
+
+    // The processor must have observed the step-finish lifecycle chunk...
+    expect(stateByChunkType['step-finish']).toBeDefined();
+    // ...and at that point it must still see the state set while handling text-delta.
+    expect(stateByChunkType['step-finish']?.sawText).toBe(true);
+  });
 });
 
 describe('OutputProcessor Metadata with Streaming (Issue #11454)', () => {

packages/core/src/loop/test-utils/options.ts

@@ -8141,6 +8141,15 @@ export function optionsTests({ loopFn, runId }: { loopFn: typeof loop; runId: st
               "runId": "test-run-id",
               "type": "step-start",
             },
+            {
+              "from": "AGENT",
+              "payload": {
+                "id": "id-2",
+                "providerMetadata": undefined,
+              },
+              "runId": "test-run-id",
+              "type": "text-start",
+            },
             {
               "from": "AGENT",
               "payload": {},
@@ -8203,7 +8212,7 @@ export function optionsTests({ loopFn, runId }: { loopFn: typeof loop; runId: st
                 },
                 "output": {
                   "steps": [],
-                  "text": "",
+                  "text": "Hello",
                   "toolCalls": [],
                   "usage": {
                     "inputTokens": 0,
@@ -8692,6 +8701,15 @@ export function optionsTests({ loopFn, runId }: { loopFn: typeof loop; runId: st
               "runId": "test-run-id",
               "type": "step-start",
             },
+            {
+              "from": "AGENT",
+              "payload": {
+                "id": "id-2",
+                "providerMetadata": undefined,
+              },
+              "runId": "test-run-id",
+              "type": "text-start",
+            },
             {
               "from": "AGENT",
               "payload": {},
@@ -8817,7 +8835,7 @@ export function optionsTests({ loopFn, runId }: { loopFn: typeof loop; runId: st
                 },
                 "output": {
                   "steps": [],
-                  "text": "",
+                  "text": "Hello",
                   "toolCalls": [],
                   "usage": {
                     "inputTokens": 3,

packages/core/src/loop/workflows/stream.ts

@@ -35,16 +35,23 @@ export function workflowLoopStream<Tools extends ToolSet = ToolSet, OUTPUT = und
       // Normalize requestContext so data-chunk processors and the agentic loop share the same instance
       const requestContext = rest.requestContext ?? new RequestContext();
 
-      // Create a ProcessorRunner for data-* chunks so they go through output processors
+      // Create a ProcessorRunner for chunks routed through outputWriter (data-* custom
+      // chunks plus lifecycle chunks like step-finish) so they go through output processors.
+      // Share the loop's processorStates map so these chunks see the same per-processor state
+      // that the main model-output stream path populates; otherwise the runner would build an
+      // isolated empty state map and break state continuity across the step lifecycle.
       const hasOutputProcessors = rest.outputProcessors && rest.outputProcessors.length > 0;
+      const dataChunkProcessorStates = hasOutputProcessors
+        ? (rest.processorStates ?? new Map<string, ProcessorState>())
+        : undefined;
       const dataChunkProcessorRunner = hasOutputProcessors
         ? new ProcessorRunner({
             outputProcessors: rest.outputProcessors,
             logger: rest.logger || new ConsoleLogger({ level: 'error' }),
             agentName: agentId || 'unknown',
+            processorStates: dataChunkProcessorStates,
           })
         : undefined;
-      const dataChunkProcessorStates = hasOutputProcessors ? new Map<string, ProcessorState>() : undefined;
 
       // Create a ProcessorStreamWriter so output processors can emit custom chunks back to the stream
       const dataChunkStreamWriter = {
@@ -69,7 +76,7 @@ export function workflowLoopStream<Tools extends ToolSet = ToolSet, OUTPUT = und
               processorId,
             } = await dataChunkProcessorRunner.processPart(
               chunk,
-              (rest.processorStates ?? dataChunkProcessorStates!) as Map<string, ProcessorState<OUTPUT>>,
+              dataChunkProcessorStates! as Map<string, ProcessorState<OUTPUT>>,
               undefined, // observabilityContext
               requestContext,
               messageList,
@@ -143,7 +150,7 @@ export function workflowLoopStream<Tools extends ToolSet = ToolSet, OUTPUT = und
             processorId,
           } = await dataChunkProcessorRunner.processPart(
             chunk,
-            (rest.processorStates ?? dataChunkProcessorStates!) as Map<string, ProcessorState<OUTPUT>>,
+            dataChunkProcessorStates! as Map<string, ProcessorState<OUTPUT>>,
             undefined,
             requestContext,
             messageList,