diff --git a/apps/sim/app/workspace/[workspaceId]/logs/components/log-details/utils.ts b/apps/sim/app/workspace/[workspaceId]/logs/components/log-details/utils.ts index 629524f244e..d41182aea7b 100644 --- a/apps/sim/app/workspace/[workspaceId]/logs/components/log-details/utils.ts +++ b/apps/sim/app/workspace/[workspaceId]/logs/components/log-details/utils.ts @@ -2,6 +2,7 @@ import type React from 'react' import { AgentSkillsIcon, WorkflowIcon } from '@/components/icons' import { formatCreditCost } from '@/lib/billing/credits/conversion' import { perceivedBrightness } from '@/lib/colors' +import { hasUnhandledError } from '@/lib/logs/execution/trace-spans/trace-spans' import type { TraceSpan } from '@/lib/logs/types' import { LoopTool } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/subflows/loop/loop-config' import { ParallelTool } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/subflows/parallel/parallel-config' @@ -43,10 +44,7 @@ export function hasErrorInTree(span: TraceSpan): boolean { } export function hasUnhandledErrorInTree(span: TraceSpan): boolean { - if (span.status === 'error' && !span.errorHandled) return true - if (span.children?.length) return span.children.some(hasUnhandledErrorInTree) - if (span.toolCalls?.length && !span.errorHandled) return span.toolCalls.some((tc) => tc.error) - return false + return hasUnhandledError(span, { includeToolCalls: true }) } export function getBlockIconAndColor( diff --git a/apps/sim/app/workspace/[workspaceId]/logs/utils.ts b/apps/sim/app/workspace/[workspaceId]/logs/utils.ts index dc1fecf2ab3..01d336f082a 100644 --- a/apps/sim/app/workspace/[workspaceId]/logs/utils.ts +++ b/apps/sim/app/workspace/[workspaceId]/logs/utils.ts @@ -85,6 +85,7 @@ const TRIGGER_VARIANT_MAP: Record['va copilot: 'pink', mothership: 'pink', workflow: 'blue-secondary', + custom_block: 'blue-secondary', } interface StatusBadgeProps { diff --git a/apps/sim/blocks/custom/build-config.test.ts b/apps/sim/blocks/custom/build-config.test.ts index 83122e6a6d3..ac721946e75 100644 --- a/apps/sim/blocks/custom/build-config.test.ts +++ b/apps/sim/blocks/custom/build-config.test.ts @@ -94,9 +94,18 @@ describe('buildCustomBlockConfig', () => { expect(findSub(config, 'docs')?.multiple).toBe(true) }) - it('exposes the full result and hides plumbing when no outputs are curated', () => { + it('advertises no data fields — and no whole-result fallback — without curation', () => { const config = buildCustomBlockConfig(row, fields, { icon }) - expect(Object.keys(config.outputs).sort()).toEqual(['error', 'result', 'success']) + // Curation is required at publish, so an uncurated row exposes only the + // system fields. `result` must not come back: it would advertise the child's + // raw terminal state (agent toolCalls/thinking, nested workflow ids). + expect(Object.keys(config.outputs).sort()).toEqual([ + 'error', + 'errorRef', + 'errorType', + 'success', + ]) + expect(config.outputs.result).toBeUndefined() expect(config.outputs.childWorkflowId).toBeUndefined() expect(config.outputs.childTraceSpans).toBeUndefined() }) diff --git a/apps/sim/blocks/custom/build-config.ts b/apps/sim/blocks/custom/build-config.ts index 5ff63f41fce..403e788cdf7 100644 --- a/apps/sim/blocks/custom/build-config.ts +++ b/apps/sim/blocks/custom/build-config.ts @@ -63,13 +63,14 @@ export const RESERVED_PARAMS = new Set([ ]) /** - * Output names the block projects itself (`success`/`error` from `buildOutputs`, - * `cost` from the executor's billing aggregation). A user-named exposed output - * must never shadow these — an output literally named `cost` would clobber the - * billed cost. `result` is deliberately NOT reserved: it only exists as a system + * Output names the block projects itself: `success`/`error`/`errorType`/`errorRef` + * from `buildOutputs`, plus `cost` (kept reserved for forward compatibility now + * that the child bills its own run). A user-named exposed output must never + * shadow these. `result` is deliberately NOT reserved: it only exists as a system * field when no outputs are curated, which cannot co-occur with a named output. + * Compared lowercased, so every entry here must be lowercase. */ -export const RESERVED_OUTPUT_NAMES = new Set(['success', 'error', 'cost']) +export const RESERVED_OUTPUT_NAMES = new Set(['success', 'error', 'cost', 'errortype', 'errorref']) /** Whether an exposed-output name collides with a system output field. */ export function isReservedOutputName(name: string): boolean { @@ -208,13 +209,15 @@ function buildOutputs(exposed: CustomBlockOutput[] | undefined): BlockConfig['ou const outputs: BlockConfig['outputs'] = { success: { type: 'boolean', description: 'Execution success status' }, error: { type: 'string', description: 'Error message' }, + errorType: { type: 'string', description: 'Machine-readable failure class' }, + errorRef: { type: 'string', description: 'Opaque reference to the failed run' }, } - if (exposed && exposed.length > 0) { - for (const out of exposed) { - outputs[out.name] = { type: 'json', description: `Output: ${out.path}` } - } - } else { - outputs.result = { type: 'json', description: 'Workflow execution result' } + // No whole-`result` fallback: curation is required at publish, so every + // consumer-visible field is one the publisher chose. A legacy row with no + // curated outputs advertises no data fields and fails loudly at invocation + // rather than silently reverting to exposing the child's raw terminal state. + for (const out of exposed ?? []) { + outputs[out.name] = { type: 'json', description: `Output: ${out.path}` } } return outputs } diff --git a/apps/sim/executor/errors/boundary.ts b/apps/sim/executor/errors/boundary.ts new file mode 100644 index 00000000000..c25d45c1648 --- /dev/null +++ b/apps/sim/executor/errors/boundary.ts @@ -0,0 +1,54 @@ +/** + * Machine-readable class of a custom-block failure. Every member describes a + * fact the CONSUMER already knows or can act on — never the source workflow's + * blocks, tools, or upstream provider responses. + */ +export type CustomBlockErrorType = + | 'missing_inputs' + | 'not_deployed' + | 'unavailable' + | 'depth_limit' + /** + * The payer had no usage headroom, so the child never ran. Safe to surface: + * a custom block always resolves within the consumer's own organization, so + * the exhausted limit is their org's, not a foreign publisher's. + */ + | 'usage_limit' + /** The child run was cancelled (the invoking run aborted or was cancelled). */ + | 'cancelled' + | 'execution_failed' + +/** What a failed custom block tells its consumer. Leaks nothing about the source run. */ +export interface CustomBlockFailure { + /** Stable, machine-readable class of the failure. */ + errorType: CustomBlockErrorType + /** Opaque handle to the child run, so the publisher can find the exact failing execution. */ + ref?: string + /** Consumer-safe sentence. Generic unless the throw site was a {@link BoundarySafeError}. */ + message: string +} + +/** + * An error whose `message` names only the CALLER's own artifacts — the block + * they placed, its input labels, its deployment state — so it may cross an + * invocation boundary verbatim. + * + * Every error that is NOT of this class is opaque at the boundary. The default + * is fail-closed, so a `throw new Error(...)` added later anywhere in the + * custom-block path is redacted without anyone remembering to redact it. This + * replaces the older convention of throwing *before* the `try` block to dodge + * the catch's sanitizer, where redaction depended on lexical position. + */ +export class BoundarySafeError extends Error { + readonly errorType: CustomBlockErrorType + + constructor(options: { message: string; errorType: CustomBlockErrorType }) { + super(options.message) + this.name = 'BoundarySafeError' + this.errorType = options.errorType + } +} + +export function isBoundarySafeError(error: unknown): error is BoundarySafeError { + return error instanceof BoundarySafeError +} diff --git a/apps/sim/executor/errors/child-workflow-error.ts b/apps/sim/executor/errors/child-workflow-error.ts index 056c2ceebe6..9c314cd0ee4 100644 --- a/apps/sim/executor/errors/child-workflow-error.ts +++ b/apps/sim/executor/errors/child-workflow-error.ts @@ -1,4 +1,5 @@ import type { TraceSpan } from '@/lib/logs/types' +import type { CustomBlockFailure } from '@/executor/errors/boundary' import type { ExecutionResult } from '@/executor/types' interface ChildWorkflowErrorOptions { @@ -8,6 +9,12 @@ interface ChildWorkflowErrorOptions { executionResult?: ExecutionResult childWorkflowSnapshotId?: string childWorkflowInstanceId?: string + /** Child workflow names from this invocation down to the failing run, outermost first. */ + workflowChain?: string[] + /** The deepest non-workflow failure, already block-name-prefixed (`"Function 1: boom"`). */ + rootErrorMessage?: string + /** Consumer-safe failure descriptor. Set only at a custom-block invocation boundary. */ + consumerFacing?: CustomBlockFailure cause?: Error } @@ -21,6 +28,15 @@ export class ChildWorkflowError extends Error { readonly childWorkflowSnapshotId?: string /** Per-invocation unique ID used to correlate child block events with this workflow block. */ readonly childWorkflowInstanceId?: string + /** + * The chain of child workflow names from this invocation down to the failing + * run. Carried structurally so nesting never has to be recovered by parsing + * the formatted message. + */ + readonly workflowChain: string[] + /** The deepest non-workflow failure, without any workflow-chain prefixes. */ + readonly rootErrorMessage: string + readonly consumerFacing?: CustomBlockFailure constructor(options: ChildWorkflowErrorOptions) { super(options.message, { cause: options.cause }) @@ -30,9 +46,24 @@ export class ChildWorkflowError extends Error { this.executionResult = options.executionResult this.childWorkflowSnapshotId = options.childWorkflowSnapshotId this.childWorkflowInstanceId = options.childWorkflowInstanceId + this.workflowChain = options.workflowChain ?? [options.childWorkflowName] + this.rootErrorMessage = options.rootErrorMessage ?? options.message + this.consumerFacing = options.consumerFacing } static isChildWorkflowError(error: unknown): error is ChildWorkflowError { return error instanceof ChildWorkflowError } } + +/** + * Renders a nested child-workflow failure for display. Kept byte-identical to + * the format this error used to carry as a raw string, so persisted logs and + * the UI are unaffected by the move to structured fields. + */ +export function formatWorkflowChainMessage(chain: string[], rootError: string): string { + if (chain.length > 1) { + return `Workflow chain: ${chain.join(' → ')} | ${rootError}` + } + return `"${chain[0]}" failed: ${rootError}` +} diff --git a/apps/sim/executor/execution/block-executor.ts b/apps/sim/executor/execution/block-executor.ts index f4b12617509..105eb517863 100644 --- a/apps/sim/executor/execution/block-executor.ts +++ b/apps/sim/executor/execution/block-executor.ts @@ -21,6 +21,7 @@ import { DEFAULTS, EDGE, isSentinelBlockType, + isWorkflowBlockType, } from '@/executor/constants' import type { DAGNode } from '@/executor/dag/builder' import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' @@ -450,10 +451,25 @@ export class BlockExecutor { errorOutput.content = partialContent } + // Only real workflow blocks surface a child workflow name. A custom block's + // source workflow is never named to its consumer — and before the handler + // resolves the real name this field still holds the source workflow id, so + // an early throw (e.g. the call-chain depth limit) would leak it outright. if (ChildWorkflowError.isChildWorkflowError(error)) { - errorOutput.childWorkflowName = error.childWorkflowName - if (error.childWorkflowSnapshotId) { - errorOutput.childWorkflowSnapshotId = error.childWorkflowSnapshotId + if (isWorkflowBlockType(block.metadata?.id)) { + errorOutput.childWorkflowName = error.childWorkflowName + if (error.childWorkflowSnapshotId) { + errorOutput.childWorkflowSnapshotId = error.childWorkflowSnapshotId + } + } + // A custom block's consumer gets a machine-readable failure class and an + // opaque handle to the failed run — enough to branch on and to quote in a + // support request, without naming anything inside the source workflow. + if (error.consumerFacing) { + errorOutput.errorType = error.consumerFacing.errorType + if (error.consumerFacing.ref) { + errorOutput.errorRef = error.consumerFacing.ref + } } } diff --git a/apps/sim/executor/handlers/agent/agent-handler.ts b/apps/sim/executor/handlers/agent/agent-handler.ts index fc721b990f2..8ad1fa99370 100644 --- a/apps/sim/executor/handlers/agent/agent-handler.ts +++ b/apps/sim/executor/handlers/agent/agent-handler.ts @@ -945,6 +945,7 @@ export class AgentBlockHandler implements BlockHandler { workflowId: ctx.workflowId, workspaceId: ctx.workspaceId, userId: ctx.userId, + executionId: ctx.executionId, stream: streaming, messages: messages?.map(({ executionId, ...msg }) => msg), environmentVariables: normalizeStringRecord(ctx.environmentVariables), @@ -1028,6 +1029,10 @@ export class AgentBlockHandler implements BlockHandler { isDeployedContext: ctx.isDeployedContext, callChain: ctx.callChain, billingAttribution: ctx.metadata.billingAttribution, + // Reaches tool `_context` via `prepareToolExecution`, so a tool that starts + // its own child execution (a custom block) correlates and cancels against + // this real run instead of minting a phantom id. + executionId: ctx.executionId, reasoningEffort: providerRequest.reasoningEffort, verbosity: providerRequest.verbosity, thinkingLevel: providerRequest.thinkingLevel, diff --git a/apps/sim/executor/handlers/generic/generic-handler.ts b/apps/sim/executor/handlers/generic/generic-handler.ts index c6ee89dd633..8336bbb2ed3 100644 --- a/apps/sim/executor/handlers/generic/generic-handler.ts +++ b/apps/sim/executor/handlers/generic/generic-handler.ts @@ -3,6 +3,7 @@ import { toError } from '@sim/utils/errors' import { getBlock } from '@/blocks/index' import { isMcpTool } from '@/executor/constants' import type { BlockHandler, ExecutionContext } from '@/executor/types' +import { readStatusCode } from '@/executor/utils/errors' import type { SerializedBlock } from '@/serializer/types' import { executeTool } from '@/tools' import { getTool } from '@/tools/utils' @@ -93,6 +94,10 @@ export class GenericBlockHandler implements BlockHandler { blockName: block.metadata?.name || 'Unnamed Block', output: result.output || {}, timestamp: new Date().toISOString(), + // `executeTool` flattens a thrown error into a result, so Sim's own + // status (hosted-key 429/503) would be lost here. Carry it onto the + // error so `getExecutionErrorStatus` can still reach the API caller. + ...(typeof result.statusCode === 'number' ? { statusCode: result.statusCode } : {}), }) throw error @@ -107,8 +112,9 @@ export class GenericBlockHandler implements BlockHandler { errorMessage += `: ${block.metadata.name}` } - if (error.status) { - errorMessage += ` (Status: ${error.status})` + const statusCode = readStatusCode(error) + if (statusCode !== undefined) { + errorMessage += ` (Status: ${statusCode})` } error.message = errorMessage diff --git a/apps/sim/executor/handlers/workflow/custom-block-tool-runner.test.ts b/apps/sim/executor/handlers/workflow/custom-block-tool-runner.test.ts index edc83663b1c..39898e9148d 100644 --- a/apps/sim/executor/handlers/workflow/custom-block-tool-runner.test.ts +++ b/apps/sim/executor/handlers/workflow/custom-block-tool-runner.test.ts @@ -88,7 +88,7 @@ describe('runCustomBlockTool', () => { expect(res.error).toContain('not deployed') }) - it('rolls up already-incurred child cost when the run fails', async () => { + it('reports no cost on failure — the child session billed its own run', async () => { const err: any = new Error('child blew up') err.name = 'ChildWorkflowError' err.childTraceSpans = [{ id: 's1', name: 'child', type: 'agent', cost: { total: 0.25 } }] @@ -98,8 +98,7 @@ describe('runCustomBlockTool', () => { const res = await runCustomBlockTool({ blockType: 'custom_block_abc', _context: {} }) expect(res.success).toBe(false) - // Partial spend must not be recorded as zero-cost. - expect((res.output as any).cost.total).toBeGreaterThan(0) + expect(res.output).toEqual({}) }) it('rejects a missing block type without invoking the handler', async () => { @@ -108,3 +107,41 @@ describe('runCustomBlockTool', () => { expect(mockExecute).not.toHaveBeenCalled() }) }) + +describe('buildCustomBlockExecutionContext invoker identity', () => { + it("adopts the invoking run's ids so correlation names a real execution", () => { + const ctx = buildCustomBlockExecutionContext({ + workspaceId: 'ws-1', + executionId: 'agent-execution-id', + requestId: 'agent-request-id', + }) + + expect(ctx.executionId).toBe('agent-execution-id') + expect(ctx.metadata.executionId).toBe('agent-execution-id') + expect(ctx.metadata.requestId).toBe('agent-request-id') + }) + + it('falls back to generated ids when the caller supplies none', () => { + const ctx = buildCustomBlockExecutionContext({ workspaceId: 'ws-1' }) + + expect(ctx.executionId).toBeTruthy() + expect(ctx.metadata.requestId).toBeTruthy() + expect(ctx.executionId).not.toBe(ctx.metadata.requestId) + }) +}) + +describe('buildCustomBlockExecutionContext cancellation', () => { + it("adopts the agent tool loop's abort signal so the bridge has something to watch", () => { + const controller = new AbortController() + const ctx = buildCustomBlockExecutionContext( + { workspaceId: 'ws-1' }, + { abortSignal: controller.signal } + ) + + expect(ctx.abortSignal).toBe(controller.signal) + }) + + it('leaves the signal undefined when the caller has none', () => { + expect(buildCustomBlockExecutionContext({ workspaceId: 'ws-1' }).abortSignal).toBeUndefined() + }) +}) diff --git a/apps/sim/executor/handlers/workflow/custom-block-tool-runner.ts b/apps/sim/executor/handlers/workflow/custom-block-tool-runner.ts index bd9db63949c..ce8ca3b67c8 100644 --- a/apps/sim/executor/handlers/workflow/custom-block-tool-runner.ts +++ b/apps/sim/executor/handlers/workflow/custom-block-tool-runner.ts @@ -2,11 +2,7 @@ import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' -import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' -import { - aggregateChildCost, - WorkflowBlockHandler, -} from '@/executor/handlers/workflow/workflow-handler' +import { WorkflowBlockHandler } from '@/executor/handlers/workflow/workflow-handler' import type { ExecutionContext } from '@/executor/types' import type { SerializedBlock } from '@/serializer/types' import type { ToolResponse } from '@/tools/types' @@ -21,6 +17,15 @@ interface CustomBlockExecutorContext { callChain?: string[] isDeployedContext?: boolean billingAttribution?: BillingAttributionSnapshot + /** + * The INVOKING agent run's execution id. Server-set, never model-supplied. + * Without it the child's correlation would name an id no log row ever has, so + * a publisher could not trace the run back to the consumer execution — and the + * cancellation bridge would subscribe to an id nothing ever cancels. + */ + executionId?: string + /** The invoking run's request id, so both sides share one trace identifier. */ + requestId?: string } interface CustomBlockToolParams { @@ -33,8 +38,9 @@ interface CustomBlockToolParams { /** * Build a minimal top-level `ExecutionContext` for running a custom block as an - * agent tool. Every value comes from the server-set `_context` (LLM-proof) plus a - * fresh executionId. `WorkflowBlockHandler`'s custom-block path re-derives owner + * agent tool. Every value comes from the server-set `_context` (LLM-proof), + * including the invoking run's execution and request ids so the child's log + * correlation names a real execution. `WorkflowBlockHandler`'s path re-derives owner * identity, env, and billing from `getCustomBlockAuthority`, so this only needs the * fields that path reads — `workspaceId` (org-scopes the authority lookup), * `metadata` (read unconditionally at `executeCore`), and `callChain` (recursion @@ -42,9 +48,12 @@ interface CustomBlockToolParams { * scaffolding. Keep in sync with `WorkflowBlockHandler.executeCore`'s custom branch. */ export function buildCustomBlockExecutionContext( - context: CustomBlockExecutorContext + context: CustomBlockExecutorContext, + options: { abortSignal?: AbortSignal } = {} ): ExecutionContext { - const executionId = generateId() + // Prefer the invoking agent run's ids so correlation and cancellation both + // point at a real execution; fall back only when a caller could not supply them. + const executionId = context.executionId ?? generateId() return { workflowId: context.workflowId ?? 'custom-block-tool', workspaceId: context.workspaceId, @@ -54,6 +63,9 @@ export function buildCustomBlockExecutionContext( // Inherit the accumulated chain so the handler appends + validates depth; // resetting to [] would let a self-referential custom block recurse unbounded. callChain: context.callChain ?? [], + // Without this the child's cancellation bridge has nothing to abort on: + // the agent tool loop owns the only signal reaching this path. + abortSignal: options.abortSignal, environmentVariables: {}, blockStates: new Map(), executedBlocks: new Set(), @@ -65,7 +77,7 @@ export function buildCustomBlockExecutionContext( // custom-block path; `duration` is the sole required field on the metadata type. metadata: { duration: 0, - requestId: generateId(), + requestId: context.requestId ?? generateId(), executionId, workflowId: context.workflowId, workspaceId: context.workspaceId, @@ -80,19 +92,25 @@ export function buildCustomBlockExecutionContext( * Runs a published custom block (deploy-as-block) as an Agent tool, in-process via * `WorkflowBlockHandler` — the same invocation boundary the canvas uses — so * authority (org-scoped owner identity, latest deployment, curated outputs, - * required-input enforcement, cost roll-up) is resolved server-side from the block - * type. No HTTP hop and no body-field trust: the block type + consumer workspace - * come from the server-set `_context`, not the model. + * required-input enforcement) is resolved server-side from the block type, and the + * child's spend is billed to the source workspace by its own logging session. No + * HTTP hop and no body-field trust: the block type + consumer workspace come from + * the server-set `_context`, not the model. * * Lives in a server-only module (dynamic-imported by `executeTool`) so the * client-bundled tool registry never pulls in the executor/db dependency graph. */ -export async function runCustomBlockTool(params: CustomBlockToolParams): Promise { +export async function runCustomBlockTool( + params: CustomBlockToolParams, + options: { abortSignal?: AbortSignal } = {} +): Promise { if (!params.blockType) { return { success: false, output: {}, error: 'Missing custom block type' } } - const ctx = buildCustomBlockExecutionContext(params._context ?? {}) + const ctx = buildCustomBlockExecutionContext(params._context ?? {}, { + abortSignal: options.abortSignal, + }) const block: SerializedBlock = { id: generateId(), position: { x: 0, y: 0 }, @@ -108,25 +126,16 @@ export async function runCustomBlockTool(params: CustomBlockToolParams): Promise inputMapping: params.inputMapping, }) // Custom blocks never stream (no `onStream` on the synthetic ctx), so the - // handler always returns the projected BlockOutput object (with `cost.total`). + // handler always returns the projected BlockOutput object. const normalized: Record = output && typeof output === 'object' && !Array.isArray(output) ? output : { result: output } return { success: true, output: normalized } } catch (error) { - // The handler throws a consumer-safe `ChildWorkflowError` on failure. Its trace - // spans are the only carrier of spend the child already incurred before failing, - // so roll that up onto the failed result too — otherwise a partially-run child is - // recorded as zero-cost. + // The handler throws a consumer-safe `ChildWorkflowError` on failure. The + // child's own logging session already billed whatever it spent before failing, + // so nothing is rolled up here. const message = getErrorMessage(error, 'Custom block execution failed') - const failedChildSpans = ChildWorkflowError.isChildWorkflowError(error) - ? error.childTraceSpans - : [] - const childCost = aggregateChildCost(failedChildSpans) logger.info('Custom block tool execution failed', { blockType: params.blockType, message }) - return { - success: false, - output: childCost > 0 ? { cost: { total: childCost } } : {}, - error: message, - } + return { success: false, output: {}, error: message } } } diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts index 4e9b8c9831a..9acbc2a1905 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.test.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.test.ts @@ -2,6 +2,7 @@ import { environmentUtilsMockFns, resetEnvironmentUtilsMock } from '@sim/testing import { afterAll, beforeAll, beforeEach, describe, expect, it, type Mock, vi } from 'vitest' import { getBlock } from '@/blocks/registry' import { BlockType } from '@/executor/constants' +import { BoundarySafeError } from '@/executor/errors/boundary' import { findMissingRequiredCustomBlockInputs, remapCustomBlockInputKeys, @@ -16,14 +17,63 @@ const { mockResolveBillingAttribution, mockGetCustomBlockAuthority, mockGetUserEmailById, + mockAdmitCustomBlockChildExecution, + mockTrackChildRun, + mockBuildTraceSpans, + mockSafeStart, + mockSafeComplete, + mockSafeCompleteWithError, + mockSafeCompleteWithCancellation, + mockDispose, executorOptions, + loggingSessionArgs, } = vi.hoisted(() => ({ mockExecutorExecute: vi.fn(), mockCreateSnapshot: vi.fn(), mockResolveBillingAttribution: vi.fn(), mockGetCustomBlockAuthority: vi.fn(), mockGetUserEmailById: vi.fn(), + mockAdmitCustomBlockChildExecution: vi.fn(), + mockTrackChildRun: vi.fn(), + mockBuildTraceSpans: vi.fn(), + mockSafeStart: vi.fn(), + mockSafeComplete: vi.fn(), + mockSafeCompleteWithError: vi.fn(), + mockSafeCompleteWithCancellation: vi.fn(), + mockDispose: vi.fn(), executorOptions: [] as Array>, + loggingSessionArgs: [] as Array, +})) + +vi.mock('@/lib/logs/execution/logging-session', () => ({ + LoggingSession: class { + constructor(...args: any[]) { + loggingSessionArgs.push(args) + } + safeStart = mockSafeStart + safeComplete = mockSafeComplete + safeCompleteWithError = mockSafeCompleteWithError + safeCompleteWithCancellation = mockSafeCompleteWithCancellation + onBlockStart = vi.fn() + onBlockComplete = vi.fn() + }, +})) + +vi.mock('@/lib/logs/execution/trace-spans/trace-spans', () => ({ + buildTraceSpans: mockBuildTraceSpans, +})) + +vi.mock('@/lib/workflows/custom-blocks/child-execution', () => ({ + admitCustomBlockChildExecution: mockAdmitCustomBlockChildExecution, + trackChildRun: mockTrackChildRun, + buildCustomBlockCorrelation: (params: Record) => + params.invokerExecutionId + ? { source: 'custom_block', executionId: params.invokerExecutionId } + : undefined, + createChildCancellationSignal: () => ({ + signal: new AbortController().signal, + dispose: mockDispose, + }), })) vi.mock('@/executor', () => ({ @@ -167,6 +217,10 @@ describe('WorkflowBlockHandler', () => { // Reset all mocks vi.clearAllMocks() executorOptions.length = 0 + loggingSessionArgs.length = 0 + mockSafeStart.mockResolvedValue(true) + mockAdmitCustomBlockChildExecution.mockResolvedValue(undefined) + mockBuildTraceSpans.mockReturnValue({ traceSpans: [], totalDuration: 0 }) // Setup default fetch mock mockFetch.mockResolvedValue({ @@ -402,7 +456,7 @@ describe('WorkflowBlockHandler', () => { workflowId: 'source-workflow-id', organizationId: 'org-1', ownerUserId: 'owner-9', - exposedOutputs: [], + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], requiredInputIds: [], }) mockGetPersonalAndWorkspaceEnv.mockResolvedValue({ @@ -465,7 +519,7 @@ describe('WorkflowBlockHandler', () => { workflowId: 'source-workflow-id', organizationId: 'org-1', ownerUserId: 'owner-9', - exposedOutputs: [], + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], requiredInputIds: [], }) mockGetPersonalAndWorkspaceEnv.mockResolvedValue({ @@ -566,7 +620,7 @@ describe('WorkflowBlockHandler', () => { workflowId: 'source-workflow-id', organizationId: 'org-1', ownerUserId: 'owner-9', - exposedOutputs: [], + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], requiredInputIds: [], }) mockGetPersonalAndWorkspaceEnv.mockResolvedValue({ @@ -963,6 +1017,425 @@ describe('WorkflowBlockHandler', () => { }) }) + describe('custom block child execution', () => { + const customBlock = () => ({ + ...mockBlock, + metadata: { id: 'custom_block_abc', name: 'Published Block' }, + }) + + function customBlockContext(overrides: Record = {}) { + return { + ...mockContext, + workspaceId: 'workspace-consumer', + executionId: 'parent-execution-id', + metadata: { ...mockContext.metadata, requestId: 'req-1' }, + ...overrides, + } as unknown as ExecutionContext + } + + beforeEach(() => { + mockGetCustomBlockAuthority.mockResolvedValue({ + workflowId: 'source-workflow-id', + organizationId: 'org-1', + ownerUserId: 'owner-9', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + requiredInputIds: [], + }) + mockGetPersonalAndWorkspaceEnv.mockResolvedValue({ + personalDecrypted: {}, + workspaceDecrypted: {}, + personalEncrypted: { SECRET: 'enc' }, + workspaceEncrypted: {}, + }) + mockResolveBillingAttribution.mockResolvedValue({ + actorUserId: 'owner-9', + workspaceId: 'workspace-source', + }) + mockFetch.mockImplementation(async (url: unknown) => { + if (String(url).includes('/deployed')) { + return { + ok: true, + json: () => + Promise.resolve({ + data: { deployedState: { blocks: {}, edges: [], loops: {}, parallels: {} } }, + }), + } + } + return { + ok: true, + json: () => + Promise.resolve({ + data: { name: 'Source Workflow', workspaceId: 'workspace-source', variables: {} }, + }), + } + }) + mockCreateSnapshot.mockResolvedValue({ snapshot: { id: 'snapshot-1' } }) + mockExecutorExecute.mockResolvedValue({ success: true, output: { data: 'ok' } }) + }) + + it('opens a session on the source workflow with a fresh id and no base charge', async () => { + await handler.execute(customBlockContext(), customBlock(), {}) + + expect(loggingSessionArgs).toHaveLength(1) + const [workflowId, executionId, trigger, requestId, reservationId, options] = + loggingSessionArgs[0] + expect(workflowId).toBe('source-workflow-id') + expect(executionId).not.toBe('parent-execution-id') + expect(reservationId).toBe(executionId) + expect(trigger).toBe('custom_block') + expect(requestId).toBe('req-1') + expect(options).toEqual({ baseExecutionCharge: 0 }) + }) + + it('starts the session against the source workspace and payer', async () => { + await handler.execute(customBlockContext(), customBlock(), {}) + + expect(mockSafeStart).toHaveBeenCalledTimes(1) + const params = mockSafeStart.mock.calls[0][0] + expect(params.workspaceId).toBe('workspace-source') + expect(params.actorUserId).toBe('owner-9') + expect(params.billingAttribution).toEqual({ + actorUserId: 'owner-9', + workspaceId: 'workspace-source', + }) + expect(params.variables).toEqual({ SECRET: 'enc' }) + expect(params.triggerData.correlation).toEqual({ + source: 'custom_block', + executionId: 'parent-execution-id', + }) + }) + + it('admits against the source payer before executing', async () => { + await handler.execute(customBlockContext(), customBlock(), {}) + + expect(mockAdmitCustomBlockChildExecution).toHaveBeenCalledWith({ + actorUserId: 'owner-9', + workspaceId: 'workspace-source', + }) + }) + + it('does not execute when admission is denied', async () => { + mockAdmitCustomBlockChildExecution.mockRejectedValue(new Error('no headroom')) + + await expect(handler.execute(customBlockContext(), customBlock(), {})).rejects.toThrow() + + expect(mockExecutorExecute).not.toHaveBeenCalled() + expect(loggingSessionArgs).toHaveLength(0) + }) + + it('runs the child under its own execution id but keeps the parent readable', async () => { + const ctx = customBlockContext() + await handler.execute(ctx, customBlock(), {}) + + const extensions = executorOptions[0].contextExtensions + expect(extensions.executionId).not.toBe('parent-execution-id') + expect(extensions.largeValueExecutionIds).toContain('parent-execution-id') + expect(ctx.largeValueExecutionIds).toContain(extensions.executionId) + }) + + it('shares one large-value id list so nested custom blocks propagate upward', async () => { + const ctx = customBlockContext() + await handler.execute(ctx, customBlock(), {}) + + const childIds = executorOptions[0].contextExtensions.largeValueExecutionIds + // Same array instance, not a copy — that is what lets a nested custom + // block's grandchild id reach the top-level invoker. + expect(childIds).toBe(ctx.largeValueExecutionIds) + + // Simulate a nested custom block appending its own child id deeper down. + childIds.push('grandchild-execution-id') + expect(ctx.largeValueExecutionIds).toContain('grandchild-execution-id') + }) + + it('does not duplicate ids across repeated invocations', async () => { + const ctx = customBlockContext() + await handler.execute(ctx, customBlock(), {}) + await handler.execute(ctx, customBlock(), {}) + + const ids = ctx.largeValueExecutionIds as string[] + expect(new Set(ids).size).toBe(ids.length) + expect(ids.filter((id) => id === 'parent-execution-id')).toHaveLength(1) + }) + + it('never forwards the consumer SSE callbacks into the source run', async () => { + const ctx = customBlockContext({ + onBlockStart: vi.fn(), + onBlockComplete: vi.fn(), + onStream: vi.fn(), + onChildWorkflowInstanceReady: vi.fn(), + }) + + await handler.execute(ctx, customBlock(), {}) + + const extensions = executorOptions[0].contextExtensions + expect(extensions.onStream).toBeUndefined() + expect(extensions.onChildWorkflowInstanceReady).toBeUndefined() + expect(extensions.childWorkflowContext).toBeUndefined() + expect(ctx.onChildWorkflowInstanceReady).not.toHaveBeenCalled() + // Block markers exist, but they belong to the CHILD's session. + expect(extensions.onBlockStart).toBeTypeOf('function') + expect(extensions.onBlockStart).not.toBe(ctx.onBlockStart) + expect(extensions.onBlockComplete).not.toBe(ctx.onBlockComplete) + }) + + it('completes the child session and disposes the cancellation bridge', async () => { + await handler.execute(customBlockContext(), customBlock(), {}) + + expect(mockSafeComplete).toHaveBeenCalledTimes(1) + expect(mockSafeCompleteWithError).not.toHaveBeenCalled() + expect(mockDispose).toHaveBeenCalledTimes(1) + }) + + it('records a cancelled child through the cancellation path', async () => { + // Production shape: the engine reports cancellation as `success: false` + // plus `status: 'cancelled'` on the ExecutionResult (never on metadata). + mockExecutorExecute.mockResolvedValue({ + success: false, + output: {}, + status: 'cancelled', + }) + + await handler.execute(customBlockContext(), customBlock(), {}).catch(() => {}) + + expect(mockSafeCompleteWithCancellation).toHaveBeenCalledTimes(1) + expect(mockSafeComplete).not.toHaveBeenCalled() + // Already finalized as cancelled — must not be re-completed as an error. + expect(mockSafeCompleteWithError).not.toHaveBeenCalled() + }) + + it('tells the consumer a cancellation was a cancellation, not a generic failure', async () => { + mockExecutorExecute.mockResolvedValue({ + success: false, + output: {}, + status: 'cancelled', + }) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('cancelled') + expect(error.message).toBe('Custom block execution was cancelled') + }) + + it('records the real failure on the child log and hides it from the consumer', async () => { + mockExecutorExecute.mockRejectedValue(new Error('Function 1: secret internals blew up')) + + await expect(handler.execute(customBlockContext(), customBlock(), {})).rejects.toMatchObject({ + message: expect.stringContaining('Custom block execution failed'), + }) + + expect(mockSafeCompleteWithError).toHaveBeenCalledTimes(1) + expect(mockSafeCompleteWithError.mock.calls[0][0].error.message).toBe( + 'Function 1: secret internals blew up' + ) + expect(mockDispose).toHaveBeenCalledTimes(1) + }) + + it('gives the consumer an opaque ref and error class, never the source detail', async () => { + mockExecutorExecute.mockRejectedValue(new Error('Function 1: secret internals blew up')) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('execution_failed') + expect(error.consumerFacing.ref).toBeDefined() + expect(error.message).toContain(error.consumerFacing.ref) + expect(error.message).not.toContain('secret internals') + expect(error.childWorkflowName).toBe('Published Block') + expect(error.childTraceSpans).toEqual([]) + expect(error.executionResult).toBeUndefined() + // The chain is severed at the trust boundary. + expect(error.cause).toBeUndefined() + }) + + it('registers the child run BEFORE executing it, and settles it when done', async () => { + // Ordering is the whole point: a cancelled parent drains while the child is + // still inside `execute`, so registering after it would find nothing. + let registeredBeforeExecute = false + mockExecutorExecute.mockImplementation(async () => { + registeredBeforeExecute = mockTrackChildRun.mock.calls.length === 1 + return { success: true, output: { data: 'ok' } } + }) + + await handler.execute(customBlockContext(), customBlock(), {}) + + expect(registeredBeforeExecute).toBe(true) + const [invokerId, childRun] = mockTrackChildRun.mock.calls[0] + expect(invokerId).toBe('parent-execution-id') + // Settled in `finally`, so the invoking run's drain can complete. + await expect(childRun).resolves.toBeUndefined() + }) + + it('settles the registered child run on the failure path too', async () => { + mockExecutorExecute.mockRejectedValue(new Error('boom')) + + await handler.execute(customBlockContext(), customBlock(), {}).catch(() => {}) + + expect(mockTrackChildRun).toHaveBeenCalledTimes(1) + expect(mockTrackChildRun.mock.calls[0][0]).toBe('parent-execution-id') + await expect(mockTrackChildRun.mock.calls[0][1]).resolves.toBeUndefined() + }) + + it('never leaks the source workflow name when the child returns success: false', async () => { + mockExecutorExecute.mockResolvedValue({ + success: false, + output: {}, + error: 'Function 1: internal detail', + }) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.message).not.toContain('Source Workflow') + expect(error.message).not.toContain('internal detail') + expect(error.consumerFacing.errorType).toBe('execution_failed') + expect(error.childWorkflowName).toBe('Published Block') + }) + + it('fails loudly on a legacy row with no curated outputs', async () => { + // Curation is required at publish; a pre-rule row must not silently fall + // back to exposing the child's raw terminal state. + mockGetCustomBlockAuthority.mockResolvedValue({ + workflowId: 'source-workflow-id', + organizationId: 'org-1', + ownerUserId: 'owner-9', + exposedOutputs: [], + requiredInputIds: [], + }) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('unavailable') + expect(error.message).toContain('re-publish') + expect(mockExecutorExecute).not.toHaveBeenCalled() + }) + + it('classifies an unavailable block so consumers can branch on it', async () => { + mockGetCustomBlockAuthority.mockResolvedValue(null) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('unavailable') + expect(error.message).toBe('This custom block is no longer available') + }) + + it('classifies an admission denial as a usage limit, not a generic failure', async () => { + // `CustomBlockAdmissionError` is a `BoundarySafeError` of this type; the + // class itself is covered in child-execution.test.ts. + mockAdmitCustomBlockChildExecution.mockRejectedValue( + new BoundarySafeError({ + errorType: 'usage_limit', + message: 'Organization usage limit exceeded', + }) + ) + + const error = await handler + .execute(customBlockContext(), customBlock(), {}) + .catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('usage_limit') + expect(error.message).toBe('Organization usage limit exceeded') + }) + + it('keeps the depth-limit classification instead of collapsing to generic', async () => { + const ctx = customBlockContext({ callChain: Array.from({ length: 30 }, (_, i) => `wf-${i}`) }) + + const error = await handler.execute(ctx, customBlock(), {}).catch((e: any) => e) + + expect(error.consumerFacing.errorType).toBe('depth_limit') + expect(error.childWorkflowName).toBe('Published Block') + }) + + it('surfaces a missing-required-input failure verbatim', async () => { + mockGetCustomBlockAuthority.mockResolvedValue({ + workflowId: 'source-workflow-id', + organizationId: 'org-1', + ownerUserId: 'owner-9', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + requiredInputIds: ['field-1'], + }) + mockFetch.mockImplementation(async (url: unknown) => { + if (String(url).includes('/deployed')) { + return { + ok: true, + json: () => + Promise.resolve({ + data: { + deployedState: { + blocks: { + starter: { + type: 'start_trigger', + subBlocks: { + inputFormat: { + value: [{ id: 'field-1', name: 'Username', type: 'string' }], + }, + }, + }, + }, + edges: [], + loops: {}, + parallels: {}, + }, + }, + }), + } + } + return { + ok: true, + json: () => + Promise.resolve({ + data: { name: 'Source Workflow', workspaceId: 'workspace-source', variables: {} }, + }), + } + }) + + const error = await handler + .execute(customBlockContext(), customBlock(), { inputMapping: '{}' }) + .catch((e: any) => e) + + expect(error.message).toContain('missing required fields') + expect(error.message).toContain('Username') + expect(error.consumerFacing.errorType).toBe('missing_inputs') + expect(mockExecutorExecute).not.toHaveBeenCalled() + }) + + it('leaves regular workflow blocks entirely alone', async () => { + const ctx = { + ...mockContext, + workspaceId: 'workspace-1', + executionId: 'parent-execution-id', + onBlockStart: vi.fn(), + onStream: vi.fn(), + } as unknown as ExecutionContext + mockFetch.mockResolvedValue({ + ok: true, + json: () => + Promise.resolve({ + data: { + name: 'Child Workflow', + workspaceId: 'workspace-1', + state: { blocks: [], edges: [], loops: {}, parallels: {} }, + }, + }), + }) + + await handler.execute(ctx, mockBlock, { workflowId: 'child-workflow-id' }) + + expect(loggingSessionArgs).toHaveLength(0) + const extensions = executorOptions[0].contextExtensions + expect(extensions.executionId).toBe('parent-execution-id') + expect(extensions.onStream).toBe(ctx.onStream) + expect(extensions.childWorkflowContext).toBeDefined() + }) + }) + describe('projectCustomBlockOutput', () => { const childResult = { success: true, @@ -971,34 +1444,30 @@ describe('WorkflowBlockHandler', () => { } it('maps each curated output to its named field plus system fields', () => { - const result = (handler as any).projectCustomBlockOutput( - childResult, - [{ blockId: 'b1', path: 'data.x', name: 'answer' }], - 0.5 - ) + const result = (handler as any).projectCustomBlockOutput(childResult, [ + { blockId: 'b1', path: 'data.x', name: 'answer' }, + ]) - expect(result).toEqual({ answer: 42, success: true, cost: { total: 0.5 } }) + expect(result).toEqual({ answer: 42, success: true }) }) - it('never lets an exposed output named cost clobber the billed cost', () => { - const result = (handler as any).projectCustomBlockOutput( - childResult, - [{ blockId: 'b1', path: 'price', name: 'cost' }], - 0.5 - ) + it('never reports cost on the consumer block — the child bills its own run', () => { + const result = (handler as any).projectCustomBlockOutput(childResult, [ + { blockId: 'b1', path: 'data.x', name: 'answer' }, + ]) - expect(result.cost).toEqual({ total: 0.5 }) - expect(result.success).toBe(true) + expect(result.cost).toBeUndefined() }) - it('exposes the whole child result when no outputs are curated', () => { - const result = (handler as any).projectCustomBlockOutput(childResult, [], 0.5) + it('never dumps the child result when no outputs are curated', () => { + // Curation is required at publish and guarded at invocation, so this path + // is unreachable in production — but it must not fall back to exposing the + // terminal block's raw state (agent toolCalls/thinking, nested workflow + // ids) if it is ever reached. + const result = (handler as any).projectCustomBlockOutput(childResult, []) - expect(result).toEqual({ - success: true, - result: { data: 'whole result' }, - cost: { total: 0.5 }, - }) + expect(result).toEqual({ success: true }) + expect((result as any).result).toBeUndefined() }) }) }) diff --git a/apps/sim/executor/handlers/workflow/workflow-handler.ts b/apps/sim/executor/handlers/workflow/workflow-handler.ts index dddd1a19010..07648a6c368 100644 --- a/apps/sim/executor/handlers/workflow/workflow-handler.ts +++ b/apps/sim/executor/handlers/workflow/workflow-handler.ts @@ -1,22 +1,36 @@ import { createLogger } from '@sim/logger' -import { getErrorMessage } from '@sim/utils/errors' +import { findCause, getErrorMessage, toError } from '@sim/utils/errors' import { generateId } from '@sim/utils/id' import { isRecordLike } from '@sim/utils/object' import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution' import { getPersonalAndWorkspaceEnv } from '@/lib/environment/utils' import { buildNextCallChain, validateCallChain } from '@/lib/execution/call-chain' -import { calculateCostSummary } from '@/lib/logs/execution/logging-factory' +import { LoggingSession } from '@/lib/logs/execution/logging-session' import { snapshotService } from '@/lib/logs/execution/snapshot/service' import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' import type { TraceSpan } from '@/lib/logs/types' import { getUserEmailById } from '@/lib/users/queries' +import { + admitCustomBlockChildExecution, + buildCustomBlockCorrelation, + createChildCancellationSignal, + trackChildRun, +} from '@/lib/workflows/custom-blocks/child-execution' import { getCustomBlockAuthority } from '@/lib/workflows/custom-blocks/operations' import { extractInputFieldsFromBlocks } from '@/lib/workflows/input-format' import { type CustomBlockOutput, isCustomBlockType } from '@/blocks/custom/build-config' import type { BlockOutput } from '@/blocks/types' import { Executor } from '@/executor' import { BlockType, DEFAULTS, HTTP } from '@/executor/constants' -import { ChildWorkflowError } from '@/executor/errors/child-workflow-error' +import { + BoundarySafeError, + type CustomBlockErrorType, + isBoundarySafeError, +} from '@/executor/errors/boundary' +import { + ChildWorkflowError, + formatWorkflowChainMessage, +} from '@/executor/errors/child-workflow-error' import type { WorkflowNodeMetadata } from '@/executor/execution/types' import { type BlockHandler, @@ -37,6 +51,13 @@ import type { SerializedBlock } from '@/serializer/types' const logger = createLogger('WorkflowBlockHandler') +/** + * Trigger recorded on a custom block child's own log row. Distinct from + * `'workflow'` (which means "invoked by another workflow") so the publisher can + * tell who used their block apart from their own nested runs. + */ +const CUSTOM_BLOCK_TRIGGER = 'custom_block' + /** Read a dot-path (e.g. `content.text`) out of a block output object. */ function getValueAtPath(source: unknown, path: string): unknown { return path.split('.').reduce((acc, key) => { @@ -68,18 +89,6 @@ function readSeededStartRunMetadata(ctx: ExecutionContext): StartBlockRunMetadat * name. Legacy fields without an id are keyed by name and pass through unchanged. * Keys that match no current field are dropped. */ -/** - * A consumer left a publisher-required custom block input empty. The message is - * consumer-safe (it names only the block's own input labels), so the catch's - * custom-block sanitizer rethrows it verbatim instead of the generic failure. - */ -export class CustomBlockMissingInputsError extends Error { - constructor(message: string) { - super(message) - this.name = 'CustomBlockMissingInputsError' - } -} - /** * Names of publisher-required custom block inputs the consumer left empty, checked * against the child's LIVE deployed Start fields — a required override whose field @@ -130,40 +139,6 @@ export function remapCustomBlockInputKeys( return remapped } -/** - * Canonical hosted-key spend of a child run: the model/tool cost the way the - * parent bills it (recursing nested/iteration spans and de-duping model - * breakdowns), minus the base execution charge the parent applies once itself. - * A naive top-level `cost.total` sum undercounts when spend sits on nested children. - */ -export function aggregateChildCost(childTraceSpans: TraceSpan[]): number { - if (childTraceSpans.length === 0) return 0 - const summary = calculateCostSummary(childTraceSpans) - return Math.max(0, summary.totalCost - summary.baseExecutionCharge) -} - -/** - * A single cost-only span so a FAILED custom block still bills the hosted-key spend - * its child already consumed (`block-executor` bills `error.childTraceSpans`), - * without exposing any of the source workflow's internal spans. Empty when free. - */ -function buildCostCarrierSpans(childCost: number, blockName: string, type: string): TraceSpan[] { - if (childCost <= 0) return [] - const now = new Date().toISOString() - return [ - { - id: generateId(), - name: blockName, - type, - duration: 0, - startTime: now, - endTime: now, - status: 'error', - cost: { total: childCost }, - }, - ] -} - type WorkflowTraceSpan = TraceSpan & { metadata?: Record children?: WorkflowTraceSpan[] @@ -216,6 +191,11 @@ export class WorkflowBlockHandler implements BlockHandler { // identity a normal deployed API/schedule/webhook run uses — so a cross-workspace // consumer needs no permission on the source workflow. Owner deletion cascade- // deletes the workflow → the custom_block row, so the block never orphans. + // Unique ID per invocation — used to correlate child block events with this specific + // workflow block execution, preventing cross-iteration child mixing in loop contexts. + // Generated up front so the pre-`try` boundary failures below can carry it too. + const instanceId = generateId() + let workflowId = inputs.workflowId let loadUserId = ctx.userId let exposedOutputs: CustomBlockOutput[] = [] @@ -223,12 +203,38 @@ export class WorkflowBlockHandler implements BlockHandler { if (isCustomBlock) { const authority = await getCustomBlockAuthority(blockTypeId as string, ctx.workspaceId) if (!authority) { - throw new Error('This custom block is no longer available') + // Routed through the same builder as in-`try` failures so the consumer gets + // `errorType: 'unavailable'` and can branch on it, rather than a bare message. + throw this.buildBoundaryFailure( + new BoundarySafeError({ + errorType: 'unavailable', + message: 'This custom block is no longer available', + }), + block, + instanceId, + undefined + ) } workflowId = authority.workflowId loadUserId = authority.ownerUserId exposedOutputs = authority.exposedOutputs requiredInputIds = authority.requiredInputIds + + // Curation is required at publish, so this only trips on a row that + // predates that rule. Fail loudly rather than fall back to exposing the + // child's raw terminal state, which is what the boundary exists to stop. + if (exposedOutputs.length === 0) { + throw this.buildBoundaryFailure( + new BoundarySafeError({ + errorType: 'unavailable', + message: + 'This custom block exposes no outputs. Its publisher must re-publish it with at least one output selected.', + }), + block, + instanceId, + undefined + ) + } } if (!workflowId) { @@ -240,33 +246,52 @@ export class WorkflowBlockHandler implements BlockHandler { let childWorkflowName = workflowId - // Unique ID per invocation — used to correlate child block events with this specific - // workflow block execution, preventing cross-iteration child mixing in loop contexts. - const instanceId = generateId() - const childCallChain = buildNextCallChain(ctx.callChain || [], workflowId) const depthError = validateCallChain(childCallChain) if (depthError) { + // The depth message names nothing about the source, so it is consumer-safe + // for a custom block too — but `childWorkflowName` is still the source + // workflow id at this point, so a custom block must carry its own block + // name instead of leaking that id across the invocation boundary. throw new ChildWorkflowError({ message: depthError, - childWorkflowName, + childWorkflowName: isCustomBlock + ? block.metadata?.name || 'Custom block' + : childWorkflowName, + childWorkflowInstanceId: instanceId, + ...(isCustomBlock + ? { consumerFacing: { errorType: 'depth_limit' as const, message: depthError } } + : {}), }) } - // A custom block runs the source's latest deployment; if the source has been - // undeployed there's nothing to run. Check + throw a clear, consumer-safe - // reason BEFORE the try so the catch's generic sanitizer doesn't mask it (the - // message names no source internals). The block still renders (its schema comes - // from stored curated inputs), so this is the only failure mode to surface. - if (isCustomBlock) { - const deployed = await this.checkChildDeployment(workflowId, loadUserId) - if (!deployed) { - throw new Error('This block’s workflow is not deployed. Redeploy it to use this block.') - } - } - let childWorkflowSnapshotId: string | undefined + /** Set for custom blocks only: the child's own execution id / log row. */ + let childExecutionId: string | undefined + let childSession: LoggingSession | undefined + let childSessionStarted = false + /** Set once the child's session reached a terminal state, so the catch doesn't re-complete it. */ + let childSessionFinalized = false + /** Large-value id list shared with the child (and any nested custom blocks). */ + let sharedLargeValueIds: string[] | undefined + let childCancellation: { signal: AbortSignal; dispose: () => void } | undefined + /** Settled in `finally` once the child is fully done — see `trackChildRun`. */ + let settleChildRun: (() => void) | undefined try { + // A custom block runs the source's latest deployment; if the source has been + // undeployed there's nothing to run. `BoundarySafeError` marks the message as + // safe to cross the invocation boundary verbatim (it names no source + // internals), so the catch forwards it instead of the generic failure. + if (isCustomBlock) { + const deployed = await this.checkChildDeployment(workflowId, loadUserId) + if (!deployed) { + throw new BoundarySafeError({ + errorType: 'not_deployed', + message: 'This block’s workflow is not deployed. Redeploy it to use this block.', + }) + } + } + if (useDeployed && !isCustomBlock) { const hasActiveDeployment = await this.checkChildDeployment(workflowId, loadUserId) if (!hasActiveDeployment) { @@ -336,9 +361,10 @@ export class WorkflowBlockHandler implements BlockHandler { childWorkflowInput ) if (missing.length > 0) { - throw new CustomBlockMissingInputsError( - `${block.metadata?.name || 'Custom block'} is missing required fields: ${missing.join(', ')}` - ) + throw new BoundarySafeError({ + errorType: 'missing_inputs', + message: `${block.metadata?.name || 'Custom block'} is missing required fields: ${missing.join(', ')}`, + }) } } @@ -349,9 +375,13 @@ export class WorkflowBlockHandler implements BlockHandler { childWorkflowSnapshotId = childSnapshotResult.snapshot.id const childDepth = (ctx.childWorkflowContext?.depth ?? 0) + 1 - const shouldPropagateCallbacks = childDepth <= DEFAULTS.MAX_SSE_CHILD_DEPTH + // A custom block is an invocation boundary: forwarding the consumer's SSE + // callbacks into the source run would stream the publisher's block names, + // inputs, outputs, and raw agent tokens to the consumer's browser — where + // the terminal silently drops them, so the leak is invisible in the UI. + const shouldPropagateCallbacks = !isCustomBlock && childDepth <= DEFAULTS.MAX_SSE_CHILD_DEPTH - if (!shouldPropagateCallbacks) { + if (!shouldPropagateCallbacks && !isCustomBlock) { logger.info('Dropping SSE callbacks beyond max child depth', { childDepth, maxDepth: DEFAULTS.MAX_SSE_CHILD_DEPTH, @@ -376,13 +406,15 @@ export class WorkflowBlockHandler implements BlockHandler { // A custom block is an invocation boundary: the child runs under the SOURCE // workflow owner's identity, workspace, and environment — not the consumer's — // so it resolves credentials/integrations/env exactly as published and the - // consumer needs no access to any of them. (Billing still lands on the - // consumer's org, aggregated onto the block above.) Regular workflow blocks - // keep running in the parent's context. + // consumer needs no access to any of them. Billing follows the same boundary: + // the child opens its own logging session against the source workspace, whose + // payer is charged for everything it spends. Regular workflow blocks keep + // running in the parent's context. let childUserId = ctx.userId let childWorkspaceId = ctx.workspaceId let childEnvVarValues = ctx.environmentVariables let childBillingAttribution = ctx.metadata.billingAttribution + let childEnvVariablesForLogging = ctx.environmentVariables if (isCustomBlock) { if (!loadUserId) { throw new Error('Custom block source workflow has no owner') @@ -390,10 +422,15 @@ export class WorkflowBlockHandler implements BlockHandler { if (!childWorkflow.workspaceId) { throw new Error('Custom block source workflow has no workspace') } + const sourceWorkspaceId = childWorkflow.workspaceId childUserId = loadUserId - childWorkspaceId = childWorkflow.workspaceId - const ownerEnv = await getPersonalAndWorkspaceEnv(loadUserId, childWorkflow.workspaceId) + childWorkspaceId = sourceWorkspaceId + const ownerEnv = await getPersonalAndWorkspaceEnv(loadUserId, sourceWorkspaceId) childEnvVarValues = { ...ownerEnv.personalDecrypted, ...ownerEnv.workspaceDecrypted } + childEnvVariablesForLogging = { + ...ownerEnv.personalEncrypted, + ...ownerEnv.workspaceEncrypted, + } // Custom-block children authenticate internal tool calls as the source // owner in the source workspace, so the consumer's snapshot would fail // the internal routes' actor/workspace scope match. Resolve the @@ -403,6 +440,75 @@ export class WorkflowBlockHandler implements BlockHandler { actorUserId: loadUserId, workspaceId: childWorkflow.workspaceId, }) + // Admit against the source payer before any spend. No reservation — see + // `admitCustomBlockChildExecution`. + await admitCustomBlockChildExecution(childBillingAttribution) + + // A custom block's child is its own execution: its own id (the log row's + // `execution_id` is unique), its own logging session against the SOURCE + // workspace, and its own ledger rows. Everything below that differs from + // a regular workflow block follows from that. + childExecutionId = generateId() + childSession = new LoggingSession( + workflowId, + childExecutionId, + CUSTOM_BLOCK_TRIGGER, + ctx.metadata.requestId, + childExecutionId, + // The consumer already pays one execution fee for the invoking run; the + // child is part of that same logical run and must not add a second. + { baseExecutionCharge: 0 } + ) + const correlation = buildCustomBlockCorrelation({ + invokerExecutionId: ctx.executionId, + invokerRequestId: ctx.metadata.requestId, + invokerWorkflowId: ctx.workflowId, + invokerWorkspaceId: ctx.workspaceId, + blockType: blockTypeId as string, + }) + childSessionStarted = await childSession.safeStart({ + userId: childUserId, + actorUserId: childUserId, + billingAttribution: childBillingAttribution, + workspaceId: sourceWorkspaceId, + variables: childEnvVariablesForLogging, + workflowState: childWorkflow.workflowState, + ...(correlation ? { triggerData: { correlation } } : {}), + }) + if (!childSessionStarted) { + logger.error('Custom block child logging failed to start; child spend will be unbilled', { + workflowId, + childExecutionId, + }) + } + // The child no longer shares the parent's execution id, so it no longer + // hears the parent's cancellation event — bridge it explicitly. + childCancellation = await createChildCancellationSignal({ + parentSignal: ctx.abortSignal, + parentExecutionId: ctx.executionId, + }) + // Registered BEFORE the child starts and settled in `finally`, so the + // tracked promise spans execution AND the terminal log write. A cancelled + // parent drains at a moment when the child is still inside `execute`, so + // registering only the finalization step would find nothing to await. + trackChildRun( + ctx.executionId, + new Promise((resolve) => { + settleChildRun = resolve + }) + ) + + // Large values are scoped by execution id, so the parent must be able to + // read a large exposed output the child produced. ONE array is shared down + // the whole chain rather than copied per hop: a nested custom block pushes + // its own child id into this same list, so a grandchild's large output is + // still materializable at the top level. Copying would strand those ids at + // the depth that created them. + ctx.largeValueExecutionIds ??= [] + sharedLargeValueIds = ctx.largeValueExecutionIds + for (const id of [ctx.executionId, childExecutionId]) { + if (id && !sharedLargeValueIds.includes(id)) sharedLargeValueIds.push(id) + } } // Trusted run metadata for the child's Start block. Every field describes @@ -439,6 +545,7 @@ export class WorkflowBlockHandler implements BlockHandler { } } + const activeSession = childSession const subExecutor = new Executor({ workflow: childWorkflow.serializedState, workflowInput: childWorkflowInput, @@ -453,7 +560,12 @@ export class WorkflowBlockHandler implements BlockHandler { enforceCredentialAccess: ctx.enforceCredentialAccess, workspaceId: childWorkspaceId, userId: childUserId, - executionId: ctx.executionId, + executionId: childExecutionId ?? ctx.executionId, + // Large values are cached per execution id, so a child running under its + // own id still needs the invoking run's id to read values in its inputs. + ...(childExecutionId && sharedLargeValueIds + ? { largeValueExecutionIds: sharedLargeValueIds } + : {}), // Same-workspace children share the parent's frozen payer decision so // internal tool calls (knowledge, guardrails, MCP, Mothership) can // attach the required billing attribution header. @@ -461,11 +573,42 @@ export class WorkflowBlockHandler implements BlockHandler { // Fall back to the inherited metadata so a toggle-off intermediate // child still carries the trusted identity chain to deeper children. startRunMetadata: childStartRunMetadata ?? inherited, - abortSignal: ctx.abortSignal, + abortSignal: childCancellation?.signal ?? ctx.abortSignal, // Propagate in-flight block-output redaction into child workflows so // nested blocks mask outputs too (recurses: each child forwards it). piiBlockOutputRedaction: ctx.piiBlockOutputRedaction, callChain: childCallChain, + // A custom block's block markers belong to ITS OWN session — the + // parent's callbacks are bound to the consumer's logging session and + // would both leak the source's block names and clobber its progress. + ...(activeSession && childSessionStarted + ? { + onBlockStart: async (blockId: string, blockName: string, blockType: string) => { + try { + await activeSession.onBlockStart( + blockId, + blockName, + blockType, + new Date().toISOString() + ) + } catch { + // A progress marker must never fail the block it describes. + } + }, + onBlockComplete: async ( + blockId: string, + blockName: string, + blockType: string, + output: unknown + ) => { + try { + await activeSession.onBlockComplete(blockId, blockName, blockType, output) + } catch { + // A progress marker must never fail the block it describes. + } + }, + } + : {}), ...(shouldPropagateCallbacks && { onBlockStart: ctx.onBlockStart, onBlockComplete: ctx.onBlockComplete, @@ -487,12 +630,32 @@ export class WorkflowBlockHandler implements BlockHandler { const executionResult = this.toExecutionResult(result) const duration = performance.now() - startTime + if (childSession && childSessionStarted) { + await this.finalizeChildSession(childSession, executionResult, duration, childWorkflowInput) + childSessionFinalized = true + } + logger.info(`Child workflow ${childWorkflowName} completed in ${Math.round(duration)}ms`, { success: executionResult.success, hasLogs: (executionResult.logs?.length ?? 0) > 0, }) - const childTraceSpans = this.captureChildWorkflowLogs(executionResult, childWorkflowName, ctx) + // A cancelled run comes back as `success: false`, so without this it would + // fall through to `mapChildOutputToParent` and reach the consumer as a + // generic `execution_failed`. Classify it instead — the message names + // nothing about the source, so it crosses the boundary verbatim. + if (isCustomBlock && executionResult.status === 'cancelled') { + throw new BoundarySafeError({ + errorType: 'cancelled', + message: 'Custom block execution was cancelled', + }) + } + + // A custom block's spans never reach the parent — they belong to the child's + // own log row in the source workspace — so don't build them here at all. + const childTraceSpans = isCustomBlock + ? [] + : this.captureChildWorkflowLogs(executionResult, childWorkflowName, ctx) const mappedResult = this.mapChildOutputToParent( executionResult, @@ -505,61 +668,41 @@ export class WorkflowBlockHandler implements BlockHandler { ) // Custom blocks expose only curated outputs — never the child workflow id, - // name, or trace spans. `mapChildOutputToParent` above still runs so failures - // surface identically; we just reshape the successful output. + // name, trace spans, or cost. `mapChildOutputToParent` above still runs so + // failures surface identically; we just reshape the successful output. The + // child's spend is billed by its own session, not rolled onto this block. if (isCustomBlock) { - // The child's spans are stripped for privacy, but they're the only carrier - // of the run's cost into billing — so roll their aggregate cost onto the - // block itself. Custom blocks are org-scoped, so this bills the same org the - // source workflow would bill if run directly, exactly as if it ran the key. - const childCost = aggregateChildCost(childTraceSpans) - return this.projectCustomBlockOutput(executionResult, exposedOutputs, childCost) + return this.projectCustomBlockOutput(executionResult, exposedOutputs) } return mappedResult } catch (error: unknown) { logger.error(`Error executing child workflow ${workflowId}:`, error) - // Custom blocks are an invocation boundary: on failure the consumer must not - // see the source workflow's name, nested error text (which names internal - // blocks), trace spans, or execution result — the success path hides all of - // these too. The real error is logged above for the publisher/ops; the - // consumer gets only a generic failure attributed to the block they placed. - // But a child that failed AFTER consuming hosted keys still owes that spend, - // so capture the child's spans server-side, distill to the aggregate cost, and - // carry only that (no internals) so `block-executor` still bills it. + // The child's own log row records the real failure in the source workspace, + // so the publisher sees what the consumer deliberately cannot. + if (childSession && childSessionStarted && !childSessionFinalized) { + await this.failChildSession(childSession, error) + } + + // A custom block is checked FIRST and unconditionally: errors this invocation + // already attributed still name the source workflow (`mapChildOutputToParent` + // formats `"" failed: `), so short-circuiting on + // them here would hand the consumer exactly what the boundary exists to hide. + // `buildBoundaryFailure` preserves an already-attached `consumerFacing`, so the + // depth guard keeps its own classification. if (isCustomBlock) { - // Missing-required-inputs is the consumer's own mistake and its message - // names only the block's input labels — surface it instead of the generic - // failure so they can actually fix it. The child never ran: no spend. - if (error instanceof CustomBlockMissingInputsError) { - throw new ChildWorkflowError({ - message: error.message, - childWorkflowName: block.metadata?.name || 'Custom block', - childWorkflowInstanceId: instanceId, - }) - } - let failedChildSpans: WorkflowTraceSpan[] = [] - if (hasExecutionResult(error) && error.executionResult.logs) { - failedChildSpans = this.captureChildWorkflowLogs( - error.executionResult, - childWorkflowName, - ctx - ) - } else if (ChildWorkflowError.isChildWorkflowError(error)) { - failedChildSpans = error.childTraceSpans - } - const blockName = block.metadata?.name || 'Custom block' - throw new ChildWorkflowError({ - message: 'Custom block execution failed', - childWorkflowName: blockName, - childTraceSpans: buildCostCarrierSpans( - aggregateChildCost(failedChildSpans), - blockName, - block.metadata?.id ?? 'custom_block' - ), - childWorkflowInstanceId: instanceId, - }) + throw this.buildBoundaryFailure(error, block, instanceId, childExecutionId) + } + + // An error this same invocation already attributed (e.g. the depth guard, or + // `mapChildOutputToParent`) is rethrown untouched — re-wrapping it would + // duplicate this workflow in the chain. + if ( + ChildWorkflowError.isChildWorkflowError(error) && + error.childWorkflowInstanceId === instanceId + ) { + throw error } let childTraceSpans: WorkflowTraceSpan[] = [] @@ -580,85 +723,152 @@ export class WorkflowBlockHandler implements BlockHandler { childTraceSpans = error.childTraceSpans } - // Build a cleaner error message for nested workflow errors - const errorMessage = this.buildNestedWorkflowErrorMessage(childWorkflowName, error) + const { chain, rootErrorMessage } = this.buildChildFailure(childWorkflowName, error) throw new ChildWorkflowError({ - message: errorMessage, + message: formatWorkflowChainMessage(chain, rootErrorMessage), childWorkflowName, + workflowChain: chain, + rootErrorMessage, childTraceSpans, executionResult, childWorkflowSnapshotId, childWorkflowInstanceId: instanceId, cause: error instanceof Error ? error : undefined, }) + } finally { + // A custom block inside a loop would otherwise leak one abort listener and + // one cancellation subscription per iteration. + childCancellation?.dispose() + // Releases the invoking run's drain: reached on every exit path, including + // when this handler's promise was abandoned by a cancelled parent engine. + settleChildRun?.() } } /** - * Builds a cleaner error message for nested workflow errors. - * Parses nested error messages to extract workflow chain and root error. + * Completes a custom-block child's own logging session, so the publisher gets a + * full run record — trace waterfall, duration, and ledger rows — in the source + * workspace. A paused child goes through the normal completion path rather than + * `safeCompleteWithPause`: a nested child has no resume path, and the pause path + * deliberately skips the reservation release, which would strand a pending row. */ - private buildNestedWorkflowErrorMessage(childWorkflowName: string, error: unknown): string { - const originalError = getErrorMessage(error, 'Unknown error') + private async finalizeChildSession( + session: LoggingSession, + executionResult: ExecutionResult, + durationMs: number, + workflowInput: Record + ): Promise { + const { traceSpans, totalDuration } = buildTraceSpans(executionResult) + const endedAt = new Date().toISOString() + const totalDurationMs = totalDuration ?? Math.round(durationMs) + + // Cancellation lives on `ExecutionResult.status` — `ExecutionMetadata.status` + // has no 'cancelled' member, so reading it there never matches. + if (executionResult.status === 'cancelled') { + await session.safeCompleteWithCancellation({ endedAt, totalDurationMs, traceSpans }) + return + } - // Extract any nested workflow names from the error message - const { chain, rootError } = this.parseNestedWorkflowError(originalError) + await session.safeComplete({ + endedAt, + totalDurationMs, + finalOutput: executionResult.output ?? {}, + traceSpans, + workflowInput, + }) + } - // Add current workflow to the beginning of the chain - chain.unshift(childWorkflowName) + /** Records a custom-block child's failure on its own log row. */ + private async failChildSession(session: LoggingSession, error: unknown): Promise { + const normalized = toError(error) + const executionResult = hasExecutionResult(error) ? error.executionResult : undefined + const { traceSpans, totalDuration } = executionResult + ? buildTraceSpans(executionResult) + : { traceSpans: [], totalDuration: 0 } + + await session.safeCompleteWithError({ + endedAt: new Date().toISOString(), + totalDurationMs: totalDuration ?? 0, + error: { message: normalized.message, stackTrace: normalized.stack }, + traceSpans, + }) + } - // If we have a chain (nested workflows), format nicely - if (chain.length > 1) { - return `Workflow chain: ${chain.join(' → ')} | ${rootError}` + /** + * The consumer-facing failure for a custom block. The invocation boundary means + * the consumer must never see the source workflow's name, its nested error text + * (which names internal blocks), its trace spans, or its execution result — the + * success path hides all of these too. The real error is logged for the + * publisher, and the child's own log row carries the full detail. + * + * A `BoundarySafeError` names only the caller's own artifacts, so its message + * crosses verbatim. Everything else collapses to a generic failure: the default + * is fail-closed, so a `throw` added later is redacted automatically. Deliberately + * sets no `cause` — that is what severs the error chain at the trust boundary. + */ + private buildBoundaryFailure( + error: unknown, + block: SerializedBlock, + instanceId: string, + childExecutionId: string | undefined + ): ChildWorkflowError { + const blockName = block.metadata?.name || 'Custom block' + + // An error this invocation already classified for the consumer (the depth + // guard) keeps its own type and message rather than collapsing to generic. + const alreadyClassified = + ChildWorkflowError.isChildWorkflowError(error) && error.consumerFacing + ? error.consumerFacing + : undefined + if (alreadyClassified) { + return new ChildWorkflowError({ + message: alreadyClassified.message, + childWorkflowName: blockName, + childWorkflowInstanceId: instanceId, + consumerFacing: alreadyClassified, + }) } - // Single workflow failure - return `"${childWorkflowName}" failed: ${rootError}` + const safe = isBoundarySafeError(error) ? error : undefined + const errorType: CustomBlockErrorType = safe?.errorType ?? 'execution_failed' + // The ref is the child run's own execution id — opaque to the consumer, and + // the handle a publisher needs to find the exact failing run in their logs. + const ref = safe ? undefined : childExecutionId + const message = safe + ? safe.message + : ref + ? `Custom block execution failed (ref: ${ref})` + : 'Custom block execution failed' + + return new ChildWorkflowError({ + message, + childWorkflowName: blockName, + childWorkflowInstanceId: instanceId, + consumerFacing: { errorType, ...(ref ? { ref } : {}), message }, + }) } /** - * Parses a potentially nested workflow error message to extract: - * - The chain of workflow names - * - The actual root error message (preserving the block name prefix for the failing block) - * - * Handles formats like: - * - "workflow-name" failed: error - * - Block Name: "workflow-name" failed: error - * - Workflow chain: A → B | error + * The workflow chain and root error for a nested failure, recovered from the + * structured fields on a wrapped {@link ChildWorkflowError} rather than by + * parsing its formatted message. */ - private parseNestedWorkflowError(message: string): { chain: string[]; rootError: string } { - const chain: string[] = [] - const remaining = message - - // First, check if it's already in chain format - const chainMatch = remaining.match(/^Workflow chain: (.+?) \| (.+)$/) - if (chainMatch) { - const chainPart = chainMatch[1] - const errorPart = chainMatch[2] - chain.push(...chainPart.split(' → ').map((s) => s.trim())) - return { chain, rootError: errorPart } + private buildChildFailure( + childWorkflowName: string, + error: unknown + ): { chain: string[]; rootErrorMessage: string } { + const nested = findCause(error, ChildWorkflowError.isChildWorkflowError) + if (nested) { + return { + chain: [childWorkflowName, ...nested.workflowChain], + rootErrorMessage: nested.rootErrorMessage, + } } - - // Extract workflow names from patterns like: - // - "workflow-name" failed: - // - Block Name: "workflow-name" failed: - const workflowPattern = /(?:\[[^\]]+\]\s*)?(?:[^:]+:\s*)?"([^"]+)"\s*failed:\s*/g - let match: RegExpExecArray | null - let lastIndex = 0 - - match = workflowPattern.exec(remaining) - while (match !== null) { - chain.push(match[1]) - lastIndex = match.index + match[0].length - match = workflowPattern.exec(remaining) + return { + chain: [childWorkflowName], + rootErrorMessage: getErrorMessage(error, 'Unknown error'), } - - // The root error is everything after the last match - // Keep the block name prefix (e.g., Function 1:) so we know which block failed - const rootError = lastIndex > 0 ? remaining.slice(lastIndex) : remaining - - return { chain, rootError: rootError.trim() || 'Unknown error' } } /** @@ -929,22 +1139,20 @@ export class WorkflowBlockHandler implements BlockHandler { } /** - * Shape a custom block's successful output. With curated `exposedOutputs`, each - * maps a child block output (blockId + dot-path, read from the child's per-block - * logs) to a named top-level field. With none, exposes the child's whole - * `result`. Never leaks child workflow id/name/trace spans. + * Shape a custom block's successful output: each curated `exposedOutput` maps a + * child block output (blockId + dot-path, read from the child's per-block logs) + * to a named top-level field. + * + * Curation is required at publish, so there is no whole-`result` fallback — + * that path would hand the consumer the terminal block's raw state, including + * an agent's `toolCalls`/`thinkingContent`/`cost` or a nested workflow block's + * identifiers. Never leaks child workflow id/name/trace spans, nor cost, which + * is billed to the source workspace by the child's own logging session. */ private projectCustomBlockOutput( executionResult: ExecutionResult, - exposedOutputs: CustomBlockOutput[], - childCost: number + exposedOutputs: CustomBlockOutput[] ): BlockOutput { - // Aggregate child cost only (never the child's spans/model breakdown) so the - // run is billed while the source workflow's internals stay hidden. - const cost = childCost > 0 ? { cost: { total: childCost } } : {} - if (exposedOutputs.length === 0) { - return { success: true, result: executionResult.output ?? {}, ...cost } - } const logs = executionResult.logs ?? [] const output: Record = {} for (const { blockId, path, name } of exposedOutputs) { @@ -953,8 +1161,8 @@ export class WorkflowBlockHandler implements BlockHandler { [...logs].reverse().find((l) => l.blockId === blockId) output[name] = log ? getValueAtPath(log.output, path) : undefined } - // System fields spread last — pre-validation rows may still name an output cost/success. - return { ...output, success: true, ...cost } as BlockOutput + // System fields spread last — pre-validation rows may still name an output success. + return { ...output, success: true } as BlockOutput } private mapChildOutputToParent( @@ -971,9 +1179,13 @@ export class WorkflowBlockHandler implements BlockHandler { if (!success) { logger.warn(`Child workflow ${childWorkflowName} failed`) + const rootErrorMessage = childResult.error || 'Child workflow execution failed' + const chain = [childWorkflowName] throw new ChildWorkflowError({ - message: `"${childWorkflowName}" failed: ${childResult.error || 'Child workflow execution failed'}`, + message: formatWorkflowChainMessage(chain, rootErrorMessage), childWorkflowName, + workflowChain: chain, + rootErrorMessage, childTraceSpans: childTraceSpans || [], childWorkflowSnapshotId, childWorkflowInstanceId: instanceId, diff --git a/apps/sim/executor/utils/errors.test.ts b/apps/sim/executor/utils/errors.test.ts new file mode 100644 index 00000000000..ea41ecea020 --- /dev/null +++ b/apps/sim/executor/utils/errors.test.ts @@ -0,0 +1,139 @@ +/** + * @vitest-environment node + */ +import { describe, expect, it } from 'vitest' +import { HttpError } from '@/lib/core/utils/http-error' +import { buildBlockExecutionError, getExecutionErrorStatus } from '@/executor/utils/errors' +import type { SerializedBlock } from '@/serializer/types' + +class TestRateLimitedError extends HttpError { + readonly statusCode = 429 +} + +class TestUnavailableError extends HttpError { + readonly statusCode = 503 +} + +class TestBadGatewayError extends HttpError { + readonly statusCode = 502 +} + +const block = { + id: 'block-1', + position: { x: 0, y: 0 }, + config: { tool: 'test_tool', params: {} }, + inputs: {}, + outputs: {}, + metadata: { id: 'generic', name: 'My Block' }, + enabled: true, +} as SerializedBlock + +describe('getExecutionErrorStatus', () => { + it('forwards a 4xx from a typed HTTP error', () => { + expect(getExecutionErrorStatus(new TestRateLimitedError('slow down'))).toBe(429) + }) + + it("forwards 503 because it describes Sim's own capacity", () => { + expect(getExecutionErrorStatus(new TestUnavailableError('no keys'))).toBe(503) + }) + + it('does not forward an upstream 502 as the workflow API status', () => { + expect(getExecutionErrorStatus(new TestBadGatewayError('upstream down'))).toBe(500) + }) + + it('falls back to 500 for an untyped error', () => { + expect(getExecutionErrorStatus(new Error('boom'))).toBe(500) + }) + + it("never adopts an upstream target's duck-typed status as our own", () => { + // `api-handler` copies the remote response's status onto the thrown error. + // Adopting it would make a remote 404 the workflow API's 404. + const error = Object.assign(new Error('HTTP 404'), { status: 404 }) + expect(getExecutionErrorStatus(error)).toBe(500) + }) + + it('still reads a Sim-owned statusCode re-attached from a ToolResponse', () => { + const error = Object.assign(new Error('rate limited'), { statusCode: 429 }) + expect(getExecutionErrorStatus(error)).toBe(429) + }) + + it('finds a status carried further down the cause chain', () => { + const wrapped = buildBlockExecutionError({ + block, + error: new TestRateLimitedError('slow down'), + }) + expect(getExecutionErrorStatus(wrapped)).toBe(429) + }) + + it('survives a cyclic cause chain', () => { + const a = new Error('a') + const b = new Error('b') + Object.assign(a, { cause: b }) + Object.assign(b, { cause: a }) + expect(getExecutionErrorStatus(a)).toBe(500) + }) +}) + +describe('buildBlockExecutionError', () => { + it('prefixes the block name and preserves the original as cause', () => { + const original = new Error('inner failure') + const wrapped = buildBlockExecutionError({ block, error: original }) + + expect(wrapped.message).toBe('My Block: inner failure') + expect(wrapped.cause).toBe(original) + }) + + it('leaves cause unset for a non-Error throw', () => { + const wrapped = buildBlockExecutionError({ block, error: 'plain string' }) + + expect(wrapped.message).toBe('My Block: plain string') + expect(wrapped.cause).toBeUndefined() + }) +}) + +describe('hosted-key status survives the ToolResponse flattening', () => { + /** + * `executeTool` catches a thrown error and returns a `ToolResponse`, so the + * status has to ride `ToolResponse.statusCode` and be re-attached by + * `generic-handler`. This reproduces that hand-off end to end. + */ + function errorFromFailedToolResponse(response: { + error: string + output: Record + statusCode?: number + }): Error { + const error = new Error(response.error) + Object.assign(error, { + output: response.output, + ...(typeof response.statusCode === 'number' ? { statusCode: response.statusCode } : {}), + }) + return buildBlockExecutionError({ block, error }) + } + + it('forwards a hosted-key 429 to the API caller', () => { + const wrapped = errorFromFailedToolResponse({ + error: 'Rate limit exceeded', + output: {}, + statusCode: 429, + }) + expect(getExecutionErrorStatus(wrapped)).toBe(429) + }) + + it('forwards a hosted-key 503 to the API caller', () => { + const wrapped = errorFromFailedToolResponse({ + error: 'No hosted keys configured', + output: {}, + statusCode: 503, + }) + expect(getExecutionErrorStatus(wrapped)).toBe(503) + }) + + it("never adopts an upstream provider's status as our own", () => { + // A provider 404 rides `output`, never `statusCode`, so it must not surface. + const wrapped = errorFromFailedToolResponse({ + error: 'HTTP 404: Not Found', + output: { status: 404, statusText: 'Not Found' }, + }) + expect(getExecutionErrorStatus(wrapped)).toBe(500) + }) +}) diff --git a/apps/sim/executor/utils/errors.ts b/apps/sim/executor/utils/errors.ts index edcff8342ea..b5a509fbf6c 100644 --- a/apps/sim/executor/utils/errors.ts +++ b/apps/sim/executor/utils/errors.ts @@ -1,3 +1,4 @@ +import { HttpError } from '@/lib/core/utils/http-error' import type { ExecutionContext, ExecutionResult } from '@/executor/types' import type { SerializedBlock } from '@/serializer/types' @@ -41,15 +42,21 @@ export interface BlockExecutionErrorDetails { additionalInfo?: Record } +/** + * Wraps a block failure with the block's identity. The original error is kept + * as `cause` so the chain stays walkable — `readStatusCode` and `describeError` + * both depend on it, and rebuilding the error without it severs the chain at + * every block boundary. + */ export function buildBlockExecutionError(details: BlockExecutionErrorDetails): Error { const errorMessage = details.error instanceof Error ? details.error.message : String(details.error) const blockName = details.block.metadata?.name || details.block.id const blockType = details.block.metadata?.id || 'unknown' - const error = new Error(`${blockName}: ${errorMessage}`) - - const innerStatusCode = readStatusCode(details.error) + const error = new Error(`${blockName}: ${errorMessage}`, { + cause: details.error instanceof Error ? details.error : undefined, + }) Object.assign(error, { blockId: details.block.id, @@ -58,56 +65,63 @@ export function buildBlockExecutionError(details: BlockExecutionErrorDetails): E workflowId: details.context?.workflowId, timestamp: new Date().toISOString(), ...details.additionalInfo, - ...(innerStatusCode !== undefined ? { statusCode: innerStatusCode } : {}), }) return error } -export function buildHTTPError(config: { - status: number - url?: string - method?: string - message?: string -}): Error { - let errorMessage = config.message || `HTTP ${config.method || 'request'} failed` +/** Maximum `.cause` links to follow before giving up. Mirrors `describeError`. */ +const MAX_CAUSE_DEPTH = 8 - if (config.url) { - errorMessage += ` - ${config.url}` - } +/** + * HTTP status carried by a thrown value, walking the `.cause` chain so a status + * set deep in a tool survives the block-level wrapping that rebuilds the error. + * + * Reads only SIM-OWNED carriers: `HttpError.statusCode` (canonical) and the + * `statusCode` field `generic-handler` re-attaches from a failed `ToolResponse`. + * Deliberately does NOT read the duck-typed `status` field: that carries an + * UPSTREAM target's status (`api-handler` copies it off the remote response, and + * transformed HTTP tool errors carry it too). Adopting it would turn a remote + * 404 into the workflow API's 404, colliding with the statuses that route owns + * (404 = workflow not found, 401 = bad API key, 429 = Sim rate limit). + */ +export function readStatusCode(value: unknown): number | undefined { + const seen = new Set() + let current = value - if (config.status) { - errorMessage += ` (Status: ${config.status})` - } + for (let depth = 0; depth < MAX_CAUSE_DEPTH; depth++) { + if (!(current instanceof Error) || seen.has(current)) return undefined + seen.add(current) - const error = new Error(errorMessage) + if (current instanceof HttpError) return current.statusCode - Object.assign(error, { - status: config.status, - url: config.url, - method: config.method, - timestamp: new Date().toISOString(), - }) + const candidate = current as unknown as { statusCode?: unknown } + if (typeof candidate.statusCode === 'number') return candidate.statusCode - return error -} + current = current.cause + } -function readStatusCode(value: unknown): number | undefined { - if (!(value instanceof Error)) return undefined - const status = (value as unknown as { statusCode?: unknown }).statusCode - return typeof status === 'number' ? status : undefined + return undefined } +/** + * 5xx statuses that describe Sim's own capacity rather than an upstream + * provider's, and are therefore safe to forward to the API caller. An + * upstream 502/504 must not become the workflow API's status. + */ +const FORWARDABLE_SERVER_STATUSES = new Set([503]) + /** * Maps an execution error to an HTTP status code. Errors thrown from the * executor that represent workflow-author mistakes (invalid field references, - * etc.) carry a 4xx `statusCode`; everything else is a 500. + * etc.) carry a 4xx status; hosted-key exhaustion carries a 503. Everything + * else is a 500. */ export function getExecutionErrorStatus(error: unknown): number { const status = readStatusCode(error) - if (status !== undefined && status >= 400 && status < 500) { - return status - } + if (status === undefined) return 500 + if (status >= 400 && status < 500) return status + if (FORWARDABLE_SERVER_STATUSES.has(status)) return status return 500 } diff --git a/apps/sim/lib/api/contracts/custom-blocks.ts b/apps/sim/lib/api/contracts/custom-blocks.ts index 77da3a0213b..70f40a1132f 100644 --- a/apps/sim/lib/api/contracts/custom-blocks.ts +++ b/apps/sim/lib/api/contracts/custom-blocks.ts @@ -102,8 +102,17 @@ export const publishCustomBlockBodySchema = z.object({ iconUrl: iconUrlSchema.optional(), /** Per-input placeholder hints keyed by Start field id; the field set itself is always derived from the deployment. */ inputs: z.array(inputPlaceholderSchema).max(50).optional(), - /** Curated outputs; omit/empty to expose the child's whole result. */ - exposedOutputs: z.array(exposedOutputWriteSchema).max(50).optional(), + /** + * Curated outputs. REQUIRED: every field a consumer receives must be one the + * publisher explicitly chose. There is deliberately no "expose everything" + * fallback — the terminal block's raw state carries execution metadata + * (an agent's `toolCalls`, `providerTiming.thinkingContent`, `cost`; a nested + * workflow block's ids) that would cross the invocation boundary unchosen. + */ + exposedOutputs: z + .array(exposedOutputWriteSchema) + .min(1, 'Select at least one output to expose to consumers') + .max(50), }) export type PublishCustomBlockBody = z.input @@ -120,7 +129,12 @@ export const updateCustomBlockBodySchema = z /** A URL (https or internal serve path) sets/replaces the icon; `null` clears it (default icon). */ iconUrl: iconUrlSchema.nullable().optional(), inputs: z.array(inputPlaceholderSchema).max(50).optional(), - exposedOutputs: z.array(exposedOutputWriteSchema).max(50).optional(), + /** Omit to leave the curated outputs unchanged; never settable to empty. */ + exposedOutputs: z + .array(exposedOutputWriteSchema) + .min(1, 'Select at least one output to expose to consumers') + .max(50) + .optional(), }) .refine((v) => Object.keys(v).length > 0, { message: 'At least one field is required' }) diff --git a/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.test.ts b/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.test.ts index 480d33a801e..54817fffbc3 100644 --- a/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.test.ts +++ b/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.test.ts @@ -341,7 +341,11 @@ describe('executeDeployCustomBlock', () => { publishCustomBlockMock.mockResolvedValue(publishedBlock) const result = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'files/icon.png' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'files/icon.png', + }, context ) @@ -364,7 +368,11 @@ describe('executeDeployCustomBlock', () => { publishCustomBlockMock.mockResolvedValue(publishedBlock) const result = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'https://example.com/icon.png' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'https://example.com/icon.png', + }, context ) @@ -379,7 +387,11 @@ describe('executeDeployCustomBlock', () => { listWorkspaceFilesMock.mockResolvedValue([]) const result = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'files/missing.png' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'files/missing.png', + }, context ) @@ -390,14 +402,22 @@ describe('executeDeployCustomBlock', () => { it('rejects non-https icon URL schemes on pass-through', async () => { const dataUri = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'data:image/svg+xml;base64,PHN2Zy8+' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'data:image/svg+xml;base64,PHN2Zy8+', + }, context ) expect(dataUri.success).toBe(false) expect(dataUri.error).toContain('https') const plainHttp = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'http://example.com/icon.png' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'http://example.com/icon.png', + }, context ) expect(plainHttp.success).toBe(false) @@ -405,7 +425,11 @@ describe('executeDeployCustomBlock', () => { publishCustomBlockMock.mockResolvedValue(publishedBlock) const servePath = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: '/api/files/serve/workspace-logos%2Ficon.png' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: '/api/files/serve/workspace-logos%2Ficon.png', + }, context ) expect(servePath.success).toBe(true) @@ -417,7 +441,11 @@ describe('executeDeployCustomBlock', () => { ]) const result = await executeDeployCustomBlock( - { name: 'Enrich Lead', iconUrl: 'files/notes.pdf' }, + { + name: 'Enrich Lead', + exposedOutputs: [{ blockId: 'b1', path: 'content', name: 'answer' }], + iconUrl: 'files/notes.pdf', + }, context ) @@ -434,4 +462,12 @@ describe('executeDeployCustomBlock', () => { expect(result.success).toBe(false) expect(result.error).toContain('organization') }) + + it('refuses to publish without curated outputs', async () => { + const result = await executeDeployCustomBlock({ name: 'Enrich Lead' }, context) + + expect(result.success).toBe(false) + expect(result.error).toContain('exposedOutputs is required') + expect(publishCustomBlockMock).not.toHaveBeenCalled() + }) }) diff --git a/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.ts b/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.ts index 36268f6216f..8b2182b6abe 100644 --- a/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.ts +++ b/apps/sim/lib/copilot/tools/handlers/deployment/custom-block.ts @@ -258,6 +258,17 @@ export async function executeDeployCustomBlock( 'Workflow must be deployed before publishing as a custom block. Use deploy_api first.', } } + // Curation is required on publish: every consumer-visible field must be one + // the publisher chose, since there is no whole-`result` fallback. + const exposedOutputs = params.exposedOutputs + if (!exposedOutputs || exposedOutputs.length === 0) { + return { + success: false, + error: + 'exposedOutputs is required: select at least one workflow output to expose to consumers', + } + } + const block = await publishCustomBlock({ organizationId, workspaceId, @@ -267,7 +278,7 @@ export async function executeDeployCustomBlock( description: description ?? '', iconUrl, inputs: params.inputs, - exposedOutputs: params.exposedOutputs, + exposedOutputs, }) recordAudit({ workspaceId, diff --git a/apps/sim/lib/core/async-jobs/types.ts b/apps/sim/lib/core/async-jobs/types.ts index 11b74f13287..12b852165f1 100644 --- a/apps/sim/lib/core/async-jobs/types.ts +++ b/apps/sim/lib/core/async-jobs/types.ts @@ -32,7 +32,7 @@ export type JobType = | 'cleanup-tasks' | 'run-data-drain' -export type AsyncExecutionCorrelationSource = 'workflow' | 'schedule' | 'webhook' +export type AsyncExecutionCorrelationSource = 'workflow' | 'schedule' | 'webhook' | 'custom_block' export interface AsyncExecutionCorrelation { executionId: string @@ -45,6 +45,11 @@ export interface AsyncExecutionCorrelation { path?: string provider?: string scheduledFor?: string + /** + * Workspace of the invoking run. Set for custom-block children, whose invoker + * lives in a different workspace than the log row this correlation lands on. + */ + invokerWorkspaceId?: string } export interface Job { diff --git a/apps/sim/lib/logs/execution/logger.ts b/apps/sim/lib/logs/execution/logger.ts index c4100005ec7..78729e44783 100644 --- a/apps/sim/lib/logs/execution/logger.ts +++ b/apps/sim/lib/logs/execution/logger.ts @@ -49,6 +49,7 @@ import { pickLatestStartedMarker, } from '@/lib/logs/execution/progress-markers' import { snapshotService } from '@/lib/logs/execution/snapshot/service' +import { traceSpansIndicateFailure } from '@/lib/logs/execution/trace-spans/trace-spans' import { externalizeExecutionData, stripSpanCosts, @@ -775,19 +776,10 @@ export class ExecutionLogger implements IExecutionLoggerService { providedActorUserId ?? this.extractActorUserId(existingLog?.executionData) - // Determine if workflow failed by checking trace spans for unhandled errors - // Errors handled by error handler paths (errorHandled: true) don't count as workflow failures - // Use the override if provided (for cost-only fallback scenarios) - const hasErrors = traceSpans?.some((span: any) => { - const checkSpanForErrors = (s: any): boolean => { - if (s.status === 'error' && !s.errorHandled) return true - if (s.children && Array.isArray(s.children)) { - return s.children.some(checkSpanForErrors) - } - return false - } - return checkSpanForErrors(span) - }) + // Determine if workflow failed by checking trace spans for unhandled errors. + // Errors handled by error handler paths (errorHandled: true) don't count as + // workflow failures. Use the override if provided (cost-only fallback). + const hasErrors = traceSpansIndicateFailure(traceSpans) const level = levelOverride ?? (hasErrors ? 'error' : 'info') const status = statusOverride ?? (hasErrors ? 'failed' : 'completed') diff --git a/apps/sim/lib/logs/execution/logging-factory.test.ts b/apps/sim/lib/logs/execution/logging-factory.test.ts index 8800312618d..7dabdc6955b 100644 --- a/apps/sim/lib/logs/execution/logging-factory.test.ts +++ b/apps/sim/lib/logs/execution/logging-factory.test.ts @@ -772,3 +772,39 @@ describe('calculateCostSummary', () => { expect(ledgerSum).toBeCloseTo(result.totalCost, 10) }) }) + +describe('calculateCostSummary base charge override', () => { + test('an invoked child adds no execution fee when given zero', () => { + const result = calculateCostSummary([], { baseExecutionCharge: 0 }) + + expect(result.baseExecutionCharge).toBe(0) + expect(result.totalCost).toBe(0) + }) + + test('total equals the span sum exactly with no base charge', () => { + const spans = [ + { + id: 's1', + name: 'Agent', + type: 'agent', + duration: 0, + startTime: '', + endTime: '', + model: 'gpt-4o', + cost: { input: 0.001, output: 0.002, total: 0.003 }, + tokens: { input: 10, output: 20, total: 30 }, + }, + ] as any + + const result = calculateCostSummary(spans, { baseExecutionCharge: 0 }) + + expect(result.baseExecutionCharge).toBe(0) + expect(result.totalCost).toBeCloseTo(0.003, 10) + }) + + test('defaults to the standard base charge when no option is passed', () => { + const withDefault = calculateCostSummary([]) + expect(withDefault.baseExecutionCharge).toBeGreaterThan(0) + expect(withDefault.totalCost).toBe(withDefault.baseExecutionCharge) + }) +}) diff --git a/apps/sim/lib/logs/execution/logging-factory.ts b/apps/sim/lib/logs/execution/logging-factory.ts index 011a6e50b70..8813fe275ea 100644 --- a/apps/sim/lib/logs/execution/logging-factory.ts +++ b/apps/sim/lib/logs/execution/logging-factory.ts @@ -143,16 +143,31 @@ function isMothershipUpdateCostOwned(span: CostTraceSpan): boolean { return span.type === 'mothership' } -export function calculateCostSummary(traceSpans: CostTraceSpan[] | undefined): CostSummary { +export interface CostSummaryOptions { + /** + * Per-run fixed charge folded into the total. Defaults to + * `BASE_EXECUTION_CHARGE`. Pass `0` for a run whose base charge is already + * paid by an invoking run — a custom block's child, for instance, is one + * logical run with its consumer and must not add a second execution fee. + */ + baseExecutionCharge?: number +} + +export function calculateCostSummary( + traceSpans: CostTraceSpan[] | undefined, + options?: CostSummaryOptions +): CostSummary { + const baseExecutionCharge = options?.baseExecutionCharge ?? BASE_EXECUTION_CHARGE + if (!traceSpans || traceSpans.length === 0) { return { - totalCost: BASE_EXECUTION_CHARGE, + totalCost: baseExecutionCharge, totalInputCost: 0, totalOutputCost: 0, totalTokens: 0, totalPromptTokens: 0, totalCompletionTokens: 0, - baseExecutionCharge: BASE_EXECUTION_CHARGE, + baseExecutionCharge, models: {}, workflowLedgerModels: {}, charges: {}, @@ -275,7 +290,7 @@ export function calculateCostSummary(traceSpans: CostTraceSpan[] | undefined): C } } - totalCost += BASE_EXECUTION_CHARGE + totalCost += baseExecutionCharge return { totalCost, @@ -284,7 +299,7 @@ export function calculateCostSummary(traceSpans: CostTraceSpan[] | undefined): C totalTokens, totalPromptTokens, totalCompletionTokens, - baseExecutionCharge: BASE_EXECUTION_CHARGE, + baseExecutionCharge, models, workflowLedgerModels, charges, diff --git a/apps/sim/lib/logs/execution/logging-session.test.ts b/apps/sim/lib/logs/execution/logging-session.test.ts index 4e9bf1f1646..fbb7d3f533f 100644 --- a/apps/sim/lib/logs/execution/logging-session.test.ts +++ b/apps/sim/lib/logs/execution/logging-session.test.ts @@ -311,7 +311,7 @@ describe('LoggingSession completion retries', () => { session.safeComplete({ finalOutput: { ok: true }, traceSpans }) ).resolves.toBeUndefined() - expect(calculateCostSummary).toHaveBeenLastCalledWith(traceSpans) + expect(calculateCostSummary).toHaveBeenLastCalledWith(traceSpans, undefined) expect(completeWorkflowExecutionMock).toHaveBeenLastCalledWith( expect.objectContaining({ executionId: 'execution-6', diff --git a/apps/sim/lib/logs/execution/logging-session.ts b/apps/sim/lib/logs/execution/logging-session.ts index d8ba08e5ad5..dcd2a0054cd 100644 --- a/apps/sim/lib/logs/execution/logging-session.ts +++ b/apps/sim/lib/logs/execution/logging-session.ts @@ -8,6 +8,7 @@ import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attr import { isRetryableInfrastructureError } from '@/lib/core/errors/retryable-infrastructure' import { executionLogger } from '@/lib/logs/execution/logger' import { + type CostSummaryOptions, calculateCostSummary, createEnvironmentObject, createTriggerObject, @@ -20,6 +21,7 @@ import { setLastCompletedBlock, setLastStartedBlock, } from '@/lib/logs/execution/progress-markers' +import { traceSpansIndicateFailure } from '@/lib/logs/execution/trace-spans/trace-spans' import type { ExecutionEnvironment, ExecutionFinalizationPath, @@ -133,6 +135,14 @@ export interface SessionPausedParams { workflowInput?: any } +export interface LoggingSessionOptions { + /** + * Overrides the per-run fixed charge. Pass `0` for a run whose base charge is + * already paid by its invoker, so it adds no second execution fee. + */ + baseExecutionCharge?: number +} + export class LoggingSession { private workflowId: string private executionId: string @@ -153,6 +163,7 @@ export class LoggingSession { private completionPromise: Promise | null = null private completionAttempt: CompletionAttempt | null = null private completionAttemptFailed = false + private costOptions?: CostSummaryOptions private pendingProgressWrites = new Set>() private postExecutionPromise: Promise | null = null @@ -161,13 +172,18 @@ export class LoggingSession { executionId: string, triggerType: ExecutionTrigger['type'], requestId?: string, - reservationId = executionId + reservationId = executionId, + options?: LoggingSessionOptions ) { this.workflowId = workflowId this.executionId = executionId this.reservationId = reservationId this.triggerType = triggerType this.requestId = requestId + this.costOptions = + options?.baseExecutionCharge !== undefined + ? { baseExecutionCharge: options.baseExecutionCharge } + : undefined } async onBlockStart( @@ -407,7 +423,7 @@ export class LoggingSession { params try { - const costSummary = calculateCostSummary(traceSpans || []) + const costSummary = calculateCostSummary(traceSpans || [], this.costOptions) const endTime = endedAt || new Date().toISOString() const duration = totalDurationMs || 0 @@ -430,16 +446,7 @@ export class LoggingSession { '@/lib/core/telemetry' ) - const hasErrors = traceSpans.some((span: any) => { - const checkForErrors = (s: any): boolean => { - if (s.status === 'error' && !s.errorHandled) return true - if (s.children && Array.isArray(s.children)) { - return s.children.some(checkForErrors) - } - return false - } - return checkForErrors(span) - }) + const hasErrors = traceSpansIndicateFailure(traceSpans) PlatformEvents.workflowExecuted({ workflowId: this.workflowId, @@ -528,7 +535,7 @@ export class LoggingSession { models: {}, charges: {}, } - : calculateCostSummary(traceSpans) + : calculateCostSummary(traceSpans, this.costOptions) const message = error?.message || 'Run failed before starting blocks' @@ -639,7 +646,7 @@ export class LoggingSession { // calculateCostSummary handles empty/undefined spans by returning the // base-charge summary, so no separate no-spans literal is needed. - const costSummary = calculateCostSummary(traceSpans) + const costSummary = calculateCostSummary(traceSpans, this.costOptions) await this.completeExecutionWithFinalization({ endedAt: endTime.toISOString(), @@ -733,7 +740,7 @@ export class LoggingSession { // calculateCostSummary handles empty/undefined spans by returning the // base-charge summary, so no separate no-spans literal is needed. - const costSummary = calculateCostSummary(traceSpans) + const costSummary = calculateCostSummary(traceSpans, this.costOptions) await this.completeExecutionWithFinalization({ endedAt: endTime.toISOString(), @@ -1138,7 +1145,7 @@ export class LoggingSession { // from the in-memory trace spans when available (this fallback fires when // persisting spans failed, not when computing them did), else just the // base execution charge. - const costSummary = calculateCostSummary(params.traceSpans) + const costSummary = calculateCostSummary(params.traceSpans, this.costOptions) const finalOutput = params.finalOutput || { _fallback: true, error: params.errorMessage } diff --git a/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts b/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts index f763e08b995..77b333bbe75 100644 --- a/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts +++ b/apps/sim/lib/logs/execution/trace-spans/trace-spans.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from 'vitest' -import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' +import { + buildTraceSpans, + hasUnhandledError, + traceSpansIndicateFailure, +} from '@/lib/logs/execution/trace-spans/trace-spans' +import type { TraceSpan } from '@/lib/logs/types' import { stripCustomToolPrefix } from '@/executor/constants' import type { ExecutionResult } from '@/executor/types' @@ -2318,3 +2323,61 @@ describe('nested subflow grouping via parentIterations', () => { }) }) }) + +describe('hasUnhandledError', () => { + const span = (overrides: Partial): TraceSpan => + ({ + id: 's', + name: 'n', + type: 'agent', + duration: 0, + startTime: '', + endTime: '', + ...overrides, + }) as TraceSpan + + it('reports an unhandled error', () => { + expect(hasUnhandledError(span({ status: 'error' }))).toBe(true) + }) + + it('ignores an error an error-handler path already handled', () => { + expect(hasUnhandledError(span({ status: 'error', errorHandled: true }))).toBe(false) + }) + + it('stops at a successful mothership boundary that recovered from a failed child', () => { + const boundary = span({ + type: 'mothership', + status: 'success', + children: [span({ status: 'error' })], + }) + + expect(hasUnhandledError(boundary)).toBe(false) + }) + + it('still descends through a successful workflow span', () => { + const parent = span({ + type: 'workflow', + status: 'success', + children: [span({ status: 'error' })], + }) + + expect(hasUnhandledError(parent)).toBe(true) + }) + + it('only counts failed tool calls when explicitly asked', () => { + const withFailedTool = span({ + status: 'success', + toolCalls: [{ name: 't', error: 'boom' }], + } as Partial) + + expect(hasUnhandledError(withFailedTool)).toBe(false) + expect(hasUnhandledError(withFailedTool, { includeToolCalls: true })).toBe(true) + }) +}) + +describe('traceSpansIndicateFailure', () => { + it('is false for no spans', () => { + expect(traceSpansIndicateFailure(undefined)).toBe(false) + expect(traceSpansIndicateFailure([])).toBe(false) + }) +}) diff --git a/apps/sim/lib/logs/execution/trace-spans/trace-spans.ts b/apps/sim/lib/logs/execution/trace-spans/trace-spans.ts index 858f9fd4939..f6fa1794a0b 100644 --- a/apps/sim/lib/logs/execution/trace-spans/trace-spans.ts +++ b/apps/sim/lib/logs/execution/trace-spans/trace-spans.ts @@ -115,7 +115,7 @@ function wrapInWorkflowRoot( duration: actualWorkflowDuration, startTime: new Date(earliestStart).toISOString(), endTime: new Date(latestEnd).toISOString(), - status: grouped.some(hasUnhandledError) ? 'error' : 'success', + status: traceSpansIndicateFailure(grouped) ? 'error' : 'success', children: grouped, ...(totalCost > 0 && { cost: { total: totalCost } }), } @@ -133,11 +133,40 @@ function addRelativeTimestamps(spans: TraceSpan[], workflowStartMs: number): voi } } -/** True if this span (or any descendant) has an unhandled error. */ -function hasUnhandledError(span: TraceSpan): boolean { +/** Options for {@link hasUnhandledError}. */ +export interface UnhandledErrorOptions { + /** + * Also treat a failed tool call as an unhandled error. Off by default: an + * agent that retried past a failed tool call still succeeded, so counting + * tool calls would flip those runs to failed. + */ + includeToolCalls?: boolean +} + +/** + * True if this span (or any descendant) has an error nothing handled. The + * single source of truth for "did this fail?" — used for the run's root span + * status, the persisted log level, and the UI's failure badge, so all three + * agree on error boundaries. + */ +export function hasUnhandledError(span: TraceSpan, options?: UnhandledErrorOptions): boolean { if (span.status === 'error' && !span.errorHandled) return true if (span.status === 'success' && SUCCESSFUL_CHILD_ERROR_BOUNDARY_BLOCK_TYPES.has(span.type)) { return false } - return span.children?.some(hasUnhandledError) ?? false + if (span.children?.length) { + return span.children.some((child) => hasUnhandledError(child, options)) + } + if (options?.includeToolCalls && span.toolCalls?.length && !span.errorHandled) { + return span.toolCalls.some((call) => call.error) + } + return false +} + +/** True when a completed run should be recorded as failed. */ +export function traceSpansIndicateFailure( + spans: TraceSpan[] | undefined, + options?: UnhandledErrorOptions +): boolean { + return spans?.some((span) => hasUnhandledError(span, options)) ?? false } diff --git a/apps/sim/lib/logs/get-trigger-options.ts b/apps/sim/lib/logs/get-trigger-options.ts index 42e3bacd610..5ff23e804ca 100644 --- a/apps/sim/lib/logs/get-trigger-options.ts +++ b/apps/sim/lib/logs/get-trigger-options.ts @@ -40,6 +40,7 @@ export function getTriggerOptions(): TriggerOption[] { { value: 'copilot', label: 'Sim agent', color: '#ec4899' }, { value: 'mothership', label: 'Sim agent', color: '#ec4899' }, { value: 'workflow', label: 'Workflow', color: '#0369a1' }, + { value: 'custom_block', label: 'Custom block', color: '#0369a1' }, ] for (const trigger of triggers) { diff --git a/apps/sim/lib/workflows/custom-blocks/child-execution.test.ts b/apps/sim/lib/workflows/custom-blocks/child-execution.test.ts new file mode 100644 index 00000000000..98bf889152b --- /dev/null +++ b/apps/sim/lib/workflows/custom-blocks/child-execution.test.ts @@ -0,0 +1,304 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockCheckAttributedUsageLimits, + mockSubscribe, + mockUnsubscribe, + mockIsExecutionCancelled, + mockIsRedisCancellationEnabled, +} = vi.hoisted(() => ({ + mockCheckAttributedUsageLimits: vi.fn(), + mockSubscribe: vi.fn(), + mockUnsubscribe: vi.fn(), + mockIsExecutionCancelled: vi.fn(), + mockIsRedisCancellationEnabled: vi.fn(), +})) + +vi.mock('@/lib/billing/core/billing-attribution', () => ({ + checkAttributedUsageLimits: mockCheckAttributedUsageLimits, +})) + +vi.mock('@/lib/execution/cancellation', () => ({ + getCancellationChannel: () => ({ subscribe: mockSubscribe }), + isExecutionCancelled: mockIsExecutionCancelled, + isRedisCancellationEnabled: mockIsRedisCancellationEnabled, +})) + +import { + admitCustomBlockChildExecution, + buildCustomBlockCorrelation, + CustomBlockAdmissionError, + createChildCancellationSignal, + trackChildRun, + waitForChildRuns, +} from '@/lib/workflows/custom-blocks/child-execution' +import { isBoundarySafeError } from '@/executor/errors/boundary' + +const attribution = { actorUserId: 'owner-1', workspaceId: 'workspace-source' } as any + +describe('admitCustomBlockChildExecution', () => { + beforeEach(() => { + vi.clearAllMocks() + mockSubscribe.mockReturnValue(mockUnsubscribe) + mockIsRedisCancellationEnabled.mockReturnValue(true) + mockIsExecutionCancelled.mockResolvedValue(false) + }) + + it('passes when the source payer has headroom', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: false }) + + await expect(admitCustomBlockChildExecution(attribution)).resolves.toBeUndefined() + }) + + it('forwards the payer-scoped message, which describes the shared org', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + scope: 'payer', + message: 'Organization usage limit exceeded: $50.00 pooled of $50.00 organization limit.', + }) + + await expect(admitCustomBlockChildExecution(attribution)).rejects.toThrow( + 'Organization usage limit exceeded: $50.00 pooled of $50.00 organization limit.' + ) + }) + + it("never forwards the owner's personal member-cap message to a consumer", async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + scope: 'member', + message: + 'Member credit limit exceeded: 900 of 1,000 credits used. Ask an admin to raise your credit limit.', + }) + + const error = await admitCustomBlockChildExecution(attribution).catch((e: Error) => e) + + expect(error).toBeInstanceOf(CustomBlockAdmissionError) + expect(error.message).not.toContain('Member credit limit') + expect(error.message).not.toContain('900') + expect(error.message).toBe( + 'This custom block is unavailable because a usage limit was reached. Ask an organization admin to review it.' + ) + }) + + it("never forwards the owner's account-block message to a consumer", async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ + isExceeded: true, + scope: 'actor', + message: 'Account frozen. Please contact support to resolve this issue.', + }) + + const error = await admitCustomBlockChildExecution(attribution).catch((e: Error) => e) + + expect(error.message).not.toContain('Account frozen') + expect(error.message).not.toContain('support') + }) + + it('takes no concurrency reservation', async () => { + mockCheckAttributedUsageLimits.mockResolvedValue({ isExceeded: false }) + + await admitCustomBlockChildExecution(attribution) + + expect(mockCheckAttributedUsageLimits).toHaveBeenCalledTimes(1) + }) +}) + +describe('buildCustomBlockCorrelation', () => { + it('records the invoking run without naming anything', () => { + const correlation = buildCustomBlockCorrelation({ + invokerExecutionId: 'exec-1', + invokerRequestId: 'req-1', + invokerWorkflowId: 'wf-1', + invokerWorkspaceId: 'ws-consumer', + blockType: 'custom_block_abc', + }) + + expect(correlation).toEqual({ + source: 'custom_block', + executionId: 'exec-1', + requestId: 'req-1', + workflowId: 'wf-1', + triggerType: 'custom_block_abc', + invokerWorkspaceId: 'ws-consumer', + }) + }) + + it('is undefined without an invoking execution id', () => { + expect(buildCustomBlockCorrelation({ blockType: 'custom_block_abc' })).toBeUndefined() + }) +}) + +describe('createChildCancellationSignal', () => { + beforeEach(() => { + vi.clearAllMocks() + mockSubscribe.mockReturnValue(mockUnsubscribe) + mockIsRedisCancellationEnabled.mockReturnValue(true) + mockIsExecutionCancelled.mockResolvedValue(false) + }) + + it('aborts when the parent signal aborts', async () => { + const parent = new AbortController() + const { signal } = await createChildCancellationSignal({ + parentSignal: parent.signal, + parentExecutionId: 'parent-1', + }) + + expect(signal.aborted).toBe(false) + parent.abort() + expect(signal.aborted).toBe(true) + }) + + it('starts aborted when the parent already aborted', async () => { + const parent = new AbortController() + parent.abort() + + const { signal } = await createChildCancellationSignal({ parentSignal: parent.signal }) + + expect(signal.aborted).toBe(true) + }) + + it('aborts on the parent cancellation event, and ignores other runs', async () => { + const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' }) + const handler = mockSubscribe.mock.calls[0][0] + + handler({ executionId: 'someone-else' }) + expect(signal.aborted).toBe(false) + + handler({ executionId: 'parent-1' }) + expect(signal.aborted).toBe(true) + }) + + it('unsubscribes on dispose so a looped block leaks nothing', async () => { + const parent = new AbortController() + const { dispose } = await createChildCancellationSignal({ + parentSignal: parent.signal, + parentExecutionId: 'parent-1', + }) + + dispose() + + expect(mockUnsubscribe).toHaveBeenCalledTimes(1) + }) +}) + +describe('CustomBlockAdmissionError', () => { + it('is boundary-safe so the consumer sees a usage-limit classification', () => { + const error = new CustomBlockAdmissionError('Organization usage limit exceeded') + + expect(isBoundarySafeError(error)).toBe(true) + expect(error.errorType).toBe('usage_limit') + expect(error.message).toBe('Organization usage limit exceeded') + }) +}) + +describe('createChildCancellationSignal durable backstop', () => { + beforeEach(() => { + vi.clearAllMocks() + mockSubscribe.mockReturnValue(mockUnsubscribe) + mockIsRedisCancellationEnabled.mockReturnValue(true) + mockIsExecutionCancelled.mockResolvedValue(false) + }) + + it('aborts on a cancel published before the bridge subscribed', async () => { + // The pub/sub event is long gone; only the durable key remains. + mockIsExecutionCancelled.mockResolvedValue(true) + + const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' }) + + expect(mockIsExecutionCancelled).toHaveBeenCalledWith('parent-1') + expect(signal.aborted).toBe(true) + }) + + it('subscribes before reading the durable key so no window is left open', async () => { + const order: string[] = [] + mockSubscribe.mockImplementation(() => { + order.push('subscribe') + return mockUnsubscribe + }) + mockIsExecutionCancelled.mockImplementation(async () => { + order.push('durable-read') + return false + }) + + await createChildCancellationSignal({ parentExecutionId: 'parent-1' }) + + expect(order).toEqual(['subscribe', 'durable-read']) + }) + + it('fails open when the durable read throws', async () => { + mockIsExecutionCancelled.mockRejectedValue(new Error('redis down')) + + const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' }) + + expect(signal.aborted).toBe(false) + }) + + it('skips the durable read when redis cancellation is unavailable', async () => { + mockIsRedisCancellationEnabled.mockReturnValue(false) + + const { signal } = await createChildCancellationSignal({ parentExecutionId: 'parent-1' }) + + expect(mockIsExecutionCancelled).not.toHaveBeenCalled() + expect(signal.aborted).toBe(false) + }) +}) + +describe('child run tracking', () => { + it('waits for a child the engine abandoned mid-execution', async () => { + // The scenario this exists for: the parent cancelled, so at drain time the + // child has NOT finished. Registration must already have happened. + let settle: () => void = () => {} + let finished = false + const childRun = new Promise((resolve) => { + settle = resolve + }).then(() => { + finished = true + }) + + trackChildRun('invoker-1', childRun) + + const drained = waitForChildRuns('invoker-1') + expect(finished).toBe(false) + settle() + await drained + + expect(finished).toBe(true) + }) + + it('gives up on a hung child rather than wedging the invoking run', async () => { + vi.useFakeTimers() + try { + // Never settles — an unbounded drain would hang the parent forever. + trackChildRun('invoker-hung', new Promise(() => {})) + + const drained = waitForChildRuns('invoker-hung') + await vi.advanceTimersByTimeAsync(10_000) + + await expect(drained).resolves.toBeUndefined() + } finally { + vi.useRealTimers() + } + }) + + it('does not let a rejected child break the drain', async () => { + trackChildRun('invoker-2', Promise.reject(new Error('db down'))) + + await expect(waitForChildRuns('invoker-2')).resolves.toBeUndefined() + }) + + it('is a no-op for a run that registered nothing', async () => { + await expect(waitForChildRuns('invoker-never')).resolves.toBeUndefined() + await expect(waitForChildRuns(undefined)).resolves.toBeUndefined() + }) + + it('releases its entry so a run that never drains cannot leak', async () => { + const childRun = Promise.resolve() + trackChildRun('invoker-3', childRun) + await childRun + await Promise.resolve() + + await expect(waitForChildRuns('invoker-3')).resolves.toBeUndefined() + }) +}) diff --git a/apps/sim/lib/workflows/custom-blocks/child-execution.ts b/apps/sim/lib/workflows/custom-blocks/child-execution.ts new file mode 100644 index 00000000000..6466dd676ea --- /dev/null +++ b/apps/sim/lib/workflows/custom-blocks/child-execution.ts @@ -0,0 +1,237 @@ +import { createLogger } from '@sim/logger' +import { + type BillingAttributionSnapshot, + checkAttributedUsageLimits, +} from '@/lib/billing/core/billing-attribution' +import type { AsyncExecutionCorrelation } from '@/lib/core/async-jobs/types' +import { + getCancellationChannel, + isExecutionCancelled, + isRedisCancellationEnabled, +} from '@/lib/execution/cancellation' +import { BoundarySafeError } from '@/executor/errors/boundary' + +const logger = createLogger('CustomBlockChildExecution') + +/** + * The payer has no headroom for this custom-block child run. + * + * Boundary-safe only as far as the message allows: see + * {@link admitCustomBlockChildExecution} for why actor-scoped denials are + * collapsed to {@link GENERIC_USAGE_LIMIT_MESSAGE} before reaching here. + */ +export class CustomBlockAdmissionError extends BoundarySafeError { + constructor(message: string) { + super({ message, errorType: 'usage_limit' }) + this.name = 'CustomBlockAdmissionError' + } +} + +/** + * Consumer-safe stand-in for a denial whose real message describes the SOURCE + * workflow owner's personal billing state rather than the shared organization. + */ +const GENERIC_USAGE_LIMIT_MESSAGE = + 'This custom block is unavailable because a usage limit was reached. Ask an organization admin to review it.' + +/** + * Admits one custom-block child run against the SOURCE workspace's payer. + * + * Performs the attributed usage check but takes no concurrency reservation. + * The consumer and source workspaces are always in the same organization (the + * publish gate plus `getCustomBlockAuthority`'s org filter), so `billingEntity` + * is identical for parent and child: reserving again would burn a second slot + * from the same payer for one logical run — N+1 slots for a custom block inside + * an N-iteration loop — and fail runs that work today. There is also no base + * charge on the child for a reservation to protect. + */ +export async function admitCustomBlockChildExecution( + attribution: BillingAttributionSnapshot +): Promise { + const usage = await checkAttributedUsageLimits(attribution) + if (!usage.isExceeded) return + + // Only the payer-scoped denial describes the shared organization ("Organization + // usage limit exceeded: $X pooled of $Y organization limit"), which both sides + // genuinely share and which the consumer can act on. + // + // The other gates are evaluated against `actorUserId` — the SOURCE workflow's + // owner — and their text is addressed to that person: their account-frozen / + // payment-method state, or "Ask an organization admin to raise YOUR credit + // limit" against their individual member cap. Forwarding those verbatim would + // show one org member another's private billing state. Being in the same + // organization makes the *payer* shared; it does not make personal quotas + // shared. The `usage_limit` type still tells the consumer what kind of failure + // this was. + const message = + usage.scope === 'payer' && usage.message ? usage.message : GENERIC_USAGE_LIMIT_MESSAGE + throw new CustomBlockAdmissionError(message) +} + +/** + * Correlation linking a custom-block child's log row back to its invoking run. + * Ids only — no names — so the publisher can trace an invocation without seeing + * anything about the consumer's workflow contents. + */ +export function buildCustomBlockCorrelation(params: { + invokerExecutionId?: string + invokerRequestId?: string + invokerWorkflowId?: string + invokerWorkspaceId?: string + blockType: string +}): AsyncExecutionCorrelation | undefined { + if (!params.invokerExecutionId) return undefined + return { + source: 'custom_block', + executionId: params.invokerExecutionId, + requestId: params.invokerRequestId ?? params.invokerExecutionId, + workflowId: params.invokerWorkflowId ?? '', + triggerType: params.blockType, + ...(params.invokerWorkspaceId ? { invokerWorkspaceId: params.invokerWorkspaceId } : {}), + } +} + +/** + * Abort signal for a custom-block child. The child runs under its own execution + * id, so it no longer receives the parent's cancellation pub/sub event (the + * engine matches on execution id) — this bridges both the parent's abort signal + * and its cancellation channel onto a signal the child engine honours. + * + * Callers MUST call `dispose()`: a custom block inside a loop would otherwise + * leak one abort listener and one channel subscription per iteration onto a + * long-lived parent signal. + */ +export async function createChildCancellationSignal(params: { + parentSignal?: AbortSignal + parentExecutionId?: string +}): Promise<{ signal: AbortSignal; dispose: () => void }> { + const controller = new AbortController() + + if (params.parentSignal?.aborted) { + controller.abort(params.parentSignal.reason) + return { signal: controller.signal, dispose: () => {} } + } + + const onParentAbort = () => controller.abort(params.parentSignal?.reason) + params.parentSignal?.addEventListener('abort', onParentAbort, { once: true }) + + let unsubscribe: (() => void) | undefined + if (params.parentExecutionId) { + const parentExecutionId = params.parentExecutionId + // Subscribe BEFORE the durable read, mirroring `markExecutionCancelled`'s + // write-durable-then-publish order: a cancel published during the read is + // caught by the subscription, and one published earlier — while the child's + // session and admission were still being set up — by the read itself. The + // child's own engine backstop cannot cover this, since it checks the CHILD's + // execution id, which is never the one marked cancelled. + unsubscribe = getCancellationChannel().subscribe((event) => { + if (event.executionId === parentExecutionId) controller.abort() + }) + if (isRedisCancellationEnabled()) { + try { + if (await isExecutionCancelled(parentExecutionId)) controller.abort() + } catch { + // Fail open, matching the engine's own backstop: a failed read must not + // stop a child whose parent was never actually cancelled. + } + } + } + + return { + signal: controller.signal, + dispose: () => { + params.parentSignal?.removeEventListener('abort', onParentAbort) + unsubscribe?.() + }, + } +} + +/** + * Custom-block child runs still in flight, keyed by the INVOKING run's execution id. + * + * A cancelled or timed-out parent engine returns without draining its in-flight + * node promises, so the child's execution AND its terminal log write are both + * abandoned mid-flight: the invoking run finishes, the worker tears down, and the + * child's row is left `running` until the stale-execution reaper sweeps it. + * + * The tracked promise therefore covers the WHOLE child lifecycle and is registered + * BEFORE the child starts. Registering only the finalization step would be useless + * on exactly the path this exists for: at drain time the child is still inside + * `execute`, so nothing would be registered yet and the drain would no-op. + * + * Only the immediate invoker is tracked. A child orphaned *below* another abandoned + * child still falls back to the reaper. + */ +const pendingChildRuns = new Map>>() + +/** How long a run will wait on abandoned children before leaving them to the reaper. */ +const CHILD_RUN_DRAIN_TIMEOUT_MS = 10_000 + +/** + * Registers a child run against its invoking run. The promise must settle when the + * child is fully done — execution finished AND its session written. + */ +export function trackChildRun( + invokerExecutionId: string | undefined, + childRun: Promise +): void { + if (!invokerExecutionId) return + + let pending = pendingChildRuns.get(invokerExecutionId) + if (!pending) { + pending = new Set() + pendingChildRuns.set(invokerExecutionId, pending) + } + pending.add(childRun) + + // Self-cleaning so an invoker that never drains cannot leak the entry. The + // `catch` also marks the rejection handled — the drain uses `allSettled`. + void childRun + .catch(() => {}) + .finally(() => { + pending.delete(childRun) + if (pending.size === 0) pendingChildRuns.delete(invokerExecutionId) + }) +} + +/** + * Awaits every child run registered for this invoking run, bounded. + * + * The bound matters: on the normal path the handler already awaited its child, so + * this returns immediately. It only blocks when a child was abandoned, where the + * bridged abort should end it in milliseconds. Waiting unbounded would trade a + * stale `running` row — which the reaper fixes — for a wedged parent run, which + * nothing fixes. + */ +export async function waitForChildRuns(invokerExecutionId: string | undefined): Promise { + if (!invokerExecutionId) return + + const pending = pendingChildRuns.get(invokerExecutionId) + if (!pending || pending.size === 0) return + + let timer: ReturnType | undefined + const timedOut = new Promise<'timeout'>((resolve) => { + timer = setTimeout(() => resolve('timeout'), CHILD_RUN_DRAIN_TIMEOUT_MS) + }) + + const drained = (async () => { + // Re-check: a nested child can still register while we await. + while (pending.size > 0) { + await Promise.allSettled([...pending]) + } + return 'drained' as const + })() + + try { + if ((await Promise.race([drained, timedOut])) === 'timeout') { + logger.warn('Abandoned custom block child did not finish; leaving it to the reaper', { + invokerExecutionId, + outstanding: pending.size, + }) + return + } + pendingChildRuns.delete(invokerExecutionId) + } finally { + if (timer) clearTimeout(timer) + } +} diff --git a/apps/sim/lib/workflows/custom-blocks/operations.ts b/apps/sim/lib/workflows/custom-blocks/operations.ts index 975b479dcba..4056e4db2c4 100644 --- a/apps/sim/lib/workflows/custom-blocks/operations.ts +++ b/apps/sim/lib/workflows/custom-blocks/operations.ts @@ -398,6 +398,26 @@ export class CustomBlockValidationError extends Error { * Reject exposed outputs whose name shadows a system output field. Authoritative * check: also covers callers that bypass the HTTP contract (copilot handler). */ +/** + * Every field a consumer receives must be one the publisher explicitly chose. + * + * There is deliberately no "expose the whole result" fallback: the child's + * terminal block state carries execution metadata that is legitimate INSIDE a + * workflow but must not cross an invocation boundary — an agent's `toolCalls` + * (arguments and results), `providerTiming.thinkingContent` and `cost`, or a + * nested workflow block's name/id/snapshot id. Curation is what makes the + * boundary's guarantee structural instead of a deny-list someone has to keep + * current as new block types gain outputs. + * + * Authoritative here as well as in the route contract, because the copilot + * publish handler bypasses the HTTP boundary. + */ +function assertCuratedOutputs(exposedOutputs: CustomBlockOutput[] | undefined): void { + if (!exposedOutputs || exposedOutputs.length === 0) { + throw new CustomBlockValidationError('Select at least one output to expose to consumers') + } +} + function assertNoReservedOutputNames(exposedOutputs: CustomBlockOutput[] | undefined): void { const reserved = exposedOutputs?.find((o) => isReservedOutputName(o.name)) if (reserved) { @@ -424,7 +444,8 @@ export async function publishCustomBlock(params: { description: string iconUrl?: string inputs?: InputPlaceholder[] - exposedOutputs?: CustomBlockOutput[] + /** Required — see {@link assertCuratedOutputs}. */ + exposedOutputs: CustomBlockOutput[] }): Promise { const { organizationId, @@ -439,6 +460,7 @@ export async function publishCustomBlock(params: { } = params assertNoReservedOutputNames(exposedOutputs) + assertCuratedOutputs(exposedOutputs) const [wf] = await db .select({ @@ -534,7 +556,10 @@ export async function updateCustomBlock( exposedOutputs?: CustomBlockOutput[] } ): Promise { - if (updates.exposedOutputs !== undefined) assertNoReservedOutputNames(updates.exposedOutputs) + if (updates.exposedOutputs !== undefined) { + assertNoReservedOutputNames(updates.exposedOutputs) + assertCuratedOutputs(updates.exposedOutputs) + } const patch: Partial = { updatedAt: new Date() } if (updates.name !== undefined) patch.name = updates.name if (updates.description !== undefined) patch.description = updates.description diff --git a/apps/sim/lib/workflows/executor/execution-core.ts b/apps/sim/lib/workflows/executor/execution-core.ts index 5733adef130..f58c89c72cb 100644 --- a/apps/sim/lib/workflows/executor/execution-core.ts +++ b/apps/sim/lib/workflows/executor/execution-core.ts @@ -22,6 +22,7 @@ import { redactLargeValueRefsInValue } from '@/lib/logs/execution/pii-large-valu import { redactObjectStrings } from '@/lib/logs/execution/pii-redaction' import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans' import { getUserEmailById } from '@/lib/users/queries' +import { waitForChildRuns } from '@/lib/workflows/custom-blocks/child-execution' import { getCustomBlockRowsForWorkspace } from '@/lib/workflows/custom-blocks/operations' import { loadDeployedWorkflowState, @@ -375,6 +376,11 @@ async function executeWorkflowCoreImpl( while (pendingLifecycleCallbacks.size > 0) { await Promise.allSettled([...pendingLifecycleCallbacks]) } + // A custom block's child is a separate execution with its own log row, and + // the engine does not drain in-flight nodes on cancel/timeout — await it here + // (bounded) so the row is not left `running` when this run finishes or the + // worker exits. + await waitForChildRuns(executionId) } try { diff --git a/apps/sim/providers/types.ts b/apps/sim/providers/types.ts index 2bb1a9d31c0..b324cdb7917 100644 --- a/apps/sim/providers/types.ts +++ b/apps/sim/providers/types.ts @@ -206,6 +206,13 @@ export interface ProviderRequest { blockId?: string isDeployedContext?: boolean callChain?: string[] + /** + * The invoking run's execution id. Propagated into the `_context` of every + * tool the LLM invokes so a tool that starts its own child execution (a + * custom block) can correlate that child back to a REAL invoking run rather + * than a freshly-minted id, and can honour its cancellation. + */ + executionId?: string /** * Immutable actor/payer decision captured before execution. Propagated into * the `_context` of every tool the LLM invokes so internal routes that diff --git a/apps/sim/providers/utils.test.ts b/apps/sim/providers/utils.test.ts index 0b10c2bf511..f959cf7911c 100644 --- a/apps/sim/providers/utils.test.ts +++ b/apps/sim/providers/utils.test.ts @@ -1730,3 +1730,40 @@ describe('transformBlockTool knowledge-base multi-instance unique IDs', () => { expect(result?.id).toBe('knowledge_search') }) }) + +describe('prepareToolExecution invoker identity hand-off', () => { + const tool = { params: {}, parameters: {} } + + /** + * A custom block invoked as an agent tool starts its own child execution, and + * correlates + cancels against the INVOKING run. That id only reaches it via + * `_context`, so this asserts the hand-off rather than any single hop — three + * separate fixes each repaired one hop and left the chain broken elsewhere. + */ + it("puts the invoking run's execution id on tool _context", () => { + const { executionParams } = prepareToolExecution( + tool, + {}, + { + workflowId: 'wf-1', + workspaceId: 'ws-1', + executionId: 'real-execution-id', + } + ) + + expect(executionParams._context.executionId).toBe('real-execution-id') + }) + + it('omits the execution id when the request carries none', () => { + const { executionParams } = prepareToolExecution( + tool, + {}, + { + workflowId: 'wf-1', + workspaceId: 'ws-1', + } + ) + + expect(executionParams._context.executionId).toBeUndefined() + }) +}) diff --git a/apps/sim/providers/utils.ts b/apps/sim/providers/utils.ts index 05d60e37a12..60ec451ea5d 100644 --- a/apps/sim/providers/utils.ts +++ b/apps/sim/providers/utils.ts @@ -1423,6 +1423,8 @@ export function prepareToolExecution( isDeployedContext?: boolean callChain?: string[] billingAttribution?: BillingAttributionSnapshot + /** Invoking run's execution id — see `ProviderRequest.executionId`. */ + executionId?: string } ): { toolParams: Record @@ -1451,6 +1453,7 @@ export function prepareToolExecution( ? { isDeployedContext: request.isDeployedContext } : {}), ...(request.callChain ? { callChain: request.callChain } : {}), + ...(request.executionId ? { executionId: request.executionId } : {}), ...(request.billingAttribution ? { billingAttribution: request.billingAttribution } : {}), diff --git a/apps/sim/stores/logs/filters/types.ts b/apps/sim/stores/logs/filters/types.ts index 1524bba87ec..c2f8c874968 100644 --- a/apps/sim/stores/logs/filters/types.ts +++ b/apps/sim/stores/logs/filters/types.ts @@ -31,6 +31,7 @@ export const CORE_TRIGGER_TYPES = [ 'copilot', 'mothership', 'workflow', + 'custom_block', ] as const export type CoreTriggerType = (typeof CORE_TRIGGER_TYPES)[number] diff --git a/apps/sim/tools/errors.ts b/apps/sim/tools/errors.ts new file mode 100644 index 00000000000..8a8809258a7 --- /dev/null +++ b/apps/sim/tools/errors.ts @@ -0,0 +1,28 @@ +import { HttpError } from '@/lib/core/utils/http-error' + +/** + * Hosted-key acquisition blocked by the workspace's own rate bucket. Carries a + * 429 so the status survives block-level wrapping and reaches the API caller + * instead of being flattened to a generic 500. + */ +export class HostedKeyRateLimitedError extends HttpError { + readonly statusCode = 429 + + constructor( + message: string, + readonly retryAfterMs?: number + ) { + super(message) + this.name = 'HostedKeyRateLimitedError' + } +} + +/** No hosted keys are configured or available for this provider. */ +export class HostedKeyUnavailableError extends HttpError { + readonly statusCode = 503 + + constructor(message: string) { + super(message) + this.name = 'HostedKeyUnavailableError' + } +} diff --git a/apps/sim/tools/index.ts b/apps/sim/tools/index.ts index 62af30c8e1a..e7fbf1bafe3 100644 --- a/apps/sim/tools/index.ts +++ b/apps/sim/tools/index.ts @@ -18,6 +18,7 @@ import { validateUrlWithDNS, } from '@/lib/core/security/input-validation.server' import { PlatformEvents } from '@/lib/core/telemetry' +import { HttpError } from '@/lib/core/utils/http-error' import { generateRequestId } from '@/lib/core/utils/request' import { isPayloadSizeLimitError, @@ -37,6 +38,7 @@ import type { ExecutionContext, UserFile } from '@/executor/types' import { resolveEnvVarReferences } from '@/executor/utils/reference-validation' import type { ErrorInfo } from '@/tools/error-extractors' import { extractErrorMessage } from '@/tools/error-extractors' +import { HostedKeyRateLimitedError, HostedKeyUnavailableError } from '@/tools/errors' import type { BYOKProviderId, OAuthTokenPayload, @@ -366,18 +368,18 @@ async function injectHostedKeyIfNeeded( workflowId, }) - const error = new Error(acquireResult.error || `Rate limit exceeded for ${tool.id}`) - ;(error as any).status = 429 - ;(error as any).retryAfterMs = acquireResult.retryAfterMs - throw error + throw new HostedKeyRateLimitedError( + acquireResult.error || `Rate limit exceeded for ${tool.id}`, + acquireResult.retryAfterMs + ) } // Handle no keys configured (503) if (!acquireResult.success) { logger.error(`[${requestId}] No hosted keys configured for ${tool.id}: ${acquireResult.error}`) - const error = new Error(acquireResult.error || `No hosted keys configured for ${tool.id}`) - ;(error as any).status = 503 - throw error + throw new HostedKeyUnavailableError( + acquireResult.error || `No hosted keys configured for ${tool.id}` + ) } params[apiKeyParam] = acquireResult.key @@ -1276,7 +1278,20 @@ export async function executeTool( const { runCustomBlockTool } = await import( '@/executor/handlers/workflow/custom-block-tool-runner' ) - const result = await runCustomBlockTool(contextParams) + // Forward the INVOKING run's identifiers so the child's log correlation + // names a real execution instead of a freshly-minted phantom id. Taken + // from the server-resolved scope, never from model-supplied params. + const result = await runCustomBlockTool( + { + ...contextParams, + _context: { + ...(contextParams._context as Record | undefined), + ...(scope.executionId ? { executionId: scope.executionId } : {}), + requestId, + }, + }, + { abortSignal: effectiveSignal } + ) const endTime = new Date() return { ...result, @@ -1501,6 +1516,10 @@ export async function executeTool( success: false, output: errorDetails, error: errorMessage, + // Sim's own status (hosted-key 429/503) survives the flattening from a + // thrown error into a result object; an upstream provider's status stays + // on `output` where it cannot be mistaken for ours. + ...(error instanceof HttpError ? { statusCode: error.statusCode } : {}), timing: { startTime: startTimeISO, endTime: endTimeISO, diff --git a/apps/sim/tools/types.ts b/apps/sim/tools/types.ts index 91426b9ad6c..53c7a2a461e 100644 --- a/apps/sim/tools/types.ts +++ b/apps/sim/tools/types.ts @@ -93,6 +93,13 @@ export interface ToolResponse { success: boolean // Whether the tool execution was successful output: Record // The structured output from the tool error?: string // Error message if success is false + /** + * HTTP status owned by SIM itself (e.g. hosted-key rate limiting or + * exhaustion), carried so it survives the throw → `ToolResponse` flattening + * and can reach the API caller. Deliberately NOT the upstream provider's + * status — a provider's 404 must never become the workflow API's status. + */ + statusCode?: number resources?: MothershipResource[] // Resources to auto-open/show in UI largeValueKeys?: string[] fileKeys?: string[] diff --git a/packages/utils/src/errors.test.ts b/packages/utils/src/errors.test.ts index 272c85e53a4..736264e051f 100644 --- a/packages/utils/src/errors.test.ts +++ b/packages/utils/src/errors.test.ts @@ -2,7 +2,7 @@ * @vitest-environment node */ import { describe, expect, it } from 'vitest' -import { describeError, getPostgresErrorCode, toError } from './errors.js' +import { describeError, findCause, getPostgresErrorCode, toError } from './errors.js' describe('toError', () => { it('returns the same Error when given an Error', () => { @@ -127,3 +127,31 @@ describe('describeError', () => { expect(described?.causeChain?.length).toBeLessThanOrEqual(10) }) }) + +describe('findCause', () => { + class Marker extends Error {} + const isMarker = (value: unknown): value is Marker => value instanceof Marker + + it('returns the error itself when it matches', () => { + const marker = new Marker('hit') + expect(findCause(marker, isMarker)).toBe(marker) + }) + + it('finds a match further down the cause chain', () => { + const marker = new Marker('deep') + const wrapped = new Error('outer', { cause: new Error('middle', { cause: marker }) }) + expect(findCause(wrapped, isMarker)).toBe(marker) + }) + + it('returns undefined when nothing matches', () => { + expect(findCause(new Error('plain'), isMarker)).toBeUndefined() + }) + + it('survives a cyclic chain', () => { + const a = new Error('a') + const b = new Error('b') + Object.assign(a, { cause: b }) + Object.assign(b, { cause: a }) + expect(findCause(a, isMarker)).toBeUndefined() + }) +}) diff --git a/packages/utils/src/errors.ts b/packages/utils/src/errors.ts index dc21d57b995..f3f4a45deda 100644 --- a/packages/utils/src/errors.ts +++ b/packages/utils/src/errors.ts @@ -93,6 +93,28 @@ export function describeError(error: unknown): DescribedError { } } +/** + * First link in the `.cause` chain (including `error` itself) matching + * `predicate`. Lets a caller recover a specific wrapped error class instead of + * re-parsing a formatted message. Cycle-safe and depth-bounded, mirroring + * {@link describeError}'s walk. + */ +export function findCause( + error: unknown, + predicate: (value: unknown) => value is T +): T | undefined { + const seen = new Set() + let current: unknown = error + + while (current instanceof Error && !seen.has(current) && seen.size < 10) { + seen.add(current) + if (predicate(current)) return current + current = current.cause + } + + return undefined +} + function readPgErrorField(error: unknown, field: string): string | undefined { const seen = new Set() let current: unknown = error