Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 64 additions & 0 deletions apps/sim/executor/execution/block-executor.retry.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,11 +5,13 @@
* client has already seen and cannot re-run the deterministic post-processing.
*/
import { beforeEach, describe, expect, it, vi } from 'vitest'
import { buildTraceSpans } from '@/lib/logs/execution/trace-spans/trace-spans'
import { BlockType, EDGE } from '@/executor/constants'
import type { DAGNode } from '@/executor/dag/builder'
import { BlockExecutor } from '@/executor/execution/block-executor'
import { ExecutionState } from '@/executor/execution/state'
import type { BlockHandler, ExecutionContext } from '@/executor/types'
import { attachTrustedExecutionCost } from '@/executor/utils/errors'
import { VariableResolver } from '@/executor/variables/resolver'
import type { SerializedBlock, SerializedWorkflow } from '@/serializer/types'

Expand Down Expand Up @@ -137,6 +139,68 @@ describe('BlockExecutor retry', () => {
expect(ctx.blockLogs[0]?.tries).toBe(2)
})

it('adds the trusted cost of failed Function tries to the successful result', async () => {
const block = createBlock(enabled)
const firstFailure = new Error('first attempt failed')
attachTrustedExecutionCost(firstFailure, { input: 0, output: 0, total: 0.125 })
const successfulOutput = {
result: 'done',
cost: { input: 0, output: 0, total: 0.25 },
}
attachTrustedExecutionCost(successfulOutput, successfulOutput.cost)
const execute = vi
.fn()
.mockRejectedValueOnce(firstFailure)
.mockResolvedValueOnce(successfulOutput)
const state = new ExecutionState()
const ctx = createContext(state)
const executor = buildExecutor(block, { canHandle: () => true, execute }, state)

const output = await executor.execute(ctx, createNode(block), block)

expect(execute).toHaveBeenCalledTimes(2)
expect(output.cost).toEqual({ input: 0, output: 0, total: 0.375 })
expect(ctx.blockLogs[0]?.output?.cost).toEqual(output.cost)
})

it('keeps earlier trusted Function costs when the final try is an infrastructure error', async () => {
const block = createBlock(enabled)
const firstFailure = new Error('first Function attempt failed')
const secondFailure = new Error('second Function attempt failed')
const finalFailure = new Error('provider unavailable')
attachTrustedExecutionCost(firstFailure, { input: 0, output: 0, total: 0.125 })
attachTrustedExecutionCost(secondFailure, { input: 0, output: 0, total: 0.25 })
const execute = vi
.fn()
.mockRejectedValueOnce(firstFailure)
.mockRejectedValueOnce(secondFailure)
.mockRejectedValueOnce(finalFailure)
const state = new ExecutionState()
const ctx = createContext(state)
const executor = buildExecutor(block, { canHandle: () => true, execute }, state)

await expect(executor.execute(ctx, createNode(block), block)).rejects.toThrow(
'provider unavailable'
)

expect(execute).toHaveBeenCalledTimes(3)
expect(ctx.blockLogs[0]?.output).toEqual({
error: 'provider unavailable',
cost: { input: 0, output: 0, total: 0.375 },
})

const { traceSpans } = buildTraceSpans({
success: false,
output: { error: 'provider unavailable' },
error: 'provider unavailable',
logs: ctx.blockLogs,
})
expect(traceSpans[0]).toMatchObject({
status: 'error',
cost: { input: 0, output: 0, total: 0.375 },
})
})

it('stops at maxTries and rethrows the final error unchanged', async () => {
const block = createBlock(enabled)
const failure = new Error('still failing')
Expand Down
77 changes: 71 additions & 6 deletions apps/sim/executor/execution/block-executor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,13 @@ import {
type StreamingExecution,
} from '@/executor/types'
import { streamingResponseFormatProcessor } from '@/executor/utils'
import { buildBlockExecutionError, normalizeError } from '@/executor/utils/errors'
import {
attachTrustedExecutionCost,
buildBlockExecutionError,
normalizeError,
readTrustedExecutionCost,
type TrustedExecutionCost,
} from '@/executor/utils/errors'
import {
buildUnifiedParentIterations,
getIterationContext,
Expand Down Expand Up @@ -76,6 +82,20 @@ import { SYSTEM_SUBBLOCK_IDS } from '@/triggers/constants'

const logger = createLogger('BlockExecutor')

function addTrustedExecutionCosts(
accumulated: TrustedExecutionCost | undefined,
current: TrustedExecutionCost | undefined
): TrustedExecutionCost | undefined {
if (!accumulated) return current
if (!current) return accumulated

return {
input: accumulated.input + current.input,
output: accumulated.output + current.output,
total: accumulated.total + current.total,
}
}

export class BlockExecutor {
private execLogger: Logger

Expand Down Expand Up @@ -229,6 +249,17 @@ export class BlockExecutor {
cleanupSelfReference?.()

let streamingPartialOutput: Record<string, any> | undefined
/**
* Cost of a handler that already finished, kept for the catch below.
*
* A Function block's sandbox is paid for the moment it completes, but the
* steps after the handler returns — base64 hydration, and large-value
* redaction that deliberately throws rather than emit unredacted data — can
* still fail the block. The error those raise carries no cost of its own, so
* without holding it here the completed sandbox would go unbilled. Hoisted
* for the same reason `streamingPartialOutput` above is.
*/
let completedHandlerCost: TrustedExecutionCost | undefined
try {
/**
* Only the handler call is retried. A streaming handler returns before any
Expand All @@ -241,6 +272,8 @@ export class BlockExecutor {
: handler.execute(blockCtx, block, resolvedInputs, nodeMetadata)
)

completedHandlerCost = readTrustedExecutionCost(output)

const isStreamingExecution =
output && typeof output === 'object' && 'stream' in output && 'execution' in output

Expand Down Expand Up @@ -416,7 +449,8 @@ export class BlockExecutor {
inputDisplayRegistry,
isSentinel,
'execution',
streamingPartialOutput
streamingPartialOutput,
completedHandlerCost
)
} finally {
commitBlockRegistry()
Expand Down Expand Up @@ -506,15 +540,40 @@ export class BlockExecutor {
const policy = resolveBlockRetryPolicy(block)
if (!policy) return invoke()

const shouldAccumulateFunctionCost = block.metadata?.id === BlockType.FUNCTION
let accumulatedFunctionCost: TrustedExecutionCost | undefined
let tries = 0
try {
for (;;) {
tries++
try {
return await invoke()
const output = await invoke()
if (!shouldAccumulateFunctionCost || !accumulatedFunctionCost || !isRecordLike(output)) {
return output
}

const totalCost = addTrustedExecutionCosts(
accumulatedFunctionCost,
readTrustedExecutionCost(output)
)
if (!totalCost) return output

const outputWithCost = { ...output, cost: totalCost }
attachTrustedExecutionCost(outputWithCost, totalCost)
return outputWithCost as T
} catch (error) {
if (shouldAccumulateFunctionCost) {
accumulatedFunctionCost = addTrustedExecutionCosts(
accumulatedFunctionCost,
readTrustedExecutionCost(error)
)
}

const isFinalTry = tries >= policy.maxTries
if (isFinalTry || ctx.abortSignal?.aborted || !isRetryableBlockError(error)) throw error
if (isFinalTry || ctx.abortSignal?.aborted || !isRetryableBlockError(error)) {
attachTrustedExecutionCost(error, accumulatedFunctionCost)
throw error
}

this.execLogger.warn('Block failed; retrying', {
blockId: block.id,
Expand All @@ -528,7 +587,10 @@ export class BlockExecutor {
if (policy.waitBetweenTriesMs > 0) await sleep(policy.waitBetweenTriesMs)

/** `sleep` is not abort-aware, so a run stopped mid-wait must not start another try. */
if (ctx.abortSignal?.aborted) throw error
if (ctx.abortSignal?.aborted) {
attachTrustedExecutionCost(error, accumulatedFunctionCost)
throw error
}
}
}
} finally {
Expand All @@ -548,7 +610,8 @@ export class BlockExecutor {
inputDisplayRegistry: ResolvedSecretTraceRegistry | undefined,
isSentinel: boolean,
phase: 'input_resolution' | 'execution',
streamingPartialOutput?: Record<string, any>
streamingPartialOutput?: Record<string, any>,
completedHandlerCost?: TrustedExecutionCost
): Promise<NormalizedBlockOutput> {
const endedAt = new Date().toISOString()
const duration = performance.now() - startTime
Expand Down Expand Up @@ -620,8 +683,10 @@ export class BlockExecutor {
return softOutput
}

const trustedExecutionCost = readTrustedExecutionCost(error) ?? completedHandlerCost
const errorOutput: NormalizedBlockOutput = {
error: errorMessage,
...(trustedExecutionCost ? { cost: trustedExecutionCost } : {}),
}

// Keep any answer text already drained before timeout/failure so logs match
Expand Down
41 changes: 41 additions & 0 deletions apps/sim/executor/handlers/function/function-handler.test.ts
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import { beforeEach, describe, expect, it, type Mock, vi } from 'vitest'
import { createTimeoutAbortController } from '@/lib/core/execution-limits'
import { DEFAULT_EXECUTION_TIMEOUT_MS } from '@/lib/execution/constants'
import { NonRetryableExecutionError } from '@/lib/execution/non-retryable-error'
import { BlockType } from '@/executor/constants'
import { FunctionBlockHandler } from '@/executor/handlers/function/function-handler'
import type { ExecutionContext } from '@/executor/types'
import { readTrustedExecutionCost } from '@/executor/utils/errors'
import {
FUNCTION_BLOCK_CONTEXT_VARS_KEY,
FUNCTION_BLOCK_DISPLAY_CODE_KEY,
Expand Down Expand Up @@ -254,6 +256,45 @@ describe('FunctionBlockHandler', () => {
expect(mockExecuteTool).toHaveBeenCalled()
})

it.each([
{ retryable: true, nonRetryable: false },
{ retryable: false, nonRetryable: true },
])(
'attaches trusted cost to a failed execution when retryable is $retryable',
async ({ retryable, nonRetryable }) => {
const cost = { input: 0, output: 0, total: 0.125 }
mockExecuteTool.mockResolvedValue({
success: false,
error: 'Remote Function failed',
retryable,
output: { result: null, stdout: '', cost },
})

let thrown: unknown
try {
await handler.execute(mockContext, mockBlock, { code: 'throw new Error("failed")' })
} catch (error) {
thrown = error
}

expect(thrown).toBeInstanceOf(Error)
expect(thrown instanceof NonRetryableExecutionError).toBe(nonRetryable)
expect(readTrustedExecutionCost(thrown)).toEqual(cost)
}
)

it('attaches trusted cost to a successful execution for retry aggregation', async () => {
const cost = { input: 0, output: 0, total: 0.25 }
mockExecuteTool.mockResolvedValue({
success: true,
output: { result: 42, stdout: '', cost },
})

const output = await handler.execute(mockContext, mockBlock, { code: 'return 42' })

expect(readTrustedExecutionCost(output)).toEqual(cost)
})

it('should pass runtime context variables to function_execute', async () => {
const contextVariables = { __blockRef_0: { result: 'from-block' } }

Expand Down
12 changes: 8 additions & 4 deletions apps/sim/executor/handlers/function/function-handler.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@ import { mergeFileKeys, mergeLargeValueKeys } from '@/lib/execution/payloads/acc
import { BlockType } from '@/executor/constants'
import type { BlockHandler, ExecutionContext } from '@/executor/types'
import { collectBlockData } from '@/executor/utils/block-data'
import { attachTrustedExecutionCost } from '@/executor/utils/errors'
import {
FUNCTION_BLOCK_CONTEXT_VARS_KEY,
FUNCTION_BLOCK_DISPLAY_CODE_KEY,
Expand Down Expand Up @@ -111,15 +112,18 @@ export class FunctionBlockHandler implements BlockHandler {
const result = await executeTool('function_execute', toolParams, { executionContext: ctx })

if (!result.success) {
if (result.retryable === false) {
throw new NonRetryableExecutionError(result.error || 'Function execution is indeterminate')
}
throw new Error(result.error || 'Function execution failed')
const error =
result.retryable === false
? new NonRetryableExecutionError(result.error || 'Function execution is indeterminate')
: new Error(result.error || 'Function execution failed')
attachTrustedExecutionCost(error, result.output?.cost)
throw error
}

mergeLargeValueKeys(ctx, result.largeValueKeys ?? [])
mergeFileKeys(ctx, result.fileKeys ?? [])

attachTrustedExecutionCost(result.output, result.output?.cost)
return result.output
}
}
5 changes: 4 additions & 1 deletion apps/sim/executor/handlers/pi/cloud/authoring/backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -435,7 +435,10 @@ async function runCloudAuthoringPi(
const lifetimeMs = resolvePiRunLifetimeMs(context.signal)
const piTimeoutMs = resolvePiTimeoutMs(lifetimeMs)

const authored = await withPiSandbox<AuthoringPhaseResult>({ lifetimeMs }, async (runner) => {
// Bound to a local so the call stays on one line: inlining the second option
// reflows this whole callback body and buries the change in re-indentation.
const sandboxOptions = { lifetimeMs, cost: context.sandboxCost }
const authored = await withPiSandbox<AuthoringPhaseResult>(sandboxOptions, async (runner) => {
try {
const clone = await raceAbort(
runner.run(params.mode === 'cloud' ? CREATE_PR_CLONE_SCRIPT : UPDATE_BRANCH_CLONE_SCRIPT, {
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/executor/handlers/pi/cloud/babysit/backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -784,7 +784,7 @@ export async function runBabysitPiWithOptions(
const lifetimeMs = resolvePiRunLifetimeMs(context.signal)
const piTimeoutMs = resolvePiTimeoutMs(lifetimeMs)

return await withPiSandbox({ lifetimeMs }, async (runner) => {
return await withPiSandbox({ lifetimeMs, cost: context.sandboxCost }, async (runner) => {
const clone = await raceAbort(
runner.run(BABYSIT_CLONE_SCRIPT, {
envs: {
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/executor/handlers/pi/cloud/plan/backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ export const runCloudPlanPi: PiBackendRun<PiCloudPlanRunParams> = async (params,
const thinking = mapThinkingLevel(params.thinkingLevel) ?? 'medium'
const lifetimeMs = resolvePiRunLifetimeMs(context.signal)

return withPiSandbox({ lifetimeMs }, async (runner) => {
return withPiSandbox({ lifetimeMs, cost: context.sandboxCost }, async (runner) => {
try {
const clone = await raceAbort(
runner.run(PLAN_CLONE_SCRIPT, {
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/executor/handlers/pi/cloud/review/backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -218,7 +218,7 @@ export const runCloudReviewPi: PiBackendRun<PiCloudReviewRunParams> = async (par
const lifetimeMs = resolvePiRunLifetimeMs(context.signal)

try {
return await withPiSandbox({ lifetimeMs }, async (runner) => {
return await withPiSandbox({ lifetimeMs, cost: context.sandboxCost }, async (runner) => {
await runner.writeFile(GIT_ASKPASS_PATH, GIT_ASKPASS_SCRIPT)
const fetched = await raceAbort(
runner.run(FETCH_PR_SCRIPT, {
Expand Down
15 changes: 15 additions & 0 deletions apps/sim/executor/handlers/pi/core/backend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
*/

import type { TSchema } from 'typebox'
import type { SandboxCostSink } from '@/lib/execution/remote-sandbox/types'
import type { SSHConnectionConfig } from '@/lib/internal/ssh/client'
import type { Message } from '@/executor/handlers/agent/types'
import type { PiEvent, PiRunTotals } from '@/executor/handlers/pi/core/events'
Expand Down Expand Up @@ -172,6 +173,20 @@ export type PiRunParams =
export interface PiRunContext {
onEvent: (event: PiEvent) => void
signal?: AbortSignal
/**
* Where a backend reports the cost of Sim-provisioned compute it used.
*
* Both modes can fill it, from different sources. Cloud modes run the agent in
* a Sim-paid sandbox and report that session. Local mode drives the caller's
* own machine over SSH, so the agent itself costs Sim nothing — but the Sim
* tools it calls still run here, and a `function_execute` among them bills its
* own remote sandbox into the same total.
*
* The handler folds whatever lands here into the block's `toolCost`, which is
* what keeps a BYOK Pi run — model unbilled by definition — from reporting no
* cost at all for compute Sim actually paid for.
*/
sandboxCost?: SandboxCostSink
}

/** Final result of a Pi run. */
Expand Down
Loading
Loading