@@ -7,6 +7,7 @@ import type { JSONSchema7 } from 'json-schema';
77import { z } from 'zod/v4' ;
88import type { MastraPrimitives , MastraUnion } from '../action' ;
99import { MastraFGAPermissions } from '../auth/ee' ;
10+ import type { ActorSignal } from '../auth/ee' ;
1011import type { AgentBackgroundConfig , ToolBackgroundConfig } from '../background-tasks' ;
1112import { MastraBase } from '../base' ;
1213import type { MastraBrowser } from '../browser/browser' ;
@@ -191,6 +192,10 @@ type AgentSnapshotMemoryInfo = {
191192 resourceId ?: string ;
192193} ;
193194
195+ function getInvocationActor ( context : unknown ) : ActorSignal | undefined {
196+ return ( context as { actor ?: ActorSignal } | undefined ) ?. actor ;
197+ }
198+
194199type ProcessorWorkflowChildrenContainer = {
195200 steps ?: Record < string , unknown > | unknown [ ] ;
196201 children ?: Record < string , unknown > | unknown [ ] ;
@@ -4130,6 +4135,7 @@ export class Agent<
41304135 // manually wrap agent tools with tracing, so that we can pass the
41314136 // current tool span onto the agent to maintain continuity of the trace
41324137 execute : async ( inputData : z . infer < typeof agentInputSchema > , context ) => {
4138+ const invocationActor = getInvocationActor ( context ) ;
41334139 const startTime = Date . now ( ) ;
41344140 const toolCallId = context ?. agent ?. toolCallId || randomUUID ( ) ;
41354141
@@ -4444,6 +4450,7 @@ export class Agent<
44444450 ? await resolvedAgent . resumeGenerate ( resumeData , {
44454451 runId : suspendedToolRunId ,
44464452 requestContext,
4453+ actor : invocationActor ,
44474454 ...resolveObservabilityContext ( context ?? { } ) ,
44484455 ...( effectiveInstructions && { instructions : effectiveInstructions } ) ,
44494456 ...( effectiveMaxSteps && { maxSteps : effectiveMaxSteps } ) ,
@@ -4461,6 +4468,7 @@ export class Agent<
44614468 } )
44624469 : await resolvedAgent . generate ( messagesForSubAgent , {
44634470 requestContext,
4471+ actor : invocationActor ,
44644472 ...resolveObservabilityContext ( context ?? { } ) ,
44654473 ...( effectiveInstructions && { instructions : effectiveInstructions } ) ,
44664474 ...( effectiveMaxSteps && { maxSteps : effectiveMaxSteps } ) ,
@@ -4556,6 +4564,7 @@ export class Agent<
45564564 }
45574565 const generateResult = await resolvedAgent . generateLegacy ( messagesForSubAgent , {
45584566 requestContext,
4567+ actor : invocationActor ,
45594568 ...resolveObservabilityContext ( context ?? { } ) ,
45604569 context : filteredContextMessages as unknown as CoreMessage [ ] ,
45614570 } ) ;
@@ -4571,6 +4580,7 @@ export class Agent<
45714580 ? await resolvedAgent . resumeStream ( resumeData , {
45724581 runId : suspendedToolRunId ,
45734582 requestContext,
4583+ actor : invocationActor ,
45744584 ...resolveObservabilityContext ( context ?? { } ) ,
45754585 ...( effectiveInstructions && { instructions : effectiveInstructions } ) ,
45764586 ...( effectiveMaxSteps && { maxSteps : effectiveMaxSteps } ) ,
@@ -4590,6 +4600,7 @@ export class Agent<
45904600 } )
45914601 : await resolvedAgent . stream ( messagesForSubAgent , {
45924602 requestContext,
4603+ actor : invocationActor ,
45934604 ...resolveObservabilityContext ( context ?? { } ) ,
45944605 ...( effectiveInstructions && { instructions : effectiveInstructions } ) ,
45954606 ...( effectiveMaxSteps && { maxSteps : effectiveMaxSteps } ) ,
@@ -4722,6 +4733,7 @@ export class Agent<
47224733 }
47234734 const streamResult = await resolvedAgent . streamLegacy ( effectivePrompt , {
47244735 requestContext,
4736+ actor : invocationActor ,
47254737 ...resolveObservabilityContext ( context ?? { } ) ,
47264738 } ) ;
47274739
@@ -5051,6 +5063,7 @@ export class Agent<
50515063 // manually wrap workflow tools with tracing, so that we can pass the
50525064 // current tool span onto the workflow to maintain continuity of the trace
50535065 execute : async ( inputData , context ) => {
5066+ const invocationActor = getInvocationActor ( context ) ;
50545067 const savedMastraMemory = requestContext . get ( 'MastraMemory' ) ;
50555068 try {
50565069 const { initialState, inputData : workflowInputData , suspendedToolRunId } = inputData as any ;
@@ -5080,12 +5093,14 @@ export class Agent<
50805093 result = await run . resume ( {
50815094 resumeData,
50825095 requestContext,
5096+ actor : invocationActor ,
50835097 ...resolveObservabilityContext ( context ?? { } ) ,
50845098 } ) ;
50855099 } else {
50865100 result = await run . start ( {
50875101 inputData : workflowInputData ,
50885102 requestContext,
5103+ actor : invocationActor ,
50895104 ...resolveObservabilityContext ( context ?? { } ) ,
50905105 ...( initialState && { initialState } ) ,
50915106 } ) ;
@@ -5094,6 +5109,7 @@ export class Agent<
50945109 const streamResult = run . streamLegacy ( {
50955110 inputData : workflowInputData ,
50965111 requestContext,
5112+ actor : invocationActor ,
50975113 ...resolveObservabilityContext ( context ?? { } ) ,
50985114 } ) ;
50995115
@@ -5111,11 +5127,13 @@ export class Agent<
51115127 ? run . resumeStream ( {
51125128 resumeData,
51135129 requestContext,
5130+ actor : invocationActor ,
51145131 ...resolveObservabilityContext ( context ?? { } ) ,
51155132 } )
51165133 : run . stream ( {
51175134 inputData : workflowInputData ,
51185135 requestContext,
5136+ actor : invocationActor ,
51195137 ...resolveObservabilityContext ( context ?? { } ) ,
51205138 ...( initialState && { initialState } ) ,
51215139 } ) ;
@@ -5881,11 +5899,13 @@ export class Agent<
58815899 memory,
58825900 runId,
58835901 snapshotMemoryInfo,
5902+ actor,
58845903 } : {
58855904 requestContext ?: RequestContext ;
58865905 memory ?: AgentExecutionOptionsBase < any > [ 'memory' ] ;
58875906 runId ?: string ;
58885907 snapshotMemoryInfo ?: AgentSnapshotMemoryInfo ;
5908+ actor ?: ActorSignal ;
58895909 } ) : Promise < void > {
58905910 const fgaProvider = this . #mastra?. getServer ( ) ?. fga ;
58915911 if ( ! fgaProvider ) {
@@ -5901,6 +5921,7 @@ export class Agent<
59015921 resource : { type : 'agent' , id : getAgentFGAResourceId ( this . id ) } ,
59025922 permission : MastraFGAPermissions . AGENTS_EXECUTE ,
59035923 requestContext,
5924+ actor,
59045925 context : {
59055926 resourceId : executionResourceId ,
59065927 } ,
@@ -6316,7 +6337,7 @@ export class Agent<
63166337 const observabilityContext = createObservabilityContext ( { currentSpan : agentSpan } ) ;
63176338 try {
63186339 const run = await executionWorkflow . createRun ( { runId : executionRunId } ) ;
6319- const result = await run . start ( { ...observabilityContext } ) ;
6340+ const result = await run . start ( { requestContext , actor : options . actor , ...observabilityContext } ) ;
63206341 return result ;
63216342 } finally {
63226343 if ( useEventedExecution ) {
@@ -6783,8 +6804,16 @@ export class Agent<
67836804 defaultOptions as Record < string , unknown > ,
67846805 ( options ?? { } ) as Record < string , unknown > ,
67856806 ) as AgentExecutionOptions < any > & { model ?: DynamicArgument < MastraModelConfig > } ;
6807+ const loopOptions = { ...mergedOptions } ;
6808+ const actor = mergedOptions . actor ;
6809+ delete loopOptions . actor ;
67866810
6787- await this . #requireAgentExecutionFGA( mergedOptions ) ;
6811+ await this . #requireAgentExecutionFGA( {
6812+ requestContext : mergedOptions . requestContext ,
6813+ memory : mergedOptions . memory ,
6814+ runId : mergedOptions . runId ,
6815+ actor,
6816+ } ) ;
67886817
67896818 const llm = await this . getLLM ( {
67906819 requestContext : mergedOptions . requestContext ,
@@ -6816,7 +6845,8 @@ export class Agent<
68166845 }
68176846
68186847 const executeOptions = {
6819- ...mergedOptions ,
6848+ ...loopOptions ,
6849+ actor,
68206850 structuredOutput : mergedOptions . structuredOutput
68216851 ? {
68226852 ...mergedOptions . structuredOutput ,
@@ -7145,6 +7175,9 @@ export class Agent<
71457175 defaultOptions as Record < string , unknown > ,
71467176 ( streamOptions ?? { } ) as Record < string , unknown > ,
71477177 ) as AgentExecutionOptions < OUTPUT > & { model ?: DynamicArgument < MastraModelConfig > } ;
7178+ const loopOptions = { ...mergedOptions } ;
7179+ const actor = mergedOptions . actor ;
7180+ delete loopOptions . actor ;
71487181
71497182 // Delegate to the idle-loop wrapper when `untilIdle` is set (from
71507183 // per-call options OR defaultOptions). Strip `untilIdle` before passing
@@ -7163,7 +7196,12 @@ export class Agent<
71637196 ) ;
71647197 }
71657198
7166- await this . #requireAgentExecutionFGA( mergedOptions ) ;
7199+ await this . #requireAgentExecutionFGA( {
7200+ requestContext : mergedOptions . requestContext ,
7201+ memory : mergedOptions . memory ,
7202+ runId : mergedOptions . runId ,
7203+ actor,
7204+ } ) ;
71677205
71687206 const llm = await this . getLLM ( {
71697207 requestContext : mergedOptions . requestContext ,
@@ -7197,7 +7235,7 @@ export class Agent<
71977235 const threadStreamPubSub = this . getPubSub ( ) ;
71987236 await agentThreadStreamRuntime . waitForCrossAgentThreadRun (
71997237 this as Agent < any , any , any , any > ,
7200- mergedOptions ,
7238+ loopOptions as AgentExecutionOptions < OUTPUT > ,
72017239 threadStreamPubSub ,
72027240 ) ;
72037241
@@ -7207,10 +7245,14 @@ export class Agent<
72077245 source : 'agent' ,
72087246 entityId : this . id ,
72097247 } ) ?? randomUUID ( ) ;
7210- const preparedOptions = agentThreadStreamRuntime . prepareRunOptions ( mergedOptions , threadStreamPubSub ) ;
7248+ const preparedOptions = agentThreadStreamRuntime . prepareRunOptions (
7249+ { ...loopOptions , runId : mergedOptions . runId , actor } as AgentExecutionOptions < OUTPUT > ,
7250+ threadStreamPubSub ,
7251+ ) ;
72117252
72127253 const executeOptions = {
72137254 ...preparedOptions ,
7255+ actor,
72147256 structuredOutput : mergedOptions . structuredOutput
72157257 ? {
72167258 ...mergedOptions . structuredOutput ,
@@ -7456,6 +7498,9 @@ export class Agent<
74567498 defaultOptions as Record < string , unknown > ,
74577499 ( streamOptions ?? { } ) as Record < string , unknown > ,
74587500 ) as typeof defaultOptions & { model ?: DynamicArgument < MastraModelConfig > } ;
7501+ const loopStreamOptions = { ...mergedStreamOptions } ;
7502+ const actor = mergedStreamOptions . actor ;
7503+ delete loopStreamOptions . actor ;
74597504
74607505 // Delegate to the idle-loop wrapper when `untilIdle` is set (from
74617506 // per-call options OR defaultOptions). Strip `untilIdle` before passing
@@ -7484,13 +7529,15 @@ export class Agent<
74847529 thread : mergedStreamOptions . memory ?. thread ?? snapshotMemoryInfo . threadId ,
74857530 resource : mergedStreamOptions . memory ?. resource ?? snapshotMemoryInfo . resourceId ,
74867531 } ;
7532+ loopStreamOptions . memory = mergedStreamOptions . memory ;
74877533 }
74887534
74897535 await this . #requireAgentExecutionFGA( {
74907536 requestContext : mergedStreamOptions . requestContext ,
74917537 memory : mergedStreamOptions . memory ,
74927538 runId : mergedStreamOptions . runId ,
74937539 snapshotMemoryInfo,
7540+ actor,
74947541 } ) ;
74957542
74967543 const llm = await this . getLLM ( {
@@ -7520,16 +7567,17 @@ export class Agent<
75207567 const threadStreamPubSub = this . getPubSub ( ) ;
75217568 await agentThreadStreamRuntime . waitForCrossAgentThreadRun (
75227569 this as Agent < any , any , any , any > ,
7523- mergedStreamOptions as unknown as AgentExecutionOptions < OUTPUT > ,
7570+ loopStreamOptions as unknown as AgentExecutionOptions < OUTPUT > ,
75247571 threadStreamPubSub ,
75257572 ) ;
75267573 const preparedOptions = agentThreadStreamRuntime . prepareRunOptions (
7527- mergedStreamOptions as unknown as AgentExecutionOptions < OUTPUT > ,
7574+ { ... loopStreamOptions , actor } as unknown as AgentExecutionOptions < OUTPUT > ,
75287575 threadStreamPubSub ,
75297576 ) ;
75307577
75317578 const result = await this . #execute( {
75327579 ...preparedOptions ,
7580+ actor,
75337581 structuredOutput : mergedStreamOptions . structuredOutput
75347582 ? {
75357583 ...mergedStreamOptions . structuredOutput ,
@@ -7629,6 +7677,9 @@ export class Agent<
76297677 defaultOptions as Record < string , unknown > ,
76307678 ( options ?? { } ) as Record < string , unknown > ,
76317679 ) as typeof defaultOptions & { model ?: DynamicArgument < MastraModelConfig > } ;
7680+ const loopOptions = { ...mergedOptions } ;
7681+ const actor = mergedOptions . actor ;
7682+ delete loopOptions . actor ;
76327683
76337684 const runId = options ?. runId ?? '' ;
76347685 const existingSnapshot = await this . #loadAgenticLoopSnapshotOrThrow( { runId, method : 'resumeGenerate' } ) ;
@@ -7637,6 +7688,7 @@ export class Agent<
76377688 memory : mergedOptions . memory ,
76387689 runId : mergedOptions . runId ,
76397690 snapshotMemoryInfo : this . #getSnapshotMemoryInfo( existingSnapshot ) ,
7691+ actor,
76407692 } ) ;
76417693
76427694 const llm = await this . getLLM ( {
@@ -7668,7 +7720,8 @@ export class Agent<
76687720 }
76697721
76707722 const result = await this . #execute( {
7671- ...mergedOptions ,
7723+ ...loopOptions ,
7724+ actor,
76727725 structuredOutput : mergedOptions . structuredOutput
76737726 ? {
76747727 ...mergedOptions . structuredOutput ,
0 commit comments