import { ControlPlaneBadRequestError, ControlPlaneNotFoundError, ControlPlaneStorageError,} from "../api/errors";import type { ElicitationResponse, OnElicitation, ToolInvocationContext, ToolInvoker,} from "@executor/codemode-core";import type { CreateExecutionPayload, ResumeExecutionPayload,} from "../api/executions/api";import { ExecutionIdSchema, ExecutionInteractionIdSchema, ExecutionStepIdSchema, type AccountId, type Execution, type ExecutionEnvelope, type ExecutionId, type ExecutionInteraction, type WorkspaceId,} from "#schema";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Schema from "effect/Schema"; import { type ResolveExecutionEnvironment } from "./execution-state";import { LiveExecutionManagerService, sanitizePersistedElicitationResponse, type LiveExecutionManager,} from "./live-execution";import { getRuntimeLocalWorkspaceOption, provideOptionalRuntimeLocalWorkspace,} from "./local-runtime-context";import { asOperationErrors, operationErrors, type OperationErrorsLike,} from "./operation-errors";import { ControlPlaneStore, type ControlPlaneStoreShape,} from "./store";import { RuntimeExecutionResolverService } from "./workspace-execution-environment"; const executionOps = { create: operationErrors("executions.create"), get: operationErrors("executions.get"), resume: operationErrors("executions.resume"),} as const; type InteractionMode = NonNullable<CreateExecutionPayload["interactionMode"]>; const EXECUTION_SUSPENDED_SENTINEL = "__EXECUTION_SUSPENDED__"; const DEFAULT_INTERACTION_MODE: InteractionMode = "detach"; const serializeJson = (value: unknown): string | null => { if (value === undefined) { return null; } return JSON.stringify(value);}; const serializeRequiredJson = (value: unknown): string => JSON.stringify(value === undefined ? null : value); const parseStoredJson = (value: string | null): unknown => { if (value === null) { return undefined; } return JSON.parse(value);}; const ElicitationActionSchema = Schema.Literal("accept", "decline", "cancel"); const ElicitationResponseSchema = Schema.Struct({ action: ElicitationActionSchema, content: Schema.optional( Schema.Record({ key: Schema.String, value: Schema.Unknown, }), ),}); const decodeElicitationResponse = Schema.decodeUnknown(ElicitationResponseSchema); const withExecutionInvocationContext = (input: { executionId: Execution["id"]; toolInvoker: ToolInvoker;}): ToolInvoker => { let sequence = 0; return { invoke: ({ path, args, context }) => { sequence += 1; return input.toolInvoker.invoke({ path, args, context: { ...context, runId: input.executionId, callId: typeof context?.callId === "string" && context.callId.length > 0 ? context.callId : `call_${String(sequence)}`, executionStepSequence: sequence, }, }); }, };}; const executionStepSequenceFromContext = ( context: ToolInvocationContext | undefined,): number | null => { const value = context?.executionStepSequence; return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null;}; const executionStepIdFor = ( executionId: Execution["id"], sequence: number,) => ExecutionStepIdSchema.make(`${executionId}:step:${String(sequence)}`); const interactionIdSuffixForRequest = (input: { interactionId: string; path: string; context?: ToolInvocationContext;}) => { const callId = typeof input.context?.callId === "string" && input.context.callId.length > 0 ? input.context.callId : null; return callId === null ? input.interactionId : `${callId}:${input.path}`;}; const executionInteractionIdForRequest = (input: { executionId: Execution["id"]; interactionId: string; path: string; context?: ToolInvocationContext;}) => { return ExecutionInteractionIdSchema.make( `${input.executionId}:${interactionIdSuffixForRequest(input)}`, );}; const resolveInteractionMode = ( value: CreateExecutionPayload["interactionMode"] | ResumeExecutionPayload["interactionMode"],): InteractionMode => value === "live" || value === "live_form" ? value : DEFAULT_INTERACTION_MODE; const createExecutionSuspendedError = (input: { executionId: Execution["id"]; interactionId: string;}): Error => new Error( `${EXECUTION_SUSPENDED_SENTINEL}:${input.executionId}:${input.interactionId}`, ); const isExecutionSuspendedValue = (value: unknown): boolean => { if (value instanceof Error) { return value.message.includes(EXECUTION_SUSPENDED_SENTINEL); } return typeof value === "string" && value.includes(EXECUTION_SUSPENDED_SENTINEL);}; const decodeStoredElicitationResponse = (input: { interactionId: string; responseJson: string | null;}) => Effect.try({ try: () => { if (input.responseJson === null) { throw new Error( `Interaction ${input.interactionId} has no stored response`, ); } return JSON.parse(input.responseJson); }, catch: (error) => error instanceof Error ? error : new Error(String(error)), }).pipe( Effect.flatMap((decoded) => decodeElicitationResponse(decoded).pipe( Effect.mapError((error) => new Error(String(error))), ) ), ); const verifyStoredStepMatches = (input: { executionId: Execution["id"]; sequence: number; expectedPath: string; expectedArgsJson: string; actualPath: string; actualArgsJson: string;}) => { if ( input.expectedPath === input.actualPath && input.expectedArgsJson === input.actualArgsJson ) { return; } throw new Error( [ `Durable execution mismatch for ${input.executionId} at tool step ${String(input.sequence)}.`, `Expected ${input.expectedPath}(${input.expectedArgsJson}) but replay reached ${input.actualPath}(${input.actualArgsJson}).`, ].join(" "), );}; const fetchExecution = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; },): Effect.Effect<Execution, ControlPlaneNotFoundError | ControlPlaneStorageError> => Effect.gen(function* () { const errors = asOperationErrors(input.operation); const existing = yield* errors.mapStorage( store.executions.getByWorkspaceAndId(input.workspaceId, input.executionId), ); if (Option.isNone(existing)) { return yield* Effect.fail( errors.notFound( "Execution not found", `workspaceId=${input.workspaceId} executionId=${input.executionId}`, ), ); } return existing.value; }); const fetchExecutionEnvelope = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; },): Effect.Effect<ExecutionEnvelope, ControlPlaneNotFoundError | ControlPlaneStorageError> => Effect.gen(function* () { const errors = asOperationErrors(input.operation); const execution = yield* fetchExecution(store, input); const pendingInteraction = yield* errors.child("pending_interaction").mapStorage( store.executionInteractions.getPendingByExecutionId(input.executionId), ); return { execution, pendingInteraction: Option.isSome(pendingInteraction) ? pendingInteraction.value : null, }; }); const waitForExecutionEnvelopeToSettle = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; previousPendingInteractionId: ExecutionInteraction["id"] | null; attemptsRemaining: number; },): Effect.Effect<ExecutionEnvelope, ControlPlaneNotFoundError | ControlPlaneStorageError, never> => Effect.gen(function* () { const envelope = yield* fetchExecutionEnvelope(store, input); if ( ( envelope.execution.status !== "running" && !( envelope.execution.status === "waiting_for_interaction" && ( envelope.pendingInteraction === null || envelope.pendingInteraction.id === input.previousPendingInteractionId ) ) ) || input.attemptsRemaining <= 0 ) { return envelope; } yield* Effect.promise( () => new Promise<void>((resolve) => setTimeout(resolve, 25)), ); return yield* waitForExecutionEnvelopeToSettle(store, { ...input, attemptsRemaining: input.attemptsRemaining - 1, }); }); const suspendExecutionForInteraction = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; liveExecutionManager: LiveExecutionManager; request: Parameters<OnElicitation>[0]; interactionId: ExecutionEnvelope["pendingInteraction"] extends infer T ? T extends { id: infer I } ? I : never : never;}) => Effect.gen(function* () { const now = Date.now(); const existing = yield* input.rows.executionInteractions.getById(input.interactionId); const stepSequence = executionStepSequenceFromContext(input.request.context); if (Option.isSome(existing) && existing.value.status !== "pending") { return yield* decodeStoredElicitationResponse({ interactionId: input.interactionId, responseJson: existing.value.responsePrivateJson ?? existing.value.responseJson, }); } if (Option.isNone(existing)) { yield* input.rows.executionInteractions.insert({ id: ExecutionInteractionIdSchema.make(input.interactionId), executionId: input.executionId, status: "pending", kind: input.request.elicitation.mode === "url" ? "url" : "form", purpose: "elicitation", payloadJson: serializeJson({ path: input.request.path, sourceKey: input.request.sourceKey, args: input.request.args, context: input.request.context, elicitation: input.request.elicitation, }) ?? "{}", responseJson: null, responsePrivateJson: null, createdAt: now, updatedAt: now, }); } if (stepSequence !== null) { yield* input.rows.executionSteps.updateByExecutionAndSequence( input.executionId, stepSequence, { status: "waiting", interactionId: ExecutionInteractionIdSchema.make(input.interactionId), updatedAt: now, }, ); } yield* input.rows.executions.update(input.executionId, { status: "waiting_for_interaction", updatedAt: now, }); yield* input.liveExecutionManager.publishState({ executionId: input.executionId, state: "waiting_for_interaction", }); return yield* Effect.fail( createExecutionSuspendedError({ executionId: input.executionId, interactionId: input.interactionId, }), ); }); const createHybridOnElicitation = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; liveExecutionManager: LiveExecutionManager; interactionMode: InteractionMode;}): OnElicitation => { const liveOnElicitation = input.liveExecutionManager.createOnElicitation({ rows: input.rows, executionId: input.executionId, }); return (request) => Effect.gen(function* () { const interactionIdSuffix = interactionIdSuffixForRequest({ interactionId: request.interactionId, path: request.path, context: request.context, }); const interactionId = executionInteractionIdForRequest({ executionId: input.executionId, interactionId: request.interactionId, path: request.path, context: request.context, }); const existing = yield* input.rows.executionInteractions.getById(interactionId); if (Option.isSome(existing) && existing.value.status !== "pending") { return yield* decodeStoredElicitationResponse({ interactionId, responseJson: existing.value.responsePrivateJson ?? existing.value.responseJson, }); } const allowLiveWait = input.interactionMode === "live" || (input.interactionMode === "live_form" && request.elicitation.mode !== "url"); if (Option.isNone(existing) && allowLiveWait) { const stepSequence = executionStepSequenceFromContext(request.context); if (stepSequence !== null) { yield* input.rows.executionSteps.updateByExecutionAndSequence( input.executionId, stepSequence, { status: "waiting", interactionId: ExecutionInteractionIdSchema.make(interactionId), updatedAt: Date.now(), }, ); } return yield* liveOnElicitation({ ...request, interactionId: interactionIdSuffix, }); } return yield* suspendExecutionForInteraction({ rows: input.rows, executionId: input.executionId, liveExecutionManager: input.liveExecutionManager, request, interactionId, }); });}; const createReplayToolInvoker = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; toolInvoker: ToolInvoker;}): ToolInvoker => ({ invoke: ({ path, args, context }) => Effect.gen(function* () { const stepSequence = executionStepSequenceFromContext(context); if (stepSequence === null) { return yield* input.toolInvoker.invoke({ path, args, context }); } const argsJson = serializeRequiredJson(args); const existing = yield* input.rows.executionSteps.getByExecutionAndSequence( input.executionId, stepSequence, ); if (Option.isSome(existing)) { verifyStoredStepMatches({ executionId: input.executionId, sequence: stepSequence, expectedPath: existing.value.path, expectedArgsJson: existing.value.argsJson, actualPath: path, actualArgsJson: argsJson, }); if (existing.value.status === "completed") { return parseStoredJson(existing.value.resultJson); } if (existing.value.status === "failed") { return yield* Effect.fail( new Error( existing.value.errorText ?? `Stored tool step ${String(stepSequence)} failed`, ), ); } } else { const now = Date.now(); yield* input.rows.executionSteps.insert({Lines 1–500 of 1008. Load more
import { ControlPlaneBadRequestError, ControlPlaneNotFoundError, ControlPlaneStorageError, } from "../api/errors"; import type { ElicitationResponse, OnElicitation, ToolInvocationContext, ToolInvoker, } from "@executor/codemode-core"; import type { CreateExecutionPayload, ResumeExecutionPayload, } from "../api/executions/api"; import { ExecutionIdSchema, ExecutionInteractionIdSchema, ExecutionStepIdSchema, type AccountId, type Execution, type ExecutionEnvelope, type ExecutionId, type ExecutionInteraction, type WorkspaceId, } from "#schema"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import { type ResolveExecutionEnvironment } from "./execution-state"; import { LiveExecutionManagerService, sanitizePersistedElicitationResponse, type LiveExecutionManager, } from "./live-execution"; import { getRuntimeLocalWorkspaceOption, provideOptionalRuntimeLocalWorkspace, } from "./local-runtime-context"; import { asOperationErrors, operationErrors, type OperationErrorsLike, } from "./operation-errors"; import { ControlPlaneStore, type ControlPlaneStoreShape, } from "./store"; import { RuntimeExecutionResolverService } from "./workspace-execution-environment"; const executionOps = { create: operationErrors("executions.create"), get: operationErrors("executions.get"), resume: operationErrors("executions.resume"), } as const; type InteractionMode = NonNullable<CreateExecutionPayload["interactionMode"]>; const EXECUTION_SUSPENDED_SENTINEL = "__EXECUTION_SUSPENDED__"; const DEFAULT_INTERACTION_MODE: InteractionMode = "detach"; const serializeJson = (value: unknown): string | null => { if (value === undefined) { return null; } return JSON.stringify(value); }; const serializeRequiredJson = (value: unknown): string => JSON.stringify(value === undefined ? null : value); const parseStoredJson = (value: string | null): unknown => { if (value === null) { return undefined; } return JSON.parse(value); }; const ElicitationActionSchema = Schema.Literal("accept", "decline", "cancel"); const ElicitationResponseSchema = Schema.Struct({ action: ElicitationActionSchema, content: Schema.optional( Schema.Record({ key: Schema.String, value: Schema.Unknown, }), ), }); const decodeElicitationResponse = Schema.decodeUnknown(ElicitationResponseSchema); const withExecutionInvocationContext = (input: { executionId: Execution["id"]; toolInvoker: ToolInvoker; }): ToolInvoker => { let sequence = 0; return { invoke: ({ path, args, context }) => { sequence += 1; return input.toolInvoker.invoke({ path, args, context: { ...context, runId: input.executionId, callId: typeof context?.callId === "string" && context.callId.length > 0 ? context.callId : `call_${String(sequence)}`, executionStepSequence: sequence, }, }); }, }; }; const executionStepSequenceFromContext = ( context: ToolInvocationContext | undefined, ): number | null => { const value = context?.executionStepSequence; return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : null; }; const executionStepIdFor = ( executionId: Execution["id"], sequence: number, ) => ExecutionStepIdSchema.make(`${executionId}:step:${String(sequence)}`); const interactionIdSuffixForRequest = (input: { interactionId: string; path: string; context?: ToolInvocationContext; }) => { const callId = typeof input.context?.callId === "string" && input.context.callId.length > 0 ? input.context.callId : null; return callId === null ? input.interactionId : `${callId}:${input.path}`; }; const executionInteractionIdForRequest = (input: { executionId: Execution["id"]; interactionId: string; path: string; context?: ToolInvocationContext; }) => { return ExecutionInteractionIdSchema.make( `${input.executionId}:${interactionIdSuffixForRequest(input)}`, ); }; const resolveInteractionMode = ( value: CreateExecutionPayload["interactionMode"] | ResumeExecutionPayload["interactionMode"], ): InteractionMode => value === "live" || value === "live_form" ? value : DEFAULT_INTERACTION_MODE; const createExecutionSuspendedError = (input: { executionId: Execution["id"]; interactionId: string; }): Error => new Error( `${EXECUTION_SUSPENDED_SENTINEL}:${input.executionId}:${input.interactionId}`, ); const isExecutionSuspendedValue = (value: unknown): boolean => { if (value instanceof Error) { return value.message.includes(EXECUTION_SUSPENDED_SENTINEL); } return typeof value === "string" && value.includes(EXECUTION_SUSPENDED_SENTINEL); }; const decodeStoredElicitationResponse = (input: { interactionId: string; responseJson: string | null; }) => Effect.try({ try: () => { if (input.responseJson === null) { throw new Error( `Interaction ${input.interactionId} has no stored response`, ); } return JSON.parse(input.responseJson); }, catch: (error) => error instanceof Error ? error : new Error(String(error)), }).pipe( Effect.flatMap((decoded) => decodeElicitationResponse(decoded).pipe( Effect.mapError((error) => new Error(String(error))), ) ), ); const verifyStoredStepMatches = (input: { executionId: Execution["id"]; sequence: number; expectedPath: string; expectedArgsJson: string; actualPath: string; actualArgsJson: string; }) => { if ( input.expectedPath === input.actualPath && input.expectedArgsJson === input.actualArgsJson ) { return; } throw new Error( [ `Durable execution mismatch for ${input.executionId} at tool step ${String(input.sequence)}.`, `Expected ${input.expectedPath}(${input.expectedArgsJson}) but replay reached ${input.actualPath}(${input.actualArgsJson}).`, ].join(" "), ); }; const fetchExecution = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; }, ): Effect.Effect<Execution, ControlPlaneNotFoundError | ControlPlaneStorageError> => Effect.gen(function* () { const errors = asOperationErrors(input.operation); const existing = yield* errors.mapStorage( store.executions.getByWorkspaceAndId(input.workspaceId, input.executionId), ); if (Option.isNone(existing)) { return yield* Effect.fail( errors.notFound( "Execution not found", `workspaceId=${input.workspaceId} executionId=${input.executionId}`, ), ); } return existing.value; }); const fetchExecutionEnvelope = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; }, ): Effect.Effect<ExecutionEnvelope, ControlPlaneNotFoundError | ControlPlaneStorageError> => Effect.gen(function* () { const errors = asOperationErrors(input.operation); const execution = yield* fetchExecution(store, input); const pendingInteraction = yield* errors.child("pending_interaction").mapStorage( store.executionInteractions.getPendingByExecutionId(input.executionId), ); return { execution, pendingInteraction: Option.isSome(pendingInteraction) ? pendingInteraction.value : null, }; }); const waitForExecutionEnvelopeToSettle = ( store: ControlPlaneStoreShape, input: { workspaceId: Execution["workspaceId"]; executionId: Execution["id"]; operation: OperationErrorsLike; previousPendingInteractionId: ExecutionInteraction["id"] | null; attemptsRemaining: number; }, ): Effect.Effect<ExecutionEnvelope, ControlPlaneNotFoundError | ControlPlaneStorageError, never> => Effect.gen(function* () { const envelope = yield* fetchExecutionEnvelope(store, input); if ( ( envelope.execution.status !== "running" && !( envelope.execution.status === "waiting_for_interaction" && ( envelope.pendingInteraction === null || envelope.pendingInteraction.id === input.previousPendingInteractionId ) ) ) || input.attemptsRemaining <= 0 ) { return envelope; } yield* Effect.promise( () => new Promise<void>((resolve) => setTimeout(resolve, 25)), ); return yield* waitForExecutionEnvelopeToSettle(store, { ...input, attemptsRemaining: input.attemptsRemaining - 1, }); }); const suspendExecutionForInteraction = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; liveExecutionManager: LiveExecutionManager; request: Parameters<OnElicitation>[0]; interactionId: ExecutionEnvelope["pendingInteraction"] extends infer T ? T extends { id: infer I } ? I : never : never; }) => Effect.gen(function* () { const now = Date.now(); const existing = yield* input.rows.executionInteractions.getById(input.interactionId); const stepSequence = executionStepSequenceFromContext(input.request.context); if (Option.isSome(existing) && existing.value.status !== "pending") { return yield* decodeStoredElicitationResponse({ interactionId: input.interactionId, responseJson: existing.value.responsePrivateJson ?? existing.value.responseJson, }); } if (Option.isNone(existing)) { yield* input.rows.executionInteractions.insert({ id: ExecutionInteractionIdSchema.make(input.interactionId), executionId: input.executionId, status: "pending", kind: input.request.elicitation.mode === "url" ? "url" : "form", purpose: "elicitation", payloadJson: serializeJson({ path: input.request.path, sourceKey: input.request.sourceKey, args: input.request.args, context: input.request.context, elicitation: input.request.elicitation, }) ?? "{}", responseJson: null, responsePrivateJson: null, createdAt: now, updatedAt: now, }); } if (stepSequence !== null) { yield* input.rows.executionSteps.updateByExecutionAndSequence( input.executionId, stepSequence, { status: "waiting", interactionId: ExecutionInteractionIdSchema.make(input.interactionId), updatedAt: now, }, ); } yield* input.rows.executions.update(input.executionId, { status: "waiting_for_interaction", updatedAt: now, }); yield* input.liveExecutionManager.publishState({ executionId: input.executionId, state: "waiting_for_interaction", }); return yield* Effect.fail( createExecutionSuspendedError({ executionId: input.executionId, interactionId: input.interactionId, }), ); }); const createHybridOnElicitation = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; liveExecutionManager: LiveExecutionManager; interactionMode: InteractionMode; }): OnElicitation => { const liveOnElicitation = input.liveExecutionManager.createOnElicitation({ rows: input.rows, executionId: input.executionId, }); return (request) => Effect.gen(function* () { const interactionIdSuffix = interactionIdSuffixForRequest({ interactionId: request.interactionId, path: request.path, context: request.context, }); const interactionId = executionInteractionIdForRequest({ executionId: input.executionId, interactionId: request.interactionId, path: request.path, context: request.context, }); const existing = yield* input.rows.executionInteractions.getById(interactionId); if (Option.isSome(existing) && existing.value.status !== "pending") { return yield* decodeStoredElicitationResponse({ interactionId, responseJson: existing.value.responsePrivateJson ?? existing.value.responseJson, }); } const allowLiveWait = input.interactionMode === "live" || (input.interactionMode === "live_form" && request.elicitation.mode !== "url"); if (Option.isNone(existing) && allowLiveWait) { const stepSequence = executionStepSequenceFromContext(request.context); if (stepSequence !== null) { yield* input.rows.executionSteps.updateByExecutionAndSequence( input.executionId, stepSequence, { status: "waiting", interactionId: ExecutionInteractionIdSchema.make(interactionId), updatedAt: Date.now(), }, ); } return yield* liveOnElicitation({ ...request, interactionId: interactionIdSuffix, }); } return yield* suspendExecutionForInteraction({ rows: input.rows, executionId: input.executionId, liveExecutionManager: input.liveExecutionManager, request, interactionId, }); }); }; const createReplayToolInvoker = (input: { rows: ControlPlaneStoreShape; executionId: Execution["id"]; toolInvoker: ToolInvoker; }): ToolInvoker => ({ invoke: ({ path, args, context }) => Effect.gen(function* () { const stepSequence = executionStepSequenceFromContext(context); if (stepSequence === null) { return yield* input.toolInvoker.invoke({ path, args, context }); } const argsJson = serializeRequiredJson(args); const existing = yield* input.rows.executionSteps.getByExecutionAndSequence( input.executionId, stepSequence, ); if (Option.isSome(existing)) { verifyStoredStepMatches({ executionId: input.executionId, sequence: stepSequence, expectedPath: existing.value.path, expectedArgsJson: existing.value.argsJson, actualPath: path, actualArgsJson: argsJson, }); if (existing.value.status === "completed") { return parseStoredJson(existing.value.resultJson); } if (existing.value.status === "failed") { return yield* Effect.fail( new Error( existing.value.errorText ?? `Stored tool step ${String(stepSequence)} failed`, ), ); } } else { const now = Date.now(); yield* input.rows.executionSteps.insert({