feat(custom-blocks): log and bill child runs in the publisher's workspace (#6023)

* feat(custom-blocks): log and bill child runs in the publisher's workspace

* fix(custom-blocks): sanitize every boundary failure and classify it for consumers

* fix(custom-blocks): surface a cancelled child as cancelled, not a generic failure

* fix(custom-blocks): share one large-value id list so nested blocks propagate

* fix(custom-blocks): add a durable cancel backstop to the child bridge

* fix(tools): carry Sim's own status through the tool-response boundary

* fix(custom-blocks): stop forwarding the publisher's personal quota to consumers

* fix(custom-blocks): correlate agent-tool runs to the real invoking execution

* fix(custom-blocks): plumb the invoking execution id and abort signal to agent tools

* fix(agent): forward the execution id through the provider payload

* fix(executor): stop adopting an upstream target's HTTP status as our own

* fix(custom-blocks): drain child log finalization when the parent is cancelled

* fix(custom-blocks): require curated outputs instead of exposing the whole result

* fix(custom-blocks): track the whole child run, not just its finalization
This commit is contained in:
Theodore Li
2026-07-30 21:20:58 -04:00
committed by GitHub
parent 64bcfead34
commit 7293b67a54
40 changed files with 2307 additions and 371 deletions
@@ -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(
@@ -85,6 +85,7 @@ const TRIGGER_VARIANT_MAP: Record<string, React.ComponentProps<typeof Badge>['va
copilot: 'pink',
mothership: 'pink',
workflow: 'blue-secondary',
custom_block: 'blue-secondary',
}
interface StatusBadgeProps {
+11 -2
View File
@@ -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()
})
+14 -11
View File
@@ -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
}
+54
View File
@@ -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
}
@@ -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}`
}
+19 -3
View File
@@ -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
}
}
}
@@ -1126,6 +1126,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),
@@ -1209,6 +1210,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,
@@ -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
@@ -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()
})
})
@@ -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<ToolResponse> {
export async function runCustomBlockTool(
params: CustomBlockToolParams,
options: { abortSignal?: AbortSignal } = {}
): Promise<ToolResponse> {
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<string, any> =
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 }
}
}
@@ -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<Record<string, any>>,
loggingSessionArgs: [] as Array<any[]>,
}))
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<string, any>) =>
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<string, any> = {}) {
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()
})
})
})
@@ -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<unknown>((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<string, unknown>
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<void>((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 `"<source name>" failed: <internal error>`), 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<string, any>
): Promise<void> {
const { traceSpans, totalDuration } = buildTraceSpans(executionResult)
const endedAt = new Date().toISOString()
const totalDurationMs = totalDuration ?? Math.round(durationMs)
// Extract any nested workflow names from the error message
const { chain, rootError } = this.parseNestedWorkflowError(originalError)
// Add current workflow to the beginning of the chain
chain.unshift(childWorkflowName)
// If we have a chain (nested workflows), format nicely
if (chain.length > 1) {
return `Workflow chain: ${chain.join(' → ')} | ${rootError}`
// 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
}
// Single workflow failure
return `"${childWorkflowName}" failed: ${rootError}`
await session.safeComplete({
endedAt,
totalDurationMs,
finalOutput: executionResult.output ?? {},
traceSpans,
workflowInput,
})
}
/** Records a custom-block child's failure on its own log row. */
private async failChildSession(session: LoggingSession, error: unknown): Promise<void> {
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,
})
}
/**
* 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)
* 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.
*
* Handles formats like:
* - "workflow-name" failed: error
* - Block Name: "workflow-name" failed: error
* - Workflow chain: A → B | error
* 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 parseNestedWorkflowError(message: string): { chain: string[]; rootError: string } {
const chain: string[] = []
const remaining = message
private buildBoundaryFailure(
error: unknown,
block: SerializedBlock,
instanceId: string,
childExecutionId: string | undefined
): ChildWorkflowError {
const blockName = block.metadata?.name || 'Custom block'
// 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 }
// 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,
})
}
// 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
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'
match = workflowPattern.exec(remaining)
while (match !== null) {
chain.push(match[1])
lastIndex = match.index + match[0].length
match = workflowPattern.exec(remaining)
return new ChildWorkflowError({
message,
childWorkflowName: blockName,
childWorkflowInstanceId: instanceId,
consumerFacing: { errorType, ...(ref ? { ref } : {}), message },
})
}
/**
* 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 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,
}
}
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<string, unknown> = {}
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,
+139
View File
@@ -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<string, unknown>
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)
})
})
+50 -36
View File
@@ -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<string, any>
}
/**
* 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<unknown>()
let current = value
for (let depth = 0; depth < MAX_CAUSE_DEPTH; depth++) {
if (!(current instanceof Error) || seen.has(current)) return undefined
seen.add(current)
if (current instanceof HttpError) return current.statusCode
const candidate = current as unknown as { statusCode?: unknown }
if (typeof candidate.statusCode === 'number') return candidate.statusCode
current = current.cause
}
if (config.status) {
errorMessage += ` (Status: ${config.status})`
}
const error = new Error(errorMessage)
Object.assign(error, {
status: config.status,
url: config.url,
method: config.method,
timestamp: new Date().toISOString(),
})
return error
return undefined
}
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
}
/**
* 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
}
+17 -3
View File
@@ -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<typeof publishCustomBlockBodySchema>
@@ -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' })
@@ -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()
})
})
@@ -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,
+6 -1
View File
@@ -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<TPayload = unknown, TOutput = unknown> {
+5 -13
View File
@@ -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')
@@ -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)
})
})
+20 -5
View File
@@ -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,
@@ -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',
+23 -16
View File
@@ -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<void> | null = null
private completionAttempt: CompletionAttempt | null = null
private completionAttemptFailed = false
private costOptions?: CostSummaryOptions
private pendingProgressWrites = new Set<Promise<void>>()
private postExecutionPromise: Promise<void> | 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 }
@@ -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>): 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<TraceSpan>)
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)
})
})
@@ -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
}
+1
View File
@@ -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) {
@@ -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<void>((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<void>(() => {}))
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()
})
})
@@ -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<void> {
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<string, Set<Promise<unknown>>>()
/** 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<unknown>
): 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<void> {
if (!invokerExecutionId) return
const pending = pendingChildRuns.get(invokerExecutionId)
if (!pending || pending.size === 0) return
let timer: ReturnType<typeof setTimeout> | 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)
}
}
@@ -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<CustomBlockWithInputs> {
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<void> {
if (updates.exposedOutputs !== undefined) assertNoReservedOutputNames(updates.exposedOutputs)
if (updates.exposedOutputs !== undefined) {
assertNoReservedOutputNames(updates.exposedOutputs)
assertCuratedOutputs(updates.exposedOutputs)
}
const patch: Partial<typeof customBlock.$inferInsert> = { updatedAt: new Date() }
if (updates.name !== undefined) patch.name = updates.name
if (updates.description !== undefined) patch.description = updates.description
@@ -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 {
+7
View File
@@ -213,6 +213,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
+37
View File
@@ -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()
})
})
+3
View File
@@ -1475,6 +1475,8 @@ export function prepareToolExecution(
isDeployedContext?: boolean
callChain?: string[]
billingAttribution?: BillingAttributionSnapshot
/** Invoking run's execution id — see `ProviderRequest.executionId`. */
executionId?: string
}
): {
toolParams: Record<string, any>
@@ -1512,6 +1514,7 @@ export function prepareToolExecution(
? { isDeployedContext: request.isDeployedContext }
: {}),
...(request.callChain ? { callChain: request.callChain } : {}),
...(request.executionId ? { executionId: request.executionId } : {}),
...(request.billingAttribution
? { billingAttribution: request.billingAttribution }
: {}),
+1
View File
@@ -31,6 +31,7 @@ export const CORE_TRIGGER_TYPES = [
'copilot',
'mothership',
'workflow',
'custom_block',
] as const
export type CoreTriggerType = (typeof CORE_TRIGGER_TYPES)[number]
+28
View File
@@ -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'
}
}
+27 -8
View File
@@ -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<string, unknown> | 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,
+7
View File
@@ -93,6 +93,13 @@ export interface ToolResponse {
success: boolean // Whether the tool execution was successful
output: Record<string, any> // 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[]
+29 -1
View File
@@ -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()
})
})
+22
View File
@@ -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<T>(
error: unknown,
predicate: (value: unknown) => value is T
): T | undefined {
const seen = new Set<unknown>()
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<unknown>()
let current: unknown = error