feat: enhanced processor system with retry, workflow orchestration, and tripwire improvements

.changeset/three-falcons-pump.md

@@ -0,0 +1,27 @@
+---
+'@mastra/playground-ui': patch
+'@mastra/client-js': patch
+'@mastra/ai-sdk': patch
+'@mastra/react': patch
+'@mastra/server': patch
+'@mastra/inngest': patch
+---
+
+Support new Workflow tripwire run status. Tripwires that are thrown from within a workflow will now bubble up and return a graceful state with information about tripwires.
+
+When a workflow contains an agent step that triggers a tripwire, the workflow returns with `status: 'tripwire'` and includes tripwire details:
+
+```typescript showLineNumbers copy
+const run = await workflow.createRun();
+const result = await run.start({ inputData: { message: 'Hello' } });
+
+if (result.status === 'tripwire') {
+  console.log('Workflow terminated by tripwire:', result.tripwire?.reason);
+  console.log('Processor ID:', result.tripwire?.processorId);
+  console.log('Retry requested:', result.tripwire?.retry);
+}
+```
+
+Adds new UI state for tripwire in agent chat and workflow UI.
+
+This is distinct from `status: 'failed'` which indicates an unexpected error. A tripwire status means a processor intentionally stopped execution (e.g., for content moderation).

.changeset/yummy-hats-tap.md

@@ -0,0 +1,54 @@
+---
+'@mastra/core': patch
+---
+
+Multiple Processor improvements including:
+
+- Workflows can now return tripwires, they bubble up from agents that return tripwires in a step
+- You can write processors as workflows using the existing Workflow primitive, every processor flow is now a workflow.
+- tripwires that you throw can now return additional information including ability to retry the step
+- New processor method `processOutputStep` added which runs after every step.
+
+**What's new:**
+
+**1. Retry mechanism with LLM feedback** - Processors can now request retries with feedback that gets sent back to the LLM:
+
+```typescript
+processOutputStep: async ({ text, abort, retryCount }) => {
+  if (isLowQuality(text)) {
+    abort('Response quality too low', { retry: true, metadata: { score: 0.6 } });
+  }
+  return [];
+};
+```
+
+Configure with `maxProcessorRetries` (default: 3). Rejected steps are preserved in `result.steps[n].tripwire`. Retries are only available in `processOutputStep` and `processInputStep`. It will replay the step with additional context added.
+
+**2. Workflow orchestration for processors** - Processors can now be composed using workflow primitives:
+
+```typescript
+import { createStep, createWorkflow } from '@mastra/core/workflows';
+import {
+  ProcessorStepSchema,
+} from '@mastra/core/processors';
+
+const moderationWorkflow = createWorkflow({ id: 'moderation', inputSchema: ProcessorStepSchema, outputSchema: ProcessorStepSchema })
+  .then(createStep(new lengthValidator({...})))
+  .parallel([createStep(new piiDetector({...}), createStep(new toxicityChecker({...}))])
+  .commit();
+
+const agent = new Agent({ inputProcessors: [moderationWorkflow] });
+```
+
+Every processor array that gets passed to an agent gets added as a workflow
+<img width="614" height="673" alt="image" src="https://github.com/user-attachments/assets/0d79f1fd-8fca-4d86-8b45-22fddea984a8" />
+
+**3. Extended tripwire API** - `abort()` now accepts options for retry control and typed metadata:
+
+```typescript
+abort('reason', { retry: true, metadata: { score: 0.8, category: 'quality' } });
+```
+
+**4. New `processOutputStep` method** - Per-step output processing with access to step number, finish reason, tool calls, and retry count.
+
+**5. Workflow tripwire status** - Workflows now have a `'tripwire'` status distinct from `'failed'`, properly bubbling up processor rejections.

client-sdks/ai-sdk/src/__tests__/convert-messages.test.ts

@@ -123,7 +123,7 @@ describe('toAISdkFormat', () => {
             type: 'tripwire',
             runId: 'test-run-id',
             payload: {
-              tripwireReason,
+              reason: tripwireReason,
             },
           });
 
@@ -154,7 +154,7 @@ describe('toAISdkFormat', () => {
       // Verify tripwire chunk was received
       expect(tripwireChunk).toBeDefined();
       expect(tripwireChunk.type).toBe('data-tripwire');
-      expect(tripwireChunk.data.tripwireReason).toBe(tripwireReason);
+      expect(tripwireChunk.data.reason).toBe(tripwireReason);
 
       // Verify finish event was sent with finishReason 'other'
       expect(finishChunk).toBeDefined();
@@ -178,7 +178,7 @@ describe('toAISdkFormat', () => {
             type: 'tripwire',
             runId: 'test-run-id',
             payload: {
-              tripwireReason,
+              reason: tripwireReason,
             },
           });
 

client-sdks/ai-sdk/src/helpers.ts

@@ -1,9 +1,27 @@
 import { DefaultGeneratedFile, DefaultGeneratedFileWithType } from '@mastra/core/stream';
-import type { PartialSchemaOutput, OutputSchema, DataChunkType, ChunkType } from '@mastra/core/stream';
-
-import type { InferUIMessageChunk, ObjectStreamPart, TextStreamPart, ToolSet, UIMessage } from 'ai';
+import type {
+  PartialSchemaOutput,
+  OutputSchema,
+  DataChunkType,
+  ChunkType,
+  MastraFinishReason,
+} from '@mastra/core/stream';
+
+import type { InferUIMessageChunk, ObjectStreamPart, TextStreamPart, ToolSet, UIMessage, FinishReason } from 'ai';
 import { isDataChunkType } from './utils';
 
+/**
+ * Maps Mastra's extended finish reasons to AI SDK-compatible values.
+ * 'tripwire' and 'retry' are Mastra-specific reasons for processor scenarios,
+ * which are mapped to 'other' for AI SDK compatibility.
+ */
+export function toAISDKFinishReason(reason: MastraFinishReason): FinishReason {
+  if (reason === 'tripwire' || reason === 'retry') {
+    return 'other';
+  }
+  return reason;
+}
+
 export type OutputChunkType<OUTPUT extends OutputSchema = undefined> =
   | TextStreamPart<ToolSet>
   | ObjectStreamPart<PartialSchemaOutput<OUTPUT>>
@@ -42,7 +60,7 @@ export function convertMastraChunkToAISDKv5<OUTPUT extends OutputSchema = undefi
     case 'finish': {
       return {
         type: 'finish',
-        finishReason: chunk.payload.stepResult.reason,
+        finishReason: toAISDKFinishReason(chunk.payload.stepResult.reason),
         totalUsage: chunk.payload.output.usage,
       };
     }
@@ -182,7 +200,7 @@ export function convertMastraChunkToAISDKv5<OUTPUT extends OutputSchema = undefi
           ...rest,
         },
         usage: chunk.payload.output.usage,
-        finishReason: chunk.payload.stepResult.reason,
+        finishReason: toAISDKFinishReason(chunk.payload.stepResult.reason),
         providerMetadata,
       };
     }
@@ -246,7 +264,10 @@ export function convertMastraChunkToAISDKv5<OUTPUT extends OutputSchema = undefi
       return {
         type: 'data-tripwire',
         data: {
-          tripwireReason: chunk.payload.tripwireReason,
+          reason: chunk.payload.reason,
+          retry: chunk.payload.retry,
+          metadata: chunk.payload.metadata,
+          processorId: chunk.payload.processorId,
         },
       };
     default:

client-sdks/ai-sdk/src/middleware.ts

@@ -21,6 +21,7 @@ import { convertFullStreamChunkToMastra } from '@mastra/core/stream';
 import type { ChunkType } from '@mastra/core/stream';
 import type { MastraEmbeddingModel, MastraVector } from '@mastra/core/vector';
 import { wrapLanguageModel } from 'ai';
+import { toAISDKFinishReason } from './helpers';
 
 /**
  * Memory context for processors that need thread/resource info
@@ -737,7 +738,7 @@ function convertMastraChunkToAISDKStreamPart(chunk: ChunkType): LanguageModelV2S
       const usage = chunk.payload.output?.usage;
       return {
         type: 'finish',
-        finishReason: chunk.payload.stepResult?.reason || 'stop',
+        finishReason: toAISDKFinishReason(chunk.payload.stepResult?.reason || 'stop'),
         usage: usage
           ? {
               inputTokens: usage.inputTokens || 0,

client-sdks/client-js/src/resources/agent.ts

@@ -855,7 +855,7 @@ export class Agent extends BaseResource {
           case 'tripwire': {
             message.parts.push({
               type: 'text',
-              text: chunk.payload.tripwireReason,
+              text: chunk.payload.reason,
             });
 
             execUpdate();

client-sdks/client-js/src/types.ts

@@ -211,6 +211,8 @@ export interface GetWorkflowResponse {
   stepGraph: Workflow['serializedStepGraph'];
   inputSchema: string;
   outputSchema: string;
+  /** Whether this workflow is a processor workflow (auto-generated from agent processors) */
+  isProcessorWorkflow?: boolean;
 }
 
 export type WorkflowRunResult = WorkflowResult<any, any, any, any>;

client-sdks/react/src/lib/ai-sdk/types.ts

@@ -1,8 +1,22 @@
 import { UIMessage } from '@ai-sdk/react';
 import { CompleteAttachment } from '@assistant-ui/react';
 
+/**
+ * Tripwire metadata included when a processor triggers a tripwire
+ */
+export type TripwireMetadata = {
+  /** Whether the agent should retry with feedback */
+  retry?: boolean;
+  /** Custom metadata from the processor */
+  tripwirePayload?: unknown;
+  /** ID of the processor that triggered the tripwire */
+  processorId?: string;
+};
+
 export type MastraUIMessageMetadata = {
-  status?: 'warning' | 'error';
+  status?: 'warning' | 'error' | 'tripwire';
+  /** Tripwire-specific metadata when status is 'tripwire' */
+  tripwire?: TripwireMetadata;
 } & (
   | {
       mode: 'generate';

client-sdks/react/src/lib/ai-sdk/utils/toUIMessage.test.ts

@@ -437,7 +437,7 @@ describe('toUIMessage', () => {
     it('should create a new assistant message for tripwire chunk', () => {
       const chunk: ChunkType = {
         type: 'tripwire',
-        payload: { tripwireReason: 'Security warning detected' },
+        payload: { reason: 'Security warning detected' },
         runId: 'run-123',
         from: ChunkFrom.AGENT,
       };
@@ -456,11 +456,53 @@ describe('toUIMessage', () => {
         ],
         metadata: {
           ...baseMetadata,
-          status: 'warning',
+          status: 'tripwire',
+          tripwire: {
+            retry: undefined,
+            tripwirePayload: undefined,
+            processorId: undefined,
+          },
         },
       });
       expect(result[0].id).toMatch(/^tripwire-run-123/);
     });
+
+    it('should include tripwire metadata when provided', () => {
+      const chunk: ChunkType = {
+        type: 'tripwire',
+        payload: {
+          reason: 'PII detected in message',
+          retry: false,
+          metadata: { detectedPII: ['email', 'phone'], severity: 'high' },
+          processorId: 'pii-detection',
+        },
+        runId: 'run-456',
+        from: ChunkFrom.AGENT,
+      };
+
+      const conversation: MastraUIMessage[] = [];
+      const result = toUIMessage({ chunk, conversation, metadata: baseMetadata });
+
+      expect(result).toHaveLength(1);
+      expect(result[0]).toMatchObject({
+        role: 'assistant',
+        parts: [
+          {
+            type: 'text',
+            text: 'PII detected in message',
+          },
+        ],
+        metadata: {
+          ...baseMetadata,
+          status: 'tripwire',
+          tripwire: {
+            retry: false,
+            tripwirePayload: { detectedPII: ['email', 'phone'], severity: 'high' },
+            processorId: 'pii-detection',
+          },
+        },
+      });
+    });
   });
 
   describe('toUIMessage - start chunk', () => {

client-sdks/react/src/lib/ai-sdk/utils/toUIMessage.ts

@@ -42,7 +42,9 @@ export const mapWorkflowStreamChunkToWatchResult = (
         ? { result: lastStep?.output }
         : finalStatus === 'failed' && lastStep?.status === 'failed'
           ? { error: lastStep?.error }
-          : {}),
+          : finalStatus === 'tripwire' && chunk.payload.tripwire
+            ? { tripwire: chunk.payload.tripwire }
+            : {}),
     };
   }
 
@@ -146,19 +148,24 @@ export const toUIMessage = ({ chunk, conversation, metadata }: ToUIMessageArgs):
 
   switch (chunk.type) {
     case 'tripwire': {
-      // Create a new assistant message
+      // Create a new assistant message with tripwire-specific metadata
       const newMessage: MastraUIMessage = {
         id: `tripwire-${chunk.runId + Date.now()}`,
         role: 'assistant',
         parts: [
           {
             type: 'text',
-            text: chunk.payload.tripwireReason,
+            text: chunk.payload.reason,
           },
         ],
         metadata: {
           ...metadata,
-          status: 'warning',
+          status: 'tripwire',
+          tripwire: {
+            retry: chunk.payload.retry,
+            tripwirePayload: chunk.payload.metadata,
+            processorId: chunk.payload.processorId,
+          },
         },
       };