fix(core): filter delegated agent stream chunks (#16071)

This fixes supervisor output processors so they can see chunks streamed
by delegated sub-agents and filter them before consumers receive them.
The existing data chunk path is unchanged, and streams without output
processors keep the same passthrough behavior.

Before, filtering inside `processOutputStream` could not stop nested
sub-agent chunks because they bypassed the processor pipeline.

```ts
async processOutputStream({ part }) {
  if (part.type === "tool-output") return null;
  return part;
}
```

After this change, supervisor agents using `agents:` can use that same
processor to drop or rewrite delegated sub-agent stream chunks.

Closes #15899.

Tested with `pnpm --filter ./packages/core check`, the new supervisor
stream processor tests, and focused non-E2E core unit coverage for
processors, loop, and agent tests.


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## ELI5

This PR makes supervisor agents apply their configured output filters to
messages coming from delegated sub-agents, so those forwarded messages
can be observed, rewritten, or dropped just like regular tool output.

## Summary of Changes

### Changeset Entry
Adds a changeset noting a patch release for @mastra/core: supervisor
output processors can now filter streamed chunks emitted by delegated
sub-agents.

### Stream Processing Pipeline Update
Updates packages/core/src/loop/workflows/stream.ts to route non-data-*
chunks injected via the output writer (e.g., `tool-output` wrappers for
forwarded sub-agent streams) through the configured output processor
pipeline (dataChunkProcessorRunner.processPart). Behavior details:
- Chunks are processed by processOutputStream and may be rewritten,
passed through, or dropped.
- If a processor blocks content, the stream emits a `tripwire` chunk
(including reason, retry, metadata, and processorId) and the blocked
chunk is not enqueued.
- If processing returns falsy, the chunk is dropped.
- If no output processors are configured, the previous passthrough
behavior is preserved.

### New Tests
Adds
packages/core/src/agent/__tests__/supervisor-output-processor-stream.test.ts
covering supervisor behavior with nested agent streams:
- Confirms nested-agent `tool-output` wrapper chunks reach
processOutputStream.
- Verifies processors can drop forwarded sub-agent chunks while leaving
the supervisor’s own `tool-call`/`tool-result` envelopes intact.
- Confirms passthrough behavior when no processors are configured.
- Validates processors can rewrite nested-agent payloads before
delivery.
- Validates processor-triggered aborts produce `tripwire` chunks with
processor id and reason, and prevent the blocked nested `text-delta`
from reaching the consumer.

### Tests / Validation
PR includes focused unit tests (supervisor stream processor tests and
processor/loop/agent coverage) and was validated with project check/test
steps (e.g., pnpm --filter ./packages/core check and new unit tests).

## Linked Issue
Closes #15899 — enables supervisor output processors to filter nested
delegated sub-agent stream chunks.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: Mastra Code (openai/gpt-5.5) <noreply@mastra.ai>

Author: roaminro

.changeset/humble-kids-post.md

@@ -0,0 +1,5 @@
+---
+'@mastra/core': patch
+---
+
+Fixed supervisor output processors so they can filter streamed chunks from delegated sub-agents.

packages/core/src/agent/__tests__/supervisor-output-processor-stream.test.ts

@@ -0,0 +1,309 @@
+import { convertArrayToReadableStream, MockLanguageModelV2 } from '@internal/ai-sdk-v5/test';
+import { describe, expect, it } from 'vitest';
+import { Mastra } from '../../mastra';
+import type { Processor, ProcessOutputStreamArgs } from '../../processors/index';
+import { InMemoryStore } from '../../storage';
+import type { ChunkType } from '../../stream';
+import { Agent } from '../agent';
+
+/**
+ * Verifies that output processors on a supervisor agent can observe and filter
+ * chunks forwarded from sub-agents delegated via the `agents:` option.
+ *
+ * Sub-agent chunks are forwarded through the parent stream's writer (not the
+ * LLM's own fullStream). Previously the writer only routed `data-*` chunks
+ * through output processors; everything else (including the `tool-output`
+ * wrapper that the synthetic agent-* tool emits around sub-agent chunks)
+ * bypassed `processOutputStream`. This test locks in the fix that runs every
+ * writer-injected chunk through processors.
+ */
+
+function makeSubAgent() {
+  const subAgentModel = new MockLanguageModelV2({
+    doStream: async () => ({
+      rawCall: { rawPrompt: null, rawSettings: {} },
+      warnings: [],
+      stream: convertArrayToReadableStream([
+        { type: 'stream-start', warnings: [] },
+        { type: 'response-metadata', id: 'sub-id-0', modelId: 'mock', timestamp: new Date(0) },
+        { type: 'text-start', id: 'sub-text-1' },
+        { type: 'text-delta', id: 'sub-text-1', delta: 'sub-agent says hi' },
+        { type: 'text-end', id: 'sub-text-1' },
+        {
+          type: 'finish',
+          finishReason: 'stop',
+          usage: { inputTokens: 5, outputTokens: 10, totalTokens: 15 },
+        },
+      ]),
+    }),
+  });
+
+  return new Agent({
+    id: 'sub-agent',
+    name: 'sub-agent',
+    description: 'A sub-agent.',
+    instructions: 'You answer briefly.',
+    model: subAgentModel,
+  });
+}
+
+function makeSupervisorModel() {
+  let callCount = 0;
+  return new MockLanguageModelV2({
+    doStream: async () => {
+      callCount++;
+      if (callCount === 1) {
+        return {
+          rawCall: { rawPrompt: null, rawSettings: {} },
+          warnings: [],
+          stream: convertArrayToReadableStream([
+            { type: 'stream-start', warnings: [] },
+            { type: 'response-metadata', id: 'sup-id-0', modelId: 'mock', timestamp: new Date(0) },
+            {
+              type: 'tool-call',
+              toolCallId: 'sup-call-1',
+              toolName: 'agent-subAgent',
+              input: JSON.stringify({ prompt: 'do the thing' }),
+            },
+            {
+              type: 'finish',
+              finishReason: 'tool-calls',
+              usage: { inputTokens: 10, outputTokens: 20, totalTokens: 30 },
+            },
+          ]),
+        };
+      }
+      return {
+        rawCall: { rawPrompt: null, rawSettings: {} },
+        warnings: [],
+        stream: convertArrayToReadableStream([
+          { type: 'stream-start', warnings: [] },
+          { type: 'response-metadata', id: 'sup-id-1', modelId: 'mock', timestamp: new Date(0) },
+          { type: 'text-start', id: 'sup-text-1' },
+          { type: 'text-delta', id: 'sup-text-1', delta: 'all done' },
+          { type: 'text-end', id: 'sup-text-1' },
+          {
+            type: 'finish',
+            finishReason: 'stop',
+            usage: { inputTokens: 10, outputTokens: 20, totalTokens: 30 },
+          },
+        ]),
+      };
+    },
+  });
+}
+
+describe('Supervisor pattern: output processor stream visibility', () => {
+  it('routes nested-agent tool-output chunks through processOutputStream', async () => {
+    const capturedChunkTypes: string[] = [];
+
+    class RecordingProcessor implements Processor {
+      readonly id = 'recording-processor';
+      readonly name = 'Recording Processor';
+
+      async processOutputStream({ part }: ProcessOutputStreamArgs) {
+        capturedChunkTypes.push(part.type);
+        return part;
+      }
+    }
+
+    const supervisor = new Agent({
+      id: 'supervisor',
+      name: 'supervisor',
+      instructions: 'You orchestrate sub-agents.',
+      model: makeSupervisorModel(),
+      agents: { subAgent: makeSubAgent() },
+      outputProcessors: [new RecordingProcessor()],
+    });
+
+    new Mastra({
+      agents: { supervisor },
+      storage: new InMemoryStore(),
+    });
+
+    const stream = await supervisor.stream('Please delegate', { maxSteps: 5 });
+    for await (const _chunk of stream.fullStream) {
+      // drain
+    }
+
+    // Sub-agent chunks are wrapped as `tool-output` by the synthetic agent-*
+    // tool's ToolStream (prefix: 'tool'). With the fix, these reach
+    // processOutputStream so processors can observe and filter them.
+    expect(capturedChunkTypes).toContain('tool-output');
+  });
+
+  it('lets a processor drop nested-agent chunks before they reach the consumer', async () => {
+    class FilterNestedAgentChunks implements Processor {
+      readonly id = 'filter-nested-agent-chunks';
+      readonly name = 'Filter Nested Agent Chunks';
+
+      async processOutputStream({ part }: ProcessOutputStreamArgs) {
+        // tool-output chunks where the wrapped output is itself an agent chunk
+        // (i.e. a chunk that has a `from` field of 'AGENT') indicate forwarded
+        // sub-agent stream content. Drop them.
+        if (part?.type === 'tool-output' && part?.payload?.output?.from === 'AGENT') {
+          return null;
+        }
+        return part;
+      }
+    }
+
+    const supervisor = new Agent({
+      id: 'supervisor',
+      name: 'supervisor',
+      instructions: 'You orchestrate sub-agents.',
+      model: makeSupervisorModel(),
+      agents: { subAgent: makeSubAgent() },
+      outputProcessors: [new FilterNestedAgentChunks()],
+    });
+
+    new Mastra({
+      agents: { supervisor },
+      storage: new InMemoryStore(),
+    });
+
+    const stream = await supervisor.stream('Please delegate', { maxSteps: 5 });
+    const consumerChunks: ChunkType[] = [];
+    for await (const chunk of stream.fullStream) {
+      consumerChunks.push(chunk);
+    }
+
+    // The consumer must NOT see any forwarded sub-agent chunks.
+    const forwardedSubAgentChunks = consumerChunks.filter(
+      c => c.type === 'tool-output' && c?.payload?.output?.from === 'AGENT',
+    );
+    expect(forwardedSubAgentChunks).toHaveLength(0);
+
+    // The surrounding tool-call / tool-result envelope (from the supervisor's
+    // own LLM stream) must still pass through unaffected.
+    const consumerChunkTypes = consumerChunks.map(c => c.type);
+    expect(consumerChunkTypes).toContain('tool-call');
+    expect(consumerChunkTypes).toContain('tool-result');
+  });
+
+  it('passes nested-agent chunks through unchanged when no output processors are configured', async () => {
+    // No outputProcessors: dataChunkProcessorRunner is undefined and the new
+    // branch must short-circuit to the original safeEnqueue fallthrough so
+    // existing behavior is preserved.
+    const supervisor = new Agent({
+      id: 'supervisor',
+      name: 'supervisor',
+      instructions: 'You orchestrate sub-agents.',
+      model: makeSupervisorModel(),
+      agents: { subAgent: makeSubAgent() },
+    });
+
+    new Mastra({
+      agents: { supervisor },
+      storage: new InMemoryStore(),
+    });
+
+    const stream = await supervisor.stream('Please delegate', { maxSteps: 5 });
+    const consumerChunks: ChunkType[] = [];
+    for await (const chunk of stream.fullStream) {
+      consumerChunks.push(chunk);
+    }
+
+    const forwardedSubAgentChunks = consumerChunks.filter(
+      c => c.type === 'tool-output' && c?.payload?.output?.from === 'AGENT',
+    );
+    // Sub-agent chunks must still reach the consumer when no processors are configured.
+    expect(forwardedSubAgentChunks.length).toBeGreaterThan(0);
+  });
+
+  it('lets a processor rewrite a nested-agent chunk before it reaches the consumer', async () => {
+    class RewriteNestedAgentChunks implements Processor {
+      readonly id = 'rewrite-nested-agent-chunks';
+      readonly name = 'Rewrite Nested Agent Chunks';
+
+      async processOutputStream({ part }: ProcessOutputStreamArgs) {
+        if (part?.type === 'tool-output' && part?.payload?.output?.from === 'AGENT') {
+          return {
+            ...part,
+            payload: { ...part.payload, rewritten: true },
+          };
+        }
+        return part;
+      }
+    }
+
+    const supervisor = new Agent({
+      id: 'supervisor',
+      name: 'supervisor',
+      instructions: 'You orchestrate sub-agents.',
+      model: makeSupervisorModel(),
+      agents: { subAgent: makeSubAgent() },
+      outputProcessors: [new RewriteNestedAgentChunks()],
+    });
+
+    new Mastra({
+      agents: { supervisor },
+      storage: new InMemoryStore(),
+    });
+
+    const stream = await supervisor.stream('Please delegate', { maxSteps: 5 });
+    const consumerChunks: ChunkType[] = [];
+    for await (const chunk of stream.fullStream) {
+      consumerChunks.push(chunk);
+    }
+
+    const forwardedSubAgentChunks = consumerChunks.filter(
+      c => c.type === 'tool-output' && c?.payload?.output?.from === 'AGENT',
+    );
+    expect(forwardedSubAgentChunks.length).toBeGreaterThan(0);
+    // Every forwarded sub-agent chunk must carry the rewritten marker.
+    for (const chunk of forwardedSubAgentChunks) {
+      expect(chunk.payload.rewritten).toBe(true);
+    }
+  });
+
+  it('emits a tripwire when a processor aborts on a nested-agent chunk', async () => {
+    class TripwireOnNestedAgentChunks implements Processor {
+      readonly id = 'tripwire-on-nested-agent-chunks';
+      readonly name = 'Tripwire On Nested Agent Chunks';
+
+      async processOutputStream({ part, abort }: ProcessOutputStreamArgs) {
+        if (
+          part?.type === 'tool-output' &&
+          part?.payload?.output?.from === 'AGENT' &&
+          part?.payload?.output?.type === 'text-delta'
+        ) {
+          abort('nested agent chunk blocked');
+        }
+        return part;
+      }
+    }
+
+    const supervisor = new Agent({
+      id: 'supervisor',
+      name: 'supervisor',
+      instructions: 'You orchestrate sub-agents.',
+      model: makeSupervisorModel(),
+      agents: { subAgent: makeSubAgent() },
+      outputProcessors: [new TripwireOnNestedAgentChunks()],
+    });
+
+    new Mastra({
+      agents: { supervisor },
+      storage: new InMemoryStore(),
+    });
+
+    const stream = await supervisor.stream('Please delegate', { maxSteps: 5 });
+    const consumerChunks: ChunkType[] = [];
+    for await (const chunk of stream.fullStream) {
+      consumerChunks.push(chunk);
+    }
+
+    const tripwireChunks = consumerChunks.filter(c => c.type === 'tripwire');
+    expect(tripwireChunks.length).toBeGreaterThan(0);
+    expect(tripwireChunks[0].payload.reason).toBe('nested agent chunk blocked');
+    expect(tripwireChunks[0].payload.processorId).toBe('tripwire-on-nested-agent-chunks');
+
+    // The specific blocked sub-agent text delta must NOT reach the consumer.
+    const forwardedSubAgentTextDeltas = consumerChunks.filter(
+      c =>
+        c.type === 'tool-output' && c?.payload?.output?.from === 'AGENT' && c?.payload?.output?.type === 'text-delta',
+    );
+    expect(forwardedSubAgentTextDeltas).toHaveLength(0);
+  });
+});

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

@@ -128,6 +128,49 @@ export function workflowLoopStream<Tools extends ToolSet = ToolSet, OUTPUT = und
           safeEnqueue(controller, processedChunk);
           return;
         }
+
+        // Non data-* chunks injected via this writer (e.g. `tool-output` from
+        // sub-agents delegated through the `agents:` option, or
+        // `workflow-step-output` from workflow tools) bypass the LLM's own
+        // processor pipeline. Route them through configured output processors
+        // here so users can filter/redact nested chunks via processOutputStream.
+        if (dataChunkProcessorRunner) {
+          const {
+            part: processed,
+            blocked,
+            reason,
+            tripwireOptions,
+            processorId,
+          } = await dataChunkProcessorRunner.processPart(
+            chunk,
+            (rest.processorStates ?? dataChunkProcessorStates!) as Map<string, ProcessorState<OUTPUT>>,
+            undefined,
+            requestContext,
+            messageList,
+            0,
+            dataChunkStreamWriter,
+          );
+
+          if (blocked) {
+            safeEnqueue(controller, {
+              type: 'tripwire',
+              runId,
+              from: ChunkFrom.AGENT,
+              payload: {
+                reason: reason || 'Output processor blocked content',
+                retry: tripwireOptions?.retry,
+                metadata: tripwireOptions?.metadata,
+                processorId,
+              },
+            } as ChunkType<OUTPUT>);
+            return;
+          }
+
+          if (!processed) return;
+          safeEnqueue(controller, processed as ChunkType<OUTPUT>);
+          return;
+        }
+
         safeEnqueue(controller, chunk);
       };