Skip to content

Commit 845a80e

Browse files
committed
Block unit-testing: variableInputs isolation runs + workflow deps command
run_block/run_from_block gain variableInputs — mocked upstream outputs keyed by block name or id, resolved with the executor's own name normalization and overlaid on the run-from-block snapshot (marked executed), so a block runs in FULL isolation: with no prior execution the run starts purely from the mocks (both the execute route and the copilot server-fallback use case synthesize the empty snapshot). Unknown names and non-object mocks fail fast with actionable messages. `sim workflow deps <workflowId> <blockId>` (agent-CLI augmentation) lists everything one block consumes — upstream blocks with the exact paths read, loop/parallel/variable refs, {{ENV}} secrets — plus a ready-made mock skeleton, using the executor's reference shapes. Claude-Session: https://claude.ai/code/session_01CgaxNAaeD3taGdghbXn17w
1 parent 9db052b commit 845a80e

15 files changed

Lines changed: 324 additions & 8 deletions

File tree

‎apps/sim/app/api/workflows/[id]/execute/route.ts‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,7 @@ import type {
165165
import type { BlockLog, NormalizedBlockOutput, StreamingExecution } from '@/executor/types'
166166
import { getExecutionErrorStatus, hasExecutionResult } from '@/executor/utils/errors'
167167
import type { ResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry'
168+
import { emptyRunFromBlockSnapshot } from '@/executor/utils/run-from-block'
168169
import { Serializer } from '@/serializer'
169170
import { CORE_TRIGGER_TYPES, type CoreTriggerType } from '@/stores/logs/filters/types'
170171

@@ -783,6 +784,7 @@ async function handleExecutePost(
783784
startBlockId: string
784785
sourceSnapshot: SerializableExecutionState
785786
sourceExecutionId?: string
787+
variableInputs?: Record<string, unknown>
786788
}
787789
| undefined
788790
if (rawRunFromBlock) {
@@ -817,10 +819,17 @@ async function handleExecutePost(
817819
startBlockId: rawRunFromBlock.startBlockId,
818820
sourceSnapshot: rawRunFromBlock.sourceSnapshot as SerializableExecutionState,
819821
}
822+
} else if (rawRunFromBlock.variableInputs && !isPublicApiAccess) {
823+
// Pure-mock isolated run: no prior execution needed — the caller supplies
824+
// every upstream output the block reads (the executor overlays them).
825+
resolvedRunFromBlock = {
826+
startBlockId: rawRunFromBlock.startBlockId,
827+
sourceSnapshot: emptyRunFromBlockSnapshot(),
828+
}
820829
} else {
821830
return NextResponse.json(
822831
{
823-
error: `No execution state found for ${rawRunFromBlock.executionId === 'latest' ? 'workflow' : `execution ${rawRunFromBlock.executionId}`}. Run the full workflow first.`,
832+
error: `No execution state found for ${rawRunFromBlock.executionId === 'latest' ? 'workflow' : `execution ${rawRunFromBlock.executionId}`}. Run the full workflow first, or pass variableInputs mocking the upstream outputs.`,
824833
},
825834
{ status: 400 }
826835
)
@@ -839,12 +848,20 @@ async function handleExecutePost(
839848
startBlockId: rawRunFromBlock.startBlockId,
840849
sourceSnapshot: rawRunFromBlock.sourceSnapshot as SerializableExecutionState,
841850
}
851+
} else if (rawRunFromBlock.variableInputs && !isPublicApiAccess) {
852+
resolvedRunFromBlock = {
853+
startBlockId: rawRunFromBlock.startBlockId,
854+
sourceSnapshot: emptyRunFromBlockSnapshot(),
855+
}
842856
} else {
843857
return NextResponse.json(
844858
{ error: 'runFromBlock requires either sourceSnapshot or executionId' },
845859
{ status: 400 }
846860
)
847861
}
862+
if (resolvedRunFromBlock && rawRunFromBlock.variableInputs && !isPublicApiAccess) {
863+
resolvedRunFromBlock.variableInputs = rawRunFromBlock.variableInputs
864+
}
848865
}
849866

850867
// For API key and internal JWT auth, the entire body is the input (except for our control fields)

‎apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -964,6 +964,8 @@ export interface WorkflowExecutionOptions {
964964
runFromBlock?: {
965965
startBlockId: string
966966
executionId?: string
967+
/** Mocked upstream outputs (block name/id → output object) overlaid server-side. */
968+
variableInputs?: Record<string, unknown>
967969
}
968970
}
969971

@@ -1037,6 +1039,9 @@ export async function executeWorkflowWithFullLogging(
10371039
runFromBlock: {
10381040
startBlockId: options.runFromBlock.startBlockId,
10391041
executionId: options.runFromBlock.executionId || 'latest',
1042+
...(options.runFromBlock.variableInputs
1043+
? { variableInputs: options.runFromBlock.variableInputs }
1044+
: {}),
10401045
},
10411046
}
10421047
: {}),

‎apps/sim/executor/execution/executor.ts‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import { type ClonedSubflowInfo, ParallelExpander } from '@/executor/utils/paral
2525
import { isResolvedSecretTraceProvenanceV1 } from '@/executor/utils/resolved-secret-trace-registry'
2626
import {
2727
computeExecutionSets,
28+
overlayVariableInputs,
2829
type RunFromBlockContext,
2930
resolveContainerToSentinelStart,
3031
validateRunFromBlock,
@@ -129,8 +130,12 @@ export class DAGExecutor {
129130
async executeFromBlock(
130131
workflowId: string,
131132
startBlockId: string,
132-
sourceSnapshot: SerializableExecutionState
133+
sourceSnapshot: SerializableExecutionState,
134+
variableInputs?: Record<string, unknown>
133135
): Promise<ExecutionResult> {
136+
if (variableInputs && Object.keys(variableInputs).length > 0) {
137+
sourceSnapshot = overlayVariableInputs(this.workflow, sourceSnapshot, variableInputs)
138+
}
134139
// Build full DAG with all blocks to compute upstream set for snapshot filtering
135140
// includeAllBlocks is needed because the startBlockId might be a trigger not reachable from the main trigger
136141
const dag = this.dagBuilder.build(this.workflow, { includeAllBlocks: true })
Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/**
2+
* @vitest-environment node
3+
*/
4+
import { describe, expect, it } from 'vitest'
5+
import { emptyRunFromBlockSnapshot, overlayVariableInputs } from '@/executor/utils/run-from-block'
6+
import type { SerializedWorkflow } from '@/serializer/types'
7+
8+
const workflow = {
9+
version: '1',
10+
blocks: [
11+
{ id: 'b-start', metadata: { id: 'starter', name: 'Start' } },
12+
{ id: 'b-fetch', metadata: { id: 'api', name: 'Fetch Orders' } },
13+
{ id: 'b-fn', metadata: { id: 'function', name: 'Transform' } },
14+
],
15+
connections: [],
16+
loops: {},
17+
} as unknown as SerializedWorkflow
18+
19+
describe('overlayVariableInputs', () => {
20+
it('resolves block names with executor normalization and marks mocks executed', () => {
21+
const out = overlayVariableInputs(workflow, emptyRunFromBlockSnapshot(), {
22+
'fetch orders': { content: 'rows' },
23+
})
24+
expect(out.blockStates['b-fetch']).toEqual({
25+
output: { content: 'rows' },
26+
executed: true,
27+
executionTime: 0,
28+
})
29+
expect(out.executedBlocks).toContain('b-fetch')
30+
})
31+
32+
it('accepts raw block ids and overrides existing snapshot state', () => {
33+
const snapshot = emptyRunFromBlockSnapshot()
34+
snapshot.blockStates['b-fetch'] = {
35+
output: { content: 'stale' },
36+
executed: true,
37+
executionTime: 5,
38+
}
39+
snapshot.executedBlocks.push('b-fetch')
40+
const out = overlayVariableInputs(workflow, snapshot, { 'b-fetch': { content: 'fresh' } })
41+
expect(out.blockStates['b-fetch'].output).toEqual({ content: 'fresh' })
42+
expect(out.executedBlocks.filter((id) => id === 'b-fetch')).toHaveLength(1)
43+
})
44+
45+
it('fails fast on unknown block names, listing what exists', () => {
46+
expect(() =>
47+
overlayVariableInputs(workflow, emptyRunFromBlockSnapshot(), { nope: { a: 1 } })
48+
).toThrow(/no block named "nope".*Fetch Orders/s)
49+
})
50+
51+
it('rejects non-object mocks with an actionable message', () => {
52+
expect(() =>
53+
overlayVariableInputs(workflow, emptyRunFromBlockSnapshot(), { Transform: 'scalar' })
54+
).toThrow(/must be an object/)
55+
})
56+
57+
it('does not mutate the input snapshot', () => {
58+
const snapshot = emptyRunFromBlockSnapshot()
59+
overlayVariableInputs(workflow, snapshot, { Transform: { ok: true } })
60+
expect(snapshot.blockStates).toEqual({})
61+
expect(snapshot.executedBlocks).toEqual([])
62+
})
63+
})

‎apps/sim/executor/utils/run-from-block.ts‎

Lines changed: 64 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,9 @@
1-
import { LOOP, PARALLEL } from '@/executor/constants'
1+
import { isPlainRecord } from '@sim/utils/object'
2+
import { LOOP, normalizeName, PARALLEL } from '@/executor/constants'
23
import type { DAG } from '@/executor/dag/builder'
4+
import type { SerializableExecutionState } from '@/executor/execution/types'
5+
import type { NormalizedBlockOutput } from '@/executor/types'
6+
import type { SerializedWorkflow } from '@/serializer/types'
37

48
/**
59
* Builds the sentinel-start node ID for a loop.
@@ -255,3 +259,62 @@ function findParentParallel(blockId: string, dag: DAG): string | undefined {
255259
}
256260
return undefined
257261
}
262+
263+
/**
264+
* Overlays caller-provided mock outputs onto a run-from-block snapshot so a
265+
* block can run in isolation: each entry (keyed by block name or id) becomes
266+
* that block's state, marked executed, overriding whatever the snapshot holds.
267+
* Name matching uses the executor's own normalization — the same rule reference
268+
* resolution applies — and an unknown key fails fast with the names that do
269+
* exist, since a silently dropped mock would just resurface as an unresolved
270+
* reference deeper in the run.
271+
*/
272+
export function overlayVariableInputs(
273+
workflow: SerializedWorkflow,
274+
snapshot: SerializableExecutionState,
275+
variableInputs: Record<string, unknown>
276+
): SerializableExecutionState {
277+
const idsByName = new Map<string, string>()
278+
for (const block of workflow.blocks) {
279+
if (block.metadata?.name) idsByName.set(normalizeName(block.metadata.name), block.id)
280+
}
281+
const blockIds = new Set(workflow.blocks.map((block) => block.id))
282+
283+
const blockStates = { ...snapshot.blockStates }
284+
const executedBlocks = [...snapshot.executedBlocks]
285+
for (const [key, value] of Object.entries(variableInputs)) {
286+
const blockId = blockIds.has(key) ? key : idsByName.get(normalizeName(key))
287+
if (!blockId) {
288+
const known = workflow.blocks
289+
.map((block) => block.metadata?.name)
290+
.filter((name): name is string => Boolean(name))
291+
throw new Error(
292+
`variableInputs: no block named "${key}" in this workflow. Blocks: ${known.join(', ')}`
293+
)
294+
}
295+
if (!isPlainRecord(value)) {
296+
throw new Error(
297+
`variableInputs["${key}"] must be an object shaped like that block's output (references read paths off it).`
298+
)
299+
}
300+
blockStates[blockId] = {
301+
output: value as NormalizedBlockOutput,
302+
executed: true,
303+
executionTime: 0,
304+
}
305+
if (!executedBlocks.includes(blockId)) executedBlocks.push(blockId)
306+
}
307+
return { ...snapshot, blockStates, executedBlocks }
308+
}
309+
310+
/** The empty prior-state a pure-mock isolated run starts from. */
311+
export function emptyRunFromBlockSnapshot(): SerializableExecutionState {
312+
return {
313+
blockStates: {},
314+
executedBlocks: [],
315+
blockLogs: [],
316+
decisions: { router: {}, condition: {} },
317+
completedLoops: [],
318+
activeExecutionPath: [],
319+
}
320+
}

‎apps/sim/lib/api/contracts/workflows.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -333,6 +333,13 @@ export type ReorderWorkflowsBody = z.input<typeof reorderWorkflowsBodySchema>
333333

334334
export const executeWorkflowRunFromBlockSchema = z.object({
335335
startBlockId: requiredFieldSchema('Start block ID is required'),
336+
/**
337+
* Mocked upstream outputs keyed by block name or id: each entry becomes that
338+
* block's state (marked executed) so the start block runs in isolation
339+
* without a prior execution — and overlays the resolved snapshot when one
340+
* exists. Names resolve server-side with the executor's own normalization.
341+
*/
342+
variableInputs: z.record(z.string(), z.unknown()).optional(),
336343
sourceSnapshot: z
337344
.object({
338345
blockStates: z.record(z.string(), z.any()),

‎apps/sim/lib/mothership/tools/client/run-tool-execution.ts‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -550,16 +550,23 @@ async function doExecuteRunTool(
550550
})()
551551

552552
const runFromBlock = (() => {
553+
// Mocked upstream outputs ride through to the server, which overlays them on
554+
// the latest snapshot — or runs purely from them when no execution exists.
555+
const variableInputs = isPlainRecord(params.variableInputs)
556+
? (params.variableInputs as Record<string, unknown>)
557+
: undefined
553558
if (toolName === RunFromBlock.id && params.startBlockId) {
554559
return {
555560
startBlockId: params.startBlockId as string,
556561
executionId: (params.executionId as string | undefined) || 'latest',
562+
...(variableInputs ? { variableInputs } : {}),
557563
}
558564
}
559565
if (toolName === RunBlock.id && params.blockId) {
560566
return {
561567
startBlockId: params.blockId as string,
562568
executionId: (params.executionId as string | undefined) || 'latest',
569+
...(variableInputs ? { variableInputs } : {}),
563570
}
564571
}
565572
return undefined
Lines changed: 123 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,123 @@
1+
import { fetchWorkflowState } from '@/lib/mothership/tools/handlers/agent-cli/commands/workflow-views'
2+
import {
3+
type AgentCliCommand,
4+
agentCliFail,
5+
agentCliOk,
6+
} from '@/lib/mothership/tools/handlers/agent-cli/types'
7+
import { normalizeName, SPECIAL_REFERENCE_PREFIXES } from '@/executor/constants'
8+
9+
/**
10+
* `workflow deps <workflowId> <blockId>` — everything one block consumes, so the
11+
* agent knows exactly what to mock before running it in isolation with
12+
* run_block's variableInputs. Tokens are extracted with the same shapes the
13+
* executor resolves (`<ref.path>` templates, `{{ENV}}` secrets) and block heads
14+
* are matched through the executor's own name normalization — this command must
15+
* never re-invent resolution semantics.
16+
*/
17+
18+
const TEMPLATE_REF = /<([^<>]+)>/g
19+
const ENV_REF = /\{\{\s*([A-Za-z0-9_-]+)\s*\}\}/g
20+
21+
interface DepView {
22+
token: string
23+
kind: 'block' | 'loop' | 'parallel' | 'variable' | 'env' | 'unknown'
24+
blockId?: string
25+
blockName?: string
26+
paths?: string[]
27+
}
28+
29+
function stringLeaves(value: unknown, out: string[]): void {
30+
if (typeof value === 'string') out.push(value)
31+
else if (Array.isArray(value)) for (const item of value) stringLeaves(item, out)
32+
else if (typeof value === 'object' && value !== null)
33+
for (const item of Object.values(value)) stringLeaves(item, out)
34+
}
35+
36+
export const workflowDepsCommand: AgentCliCommand = {
37+
path: ['workflow', 'deps'],
38+
summary:
39+
'List every reference one block consumes (upstream blocks, variables, env) — what to mock for an isolated run',
40+
usage: 'workflow deps <workflowId> <blockId>',
41+
async execute(rest, runtime) {
42+
const [workflowId, blockId] = rest
43+
if (!workflowId || !blockId)
44+
return agentCliFail('Usage: sim workflow deps <workflowId> <blockId>')
45+
const state = await fetchWorkflowState(runtime, workflowId)
46+
const blocks = (state.blocks ?? {}) as Record<string, Record<string, unknown>>
47+
const block = blocks[blockId]
48+
if (!block) return agentCliFail(`No block ${blockId} in workflow ${workflowId}`)
49+
50+
const nameToId = new Map<string, string>()
51+
const idToName = new Map<string, string>()
52+
for (const [id, raw] of Object.entries(blocks)) {
53+
const name = typeof raw.name === 'string' ? raw.name : undefined
54+
if (name) {
55+
nameToId.set(normalizeName(name), id)
56+
idToName.set(id, name)
57+
}
58+
}
59+
60+
const leaves: string[] = []
61+
stringLeaves(block.subBlocks ?? block, leaves)
62+
63+
const byToken = new Map<string, DepView>()
64+
const envs = new Set<string>()
65+
for (const leaf of leaves) {
66+
for (const match of leaf.matchAll(TEMPLATE_REF)) {
67+
const token = match[1]
68+
if (!token || byToken.has(token)) continue
69+
const [head = '', ...pathParts] = token.split('.')
70+
const path = pathParts.join('.')
71+
const special = (SPECIAL_REFERENCE_PREFIXES as readonly string[]).includes(head)
72+
if (special) {
73+
byToken.set(token, { token, kind: head as 'loop' | 'parallel' | 'variable' })
74+
continue
75+
}
76+
const refBlockId = blocks[head] ? head : nameToId.get(normalizeName(head))
77+
if (refBlockId && refBlockId !== blockId) {
78+
const existing = [...byToken.values()].find((d) => d.blockId === refBlockId)
79+
if (existing) {
80+
if (path && !existing.paths?.includes(path)) existing.paths?.push(path)
81+
byToken.set(token, existing)
82+
} else {
83+
byToken.set(token, {
84+
token,
85+
kind: 'block',
86+
blockId: refBlockId,
87+
blockName: idToName.get(refBlockId),
88+
paths: path ? [path] : [],
89+
})
90+
}
91+
} else if (!refBlockId) {
92+
byToken.set(token, { token, kind: 'unknown' })
93+
}
94+
}
95+
for (const match of leaf.matchAll(ENV_REF)) {
96+
if (match[1]) envs.add(match[1])
97+
}
98+
}
99+
100+
const deps = [...new Set(byToken.values())]
101+
const blockDeps = deps.filter((d) => d.kind === 'block')
102+
return agentCliOk(
103+
JSON.stringify(
104+
{
105+
blockId,
106+
blockName: idToName.get(blockId),
107+
references: deps,
108+
env: [...envs].sort(),
109+
// Ready-made skeleton for run_block's variableInputs: mock each upstream
110+
// block's output at the paths this block actually reads.
111+
mock: Object.fromEntries(
112+
blockDeps.map((d) => [
113+
d.blockName ?? d.blockId,
114+
d.paths?.length ? d.paths : ['<output>'],
115+
])
116+
),
117+
},
118+
null,
119+
2
120+
)
121+
)
122+
},
123+
}

0 commit comments

Comments
 (0)