Compare commits

..

42 Commits

Author SHA1 Message Date
Waleed
e9c4251c1c v0.5.68: router block reasoning, executor improvements, variable resolution consolidation, helm updates (#2946)
* improvement(workflow-item): stabilize avatar layout and fix name truncation (#2939)

* improvement(workflow-item): stabilize avatar layout and fix name truncation

* fix(avatars): revert overflow bg to hardcoded color for contrast

* fix(executor): stop parallel execution when block errors (#2940)

* improvement(helm): add per-deployment extraVolumes support (#2942)

* fix(gmail): expose messageId field in read email block (#2943)

* fix(resolver): consolidate reference resolution  (#2941)

* fix(resolver): consolidate code to resolve references

* fix edge cases

* use already formatted error

* fix multi index

* fix backwards compat reachability

* handle backwards compatibility accurately

* use shared constant correctly

* feat(router): expose reasoning output in router v2 block (#2945)

* fix(copilot): always allow, credential masking (#2947)

* Fix always allow, credential validation

* Credential masking

* Autoload

* fix(executor): handle condition dead-end branches in loops (#2944)

---------

Co-authored-by: Vikhyath Mondreti <vikhyathvikku@gmail.com>
Co-authored-by: Siddharth Ganesan <33737564+Sg312@users.noreply.github.com>
2026-01-22 13:48:15 -08:00
Waleed
748793e07d fix(executor): handle condition dead-end branches in loops (#2944) 2026-01-22 13:30:11 -08:00
Siddharth Ganesan
91da7e183a fix(copilot): always allow, credential masking (#2947)
* Fix always allow, credential validation

* Credential masking

* Autoload
2026-01-22 13:07:16 -08:00
Waleed
ab09a5ad23 feat(router): expose reasoning output in router v2 block (#2945) 2026-01-22 12:43:57 -08:00
Vikhyath Mondreti
fcd0240db6 fix(resolver): consolidate reference resolution (#2941)
* fix(resolver): consolidate code to resolve references

* fix edge cases

* use already formatted error

* fix multi index

* fix backwards compat reachability

* handle backwards compatibility accurately

* use shared constant correctly
2026-01-22 12:38:50 -08:00
Waleed
4e4149792a fix(gmail): expose messageId field in read email block (#2943) 2026-01-22 11:46:34 -08:00
Waleed
9a8b591257 improvement(helm): add per-deployment extraVolumes support (#2942) 2026-01-22 11:35:23 -08:00
Waleed
f3ae3f8442 fix(executor): stop parallel execution when block errors (#2940) 2026-01-22 11:34:40 -08:00
Waleed
cc2be33d6b v0.5.67: loading, password reset, ui improvements, helm updates (#2928)
* fix(zustand): updated to useShallow from deprecated createWithEqualityFn (#2919)

* fix(logger): use direct env access for webpack inlining (#2920)

* fix(notifications): text overflow with line-clamp (#2921)

* chore(helm): add env vars for Vertex AI, orgs, and telemetry (#2922)

* fix(auth): improve reset password flow and consolidate brand detection (#2924)

* fix(auth): improve reset password flow and consolidate brand detection

* fix(auth): set errorHandled for EMAIL_NOT_VERIFIED to prevent duplicate error

* fix(auth): clear success message on login errors

* chore(auth): fix import order per lint

* fix(action-bar): duplicate subflows with children (#2923)

* fix(action-bar): duplicate subflows with children

* fix(action-bar): add validateTriggerPaste for subflow duplicate

* fix(resolver): agent response format, input formats, root level (#2925)

* fix(resolvers): agent response format, input formats, root level

* fix response block initial seeding

* fix tests

* fix(messages-input): fix cursor alignment and auto-resize with overlay (#2926)

* fix(messages-input): fix cursor alignment and auto-resize with overlay

* fixed remaining zustand warnings

* fix(stores): remove dead code causing log spam on startup (#2927)

* fix(stores): remove dead code causing log spam on startup

* fix(stores): replace custom tools zustand store with react query cache

* improvement(ui): use BrandedButton and BrandedLink components (#2930)

- Refactor auth forms to use BrandedButton component
- Add BrandedLink component for changelog page
- Reduce code duplication in login, signup, reset-password forms
- Update star count default value

* fix(custom-tools): remove unsafe title fallback in getCustomTool (#2929)

* fix(custom-tools): remove unsafe title fallback in getCustomTool

* fix(custom-tools): restore title fallback in getCustomTool lookup

Custom tools are referenced by title (custom_${title}), not database ID.
The title fallback is required for client-side tool resolution to work.

* fix(null-bodies): empty bodies handling (#2931)

* fix(null-statuses): empty bodies handling

* address bugbot comment

* fix(token-refresh): microsoft, notion, x, linear (#2933)

* fix(microsoft): proactive refresh needed

* fix(x): missing token refresh flag

* notion and linear missing flag too

* address bugbot comment

* fix(auth): handle EMAIL_NOT_VERIFIED in onError callback (#2932)

* fix(auth): handle EMAIL_NOT_VERIFIED in onError callback

* refactor(auth): extract redirectToVerify helper to reduce duplication

* fix(workflow-selector): use dedicated selector for workflow dropdown (#2934)

* feat(workflow-block): preview (#2935)

* improvement(copilot): tool configs to show nested props (#2936)

* fix(auth): add genericOAuth providers to trustedProviders (#2937)

---------

Co-authored-by: Vikhyath Mondreti <vikhyathvikku@gmail.com>
Co-authored-by: Emir Karabeg <78010029+emir-karabeg@users.noreply.github.com>
2026-01-21 22:53:25 -08:00
Vikhyath Mondreti
45371e521e v0.5.66: external http requests fix, ring highlighting 2026-01-21 02:55:39 -08:00
Waleed
0ce0f98aa5 v0.5.65: gemini updates, textract integration, ui updates (#2909)
* fix(google): wrap primitive tool responses for Gemini API compatibility (#2900)

* fix(canonical): copilot path + update parent (#2901)

* fix(rss): add top-level title, link, pubDate fields to RSS trigger output (#2902)

* fix(rss): add top-level title, link, pubDate fields to RSS trigger output

* fix(imap): add top-level fields to IMAP trigger output

* improvement(browseruse): add profile id param (#2903)

* improvement(browseruse): add profile id param

* make request a stub since we have directExec

* improvement(executor): upgraded abort controller to handle aborts for loops and parallels (#2880)

* improvement(executor): upgraded abort controller to handle aborts for loops and parallels

* comments

* improvement(files): update execution for passing base64 strings (#2906)

* progress

* improvement(execution): update execution for passing base64 strings

* fix types

* cleanup comments

* path security vuln

* reject promise correctly

* fix redirect case

* remove proxy routes

* fix tests

* use ipaddr

* feat(tools): added textract, added v2 for mistral, updated tag dropdown (#2904)

* feat(tools): added textract

* cleanup

* ack pr comments

* reorder

* removed upload for textract async version

* fix additional fields dropdown in editor, update parser to leave validation to be done on the server

* added mistral v2, files v2, and finalized textract

* updated the rest of the old file patterns, updated mistral outputs for v2

* updated tag dropdown to parse non-operation fields as well

* updated extension finder

* cleanup

* added description for inputs to workflow

* use helper for internal route check

* fix tag dropdown merge conflict change

* remove duplicate code

---------

Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>

* fix(ui): change add inputs button to match output selector (#2907)

* fix(canvas): removed invite to workspace from canvas popover (#2908)

* fix(canvas): removed invite to workspace

* removed unused props

* fix(copilot): legacy tool display names (#2911)

* fix(a2a): canonical merge  (#2912)

* fix canonical merge

* fix empty array case

* fix(change-detection): copilot diffs have extra field (#2913)

* improvement(logs): improved logs ui bugs, added subflow disable UI (#2910)

* improvement(logs): improved logs ui bugs, added subflow disable UI

* added duplicate to action bar for subflows

* feat(broadcast): email v0.5 (#2905)

---------

Co-authored-by: Vikhyath Mondreti <vikhyathvikku@gmail.com>
Co-authored-by: Vikhyath Mondreti <vikhyath@simstudio.ai>
Co-authored-by: Emir Karabeg <78010029+emir-karabeg@users.noreply.github.com>
2026-01-20 23:54:55 -08:00
Waleed
dff1c9d083 v0.5.64: unsubscribe, search improvements, metrics, additional SSO configuration 2026-01-20 00:34:11 -08:00
Vikhyath Mondreti
b09f683072 v0.5.63: ui and performance improvements, more google tools 2026-01-18 15:22:42 -08:00
Vikhyath Mondreti
a8bb0db660 v0.5.62: webhook bug fixes, seeding default subblock values, block selection fixes 2026-01-16 20:27:06 -08:00
Waleed
af82820a28 v0.5.61: webhook improvements, workflow controls, react query for deployment status, chat fixes, reducto and pulse OCR, linear fixes 2026-01-16 18:06:23 -08:00
Waleed
4372841797 v0.5.60: invitation flow improvements, chat fixes, a2a improvements, additional copilot actions 2026-01-15 00:02:18 -08:00
Waleed
5e8c843241 v0.5.59: a2a support, documentation 2026-01-13 13:21:21 -08:00
Waleed
7bf3d73ee6 v0.5.58: export folders, new tools, permissions groups enhancements 2026-01-13 00:56:59 -08:00
Vikhyath Mondreti
7ffc11a738 v0.5.57: subagents, context menu improvements, bug fixes 2026-01-11 11:38:40 -08:00
Waleed
be578e2ed7 v0.5.56: batch operations, access control and permission groups, billing fixes 2026-01-10 00:31:34 -08:00
Waleed
f415e5edc4 v0.5.55: polling groups, bedrock provider, devcontainer fixes, workflow preview enhancements 2026-01-08 23:36:56 -08:00
Waleed
13a6e6c3fa v0.5.54: seo, model blacklist, helm chart updates, fireflies integration, autoconnect improvements, billing fixes 2026-01-07 16:09:45 -08:00
Waleed
f5ab7f21ae v0.5.53: hotkey improvements, added redis fallback, fixes for workflow tool 2026-01-06 23:34:52 -08:00
Waleed
bfb6fffe38 v0.5.52: new port-based router block, combobox expression and variable support 2026-01-06 16:14:10 -08:00
Waleed
4fbec0a43f v0.5.51: triggers, kb, condition block improvements, supabase and grain integration updates 2026-01-06 14:26:46 -08:00
Waleed
585f5e365b v0.5.50: import improvements, ui upgrades, kb styling and performance improvements 2026-01-05 00:35:55 -08:00
Waleed
3792bdd252 v0.5.49: hitl improvements, new email styles, imap trigger, logs context menu (#2672)
* feat(logs-context-menu): consolidated logs utils and types, added logs record context menu (#2659)

* feat(email): welcome email; improvement(emails): ui/ux (#2658)

* feat(email): welcome email; improvement(emails): ui/ux

* improvement(emails): links, accounts, preview

* refactor(emails): file structure and wrapper components

* added envvar for personal emails sent, added isHosted gate

* fixed failing tests, added env mock

* fix: removed comment

---------

Co-authored-by: waleed <walif6@gmail.com>

* fix(logging): hitl + trigger dev crash protection (#2664)

* hitl gaps

* deal with trigger worker crashes

* cleanup import strcuture

* feat(imap): added support for imap trigger (#2663)

* feat(tools): added support for imap trigger

* feat(imap): added parity, tested

* ack PR comments

* final cleanup

* feat(i18n): update translations (#2665)

Co-authored-by: waleedlatif1 <waleedlatif1@users.noreply.github.com>

* fix(grain): updated grain trigger to auto-establish trigger (#2666)

Co-authored-by: aadamgough <adam@sim.ai>

* feat(admin): routes to manage deployments (#2667)

* feat(admin): routes to manage deployments

* fix naming fo deployed by

* feat(time-picker): added timepicker emcn component, added to playground, added searchable prop for dropdown, added more timezones for schedule, updated license and notice date (#2668)

* feat(time-picker): added timepicker emcn component, added to playground, added searchable prop for dropdown, added more timezones for schedule, updated license and notice date

* removed unused params, cleaned up redundant utils

* improvement(invite): aligned styling (#2669)

* improvement(invite): aligned with rest of app

* fix(invite): error handling

* fix: addressed comments

---------

Co-authored-by: Emir Karabeg <78010029+emir-karabeg@users.noreply.github.com>
Co-authored-by: Vikhyath Mondreti <vikhyathvikku@gmail.com>
Co-authored-by: waleedlatif1 <waleedlatif1@users.noreply.github.com>
Co-authored-by: Adam Gough <77861281+aadamgough@users.noreply.github.com>
Co-authored-by: aadamgough <adam@sim.ai>
2026-01-03 13:19:18 -08:00
Waleed
eb5d1f3e5b v0.5.48: copy-paste workflow blocks, docs updates, mcp tool fixes 2025-12-31 18:00:04 -08:00
Waleed
54ab82c8dd v0.5.47: deploy workflow as mcp, kb chunks tokenizer, UI improvements, jira service management tools 2025-12-30 23:18:58 -08:00
Waleed
f895bf469b v0.5.46: build improvements, greptile, light mode improvements 2025-12-29 02:17:52 -08:00
Waleed
dd3209af06 v0.5.45: light mode fixes, realtime usage indicator, docker build improvements 2025-12-27 19:57:42 -08:00
Waleed
b6ba3b50a7 v0.5.44: keyboard shortcuts, autolayout, light mode, byok, testing improvements 2025-12-26 21:25:19 -08:00
Waleed
b304233062 v0.5.43: export logs, circleback, grain, vertex, code hygiene, schedule improvements 2025-12-23 19:19:18 -08:00
Vikhyath Mondreti
57e4b49bd6 v0.5.42: fix memory migration 2025-12-23 01:24:54 -08:00
Vikhyath Mondreti
e12dd204ed v0.5.41: memory fixes, copilot improvements, knowledgebase improvements, LLM providers standardization 2025-12-23 00:15:18 -08:00
Vikhyath Mondreti
3d9d9cbc54 v0.5.40: supabase ops to allow non-public schemas, jira uuid 2025-12-21 22:28:05 -08:00
Waleed
0f4ec962ad v0.5.39: notion, workflow variables fixes 2025-12-20 20:44:00 -08:00
Waleed
4827866f9a v0.5.38: snap to grid, copilot ux improvements, billing line items 2025-12-20 17:24:38 -08:00
Waleed
3e697d9ed9 v0.5.37: redaction utils consolidation, logs updates, autoconnect improvements, additional kb tag types 2025-12-19 22:31:55 -08:00
Martin Yankov
4431a1a484 fix(helm): add custom egress rules to realtime network policy (#2481)
The realtime service network policy was missing the custom egress rules section
that allows configuration of additional egress rules via values.yaml. This caused
the realtime pods to be unable to connect to external databases (e.g., PostgreSQL
on port 5432) when using external database configurations.

The app network policy already had this section, but the realtime network policy
was missing it, creating an inconsistency and preventing the realtime service
from accessing external databases configured via networkPolicy.egress values.

This fix adds the same custom egress rules template section to the realtime
network policy, matching the app network policy behavior and allowing users to
configure database connectivity via values.yaml.
2025-12-19 18:59:08 -08:00
Waleed
4d1a9a3f22 v0.5.36: hitl improvements, opengraph, slack fixes, one-click unsubscribe, auth checks, new db indexes 2025-12-19 01:27:49 -08:00
Vikhyath Mondreti
eb07a080fb v0.5.35: helm updates, copilot improvements, 404 for docs, salesforce fixes, subflow resize clamping 2025-12-18 16:23:19 -08:00
32 changed files with 2923 additions and 432 deletions

View File

@@ -313,7 +313,7 @@ describe('Function Execute API Route', () => {
'block-2': 'world', 'block-2': 'world',
}, },
blockNameMapping: { blockNameMapping: {
validVar: 'block-1', validvar: 'block-1',
another_valid: 'block-2', another_valid: 'block-2',
}, },
}) })
@@ -539,7 +539,7 @@ describe('Function Execute API Route', () => {
'block-complex': complexData, 'block-complex': complexData,
}, },
blockNameMapping: { blockNameMapping: {
complexData: 'block-complex', complexdata: 'block-complex',
}, },
}) })

View File

@@ -6,11 +6,11 @@ import { executeInE2B } from '@/lib/execution/e2b'
import { executeInIsolatedVM } from '@/lib/execution/isolated-vm' import { executeInIsolatedVM } from '@/lib/execution/isolated-vm'
import { CodeLanguage, DEFAULT_CODE_LANGUAGE, isValidCodeLanguage } from '@/lib/execution/languages' import { CodeLanguage, DEFAULT_CODE_LANGUAGE, isValidCodeLanguage } from '@/lib/execution/languages'
import { escapeRegExp, normalizeName, REFERENCE } from '@/executor/constants' import { escapeRegExp, normalizeName, REFERENCE } from '@/executor/constants'
import { type OutputSchema, resolveBlockReference } from '@/executor/utils/block-reference'
import { import {
createEnvVarPattern, createEnvVarPattern,
createWorkflowVariablePattern, createWorkflowVariablePattern,
} from '@/executor/utils/reference-validation' } from '@/executor/utils/reference-validation'
import { navigatePath } from '@/executor/variables/resolvers/reference'
export const dynamic = 'force-dynamic' export const dynamic = 'force-dynamic'
export const runtime = 'nodejs' export const runtime = 'nodejs'
@@ -470,14 +470,17 @@ function resolveEnvironmentVariables(
function resolveTagVariables( function resolveTagVariables(
code: string, code: string,
blockData: Record<string, any>, blockData: Record<string, unknown>,
blockNameMapping: Record<string, string>, blockNameMapping: Record<string, string>,
contextVariables: Record<string, any> blockOutputSchemas: Record<string, OutputSchema>,
contextVariables: Record<string, unknown>,
language = 'javascript'
): string { ): string {
let resolvedCode = code let resolvedCode = code
const undefinedLiteral = language === 'python' ? 'None' : 'undefined'
const tagPattern = new RegExp( const tagPattern = new RegExp(
`${REFERENCE.START}([a-zA-Z_][a-zA-Z0-9_${REFERENCE.PATH_DELIMITER}]*[a-zA-Z0-9_])${REFERENCE.END}`, `${REFERENCE.START}([a-zA-Z_](?:[a-zA-Z0-9_${REFERENCE.PATH_DELIMITER}]*[a-zA-Z0-9_])?)${REFERENCE.END}`,
'g' 'g'
) )
const tagMatches = resolvedCode.match(tagPattern) || [] const tagMatches = resolvedCode.match(tagPattern) || []
@@ -486,41 +489,37 @@ function resolveTagVariables(
const tagName = match.slice(REFERENCE.START.length, -REFERENCE.END.length).trim() const tagName = match.slice(REFERENCE.START.length, -REFERENCE.END.length).trim()
const pathParts = tagName.split(REFERENCE.PATH_DELIMITER) const pathParts = tagName.split(REFERENCE.PATH_DELIMITER)
const blockName = pathParts[0] const blockName = pathParts[0]
const fieldPath = pathParts.slice(1)
const blockId = blockNameMapping[blockName] const result = resolveBlockReference(blockName, fieldPath, {
if (!blockId) { blockNameMapping,
blockData,
blockOutputSchemas,
})
if (!result) {
continue continue
} }
const blockOutput = blockData[blockId] let tagValue = result.value
if (blockOutput === undefined) {
continue
}
let tagValue: any
if (pathParts.length === 1) {
tagValue = blockOutput
} else {
tagValue = navigatePath(blockOutput, pathParts.slice(1))
}
if (tagValue === undefined) { if (tagValue === undefined) {
resolvedCode = resolvedCode.replace(new RegExp(escapeRegExp(match), 'g'), undefinedLiteral)
continue continue
} }
if ( if (typeof tagValue === 'string') {
typeof tagValue === 'string' && const trimmed = tagValue.trimStart()
tagValue.length > 100 && if (trimmed.startsWith('{') || trimmed.startsWith('[')) {
(tagValue.startsWith('{') || tagValue.startsWith('[')) try {
) { tagValue = JSON.parse(tagValue)
try { } catch {
tagValue = JSON.parse(tagValue) // Keep as string if not valid JSON
} catch { }
// Keep as-is
} }
} }
const safeVarName = `__tag_${tagName.replace(/[^a-zA-Z0-9_]/g, '_')}` const safeVarName = `__tag_${tagName.replace(/_/g, '_1').replace(/\./g, '_0')}`
contextVariables[safeVarName] = tagValue contextVariables[safeVarName] = tagValue
resolvedCode = resolvedCode.replace(new RegExp(escapeRegExp(match), 'g'), safeVarName) resolvedCode = resolvedCode.replace(new RegExp(escapeRegExp(match), 'g'), safeVarName)
} }
@@ -537,18 +536,27 @@ function resolveTagVariables(
*/ */
function resolveCodeVariables( function resolveCodeVariables(
code: string, code: string,
params: Record<string, any>, params: Record<string, unknown>,
envVars: Record<string, string> = {}, envVars: Record<string, string> = {},
blockData: Record<string, any> = {}, blockData: Record<string, unknown> = {},
blockNameMapping: Record<string, string> = {}, blockNameMapping: Record<string, string> = {},
workflowVariables: Record<string, any> = {} blockOutputSchemas: Record<string, OutputSchema> = {},
): { resolvedCode: string; contextVariables: Record<string, any> } { workflowVariables: Record<string, unknown> = {},
language = 'javascript'
): { resolvedCode: string; contextVariables: Record<string, unknown> } {
let resolvedCode = code let resolvedCode = code
const contextVariables: Record<string, any> = {} const contextVariables: Record<string, unknown> = {}
resolvedCode = resolveWorkflowVariables(resolvedCode, workflowVariables, contextVariables) resolvedCode = resolveWorkflowVariables(resolvedCode, workflowVariables, contextVariables)
resolvedCode = resolveEnvironmentVariables(resolvedCode, params, envVars, contextVariables) resolvedCode = resolveEnvironmentVariables(resolvedCode, params, envVars, contextVariables)
resolvedCode = resolveTagVariables(resolvedCode, blockData, blockNameMapping, contextVariables) resolvedCode = resolveTagVariables(
resolvedCode,
blockData,
blockNameMapping,
blockOutputSchemas,
contextVariables,
language
)
return { resolvedCode, contextVariables } return { resolvedCode, contextVariables }
} }
@@ -585,6 +593,7 @@ export async function POST(req: NextRequest) {
envVars = {}, envVars = {},
blockData = {}, blockData = {},
blockNameMapping = {}, blockNameMapping = {},
blockOutputSchemas = {},
workflowVariables = {}, workflowVariables = {},
workflowId, workflowId,
isCustomTool = false, isCustomTool = false,
@@ -601,20 +610,21 @@ export async function POST(req: NextRequest) {
isCustomTool, isCustomTool,
}) })
// Resolve variables in the code with workflow environment variables const lang = isValidCodeLanguage(language) ? language : DEFAULT_CODE_LANGUAGE
const codeResolution = resolveCodeVariables( const codeResolution = resolveCodeVariables(
code, code,
executionParams, executionParams,
envVars, envVars,
blockData, blockData,
blockNameMapping, blockNameMapping,
workflowVariables blockOutputSchemas,
workflowVariables,
lang
) )
resolvedCode = codeResolution.resolvedCode resolvedCode = codeResolution.resolvedCode
const contextVariables = codeResolution.contextVariables const contextVariables = codeResolution.contextVariables
const lang = isValidCodeLanguage(language) ? language : DEFAULT_CODE_LANGUAGE
let jsImports = '' let jsImports = ''
let jsRemainingCode = resolvedCode let jsRemainingCode = resolvedCode
let hasImports = false let hasImports = false
@@ -670,7 +680,11 @@ export async function POST(req: NextRequest) {
prologue += `const environmentVariables = JSON.parse(${JSON.stringify(JSON.stringify(envVars))});\n` prologue += `const environmentVariables = JSON.parse(${JSON.stringify(JSON.stringify(envVars))});\n`
prologueLineCount++ prologueLineCount++
for (const [k, v] of Object.entries(contextVariables)) { for (const [k, v] of Object.entries(contextVariables)) {
prologue += `const ${k} = JSON.parse(${JSON.stringify(JSON.stringify(v))});\n` if (v === undefined) {
prologue += `const ${k} = undefined;\n`
} else {
prologue += `const ${k} = JSON.parse(${JSON.stringify(JSON.stringify(v))});\n`
}
prologueLineCount++ prologueLineCount++
} }
@@ -741,7 +755,11 @@ export async function POST(req: NextRequest) {
prologue += `environmentVariables = json.loads(${JSON.stringify(JSON.stringify(envVars))})\n` prologue += `environmentVariables = json.loads(${JSON.stringify(JSON.stringify(envVars))})\n`
prologueLineCount++ prologueLineCount++
for (const [k, v] of Object.entries(contextVariables)) { for (const [k, v] of Object.entries(contextVariables)) {
prologue += `${k} = json.loads(${JSON.stringify(JSON.stringify(v))})\n` if (v === undefined) {
prologue += `${k} = None\n`
} else {
prologue += `${k} = json.loads(${JSON.stringify(JSON.stringify(v))})\n`
}
prologueLineCount++ prologueLineCount++
} }
const wrapped = [ const wrapped = [

View File

@@ -151,6 +151,29 @@ export const ActionBar = memo(
</Tooltip.Root> </Tooltip.Root>
)} )}
{isSubflowBlock && (
<Tooltip.Root>
<Tooltip.Trigger asChild>
<Button
variant='ghost'
onClick={(e) => {
e.stopPropagation()
if (!disabled) {
collaborativeBatchToggleBlockEnabled([blockId])
}
}}
className={ACTION_BUTTON_STYLES}
disabled={disabled}
>
{isEnabled ? <Circle className={ICON_SIZE} /> : <CircleOff className={ICON_SIZE} />}
</Button>
</Tooltip.Trigger>
<Tooltip.Content side='top'>
{getTooltipMessage(isEnabled ? 'Disable Block' : 'Enable Block')}
</Tooltip.Content>
</Tooltip.Root>
)}
{!isStartBlock && !isResponseBlock && ( {!isStartBlock && !isResponseBlock && (
<Tooltip.Root> <Tooltip.Root>
<Tooltip.Trigger asChild> <Tooltip.Trigger asChild>
@@ -222,29 +245,6 @@ export const ActionBar = memo(
</Tooltip.Root> </Tooltip.Root>
)} )}
{isSubflowBlock && (
<Tooltip.Root>
<Tooltip.Trigger asChild>
<Button
variant='ghost'
onClick={(e) => {
e.stopPropagation()
if (!disabled) {
collaborativeBatchToggleBlockEnabled([blockId])
}
}}
className={ACTION_BUTTON_STYLES}
disabled={disabled}
>
{isEnabled ? <Circle className={ICON_SIZE} /> : <CircleOff className={ICON_SIZE} />}
</Button>
</Tooltip.Trigger>
<Tooltip.Content side='top'>
{getTooltipMessage(isEnabled ? 'Disable Block' : 'Enable Block')}
</Tooltip.Content>
</Tooltip.Root>
)}
<Tooltip.Root> <Tooltip.Root>
<Tooltip.Trigger asChild> <Tooltip.Trigger asChild>
<Button <Button

View File

@@ -242,15 +242,9 @@ Return ONLY the email body - no explanations, no extra text.`,
id: 'messageId', id: 'messageId',
title: 'Message ID', title: 'Message ID',
type: 'short-input', type: 'short-input',
placeholder: 'Enter message ID to read (optional)', placeholder: 'Read specific email by ID (overrides label/folder)',
condition: { condition: { field: 'operation', value: 'read_gmail' },
field: 'operation', mode: 'advanced',
value: 'read_gmail',
and: {
field: 'folder',
value: '',
},
},
}, },
// Search Fields // Search Fields
{ {

View File

@@ -129,12 +129,9 @@ ROUTING RULES:
3. If the context is even partially related to a route's description, select that route 3. If the context is even partially related to a route's description, select that route
4. ONLY output NO_MATCH if the context is completely unrelated to ALL route descriptions 4. ONLY output NO_MATCH if the context is completely unrelated to ALL route descriptions
OUTPUT FORMAT: Respond with a JSON object containing:
- Output EXACTLY one route ID (copied exactly as shown above) OR "NO_MATCH" - route: EXACTLY one route ID (copied exactly as shown above) OR "NO_MATCH"
- No explanation, no punctuation, no additional text - reasoning: A brief explanation (1-2 sentences) of why you chose this route`
- Just the route ID or NO_MATCH
Your response:`
} }
/** /**
@@ -272,6 +269,7 @@ interface RouterV2Response extends ToolResponse {
total: number total: number
} }
selectedRoute: string selectedRoute: string
reasoning: string
selectedPath: { selectedPath: {
blockId: string blockId: string
blockType: string blockType: string
@@ -355,6 +353,7 @@ export const RouterV2Block: BlockConfig<RouterV2Response> = {
tokens: { type: 'json', description: 'Token usage' }, tokens: { type: 'json', description: 'Token usage' },
cost: { type: 'json', description: 'Cost information' }, cost: { type: 'json', description: 'Cost information' },
selectedRoute: { type: 'string', description: 'Selected route ID' }, selectedRoute: { type: 'string', description: 'Selected route ID' },
reasoning: { type: 'string', description: 'Explanation of why this route was chosen' },
selectedPath: { type: 'json', description: 'Selected routing path' }, selectedPath: { type: 'json', description: 'Selected routing path' },
}, },
} }

View File

@@ -24,6 +24,71 @@ function createBlock(id: string, metadataId: string): SerializedBlock {
} }
} }
describe('DAGBuilder disabled subflow validation', () => {
it('skips validation for disabled loops with no blocks inside', () => {
const workflow: SerializedWorkflow = {
version: '1',
blocks: [
createBlock('start', BlockType.STARTER),
{ ...createBlock('loop-block', BlockType.FUNCTION), enabled: false },
],
connections: [],
loops: {
'loop-1': {
id: 'loop-1',
nodes: [], // Empty loop - would normally throw
iterations: 3,
},
},
}
const builder = new DAGBuilder()
// Should not throw even though loop has no blocks inside
expect(() => builder.build(workflow)).not.toThrow()
})
it('skips validation for disabled parallels with no blocks inside', () => {
const workflow: SerializedWorkflow = {
version: '1',
blocks: [createBlock('start', BlockType.STARTER)],
connections: [],
loops: {},
parallels: {
'parallel-1': {
id: 'parallel-1',
nodes: [], // Empty parallel - would normally throw
},
},
}
const builder = new DAGBuilder()
// Should not throw even though parallel has no blocks inside
expect(() => builder.build(workflow)).not.toThrow()
})
it('skips validation for loops where all inner blocks are disabled', () => {
const workflow: SerializedWorkflow = {
version: '1',
blocks: [
createBlock('start', BlockType.STARTER),
{ ...createBlock('inner-block', BlockType.FUNCTION), enabled: false },
],
connections: [],
loops: {
'loop-1': {
id: 'loop-1',
nodes: ['inner-block'], // Has node but it's disabled
iterations: 3,
},
},
}
const builder = new DAGBuilder()
// Should not throw - loop is effectively disabled since all inner blocks are disabled
expect(() => builder.build(workflow)).not.toThrow()
})
})
describe('DAGBuilder human-in-the-loop transformation', () => { describe('DAGBuilder human-in-the-loop transformation', () => {
it('creates trigger nodes and rewires edges for pause blocks', () => { it('creates trigger nodes and rewires edges for pause blocks', () => {
const workflow: SerializedWorkflow = { const workflow: SerializedWorkflow = {

View File

@@ -136,17 +136,18 @@ export class DAGBuilder {
nodes: string[] | undefined, nodes: string[] | undefined,
type: 'Loop' | 'Parallel' type: 'Loop' | 'Parallel'
): void { ): void {
const sentinelStartId =
type === 'Loop' ? buildSentinelStartId(id) : buildParallelSentinelStartId(id)
const sentinelStartNode = dag.nodes.get(sentinelStartId)
if (!sentinelStartNode) return
if (!nodes || nodes.length === 0) { if (!nodes || nodes.length === 0) {
throw new Error( throw new Error(
`${type} has no blocks inside. Add at least one block to the ${type.toLowerCase()}.` `${type} has no blocks inside. Add at least one block to the ${type.toLowerCase()}.`
) )
} }
const sentinelStartId =
type === 'Loop' ? buildSentinelStartId(id) : buildParallelSentinelStartId(id)
const sentinelStartNode = dag.nodes.get(sentinelStartId)
if (!sentinelStartNode) return
const hasConnections = Array.from(sentinelStartNode.outgoingEdges.values()).some((edge) => const hasConnections = Array.from(sentinelStartNode.outgoingEdges.values()).some((edge) =>
nodes.includes(extractBaseBlockId(edge.target)) nodes.includes(extractBaseBlockId(edge.target))
) )

File diff suppressed because it is too large Load Diff

View File

@@ -20,21 +20,13 @@ export class EdgeManager {
const activatedTargets: string[] = [] const activatedTargets: string[] = []
const edgesToDeactivate: Array<{ target: string; handle?: string }> = [] const edgesToDeactivate: Array<{ target: string; handle?: string }> = []
// First pass: categorize edges as activating or deactivating for (const [, edge] of node.outgoingEdges) {
// Don't modify incomingEdges yet - we need the original state for deactivation checks
for (const [edgeId, edge] of node.outgoingEdges) {
if (skipBackwardsEdge && this.isBackwardsEdge(edge.sourceHandle)) { if (skipBackwardsEdge && this.isBackwardsEdge(edge.sourceHandle)) {
continue continue
} }
const shouldActivate = this.shouldActivateEdge(edge, output) if (!this.shouldActivateEdge(edge, output)) {
if (!shouldActivate) { if (!this.isLoopEdge(edge.sourceHandle)) {
const isLoopEdge =
edge.sourceHandle === EDGE.LOOP_CONTINUE ||
edge.sourceHandle === EDGE.LOOP_CONTINUE_ALT ||
edge.sourceHandle === EDGE.LOOP_EXIT
if (!isLoopEdge) {
edgesToDeactivate.push({ target: edge.target, handle: edge.sourceHandle }) edgesToDeactivate.push({ target: edge.target, handle: edge.sourceHandle })
} }
continue continue
@@ -43,13 +35,19 @@ export class EdgeManager {
activatedTargets.push(edge.target) activatedTargets.push(edge.target)
} }
// Second pass: process deactivations while incomingEdges is still intact const cascadeTargets = new Set<string>()
// This ensures hasActiveIncomingEdges can find all potential sources
for (const { target, handle } of edgesToDeactivate) { for (const { target, handle } of edgesToDeactivate) {
this.deactivateEdgeAndDescendants(node.id, target, handle) this.deactivateEdgeAndDescendants(node.id, target, handle, cascadeTargets)
}
if (activatedTargets.length === 0) {
for (const { target } of edgesToDeactivate) {
if (this.isTerminalControlNode(target)) {
cascadeTargets.add(target)
}
}
} }
// Third pass: update incomingEdges for activated targets
for (const targetId of activatedTargets) { for (const targetId of activatedTargets) {
const targetNode = this.dag.nodes.get(targetId) const targetNode = this.dag.nodes.get(targetId)
if (!targetNode) { if (!targetNode) {
@@ -59,28 +57,25 @@ export class EdgeManager {
targetNode.incomingEdges.delete(node.id) targetNode.incomingEdges.delete(node.id)
} }
// Fourth pass: check readiness after all edge processing is complete
for (const targetId of activatedTargets) { for (const targetId of activatedTargets) {
const targetNode = this.dag.nodes.get(targetId) if (this.isTargetReady(targetId)) {
if (targetNode && this.isNodeReady(targetNode)) {
readyNodes.push(targetId) readyNodes.push(targetId)
} }
} }
for (const targetId of cascadeTargets) {
if (!readyNodes.includes(targetId) && !activatedTargets.includes(targetId)) {
if (this.isTargetReady(targetId)) {
readyNodes.push(targetId)
}
}
}
return readyNodes return readyNodes
} }
isNodeReady(node: DAGNode): boolean { isNodeReady(node: DAGNode): boolean {
if (node.incomingEdges.size === 0) { return node.incomingEdges.size === 0 || this.countActiveIncomingEdges(node) === 0
return true
}
const activeIncomingCount = this.countActiveIncomingEdges(node)
if (activeIncomingCount > 0) {
return false
}
return true
} }
restoreIncomingEdge(targetNodeId: string, sourceNodeId: string): void { restoreIncomingEdge(targetNodeId: string, sourceNodeId: string): void {
@@ -99,13 +94,10 @@ export class EdgeManager {
/** /**
* Clear deactivated edges for a set of nodes (used when restoring loop state for next iteration). * Clear deactivated edges for a set of nodes (used when restoring loop state for next iteration).
* This ensures error/success edges can be re-evaluated on each iteration.
*/ */
clearDeactivatedEdgesForNodes(nodeIds: Set<string>): void { clearDeactivatedEdgesForNodes(nodeIds: Set<string>): void {
const edgesToRemove: string[] = [] const edgesToRemove: string[] = []
for (const edgeKey of this.deactivatedEdges) { for (const edgeKey of this.deactivatedEdges) {
// Edge key format is "sourceId-targetId-handle"
// Check if either source or target is in the nodeIds set
for (const nodeId of nodeIds) { for (const nodeId of nodeIds) {
if (edgeKey.startsWith(`${nodeId}-`) || edgeKey.includes(`-${nodeId}-`)) { if (edgeKey.startsWith(`${nodeId}-`) || edgeKey.includes(`-${nodeId}-`)) {
edgesToRemove.push(edgeKey) edgesToRemove.push(edgeKey)
@@ -118,6 +110,44 @@ export class EdgeManager {
} }
} }
private isTargetReady(targetId: string): boolean {
const targetNode = this.dag.nodes.get(targetId)
return targetNode ? this.isNodeReady(targetNode) : false
}
private isLoopEdge(handle?: string): boolean {
return (
handle === EDGE.LOOP_CONTINUE ||
handle === EDGE.LOOP_CONTINUE_ALT ||
handle === EDGE.LOOP_EXIT
)
}
private isControlEdge(handle?: string): boolean {
return (
handle === EDGE.LOOP_CONTINUE ||
handle === EDGE.LOOP_CONTINUE_ALT ||
handle === EDGE.LOOP_EXIT ||
handle === EDGE.PARALLEL_EXIT
)
}
private isBackwardsEdge(sourceHandle?: string): boolean {
return sourceHandle === EDGE.LOOP_CONTINUE || sourceHandle === EDGE.LOOP_CONTINUE_ALT
}
private isTerminalControlNode(nodeId: string): boolean {
const node = this.dag.nodes.get(nodeId)
if (!node || node.outgoingEdges.size === 0) return false
for (const [, edge] of node.outgoingEdges) {
if (!this.isControlEdge(edge.sourceHandle)) {
return false
}
}
return true
}
private shouldActivateEdge(edge: DAGEdge, output: NormalizedBlockOutput): boolean { private shouldActivateEdge(edge: DAGEdge, output: NormalizedBlockOutput): boolean {
const handle = edge.sourceHandle const handle = edge.sourceHandle
@@ -159,14 +189,12 @@ export class EdgeManager {
} }
} }
private isBackwardsEdge(sourceHandle?: string): boolean {
return sourceHandle === EDGE.LOOP_CONTINUE || sourceHandle === EDGE.LOOP_CONTINUE_ALT
}
private deactivateEdgeAndDescendants( private deactivateEdgeAndDescendants(
sourceId: string, sourceId: string,
targetId: string, targetId: string,
sourceHandle?: string sourceHandle?: string,
cascadeTargets?: Set<string>,
isCascade = false
): void { ): void {
const edgeKey = this.createEdgeKey(sourceId, targetId, sourceHandle) const edgeKey = this.createEdgeKey(sourceId, targetId, sourceHandle)
if (this.deactivatedEdges.has(edgeKey)) { if (this.deactivatedEdges.has(edgeKey)) {
@@ -174,38 +202,46 @@ export class EdgeManager {
} }
this.deactivatedEdges.add(edgeKey) this.deactivatedEdges.add(edgeKey)
const targetNode = this.dag.nodes.get(targetId) const targetNode = this.dag.nodes.get(targetId)
if (!targetNode) return if (!targetNode) return
// Check if target has other active incoming edges if (isCascade && this.isTerminalControlNode(targetId)) {
// Pass the specific edge key being deactivated, not just source ID, cascadeTargets?.add(targetId)
// to handle multiple edges from same source to same target (e.g., condition branches) }
const hasOtherActiveIncoming = this.hasActiveIncomingEdges(targetNode, edgeKey)
if (!hasOtherActiveIncoming) { if (this.hasActiveIncomingEdges(targetNode, edgeKey)) {
for (const [_, outgoingEdge] of targetNode.outgoingEdges) { return
this.deactivateEdgeAndDescendants(targetId, outgoingEdge.target, outgoingEdge.sourceHandle) }
for (const [, outgoingEdge] of targetNode.outgoingEdges) {
if (!this.isControlEdge(outgoingEdge.sourceHandle)) {
this.deactivateEdgeAndDescendants(
targetId,
outgoingEdge.target,
outgoingEdge.sourceHandle,
cascadeTargets,
true
)
} }
} }
} }
/** /**
* Checks if a node has any active incoming edges besides the one being excluded. * Checks if a node has any active incoming edges besides the one being excluded.
* This properly handles the case where multiple edges from the same source go to
* the same target (e.g., multiple condition branches pointing to one block).
*/ */
private hasActiveIncomingEdges(node: DAGNode, excludeEdgeKey: string): boolean { private hasActiveIncomingEdges(node: DAGNode, excludeEdgeKey: string): boolean {
for (const incomingSourceId of node.incomingEdges) { for (const incomingSourceId of node.incomingEdges) {
const incomingNode = this.dag.nodes.get(incomingSourceId) const incomingNode = this.dag.nodes.get(incomingSourceId)
if (!incomingNode) continue if (!incomingNode) continue
for (const [_, incomingEdge] of incomingNode.outgoingEdges) { for (const [, incomingEdge] of incomingNode.outgoingEdges) {
if (incomingEdge.target === node.id) { if (incomingEdge.target === node.id) {
const incomingEdgeKey = this.createEdgeKey( const incomingEdgeKey = this.createEdgeKey(
incomingSourceId, incomingSourceId,
node.id, node.id,
incomingEdge.sourceHandle incomingEdge.sourceHandle
) )
// Skip the specific edge being excluded, but check other edges from same source
if (incomingEdgeKey === excludeEdgeKey) continue if (incomingEdgeKey === excludeEdgeKey) continue
if (!this.deactivatedEdges.has(incomingEdgeKey)) { if (!this.deactivatedEdges.has(incomingEdgeKey)) {
return true return true

View File

@@ -554,6 +554,413 @@ describe('ExecutionEngine', () => {
}) })
}) })
describe('Error handling in execution', () => {
it('should fail execution when a single node throws an error', async () => {
const startNode = createMockNode('start', 'starter')
const errorNode = createMockNode('error-node', 'function')
startNode.outgoingEdges.set('edge1', { target: 'error-node' })
const dag = createMockDAG([startNode, errorNode])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['error-node']
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'error-node') {
throw new Error('Block execution failed')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('Block execution failed')
})
it('should stop parallel branches when one branch throws an error', async () => {
const startNode = createMockNode('start', 'starter')
const parallelNodes = Array.from({ length: 5 }, (_, i) =>
createMockNode(`parallel${i}`, 'function')
)
parallelNodes.forEach((_, i) => {
startNode.outgoingEdges.set(`edge${i}`, { target: `parallel${i}` })
})
const dag = createMockDAG([startNode, ...parallelNodes])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return parallelNodes.map((_, i) => `parallel${i}`)
return []
})
const executedNodes: string[] = []
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
executedNodes.push(nodeId)
if (nodeId === 'parallel0') {
await new Promise((resolve) => setTimeout(resolve, 10))
throw new Error('Parallel branch failed')
}
await new Promise((resolve) => setTimeout(resolve, 100))
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('Parallel branch failed')
})
it('should capture only the first error when multiple parallel branches fail', async () => {
const startNode = createMockNode('start', 'starter')
const parallelNodes = Array.from({ length: 3 }, (_, i) =>
createMockNode(`parallel${i}`, 'function')
)
parallelNodes.forEach((_, i) => {
startNode.outgoingEdges.set(`edge${i}`, { target: `parallel${i}` })
})
const dag = createMockDAG([startNode, ...parallelNodes])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return parallelNodes.map((_, i) => `parallel${i}`)
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'parallel0') {
await new Promise((resolve) => setTimeout(resolve, 10))
throw new Error('First error')
}
if (nodeId === 'parallel1') {
await new Promise((resolve) => setTimeout(resolve, 20))
throw new Error('Second error')
}
if (nodeId === 'parallel2') {
await new Promise((resolve) => setTimeout(resolve, 30))
throw new Error('Third error')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('First error')
})
it('should wait for ongoing executions to complete before throwing error', async () => {
const startNode = createMockNode('start', 'starter')
const fastErrorNode = createMockNode('fast-error', 'function')
const slowNode = createMockNode('slow', 'function')
startNode.outgoingEdges.set('edge1', { target: 'fast-error' })
startNode.outgoingEdges.set('edge2', { target: 'slow' })
const dag = createMockDAG([startNode, fastErrorNode, slowNode])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['fast-error', 'slow']
return []
})
let slowNodeCompleted = false
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'fast-error') {
await new Promise((resolve) => setTimeout(resolve, 10))
throw new Error('Fast error')
}
if (nodeId === 'slow') {
await new Promise((resolve) => setTimeout(resolve, 50))
slowNodeCompleted = true
return { nodeId, output: {}, isFinalOutput: false }
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('Fast error')
expect(slowNodeCompleted).toBe(true)
})
it('should not queue new nodes after an error occurs', async () => {
const startNode = createMockNode('start', 'starter')
const errorNode = createMockNode('error-node', 'function')
const afterErrorNode = createMockNode('after-error', 'function')
startNode.outgoingEdges.set('edge1', { target: 'error-node' })
errorNode.outgoingEdges.set('edge2', { target: 'after-error' })
const dag = createMockDAG([startNode, errorNode, afterErrorNode])
const context = createMockContext()
const queuedNodes: string[] = []
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') {
queuedNodes.push('error-node')
return ['error-node']
}
if (node.id === 'error-node') {
queuedNodes.push('after-error')
return ['after-error']
}
return []
})
const executedNodes: string[] = []
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
executedNodes.push(nodeId)
if (nodeId === 'error-node') {
throw new Error('Node error')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('Node error')
expect(executedNodes).not.toContain('after-error')
})
it('should populate error result with metadata when execution fails', async () => {
const startNode = createMockNode('start', 'starter')
const errorNode = createMockNode('error-node', 'function')
startNode.outgoingEdges.set('edge1', { target: 'error-node' })
const dag = createMockDAG([startNode, errorNode])
const context = createMockContext()
context.blockLogs.push({
blockId: 'start',
blockName: 'Start',
blockType: 'starter',
startedAt: new Date().toISOString(),
endedAt: new Date().toISOString(),
durationMs: 10,
success: true,
})
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['error-node']
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'error-node') {
const error = new Error('Execution failed') as any
error.executionResult = {
success: false,
output: { partial: 'data' },
logs: context.blockLogs,
metadata: context.metadata,
}
throw error
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
try {
await engine.run('start')
expect.fail('Should have thrown')
} catch (error: any) {
expect(error.executionResult).toBeDefined()
expect(error.executionResult.metadata.endTime).toBeDefined()
expect(error.executionResult.metadata.duration).toBeDefined()
}
})
it('should prefer cancellation status over error when both occur', async () => {
const abortController = new AbortController()
const startNode = createMockNode('start', 'starter')
const errorNode = createMockNode('error-node', 'function')
startNode.outgoingEdges.set('edge1', { target: 'error-node' })
const dag = createMockDAG([startNode, errorNode])
const context = createMockContext({ abortSignal: abortController.signal })
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['error-node']
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'error-node') {
abortController.abort()
throw new Error('Node error')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
const result = await engine.run('start')
expect(result.status).toBe('cancelled')
expect(result.success).toBe(false)
})
it('should stop loop iteration when error occurs in loop body', async () => {
const loopStartNode = createMockNode('loop-start', 'loop_sentinel')
loopStartNode.metadata = { isSentinel: true, sentinelType: 'start', loopId: 'loop1' }
const loopBodyNode = createMockNode('loop-body', 'function')
loopBodyNode.metadata = { isLoopNode: true, loopId: 'loop1' }
const loopEndNode = createMockNode('loop-end', 'loop_sentinel')
loopEndNode.metadata = { isSentinel: true, sentinelType: 'end', loopId: 'loop1' }
const afterLoopNode = createMockNode('after-loop', 'function')
loopStartNode.outgoingEdges.set('edge1', { target: 'loop-body' })
loopBodyNode.outgoingEdges.set('edge2', { target: 'loop-end' })
loopEndNode.outgoingEdges.set('loop_continue', {
target: 'loop-start',
sourceHandle: 'loop_continue',
})
loopEndNode.outgoingEdges.set('loop_complete', {
target: 'after-loop',
sourceHandle: 'loop_complete',
})
const dag = createMockDAG([loopStartNode, loopBodyNode, loopEndNode, afterLoopNode])
const context = createMockContext()
let iterationCount = 0
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'loop-start') return ['loop-body']
if (node.id === 'loop-body') return ['loop-end']
if (node.id === 'loop-end') {
iterationCount++
if (iterationCount < 5) return ['loop-start']
return ['after-loop']
}
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'loop-body' && iterationCount >= 2) {
throw new Error('Loop body error on iteration 3')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('loop-start')).rejects.toThrow('Loop body error on iteration 3')
expect(iterationCount).toBeLessThanOrEqual(3)
})
it('should handle error that is not an Error instance', async () => {
const startNode = createMockNode('start', 'starter')
const errorNode = createMockNode('error-node', 'function')
startNode.outgoingEdges.set('edge1', { target: 'error-node' })
const dag = createMockDAG([startNode, errorNode])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['error-node']
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'error-node') {
throw 'String error message'
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
await expect(engine.run('start')).rejects.toThrow('String error message')
})
it('should preserve partial output when error occurs after some blocks complete', async () => {
const startNode = createMockNode('start', 'starter')
const successNode = createMockNode('success', 'function')
const errorNode = createMockNode('error-node', 'function')
startNode.outgoingEdges.set('edge1', { target: 'success' })
successNode.outgoingEdges.set('edge2', { target: 'error-node' })
const dag = createMockDAG([startNode, successNode, errorNode])
const context = createMockContext()
const edgeManager = createMockEdgeManager((node) => {
if (node.id === 'start') return ['success']
if (node.id === 'success') return ['error-node']
return []
})
const nodeOrchestrator = {
executionCount: 0,
executeNode: vi.fn().mockImplementation(async (_ctx: ExecutionContext, nodeId: string) => {
if (nodeId === 'success') {
return { nodeId, output: { successData: 'preserved' }, isFinalOutput: false }
}
if (nodeId === 'error-node') {
throw new Error('Late error')
}
return { nodeId, output: {}, isFinalOutput: false }
}),
handleNodeCompletion: vi.fn(),
} as unknown as MockNodeOrchestrator
const engine = new ExecutionEngine(context, dag, edgeManager, nodeOrchestrator)
try {
await engine.run('start')
expect.fail('Should have thrown')
} catch (error: any) {
// Verify the error was thrown
expect(error.message).toBe('Late error')
// The partial output should be available in executionResult if attached
if (error.executionResult) {
expect(error.executionResult.output).toBeDefined()
}
}
})
})
describe('Cancellation flag behavior', () => { describe('Cancellation flag behavior', () => {
it('should set cancelledFlag when abort signal fires', async () => { it('should set cancelledFlag when abort signal fires', async () => {
const abortController = new AbortController() const abortController = new AbortController()

View File

@@ -25,6 +25,8 @@ export class ExecutionEngine {
private pausedBlocks: Map<string, PauseMetadata> = new Map() private pausedBlocks: Map<string, PauseMetadata> = new Map()
private allowResumeTriggers: boolean private allowResumeTriggers: boolean
private cancelledFlag = false private cancelledFlag = false
private errorFlag = false
private executionError: Error | null = null
private lastCancellationCheck = 0 private lastCancellationCheck = 0
private readonly useRedisCancellation: boolean private readonly useRedisCancellation: boolean
private readonly CANCELLATION_CHECK_INTERVAL_MS = 500 private readonly CANCELLATION_CHECK_INTERVAL_MS = 500
@@ -103,7 +105,7 @@ export class ExecutionEngine {
this.initializeQueue(triggerBlockId) this.initializeQueue(triggerBlockId)
while (this.hasWork()) { while (this.hasWork()) {
if (await this.checkCancellation()) { if ((await this.checkCancellation()) || this.errorFlag) {
break break
} }
await this.processQueue() await this.processQueue()
@@ -113,6 +115,11 @@ export class ExecutionEngine {
await this.waitForAllExecutions() await this.waitForAllExecutions()
} }
// Rethrow the captured error so it's handled by the catch block
if (this.errorFlag && this.executionError) {
throw this.executionError
}
if (this.pausedBlocks.size > 0) { if (this.pausedBlocks.size > 0) {
return this.buildPausedResult(startTime) return this.buildPausedResult(startTime)
} }
@@ -196,11 +203,17 @@ export class ExecutionEngine {
} }
private trackExecution(promise: Promise<void>): void { private trackExecution(promise: Promise<void>): void {
this.executing.add(promise) const trackedPromise = promise
promise.catch(() => {}) .catch((error) => {
promise.finally(() => { if (!this.errorFlag) {
this.executing.delete(promise) this.errorFlag = true
}) this.executionError = error instanceof Error ? error : new Error(String(error))
}
})
.finally(() => {
this.executing.delete(trackedPromise)
})
this.executing.add(trackedPromise)
} }
private async waitForAnyExecution(): Promise<void> { private async waitForAnyExecution(): Promise<void> {
@@ -315,7 +328,7 @@ export class ExecutionEngine {
private async processQueue(): Promise<void> { private async processQueue(): Promise<void> {
while (this.readyQueue.length > 0) { while (this.readyQueue.length > 0) {
if (await this.checkCancellation()) { if ((await this.checkCancellation()) || this.errorFlag) {
break break
} }
const nodeId = this.dequeue() const nodeId = this.dequeue()
@@ -324,7 +337,7 @@ export class ExecutionEngine {
this.trackExecution(promise) this.trackExecution(promise)
} }
if (this.executing.size > 0 && !this.cancelledFlag) { if (this.executing.size > 0 && !this.cancelledFlag && !this.errorFlag) {
await this.waitForAnyExecution() await this.waitForAnyExecution()
} }
} }

View File

@@ -305,7 +305,7 @@ export class AgentBlockHandler implements BlockHandler {
base.executeFunction = async (callParams: Record<string, any>) => { base.executeFunction = async (callParams: Record<string, any>) => {
const mergedParams = mergeToolParameters(userProvidedParams, callParams) const mergedParams = mergeToolParameters(userProvidedParams, callParams)
const { blockData, blockNameMapping } = collectBlockData(ctx) const { blockData, blockNameMapping, blockOutputSchemas } = collectBlockData(ctx)
const result = await executeTool( const result = await executeTool(
'function_execute', 'function_execute',
@@ -317,6 +317,7 @@ export class AgentBlockHandler implements BlockHandler {
workflowVariables: ctx.workflowVariables || {}, workflowVariables: ctx.workflowVariables || {},
blockData, blockData,
blockNameMapping, blockNameMapping,
blockOutputSchemas,
isCustomTool: true, isCustomTool: true,
_context: { _context: {
workflowId: ctx.workflowId, workflowId: ctx.workflowId,

View File

@@ -26,7 +26,7 @@ export async function evaluateConditionExpression(
const contextSetup = `const context = ${JSON.stringify(evalContext)};` const contextSetup = `const context = ${JSON.stringify(evalContext)};`
const code = `${contextSetup}\nreturn Boolean(${conditionExpression})` const code = `${contextSetup}\nreturn Boolean(${conditionExpression})`
const { blockData, blockNameMapping } = collectBlockData(ctx) const { blockData, blockNameMapping, blockOutputSchemas } = collectBlockData(ctx)
const result = await executeTool( const result = await executeTool(
'function_execute', 'function_execute',
@@ -37,6 +37,7 @@ export async function evaluateConditionExpression(
workflowVariables: ctx.workflowVariables || {}, workflowVariables: ctx.workflowVariables || {},
blockData, blockData,
blockNameMapping, blockNameMapping,
blockOutputSchemas,
_context: { _context: {
workflowId: ctx.workflowId, workflowId: ctx.workflowId,
workspaceId: ctx.workspaceId, workspaceId: ctx.workspaceId,

View File

@@ -75,7 +75,12 @@ describe('FunctionBlockHandler', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
_context: { workflowId: mockContext.workflowId, workspaceId: mockContext.workspaceId }, blockOutputSchemas: {},
_context: {
workflowId: mockContext.workflowId,
workspaceId: mockContext.workspaceId,
isDeployedContext: mockContext.isDeployedContext,
},
} }
const expectedOutput: any = { result: 'Success' } const expectedOutput: any = { result: 'Success' }
@@ -84,8 +89,8 @@ describe('FunctionBlockHandler', () => {
expect(mockExecuteTool).toHaveBeenCalledWith( expect(mockExecuteTool).toHaveBeenCalledWith(
'function_execute', 'function_execute',
expectedToolParams, expectedToolParams,
false, // skipPostProcess false,
mockContext // execution context mockContext
) )
expect(result).toEqual(expectedOutput) expect(result).toEqual(expectedOutput)
}) })
@@ -107,7 +112,12 @@ describe('FunctionBlockHandler', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
_context: { workflowId: mockContext.workflowId, workspaceId: mockContext.workspaceId }, blockOutputSchemas: {},
_context: {
workflowId: mockContext.workflowId,
workspaceId: mockContext.workspaceId,
isDeployedContext: mockContext.isDeployedContext,
},
} }
const expectedOutput: any = { result: 'Success' } const expectedOutput: any = { result: 'Success' }
@@ -116,8 +126,8 @@ describe('FunctionBlockHandler', () => {
expect(mockExecuteTool).toHaveBeenCalledWith( expect(mockExecuteTool).toHaveBeenCalledWith(
'function_execute', 'function_execute',
expectedToolParams, expectedToolParams,
false, // skipPostProcess false,
mockContext // execution context mockContext
) )
expect(result).toEqual(expectedOutput) expect(result).toEqual(expectedOutput)
}) })
@@ -132,7 +142,12 @@ describe('FunctionBlockHandler', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
_context: { workflowId: mockContext.workflowId, workspaceId: mockContext.workspaceId }, blockOutputSchemas: {},
_context: {
workflowId: mockContext.workflowId,
workspaceId: mockContext.workspaceId,
isDeployedContext: mockContext.isDeployedContext,
},
} }
await handler.execute(mockContext, mockBlock, inputs) await handler.execute(mockContext, mockBlock, inputs)

View File

@@ -23,7 +23,7 @@ export class FunctionBlockHandler implements BlockHandler {
? inputs.code.map((c: { content: string }) => c.content).join('\n') ? inputs.code.map((c: { content: string }) => c.content).join('\n')
: inputs.code : inputs.code
const { blockData, blockNameMapping } = collectBlockData(ctx) const { blockData, blockNameMapping, blockOutputSchemas } = collectBlockData(ctx)
const result = await executeTool( const result = await executeTool(
'function_execute', 'function_execute',
@@ -35,6 +35,7 @@ export class FunctionBlockHandler implements BlockHandler {
workflowVariables: ctx.workflowVariables || {}, workflowVariables: ctx.workflowVariables || {},
blockData, blockData,
blockNameMapping, blockNameMapping,
blockOutputSchemas,
_context: { _context: {
workflowId: ctx.workflowId, workflowId: ctx.workflowId,
workspaceId: ctx.workspaceId, workspaceId: ctx.workspaceId,

View File

@@ -1,7 +1,7 @@
import '@sim/testing/mocks/executor' import '@sim/testing/mocks/executor'
import { beforeEach, describe, expect, it, type Mock, vi } from 'vitest' import { beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
import { generateRouterPrompt } from '@/blocks/blocks/router' import { generateRouterPrompt, generateRouterV2Prompt } from '@/blocks/blocks/router'
import { BlockType } from '@/executor/constants' import { BlockType } from '@/executor/constants'
import { RouterBlockHandler } from '@/executor/handlers/router/router-handler' import { RouterBlockHandler } from '@/executor/handlers/router/router-handler'
import type { ExecutionContext } from '@/executor/types' import type { ExecutionContext } from '@/executor/types'
@@ -9,6 +9,7 @@ import { getProviderFromModel } from '@/providers/utils'
import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types' import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
const mockGenerateRouterPrompt = generateRouterPrompt as Mock const mockGenerateRouterPrompt = generateRouterPrompt as Mock
const mockGenerateRouterV2Prompt = generateRouterV2Prompt as Mock
const mockGetProviderFromModel = getProviderFromModel as Mock const mockGetProviderFromModel = getProviderFromModel as Mock
const mockFetch = global.fetch as unknown as Mock const mockFetch = global.fetch as unknown as Mock
@@ -44,7 +45,7 @@ describe('RouterBlockHandler', () => {
metadata: { id: BlockType.ROUTER, name: 'Test Router' }, metadata: { id: BlockType.ROUTER, name: 'Test Router' },
position: { x: 50, y: 50 }, position: { x: 50, y: 50 },
config: { tool: BlockType.ROUTER, params: {} }, config: { tool: BlockType.ROUTER, params: {} },
inputs: { prompt: 'string', model: 'string' }, // Using ParamType strings inputs: { prompt: 'string', model: 'string' },
outputs: {}, outputs: {},
enabled: true, enabled: true,
} }
@@ -72,14 +73,11 @@ describe('RouterBlockHandler', () => {
workflow: mockWorkflow as SerializedWorkflow, workflow: mockWorkflow as SerializedWorkflow,
} }
// Reset mocks using vi
vi.clearAllMocks() vi.clearAllMocks()
// Default mock implementations
mockGetProviderFromModel.mockReturnValue('openai') mockGetProviderFromModel.mockReturnValue('openai')
mockGenerateRouterPrompt.mockReturnValue('Generated System Prompt') mockGenerateRouterPrompt.mockReturnValue('Generated System Prompt')
// Set up fetch mock to return a successful response
mockFetch.mockImplementation(() => { mockFetch.mockImplementation(() => {
return Promise.resolve({ return Promise.resolve({
ok: true, ok: true,
@@ -147,7 +145,6 @@ describe('RouterBlockHandler', () => {
}) })
) )
// Verify the request body contains the expected data
const fetchCallArgs = mockFetch.mock.calls[0] const fetchCallArgs = mockFetch.mock.calls[0]
const requestBody = JSON.parse(fetchCallArgs[1].body) const requestBody = JSON.parse(fetchCallArgs[1].body)
expect(requestBody).toMatchObject({ expect(requestBody).toMatchObject({
@@ -180,7 +177,6 @@ describe('RouterBlockHandler', () => {
const inputs = { prompt: 'Test' } const inputs = { prompt: 'Test' }
mockContext.workflow!.blocks = [mockBlock, mockTargetBlock2] mockContext.workflow!.blocks = [mockBlock, mockTargetBlock2]
// Expect execute to throw because getTargetBlocks (called internally) will throw
await expect(handler.execute(mockContext, mockBlock, inputs)).rejects.toThrow( await expect(handler.execute(mockContext, mockBlock, inputs)).rejects.toThrow(
'Target block target-block-1 not found' 'Target block target-block-1 not found'
) )
@@ -190,7 +186,6 @@ describe('RouterBlockHandler', () => {
it('should throw error if LLM response is not a valid target block ID', async () => { it('should throw error if LLM response is not a valid target block ID', async () => {
const inputs = { prompt: 'Test', apiKey: 'test-api-key' } const inputs = { prompt: 'Test', apiKey: 'test-api-key' }
// Override fetch mock to return an invalid block ID
mockFetch.mockImplementationOnce(() => { mockFetch.mockImplementationOnce(() => {
return Promise.resolve({ return Promise.resolve({
ok: true, ok: true,
@@ -228,7 +223,6 @@ describe('RouterBlockHandler', () => {
it('should handle server error responses', async () => { it('should handle server error responses', async () => {
const inputs = { prompt: 'Test error handling.', apiKey: 'test-api-key' } const inputs = { prompt: 'Test error handling.', apiKey: 'test-api-key' }
// Override fetch mock to return an error
mockFetch.mockImplementationOnce(() => { mockFetch.mockImplementationOnce(() => {
return Promise.resolve({ return Promise.resolve({
ok: false, ok: false,
@@ -276,13 +270,12 @@ describe('RouterBlockHandler', () => {
mockGetProviderFromModel.mockReturnValue('vertex') mockGetProviderFromModel.mockReturnValue('vertex')
// Mock the database query for Vertex credential
const mockDb = await import('@sim/db') const mockDb = await import('@sim/db')
const mockAccount = { const mockAccount = {
id: 'test-vertex-credential-id', id: 'test-vertex-credential-id',
accessToken: 'mock-access-token', accessToken: 'mock-access-token',
refreshToken: 'mock-refresh-token', refreshToken: 'mock-refresh-token',
expiresAt: new Date(Date.now() + 3600000), // 1 hour from now expiresAt: new Date(Date.now() + 3600000),
} }
vi.spyOn(mockDb.db.query.account, 'findFirst').mockResolvedValue(mockAccount as any) vi.spyOn(mockDb.db.query.account, 'findFirst').mockResolvedValue(mockAccount as any)
@@ -300,3 +293,287 @@ describe('RouterBlockHandler', () => {
expect(requestBody.apiKey).toBe('mock-access-token') expect(requestBody.apiKey).toBe('mock-access-token')
}) })
}) })
describe('RouterBlockHandler V2', () => {
let handler: RouterBlockHandler
let mockRouterV2Block: SerializedBlock
let mockContext: ExecutionContext
let mockWorkflow: Partial<SerializedWorkflow>
let mockTargetBlock1: SerializedBlock
let mockTargetBlock2: SerializedBlock
beforeEach(() => {
mockTargetBlock1 = {
id: 'target-block-1',
metadata: { id: 'agent', name: 'Support Agent' },
position: { x: 100, y: 100 },
config: { tool: 'agent', params: {} },
inputs: {},
outputs: {},
enabled: true,
}
mockTargetBlock2 = {
id: 'target-block-2',
metadata: { id: 'agent', name: 'Sales Agent' },
position: { x: 100, y: 150 },
config: { tool: 'agent', params: {} },
inputs: {},
outputs: {},
enabled: true,
}
mockRouterV2Block = {
id: 'router-v2-block-1',
metadata: { id: BlockType.ROUTER_V2, name: 'Test Router V2' },
position: { x: 50, y: 50 },
config: { tool: BlockType.ROUTER_V2, params: {} },
inputs: {},
outputs: {},
enabled: true,
}
mockWorkflow = {
blocks: [mockRouterV2Block, mockTargetBlock1, mockTargetBlock2],
connections: [
{
source: mockRouterV2Block.id,
target: mockTargetBlock1.id,
sourceHandle: 'router-route-support',
},
{
source: mockRouterV2Block.id,
target: mockTargetBlock2.id,
sourceHandle: 'router-route-sales',
},
],
}
handler = new RouterBlockHandler({})
mockContext = {
workflowId: 'test-workflow-id',
blockStates: new Map(),
blockLogs: [],
metadata: { duration: 0 },
environmentVariables: {},
decisions: { router: new Map(), condition: new Map() },
loopExecutions: new Map(),
completedLoops: new Set(),
executedBlocks: new Set(),
activeExecutionPath: new Set(),
workflow: mockWorkflow as SerializedWorkflow,
}
vi.clearAllMocks()
mockGetProviderFromModel.mockReturnValue('openai')
mockGenerateRouterV2Prompt.mockReturnValue('Generated V2 System Prompt')
})
it('should handle router_v2 blocks', () => {
expect(handler.canHandle(mockRouterV2Block)).toBe(true)
})
it('should execute router V2 and return reasoning', async () => {
const inputs = {
context: 'I need help with a billing issue',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: JSON.stringify([
{ id: 'route-support', title: 'Support', value: 'Customer support inquiries' },
{ id: 'route-sales', title: 'Sales', value: 'Sales and pricing questions' },
]),
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: JSON.stringify({
route: 'route-support',
reasoning: 'The user mentioned a billing issue which is a customer support matter.',
}),
model: 'gpt-4o',
tokens: { input: 150, output: 25, total: 175 },
}),
})
})
const result = await handler.execute(mockContext, mockRouterV2Block, inputs)
expect(result).toMatchObject({
context: 'I need help with a billing issue',
model: 'gpt-4o',
selectedRoute: 'route-support',
reasoning: 'The user mentioned a billing issue which is a customer support matter.',
selectedPath: {
blockId: 'target-block-1',
blockType: 'agent',
blockTitle: 'Support Agent',
},
})
})
it('should include responseFormat in provider request', async () => {
const inputs = {
context: 'Test context',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: JSON.stringify([{ id: 'route-1', title: 'Route 1', value: 'Description 1' }]),
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: JSON.stringify({ route: 'route-1', reasoning: 'Test reasoning' }),
model: 'gpt-4o',
tokens: { input: 100, output: 20, total: 120 },
}),
})
})
await handler.execute(mockContext, mockRouterV2Block, inputs)
const fetchCallArgs = mockFetch.mock.calls[0]
const requestBody = JSON.parse(fetchCallArgs[1].body)
expect(requestBody.responseFormat).toEqual({
name: 'router_response',
schema: {
type: 'object',
properties: {
route: {
type: 'string',
description: 'The selected route ID or NO_MATCH',
},
reasoning: {
type: 'string',
description: 'Brief explanation of why this route was chosen',
},
},
required: ['route', 'reasoning'],
additionalProperties: false,
},
strict: true,
})
})
it('should handle NO_MATCH response with reasoning', async () => {
const inputs = {
context: 'Random unrelated query',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: JSON.stringify([{ id: 'route-1', title: 'Route 1', value: 'Specific topic' }]),
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: JSON.stringify({
route: 'NO_MATCH',
reasoning: 'The query does not relate to any available route.',
}),
model: 'gpt-4o',
tokens: { input: 100, output: 20, total: 120 },
}),
})
})
await expect(handler.execute(mockContext, mockRouterV2Block, inputs)).rejects.toThrow(
'Router could not determine a matching route: The query does not relate to any available route.'
)
})
it('should throw error for invalid route ID in response', async () => {
const inputs = {
context: 'Test context',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: JSON.stringify([{ id: 'route-1', title: 'Route 1', value: 'Description' }]),
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: JSON.stringify({ route: 'invalid-route', reasoning: 'Some reasoning' }),
model: 'gpt-4o',
tokens: { input: 100, output: 20, total: 120 },
}),
})
})
await expect(handler.execute(mockContext, mockRouterV2Block, inputs)).rejects.toThrow(
/Router could not determine a valid route/
)
})
it('should handle routes passed as array instead of JSON string', async () => {
const inputs = {
context: 'Test context',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: [{ id: 'route-1', title: 'Route 1', value: 'Description' }],
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: JSON.stringify({ route: 'route-1', reasoning: 'Matched route 1' }),
model: 'gpt-4o',
tokens: { input: 100, output: 20, total: 120 },
}),
})
})
const result = await handler.execute(mockContext, mockRouterV2Block, inputs)
expect(result.selectedRoute).toBe('route-1')
expect(result.reasoning).toBe('Matched route 1')
})
it('should throw error when no routes are defined', async () => {
const inputs = {
context: 'Test context',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: '[]',
}
await expect(handler.execute(mockContext, mockRouterV2Block, inputs)).rejects.toThrow(
'No routes defined for router'
)
})
it('should handle fallback when JSON parsing fails', async () => {
const inputs = {
context: 'Test context',
model: 'gpt-4o',
apiKey: 'test-api-key',
routes: JSON.stringify([{ id: 'route-1', title: 'Route 1', value: 'Description' }]),
}
mockFetch.mockImplementationOnce(() => {
return Promise.resolve({
ok: true,
json: () =>
Promise.resolve({
content: 'route-1',
model: 'gpt-4o',
tokens: { input: 100, output: 5, total: 105 },
}),
})
})
const result = await handler.execute(mockContext, mockRouterV2Block, inputs)
expect(result.selectedRoute).toBe('route-1')
expect(result.reasoning).toBe('')
})
})

View File

@@ -238,6 +238,25 @@ export class RouterBlockHandler implements BlockHandler {
apiKey: finalApiKey, apiKey: finalApiKey,
workflowId: ctx.workflowId, workflowId: ctx.workflowId,
workspaceId: ctx.workspaceId, workspaceId: ctx.workspaceId,
responseFormat: {
name: 'router_response',
schema: {
type: 'object',
properties: {
route: {
type: 'string',
description: 'The selected route ID or NO_MATCH',
},
reasoning: {
type: 'string',
description: 'Brief explanation of why this route was chosen',
},
},
required: ['route', 'reasoning'],
additionalProperties: false,
},
strict: true,
},
} }
if (providerId === 'vertex') { if (providerId === 'vertex') {
@@ -277,16 +296,31 @@ export class RouterBlockHandler implements BlockHandler {
const result = await response.json() const result = await response.json()
const chosenRouteId = result.content.trim() let chosenRouteId: string
let reasoning = ''
try {
const parsedResponse = JSON.parse(result.content)
chosenRouteId = parsedResponse.route?.trim() || ''
reasoning = parsedResponse.reasoning || ''
} catch (_parseError) {
logger.error('Router response was not valid JSON despite responseFormat', {
content: result.content,
})
chosenRouteId = result.content.trim()
}
if (chosenRouteId === 'NO_MATCH' || chosenRouteId.toUpperCase() === 'NO_MATCH') { if (chosenRouteId === 'NO_MATCH' || chosenRouteId.toUpperCase() === 'NO_MATCH') {
logger.info('Router determined no route matches the context, routing to error path') logger.info('Router determined no route matches the context, routing to error path')
throw new Error('Router could not determine a matching route for the given context') throw new Error(
reasoning
? `Router could not determine a matching route: ${reasoning}`
: 'Router could not determine a matching route for the given context'
)
} }
const chosenRoute = routes.find((r) => r.id === chosenRouteId) const chosenRoute = routes.find((r) => r.id === chosenRouteId)
// Throw error if LLM returns invalid route ID - this routes through error path
if (!chosenRoute) { if (!chosenRoute) {
const availableRoutes = routes.map((r) => ({ id: r.id, title: r.title })) const availableRoutes = routes.map((r) => ({ id: r.id, title: r.title }))
logger.error( logger.error(
@@ -298,7 +332,6 @@ export class RouterBlockHandler implements BlockHandler {
) )
} }
// Find the target block connected to this route's handle
const connection = ctx.workflow?.connections.find( const connection = ctx.workflow?.connections.find(
(conn) => conn.source === block.id && conn.sourceHandle === `router-${chosenRoute.id}` (conn) => conn.source === block.id && conn.sourceHandle === `router-${chosenRoute.id}`
) )
@@ -334,6 +367,7 @@ export class RouterBlockHandler implements BlockHandler {
total: cost.total, total: cost.total,
}, },
selectedRoute: chosenRoute.id, selectedRoute: chosenRoute.id,
reasoning,
selectedPath: targetBlock selectedPath: targetBlock
? { ? {
blockId: targetBlock.id, blockId: targetBlock.id,
@@ -353,7 +387,7 @@ export class RouterBlockHandler implements BlockHandler {
} }
/** /**
* Parse routes from input (can be JSON string or array). * Parse routes from input (can be JSON string or array)
*/ */
private parseRoutes(input: any): RouteDefinition[] { private parseRoutes(input: any): RouteDefinition[] {
try { try {

View File

@@ -204,26 +204,21 @@ describe('WorkflowBlockHandler', () => {
}) })
}) })
it('should map failed child output correctly', () => { it('should throw error for failed child output so BlockExecutor can check error port', () => {
const childResult = { const childResult = {
success: false, success: false,
error: 'Child workflow failed', error: 'Child workflow failed',
} }
const result = (handler as any).mapChildOutputToParent( expect(() =>
childResult, (handler as any).mapChildOutputToParent(childResult, 'child-id', 'Child Workflow', 100)
'child-id', ).toThrow('Error in child workflow "Child Workflow": Child workflow failed')
'Child Workflow',
100
)
expect(result).toEqual({ try {
success: false, ;(handler as any).mapChildOutputToParent(childResult, 'child-id', 'Child Workflow', 100)
childWorkflowName: 'Child Workflow', } catch (error: any) {
result: {}, expect(error.childTraceSpans).toEqual([])
error: 'Child workflow failed', }
childTraceSpans: [],
})
}) })
it('should handle nested response structures', () => { it('should handle nested response structures', () => {

View File

@@ -144,6 +144,11 @@ export class WorkflowBlockHandler implements BlockHandler {
const workflowMetadata = workflows[workflowId] const workflowMetadata = workflows[workflowId]
const childWorkflowName = workflowMetadata?.name || workflowId const childWorkflowName = workflowMetadata?.name || workflowId
const originalError = error.message || 'Unknown error'
const wrappedError = new Error(
`Error in child workflow "${childWorkflowName}": ${originalError}`
)
if (error.executionResult?.logs) { if (error.executionResult?.logs) {
const executionResult = error.executionResult as ExecutionResult const executionResult = error.executionResult as ExecutionResult
@@ -159,28 +164,12 @@ export class WorkflowBlockHandler implements BlockHandler {
) )
logger.info(`Captured ${childTraceSpans.length} child trace spans from failed execution`) logger.info(`Captured ${childTraceSpans.length} child trace spans from failed execution`)
;(wrappedError as any).childTraceSpans = childTraceSpans
return { } else if (error.childTraceSpans && Array.isArray(error.childTraceSpans)) {
success: false, ;(wrappedError as any).childTraceSpans = error.childTraceSpans
childWorkflowName,
result: {},
error: error.message || 'Child workflow execution failed',
childTraceSpans: childTraceSpans,
} as Record<string, any>
} }
if (error.childTraceSpans && Array.isArray(error.childTraceSpans)) { throw wrappedError
return {
success: false,
childWorkflowName,
result: {},
error: error.message || 'Child workflow execution failed',
childTraceSpans: error.childTraceSpans,
} as Record<string, any>
}
const originalError = error.message || 'Unknown error'
throw new Error(`Error in child workflow "${childWorkflowName}": ${originalError}`)
} }
} }
@@ -452,17 +441,13 @@ export class WorkflowBlockHandler implements BlockHandler {
if (!success) { if (!success) {
logger.warn(`Child workflow ${childWorkflowName} failed`) logger.warn(`Child workflow ${childWorkflowName} failed`)
// Return failure with child trace spans so they can be displayed const error = new Error(
return { `Error in child workflow "${childWorkflowName}": ${childResult.error || 'Child workflow execution failed'}`
success: false, )
childWorkflowName, ;(error as any).childTraceSpans = childTraceSpans || []
result, throw error
error: childResult.error || 'Child workflow execution failed',
childTraceSpans: childTraceSpans || [],
} as Record<string, any>
} }
// Success case
return { return {
success: true, success: true,
childWorkflowName, childWorkflowName,

View File

@@ -1,24 +1,43 @@
import { getBlockOutputs } from '@/lib/workflows/blocks/block-outputs'
import { normalizeName } from '@/executor/constants' import { normalizeName } from '@/executor/constants'
import type { ExecutionContext } from '@/executor/types' import type { ExecutionContext } from '@/executor/types'
import type { OutputSchema } from '@/executor/utils/block-reference'
export interface BlockDataCollection { export interface BlockDataCollection {
blockData: Record<string, any> blockData: Record<string, unknown>
blockNameMapping: Record<string, string> blockNameMapping: Record<string, string>
blockOutputSchemas: Record<string, OutputSchema>
} }
export function collectBlockData(ctx: ExecutionContext): BlockDataCollection { export function collectBlockData(ctx: ExecutionContext): BlockDataCollection {
const blockData: Record<string, any> = {} const blockData: Record<string, unknown> = {}
const blockNameMapping: Record<string, string> = {} const blockNameMapping: Record<string, string> = {}
const blockOutputSchemas: Record<string, OutputSchema> = {}
for (const [id, state] of ctx.blockStates.entries()) { for (const [id, state] of ctx.blockStates.entries()) {
if (state.output !== undefined) { if (state.output !== undefined) {
blockData[id] = state.output blockData[id] = state.output
const workflowBlock = ctx.workflow?.blocks?.find((b) => b.id === id) }
if (workflowBlock?.metadata?.name) {
blockNameMapping[normalizeName(workflowBlock.metadata.name)] = id const workflowBlock = ctx.workflow?.blocks?.find((b) => b.id === id)
if (!workflowBlock) continue
if (workflowBlock.metadata?.name) {
blockNameMapping[normalizeName(workflowBlock.metadata.name)] = id
}
const blockType = workflowBlock.metadata?.id
if (blockType) {
const params = workflowBlock.config?.params as Record<string, unknown> | undefined
const subBlocks = params
? Object.fromEntries(Object.entries(params).map(([k, v]) => [k, { value: v }]))
: undefined
const schema = getBlockOutputs(blockType, subBlocks)
if (schema && Object.keys(schema).length > 0) {
blockOutputSchemas[id] = schema
} }
} }
} }
return { blockData, blockNameMapping } return { blockData, blockNameMapping, blockOutputSchemas }
} }

View File

@@ -0,0 +1,255 @@
/**
* @vitest-environment node
*/
import { describe, expect, it } from 'vitest'
import {
type BlockReferenceContext,
InvalidFieldError,
resolveBlockReference,
} from './block-reference'
describe('resolveBlockReference', () => {
const createContext = (
overrides: Partial<BlockReferenceContext> = {}
): BlockReferenceContext => ({
blockNameMapping: { start: 'block-1', agent: 'block-2' },
blockData: {},
blockOutputSchemas: {},
...overrides,
})
describe('block name resolution', () => {
it('should return undefined when block name does not exist', () => {
const ctx = createContext()
const result = resolveBlockReference('unknown', ['field'], ctx)
expect(result).toBeUndefined()
})
it('should normalize block name before lookup', () => {
const ctx = createContext({
blockNameMapping: { myblock: 'block-1' },
blockData: { 'block-1': { value: 'test' } },
})
const result = resolveBlockReference('MyBlock', ['value'], ctx)
expect(result).toEqual({ value: 'test', blockId: 'block-1' })
})
it('should handle block names with spaces', () => {
const ctx = createContext({
blockNameMapping: { myblock: 'block-1' },
blockData: { 'block-1': { value: 'test' } },
})
const result = resolveBlockReference('My Block', ['value'], ctx)
expect(result).toEqual({ value: 'test', blockId: 'block-1' })
})
})
describe('field resolution', () => {
it('should return entire block output when no path specified', () => {
const ctx = createContext({
blockData: { 'block-1': { input: 'hello', other: 'data' } },
})
const result = resolveBlockReference('start', [], ctx)
expect(result).toEqual({
value: { input: 'hello', other: 'data' },
blockId: 'block-1',
})
})
it('should resolve simple field path', () => {
const ctx = createContext({
blockData: { 'block-1': { input: 'hello' } },
})
const result = resolveBlockReference('start', ['input'], ctx)
expect(result).toEqual({ value: 'hello', blockId: 'block-1' })
})
it('should resolve nested field path', () => {
const ctx = createContext({
blockData: { 'block-1': { response: { data: { name: 'test' } } } },
})
const result = resolveBlockReference('start', ['response', 'data', 'name'], ctx)
expect(result).toEqual({ value: 'test', blockId: 'block-1' })
})
it('should resolve array index path', () => {
const ctx = createContext({
blockData: { 'block-1': { items: ['a', 'b', 'c'] } },
})
const result = resolveBlockReference('start', ['items', '1'], ctx)
expect(result).toEqual({ value: 'b', blockId: 'block-1' })
})
it('should return undefined value when field exists but has no value', () => {
const ctx = createContext({
blockData: { 'block-1': { input: undefined } },
blockOutputSchemas: {
'block-1': { input: { type: 'string' } },
},
})
const result = resolveBlockReference('start', ['input'], ctx)
expect(result).toEqual({ value: undefined, blockId: 'block-1' })
})
it('should return null value when field has null', () => {
const ctx = createContext({
blockData: { 'block-1': { input: null } },
})
const result = resolveBlockReference('start', ['input'], ctx)
expect(result).toEqual({ value: null, blockId: 'block-1' })
})
})
describe('schema validation', () => {
it('should throw InvalidFieldError when field not in schema', () => {
const ctx = createContext({
blockData: { 'block-1': { existing: 'value' } },
blockOutputSchemas: {
'block-1': {
input: { type: 'string' },
conversationId: { type: 'string' },
},
},
})
expect(() => resolveBlockReference('start', ['invalid'], ctx)).toThrow(InvalidFieldError)
expect(() => resolveBlockReference('start', ['invalid'], ctx)).toThrow(
/"invalid" doesn't exist on block "start"/
)
})
it('should include available fields in error message', () => {
const ctx = createContext({
blockData: { 'block-1': {} },
blockOutputSchemas: {
'block-1': {
input: { type: 'string' },
conversationId: { type: 'string' },
files: { type: 'files' },
},
},
})
try {
resolveBlockReference('start', ['typo'], ctx)
expect.fail('Should have thrown')
} catch (error) {
expect(error).toBeInstanceOf(InvalidFieldError)
const fieldError = error as InvalidFieldError
expect(fieldError.availableFields).toContain('input')
expect(fieldError.availableFields).toContain('conversationId')
expect(fieldError.availableFields).toContain('files')
}
})
it('should allow valid field even when value is undefined', () => {
const ctx = createContext({
blockData: { 'block-1': {} },
blockOutputSchemas: {
'block-1': { input: { type: 'string' } },
},
})
const result = resolveBlockReference('start', ['input'], ctx)
expect(result).toEqual({ value: undefined, blockId: 'block-1' })
})
it('should validate path when block has no output yet', () => {
const ctx = createContext({
blockData: {},
blockOutputSchemas: {
'block-1': { input: { type: 'string' } },
},
})
expect(() => resolveBlockReference('start', ['invalid'], ctx)).toThrow(InvalidFieldError)
})
it('should return undefined for valid field when block has no output', () => {
const ctx = createContext({
blockData: {},
blockOutputSchemas: {
'block-1': { input: { type: 'string' } },
},
})
const result = resolveBlockReference('start', ['input'], ctx)
expect(result).toEqual({ value: undefined, blockId: 'block-1' })
})
})
describe('without schema (pass-through mode)', () => {
it('should return undefined value without throwing when no schema', () => {
const ctx = createContext({
blockData: { 'block-1': { existing: 'value' } },
})
const result = resolveBlockReference('start', ['missing'], ctx)
expect(result).toEqual({ value: undefined, blockId: 'block-1' })
})
})
describe('file type handling', () => {
it('should allow file property access', () => {
const ctx = createContext({
blockData: {
'block-1': {
files: [{ name: 'test.txt', url: 'http://example.com/file' }],
},
},
blockOutputSchemas: {
'block-1': { files: { type: 'files' } },
},
})
const result = resolveBlockReference('start', ['files', '0', 'name'], ctx)
expect(result).toEqual({ value: 'test.txt', blockId: 'block-1' })
})
it('should validate file property names', () => {
const ctx = createContext({
blockData: { 'block-1': { files: [] } },
blockOutputSchemas: {
'block-1': { files: { type: 'files' } },
},
})
expect(() => resolveBlockReference('start', ['files', '0', 'invalid'], ctx)).toThrow(
InvalidFieldError
)
})
})
})
describe('InvalidFieldError', () => {
it('should have correct properties', () => {
const error = new InvalidFieldError('myBlock', 'invalid.path', ['field1', 'field2'])
expect(error.blockName).toBe('myBlock')
expect(error.fieldPath).toBe('invalid.path')
expect(error.availableFields).toEqual(['field1', 'field2'])
expect(error.name).toBe('InvalidFieldError')
})
it('should format message correctly', () => {
const error = new InvalidFieldError('start', 'typo', ['input', 'files'])
expect(error.message).toBe(
'"typo" doesn\'t exist on block "start". Available fields: input, files'
)
})
it('should handle empty available fields', () => {
const error = new InvalidFieldError('start', 'field', [])
expect(error.message).toBe('"field" doesn\'t exist on block "start". Available fields: none')
})
})

View File

@@ -0,0 +1,210 @@
import { USER_FILE_ACCESSIBLE_PROPERTIES } from '@/lib/workflows/types'
import { normalizeName } from '@/executor/constants'
import { navigatePath } from '@/executor/variables/resolvers/reference'
export type OutputSchema = Record<string, { type?: string; description?: string } | unknown>
export interface BlockReferenceContext {
blockNameMapping: Record<string, string>
blockData: Record<string, unknown>
blockOutputSchemas?: Record<string, OutputSchema>
}
export interface BlockReferenceResult {
value: unknown
blockId: string
}
export class InvalidFieldError extends Error {
constructor(
public readonly blockName: string,
public readonly fieldPath: string,
public readonly availableFields: string[]
) {
super(
`"${fieldPath}" doesn't exist on block "${blockName}". ` +
`Available fields: ${availableFields.length > 0 ? availableFields.join(', ') : 'none'}`
)
this.name = 'InvalidFieldError'
}
}
function isFileType(value: unknown): boolean {
if (typeof value !== 'object' || value === null) return false
const typed = value as { type?: string }
return typed.type === 'file[]' || typed.type === 'files'
}
function isArrayType(value: unknown): value is { type: 'array'; items?: unknown } {
if (typeof value !== 'object' || value === null) return false
return (value as { type?: string }).type === 'array'
}
function getArrayItems(schema: unknown): unknown {
if (typeof schema !== 'object' || schema === null) return undefined
return (schema as { items?: unknown }).items
}
function getProperties(schema: unknown): Record<string, unknown> | undefined {
if (typeof schema !== 'object' || schema === null) return undefined
const props = (schema as { properties?: unknown }).properties
return typeof props === 'object' && props !== null
? (props as Record<string, unknown>)
: undefined
}
function lookupField(schema: unknown, fieldName: string): unknown | undefined {
if (typeof schema !== 'object' || schema === null) return undefined
const typed = schema as Record<string, unknown>
if (fieldName in typed) {
return typed[fieldName]
}
const props = getProperties(schema)
if (props && fieldName in props) {
return props[fieldName]
}
return undefined
}
function isPathInSchema(schema: OutputSchema | undefined, pathParts: string[]): boolean {
if (!schema || pathParts.length === 0) {
return true
}
let current: unknown = schema
for (let i = 0; i < pathParts.length; i++) {
const part = pathParts[i]
if (current === null || current === undefined) {
return false
}
if (/^\d+$/.test(part)) {
if (isFileType(current)) {
const nextPart = pathParts[i + 1]
return (
!nextPart ||
USER_FILE_ACCESSIBLE_PROPERTIES.includes(
nextPart as (typeof USER_FILE_ACCESSIBLE_PROPERTIES)[number]
)
)
}
if (isArrayType(current)) {
current = getArrayItems(current)
}
continue
}
const arrayMatch = part.match(/^([^[]+)\[(\d+)\]$/)
if (arrayMatch) {
const [, prop] = arrayMatch
const fieldDef = lookupField(current, prop)
if (!fieldDef) return false
if (isFileType(fieldDef)) {
const nextPart = pathParts[i + 1]
return (
!nextPart ||
USER_FILE_ACCESSIBLE_PROPERTIES.includes(
nextPart as (typeof USER_FILE_ACCESSIBLE_PROPERTIES)[number]
)
)
}
current = isArrayType(fieldDef) ? getArrayItems(fieldDef) : fieldDef
continue
}
if (
isFileType(current) &&
USER_FILE_ACCESSIBLE_PROPERTIES.includes(
part as (typeof USER_FILE_ACCESSIBLE_PROPERTIES)[number]
)
) {
return true
}
const fieldDef = lookupField(current, part)
if (fieldDef !== undefined) {
if (isFileType(fieldDef)) {
const nextPart = pathParts[i + 1]
if (!nextPart) return true
if (/^\d+$/.test(nextPart)) {
const afterIndex = pathParts[i + 2]
return (
!afterIndex ||
USER_FILE_ACCESSIBLE_PROPERTIES.includes(
afterIndex as (typeof USER_FILE_ACCESSIBLE_PROPERTIES)[number]
)
)
}
return USER_FILE_ACCESSIBLE_PROPERTIES.includes(
nextPart as (typeof USER_FILE_ACCESSIBLE_PROPERTIES)[number]
)
}
current = fieldDef
continue
}
if (isArrayType(current)) {
const items = getArrayItems(current)
const itemField = lookupField(items, part)
if (itemField !== undefined) {
current = itemField
continue
}
}
return false
}
return true
}
function getSchemaFieldNames(schema: OutputSchema | undefined): string[] {
if (!schema) return []
return Object.keys(schema)
}
export function resolveBlockReference(
blockName: string,
pathParts: string[],
context: BlockReferenceContext
): BlockReferenceResult | undefined {
const normalizedName = normalizeName(blockName)
const blockId = context.blockNameMapping[normalizedName]
if (!blockId) {
return undefined
}
const blockOutput = context.blockData[blockId]
const schema = context.blockOutputSchemas?.[blockId]
if (blockOutput === undefined) {
if (schema && pathParts.length > 0) {
if (!isPathInSchema(schema, pathParts)) {
throw new InvalidFieldError(blockName, pathParts.join('.'), getSchemaFieldNames(schema))
}
}
return { value: undefined, blockId }
}
if (pathParts.length === 0) {
return { value: blockOutput, blockId }
}
const value = navigatePath(blockOutput, pathParts)
if (value === undefined && schema) {
if (!isPathInSchema(schema, pathParts)) {
throw new InvalidFieldError(blockName, pathParts.join('.'), getSchemaFieldNames(schema))
}
}
return { value, blockId }
}

View File

@@ -1,11 +1,15 @@
import { getBlockOutputs } from '@/lib/workflows/blocks/block-outputs' import { getBlockOutputs } from '@/lib/workflows/blocks/block-outputs'
import { USER_FILE_ACCESSIBLE_PROPERTIES } from '@/lib/workflows/types'
import { import {
isReference, isReference,
normalizeName, normalizeName,
parseReferencePath, parseReferencePath,
SPECIAL_REFERENCE_PREFIXES, SPECIAL_REFERENCE_PREFIXES,
} from '@/executor/constants' } from '@/executor/constants'
import {
InvalidFieldError,
type OutputSchema,
resolveBlockReference,
} from '@/executor/utils/block-reference'
import { import {
navigatePath, navigatePath,
type ResolutionContext, type ResolutionContext,
@@ -14,123 +18,6 @@ import {
import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types' import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'
import { getTool } from '@/tools/utils' import { getTool } from '@/tools/utils'
function isPathInOutputSchema(
outputs: Record<string, any> | undefined,
pathParts: string[]
): boolean {
if (!outputs || pathParts.length === 0) {
return true
}
const isFileArrayType = (value: any): boolean =>
value?.type === 'file[]' || value?.type === 'files'
let current: any = outputs
for (let i = 0; i < pathParts.length; i++) {
const part = pathParts[i]
const arrayMatch = part.match(/^([^[]+)\[(\d+)\]$/)
if (arrayMatch) {
const [, prop] = arrayMatch
let fieldDef: any
if (prop in current) {
fieldDef = current[prop]
} else if (current.properties && prop in current.properties) {
fieldDef = current.properties[prop]
} else if (current.type === 'array' && current.items) {
if (current.items.properties && prop in current.items.properties) {
fieldDef = current.items.properties[prop]
} else if (prop in current.items) {
fieldDef = current.items[prop]
}
}
if (!fieldDef) {
return false
}
if (isFileArrayType(fieldDef)) {
if (i + 1 < pathParts.length) {
return USER_FILE_ACCESSIBLE_PROPERTIES.includes(pathParts[i + 1] as any)
}
return true
}
if (fieldDef.type === 'array' && fieldDef.items) {
current = fieldDef.items
continue
}
current = fieldDef
continue
}
if (/^\d+$/.test(part)) {
if (isFileArrayType(current)) {
if (i + 1 < pathParts.length) {
const nextPart = pathParts[i + 1]
return USER_FILE_ACCESSIBLE_PROPERTIES.includes(nextPart as any)
}
return true
}
continue
}
if (current === null || current === undefined) {
return false
}
if (part in current) {
const nextCurrent = current[part]
if (nextCurrent?.type === 'file[]' && i + 1 < pathParts.length) {
const nextPart = pathParts[i + 1]
if (/^\d+$/.test(nextPart) && i + 2 < pathParts.length) {
const propertyPart = pathParts[i + 2]
return USER_FILE_ACCESSIBLE_PROPERTIES.includes(propertyPart as any)
}
}
current = nextCurrent
continue
}
if (current.properties && part in current.properties) {
current = current.properties[part]
continue
}
if (current.type === 'array' && current.items) {
if (current.items.properties && part in current.items.properties) {
current = current.items.properties[part]
continue
}
if (part in current.items) {
current = current.items[part]
continue
}
}
if (isFileArrayType(current) && USER_FILE_ACCESSIBLE_PROPERTIES.includes(part as any)) {
return true
}
if ('type' in current && typeof current.type === 'string') {
if (!current.properties && !current.items) {
return false
}
}
return false
}
return true
}
function getSchemaFieldNames(outputs: Record<string, any> | undefined): string[] {
if (!outputs) return []
return Object.keys(outputs)
}
export class BlockResolver implements Resolver { export class BlockResolver implements Resolver {
private nameToBlockId: Map<string, string> private nameToBlockId: Map<string, string>
private blockById: Map<string, SerializedBlock> private blockById: Map<string, SerializedBlock>
@@ -170,83 +57,94 @@ export class BlockResolver implements Resolver {
return undefined return undefined
} }
const block = this.blockById.get(blockId) const block = this.blockById.get(blockId)!
const output = this.getBlockOutput(blockId, context) const output = this.getBlockOutput(blockId, context)
if (output === undefined) { const blockData: Record<string, unknown> = {}
const blockOutputSchemas: Record<string, OutputSchema> = {}
if (output !== undefined) {
blockData[blockId] = output
}
const blockType = block.metadata?.id
const params = block.config?.params as Record<string, unknown> | undefined
const subBlocks = params
? Object.fromEntries(Object.entries(params).map(([k, v]) => [k, { value: v }]))
: undefined
const toolId = block.config?.tool
const toolConfig = toolId ? getTool(toolId) : undefined
const outputSchema =
toolConfig?.outputs ?? (blockType ? getBlockOutputs(blockType, subBlocks) : block.outputs)
if (outputSchema && Object.keys(outputSchema).length > 0) {
blockOutputSchemas[blockId] = outputSchema
}
try {
const result = resolveBlockReference(blockName, pathParts, {
blockNameMapping: Object.fromEntries(this.nameToBlockId),
blockData,
blockOutputSchemas,
})!
if (result.value !== undefined) {
return result.value
}
return this.handleBackwardsCompat(block, output, pathParts)
} catch (error) {
if (error instanceof InvalidFieldError) {
const fallback = this.handleBackwardsCompat(block, output, pathParts)
if (fallback !== undefined) {
return fallback
}
throw new Error(error.message)
}
throw error
}
}
private handleBackwardsCompat(
block: SerializedBlock,
output: unknown,
pathParts: string[]
): unknown {
if (output === undefined || pathParts.length === 0) {
return undefined return undefined
} }
if (pathParts.length === 0) {
return output
}
// Try the original path first
let result = navigatePath(output, pathParts)
// If successful, return it immediately
if (result !== undefined) {
return result
}
// Response block backwards compatibility:
// Old: <responseBlock.response.data> -> New: <responseBlock.data>
// Only apply fallback if:
// 1. Block type is 'response'
// 2. Path starts with 'response.'
// 3. Output doesn't have a 'response' key (confirming it's the new format)
if ( if (
block?.metadata?.id === 'response' && block.metadata?.id === 'response' &&
pathParts[0] === 'response' && pathParts[0] === 'response' &&
output?.response === undefined (output as Record<string, unknown>)?.response === undefined
) { ) {
const adjustedPathParts = pathParts.slice(1) const adjustedPathParts = pathParts.slice(1)
if (adjustedPathParts.length === 0) { if (adjustedPathParts.length === 0) {
return output return output
} }
result = navigatePath(output, adjustedPathParts) const fallbackResult = navigatePath(output, adjustedPathParts)
if (result !== undefined) { if (fallbackResult !== undefined) {
return result return fallbackResult
} }
} }
// Workflow block backwards compatibility:
// Old: <workflowBlock.result.response.data> -> New: <workflowBlock.result.data>
// Only apply fallback if:
// 1. Block type is 'workflow' or 'workflow_input'
// 2. Path starts with 'result.response.'
// 3. output.result.response doesn't exist (confirming child used new format)
const isWorkflowBlock = const isWorkflowBlock =
block?.metadata?.id === 'workflow' || block?.metadata?.id === 'workflow_input' block.metadata?.id === 'workflow' || block.metadata?.id === 'workflow_input'
const outputRecord = output as Record<string, Record<string, unknown> | undefined>
if ( if (
isWorkflowBlock && isWorkflowBlock &&
pathParts[0] === 'result' && pathParts[0] === 'result' &&
pathParts[1] === 'response' && pathParts[1] === 'response' &&
output?.result?.response === undefined outputRecord?.result?.response === undefined
) { ) {
const adjustedPathParts = ['result', ...pathParts.slice(2)] const adjustedPathParts = ['result', ...pathParts.slice(2)]
result = navigatePath(output, adjustedPathParts) const fallbackResult = navigatePath(output, adjustedPathParts)
if (result !== undefined) { if (fallbackResult !== undefined) {
return result return fallbackResult
} }
} }
const blockType = block?.metadata?.id
const params = block?.config?.params as Record<string, unknown> | undefined
const subBlocks = params
? Object.fromEntries(Object.entries(params).map(([k, v]) => [k, { value: v }]))
: undefined
const toolId = block?.config?.tool
const toolConfig = toolId ? getTool(toolId) : undefined
const outputSchema =
toolConfig?.outputs ?? (blockType ? getBlockOutputs(blockType, subBlocks) : block?.outputs)
const schemaFields = getSchemaFieldNames(outputSchema)
if (schemaFields.length > 0 && !isPathInOutputSchema(outputSchema, pathParts)) {
throw new Error(
`"${pathParts.join('.')}" doesn't exist on block "${blockName}". ` +
`Available fields: ${schemaFields.join(', ')}`
)
}
return undefined return undefined
} }

View File

@@ -27,23 +27,28 @@ export function navigatePath(obj: any, path: string[]): any {
return undefined return undefined
} }
// Handle array indexing like "items[0]" or just numeric indices const arrayMatch = part.match(/^([^[]+)(\[.+)$/)
const arrayMatch = part.match(/^([^[]+)\[(\d+)\](.*)$/)
if (arrayMatch) { if (arrayMatch) {
// Handle complex array access like "items[0]" const [, prop, bracketsPart] = arrayMatch
const [, prop, index] = arrayMatch
current = current[prop] current = current[prop]
if (current === undefined || current === null) { if (current === undefined || current === null) {
return undefined return undefined
} }
const idx = Number.parseInt(index, 10)
current = Array.isArray(current) ? current[idx] : undefined const indices = bracketsPart.match(/\[(\d+)\]/g)
if (indices) {
for (const indexMatch of indices) {
if (current === null || current === undefined) {
return undefined
}
const idx = Number.parseInt(indexMatch.slice(1, -1), 10)
current = Array.isArray(current) ? current[idx] : undefined
}
}
} else if (/^\d+$/.test(part)) { } else if (/^\d+$/.test(part)) {
// Handle plain numeric index
const index = Number.parseInt(part, 10) const index = Number.parseInt(part, 10)
current = Array.isArray(current) ? current[index] : undefined current = Array.isArray(current) ? current[index] : undefined
} else { } else {
// Handle regular property access
current = current[part] current = current[part]
} }
} }

View File

@@ -130,7 +130,11 @@ async function executeCode(request) {
await jail.set('environmentVariables', new ivm.ExternalCopy(envVars).copyInto()) await jail.set('environmentVariables', new ivm.ExternalCopy(envVars).copyInto())
for (const [key, value] of Object.entries(contextVariables)) { for (const [key, value] of Object.entries(contextVariables)) {
await jail.set(key, new ivm.ExternalCopy(value).copyInto()) if (value === undefined) {
await jail.set(key, undefined)
} else {
await jail.set(key, new ivm.ExternalCopy(value).copyInto())
}
} }
const fetchCallback = new ivm.Reference(async (url, optionsJson) => { const fetchCallback = new ivm.Reference(async (url, optionsJson) => {

View File

@@ -56,6 +56,7 @@ describe('Function Execute Tool', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
blockOutputSchemas: {},
isCustomTool: false, isCustomTool: false,
language: 'javascript', language: 'javascript',
timeout: 5000, timeout: 5000,
@@ -83,6 +84,7 @@ describe('Function Execute Tool', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
blockOutputSchemas: {},
isCustomTool: false, isCustomTool: false,
language: 'javascript', language: 'javascript',
workflowId: undefined, workflowId: undefined,
@@ -101,6 +103,7 @@ describe('Function Execute Tool', () => {
workflowVariables: {}, workflowVariables: {},
blockData: {}, blockData: {},
blockNameMapping: {}, blockNameMapping: {},
blockOutputSchemas: {},
isCustomTool: false, isCustomTool: false,
language: 'javascript', language: 'javascript',
workflowId: undefined, workflowId: undefined,

View File

@@ -53,6 +53,13 @@ export const functionExecuteTool: ToolConfig<CodeExecutionInput, CodeExecutionOu
description: 'Mapping of block names to block IDs', description: 'Mapping of block names to block IDs',
default: {}, default: {},
}, },
blockOutputSchemas: {
type: 'object',
required: false,
visibility: 'hidden',
description: 'Mapping of block IDs to their output schemas for validation',
default: {},
},
workflowVariables: { workflowVariables: {
type: 'object', type: 'object',
required: false, required: false,
@@ -81,6 +88,7 @@ export const functionExecuteTool: ToolConfig<CodeExecutionInput, CodeExecutionOu
workflowVariables: params.workflowVariables || {}, workflowVariables: params.workflowVariables || {},
blockData: params.blockData || {}, blockData: params.blockData || {},
blockNameMapping: params.blockNameMapping || {}, blockNameMapping: params.blockNameMapping || {},
blockOutputSchemas: params.blockOutputSchemas || {},
workflowId: params._context?.workflowId, workflowId: params._context?.workflowId,
isCustomTool: params.isCustomTool || false, isCustomTool: params.isCustomTool || false,
} }

View File

@@ -11,6 +11,7 @@ export interface CodeExecutionInput {
workflowVariables?: Record<string, unknown> workflowVariables?: Record<string, unknown>
blockData?: Record<string, unknown> blockData?: Record<string, unknown>
blockNameMapping?: Record<string, string> blockNameMapping?: Record<string, string>
blockOutputSchemas?: Record<string, Record<string, unknown>>
_context?: { _context?: {
workflowId?: string workflowId?: string
} }

View File

@@ -110,12 +110,22 @@ spec:
{{- end }} {{- end }}
{{- include "sim.resources" .Values.app | nindent 10 }} {{- include "sim.resources" .Values.app | nindent 10 }}
{{- include "sim.securityContext" .Values.app | nindent 10 }} {{- include "sim.securityContext" .Values.app | nindent 10 }}
{{- with .Values.extraVolumeMounts }} {{- if or .Values.extraVolumeMounts .Values.app.extraVolumeMounts }}
volumeMounts: volumeMounts:
{{- with .Values.extraVolumeMounts }}
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }}
{{- with .Values.app.extraVolumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
{{- end }} {{- end }}
{{- with .Values.extraVolumes }} {{- if or .Values.extraVolumes .Values.app.extraVolumes }}
volumes: volumes:
{{- with .Values.extraVolumes }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.app.extraVolumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }} {{- end }}
{{- end }} {{- end }}

View File

@@ -92,6 +92,7 @@ spec:
{{- toYaml .Values.ollama.readinessProbe | nindent 12 }} {{- toYaml .Values.ollama.readinessProbe | nindent 12 }}
{{- end }} {{- end }}
{{- include "sim.resources" .Values.ollama | nindent 10 }} {{- include "sim.resources" .Values.ollama | nindent 10 }}
{{- if or .Values.ollama.persistence.enabled .Values.extraVolumeMounts .Values.ollama.extraVolumeMounts }}
volumeMounts: volumeMounts:
{{- if .Values.ollama.persistence.enabled }} {{- if .Values.ollama.persistence.enabled }}
- name: ollama-data - name: ollama-data
@@ -100,13 +101,22 @@ spec:
{{- with .Values.extraVolumeMounts }} {{- with .Values.extraVolumeMounts }}
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }} {{- end }}
{{- if .Values.ollama.persistence.enabled }} {{- with .Values.ollama.extraVolumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
{{- end }}
{{- if or .Values.ollama.persistence.enabled .Values.extraVolumes .Values.ollama.extraVolumes }}
volumes: volumes:
{{- if .Values.ollama.persistence.enabled }}
- name: ollama-data - name: ollama-data
persistentVolumeClaim: persistentVolumeClaim:
claimName: {{ include "sim.fullname" . }}-ollama-data claimName: {{ include "sim.fullname" . }}-ollama-data
{{- end }}
{{- with .Values.extraVolumes }} {{- with .Values.extraVolumes }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }} {{- end }}
{{- with .Values.ollama.extraVolumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }} {{- end }}
{{- end }} {{- end }}

View File

@@ -84,12 +84,22 @@ spec:
{{- end }} {{- end }}
{{- include "sim.resources" .Values.realtime | nindent 10 }} {{- include "sim.resources" .Values.realtime | nindent 10 }}
{{- include "sim.securityContext" .Values.realtime | nindent 10 }} {{- include "sim.securityContext" .Values.realtime | nindent 10 }}
{{- with .Values.extraVolumeMounts }} {{- if or .Values.extraVolumeMounts .Values.realtime.extraVolumeMounts }}
volumeMounts: volumeMounts:
{{- with .Values.extraVolumeMounts }}
{{- toYaml . | nindent 12 }} {{- toYaml . | nindent 12 }}
{{- end }}
{{- with .Values.realtime.extraVolumeMounts }}
{{- toYaml . | nindent 12 }}
{{- end }}
{{- end }} {{- end }}
{{- with .Values.extraVolumes }} {{- if or .Values.extraVolumes .Values.realtime.extraVolumes }}
volumes: volumes:
{{- with .Values.extraVolumes }}
{{- toYaml . | nindent 8 }} {{- toYaml . | nindent 8 }}
{{- end }}
{{- with .Values.realtime.extraVolumes }}
{{- toYaml . | nindent 8 }}
{{- end }}
{{- end }} {{- end }}
{{- end }} {{- end }}

View File

@@ -224,6 +224,10 @@ app:
timeoutSeconds: 5 timeoutSeconds: 5
failureThreshold: 3 failureThreshold: 3
# Additional volumes for app deployment (e.g., branding assets, custom configs)
extraVolumes: []
extraVolumeMounts: []
# Realtime socket server configuration # Realtime socket server configuration
realtime: realtime:
# Enable/disable the realtime service # Enable/disable the realtime service
@@ -301,6 +305,10 @@ realtime:
timeoutSeconds: 5 timeoutSeconds: 5
failureThreshold: 3 failureThreshold: 3
# Additional volumes for realtime deployment
extraVolumes: []
extraVolumeMounts: []
# Database migrations job configuration # Database migrations job configuration
migrations: migrations:
# Enable/disable migrations job # Enable/disable migrations job
@@ -539,6 +547,10 @@ ollama:
timeoutSeconds: 5 timeoutSeconds: 5
failureThreshold: 3 failureThreshold: 3
# Additional volumes for ollama deployment
extraVolumes: []
extraVolumeMounts: []
# Ingress configuration # Ingress configuration
ingress: ingress:
# Enable/disable ingress # Enable/disable ingress