feat(inngest): forward FGA actor signal through execution engine

The @mastra/inngest execution engine now accepts a trusted `actor`
(ActorSignal) on its run/start path and re-threads it across durable
step and nested-workflow boundaries, so trusted background workflows on
the Inngest engine get the same FGA membership bypass the default engine
already has.

- run.ts: accept `actor` on start/startAsync/resume/resumeAsync/stream/
  streamLegacy/timeTravel and include it in each Inngest event payload.
- workflow.ts: read `actor` from event.data in the function handler and
  pass it to engine.execute().
- execution-engine.ts: forward `actor` into every nested
  inngestStep.invoke() so nested workflows re-thread it across the
  serialization boundary.

Mirrors core PR mastra-ai#17487 (default engine) for the durable runtime.
Refs mastra-ai#17216, mastra-ai#17487.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tiffanybalc

.changeset/early-spoons-joke.md

@@ -0,0 +1,18 @@
+---
+'@mastra/inngest': minor
+---
+
+Added support for the fine-grained authorization (FGA) `actor` signal on the Inngest execution engine.
+
+Workflows running on the Inngest engine can now pass a trusted `actor` through `run.start()`, `startAsync()`, `resume()`, `stream()`, and `timeTravel()`. The signal is re-threaded across durable step and nested-workflow boundaries, so every nested agent, tool, and memory FGA check sees the same actor. Previously `actor` was only threaded through the default engine, so trusted background workflows on Inngest lost the membership bypass at each step re-entry.
+
+**Usage**
+
+```ts
+const run = await workflow.createRun();
+await run.start({
+  inputData,
+  requestContext, // includes organizationId / tenant scope
+  actor: { actorKind: 'system', sourceWorkflow: 'nightly-sync' },
+});
+```

workflows/inngest/src/actor-signal.test.ts

@@ -0,0 +1,177 @@
+import type { ActorSignal } from '@mastra/core/auth/ee';
+import { Mastra } from '@mastra/core/mastra';
+import { MockStore } from '@mastra/core/storage';
+import { Inngest } from 'inngest';
+import { describe, expect, it, vi } from 'vitest';
+import { z } from 'zod';
+
+import { InngestExecutionEngine } from './execution-engine';
+import { init } from './index';
+
+/**
+ * Hermetic tests for FGA `actor` signal threading through @mastra/inngest.
+ *
+ * These do NOT require an Inngest dev server. They exercise the two seams the
+ * engine owns directly:
+ *   1. The run/start path serializes `actor` into the Inngest event payload.
+ *   2. `executeWorkflowStep` forwards `actor` across the nested-workflow
+ *      `step.invoke()` serialization boundary so the nested run re-threads it.
+ */
+describe('@mastra/inngest actor signal threading (hermetic)', () => {
+  const actor: ActorSignal = { actorKind: 'system', sourceWorkflow: 'nightly-workflow' };
+
+  it('forwards actor into the nested-workflow invoke payload (durable step boundary)', async () => {
+    const inngest = new Inngest({ id: 'mastra-test' });
+    const { createWorkflow, createStep } = init(inngest);
+
+    const nestedStep = createStep({
+      id: 'nested-step',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      execute: async ({ inputData }) => inputData,
+    });
+
+    const nestedWorkflow = createWorkflow({
+      id: 'nested-workflow',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      steps: [nestedStep],
+    })
+      .then(nestedStep)
+      .commit();
+
+    // Capture the data handed to inngestStep.invoke (the serialization boundary).
+    const invokeData: any[] = [];
+    const fakeStep: any = {
+      run: async (_id: string, fn: () => Promise<any>) => fn(),
+      invoke: async (_id: string, opts: { function: any; data: any }) => {
+        invokeData.push(opts.data);
+        return { result: { status: 'success', result: { value: 'ok' }, state: {} }, runId: 'nested-run' };
+      },
+      sleep: async () => {},
+      sleepUntil: async () => {},
+    };
+
+    const engine = new InngestExecutionEngine({} as Mastra, fakeStep, 0, {});
+    const pubsub: any = { publish: vi.fn().mockResolvedValue(undefined) };
+
+    const result = await engine.executeWorkflowStep({
+      step: nestedWorkflow as any,
+      stepResults: {},
+      executionContext: {
+        workflowId: 'parent-workflow',
+        runId: 'parent-run',
+        executionPath: [0],
+        suspendedPaths: {},
+        state: {},
+      } as any,
+      prevOutput: {},
+      inputData: { value: 'ok' },
+      pubsub,
+      startedAt: Date.now(),
+      actor,
+    } as any);
+
+    expect(result?.status).toBe('success');
+    expect(invokeData).toHaveLength(1);
+    expect(invokeData[0].actor).toEqual(actor);
+  });
+
+  it('serializes actor into the Inngest event payload on the start path', async () => {
+    const inngest = new Inngest({ id: 'mastra-test' });
+    const { createWorkflow, createStep } = init(inngest);
+
+    const step = createStep({
+      id: 'step',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      execute: async ({ inputData }) => inputData,
+    });
+
+    const workflow = createWorkflow({
+      id: 'actor-start-workflow',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      steps: [step],
+    })
+      .then(step)
+      .commit();
+
+    const mastra = new Mastra({
+      logger: false,
+      storage: new MockStore(),
+      workflows: { 'actor-start-workflow': workflow as any },
+    });
+    workflow.__registerMastra(mastra);
+
+    const sendSpy = vi.spyOn(inngest, 'send').mockResolvedValue({ ids: ['evt-1'] } as any);
+
+    const run = await workflow.createRun();
+    await run.startAsync({ inputData: { value: 'ok' }, actor });
+
+    expect(sendSpy).toHaveBeenCalledTimes(1);
+    const sentData = (sendSpy.mock.calls[0]![0] as any).data;
+    expect(sentData.actor).toEqual(actor);
+  });
+
+  it('forwards a re-supplied actor through the resume event payload (per-call contract)', async () => {
+    // `actor` is intentionally NOT rehydrated from the snapshot (matching the default
+    // engine — see packages/core/src/workflows/workflow.ts `_resume`). A trusted resumer
+    // re-supplies it on each resume; this locks in that per-call contract.
+    const inngest = new Inngest({ id: 'mastra-test' });
+    const { createWorkflow, createStep } = init(inngest);
+
+    const step = createStep({
+      id: 'suspendable-step',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      execute: async ({ inputData }) => inputData,
+    });
+
+    const workflow = createWorkflow({
+      id: 'actor-resume-workflow',
+      inputSchema: z.object({ value: z.string() }),
+      outputSchema: z.object({ value: z.string() }),
+      steps: [step],
+    })
+      .then(step)
+      .commit();
+
+    const mastra = new Mastra({
+      logger: false,
+      storage: new MockStore(),
+      workflows: { 'actor-resume-workflow': workflow as any },
+    });
+    workflow.__registerMastra(mastra);
+
+    const run = await workflow.createRun();
+
+    // Persist a suspended snapshot so the resume path has something to load.
+    const workflowsStore = await mastra.getStorage()!.getStore('workflows');
+    await workflowsStore!.persistWorkflowSnapshot({
+      workflowName: 'actor-resume-workflow',
+      runId: run.runId,
+      snapshot: {
+        runId: run.runId,
+        serializedStepGraph: [],
+        status: 'suspended',
+        value: {},
+        context: { input: { value: 'ok' } },
+        activePaths: [],
+        suspendedPaths: { 'suspendable-step': [0] },
+        activeStepsPath: {},
+        resumeLabels: {},
+        waitingPaths: {},
+        timestamp: Date.now(),
+      } as any,
+    });
+
+    const sendSpy = vi.spyOn(inngest, 'send').mockResolvedValue({ ids: ['evt-resume'] } as any);
+
+    await run.resumeAsync({ resumeData: { value: 'ok' }, step: 'suspendable-step', actor });
+
+    expect(sendSpy).toHaveBeenCalledTimes(1);
+    const sentData = (sendSpy.mock.calls[0]![0] as any).data;
+    expect(sentData.actor).toEqual(actor);
+  });
+});

workflows/inngest/src/execution-engine.ts

@@ -1,4 +1,5 @@
 import { randomUUID } from 'node:crypto';
+import type { ActorSignal } from '@mastra/core/auth/ee';
 import type { RequestContext } from '@mastra/core/di';
 import { getErrorFromUnknown } from '@mastra/core/error';
 import type { SerializedError } from '@mastra/core/error';
@@ -424,6 +425,7 @@ export class InngestExecutionEngine extends DefaultExecutionEngine {
     startedAt: number;
     perStep?: boolean;
     stepSpan?: any;
+    actor?: ActorSignal;
   }): Promise<StepResult<any, any, any, any> | null> {
     // Only handle InngestWorkflow instances
     if (!(params.step instanceof InngestWorkflow)) {
@@ -442,6 +444,7 @@ export class InngestExecutionEngine extends DefaultExecutionEngine {
       startedAt,
       perStep,
       stepSpan,
+      actor,
     } = params;
 
     // Build trace context to propagate to nested workflow
@@ -483,6 +486,7 @@ export class InngestExecutionEngine extends DefaultExecutionEngine {
             outputOptions: { includeState: true },
             perStep,
             tracingOptions: nestedTracingContext,
+            actor,
           },
         })) as any;
         result = invokeResp.result;
@@ -512,6 +516,7 @@ export class InngestExecutionEngine extends DefaultExecutionEngine {
             outputOptions: { includeState: true },
             perStep,
             tracingOptions: nestedTracingContext,
+            actor,
           },
         })) as any;
         result = invokeResp.result;
@@ -526,6 +531,7 @@ export class InngestExecutionEngine extends DefaultExecutionEngine {
             outputOptions: { includeState: true },
             perStep,
             tracingOptions: nestedTracingContext,
+            actor,
           },
         })) as any;
         result = invokeResp.result;

workflows/inngest/src/index.test.ts

@@ -237,6 +237,119 @@ describe('MastraInngestWorkflow', () => {
     }
   });
 
+  describe.sequential('FGA actor signal', () => {
+    it('bypasses membership resolution for a trusted system actor across a nested-workflow step boundary', async ctx => {
+      const inngest = new Inngest({
+        id: 'mastra',
+        baseUrl: `http://localhost:${(ctx as any).inngestPort}`,
+      });
+
+      const { createWorkflow, createStep } = init(inngest);
+
+      const fgaProvider = {
+        require: vi.fn().mockResolvedValue(undefined),
+        check: vi.fn(),
+        filterAccessible: vi.fn(),
+      };
+
+      const agent = new Agent({
+        id: 'membership-agent',
+        name: 'Membership Agent',
+        instructions: 'Say ok',
+        model: new MockLanguageModelV2({
+          doGenerate: async () => ({
+            rawCall: { rawPrompt: null, rawSettings: {} },
+            finishReason: 'stop',
+            usage: { inputTokens: 1, outputTokens: 1, totalTokens: 2 },
+            content: [{ type: 'text', text: 'ok' }],
+            warnings: [],
+          }),
+        }),
+      });
+
+      // Inside a durable step, forward the per-call actor + tenant-scoped requestContext
+      // to a nested agent FGA check, exactly as a trusted background workflow would.
+      const callAgentStep = createStep({
+        id: 'call-agent',
+        inputSchema: z.object({}),
+        outputSchema: z.object({ text: z.string() }),
+        execute: async ({ actor, requestContext, mastra }) => {
+          const res = await mastra!.getAgent('membership-agent').generate('hello', { actor, requestContext });
+          return { text: res.text };
+        },
+      });
+
+      const nestedWorkflow = createWorkflow({
+        id: 'nested-actor-workflow',
+        inputSchema: z.object({}),
+        outputSchema: z.object({ text: z.string() }),
+        steps: [callAgentStep],
+      })
+        .then(callAgentStep)
+        .commit();
+
+      const workflow = createWorkflow({
+        id: 'actor-parent-workflow',
+        inputSchema: z.object({}),
+        outputSchema: z.object({ text: z.string() }),
+        steps: [nestedWorkflow],
+      })
+        .then(nestedWorkflow)
+        .commit();
+
+      const mastra = new Mastra({
+        logger: false,
+        storage: new DefaultStorage({
+          id: 'test-storage',
+          url: ':memory:',
+        }),
+        agents: { 'membership-agent': agent },
+        workflows: {
+          'actor-parent-workflow': workflow,
+        },
+        server: {
+          fga: fgaProvider,
+          apiRoutes: [
+            {
+              path: '/inngest/api',
+              method: 'ALL',
+              createHandler: async ({ mastra }) => inngestServe({ mastra, inngest, ...getDockerRegisterOptions() }),
+            },
+          ],
+        },
+      });
+
+      const app = await createHonoServer(mastra);
+
+      const srv = (globServer = serve({
+        fetch: app.fetch,
+        port: (ctx as any).handlerPort,
+      }));
+      await resetInngest();
+
+      const requestContext = new RequestContext();
+      requestContext.set('organizationId', 'org-1');
+
+      const run = await workflow.createRun();
+      const result = await run.start({
+        inputData: {},
+        requestContext,
+        actor: { actorKind: 'system', sourceWorkflow: 'nightly-workflow' },
+      });
+
+      srv.close();
+
+      // The trusted actor bypasses membership resolution at the nested agent FGA check,
+      // which only happens if `actor` survived the parent -> nested-workflow step boundary.
+      expect(fgaProvider.require).not.toHaveBeenCalled();
+      expect(result.status).toBe('success');
+      expect(result.steps['nested-actor-workflow']).toMatchObject({
+        status: 'success',
+        output: { text: 'ok' },
+      });
+    });
+  });
+
   describe.sequential('Basic Workflow Execution', () => {
     it('should be able to bail workflow execution', async ctx => {
       const t0 = Date.now();