payload: WorkflowExecutionJobPayload

): AsyncExecutionCorrelation {

const executionId = payload.executionId || generateId()

const requestId = payload.requestId || payload.correlation?.requestId || executionId.slice(0, 8)

return {

executionId,

requestId,

source: 'workflow',

workflowId: payload.workflowId,

...(payload.correlation?.copilotToolCallId

? { copilotToolCallId: payload.correlation.copilotToolCallId }

: {}),

triggerType: payload.triggerType || payload.correlation?.triggerType || 'api',

}

workflowId: string

principal: SerializedPrincipalV1

userId: string

billingAttribution: BillingAttributionSnapshot

workspaceId: string

input?: any

triggerType?: CoreTriggerType

triggerBlockId?: string

executionId?: string

requestId?: string

correlation?: AsyncExecutionCorrelation

metadata?: Record<string, any>

callChain?: string[]

/** Who queued the run, restored into the job's context so its events stay attributed. */

attribution?: RequestAttribution

executionMode?: 'sync' | 'stream' | 'async'

/** Upstream preprocessing already consumed rate-limit quota and owns the usage reservation. */

admissionCompleted?: boolean

/** Optional trusted cap already resolved by the async API admission boundary. */

executionTimeoutMs?: number

/** Authenticated input provenance validated by the workflow execution boundary. */

trustedInitialResolvedSecretTraceProvenance?: ResolvedSecretTraceProvenanceV1

/**

enforceCredentialAccess?: boolean

isPublicApiAccess?: boolean

rejection: PreprocessExecutionError,

rejectionLogged: boolean

): boolean {

if (rejection.retryable === true) return false

if (isWorkflowStateRejection(rejection)) return true

const quotaRejection =

getDeterministicAdmissionRejectionCode(rejection) !== undefined ||

isReservationHeadroomDenial(rejection) ||

rejection.code === RATE_LIMIT_EXCEEDED_CODE

return quotaRejection && rejectionLogged

const failure = toError(error)

trace.getActiveSpan()?.recordException(failure)

return buildRecordedJobFailureResult({ workflowId, executionId, error: failure.message })

if (typeof value !== 'string' || !value.trim()) {

throw new Error(`Legacy workflow job ${field} must be a non-empty string`)

}

return value

payload: WorkflowExecutionJobPayload

): WorkflowExecutionPrincipal {

if (payload.principal !== undefined) return parsePrincipal(payload.principal)

const workflowId = requireLegacyWorkflowJobString(payload.workflowId, 'workflowId')

const workspaceId = requireLegacyWorkflowJobString(payload.workspaceId, 'workspaceId')

if (payload.enforceCredentialAccess === true) {

return {

kind: 'session',

userId: requireLegacyWorkflowJobString(payload.userId, 'userId'),

sessionId: LEGACY_WORKFLOW_JOB_SESSION_ID,

}

}

return { kind: 'system', serviceId: 'internal', workspaceId, workflowId }

payload: WorkflowExecutionJobPayload,

externalAbortSignal?: AbortSignal

) {

const workflowId = payload.workflowId

const correlation = buildWorkflowCorrelation(payload)

const executionId = correlation.executionId

const requestId = correlation.requestId

let principal

let billingAttribution: BillingAttributionSnapshot

try {

principal = parseWorkflowJobPrincipal(payload)

billingAttribution = assertBillingAttributionSnapshot(payload.billingAttribution)

if (

billingAttribution.actorUserId !== payload.userId ||

billingAttribution.workspaceId !== payload.workspaceId

) {

throw new Error('Workflow job billing attribution does not match its actor and workspace')

}

} catch (error) {

await releaseExecutionSlot(executionId)

throw error

}

const timeoutController = createTimeoutAbortController(

capExecutionTimeoutMs(

getAsyncExecutionTimeoutForBillingAttribution(billingAttribution),

payload.executionTimeoutMs

),

externalAbortSignal

)

try {

const executionDeadlineAt = getExecutionDeadlineAt(timeoutController.signal)?.getTime()

let admissionCompleted = payload.admissionCompleted === true

if (admissionCompleted && executionDeadlineAt !== undefined) {

admissionCompleted = await refreshExecutionSlotExpiry(

executionId,

executionDeadlineAt + RESERVATION_TTL_BUFFER_MS

)

if (!admissionCompleted) {

logger.warn('Queued workflow reservation expired; repeating usage admission', {

workflowId,

executionId,

})

}

}

return await runWithRequestContext({ requestId, ...payload.attribution }, async () => {

logger.info(`[${requestId}] Starting workflow execution job: ${workflowId}`, {

userId: payload.userId,

triggerType: payload.triggerType,

executionId,

})

const triggerType = (correlation.triggerType || 'api') as CoreTriggerType

const loggingSession = new LoggingSession(workflowId, executionId, triggerType, requestId)

if (correlation.copilotToolCallId) {

loggingSession.setTrustedExecutionCorrelation(correlation)

}

loggingSession.setExecutionDeadlineAt(getExecutionDeadlineAt(timeoutController.signal))

let preprocessingRejection: PreprocessExecutionError | undefined

try {

const preprocessResult = await preprocessExecution({

workflowId: payload.workflowId,

userId: payload.userId,

triggerType: triggerType,

executionId: executionId,

requestId: requestId,

checkRateLimit: !admissionCompleted,

checkDeployment: true,

skipUsageLimits: admissionCompleted,

loggingSession: loggingSession,

triggerData: { correlation },

billingAttribution,

executionType: 'async',

executionDeadlineAt,

})

if (!preprocessResult.success) {

logger.error(`[${requestId}] Preprocessing failed: ${preprocessResult.error?.message}`, {

workflowId,

statusCode: preprocessResult.error?.statusCode,

})

preprocessingRejection = preprocessResult.error

throw new Error(preprocessResult.error?.message || 'Preprocessing failed')

}

const actorUserId = preprocessResult.actorUserId!

const workspaceId = preprocessResult.workflowRecord?.workspaceId

if (!workspaceId) {

throw new Error(`Workflow ${workflowId} has no associated workspace`)

}

logger.info(`[${requestId}] Preprocessing passed. Using actor: ${actorUserId}`)

const workflow = preprocessResult.workflowRecord!

const metadata: ExecutionMetadata = {

requestId,

executionId,

workflowId,

workspaceId,

userId: actorUserId,

principal,

billingAttribution: preprocessResult.billingAttribution,

sessionUserId: undefined,

workflowUserId: workflow.userId,

triggerType: payload.triggerType || 'api',

triggerBlockId: payload.triggerBlockId,

useDraftState: false,

startTime: new Date().toISOString(),

isClientSession: false,

enforceCredentialAccess: payload.enforceCredentialAccess ?? false,

isPublicApiAccess: payload.isPublicApiAccess ?? false,

callChain: payload.callChain,

correlation,

executionMode: payload.executionMode ?? 'async',

}

const snapshot = new ExecutionSnapshot(

metadata,

workflow,

payload.input,

workflow.variables || {},

[]

)

const result = await executeWorkflowCore({

snapshot,

callbacks: {},

loggingSession,

includeFileBase64: true,

base64MaxBytes: undefined,

abortSignal: timeoutController.signal,

trustedInitialResolvedSecretTraceProvenance:

payload.trustedInitialResolvedSecretTraceProvenance,

})

let timeoutError: ExecutionTimeoutError | undefined

if (

result.status === 'cancelled' &&

timeoutController.isTimedOut() &&

timeoutController.timeoutMs

) {

const timeoutErrorMessage = getTimeoutErrorMessage(timeoutController.timeoutMs)

logger.info(`[${requestId}] Workflow execution timed out`, {

timeoutMs: timeoutController.timeoutMs,

})

await loggingSession.markAsFailed(timeoutErrorMessage)

timeoutError = new ExecutionTimeoutError(timeoutErrorMessage)

} else {

await handlePostExecutionPauseState({ result, workflowId, executionId, loggingSession })

}

await loggingSession.waitForPostExecution()

if (timeoutError) throw timeoutError

logger.info(`[${requestId}] Workflow execution completed: ${workflowId}`, {

success: result.success,

executionTime: result.metadata?.duration,

executionId,

})

return {

success: result.success,

workflowId: payload.workflowId,

executionId,

output: result.output,

executedAt: new Date().toISOString(),

metadata: payload.metadata,

}

} catch (error: unknown) {

logFailureOnce(logger, `[${requestId}] Workflow execution failed: ${workflowId}`, error, {

metadata: () => loggingSession.projectDiagnosticError(error),

executionId,

})

const completeWithDeletedScope = (scope: ExecutionScope) => {

logger.info(`[${requestId}] Workflow execution ended: its ${scope} was deleted`, {

workflowId,

executionId,

})

return completeWithRecordedFailure(workflowId, executionId, error)

}

// Checked before the timeout: a deleted scope leaves nobody to alert about a slow run.

const deletedScope = await resolveDeletedExecutionScope(error, loggingSession)

if (deletedScope) return completeWithDeletedScope(deletedScope)

if (error instanceof ExecutionTimeoutError) throw error

// A failure core already recorded is the workflow's outcome, not a job fault; a run past

// its deadline (checked after the await, which can outlast it) still fails the job.

if (await isRecordedWorkflowFailure(error, executionId, loggingSession)) {

if (timeoutController.isTimedOut()) throw error

return completeWithRecordedFailure(workflowId, executionId, error)

}

const executionResult = hasExecutionResult(error) ? error.executionResult : undefined

const { traceSpans } = executionResult

? buildTraceSpans(executionResult)

: { traceSpans: [] }

await loggingSession.safeCompleteWithError({

error: {

message: toError(error).message,

stackTrace: error instanceof Error ? error.stack : undefined,

},

traceSpans,

executionState: executionResult?.executionState,

})

const scopeDeletedOnCompletion = loggingSession.getDeletedScope()

if (scopeDeletedOnCompletion) return completeWithDeletedScope(scopeDeletedOnCompletion)

if (

preprocessingRejection &&

isNonFaultPreprocessingRejection(preprocessingRejection, loggingSession.hasCompleted())

) {

return completeWithRecordedFailure(workflowId, executionId, error)

}

throw error

} finally {

void cleanupExecutionBase64Cache(executionId)

}

})

} finally {

timeoutController.cleanup()

}