{"record":{"id":"56cc2fc6216461c4","repo":"mastra-ai/mastra","slug":"stream-part-blocked-by-processor-id","errorCode":null,"errorMessage":"Stream part blocked by ${processor.id}","messagePattern":"Stream part blocked by (.+?)","errorType":"exception","errorClass":"TripWire","httpStatus":null,"severity":"warning","filePath":"packages/core/src/processors/runner.ts","lineNumber":906,"sourceCode":"              state = new ProcessorState<OUTPUT>({\n                processorName: processor.name ?? processor.id,\n                ...observabilityContext,\n                processorIndex: index,\n                createSpan: true,\n              });\n              processorStates.set(processor.id, state);\n            }\n\n            // Track input chunk (before processor transformation)\n            state.addInputPart(processedPart);\n\n            const result = await processor.processOutputStream({\n              part: processedPart as ChunkType,\n              streamParts: state.streamParts as ChunkType[],\n              state: state.customState,\n              agent: this.agent,\n              abort: <TMetadata = unknown>(reason?: string, options?: TripWireOptions<TMetadata>): never => {\n                throw new TripWire(reason || `Stream part blocked by ${processor.id}`, options, processor.id);\n              },\n              ...createObservabilityContext({ currentSpan: state.span }),\n              requestContext,\n              messageList,\n              retryCount,\n              writer,\n            });\n\n            // Track output chunk and update processedPart\n            processedPart = result as ChunkType<OUTPUT> | null | undefined;\n            state.addOutputPart(processedPart);\n          }\n        } catch (error) {\n          if (error instanceof TripWire) {\n            // Error span for trip-wire abort so it shows as ERROR in traces\n            const state = processorStates.get(processor.id);\n            state?.span?.error({\n              error,","sourceCodeStart":888,"sourceCodeEnd":924,"githubUrl":"https://github.com/mastra-ai/mastra/blob/75dd419e613fe9c39f846ffc500716141b74fda6/packages/core/src/processors/runner.ts#L888-L924","documentation":"This TripWire is thrown when a processor's processOutputStream calls the abort() helper for an individual stream chunk. It blocks that stream part (and can trip the whole stream depending on tripwireOptions). The generic message appears because the processor called abort() without a custom reason.","triggerScenarios":"A stream processor inspecting each ChunkType via processOutputStream calls abort() when a chunk violates policy (e.g. streaming output containing banned words), without providing a reason string.","commonSituations":"Token-level moderation during streaming; redaction processors aborting mid-stream; debugging blocked streams where the generic message obscures which check fired.","solutions":["Inspect the thrown TripWire's processorId to find the offending processor.","Pass a specific reason to abort(), e.g. abort('Blocked token stream: policy hit').","Adjust or remove the stream processor if blocking is not intended; check its tripwireOptions for blocking scope.","Catch TripWire in stream error handling if mid-stream aborts are expected."],"exampleFix":"// before\nprocessOutputStream: async ({ part, abort }) => {\n  if (isBad(part)) abort();\n  return part;\n}\n// after\nprocessOutputStream: async ({ part, abort }) => {\n  if (isBad(part)) abort('Stream chunk blocked by profanity-filter', { suppress: true });\n  return part;\n}","handlingStrategy":"try-catch","validationCode":"// Review stream processors for abort() without reason before deploying:\nconst audit = (p) => p.processOutputStream?.toString().includes('abort()') ? warn(p) : ok(p);","typeGuard":"function isTripWire(e: unknown): e is TripWire {\n  return e instanceof TripWire;\n}","tryCatchPattern":"try {\n  for await (const chunk of stream) handle(chunk);\n} catch (e) {\n  if (isTripWire(e)) {\n    console.warn('Stream part blocked by processor:', e.processorId, e.message);\n  } else throw e;\n}","preventionTips":["Pass a reason to abort() in processOutputStream","Decide blocking scope via TripWireOptions explicitly","Test stream processors with chunk-level fixtures","Monitor blocked-stream telemetry keyed by processorId"],"tags":["agent","processors","streaming","tripwire"],"backgroundTag":"processor-tripwire-aborted","analyzedSha":"75dd419e613fe9c39f846ffc500716141b74fda6","analyzedAt":"2026-08-30T00:15:31.844Z","schemaVersion":2},"datasetVersion":"2026-08-30T03:17:51.788Z"}