36:set-merge-loop-code node - #36
Conversation
📝 WalkthroughWalkthroughThis PR adds four new node types to the workflow execution system: SET (variable assignment), MERGE (data combining), LOOP_OVER_ITEMS (iteration), and CODE (JavaScript sandbox execution). It includes database migrations, UI components, executor implementations, realtime channels, and updates to workflow planning/validation infrastructure. Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant UI as React Flow UI
participant Executor as Node Executor
participant Channel as Realtime Channel
participant Context as Execution Context
User->>UI: Configure node settings
UI->>UI: Render dialog form
User->>UI: Submit configuration
UI->>Context: Update node.data in React Flow
rect rgb(100, 150, 200, 0.5)
Note over Executor,Context: SET Node Execution
Executor->>Channel: Publish "loading"
Executor->>Context: Build output object (with/without previous data)
Executor->>Context: Resolve field values (literals or expressions)
Executor->>Context: Coerce by declared types
Executor->>Context: Assign to output variable
Executor->>Channel: Publish "success"
Executor-->>Context: Return augmented context
end
rect rgb(150, 100, 200, 0.5)
Note over Executor,Context: LOOP_OVER_ITEMS Execution
Executor->>Channel: Publish "loading"
Executor->>Context: Resolve itemsPath to array
Executor->>Executor: Build loop plan (chunks/batch if needed)
Executor->>Executor: Execute loop units (parallel/sequential)
loop For each unit
Executor->>Context: Run downstream node chain
Executor->>Context: Update loop context
end
Executor->>Channel: Publish "success" with stats
Executor-->>Context: Return context with loop metadata
end
rect rgb(200, 100, 150, 0.5)
Note over Executor,Context: MERGE Node Execution
Executor->>Channel: Publish "loading"
Executor->>Context: Resolve inputAPath & inputBPath
Executor->>Executor: Apply selected merge mode logic
Executor->>Context: Store merged result under outputVariableName
Executor->>Channel: Publish "success"
Executor-->>Context: Return augmented context
end
rect rgb(150, 200, 100, 0.5)
Note over Executor,Context: CODE Node Execution
Executor->>Channel: Publish "loading"
Executor->>Executor: Create sandboxed VM with context
Executor->>Executor: Execute user code as IIFE with timeout
Executor->>Executor: Capture console.log output
Executor->>Context: Validate and coerce code result
Executor->>Context: Store under variableName
Executor->>Channel: Publish "success"
Executor-->>Context: Return augmented context
end
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Pull request overview
This PR introduces four new workflow node types (SET, MERGE, LOOP_OVER_ITEMS, CODE) across the database schema, execution engine, UI node palette/components, realtime channels, and the AI workflow planner/validator.
Changes:
- Add new node types to Prisma
NodeTypeenum + migration, and wire them into the node selector/components registry. - Implement executors and configuration dialogs for Set, Merge, Loop Over Items, and Code nodes, including realtime status channels/tokens.
- Extend AI planner/validator/catalog to recognize and validate the new nodes, plus improve
step.runkey uniqueness by includingnodeId.
Reviewed changes
Copilot reviewed 40 out of 40 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| src/lib/node-type.ts | Adds exported constants for new node types. |
| src/inngest/functions.ts | Registers new realtime channels and adds loop execution logic inside the workflow runner. |
| src/inngest/channels/set-node.ts | New realtime channel for Set node status updates. |
| src/inngest/channels/merge-node.ts | New realtime channel for Merge node status updates. |
| src/inngest/channels/loop-over-items.ts | New realtime channel for Loop Over Items status/progress updates. |
| src/inngest/channels/code-node.ts | New realtime channel for Code node status updates. |
| src/features/workflows/server/ai-builder.ts | Expands node catalog/defaults/fallback plan/missing-input detection for new nodes. |
| src/features/workflows/lib/workflow-validator.ts | Adds validation for new node types and improves output-variable detection. |
| src/features/workflows/lib/node-schemas.ts | Adds Zod schemas/output schemas/requirements for new nodes + dynamic outputVariableName support. |
| src/features/workflows/lib/ai-workflow-schema.ts | Allows AI planner to emit the new node types. |
| src/features/executions/lib/executor-registry.ts | Registers new executors (set/merge/loop/code). |
| src/features/executions/components/telegram/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/slack/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/openai/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/http-request/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/google-sheets/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/gemini/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/email/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/discord/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/anthropic/executor.ts | Makes step.run keys node-specific to avoid collisions. |
| src/features/executions/components/set/node.tsx | Adds Set node UI component with dialog + realtime status integration. |
| src/features/executions/components/set/executor.ts | Implements Set node execution (field setting, coercion, templates). |
| src/features/executions/components/set/dialog.tsx | Adds Set node configuration dialog with preview. |
| src/features/executions/components/set/actions.ts | Adds realtime token fetch for Set node channel. |
| src/features/executions/components/merge/node.tsx | Adds Merge node UI with dual inputs and settings dialog. |
| src/features/executions/components/merge/executor.ts | Implements Merge node execution modes and input resolution. |
| src/features/executions/components/merge/dialog.tsx | Adds Merge node configuration dialog. |
| src/features/executions/components/merge/actions.ts | Adds realtime token fetch for Merge node channel. |
| src/features/executions/components/loop-over-items/node.tsx | Adds Loop Over Items node UI with progress display. |
| src/features/executions/components/loop-over-items/executor.ts | Implements loop plan building + initializes loop status/output. |
| src/features/executions/components/loop-over-items/dialog.tsx | Adds Loop Over Items configuration dialog. |
| src/features/executions/components/loop-over-items/actions.ts | Adds realtime token fetch for Loop Over Items channel. |
| src/features/executions/components/code/node.tsx | Adds Code node UI component with dialog + realtime status integration. |
| src/features/executions/components/code/executor.ts | Implements server-side JS execution for Code node. |
| src/features/executions/components/code/dialog.tsx | Adds Code node configuration dialog with templates/testing/highlighting. |
| src/features/executions/components/code/actions.ts | Adds realtime token fetch for Code node channel. |
| src/config/node-components.ts | Registers new node React components in the node type map. |
| src/components/node-selector.tsx | Adds new node types to the node picker UI. |
| prisma/schema.prisma | Adds SET/MERGE/LOOP_OVER_ITEMS/CODE to Prisma NodeType enum. |
| prisma/migrations/20260420193000_core_logic_nodes/migration.sql | Migration to add the new enum values in Postgres. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| function setValueByPath( | ||
| target: Record<string, unknown>, | ||
| rawPath: string, | ||
| value: unknown, | ||
| ) { | ||
| const tokens = parsePathToken(rawPath); | ||
| if (tokens.length === 0) { | ||
| throw new NonRetriableError("SET node field name is invalid."); | ||
| } | ||
| let current: Record<string, unknown> = target; | ||
| for (let i = 0; i < tokens.length - 1; i += 1) { | ||
| const token = tokens[i]; | ||
| if (!token) continue; | ||
| const existing = current[token]; | ||
| if (!existing || typeof existing !== "object" || Array.isArray(existing)) { | ||
| current[token] = {}; | ||
| } | ||
| current = current[token] as Record<string, unknown>; | ||
| } |
There was a problem hiding this comment.
The UI/help text suggests array index paths like jobs[0].title, and parsePathToken supports [0], but setValueByPath always creates plain objects for intermediate segments and never creates/updates arrays. This means paths with indexes won’t behave as advertised. Either remove index support from the UX/docs or extend setValueByPath to correctly create arrays when numeric tokens are used.
| if ( | ||
| node.type === NodeType.MERGE && | ||
| !String(node.data.inputBPath ?? "").trim() | ||
| ) { | ||
| push({ | ||
| nodeId: node.id, | ||
| field: "inputBPath", | ||
| question: "What should be used as Merge input B?", | ||
| whyItMatters: "Merge requires two inputs to combine branch data.", | ||
| }); | ||
| } |
There was a problem hiding this comment.
computeMissingInputs only prompts for MERGE.inputBPath, but WorkflowValidator.validateMergeNode requires both inputAPath and inputBPath. This can leave AI-generated plans stuck with an empty/placeholder inputAPath and still failing validation/execution. Add a similar missing-input question for inputAPath (and consider whether outputVariableName also needs prompting when omitted).
| const nodeStatus = useNodeStatus({ | ||
| nodeId: props.id, | ||
| channel: LOOP_OVER_ITEMS_CHANNEL_NAME, | ||
| topic: "status", | ||
| refreshToken: fetchLoopOverItemsRealtimeToken, | ||
| }); | ||
| const { data } = useInngestSubscription({ | ||
| refreshToken: fetchLoopOverItemsRealtimeToken, | ||
| enabled: true, | ||
| }); | ||
|
|
||
| const handleOpenSettings = () => setDialogOpen(true); | ||
|
|
||
| const handleSubmit = (values: LoopOverItemsFormValues) => { | ||
| setNodes((nodes) => | ||
| nodes.map((node) => | ||
| node.id === props.id | ||
| ? { | ||
| ...node, | ||
| data: { | ||
| ...node.data, | ||
| ...values, | ||
| }, | ||
| } | ||
| : node, | ||
| ), | ||
| ); | ||
| }; | ||
|
|
||
| const modeLabel = | ||
| props.data?.mode === "parallel" | ||
| ? "Parallel" | ||
| : props.data?.mode === "batch" | ||
| ? `Batch (${props.data.batchSize ?? 10})` | ||
| : "Sequential"; | ||
| const description = props.data?.itemsPath | ||
| ? `${modeLabel} · ${props.data.itemsPath}` | ||
| : "Not configured"; | ||
| const progressLabel = useMemo(() => { | ||
| const lastMessage = data | ||
| ?.filter( | ||
| (message) => | ||
| message.kind === "data" && | ||
| message.channel === LOOP_OVER_ITEMS_CHANNEL_NAME && | ||
| message.topic === "status" && | ||
| message.data.nodeId === props.id, | ||
| ) | ||
| .sort((a, b) => { | ||
| if (a.kind !== "data" || b.kind !== "data") return 0; | ||
| return ( | ||
| new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime() | ||
| ); | ||
| })[0]; | ||
|
|
||
| if (!lastMessage || lastMessage.kind !== "data") return null; | ||
| const processed = Number(lastMessage.data.processed ?? 0); | ||
| const total = Number(lastMessage.data.totalItems ?? 0); | ||
| const failed = Number(lastMessage.data.failed ?? 0); | ||
| if (!Number.isFinite(total) || total <= 0) return null; | ||
| return `${Math.max(0, processed)}/${total} processed${failed > 0 ? ` · ${failed} failed` : ""}`; | ||
| }, [data, props.id]); |
There was a problem hiding this comment.
This component opens a second useInngestSubscription in addition to useNodeStatus, then filters+sorts all received messages on every update to compute progress. That’s extra websocket load and O(n log n) work per message. Consider extending useNodeStatus (or adding a dedicated hook) to expose the latest matching message so Loop Over Items can get progress without a separate subscription and repeated sorting.
| while (currentNodeId) { | ||
| const currentNode = nodeById.get(currentNodeId); | ||
| if (!currentNode) break; | ||
| linearChain.push(currentNodeId); | ||
|
|
||
| const currentOutgoing = (outgoingByNodeId.get(currentNodeId) ?? | ||
| []) as typeof workflowConnections; | ||
| const currentIncoming = (incomingByNodeId.get(currentNodeId) ?? | ||
| []) as typeof workflowConnections; | ||
| if (currentIncoming.length > 1 || currentOutgoing.length !== 1) { | ||
| break; | ||
| } | ||
| currentNodeId = currentOutgoing[0]?.toNodeId; | ||
| } |
There was a problem hiding this comment.
The loop runner builds a linearChain but simply breaks when it encounters nodes with multiple incoming edges or branching outputs. Because the loop node also overrides routing, anything past that break point will be silently skipped by the outer graph execution. Consider throwing a clear NonRetriableError when the loop body contains unsupported joins/branches (incoming > 1 or outgoing > 1), so workflows don’t partially execute without an obvious reason.
| const fallbackItems = Array.isArray(context.items) ? context.items : []; | ||
| const sandbox = { | ||
| input: context, | ||
| items: fallbackItems, | ||
| payload: context, | ||
| console: { | ||
| log: (..._args: unknown[]) => {}, | ||
| }, | ||
| }; |
There was a problem hiding this comment.
The sandbox assigns payload: context, which makes payload identical to input and ignores any context.payload value set by upstream nodes (notably the Loop Over Items runner sets payload to the current item/batch). If the intent is for code to access the per-item payload, this should likely be payload: context.payload (and similarly consider whether input should be the whole context or a specific object).
| }, | ||
| }, | ||
| nodeId, | ||
| ["__loop_internal__"], |
There was a problem hiding this comment.
loopOverItemsExecutor sets a route output of __loop_internal__ via withNodeRoute(...). Since the Loop Over Items UI uses the default source handle id (source-1), this route will not match any outgoing connection’s fromOutput, so downstream nodes will never execute via the normal graph runner. Either route to the actual outgoing handle id(s) (e.g. source-1) or avoid setting a route here and handle loop control exclusively in executeWorkflow with explicit skipping/advancement logic.
| ["__loop_internal__"], | |
| ["source-1"], |
| if (loopPlan.mode === "parallel") { | ||
| await Promise.all( | ||
| loopPlan.units.map((unit, index) => runLoopUnit(unit, index)), | ||
| ); | ||
| } else { |
There was a problem hiding this comment.
Promise.all(loopPlan.units.map(...)) can create unbounded concurrency for large arrays, potentially overwhelming external APIs and the worker (memory/CPU) and spamming realtime progress messages. Consider adding a concurrency limit (e.g., a small pool) for parallel mode or enforcing a maxItems cap when running in parallel.
| const wrapped = `(async () => {\n${userCode}\n})()`; | ||
| let result: unknown; | ||
| try { | ||
| const script = new vm.Script(wrapped); | ||
| result = await script.runInNewContext(sandbox, { timeout: timeoutMs }); | ||
| } catch (error) { |
There was a problem hiding this comment.
This executes user-supplied JavaScript on the server using Node’s vm module. vm.runInNewContext is not a security boundary; malicious code can often escape the sandbox (e.g., via Function constructors) and access process, environment variables, filesystem, network, etc. If workflows are user-generated in a multi-tenant environment, this is a critical RCE vector—consider moving execution to a hardened sandbox (separate process/container, allowlist-only API) or using a library designed for isolation.
| if (type === "json" || type === "array") { | ||
| const parsed = | ||
| typeof value === "string" ? JSON.parse(value) : JSON.parse(String(value)); | ||
| if (type === "array" && !Array.isArray(parsed)) { | ||
| throw new NonRetriableError("SET node array field must resolve to an array."); | ||
| } | ||
| return parsed; | ||
| } |
There was a problem hiding this comment.
For type === "json" | "array", coerceSetValue always runs JSON.parse(...) even when the resolved value is already an object/array (e.g. when the value is a full-path expression like {{someArray}}). This will throw (String(array) => "a,b" / String(object) => "[object Object]") and prevents using SET to copy structured values from context. Consider returning the value directly when it is already the expected type, and only JSON.parse when the resolved value is a string.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/features/workflows/lib/workflow-validator.ts (1)
807-822:⚠️ Potential issue | 🟠 Major
availableVariablesis computed but never read — dead code; and itsdynamicNamedisagrees with the shared registry.Two issues in this block:
availableVariablesis populated but never consumed; the subsequent template check on line 831-834 usesthis.variableRegistry(built once in the constructor). The per-iteration accumulation intoavailableVariableshas no effect — this phase is not actually doing the "data available at this step" check its header comment promises.- The
dynamicNameresolution here uses the.trim()ed string, whilebuildVariableRegistryinsrc/features/workflows/lib/node-schemas.ts(lines 503-510) stores the untrimmednode.data.variableName/outputVariableNameas the registry key. A config likevariableName: " jobs "will register as" jobs "but be referenced as{{jobs}}(the regex captures word chars only), causingvalidateTemplateVariableto falsely report "undefined variable".Recommend either (a) aligning both sides to use the trimmed name (preferred), or (b) removing this unused local state.
🔧 Suggested fix in node-schemas.ts (root cause)
- const dynamicVariableName = - typeof node.data.variableName === "string" && - node.data.variableName.trim() - ? node.data.variableName - : typeof node.data.outputVariableName === "string" && - node.data.outputVariableName.trim() - ? node.data.outputVariableName - : ""; + const trimmedVariableName = + typeof node.data.variableName === "string" + ? node.data.variableName.trim() + : ""; + const trimmedOutputVariableName = + typeof node.data.outputVariableName === "string" + ? node.data.outputVariableName.trim() + : ""; + const dynamicVariableName = + trimmedVariableName || trimmedOutputVariableName || "";🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/workflows/lib/workflow-validator.ts` around lines 807 - 822, The loop builds availableVariables and computes dynamicName but never uses it, and dynamicName trims names while buildVariableRegistry stores untrimmed keys causing mismatches; update the code so the per-node availability check actually consults the same registry by either (preferred) trimming names consistently at registration and lookup: modify buildVariableRegistry (function buildVariableRegistry in node-schemas.ts) to store trimmed node.data.variableName/outputVariableName and change this loop in workflow-validator.ts to use the same trimmed dynamicName when adding to availableVariables and then replace subsequent uses of this.variableRegistry in validateTemplateVariable checks with availableVariables for step-local validation, or if you choose the other option remove availableVariables and keep a single source of truth—ensure variableRegistry and dynamicName use the identical (trimmed) normalization function.src/features/workflows/lib/node-schemas.ts (1)
499-532:⚠️ Potential issue | 🟠 MajorRegistry keys use untrimmed variable names — templates referencing
{{foo}}won't match registrations of" foo ".The ternary assigns
dynamicVariableName = node.data.variableName/node.data.outputVariableNamewithout trimming, even though the truthiness guard uses.trim(). So a user-enteredvariableNameof" jobs "registers as:registry[" jobs "] = { … }But
validateTemplateVariableextracts references with/\{\{(\w+)/g, which capturesjobs(word chars only). Lookup misses → template reported as "undefined variable" inWorkflowValidator.validateNodeTemplatesandvalidateDataFlow. The same input passes the Zod schema (z.string().min(1)doesn't require trimming).Trim once at the source so the registry is the single source of truth:
🔧 Suggested fix
- const dynamicVariableName = - typeof node.data.variableName === "string" && - node.data.variableName.trim() - ? node.data.variableName - : typeof node.data.outputVariableName === "string" && - node.data.outputVariableName.trim() - ? node.data.outputVariableName - : ""; + const trimmedVar = + typeof node.data.variableName === "string" + ? node.data.variableName.trim() + : ""; + const trimmedOutputVar = + typeof node.data.outputVariableName === "string" + ? node.data.outputVariableName.trim() + : ""; + const dynamicVariableName = trimmedVar || trimmedOutputVar || ""; // Configurable variable name (HTTP_REQUEST/AI/CODE or outputVariableName nodes) if (!schema.variableName && dynamicVariableName) { - const varName = String(dynamicVariableName); + const varName = dynamicVariableName;Pair this with the corresponding fix in
workflow-validator.ts:812-818so both sides converge.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/workflows/lib/node-schemas.ts` around lines 499 - 532, The registry is being keyed with untrimmed variable names because dynamicVariableName is chosen from node.data.variableName / node.data.outputVariableName without trimming; update the logic in the loop that builds registry (referencing nodeOutputSchemas, dynamicVariableName, and registry) to trim the chosen string once (e.g., call .trim() on variableName/outputVariableName when assigning dynamicVariableName and use that trimmed value for registry keys and path) so lookups from validateTemplateVariable / validateNodeTemplates / validateDataFlow match; also ensure the branch that uses schema.variableName similarly uses a trimmed string if schema.variableName can contain user input so both producer (node-schemas.ts) and consumer (workflow-validator.ts validateNodeTemplates/validateDataFlow) use the same trimmed canonical variable name.
♻️ Duplicate comments (1)
src/features/executions/components/set/dialog.tsx (1)
189-218:⚠️ Potential issue | 🟡 MinorPreview coercion has the same JSON-parse-on-non-string defect as the executor.
coercePreviewValueat lines 209-216 mirrors the executor'scoerceSetValuebug: when the resolved expression is already an object/array (from thefullPatternbranch ofresolveValueExpression),typeof value === "string"is false andJSON.parse(String(value))throws on"[object Object]". The preview then silently collapses to "Preview unavailable" (the outertry/catchat line 303), giving users no feedback on what went wrong even though the real SET operation would succeed after the executor fix.Apply the same fix as in the executor:
🔧 Suggested fix
if (type === "json" || type === "array") { - const parsed = - typeof value === "string" ? JSON.parse(value) : JSON.parse(String(value)); + const parsed = typeof value === "string" ? JSON.parse(value) : value; if (type === "array" && !Array.isArray(parsed)) { throw new Error("array expected"); } return parsed; }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/set/dialog.tsx` around lines 189 - 218, coercePreviewValue incorrectly calls JSON.parse on non-strings (causing "[object Object]" parse failures); update the json/array branch in coercePreviewValue to only JSON.parse when typeof value === "string" and otherwise use the value directly (then for type "array" still validate Array.isArray(parsedOrValue") and throw "array expected" if not); mirror the same logic used in the executor's coerceSetValue/resolveValueExpression fix so preview succeeds when given already-parsed objects/arrays.
🧹 Nitpick comments (8)
src/config/node-components.ts (1)
53-53: Remove debugconsole.log.This top-level
console.logruns at module import time on every environment. Consider removing it or gating it behind a debug flag.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/config/node-components.ts` at line 53, Remove the top-level debug console.log that runs at module import; locate the statement logging Object.keys(nodeComponents) (the console.log("[flowforge][nodeComponents]", Object.keys(nodeComponents)) line) and either delete it or replace it with a gated/logged call using the existing logging utility or a debug flag (e.g., check process.env.DEBUG or use the project logger) so it only emits in debug mode rather than at every import.src/features/executions/components/merge/node.tsx (1)
60-69: Replace the 4-deep nested ternary with a lookup.Readability nit — a simple map makes the mode→label relationship clearer and easier to extend:
♻️ Proposed refactor
- const modeLabel = - props.data?.mode === "append_arrays" - ? "Append Arrays" - : props.data?.mode === "merge_by_index" - ? "Merge by Index" - : props.data?.mode === "merge_by_key" - ? "Merge by Key" - : props.data?.mode === "wait_for_both" - ? "Wait for Both Inputs" - : "Combine Objects"; + const MODE_LABELS = { + combine_objects: "Combine Objects", + append_arrays: "Append Arrays", + merge_by_index: "Merge by Index", + merge_by_key: "Merge by Key", + wait_for_both: "Wait for Both Inputs", + } as const; + const modeLabel = MODE_LABELS[props.data?.mode ?? "combine_objects"];🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/merge/node.tsx` around lines 60 - 69, The nested ternary that builds modeLabel from props.data?.mode is hard to read and should be replaced with a simple lookup map; create a constant object (e.g., modeLabelMap) mapping keys "append_arrays", "merge_by_index", "merge_by_key", "wait_for_both" to their respective labels and then set modeLabel = modeLabelMap[props.data?.mode] ?? "Combine Objects" so the mapping is clearer and easier to extend (refer to the existing modeLabel variable and props.data?.mode).src/features/workflows/server/ai-builder.ts (1)
387-399: CODE fallback default is domain-specific and may confuse non-jobs users.The default
codeliteral filters items by"React"in the title — reasonable for the job-scraping demo but misleading when the prompt is generic. Consider picking the template body dynamically based on the selectedtemplate(the UI already hastemplates[key]mapping incode/dialog.tsx), or fall back to a neutralreturn items;.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/workflows/server/ai-builder.ts` around lines 387 - 399, The NodeType.CODE case currently uses a domain-specific default code string that filters for "React"; change it so the default is neutral or derived from the selected template: when building the node in the NodeType.CODE branch, if data.code is empty either (a) set code to a neutral fallback like "return items;" or (b) look up the template body from the existing templates mapping used in code/dialog.tsx using data.template (e.g., templates[String(data.template)]), and use that template body as the default; ensure you still coerce variableName, timeoutMs and template as done now and only replace the hardcoded "React" example with the neutral or template-derived fallback.src/features/executions/components/loop-over-items/dialog.tsx (1)
186-200: Batch size field: clearing input produces0(sets invalid state).
Number("")is0, so clearing the field yieldsbatchSize=0, which failspositive()validation without a clear "required" prompt. For parity withmaxItems(which converts empty string toundefined), consider the same empty-string handling so the superRefine message ("Batch size must be greater than zero") is what the user sees, not a "positive" min-violation.🔧 Small consistency fix
onChange={(event) => - field.onChange(Number(event.target.value)) + field.onChange( + event.target.value + ? Number(event.target.value) + : undefined, + ) }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/loop-over-items/dialog.tsx` around lines 186 - 200, The batchSize input converts an empty string to 0 via Number(event.target.value), causing a validation "positive" error instead of treating empty as undefined like maxItems; update the onChange for the batchSize Input (the handler that calls field.onChange) to map "" to undefined and otherwise pass Number(value) so clearing the field sets undefined (matching maxItems) and lets your superRefine message ("Batch size must be greater than zero") surface.src/features/executions/components/code/executor.ts (1)
62-69: Sandboxconsole.logis silently dropped — consider capturing for debugging.Discarding logs makes CODE node issues very hard to diagnose in production. The dialog already captures logs for the client-side dry run, but the server-side executor throws them away. Consider collecting them into
context(e.g.,{ [variableName]: result, [${variableName}_logs]: logs }) or at minimum forwarding to the run logger so users have parity with the dialog's "Console Logs" panel.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/code/executor.ts` around lines 62 - 69, The sandbox currently swallows console output because sandbox.console.log is a no-op; change it to capture messages into an array and attach or forward them so server-side runs match the dialog's "Console Logs" behavior. Specifically, inside the executor that creates the sandbox (references: sandbox, sandbox.console.log, input, payload, items, context, fallbackItems), replace the no-op with a logger that pushes serialized args into a local logs array and then either merge that array into the execution result/context (e.g., add `${variableName}_logs` or a dedicated logs field on context/payload) or forward each entry to the run logger (e.g., runLogger.log) if available; ensure the logs array is included in the value returned from the executor so callers can surface them.src/features/executions/components/code/dialog.tsx (1)
129-150:formatJavaScriptbreaks on braces inside strings/comments.The line-based brace counter can't distinguish
"}"inside a string literal or//}in a comment from real block endings, so "Format code" will produce visibly wrong indentation on realistic snippets. If full formatting isn't in scope, consider removing the button (or pulling in a real formatter like prettier-standalone for just JS) rather than shipping a heuristic that can silently corrupt the layout.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/code/dialog.tsx` around lines 129 - 150, The current line-based formatter formatJavaScript incorrectly changes indentation when braces appear inside string literals or comments; replace this heuristic by using a real JS formatter (e.g., import and call prettier-standalone with parser "babel") inside the same code path instead of formatJavaScript, or else remove/disable the "Format code" action that calls formatJavaScript; update references to formatJavaScript to call the prettier formatting function (or to no-op/remove the button) and ensure errors from prettier are handled so the original code is left unchanged on failure.src/features/executions/components/set/dialog.tsx (1)
114-168: Path-helpers are duplicated between dialog and executor — extract to a shared module.
parsePathToken,getValueByPath,setValueByPath, and thefullTemplatePatternregex are copy-pasted fromsrc/features/executions/components/set/executor.ts(and again, partially, insrc/features/executions/components/merge/executor.tsandsrc/features/executions/components/loop-over-items/executor.ts). Any fix to one — e.g., the JSON-coercion bug below or the registry-trim fix — must be applied in multiple places to stay consistent.Consider extracting these into
src/features/executions/lib/path-utils.ts(or similar) and importing from both the dialog preview and the executors.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/features/executions/components/set/dialog.tsx` around lines 114 - 168, The path helper functions parsePathToken, getValueByPath, setValueByPath (and the fullTemplatePattern regex) are duplicated across dialog and several executor files; extract them into a single shared module (e.g., create src/features/executions/lib/path-utils.ts) that exports these symbols, update src/features/executions/components/set/dialog.tsx and the executor files (set/executor.ts, merge/executor.ts, loop-over-items/executor.ts) to import those exports, and remove the local duplicates so all callers use the single implementation.src/inngest/functions.ts (1)
252-262:buildLoopPlanis computed twice per loop node.The LOOP_OVER_ITEMS executor at line 243 already calls
buildLoopPlan(data, context)internally (seesrc/features/executions/components/loop-over-items/executor.ts:175) and attaches the plan-derived metadata to the context. Recomputing it here discards that work and re-resolvesitemsPath, which:
- Re-parses
itemsPath(extra Handlebars compile when the value contains{{…}}), and- Creates a subtle divergence risk if anything in
contextchanged between the two calls (unlikely today, but easy to break when the executor is extended).Consider exposing the computed plan from the executor (e.g., stash it under a reserved runtime key via
withNodeRoute/runtime state, or callbuildLoopPlanonce here and skip calling the executor's planning branch). At minimum, leave a comment flagging the duplication.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/inngest/functions.ts` around lines 252 - 262, The code recomputes buildLoopPlan(...) for NodeType.LOOP_OVER_ITEMS which duplicates work already done inside the loop-over-items executor; update the flow so the precomputed plan is reused instead of calling buildLoopPlan again: have the executor expose the computed plan on the runtime/context (e.g., stash under a reserved runtime key via withNodeRoute or runtime state) and read that plan here when present (fall back to buildLoopPlan only if absent), or alternatively compute the plan here once and skip planning in the executor; reference NodeType.LOOP_OVER_ITEMS, buildLoopPlan, withNodeRoute and the loop-over-items executor to locate the relevant code and add a short comment explaining the deduplication decision.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@src/features/executions/components/code/dialog.tsx`:
- Around line 279-286: handleTemplateChange currently unconditionally overwrites
the code field with templates[value], which can silently clobber user edits;
update handleTemplateChange to only replace form.setValue("code",
templates[value]) when the current code is unchanged from the previously
selected template body or when form.formState.dirty for the "code" field is
false (otherwise prompt the user with confirm/inline warning before
overwriting). Similarly update handleResetTemplate to only reset when code
equals the current template body or when the user confirms; use the existing
form, templates, selectedTemplate and the CodeNodeFormValues["template"] types
to locate these handlers and gate the setValue calls accordingly.
In `@src/features/executions/components/code/executor.ts`:
- Around line 71-93: The async IIFE wrapping userCode allows awaitable Promises
to escape vm timeout because runInNewContext returns a Promise synchronously;
update the execution in executor (where vm.Script is created and runInNewContext
is called) to enforce timeout on async code by either adding microtaskMode:
'afterEvaluate' to the runInNewContext options (i.e.,
script.runInNewContext(sandbox, { timeout: timeoutMs, microtaskMode:
'afterEvaluate' })) or, for broader coverage, wrap the awaited result in an
explicit timeout-bound Promise (use Promise.race between the execution Promise
and a timer that rejects after timeoutMs) and throw the same NonRetriableError
message on timeout; keep existing SyntaxError handling intact and ensure the new
timeout path rejects with the same NonRetriableError ("CODE node execution timed
out.").
In `@src/features/executions/components/loop-over-items/actions.ts`:
- Around line 12-18: The fetchLoopOverItemsRealtimeToken function returns a
global token; update it to authenticate the caller and scope the subscription to
a specific execution (or workflow/user) before returning: require an executionId
(or derive it from request context/session), verify ownership/permission using
your auth helper (e.g., checkExecutionOwnership or currentUser) and reject
unauthorized callers, then call getSubscriptionToken with a scoped channel
(e.g., loopOverItemsChannel(executionId) or include an execution-specific topic)
and attach any identifying metadata/claims to the token so it cannot be used for
other executions; ensure functions referenced are
fetchLoopOverItemsRealtimeToken, getSubscriptionToken, and loopOverItemsChannel
and throw/return an authorization error if the ownership check fails.
In `@src/features/executions/components/loop-over-items/executor.ts`:
- Around line 164-211: The executor is pre-writing processed/failed zeros into
context and publishing them, which can be read as stale before the actual loop
runs; remove the early "processed" and "failed" fields from both the initial
loopOverItemsChannel().status publish and the object you inject into context via
withNodeRoute (leave mode, totalItems, totalUnits, delayBetweenItemsMs,
continueOnItemError only), so the final write performed by the loop
orchestration remains authoritative for processed/failed on
plan.outputVariableName; keep the existing success publish for plan.totalItems
=== 0 as-is.
In `@src/features/executions/components/loop-over-items/node.tsx`:
- Around line 30-39: The component is creating two realtime subscriptions:
useNodeStatus(...) already calls useInngestSubscription internally, and the
component separately calls useInngestSubscription(...) causing duplicate
subscriptions; instead modify useNodeStatus to return the full message payload
(including progress fields processed, totalItems, failed) or add a dedicated
hook that reuses its internal subscription, then remove the second
useInngestSubscription call in the component (the variables data and
progressLabel should be derived from the returned nodeStatus payload). Update
references to LOOP_OVER_ITEMS_CHANNEL_NAME and fetchLoopOverItemsRealtimeToken
in the adjusted hook so the same refresh token/channel are used, and remove
redundant kind === "data" checks in the sort/comparator logic if desired.
In `@src/features/executions/components/merge/executor.ts`:
- Around line 139-190: mergeByKey currently throws a misleading error when input
B is empty and handles non-object items asymmetrically between A and B; fix by
removing the post-build check that errors when rightByKey.size === 0 (so empty B
is allowed and leftover A items are still emitted) and make the non-object
handling consistent: either both throw or both skip — change the A-side
early-filter (the loop that iterates over a and currently "continue"s on
non-object items) to throw the same NonRetriableError as the B-side (so both
loops validate objects consistently using keyField), keeping the later logic
that looks up rightByKey and calls mergeObjects(left, right, strategy)
unchanged.
In `@src/features/executions/components/set/executor.ts`:
- Around line 102-128: coerceSetValue currently always calls JSON.parse on
non-string values which turns objects/arrays from resolved expressions into
invalid strings; update coerceSetValue so that for type "json" or "array" it
returns the value directly if it's already a non-null object (for "array" ensure
Array.isArray), otherwise only attempt JSON.parse when value is a string (and
handle parse errors the same way); apply the same change to coercePreviewValue
so the preview behavior matches runtime behavior and keep the existing
NonRetriableError messages for invalid JSON/invalid array cases.
In `@src/features/workflows/lib/workflow-validator.ts`:
- Around line 537-556: The validateCodeNode method currently emits a "warning"
for out-of-range timeoutMs which contradicts nodeInputSchemas[NodeType.CODE]
(timeoutMs min 250 max 10000) that would reject such values; update
validateCodeNode to call this.addError with severity "error" for the timeoutMs
check (instead of "warning") so the validator matches the Zod schema contract
for timeoutMs, or if you prefer warnings change the schema to optional/loosen
bounds—adjust the code in validateCodeNode where timeout is validated and the
addError call for "timeoutMs".
In `@src/features/workflows/server/ai-builder.ts`:
- Around line 1860-1879: The MERGE fallback branch in buildFallbackPlan creates
a non-executable MERGE node (NodeType.MERGE) because inputAPath/inputBPath are
literal placeholders and no second upstream is wired; fix by only adding the
MERGE node when there are two independent upstream branches available, wire both
upstreams to the MERGE node using distinct edge targets (e.g., "target-a" and
"target-b"), and set data.inputAPath and data.inputBPath to template references
pointing at the upstream outputs (e.g., "{{upstreamA.variableName}}",
"{{upstreamB.variableName}}") so mergeExecutor.resolveInput can find actual
values; alternatively remove the MERGE pushNode call from buildFallbackPlan if
two upstreams cannot be guaranteed.
- Around line 1531-1541: computeMissingInputs currently only checks for missing
inputBPath on MERGE nodes, but mergeExecutor can throw "MERGE node input A is
missing." Add a symmetric check for inputAPath in the same block that checks
node.type === NodeType.MERGE: if String(node.data.inputAPath ?? "").trim() is
falsy, call push({ nodeId: node.id, field: "inputAPath", question: "What should
be used as Merge input A?", whyItMatters: "Merge requires two inputs to combine
branch data." }); so both inputAPath and inputBPath are prompted before
execution.
In `@src/inngest/channels/code-node.ts`:
- Around line 3-9: fetchCodeNodeRealtimeToken currently issues a global token
for codeNodeChannel without verifying the requesting user owns the execution;
update fetchCodeNodeRealtimeToken in
src/features/executions/components/code/actions.ts to call
auth.api.getSession(), load the execution record (e.g., via the same execution
service/DB code used elsewhere), and compare the session user id to the
execution.ownerId (or equivalent) before creating the token; if the user is not
the owner, throw an authorization error and do not return a token. Ensure you
reference the same channel/topic usage (CODE_NODE_CHANNEL_NAME / codeNodeChannel
/ "status") when creating the token so the validation precedes token issuance.
In `@src/inngest/functions.ts`:
- Around line 314-343: The batch accounting currently treats a thrown error as
failing the entire unit (failed += unit.items.length) and on non-continue mode
rethrows without populating errors, losing per-item context; change the logic
inside the try/catch around the linearChain execution so that you track failures
at item granularity: when a chainExecutor throws for a specific
runtimeLoopNodeId (and unitIndex), increment failed by 1 (not unit.items.length)
and only increment processed when that specific item completes successfully;
also, before rethrowing when loopPlan.continueOnItemError === false, push a
descriptive entry into errors (e.g., `Unit ${unitIndex + 1}: ${error.message}`
or similar) so the outer catch and the final loop summary can include the
per-item error, and keep existing publishLoopProgress calls
(publishLoopProgress("loading")) unchanged.
---
Outside diff comments:
In `@src/features/workflows/lib/node-schemas.ts`:
- Around line 499-532: The registry is being keyed with untrimmed variable names
because dynamicVariableName is chosen from node.data.variableName /
node.data.outputVariableName without trimming; update the logic in the loop that
builds registry (referencing nodeOutputSchemas, dynamicVariableName, and
registry) to trim the chosen string once (e.g., call .trim() on
variableName/outputVariableName when assigning dynamicVariableName and use that
trimmed value for registry keys and path) so lookups from
validateTemplateVariable / validateNodeTemplates / validateDataFlow match; also
ensure the branch that uses schema.variableName similarly uses a trimmed string
if schema.variableName can contain user input so both producer (node-schemas.ts)
and consumer (workflow-validator.ts validateNodeTemplates/validateDataFlow) use
the same trimmed canonical variable name.
In `@src/features/workflows/lib/workflow-validator.ts`:
- Around line 807-822: The loop builds availableVariables and computes
dynamicName but never uses it, and dynamicName trims names while
buildVariableRegistry stores untrimmed keys causing mismatches; update the code
so the per-node availability check actually consults the same registry by either
(preferred) trimming names consistently at registration and lookup: modify
buildVariableRegistry (function buildVariableRegistry in node-schemas.ts) to
store trimmed node.data.variableName/outputVariableName and change this loop in
workflow-validator.ts to use the same trimmed dynamicName when adding to
availableVariables and then replace subsequent uses of this.variableRegistry in
validateTemplateVariable checks with availableVariables for step-local
validation, or if you choose the other option remove availableVariables and keep
a single source of truth—ensure variableRegistry and dynamicName use the
identical (trimmed) normalization function.
---
Duplicate comments:
In `@src/features/executions/components/set/dialog.tsx`:
- Around line 189-218: coercePreviewValue incorrectly calls JSON.parse on
non-strings (causing "[object Object]" parse failures); update the json/array
branch in coercePreviewValue to only JSON.parse when typeof value === "string"
and otherwise use the value directly (then for type "array" still validate
Array.isArray(parsedOrValue") and throw "array expected" if not); mirror the
same logic used in the executor's coerceSetValue/resolveValueExpression fix so
preview succeeds when given already-parsed objects/arrays.
---
Nitpick comments:
In `@src/config/node-components.ts`:
- Line 53: Remove the top-level debug console.log that runs at module import;
locate the statement logging Object.keys(nodeComponents) (the
console.log("[flowforge][nodeComponents]", Object.keys(nodeComponents)) line)
and either delete it or replace it with a gated/logged call using the existing
logging utility or a debug flag (e.g., check process.env.DEBUG or use the
project logger) so it only emits in debug mode rather than at every import.
In `@src/features/executions/components/code/dialog.tsx`:
- Around line 129-150: The current line-based formatter formatJavaScript
incorrectly changes indentation when braces appear inside string literals or
comments; replace this heuristic by using a real JS formatter (e.g., import and
call prettier-standalone with parser "babel") inside the same code path instead
of formatJavaScript, or else remove/disable the "Format code" action that calls
formatJavaScript; update references to formatJavaScript to call the prettier
formatting function (or to no-op/remove the button) and ensure errors from
prettier are handled so the original code is left unchanged on failure.
In `@src/features/executions/components/code/executor.ts`:
- Around line 62-69: The sandbox currently swallows console output because
sandbox.console.log is a no-op; change it to capture messages into an array and
attach or forward them so server-side runs match the dialog's "Console Logs"
behavior. Specifically, inside the executor that creates the sandbox
(references: sandbox, sandbox.console.log, input, payload, items, context,
fallbackItems), replace the no-op with a logger that pushes serialized args into
a local logs array and then either merge that array into the execution
result/context (e.g., add `${variableName}_logs` or a dedicated logs field on
context/payload) or forward each entry to the run logger (e.g., runLogger.log)
if available; ensure the logs array is included in the value returned from the
executor so callers can surface them.
In `@src/features/executions/components/loop-over-items/dialog.tsx`:
- Around line 186-200: The batchSize input converts an empty string to 0 via
Number(event.target.value), causing a validation "positive" error instead of
treating empty as undefined like maxItems; update the onChange for the batchSize
Input (the handler that calls field.onChange) to map "" to undefined and
otherwise pass Number(value) so clearing the field sets undefined (matching
maxItems) and lets your superRefine message ("Batch size must be greater than
zero") surface.
In `@src/features/executions/components/merge/node.tsx`:
- Around line 60-69: The nested ternary that builds modeLabel from
props.data?.mode is hard to read and should be replaced with a simple lookup
map; create a constant object (e.g., modeLabelMap) mapping keys "append_arrays",
"merge_by_index", "merge_by_key", "wait_for_both" to their respective labels and
then set modeLabel = modeLabelMap[props.data?.mode] ?? "Combine Objects" so the
mapping is clearer and easier to extend (refer to the existing modeLabel
variable and props.data?.mode).
In `@src/features/executions/components/set/dialog.tsx`:
- Around line 114-168: The path helper functions parsePathToken, getValueByPath,
setValueByPath (and the fullTemplatePattern regex) are duplicated across dialog
and several executor files; extract them into a single shared module (e.g.,
create src/features/executions/lib/path-utils.ts) that exports these symbols,
update src/features/executions/components/set/dialog.tsx and the executor files
(set/executor.ts, merge/executor.ts, loop-over-items/executor.ts) to import
those exports, and remove the local duplicates so all callers use the single
implementation.
In `@src/features/workflows/server/ai-builder.ts`:
- Around line 387-399: The NodeType.CODE case currently uses a domain-specific
default code string that filters for "React"; change it so the default is
neutral or derived from the selected template: when building the node in the
NodeType.CODE branch, if data.code is empty either (a) set code to a neutral
fallback like "return items;" or (b) look up the template body from the existing
templates mapping used in code/dialog.tsx using data.template (e.g.,
templates[String(data.template)]), and use that template body as the default;
ensure you still coerce variableName, timeoutMs and template as done now and
only replace the hardcoded "React" example with the neutral or template-derived
fallback.
In `@src/inngest/functions.ts`:
- Around line 252-262: The code recomputes buildLoopPlan(...) for
NodeType.LOOP_OVER_ITEMS which duplicates work already done inside the
loop-over-items executor; update the flow so the precomputed plan is reused
instead of calling buildLoopPlan again: have the executor expose the computed
plan on the runtime/context (e.g., stash under a reserved runtime key via
withNodeRoute or runtime state) and read that plan here when present (fall back
to buildLoopPlan only if absent), or alternatively compute the plan here once
and skip planning in the executor; reference NodeType.LOOP_OVER_ITEMS,
buildLoopPlan, withNodeRoute and the loop-over-items executor to locate the
relevant code and add a short comment explaining the deduplication decision.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 415ff971-7641-4989-baf8-2b43a54c4212
📒 Files selected for processing (40)
prisma/migrations/20260420193000_core_logic_nodes/migration.sqlprisma/schema.prismasrc/components/node-selector.tsxsrc/config/node-components.tssrc/features/executions/components/anthropic/executor.tssrc/features/executions/components/code/actions.tssrc/features/executions/components/code/dialog.tsxsrc/features/executions/components/code/executor.tssrc/features/executions/components/code/node.tsxsrc/features/executions/components/discord/executor.tssrc/features/executions/components/email/executor.tssrc/features/executions/components/gemini/executor.tssrc/features/executions/components/google-sheets/executor.tssrc/features/executions/components/http-request/executor.tssrc/features/executions/components/loop-over-items/actions.tssrc/features/executions/components/loop-over-items/dialog.tsxsrc/features/executions/components/loop-over-items/executor.tssrc/features/executions/components/loop-over-items/node.tsxsrc/features/executions/components/merge/actions.tssrc/features/executions/components/merge/dialog.tsxsrc/features/executions/components/merge/executor.tssrc/features/executions/components/merge/node.tsxsrc/features/executions/components/openai/executor.tssrc/features/executions/components/set/actions.tssrc/features/executions/components/set/dialog.tsxsrc/features/executions/components/set/executor.tssrc/features/executions/components/set/node.tsxsrc/features/executions/components/slack/executor.tssrc/features/executions/components/telegram/executor.tssrc/features/executions/lib/executor-registry.tssrc/features/workflows/lib/ai-workflow-schema.tssrc/features/workflows/lib/node-schemas.tssrc/features/workflows/lib/workflow-validator.tssrc/features/workflows/server/ai-builder.tssrc/inngest/channels/code-node.tssrc/inngest/channels/loop-over-items.tssrc/inngest/channels/merge-node.tssrc/inngest/channels/set-node.tssrc/inngest/functions.tssrc/lib/node-type.ts
| const handleTemplateChange = (value: CodeNodeFormValues["template"]) => { | ||
| form.setValue("template", value); | ||
| form.setValue("code", templates[value]); | ||
| }; | ||
|
|
||
| const handleResetTemplate = () => { | ||
| form.setValue("code", templates[selectedTemplate]); | ||
| }; |
There was a problem hiding this comment.
Silently overwriting user-edited code on template change.
handleTemplateChange replaces the current code field with the template body whenever the user selects a template from the dropdown — easy way to destroy in-progress edits without warning. At minimum, gate the replacement behind form.formState.isDirty for the code field (ask via confirm or a small inline warning), or only prime the template body when code is still equal to the previous template's body.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/code/dialog.tsx` around lines 279 - 286,
handleTemplateChange currently unconditionally overwrites the code field with
templates[value], which can silently clobber user edits; update
handleTemplateChange to only replace form.setValue("code", templates[value])
when the current code is unchanged from the previously selected template body or
when form.formState.dirty for the "code" field is false (otherwise prompt the
user with confirm/inline warning before overwriting). Similarly update
handleResetTemplate to only reset when code equals the current template body or
when the user confirms; use the existing form, templates, selectedTemplate and
the CodeNodeFormValues["template"] types to locate these handlers and gate the
setValue calls accordingly.
| const wrapped = `(async () => {\n${userCode}\n})()`; | ||
| let result: unknown; | ||
| try { | ||
| const script = new vm.Script(wrapped); | ||
| result = await script.runInNewContext(sandbox, { timeout: timeoutMs }); | ||
| } catch (error) { | ||
| if (error instanceof SyntaxError) { | ||
| const lineMatch = (error.stack ?? "").match(/<anonymous>:(\d+):(\d+)/); | ||
| const lineSuffix = lineMatch?.[1] | ||
| ? ` (line ${Math.max(1, Number(lineMatch[1]) - 1)})` | ||
| : ""; | ||
| throw new NonRetriableError( | ||
| `CODE node syntax error${lineSuffix}: ${error.message}`, | ||
| ); | ||
| } | ||
| if ( | ||
| error instanceof Error && | ||
| error.message.toLowerCase().includes("script execution timed out") | ||
| ) { | ||
| throw new NonRetriableError("CODE node execution timed out."); | ||
| } | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Does Node.js vm.Script runInNewContext timeout option apply to asynchronous code and awaited promises?
💡 Result:
No, the timeout option in Node.js vm.Script's runInNewContext does not reliably apply to asynchronous code and awaited promises by default. The timeout primarily enforces limits on synchronous script execution and the main event loop tick. Asynchronous tasks like Promise.then callbacks or setTimeout are scheduled as microtasks or macrotasks that execute after the runInNewContext call returns, potentially bypassing the timeout. Official Node.js documentation explicitly warns about this in the "Timeout interactions with asynchronous tasks and Promises" section. For example, code like 'Promise.resolve.then( => infiniteLoop);' with { timeout: 5 } will return immediately after scheduling the microtask, printing "done executing" before the loop runs, allowing the async code to execute without timeout interruption. To make timeout apply to microtasks (including those from promises and async/await, since async functions use promises), use the microtaskMode: 'afterEvaluate' option. This runs queued microtasks before returning from runInNewContext, subjecting them to the timeout. With this option, the example above times out as expected. Macrotasks (e.g., setTimeout, setImmediate) run in a new event loop created by the timeout option but may still execute after the main script if not drained fully. For comprehensive async timeout coverage, combine microtaskMode with careful context setup, avoiding exposing async scheduling functions or using isolated contexts. This behavior is consistent across recent Node.js versions (v14+), with perf notes on new event loops/threads for timeout.
Citations:
- 1: https://nodejs.org/docs/latest-v20.x/api/vm.html
- 2: https://nodejs.org/api/vm.md
- 3: https://nodejs.org/api/vm.html
- 4: https://nodejs.org/docs/latest/api/vm.html
🏁 Script executed:
# Find package.json to check Node.js version constraints
fd -t f "package.json" | head -5 | xargs -I {} sh -c 'echo "=== {} ===" && cat {}'Repository: newbie-del/FlowForge
Length of output: 3733
🏁 Script executed:
# Search for any existing use of microtaskMode in the codebase
rg "microtaskMode" --type ts --type jsRepository: newbie-del/FlowForge
Length of output: 46
🏁 Script executed:
# Check if there's any documentation or comments about vm timeout handling
rg "vm\.Script|runInNewContext|timeout" --type ts --type js -A 2 -B 2 | head -100Repository: newbie-del/FlowForge
Length of output: 7287
🏁 Script executed:
sed -n '60,100p' src/features/executions/components/code/executor.tsRepository: newbie-del/FlowForge
Length of output: 1278
vm timeout does not bound async user code—awaitable hangs bypass timeout protection.
Node.js vm.Script.runInNewContext's timeout option only guards synchronous execution within the script tick. Because user code is wrapped in an async IIFE, runInNewContext returns the Promise synchronously and the outer await waits on it without timeout bounds. User code like await new Promise(() => {}), a never-resolving promise, or a long async chain will hang the executor indefinitely, preventing status updates and never triggering the timeout error handler at lines 86–91.
Options to fix:
-
Add
microtaskMode: 'afterEvaluate'(simpler): Node.js will drain queued microtasks before returning fromrunInNewContext, subjecting async/await code to the timeout. Requires only:script.runInNewContext(sandbox, { timeout: timeoutMs, microtaskMode: 'afterEvaluate' }). Available in Node.js v14+. -
Use
Promise.racefor comprehensive coverage (fallback): Wraps the execution promise with an explicit timeout to handle both microtasks and scheduled async operations:
Promise.race approach
const wrapped = `(async () => {\n${userCode}\n})()`;
let result: unknown;
try {
const script = new vm.Script(wrapped);
- result = await script.runInNewContext(sandbox, { timeout: timeoutMs });
+ const execPromise = script.runInNewContext(sandbox, {
+ timeout: timeoutMs,
+ microtaskMode: 'afterEvaluate',
+ }) as Promise<unknown>;
+ let timer: NodeJS.Timeout | undefined;
+ const timeoutPromise = new Promise<never>((_, reject) => {
+ timer = setTimeout(
+ () => reject(new Error("Script execution timed out")),
+ timeoutMs,
+ );
+ });
+ try {
+ result = await Promise.race([execPromise, timeoutPromise]);
+ } finally {
+ if (timer) clearTimeout(timer);
+ }
} catch (error) {Note: Neither approach can truly terminate rogue microtask loops (V8 keeps scheduling them). For hard isolation, worker threads with worker.terminate() is the robust long-term option.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/code/executor.ts` around lines 71 - 93,
The async IIFE wrapping userCode allows awaitable Promises to escape vm timeout
because runInNewContext returns a Promise synchronously; update the execution in
executor (where vm.Script is created and runInNewContext is called) to enforce
timeout on async code by either adding microtaskMode: 'afterEvaluate' to the
runInNewContext options (i.e., script.runInNewContext(sandbox, { timeout:
timeoutMs, microtaskMode: 'afterEvaluate' })) or, for broader coverage, wrap the
awaited result in an explicit timeout-bound Promise (use Promise.race between
the execution Promise and a timer that rejects after timeoutMs) and throw the
same NonRetriableError message on timeout; keep existing SyntaxError handling
intact and ensure the new timeout path rejects with the same NonRetriableError
("CODE node execution timed out.").
| export async function fetchLoopOverItemsRealtimeToken(): Promise<LoopOverItemsToken> { | ||
| const token = await getSubscriptionToken(inngest, { | ||
| channel: loopOverItemsChannel(), | ||
| topics: ["status"], | ||
| }); | ||
|
|
||
| return token; |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect realtime token fetchers for auth/ownership checks and scoped channel usage.
# Expectation: each token action validates the current user and limits subscription scope.
rg -n -C5 'export async function fetch.*RealtimeToken|getSubscriptionToken' --type ts
rg -n -C5 '\bauth\(|getServerSession|currentUser|session|userId|workflowId|executionId' --type tsRepository: newbie-del/FlowForge
Length of output: 50377
🏁 Script executed:
# Check the loopOverItemsChannel definition and its scoping
rg -n 'loopOverItemsChannel|loopOverItemsChannel\(' --type ts -A3 -B1Repository: newbie-del/FlowForge
Length of output: 4459
🏁 Script executed:
# Check if any realtime token fetchers have params or auth checks
rg -n 'export async function fetch.*RealtimeToken.*\(' --type ts -A10 | head -100Repository: newbie-del/FlowForge
Length of output: 7957
🏁 Script executed:
# Get full loopOverItemsChannel definition
cat src/inngest/channels/loop-over-items.tsRepository: newbie-del/FlowForge
Length of output: 450
🏁 Script executed:
# Check a few other channel definitions for comparison
cat src/inngest/channels/if-node.tsRepository: newbie-del/FlowForge
Length of output: 344
🏁 Script executed:
# Look for any auth guards or access control in the token fetcher files
rg -n 'auth|getServerSession|currentUser|session|requireAuth' src/features/executions/components/loop-over-items/ src/features/executions/components/anthropic/ --type tsRepository: newbie-del/FlowForge
Length of output: 46
🏁 Script executed:
# Check if getSubscriptionToken does any validation/authorization
rg -n 'getSubscriptionToken|@inngest/realtime' --type ts -B2 -A8 src/ | head -80Repository: newbie-del/FlowForge
Length of output: 6795
🏁 Script executed:
# Look for who calls fetchLoopOverItemsRealtimeToken to understand the usage pattern
rg -n 'fetchLoopOverItemsRealtimeToken' --type ts -B2 -A5Repository: newbie-del/FlowForge
Length of output: 2647
🏁 Script executed:
# Check if Inngest getSubscriptionToken is a custom wrapper or if it's from the library
find src -name '*.ts' -type f | xargs grep -l 'getSubscriptionToken' | head -5Repository: newbie-del/FlowForge
Length of output: 329
🏁 Script executed:
# Check how the node.tsx component uses the token fetcher - is execution/workflow context passed?
cat src/features/executions/components/loop-over-items/node.tsx | head -80Repository: newbie-del/FlowForge
Length of output: 2590
🏁 Script executed:
# Check BaseExecutionNode to see if context is available there
rg -n 'BaseExecutionNode|useNodeStatus' src/features/executions/components/base-execution-node.tsx -A5 -B2 | head -60Repository: newbie-del/FlowForge
Length of output: 801
🏁 Script executed:
# Check how other token fetchers in schedule-trigger handle it (they had auth imports)
cat src/features/triggers/components/schedule-trigger/actions.tsRepository: newbie-del/FlowForge
Length of output: 3025
🏁 Script executed:
# Check what data is actually published on these channels and if it's sensitive
rg -n 'publish.*Channel\(\)\.status\(' src/ -A3 -B3 | head -100Repository: newbie-del/FlowForge
Length of output: 46
🏁 Script executed:
# Verify if there are any ownership/scoping checks in the Inngest realtime middleware
cat src/inngest/client.tsRepository: newbie-del/FlowForge
Length of output: 264
Authorize and scope the realtime subscription token.
This action returns a token for a global loop-over-items status channel without verifying the caller owns the execution. Add authentication and scope the token/channel to a specific execution, workflow, or user before returning it.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/loop-over-items/actions.ts` around lines
12 - 18, The fetchLoopOverItemsRealtimeToken function returns a global token;
update it to authenticate the caller and scope the subscription to a specific
execution (or workflow/user) before returning: require an executionId (or derive
it from request context/session), verify ownership/permission using your auth
helper (e.g., checkExecutionOwnership or currentUser) and reject unauthorized
callers, then call getSubscriptionToken with a scoped channel (e.g.,
loopOverItemsChannel(executionId) or include an execution-specific topic) and
attach any identifying metadata/claims to the token so it cannot be used for
other executions; ensure functions referenced are
fetchLoopOverItemsRealtimeToken, getSubscriptionToken, and loopOverItemsChannel
and throw/return an authorization error if the ownership check fails.
| export const loopOverItemsExecutor: NodeExecutor< | ||
| LoopOverItemsNodeData | ||
| > = async ({ data, nodeId, context, publish }) => { | ||
| await publish( | ||
| loopOverItemsChannel().status({ | ||
| nodeId, | ||
| status: "loading", | ||
| }), | ||
| ); | ||
|
|
||
| try { | ||
| const plan = buildLoopPlan(data, context); | ||
|
|
||
| await publish( | ||
| loopOverItemsChannel().status({ | ||
| nodeId, | ||
| status: plan.totalItems === 0 ? "success" : "loading", | ||
| processed: 0, | ||
| totalItems: plan.totalItems, | ||
| failed: 0, | ||
| }), | ||
| ); | ||
|
|
||
| return withNodeRoute( | ||
| { | ||
| ...context, | ||
| [plan.outputVariableName]: { | ||
| mode: plan.mode, | ||
| totalItems: plan.totalItems, | ||
| totalUnits: plan.units.length, | ||
| delayBetweenItemsMs: plan.delayBetweenItemsMs, | ||
| continueOnItemError: plan.continueOnItemError, | ||
| processed: 0, | ||
| failed: 0, | ||
| }, | ||
| }, | ||
| nodeId, | ||
| ["__loop_internal__"], | ||
| ); | ||
| } catch (error) { | ||
| await publish( | ||
| loopOverItemsChannel().status({ | ||
| nodeId, | ||
| status: "error", | ||
| }), | ||
| ); | ||
| throw error; | ||
| } |
There was a problem hiding this comment.
Executor returns immediately with processed: 0, failed: 0 before loop runs — downstream consumers keying off [outputVariableName] may read stale zeros.
loopOverItemsExecutor writes context[plan.outputVariableName] = { …, processed: 0, failed: 0 } and returns. The actual loop orchestration in src/inngest/functions.ts:376-386 overwrites this same key after all units complete. This works for nodes after the loop (topo-sorted later) but is easy to misread:
- Any helper that eagerly inspects the context between
executor(...)returning and the loop block starting sees zeros. - If the loop block is ever skipped (e.g.,
outgoing.length === 0in functions.ts line 264 gate — no outgoing edge means no units ran) the context keeps the zero-filled summary forever, even though the executor already publishedstatus: "success"at line 180 for thetotalItems === 0case, but there's no such publish for thetotalItems > 0 && no outgoingcase. The node will look "in progress" indefinitely.
Consider:
- Emitting
status: "success"(with zeros) when there's no downstream chain, or - Not pre-writing
processed/failedhere so the final write is authoritative and the absence is obvious during debugging.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/loop-over-items/executor.ts` around lines
164 - 211, The executor is pre-writing processed/failed zeros into context and
publishing them, which can be read as stale before the actual loop runs; remove
the early "processed" and "failed" fields from both the initial
loopOverItemsChannel().status publish and the object you inject into context via
withNodeRoute (leave mode, totalItems, totalUnits, delayBetweenItemsMs,
continueOnItemError only), so the final write performed by the loop
orchestration remains authoritative for processed/failed on
plan.outputVariableName; keep the existing success publish for plan.totalItems
=== 0 as-is.
| const nodeStatus = useNodeStatus({ | ||
| nodeId: props.id, | ||
| channel: LOOP_OVER_ITEMS_CHANNEL_NAME, | ||
| topic: "status", | ||
| refreshToken: fetchLoopOverItemsRealtimeToken, | ||
| }); | ||
| const { data } = useInngestSubscription({ | ||
| refreshToken: fetchLoopOverItemsRealtimeToken, | ||
| enabled: true, | ||
| }); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether useNodeStatus already subscribes internally and exposes message data.
rg -nP -C5 'export function useNodeStatus' --type=ts
rg -nP -C3 'useInngestSubscription' --type=ts src/features/executions/hooks/Repository: newbie-del/FlowForge
Length of output: 1907
🏁 Script executed:
# Get the complete useNodeStatus hook implementation
cat src/features/executions/hooks/use-node-status.tsRepository: newbie-del/FlowForge
Length of output: 2076
🏁 Script executed:
# Get the loop-over-items/node.tsx to see the actual usage
cat src/features/executions/components/loop-over-items/node.tsxRepository: newbie-del/FlowForge
Length of output: 4065
Double subscription to the same realtime channel.
useNodeStatus internally calls useInngestSubscription({ refreshToken, enabled: true }) and filters the stream by channel, topic, and nodeId to extract the status. The component then calls useInngestSubscription separately with the same refreshToken, creating a second subscription to the same stream just to extract progress metrics (processed, totalItems, failed) from the same filtered messages.
Consolidate to a single subscription: either have useNodeStatus return the full message data so the component can extract both status and progress metrics, or expose a separate hook for progress metrics that reuses the same subscription internally.
Minor: The sort comparators in both useNodeStatus and the progressLabel computation check kind === "data" after the filter already ensures only data messages remain—this is redundant but harmless.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/executions/components/loop-over-items/node.tsx` around lines 30
- 39, The component is creating two realtime subscriptions: useNodeStatus(...)
already calls useInngestSubscription internally, and the component separately
calls useInngestSubscription(...) causing duplicate subscriptions; instead
modify useNodeStatus to return the full message payload (including progress
fields processed, totalItems, failed) or add a dedicated hook that reuses its
internal subscription, then remove the second useInngestSubscription call in the
component (the variables data and progressLabel should be derived from the
returned nodeStatus payload). Update references to LOOP_OVER_ITEMS_CHANNEL_NAME
and fetchLoopOverItemsRealtimeToken in the adjusted hook so the same refresh
token/channel are used, and remove redundant kind === "data" checks in the
sort/comparator logic if desired.
| private validateCodeNode(node: AiWorkflowNode): void { | ||
| const code = String(node.data.code ?? "").trim(); | ||
| if (!code) { | ||
| this.addError( | ||
| "error", | ||
| node.id, | ||
| "code", | ||
| "Code node requires JavaScript code", | ||
| ); | ||
| } | ||
| const timeout = Number(node.data.timeoutMs ?? 3000); | ||
| if (!Number.isFinite(timeout) || timeout < 250 || timeout > 10000) { | ||
| this.addError( | ||
| "warning", | ||
| node.id, | ||
| "timeoutMs", | ||
| "Code timeout should be between 250 and 10000 ms", | ||
| ); | ||
| } | ||
| } |
There was a problem hiding this comment.
Severity mismatch with schema: timeoutMs out-of-range should be an error.
nodeInputSchemas[NodeType.CODE] (node-schemas.ts) declares timeoutMs: z.number().int().min(250).max(10000).optional(), which will reject values outside [250, 10000] during schema validation. Emitting only a "warning" here contradicts that contract — a CODE node with timeoutMs: 50 would pass WorkflowValidator with just a warning yet fail the Zod schema later.
Consider raising this to "error", or loosening the schema if warnings are the intended behavior.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/workflows/lib/workflow-validator.ts` around lines 537 - 556, The
validateCodeNode method currently emits a "warning" for out-of-range timeoutMs
which contradicts nodeInputSchemas[NodeType.CODE] (timeoutMs min 250 max 10000)
that would reject such values; update validateCodeNode to call this.addError
with severity "error" for the timeoutMs check (instead of "warning") so the
validator matches the Zod schema contract for timeoutMs, or if you prefer
warnings change the schema to optional/loosen bounds—adjust the code in
validateCodeNode where timeout is validated and the addError call for
"timeoutMs".
| if ( | ||
| node.type === NodeType.MERGE && | ||
| !String(node.data.inputBPath ?? "").trim() | ||
| ) { | ||
| push({ | ||
| nodeId: node.id, | ||
| field: "inputBPath", | ||
| question: "What should be used as Merge input B?", | ||
| whyItMatters: "Merge requires two inputs to combine branch data.", | ||
| }); | ||
| } |
There was a problem hiding this comment.
Also flag missing inputAPath on MERGE — executor requires both.
mergeExecutor throws "MERGE node input A is missing." when inputAPath resolves to undefined, but computeMissingInputs only asks about inputBPath. Add the symmetric check so users aren't blindsided at runtime.
🔧 Proposed fix
if (
node.type === NodeType.MERGE &&
+ !String(node.data.inputAPath ?? "").trim()
+ ) {
+ push({
+ nodeId: node.id,
+ field: "inputAPath",
+ question: "What should be used as Merge input A?",
+ whyItMatters: "Merge requires two inputs to combine branch data.",
+ });
+ }
+
+ if (
+ node.type === NodeType.MERGE &&
!String(node.data.inputBPath ?? "").trim()
) {
push({🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/workflows/server/ai-builder.ts` around lines 1531 - 1541,
computeMissingInputs currently only checks for missing inputBPath on MERGE
nodes, but mergeExecutor can throw "MERGE node input A is missing." Add a
symmetric check for inputAPath in the same block that checks node.type ===
NodeType.MERGE: if String(node.data.inputAPath ?? "").trim() is falsy, call
push({ nodeId: node.id, field: "inputAPath", question: "What should be used as
Merge input A?", whyItMatters: "Merge requires two inputs to combine branch
data." }); so both inputAPath and inputBPath are prompted before execution.
| if ( | ||
| normalizedPrompt.includes("merge") || | ||
| normalizedPrompt.includes("combine branches") || | ||
| normalizedPrompt.includes("combine outputs") | ||
| ) { | ||
| pushNode({ | ||
| id: makeId("merge"), | ||
| type: NodeType.MERGE, | ||
| title: "Merge", | ||
| description: "Merge branch outputs", | ||
| data: { | ||
| mode: "combine_objects", | ||
| keyField: "", | ||
| conflictStrategy: "prefer_b", | ||
| inputAPath: "sourceA", | ||
| inputBPath: "sourceB", | ||
| outputVariableName: "merged", | ||
| }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
MERGE fallback produces a non-executable node.
Two issues on this branch of buildFallbackPlan:
inputAPath: "sourceA"/inputBPath: "sourceB"are literal placeholder strings.mergeExecutor.resolveInputwill treat them as paths intocontext, both resolve toundefined, and execution fails with"MERGE node input A is missing."every time.- The sequential edge builder at lines 2018‑2025 only wires
prev → mergeas a single connection. The MERGE node needs two distinct upstream sources (target-a,target-b). A fallback that appends MERGE without a branched upstream will never produce two inputs.
Either drop MERGE from fallback generation, or only append it when there are already two independent upstream branches to wire to target-a/target-b, and set inputAPath/inputBPath to template references (e.g. {{upstreamA.variableName}}).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/features/workflows/server/ai-builder.ts` around lines 1860 - 1879, The
MERGE fallback branch in buildFallbackPlan creates a non-executable MERGE node
(NodeType.MERGE) because inputAPath/inputBPath are literal placeholders and no
second upstream is wired; fix by only adding the MERGE node when there are two
independent upstream branches available, wire both upstreams to the MERGE node
using distinct edge targets (e.g., "target-a" and "target-b"), and set
data.inputAPath and data.inputBPath to template references pointing at the
upstream outputs (e.g., "{{upstreamA.variableName}}",
"{{upstreamB.variableName}}") so mergeExecutor.resolveInput can find actual
values; alternatively remove the MERGE pushNode call from buildFallbackPlan if
two upstreams cannot be guaranteed.
| export const CODE_NODE_CHANNEL_NAME = "code-node-execution"; | ||
|
|
||
| export const codeNodeChannel = channel(CODE_NODE_CHANNEL_NAME).addTopic( | ||
| topic("status").type<{ | ||
| nodeId: string; | ||
| status: "loading" | "success" | "error"; | ||
| }>(), |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect code-node realtime publishing/subscription for auth and channel scoping.
# Expectation: token issuance checks auth/ownership and publisher/subscriber use a user/workflow/execution-scoped channel.
rg -n -C4 'CODE_NODE_CHANNEL_NAME|codeNodeChannel|getSubscriptionToken|fetchCodeNodeRealtimeToken|codeNodeChannel\(\)\.status' --type ts
rg -n -C4 '\bauth\(|getServerSession|currentUser|session|userId|workflowId|executionId' --type tsRepository: newbie-del/FlowForge
Length of output: 50377
Add ownership validation to the fetchCodeNodeRealtimeToken() function.
The fetchCodeNodeRealtimeToken() in src/features/executions/components/code/actions.ts issues a token without verifying the authenticated user owns the execution. Since codeNodeChannel() is global ("code-node-execution") and scoped only to ["status"], any token holder can observe status events for all code-node executions in the system, not just their own. Before issuing the token, verify that the current user owns the execution being monitored (fetch session via auth.api.getSession() and validate ownership against the execution record).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/inngest/channels/code-node.ts` around lines 3 - 9,
fetchCodeNodeRealtimeToken currently issues a global token for codeNodeChannel
without verifying the requesting user owns the execution; update
fetchCodeNodeRealtimeToken in src/features/executions/components/code/actions.ts
to call auth.api.getSession(), load the execution record (e.g., via the same
execution service/DB code used elsewhere), and compare the session user id to
the execution.ownerId (or equivalent) before creating the token; if the user is
not the owner, throw an authorization error and do not return a token. Ensure
you reference the same channel/topic usage (CODE_NODE_CHANNEL_NAME /
codeNodeChannel / "status") when creating the token so the validation precedes
token issuance.
| try { | ||
| for (const chainNodeId of linearChain) { | ||
| const chainNode = nodeById.get(chainNodeId); | ||
| if (!chainNode) continue; | ||
| const chainExecutor = getExecutor(chainNode.type as NodeType); | ||
| const runtimeLoopNodeId = `${chainNode.id}__loop_${unitIndex + 1}`; | ||
| loopContext = await chainExecutor({ | ||
| data: chainNode.data as Record<string, unknown>, | ||
| nodeId: runtimeLoopNodeId, | ||
| userId, | ||
| context: loopContext, | ||
| step, | ||
| publish, | ||
| }); | ||
| } | ||
| processed += unit.items.length; | ||
| await publishLoopProgress("loading"); | ||
| } catch (error) { | ||
| failed += unit.items.length; | ||
| await publishLoopProgress("loading"); | ||
| if (!loopPlan.continueOnItemError) { | ||
| throw error; | ||
| } | ||
| errors.push( | ||
| error instanceof Error | ||
| ? `Unit ${unitIndex + 1}: ${error.message}` | ||
| : `Unit ${unitIndex + 1}: loop unit failed`, | ||
| ); | ||
| } | ||
| }; |
There was a problem hiding this comment.
Batch failure accounting is all-or-nothing and loses the original error.
For batch mode, a single chain node throwing marks every item in that batch as failed (failed += unit.items.length) and every item as processed on success. A 50-item batch with one bad record reports 50 failed, which will be misleading when users look at the failed totals or the realtime progress UI.
Additionally, on error in sequential/batch non-continue mode (line 334-336), the error is re-thrown but the errors: string[] array is not populated — the outer catch at 371-374 publishes "error" status but the loop summary written at line 376-386 is never reached, so the per-unit error context is lost from context[loopPlan.outputVariableName].
Not blocking for v1, but worth filing a follow-up so failures are attributable to specific items/units.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@src/inngest/functions.ts` around lines 314 - 343, The batch accounting
currently treats a thrown error as failing the entire unit (failed +=
unit.items.length) and on non-continue mode rethrows without populating errors,
losing per-item context; change the logic inside the try/catch around the
linearChain execution so that you track failures at item granularity: when a
chainExecutor throws for a specific runtimeLoopNodeId (and unitIndex), increment
failed by 1 (not unit.items.length) and only increment processed when that
specific item completes successfully; also, before rethrowing when
loopPlan.continueOnItemError === false, push a descriptive entry into errors
(e.g., `Unit ${unitIndex + 1}: ${error.message}` or similar) so the outer catch
and the final loop summary can include the per-item error, and keep existing
publishLoopProgress calls (publishLoopProgress("loading")) unchanged.
Summary by CodeRabbit