mirror of
https://github.com/simstudioai/sim.git
synced 2026-09-24 15:45:35 +08:00
feat(start-block): add run metadata toggle with trusted <start.metadata> outputs (#5700)
* feat(start-block): add run metadata toggle with trusted <start.metadata> outputs * fix(start-block): fail-soft email lookup, consistent nested metadata propagation, toggle parsing parity * improvement(start-block): enumerate metadata fields in toggle description for agent schema * fix(start-block): carry metadata chain through toggle-off children, preserve fail-soft null email * fix(start-block): recover metadata chain from seeded start output after resume
This commit is contained in:
@@ -25,6 +25,15 @@ export const StartTriggerBlock: BlockConfig = {
|
||||
type: 'input-format',
|
||||
description: 'Add custom fields beyond the built-in input, conversationId, and files fields.',
|
||||
},
|
||||
{
|
||||
id: 'runMetadata',
|
||||
title: 'Add run metadata',
|
||||
type: 'switch',
|
||||
mode: 'advanced',
|
||||
defaultValue: false,
|
||||
description:
|
||||
'Expose trusted, server-injected run metadata under <start.metadata>: userEmail, workspaceId, workflowId, executionId, executionType, executionMode, startTime. Fields describe the invoking run — inside a custom block they identify the calling user and workflow.',
|
||||
},
|
||||
],
|
||||
tools: {
|
||||
access: [],
|
||||
|
||||
@@ -436,6 +436,7 @@ export class DAGExecutor {
|
||||
this.contextExtensions.metadata?.useDraftState ??
|
||||
this.contextExtensions.isDeployedContext !== true,
|
||||
},
|
||||
startRunMetadata: this.contextExtensions.startRunMetadata,
|
||||
environmentVariables: this.environmentVariables,
|
||||
workflowVariables: this.workflowVariables,
|
||||
decisions: {
|
||||
@@ -620,6 +621,7 @@ export class DAGExecutor {
|
||||
const blockOutput = buildStartBlockOutput({
|
||||
resolution: startResolution,
|
||||
workflowInput: this.workflowInput,
|
||||
runMetadata: this.contextExtensions.startRunMetadata,
|
||||
})
|
||||
|
||||
state.setBlockState(startResolution.block.id, {
|
||||
|
||||
@@ -6,6 +6,7 @@ import type {
|
||||
BlockLog,
|
||||
BlockState,
|
||||
NormalizedBlockOutput,
|
||||
StartBlockRunMetadata,
|
||||
StreamingExecution,
|
||||
} from '@/executor/types'
|
||||
import type { RunFromBlockContext } from '@/executor/utils/run-from-block'
|
||||
@@ -192,6 +193,12 @@ export interface ContextExtensions {
|
||||
dagIncomingEdges?: Record<string, string[]>
|
||||
snapshotState?: SerializableExecutionState
|
||||
metadata?: ExecutionMetadata
|
||||
/**
|
||||
* Trusted run metadata injected into the Start block output when its
|
||||
* "Add run metadata" toggle is enabled. Built server-side at the two
|
||||
* Executor construction sites — never from caller-supplied input.
|
||||
*/
|
||||
startRunMetadata?: StartBlockRunMetadata
|
||||
/**
|
||||
* AbortSignal for cancellation support.
|
||||
* When aborted, the execution should stop gracefully.
|
||||
|
||||
@@ -15,6 +15,7 @@ const {
|
||||
mockResolveBillingAttribution,
|
||||
mockGetCustomBlockAuthority,
|
||||
mockGetPersonalAndWorkspaceEnv,
|
||||
mockGetUserEmailById,
|
||||
executorOptions,
|
||||
} = vi.hoisted(() => ({
|
||||
mockExecutorExecute: vi.fn(),
|
||||
@@ -22,6 +23,7 @@ const {
|
||||
mockResolveBillingAttribution: vi.fn(),
|
||||
mockGetCustomBlockAuthority: vi.fn(),
|
||||
mockGetPersonalAndWorkspaceEnv: vi.fn(),
|
||||
mockGetUserEmailById: vi.fn(),
|
||||
executorOptions: [] as Array<Record<string, any>>,
|
||||
}))
|
||||
|
||||
@@ -46,6 +48,45 @@ vi.mock('@/lib/workflows/custom-blocks/operations', () => ({
|
||||
getCustomBlockAuthority: mockGetCustomBlockAuthority,
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/users/queries', () => ({
|
||||
getUserEmailById: mockGetUserEmailById,
|
||||
}))
|
||||
|
||||
// Override the global registry mock so the Serializer can carry the start
|
||||
// block's runMetadata param through child deployed-state serialization.
|
||||
vi.mock('@/blocks/registry', () => ({
|
||||
getBlock: vi.fn((type: string) => {
|
||||
if (type === 'start_trigger') {
|
||||
return {
|
||||
name: 'Start',
|
||||
description: 'Unified workflow entry point',
|
||||
category: 'triggers',
|
||||
bgColor: '#34B5FF',
|
||||
icon: () => null,
|
||||
subBlocks: [
|
||||
{ id: 'inputFormat', title: 'Inputs', type: 'input-format' },
|
||||
{ id: 'runMetadata', title: 'Add run metadata', type: 'switch', defaultValue: false },
|
||||
],
|
||||
inputs: {},
|
||||
outputs: {},
|
||||
tools: { access: [] },
|
||||
triggers: { enabled: true, available: ['chat', 'manual', 'api'] },
|
||||
}
|
||||
}
|
||||
return {
|
||||
name: 'Mock Block',
|
||||
description: 'Mock block description',
|
||||
icon: () => null,
|
||||
subBlocks: [],
|
||||
inputs: {},
|
||||
outputs: {},
|
||||
tools: { access: [] },
|
||||
}
|
||||
}),
|
||||
getAllBlocks: vi.fn(() => ({})),
|
||||
getLatestBlock: vi.fn(() => undefined),
|
||||
}))
|
||||
|
||||
vi.mock('@/lib/logs/execution/snapshot/service', () => ({
|
||||
snapshotService: { createSnapshotWithDeduplication: mockCreateSnapshot },
|
||||
}))
|
||||
@@ -401,6 +442,402 @@ describe('WorkflowBlockHandler', () => {
|
||||
expect(executorOptions[0].contextExtensions.workspaceId).toBe('workspace-source')
|
||||
})
|
||||
|
||||
it('builds trusted caller metadata for custom block children with the toggle on', async () => {
|
||||
const customBlock = {
|
||||
...mockBlock,
|
||||
metadata: { id: 'custom_block_abc', name: 'Published Block' },
|
||||
}
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'consumer-1',
|
||||
workspaceId: 'workspace-consumer',
|
||||
executionId: 'exec-1',
|
||||
} as ExecutionContext
|
||||
|
||||
mockGetCustomBlockAuthority.mockResolvedValue({
|
||||
workflowId: 'source-workflow-id',
|
||||
organizationId: 'org-1',
|
||||
ownerUserId: 'owner-9',
|
||||
exposedOutputs: [],
|
||||
requiredInputIds: [],
|
||||
})
|
||||
mockGetPersonalAndWorkspaceEnv.mockResolvedValue({
|
||||
personalDecrypted: {},
|
||||
workspaceDecrypted: {},
|
||||
})
|
||||
mockResolveBillingAttribution.mockResolvedValue({
|
||||
actorUserId: 'owner-9',
|
||||
workspaceId: 'workspace-source',
|
||||
})
|
||||
mockGetUserEmailById.mockImplementation(async (userId: string) =>
|
||||
userId === 'owner-9' ? 'owner@source.com' : userId === 'consumer-1' ? 'a@corp.com' : null
|
||||
)
|
||||
mockFetch.mockImplementation(async (url: unknown) => {
|
||||
if (String(url).includes('/deployed')) {
|
||||
return {
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
deployedState: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {
|
||||
runMetadata: { id: 'runMetadata', type: 'switch', value: true },
|
||||
},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
}
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Source Workflow',
|
||||
workspaceId: 'workspace-source',
|
||||
variables: {},
|
||||
},
|
||||
}),
|
||||
}
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, customBlock, {})
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
const startRunMetadata = executorOptions[0].contextExtensions.startRunMetadata
|
||||
expect(startRunMetadata).toMatchObject({
|
||||
userEmail: 'a@corp.com',
|
||||
workspaceId: 'workspace-consumer',
|
||||
workflowId: 'parent-workflow-id',
|
||||
executionId: 'exec-1',
|
||||
executionType: 'workflow',
|
||||
})
|
||||
expect(mockGetUserEmailById).toHaveBeenCalledWith('consumer-1')
|
||||
expect(mockGetUserEmailById).not.toHaveBeenCalledWith('owner-9')
|
||||
expect(startRunMetadata).not.toHaveProperty('userId')
|
||||
expect(typeof startRunMetadata.startTime).toBe('string')
|
||||
})
|
||||
|
||||
it('propagates the parent run metadata wholesale to nested children', async () => {
|
||||
const customBlock = {
|
||||
...mockBlock,
|
||||
metadata: { id: 'custom_block_abc', name: 'Published Block' },
|
||||
}
|
||||
const inheritedMetadata = {
|
||||
userEmail: 'original@corp.com',
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
executionId: 'exec-1',
|
||||
executionType: 'api',
|
||||
executionMode: 'async' as const,
|
||||
startTime: '2026-07-15T00:00:00.000Z',
|
||||
}
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'publisher-1',
|
||||
workspaceId: 'workspace-intermediate',
|
||||
executionId: 'exec-1',
|
||||
startRunMetadata: inheritedMetadata,
|
||||
} as ExecutionContext
|
||||
|
||||
mockGetCustomBlockAuthority.mockResolvedValue({
|
||||
workflowId: 'source-workflow-id',
|
||||
organizationId: 'org-1',
|
||||
ownerUserId: 'owner-9',
|
||||
exposedOutputs: [],
|
||||
requiredInputIds: [],
|
||||
})
|
||||
mockGetPersonalAndWorkspaceEnv.mockResolvedValue({
|
||||
personalDecrypted: {},
|
||||
workspaceDecrypted: {},
|
||||
})
|
||||
mockResolveBillingAttribution.mockResolvedValue({
|
||||
actorUserId: 'owner-9',
|
||||
workspaceId: 'workspace-source',
|
||||
})
|
||||
mockFetch.mockImplementation(async (url: unknown) => {
|
||||
if (String(url).includes('/deployed')) {
|
||||
return {
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
deployedState: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {
|
||||
runMetadata: { id: 'runMetadata', type: 'switch', value: true },
|
||||
},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
}
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Source Workflow',
|
||||
workspaceId: 'workspace-source',
|
||||
variables: {},
|
||||
},
|
||||
}),
|
||||
}
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, customBlock, {})
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
expect(executorOptions[0].contextExtensions.startRunMetadata).toMatchObject({
|
||||
userEmail: 'original@corp.com',
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
executionMode: 'async',
|
||||
})
|
||||
expect(mockGetUserEmailById).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('preserves a fail-soft null inherited email instead of re-resolving it', async () => {
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'publisher-1',
|
||||
workspaceId: 'workspace-parent',
|
||||
startRunMetadata: {
|
||||
userEmail: null,
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
},
|
||||
} as ExecutionContext
|
||||
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Child Workflow',
|
||||
workspaceId: 'workspace-parent',
|
||||
state: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {
|
||||
runMetadata: { id: 'runMetadata', type: 'switch', value: true },
|
||||
},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, mockBlock, inputs)
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
expect(executorOptions[0].contextExtensions.startRunMetadata.userEmail).toBeNull()
|
||||
expect(mockGetUserEmailById).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('recovers inherited metadata from the seeded start-block state after resume', async () => {
|
||||
const seededMetadata = {
|
||||
userEmail: 'original@corp.com',
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
executionMode: 'sync',
|
||||
}
|
||||
const parentStartBlock = {
|
||||
id: 'parent-start',
|
||||
position: { x: 0, y: 0 },
|
||||
config: { tool: 'start_trigger', params: { runMetadata: true } },
|
||||
inputs: {},
|
||||
outputs: {},
|
||||
metadata: { id: 'start_trigger', name: 'Start', category: 'triggers' },
|
||||
enabled: true,
|
||||
}
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'user-1',
|
||||
workspaceId: 'workspace-parent',
|
||||
workflow: { ...mockContext.workflow, blocks: [parentStartBlock] },
|
||||
blockStates: new Map([
|
||||
[
|
||||
'parent-start',
|
||||
{ output: { metadata: seededMetadata }, executed: true, executionTime: 0 },
|
||||
],
|
||||
]),
|
||||
} as unknown as ExecutionContext
|
||||
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Child Workflow',
|
||||
workspaceId: 'workspace-parent',
|
||||
state: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {
|
||||
runMetadata: { id: 'runMetadata', type: 'switch', value: true },
|
||||
},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, mockBlock, inputs)
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
expect(executorOptions[0].contextExtensions.startRunMetadata).toMatchObject({
|
||||
userEmail: 'original@corp.com',
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
})
|
||||
expect(mockGetUserEmailById).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('passes inherited metadata through a toggle-off child so deeper children keep it', async () => {
|
||||
const inheritedMetadata = {
|
||||
userEmail: 'original@corp.com',
|
||||
workspaceId: 'workspace-original',
|
||||
workflowId: 'workflow-original',
|
||||
}
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'publisher-1',
|
||||
workspaceId: 'workspace-parent',
|
||||
startRunMetadata: inheritedMetadata,
|
||||
} as ExecutionContext
|
||||
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Child Workflow',
|
||||
workspaceId: 'workspace-parent',
|
||||
state: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, mockBlock, inputs)
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
expect(executorOptions[0].contextExtensions.startRunMetadata).toBe(inheritedMetadata)
|
||||
})
|
||||
|
||||
it('passes no run metadata when the child start block toggle is off', async () => {
|
||||
const ctx = {
|
||||
...mockContext,
|
||||
userId: 'consumer-1',
|
||||
workspaceId: 'workspace-parent',
|
||||
} as ExecutionContext
|
||||
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
json: () =>
|
||||
Promise.resolve({
|
||||
data: {
|
||||
name: 'Child Workflow',
|
||||
workspaceId: 'workspace-parent',
|
||||
state: {
|
||||
blocks: {
|
||||
start: {
|
||||
id: 'start',
|
||||
type: 'start_trigger',
|
||||
name: 'Start',
|
||||
position: { x: 0, y: 0 },
|
||||
subBlocks: {},
|
||||
outputs: {},
|
||||
enabled: true,
|
||||
},
|
||||
},
|
||||
edges: [],
|
||||
loops: {},
|
||||
parallels: {},
|
||||
},
|
||||
},
|
||||
}),
|
||||
})
|
||||
mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } })
|
||||
mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } })
|
||||
|
||||
await handler.execute(ctx, mockBlock, inputs)
|
||||
|
||||
expect(executorOptions).toHaveLength(1)
|
||||
expect(executorOptions[0].contextExtensions.startRunMetadata).toBeUndefined()
|
||||
expect(mockGetUserEmailById).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('should fail closed when the executing context has no workspace', async () => {
|
||||
mockFetch.mockResolvedValueOnce({
|
||||
ok: true,
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { getErrorMessage } from '@sim/utils/errors'
|
||||
import { generateId } from '@sim/utils/id'
|
||||
import { isRecordLike } from '@sim/utils/object'
|
||||
import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution'
|
||||
import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils'
|
||||
import { buildNextCallChain, validateCallChain } from '@/lib/execution/call-chain'
|
||||
@@ -8,6 +9,7 @@ import { calculateCostSummary } from '@/lib/logs/execution/logging-factory'
|
||||
import { snapshotService } from '@/lib/logs/execution/snapshot/service'
|
||||
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
|
||||
import type { TraceSpan } from '@/lib/logs/types'
|
||||
import { getUserEmailById } from '@/lib/users/queries'
|
||||
import { getCustomBlockAuthority } from '@/lib/workflows/custom-blocks/operations'
|
||||
import { extractInputFieldsFromBlocks } from '@/lib/workflows/input-format'
|
||||
import { type CustomBlockOutput, isCustomBlockType } from '@/blocks/custom/build-config'
|
||||
@@ -16,17 +18,20 @@ import { Executor } from '@/executor'
|
||||
import { BlockType, DEFAULTS, HTTP } from '@/executor/constants'
|
||||
import { ChildWorkflowError } from '@/executor/errors/child-workflow-error'
|
||||
import type { WorkflowNodeMetadata } from '@/executor/execution/types'
|
||||
import type {
|
||||
BlockHandler,
|
||||
ExecutionContext,
|
||||
ExecutionResult,
|
||||
StreamingExecution,
|
||||
import {
|
||||
type BlockHandler,
|
||||
type ExecutionContext,
|
||||
type ExecutionResult,
|
||||
START_BLOCK_METADATA_FIELD,
|
||||
type StartBlockRunMetadata,
|
||||
type StreamingExecution,
|
||||
} from '@/executor/types'
|
||||
import { hasExecutionResult } from '@/executor/utils/errors'
|
||||
import { buildAPIUrl, buildAuthHeaders } from '@/executor/utils/http'
|
||||
import { getIterationContext } from '@/executor/utils/iteration-context'
|
||||
import { parseJSON } from '@/executor/utils/json'
|
||||
import { lazyCleanupInputMapping } from '@/executor/utils/lazy-cleanup'
|
||||
import { isRunMetadataEnabled, resolveExecutorStartBlock } from '@/executor/utils/start-block'
|
||||
import { Serializer } from '@/serializer'
|
||||
import type { SerializedBlock } from '@/serializer/types'
|
||||
|
||||
@@ -40,6 +45,22 @@ function getValueAtPath(source: unknown, path: string): unknown {
|
||||
}, source)
|
||||
}
|
||||
|
||||
/**
|
||||
* Recover the trusted run metadata from the executing workflow's seeded
|
||||
* start-block output. Resume restores block states from the snapshot but never
|
||||
* rebuilds `ctx.startRunMetadata`, so the seeded output is the surviving copy.
|
||||
*/
|
||||
function readSeededStartRunMetadata(ctx: ExecutionContext): StartBlockRunMetadata | undefined {
|
||||
const resolution = resolveExecutorStartBlock(ctx.workflow?.blocks ?? [], {
|
||||
execution: 'manual',
|
||||
isChildWorkflow: false,
|
||||
})
|
||||
if (!resolution || !isRunMetadataEnabled(resolution.block)) return undefined
|
||||
|
||||
const seeded = ctx.blockStates.get(resolution.blockId)?.output?.[START_BLOCK_METADATA_FIELD]
|
||||
return isRecordLike(seeded) ? (seeded as StartBlockRunMetadata) : undefined
|
||||
}
|
||||
|
||||
/**
|
||||
* Remap a custom block's resolved input mapping from source-field ids to the
|
||||
* child workflow's current field names. The consumer's sub-block values are keyed
|
||||
@@ -384,6 +405,40 @@ export class WorkflowBlockHandler implements BlockHandler {
|
||||
})
|
||||
}
|
||||
|
||||
// Trusted run metadata for the child's Start block. Every field describes
|
||||
// the INVOKING run (the caller's email, workspace, and workflow — never the
|
||||
// child's own static, authoring-time-known identity), delivered on a
|
||||
// server-verified channel a consumer's inputs can never spoof.
|
||||
let childStartRunMetadata: StartBlockRunMetadata | undefined
|
||||
const childStartResolution = resolveExecutorStartBlock(childWorkflow.serializedState.blocks, {
|
||||
execution: 'manual',
|
||||
isChildWorkflow: false,
|
||||
})
|
||||
// Resumed executions never rebuild `ctx.startRunMetadata`, so fall back to
|
||||
// the parent's own seeded start-block output — the persisted copy of the
|
||||
// same trusted object, restored from the snapshot on resume.
|
||||
const inherited = ctx.startRunMetadata ?? readSeededStartRunMetadata(ctx)
|
||||
if (childStartResolution && isRunMetadataEnabled(childStartResolution.block)) {
|
||||
// When the parent run already carries trusted metadata, propagate ALL of
|
||||
// it so nested children see one consistent invoking identity (the
|
||||
// original consumer) instead of a mix of original and intermediate.
|
||||
// Inherited email is taken verbatim — a fail-soft null must stay null,
|
||||
// not be re-resolved to the intermediate (publisher) identity.
|
||||
childStartRunMetadata = {
|
||||
userEmail: inherited
|
||||
? (inherited.userEmail ?? null)
|
||||
: ctx.userId
|
||||
? await getUserEmailById(ctx.userId)
|
||||
: null,
|
||||
workspaceId: inherited?.workspaceId ?? ctx.workspaceId ?? null,
|
||||
workflowId: inherited?.workflowId ?? ctx.workflowId ?? null,
|
||||
executionId: ctx.executionId,
|
||||
executionType: 'workflow',
|
||||
executionMode: inherited?.executionMode ?? ctx.metadata.executionMode,
|
||||
startTime: new Date().toISOString(),
|
||||
}
|
||||
}
|
||||
|
||||
const subExecutor = new Executor({
|
||||
workflow: childWorkflow.serializedState,
|
||||
workflowInput: childWorkflowInput,
|
||||
@@ -403,6 +458,9 @@ export class WorkflowBlockHandler implements BlockHandler {
|
||||
// internal tool calls (knowledge, guardrails, MCP, Mothership) can
|
||||
// attach the required billing attribution header.
|
||||
billingAttribution: childBillingAttribution,
|
||||
// Fall back to the inherited metadata so a toggle-off intermediate
|
||||
// child still carries the trusted identity chain to deeper children.
|
||||
startRunMetadata: childStartRunMetadata ?? inherited,
|
||||
abortSignal: ctx.abortSignal,
|
||||
// Propagate in-flight block-output redaction into child workflows so
|
||||
// nested blocks mask outputs too (recurses: each child forwards it).
|
||||
|
||||
@@ -231,6 +231,28 @@ export const EXECUTION_CONTROL_OUTPUT_FIELD_NAMES = [
|
||||
|
||||
export type ExecutionControlOutputFieldName = (typeof EXECUTION_CONTROL_OUTPUT_FIELD_NAMES)[number]
|
||||
|
||||
/** Start block output key that carries trusted, server-injected run metadata. */
|
||||
export const START_BLOCK_METADATA_FIELD = 'metadata'
|
||||
|
||||
/**
|
||||
* Trusted run metadata surfaced under `<start.metadata.*>` when the Start
|
||||
* block's "Add run metadata" toggle is enabled. Built server-side from the
|
||||
* authenticated execution context — never from caller-supplied input.
|
||||
* Every field describes the INVOKING run: on top-level runs that is the run
|
||||
* itself; on child and custom-block executions it is the parent run (its
|
||||
* actor's email, workspace, and workflow) — never the child's own static,
|
||||
* authoring-time-known identity.
|
||||
*/
|
||||
export interface StartBlockRunMetadata {
|
||||
userEmail?: string | null
|
||||
workspaceId?: string | null
|
||||
workflowId?: string | null
|
||||
executionId?: string
|
||||
executionType?: string
|
||||
executionMode?: 'sync' | 'stream' | 'async'
|
||||
startTime?: string
|
||||
}
|
||||
|
||||
export interface BlockLog {
|
||||
blockId: string
|
||||
blockName?: string
|
||||
@@ -291,6 +313,7 @@ interface ExecutionMetadata {
|
||||
useDraftState?: boolean
|
||||
resumeFromSnapshot?: boolean
|
||||
resumeTerminalNoop?: boolean
|
||||
executionMode?: 'sync' | 'stream' | 'async'
|
||||
}
|
||||
|
||||
export interface BlockState {
|
||||
@@ -322,6 +345,8 @@ export interface ExecutionContext {
|
||||
|
||||
blockLogs: BlockLog[]
|
||||
metadata: ExecutionMetadata
|
||||
/** Trusted run metadata for the Start block's "Add run metadata" toggle. */
|
||||
startRunMetadata?: StartBlockRunMetadata
|
||||
environmentVariables: Record<string, string>
|
||||
workflowVariables?: Record<string, any>
|
||||
|
||||
|
||||
@@ -541,4 +541,99 @@ describe('start-block utilities', () => {
|
||||
expect(output.extra).toBe('untouched')
|
||||
})
|
||||
})
|
||||
|
||||
describe('run metadata injection', () => {
|
||||
const runMetadata = {
|
||||
userEmail: 'real@sim.ai',
|
||||
workspaceId: 'ws-1',
|
||||
workflowId: 'wf-1',
|
||||
executionId: 'exec-1',
|
||||
executionType: 'api',
|
||||
executionMode: 'sync' as const,
|
||||
startTime: '2026-07-15T00:00:00.000Z',
|
||||
}
|
||||
|
||||
function createUnifiedResolution(subBlocks?: Record<string, unknown>) {
|
||||
const block = createBlock('start_trigger', 'start', subBlocks ? { subBlocks } : undefined)
|
||||
return {
|
||||
blockId: 'start',
|
||||
block,
|
||||
path: StartBlockPath.UNIFIED,
|
||||
} as const
|
||||
}
|
||||
|
||||
it.concurrent('server metadata overrides caller-supplied metadata key', () => {
|
||||
const resolution = createUnifiedResolution({ runMetadata: { value: true } })
|
||||
|
||||
const output = buildStartBlockOutput({
|
||||
resolution,
|
||||
workflowInput: {
|
||||
metadata: { userEmail: 'attacker@x.com' },
|
||||
simUserEmail: 'attacker@x.com',
|
||||
payload: 'value',
|
||||
},
|
||||
runMetadata,
|
||||
})
|
||||
|
||||
expect(output.metadata).toEqual(runMetadata)
|
||||
expect(output.payload).toBe('value')
|
||||
expect(output.simUserEmail).toBe('attacker@x.com')
|
||||
})
|
||||
|
||||
it.concurrent('strips caller-supplied metadata key when no trusted metadata exists', () => {
|
||||
const resolution = createUnifiedResolution({ runMetadata: { value: true } })
|
||||
|
||||
const output = buildStartBlockOutput({
|
||||
resolution,
|
||||
workflowInput: { metadata: { userEmail: 'attacker@x.com' } },
|
||||
})
|
||||
|
||||
expect(output).not.toHaveProperty('metadata')
|
||||
})
|
||||
|
||||
it.concurrent('throws when an input format field is named metadata', () => {
|
||||
const resolution = createUnifiedResolution({
|
||||
runMetadata: { value: true },
|
||||
inputFormat: { value: [{ name: 'metadata', type: 'string' }] },
|
||||
})
|
||||
|
||||
expect(() =>
|
||||
buildStartBlockOutput({
|
||||
resolution,
|
||||
workflowInput: {},
|
||||
runMetadata,
|
||||
})
|
||||
).toThrow('reserves the "metadata" output')
|
||||
})
|
||||
|
||||
it.concurrent('toggle off leaves caller-supplied metadata untouched', () => {
|
||||
const resolution = createUnifiedResolution()
|
||||
|
||||
const output = buildStartBlockOutput({
|
||||
resolution,
|
||||
workflowInput: { metadata: { custom: 'value' } },
|
||||
runMetadata,
|
||||
})
|
||||
|
||||
expect(output.metadata).toEqual({ custom: 'value' })
|
||||
})
|
||||
|
||||
it.concurrent('reads the toggle from config params when metadata subBlocks are absent', () => {
|
||||
const block = createBlock('start_trigger', 'start')
|
||||
block.config.params.runMetadata = true
|
||||
const resolution = {
|
||||
blockId: 'start',
|
||||
block,
|
||||
path: StartBlockPath.UNIFIED,
|
||||
} as const
|
||||
|
||||
const output = buildStartBlockOutput({
|
||||
resolution,
|
||||
workflowInput: { metadata: 'spoof' },
|
||||
runMetadata,
|
||||
})
|
||||
|
||||
expect(output.metadata).toEqual(runMetadata)
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -13,6 +13,8 @@ import type { InputFormatField } from '@/lib/workflows/types'
|
||||
import {
|
||||
EXECUTION_CONTROL_OUTPUT_FIELD_NAMES,
|
||||
type NormalizedBlockOutput,
|
||||
START_BLOCK_METADATA_FIELD,
|
||||
type StartBlockRunMetadata,
|
||||
type UserFile,
|
||||
} from '@/executor/types'
|
||||
import type { SerializedBlock } from '@/serializer/types'
|
||||
@@ -141,6 +143,14 @@ function extractInputFormat(block: SerializedBlock): InputFormatField[] {
|
||||
.map((field) => field)
|
||||
}
|
||||
|
||||
/**
|
||||
* Reads the Start block's "Add run metadata" toggle from the serialized block.
|
||||
*/
|
||||
export function isRunMetadataEnabled(block: SerializedBlock): boolean {
|
||||
const value = readMetadataSubBlockValue(block, 'runMetadata') ?? block.config?.params?.runMetadata
|
||||
return value === true || value === 'true'
|
||||
}
|
||||
|
||||
function normalizeLegacyStarterMode(modeValue: unknown): 'manual' | 'api' | 'chat' | null {
|
||||
if (modeValue === 'chat') return 'chat'
|
||||
if (modeValue === 'api' || modeValue === 'run') return 'api'
|
||||
@@ -580,11 +590,32 @@ function extractSubBlocks(block: SerializedBlock): Record<string, unknown> | und
|
||||
export interface StartBlockOutputOptions {
|
||||
resolution: ExecutorStartResolution
|
||||
workflowInput: unknown
|
||||
/** Trusted, server-built run metadata. Only applied when the block's toggle is on. */
|
||||
runMetadata?: StartBlockRunMetadata
|
||||
}
|
||||
|
||||
function assertNoMetadataInputFormatField(
|
||||
inputFormat: InputFormatField[],
|
||||
block: SerializedBlock
|
||||
): void {
|
||||
const hasMetadataField = inputFormat.some(
|
||||
(field) => readInputFormatFieldName(field) === START_BLOCK_METADATA_FIELD
|
||||
)
|
||||
|
||||
if (!hasMetadataField) {
|
||||
return
|
||||
}
|
||||
|
||||
const blockName = block.metadata?.name ?? block.id
|
||||
throw new Error(
|
||||
`Start block "${blockName}" has "Add run metadata" enabled, which reserves the "${START_BLOCK_METADATA_FIELD}" output. Rename the "${START_BLOCK_METADATA_FIELD}" input format field or disable the toggle.`
|
||||
)
|
||||
}
|
||||
|
||||
export function buildStartBlockOutput(options: StartBlockOutputOptions): NormalizedBlockOutput {
|
||||
const { resolution, workflowInput } = options
|
||||
const inputFormat = extractInputFormat(resolution.block)
|
||||
const runMetadataEnabled = isRunMetadataEnabled(resolution.block)
|
||||
const legacyStarterMode =
|
||||
resolution.path === StartBlockPath.LEGACY_STARTER
|
||||
? getSerializedLegacyStarterMode(resolution.block)
|
||||
@@ -592,6 +623,9 @@ export function buildStartBlockOutput(options: StartBlockOutputOptions): Normali
|
||||
|
||||
if (pathConsumesInputFormat(resolution.path, legacyStarterMode)) {
|
||||
assertNoReservedInputFormatFields(inputFormat, resolution.block)
|
||||
if (runMetadataEnabled) {
|
||||
assertNoMetadataInputFormatField(inputFormat, resolution.block)
|
||||
}
|
||||
}
|
||||
|
||||
const { finalInput, structuredInput, hasStructured } = deriveInputFromFormat(
|
||||
@@ -631,6 +665,15 @@ export function buildStartBlockOutput(options: StartBlockOutputOptions): Normali
|
||||
output = buildManualTriggerOutput(finalInput, workflowInput)
|
||||
}
|
||||
|
||||
if (runMetadataEnabled) {
|
||||
// The metadata key is server-owned when the toggle is on: any caller-supplied
|
||||
// value is dropped, and absent trusted metadata leaves the key absent (fail closed).
|
||||
delete output[START_BLOCK_METADATA_FIELD]
|
||||
if (options.runMetadata) {
|
||||
output[START_BLOCK_METADATA_FIELD] = { ...options.runMetadata }
|
||||
}
|
||||
}
|
||||
|
||||
assertNoReservedStartOutputFields(output, resolution.block)
|
||||
return output
|
||||
}
|
||||
|
||||
@@ -1,8 +1,12 @@
|
||||
import { db } from '@sim/db'
|
||||
import { settings, user } from '@sim/db/schema'
|
||||
import { createLogger } from '@sim/logger'
|
||||
import { getErrorMessage } from '@sim/utils/errors'
|
||||
import { eq } from 'drizzle-orm'
|
||||
import type { UserSettingsApi } from '@/lib/api/contracts/user'
|
||||
|
||||
const logger = createLogger('UserQueries')
|
||||
|
||||
/**
|
||||
* Default user settings returned for unauthenticated users or when no
|
||||
* settings row exists yet.
|
||||
@@ -76,6 +80,26 @@ export async function getUserSettings(userId: string | null): Promise<UserSettin
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Loads a user's email address, or `null` when no matching user exists or the
|
||||
* lookup fails. Fail-soft: callers use this for optional run metadata, and a
|
||||
* transient lookup error must not abort an otherwise valid execution.
|
||||
*/
|
||||
export async function getUserEmailById(userId: string): Promise<string | null> {
|
||||
try {
|
||||
const [userRecord] = await db
|
||||
.select({ email: user.email })
|
||||
.from(user)
|
||||
.where(eq(user.id, userId))
|
||||
.limit(1)
|
||||
|
||||
return userRecord?.email ?? null
|
||||
} catch (error) {
|
||||
logger.warn('Failed to load user email', { userId, error: getErrorMessage(error) })
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Loads a user's public profile fields, or `null` when no matching user exists.
|
||||
*/
|
||||
|
||||
@@ -70,4 +70,25 @@ describe('block outputs parity', () => {
|
||||
expect(getEffectiveBlockOutputType('agent', 'min', subBlocks, options)).toBe('number')
|
||||
expect(getEffectiveBlockOutputType('agent', 'max', subBlocks, options)).toBe('number')
|
||||
})
|
||||
|
||||
it.concurrent('surfaces start run metadata paths only when the toggle is on', () => {
|
||||
const subBlocks: SubBlocks = { runMetadata: { value: true } }
|
||||
const options = { triggerMode: true }
|
||||
|
||||
const outputs = getEffectiveBlockOutputs('start_trigger', subBlocks, options)
|
||||
const paths = getEffectiveBlockOutputPaths('start_trigger', subBlocks, options)
|
||||
|
||||
expect(outputs).toHaveProperty('metadata')
|
||||
expect(paths).toContain('metadata.userEmail')
|
||||
expect(paths).toContain('metadata.executionType')
|
||||
expect(paths).toContain('metadata.workflowId')
|
||||
expect(
|
||||
getEffectiveBlockOutputType('start_trigger', 'metadata.userEmail', subBlocks, options)
|
||||
).toBe('string')
|
||||
|
||||
const offOutputs = getEffectiveBlockOutputs('start_trigger', {}, options)
|
||||
const offPaths = getEffectiveBlockOutputPaths('start_trigger', {}, options)
|
||||
expect(offOutputs).not.toHaveProperty('metadata')
|
||||
expect(offPaths.some((path) => path.startsWith('metadata'))).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
@@ -144,6 +144,35 @@ const UNIFIED_START_OUTPUTS: OutputDefinition = {
|
||||
files: { type: 'file[]', description: 'User uploaded files' },
|
||||
}
|
||||
|
||||
const START_RUN_METADATA_OUTPUT = {
|
||||
type: 'json',
|
||||
description: 'Trusted run metadata (server-injected)',
|
||||
properties: {
|
||||
userEmail: {
|
||||
type: 'string',
|
||||
description: 'Email of the user who invoked the run (for custom blocks, the invoking user)',
|
||||
},
|
||||
workspaceId: {
|
||||
type: 'string',
|
||||
description: 'Workspace ID of the invoking run (for custom blocks, the invoking workspace)',
|
||||
},
|
||||
workflowId: {
|
||||
type: 'string',
|
||||
description: 'ID of the invoking workflow (for custom blocks, the invoking workflow)',
|
||||
},
|
||||
executionId: {
|
||||
type: 'string',
|
||||
description: 'Execution ID (child workflows share the parent execution ID)',
|
||||
},
|
||||
executionType: {
|
||||
type: 'string',
|
||||
description: 'How the run started: manual, api, chat, mcp, copilot, table, or workflow',
|
||||
},
|
||||
executionMode: { type: 'string', description: 'Execution mode: sync, stream, or async' },
|
||||
startTime: { type: 'string', description: 'ISO timestamp when the execution started' },
|
||||
},
|
||||
} as OutputFieldDefinition
|
||||
|
||||
function applyInputFormatFields(
|
||||
inputFormat: InputFormatField[],
|
||||
outputs: OutputDefinition
|
||||
@@ -188,7 +217,14 @@ function getUnifiedStartOutputs(
|
||||
): OutputDefinition {
|
||||
const outputs = { ...UNIFIED_START_OUTPUTS }
|
||||
const normalizedInputFormat = normalizeInputFormatValue(subBlocks?.inputFormat?.value)
|
||||
return applyInputFormatFields(normalizedInputFormat, outputs)
|
||||
const withFields = applyInputFormatFields(normalizedInputFormat, outputs)
|
||||
|
||||
const runMetadataValue = subBlocks?.runMetadata?.value
|
||||
if (runMetadataValue === true || runMetadataValue === 'true') {
|
||||
withFields.metadata = START_RUN_METADATA_OUTPUT
|
||||
}
|
||||
|
||||
return withFields
|
||||
}
|
||||
|
||||
function getLegacyStarterOutputs(
|
||||
@@ -491,7 +527,7 @@ function traverseOutputPath(outputs: OutputDefinition, pathParts: string[]): unk
|
||||
current = currentObj[part]
|
||||
} else if (
|
||||
'type' in currentObj &&
|
||||
currentObj.type === 'object' &&
|
||||
(currentObj.type === 'object' || currentObj.type === 'json') &&
|
||||
'properties' in currentObj &&
|
||||
currentObj.properties &&
|
||||
typeof currentObj.properties === 'object'
|
||||
|
||||
@@ -174,7 +174,7 @@ describe('executeWorkflowCore terminal finalization sequencing', () => {
|
||||
})
|
||||
|
||||
mergeSubblockStateWithValuesMock.mockImplementation((blocks) => blocks)
|
||||
serializeWorkflowMock.mockReturnValue({ loops: {}, parallels: {} })
|
||||
serializeWorkflowMock.mockReturnValue({ blocks: [], loops: {}, parallels: {} })
|
||||
buildTraceSpansMock.mockReturnValue({ traceSpans: [{ id: 'span-1' }], totalDuration: 123 })
|
||||
findStartBlockMock.mockReturnValue({
|
||||
blockId: 'start-block',
|
||||
|
||||
@@ -21,6 +21,7 @@ import type { LoggingSession } from '@/lib/logs/execution/logging-session'
|
||||
import { redactLargeValueRefsInValue } from '@/lib/logs/execution/pii-large-values'
|
||||
import { redactObjectStrings } from '@/lib/logs/execution/pii-redaction'
|
||||
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
|
||||
import { getUserEmailById } from '@/lib/users/queries'
|
||||
import { getCustomBlockRowsForWorkspace } from '@/lib/workflows/custom-blocks/operations'
|
||||
import {
|
||||
loadDeployedWorkflowState,
|
||||
@@ -38,8 +39,13 @@ import type {
|
||||
IterationContext,
|
||||
SerializableExecutionState,
|
||||
} from '@/executor/execution/types'
|
||||
import type { ExecutionResult, NormalizedBlockOutput } from '@/executor/types'
|
||||
import type {
|
||||
ExecutionResult,
|
||||
NormalizedBlockOutput,
|
||||
StartBlockRunMetadata,
|
||||
} from '@/executor/types'
|
||||
import { hasExecutionResult } from '@/executor/utils/errors'
|
||||
import { isRunMetadataEnabled } from '@/executor/utils/start-block'
|
||||
import { buildParallelSentinelEndId, buildSentinelEndId } from '@/executor/utils/subflow-utils'
|
||||
import { Serializer } from '@/serializer'
|
||||
|
||||
@@ -743,6 +749,24 @@ async function executeWorkflowCoreImpl(
|
||||
}
|
||||
}
|
||||
|
||||
let startRunMetadata: StartBlockRunMetadata | undefined
|
||||
if (resolvedTriggerBlockId) {
|
||||
const entryBlock = serializedWorkflow.blocks.find(
|
||||
(block) => block.id === resolvedTriggerBlockId
|
||||
)
|
||||
if (entryBlock && isRunMetadataEnabled(entryBlock)) {
|
||||
startRunMetadata = {
|
||||
userEmail: await getUserEmailById(userId),
|
||||
workspaceId: providedWorkspaceId,
|
||||
workflowId,
|
||||
executionId,
|
||||
executionType: triggerType,
|
||||
executionMode: metadata.executionMode ?? 'sync',
|
||||
startTime: metadata.startTime,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const contextExtensions: ContextExtensions = {
|
||||
stream: !!onStream,
|
||||
selectedOutputs,
|
||||
@@ -770,6 +794,7 @@ async function executeWorkflowCoreImpl(
|
||||
dagIncomingEdges: snapshot.state?.dagIncomingEdges,
|
||||
snapshotState: snapshot.state,
|
||||
metadata,
|
||||
startRunMetadata,
|
||||
abortSignal,
|
||||
includeFileBase64,
|
||||
base64MaxBytes,
|
||||
|
||||
Reference in New Issue
Block a user