diff --git a/.changeset/configure-workflow-agent-steps.md b/.changeset/configure-workflow-agent-steps.md new file mode 100644 index 0000000000..7bf1cb2915 --- /dev/null +++ b/.changeset/configure-workflow-agent-steps.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Prepare workflow execution to run a configurable number of agent loop steps within each durable Workflow step. Completed logical steps are journaled for cancellation and Workflow retry recovery, while background task launches and requested sleeps remain batching barriers; the limit stays at one. diff --git a/packages/eve/src/context/serialized-dynamic-model-selection.ts b/packages/eve/src/context/serialized-dynamic-model-selection.ts deleted file mode 100644 index dce5e8ef57..0000000000 --- a/packages/eve/src/context/serialized-dynamic-model-selection.ts +++ /dev/null @@ -1,12 +0,0 @@ -import { SessionDynamicModelReferenceKey } from "#context/keys.js"; - -/** Keeps a completed session-scoped model selection when its turn is cancelled. */ -export function preserveSerializedSessionDynamicModelSelection( - original: Record, - interrupted: Record, -): Record { - const selection = interrupted[SessionDynamicModelReferenceKey.name]; - return selection === undefined - ? original - : { ...original, [SessionDynamicModelReferenceKey.name]: selection }; -} diff --git a/packages/eve/src/context/serialized-session-preamble-state.test.ts b/packages/eve/src/context/serialized-session-preamble-state.test.ts new file mode 100644 index 0000000000..4267e6cb58 --- /dev/null +++ b/packages/eve/src/context/serialized-session-preamble-state.test.ts @@ -0,0 +1,46 @@ +import { describe, expect, it } from "vitest"; + +import { + DynamicSkillManifestKey, + SessionDynamicInstructionsKey, + SessionDynamicModelReferenceKey, + SessionDynamicSubagentRuntimeRevisionKey, + SessionDynamicSubagentSelectionsKey, + SessionDynamicToolMetadataKey, + SessionDynamicToolRuntimeRevisionKey, + TurnDynamicModelReferenceKey, +} from "#context/keys.js"; +import { preserveSerializedSessionPreambleState } from "#context/serialized-session-preamble-state.js"; + +describe("preserveSerializedSessionPreambleState", () => { + it("preserves every durable session.started output but not turn state", () => { + const interrupted = { + [SessionDynamicModelReferenceKey.name]: { id: "model" }, + [SessionDynamicToolMetadataKey.name]: [{ name: "tool" }], + [SessionDynamicToolRuntimeRevisionKey.name]: "tools-revision", + [SessionDynamicSubagentSelectionsKey.name]: { researcher: null }, + [SessionDynamicSubagentRuntimeRevisionKey.name]: "subagents-revision", + [DynamicSkillManifestKey.name]: { skills: [{ name: "skill" }] }, + [SessionDynamicInstructionsKey.name]: { + instructions: [{ content: "instruction", role: "system" }], + }, + [TurnDynamicModelReferenceKey.name]: { id: "turn-model" }, + }; + + const preserved = preserveSerializedSessionPreambleState({ original: true }, interrupted); + + for (const key of [ + SessionDynamicModelReferenceKey, + SessionDynamicToolMetadataKey, + SessionDynamicToolRuntimeRevisionKey, + SessionDynamicSubagentSelectionsKey, + SessionDynamicSubagentRuntimeRevisionKey, + DynamicSkillManifestKey, + SessionDynamicInstructionsKey, + ]) { + expect(preserved[key.name]).toEqual(interrupted[key.name]); + } + expect(preserved).not.toHaveProperty(TurnDynamicModelReferenceKey.name); + expect(preserved.original).toBe(true); + }); +}); diff --git a/packages/eve/src/context/serialized-session-preamble-state.ts b/packages/eve/src/context/serialized-session-preamble-state.ts new file mode 100644 index 0000000000..7b037d477b --- /dev/null +++ b/packages/eve/src/context/serialized-session-preamble-state.ts @@ -0,0 +1,32 @@ +import { + DynamicSkillManifestKey, + SessionDynamicInstructionsKey, + SessionDynamicModelReferenceKey, + SessionDynamicSubagentRuntimeRevisionKey, + SessionDynamicSubagentSelectionsKey, + SessionDynamicToolMetadataKey, + SessionDynamicToolRuntimeRevisionKey, +} from "#context/keys.js"; + +const SESSION_PREAMBLE_KEYS = [ + SessionDynamicModelReferenceKey, + SessionDynamicToolMetadataKey, + SessionDynamicToolRuntimeRevisionKey, + SessionDynamicSubagentSelectionsKey, + SessionDynamicSubagentRuntimeRevisionKey, + DynamicSkillManifestKey, + SessionDynamicInstructionsKey, +] as const; + +/** Keeps durable session.started resolver output when its turn is cancelled. */ +export function preserveSerializedSessionPreambleState( + original: Record, + interrupted: Record, +): Record { + let preserved = original; + for (const key of SESSION_PREAMBLE_KEYS) { + const value = interrupted[key.name]; + if (value !== undefined) preserved = { ...preserved, [key.name]: value }; + } + return preserved; +} diff --git a/packages/eve/src/execution/agent-loop-batch.ts b/packages/eve/src/execution/agent-loop-batch.ts new file mode 100644 index 0000000000..abfed1d1af --- /dev/null +++ b/packages/eve/src/execution/agent-loop-batch.ts @@ -0,0 +1,115 @@ +import type { ContextContainer } from "#context/container.js"; +import { HandleEventKey } from "#context/keys.js"; +import { serializeContext } from "#context/serialize.js"; +import { + type AgentLoopCheckpoint, + writeAgentLoopCheckpoint, +} from "#execution/agent-loop-checkpoint.js"; +import { createDurableSessionState } from "#execution/durable-session-store.js"; +import type { DurableTransition } from "#execution/next-driver-action.js"; +import { reconcileSessionContinuationToken } from "#execution/reconcile-session-continuation-token.js"; +import { runBackgroundStep } from "#execution/tasks/parent/tool-execution.js"; +import { parseJsonObject } from "#shared/json.js"; +import { stageAttachmentsToSandbox } from "#harness/attachment-staging.js"; +import { normalizeUserContent } from "#harness/messages.js"; +import { throwIfTurnAborted } from "#harness/turn-cancellation.js"; +import type { HandleEventFn, HarnessSession, StepInput, StepResult } from "#harness/types.js"; + +export async function preserveCancelledTurnMessage( + session: HarnessSession, + input: StepInput | undefined, +): Promise { + const message = normalizeUserContent(input?.message); + if (message === undefined) return session; + const content = await stageAttachmentsToSandbox(message); + return { ...session, history: [...session.history, { content, role: "user" }] }; +} + +export class AgentLoopBatch { + private readonly abortSignal: AbortSignal | undefined; + private readonly ctx: ContextContainer; + private readonly initialSessionState: AgentLoopCheckpoint["sessionState"]; + private readonly initialSerializedContext: Record; + private latestSession: HarnessSession; + private checkpoint: AgentLoopCheckpoint | undefined; + completedSteps: number; + + constructor( + abortSignal: AbortSignal | undefined, + ctx: ContextContainer, + initialSession: HarnessSession, + initialSessionState: AgentLoopCheckpoint["sessionState"], + initialSerializedContext: Record, + checkpoint: AgentLoopCheckpoint | undefined, + ) { + this.abortSignal = abortSignal; + this.ctx = ctx; + this.initialSessionState = initialSessionState; + this.initialSerializedContext = initialSerializedContext; + this.latestSession = initialSession; + this.checkpoint = checkpoint; + this.completedSteps = checkpoint?.completedSteps ?? 0; + } + + cancellationTransition(): DurableTransition | undefined { + return this.checkpoint === undefined + ? undefined + : { + serializedContext: this.checkpoint.serializedContext, + sessionState: this.checkpoint.sessionState, + }; + } + + checkpointTransition(): DurableTransition { + return this.checkpoint === undefined + ? { + serializedContext: this.initialSerializedContext, + sessionState: this.initialSessionState, + } + : { + serializedContext: this.checkpoint.serializedContext, + sessionState: this.checkpoint.sessionState, + }; + } + + checkpointSession(): HarnessSession { + return this.latestSession; + } + + checkpointSessionState(): AgentLoopCheckpoint["sessionState"] | undefined { + return this.checkpoint?.sessionState; + } + + checkpointSerializedContext(): Record { + return this.checkpoint?.serializedContext ?? this.initialSerializedContext; + } + + async commitContinuation(): Promise { + this.checkpoint = await writeAgentLoopCheckpoint({ + completedSteps: this.completedSteps, + serializedContext: parseJsonObject(serializeContext(this.ctx)), + sessionState: createDurableSessionState({ session: this.latestSession }), + }); + } + + async run( + session: HarnessSession, + handleEvent: HandleEventFn, + callback: (enrichedSession: HarnessSession) => Promise, + ): Promise { + throwIfTurnAborted(this.abortSignal); + let result = await runBackgroundStep(this.ctx, session, async (enrichedSession) => { + this.ctx.setVirtualContext(HandleEventKey, handleEvent); + return callback(enrichedSession); + }); + if (result.backgroundTasks === undefined) throwIfTurnAborted(this.abortSignal); + + result = { + ...result, + session: reconcileSessionContinuationToken(this.ctx, result.session), + }; + this.latestSession = result.session; + this.completedSteps += 1; + return result; + } +} diff --git a/packages/eve/src/execution/agent-loop-checkpoint.ts b/packages/eve/src/execution/agent-loop-checkpoint.ts new file mode 100644 index 0000000000..6430904e7e --- /dev/null +++ b/packages/eve/src/execution/agent-loop-checkpoint.ts @@ -0,0 +1,94 @@ +import { + getStepMetadata, + getWorkflowMetadata, + getWritable, +} from "#compiled/@workflow/core/index.js"; + +import type { DurableSessionState } from "#execution/durable-session-store.js"; +import type { TurnStepInput } from "#execution/durable-session-migrations/turn-workflow.js"; +import { getRun } from "#internal/workflow/runtime.js"; + +const AGENT_LOOP_CHECKPOINT_VERSION = 1; +const AGENT_LOOP_CHECKPOINT_NAMESPACE_PREFIX = "eve.agent-loop-checkpoint"; + +export interface AgentLoopCheckpoint { + readonly completedSteps: number; + readonly serializedContext: Record; + readonly sessionState: DurableSessionState; + readonly version: typeof AGENT_LOOP_CHECKPOINT_VERSION; +} + +export async function resumeAgentLoopCheckpoint(input: { + readonly enabled: boolean; + readonly rawInput: TurnStepInput; +}): Promise<{ readonly checkpoint?: AgentLoopCheckpoint; readonly stepInput: TurnStepInput }> { + const checkpoint = input.enabled ? await readAgentLoopCheckpoint() : undefined; + return checkpoint === undefined + ? { stepInput: input.rawInput } + : { + checkpoint, + stepInput: { + ...input.rawInput, + input: undefined, + serializedContext: checkpoint.serializedContext, + sessionState: checkpoint.sessionState, + }, + }; +} + +export async function readAgentLoopCheckpoint(): Promise { + const { attempt, stepId } = getStepMetadata(); + if (attempt === 1) return undefined; + return readCheckpointTail(checkpointNamespace(stepId)); +} + +export async function writeAgentLoopCheckpoint( + checkpoint: Omit, +): Promise { + const namespace = checkpointNamespace(getStepMetadata().stepId); + const writer = getWritable({ namespace }).getWriter(); + const persisted: AgentLoopCheckpoint = { + ...checkpoint, + version: AGENT_LOOP_CHECKPOINT_VERSION, + }; + try { + await writer.write(persisted); + } finally { + writer.releaseLock(); + } + return persisted; +} + +async function readCheckpointTail(namespace: string): Promise { + const metadata = getWorkflowMetadata(); + const runId = metadata.workflowRunId; + if (typeof runId !== "string") { + throw new Error("Agent loop checkpointing requires a Workflow run id."); + } + const run = getRun(runId); + const tail = run.getReadable({ namespace }); + if ((await tail.getTailIndex()) === -1) return undefined; + const reader = run.getReadable({ namespace, startIndex: -1 }).getReader(); + try { + const result = await reader.read(); + return result.done ? undefined : parseAgentLoopCheckpoint(result.value); + } finally { + await reader.cancel("eve agent loop checkpoint tail read complete").catch(() => {}); + reader.releaseLock(); + } +} + +function parseAgentLoopCheckpoint(value: unknown): AgentLoopCheckpoint { + if ( + typeof value !== "object" || + value === null || + (value as { readonly version?: unknown }).version !== AGENT_LOOP_CHECKPOINT_VERSION + ) { + throw new Error("Agent loop checkpoint has an unsupported or malformed version."); + } + return value as AgentLoopCheckpoint; +} + +function checkpointNamespace(stepId: string): string { + return `${AGENT_LOOP_CHECKPOINT_NAMESPACE_PREFIX}:${stepId}`; +} diff --git a/packages/eve/src/execution/agent-loop-config.ts b/packages/eve/src/execution/agent-loop-config.ts new file mode 100644 index 0000000000..577e2b371e --- /dev/null +++ b/packages/eve/src/execution/agent-loop-config.ts @@ -0,0 +1 @@ +export const AGENT_LOOP_STEPS_PER_WORKFLOW_STEP = 1; diff --git a/packages/eve/src/execution/durable-session-store.integration.test.ts b/packages/eve/src/execution/durable-session-store.integration.test.ts index 9daf8db46a..6171c9af43 100644 --- a/packages/eve/src/execution/durable-session-store.integration.test.ts +++ b/packages/eve/src/execution/durable-session-store.integration.test.ts @@ -3,6 +3,7 @@ import { start } from "#internal/workflow/runtime.js"; import { createTestRuntime } from "#internal/testing/app-harness.js"; import { + agentLoopCheckpointRetryFixtureWorkflow, durableSessionRetryFixtureWorkflow, durableSessionStoreFixtureWorkflow, } from "#internal/testing/durable-session-workflow.js"; @@ -64,6 +65,36 @@ describe("durableSessionStore integration", () => { }); }); + it("retains an agent-loop checkpoint across a physical step retry", async () => { + const runtime = createTestRuntime({ agent: { name: "agent-loop-checkpoint-retry" } }); + + await runtime.run(async () => { + const run = await start(agentLoopCheckpointRetryFixtureWorkflow, [ + { writeBeforeFailure: true }, + ]); + await expect(run.returnValue).resolves.toEqual({ + attempt: 2, + completedSteps: 3, + resumed: true, + }); + }); + }); + + it("retries logical step one without blocking on an empty checkpoint journal", async () => { + const runtime = createTestRuntime({ agent: { name: "agent-loop-checkpoint-empty" } }); + + await runtime.run(async () => { + const run = await start(agentLoopCheckpointRetryFixtureWorkflow, [ + { writeBeforeFailure: false }, + ]); + await expect(run.returnValue).resolves.toEqual({ + attempt: 2, + completedSteps: 0, + resumed: false, + }); + }); + }); + it("a write-step retry's returned state is what the subsequent read returns", async () => { const runtime = createTestRuntime({ agent: { name: "durable-session-store-fixture-retry" } }); diff --git a/packages/eve/src/execution/finalize-turn-step-result.ts b/packages/eve/src/execution/finalize-turn-step-result.ts new file mode 100644 index 0000000000..c12770a1fa --- /dev/null +++ b/packages/eve/src/execution/finalize-turn-step-result.ts @@ -0,0 +1,147 @@ +import type { ContextContainer } from "#context/container.js"; +import type { AgentLoopBatch } from "#execution/agent-loop-batch.js"; +import { preserveSerializedSessionPreambleState } from "#context/serialized-session-preamble-state.js"; +import { createDurableSessionState } from "#execution/durable-session-store.js"; +import type { DurableStepResult } from "#execution/next-driver-action.js"; +import { derivePendingState } from "#execution/pending-turn-state.js"; +import { reconcileSessionContinuationToken } from "#execution/reconcile-session-continuation-token.js"; +import { + getRuntimeActionKeysFromWorkflowInterrupt, + isWorkflowRuntimeActionInterrupt, +} from "#harness/workflow-runtime-action-state.js"; +import { getPendingWorkflowInterrupt } from "#harness/workflow-interrupt-state.js"; +import { hasPendingInputBatch } from "#harness/input-requests.js"; +import { readTurnSleepDurationMs } from "#harness/turn-sleep.js"; +import { getTurnUsageState, takeSessionUsageDelta, toUsage } from "#harness/turn-tag-state.js"; +import type { StepResult } from "#harness/types.js"; +import type { RunMode } from "#shared/run-mode.js"; +import { serializeContext } from "#context/serialize.js"; + +const TASK_DONE_WITH_PENDING_INPUT_ERROR_MESSAGE = + "Task mode cannot complete while input requests remain pending."; + +export async function finalizeTurnStepResult(input: { + readonly batch: AgentLoopBatch; + readonly ctx: ContextContainer; + readonly mode: RunMode; + readonly tasksEnabled: boolean; + readonly writer: WritableStreamDefaultWriter; + readonly stepResult: StepResult; +}): Promise { + let stepResult = { + ...input.stepResult, + session: reconcileSessionContinuationToken(input.ctx, input.stepResult.session), + }; + const serializedContext = serializeContext(input.ctx); + const sessionState = createDurableSessionState({ session: stepResult.session }); + const checkpointTransition = input.batch.cancellationTransition(); + const cancellationContext = preserveSerializedSessionPreambleState( + checkpointTransition?.serializedContext ?? input.batch.checkpointSerializedContext(), + serializedContext, + ); + const backgroundTaskState = + stepResult.backgroundTasks === undefined || stepResult.backgroundTaskSession === undefined + ? undefined + : createDurableSessionState({ session: stepResult.backgroundTaskSession }); + const cancellationTransition = + backgroundTaskState === undefined + ? checkpointTransition === undefined + ? undefined + : { ...checkpointTransition, serializedContext: cancellationContext } + : { + serializedContext: cancellationContext, + sessionState: backgroundTaskState, + }; + const commitBarrier = + backgroundTaskState === undefined || stepResult.backgroundTasks === undefined + ? undefined + : { + effect: { kind: "release-background-tasks" as const, tasks: stepResult.backgroundTasks }, + transition: cancellationTransition!, + }; + const transitions = { cancellationTransition, commitBarrier }; + const sleepDurationMs = readTurnSleepDurationMs(input.ctx); + const sleep = sleepDurationMs === undefined ? {} : { sleepDurationMs }; + + if ( + stepResult.next !== null && + typeof stepResult.next === "object" && + "done" in stepResult.next + ) { + if (input.mode === "task" && hasPendingInputBatch(stepResult.session.state)) { + input.writer.releaseLock(); + throw new Error(TASK_DONE_WITH_PENDING_INPUT_ERROR_MESSAGE); + } + await input.writer.close(); + const sessionTotals = getTurnUsageState(stepResult.session.state)?.session; + return { + action: "done", + ...transitions, + output: stepResult.next.output, + isError: stepResult.next.isError, + ...sleep, + serializedContext, + sessionState, + usage: sessionTotals === undefined ? undefined : toUsage(sessionTotals), + usageDelta: takeSessionUsageDelta(stepResult.session).delta, + }; + } + + if (stepResult.next === null) { + input.writer.releaseLock(); + const workflowInterrupt = getPendingWorkflowInterrupt(stepResult.session.state); + if ( + workflowInterrupt !== undefined && + isWorkflowRuntimeActionInterrupt(workflowInterrupt.interrupt) + ) { + return { + action: "dispatch-workflow-runtime-actions", + ...transitions, + pendingRuntimeActionKeys: getRuntimeActionKeysFromWorkflowInterrupt( + workflowInterrupt.interrupt, + ), + ...sleep, + serializedContext, + sessionState, + }; + } + + const pending = derivePendingState(stepResult.session); + if (stepResult.settledTurn !== undefined) { + const { delta, session: reportedSession } = takeSessionUsageDelta(stepResult.session); + return { + action: "park", + ...transitions, + ...pending, + ...sleep, + serializedContext, + sessionState: createDurableSessionState({ session: reportedSession }), + settled: { + output: stepResult.settledTurn.output, + isError: stepResult.settledTurn.isError, + usage: delta, + }, + tasksEnabled: input.tasksEnabled, + }; + } + + return { + action: "park", + ...transitions, + ...pending, + ...sleep, + serializedContext, + sessionState, + tasksEnabled: input.tasksEnabled, + }; + } + + input.writer.releaseLock(); + return { + action: "continue", + ...transitions, + ...sleep, + serializedContext, + sessionState, + }; +} diff --git a/packages/eve/src/execution/next-driver-action.ts b/packages/eve/src/execution/next-driver-action.ts index 3ccdb0cf4e..aa2aed81b7 100644 --- a/packages/eve/src/execution/next-driver-action.ts +++ b/packages/eve/src/execution/next-driver-action.ts @@ -15,13 +15,24 @@ import type { DurableSessionState } from "#execution/durable-session-store.js"; import type { SettledTurn, StepResult } from "#harness/types.js"; import type { TokenUsage } from "#shared/token-usage.js"; -interface DurableStepResultFields { - readonly backgroundTaskState?: DurableSessionState; - readonly backgroundTasks?: StepResult["backgroundTasks"]; +export interface DurableTransition { readonly serializedContext: Record; readonly sessionState: DurableSessionState; } +export interface DurableCommitBarrier { + readonly effect: { + readonly kind: "release-background-tasks"; + readonly tasks: NonNullable; + }; + readonly transition: DurableTransition; +} + +interface DurableStepResultFields extends DurableTransition { + readonly cancellationTransition?: DurableTransition; + readonly commitBarrier?: DurableCommitBarrier; +} + /** Result returned by the latest turn step to its durable driver workflow. */ export type DurableStepResult = ( | { diff --git a/packages/eve/src/execution/turn-workflow.test.ts b/packages/eve/src/execution/turn-workflow.test.ts index 24f21ace67..8ce156f045 100644 --- a/packages/eve/src/execution/turn-workflow.test.ts +++ b/packages/eve/src/execution/turn-workflow.test.ts @@ -219,6 +219,37 @@ describe("turnWorkflow", () => { ); }); + it("keeps completed step state when cancellation interrupts a durable sleep", async () => { + const initialState = createSessionState({ continuationToken: "http:initial" }); + const sleepingState = createSessionState({ continuationToken: "http:sleeping" }); + let requestCancel!: () => void; + const cancelPayload = new Promise((resolve) => { + requestCancel = () => resolve({}); + }); + installInbox([], { cancelPayloads: [cancelPayload] }); + sleepMock.mockImplementationOnce(async () => { + requestCancel(); + await new Promise(() => {}); + }); + vi.mocked(turnStep).mockResolvedValueOnce({ + action: "continue", + sleepDurationMs: 2_500, + serializedContext: { state: "sleeping" }, + sessionState: sleepingState, + }); + + const { input } = createInput({ + driverCapabilities: { cancelledTurnSettle: true, turnInbox: true }, + sessionState: initialState, + }); + await turnWorkflow(input); + + expect(cancelDescendantTurnsStep).toHaveBeenCalledWith({ + serializedContext: { state: "sleeping" }, + sessionState: sleepingState, + }); + }); + it("parks when an authorization is pending", async () => { const sessionState = createSessionState(); vi.mocked(turnStep).mockResolvedValueOnce({ @@ -373,10 +404,19 @@ describe("turnWorkflow", () => { installInbox([]); vi.mocked(turnStep).mockResolvedValueOnce({ action: "cancelled", - backgroundTaskState: backgroundState, - backgroundTasks, + cancellationTransition: { + serializedContext: { state: "cancelled" }, + sessionState: backgroundState, + }, + commitBarrier: { + effect: { kind: "release-background-tasks", tasks: backgroundTasks }, + transition: { + serializedContext: { state: "cancelled" }, + sessionState: backgroundState, + }, + }, serializedContext: { state: "cancelled" }, - sessionState: initialState, + sessionState: backgroundState, }); const { input } = createInput({ @@ -393,6 +433,41 @@ describe("turnWorkflow", () => { }); }); + it("keeps earlier logical context when cancellation races background tasks", async () => { + const initialState = createSessionState({ continuationToken: "http:initial" }); + const backgroundState = createSessionState({ continuationToken: "http:background" }); + installInbox([], { cancelPayloads: [{}] }); + vi.mocked(turnStep).mockImplementationOnce(async (stepInput) => { + await vi.waitFor(() => expect(stepInput.abortSignal?.aborted).toBe(true)); + const tasks = [{ taskId: "task-1", taskInboxToken: "task-inbox-1", taskRunId: "task-run-1" }]; + const cancellationTransition = { + serializedContext: { state: "checkpoint" }, + sessionState: backgroundState, + }; + return { + action: "continue", + cancellationTransition, + commitBarrier: { + effect: { kind: "release-background-tasks" as const, tasks }, + transition: cancellationTransition, + }, + serializedContext: { state: "task-result" }, + sessionState: backgroundState, + }; + }); + + const { input } = createInput({ + driverCapabilities: { cancelledTurnSettle: true, turnInbox: true }, + sessionState: initialState, + }); + await turnWorkflow(input); + + expect(cancelDescendantTurnsStep).toHaveBeenCalledWith({ + serializedContext: { state: "checkpoint" }, + sessionState: backgroundState, + }); + }); + it("honors cancellation observed while a durable turn step returns", async () => { const sessionState = createSessionState(); const sessionModel = { @@ -444,6 +519,36 @@ describe("turnWorkflow", () => { ); }); + it("uses the logical-step checkpoint when cancellation races a batched result", async () => { + const initialState = createSessionState({ continuationToken: "http:initial" }); + const checkpointState = createSessionState({ continuationToken: "http:checkpoint" }); + const resultState = createSessionState({ continuationToken: "http:result" }); + installInbox([], { cancelPayloads: [{}] }); + vi.mocked(turnStep).mockImplementationOnce(async (stepInput) => { + await vi.waitFor(() => expect(stepInput.abortSignal?.aborted).toBe(true)); + return { + action: "continue", + cancellationTransition: { + serializedContext: { state: "checkpoint" }, + sessionState: checkpointState, + }, + serializedContext: { state: "result" }, + sessionState: resultState, + }; + }); + + const { input } = createInput({ + driverCapabilities: { cancelledTurnSettle: true, turnInbox: true }, + sessionState: initialState, + }); + await turnWorkflow(input); + + expect(cancelDescendantTurnsStep).toHaveBeenCalledWith({ + serializedContext: { state: "checkpoint" }, + sessionState: checkpointState, + }); + }); + it("runs uncancellable when the session cancel token is claimed by another run", async () => { const sessionState = createSessionState(); installInbox([], { cancelConflict: { runId: "wrun_stale_prior_turn" } }); @@ -1274,11 +1379,11 @@ function createCancelHookMock( dispose: vi.fn(), [Symbol.asyncIterator](): AsyncIterator { return { - next: () => { + next: async () => { const value = queue.shift(); return value === undefined - ? new Promise>(() => {}) - : Promise.resolve({ done: false, value }); + ? await new Promise>(() => {}) + : { done: false, value: await value }; }, return: vi.fn(async () => ({ done: true, value: undefined })), }; diff --git a/packages/eve/src/execution/turn-workflow.ts b/packages/eve/src/execution/turn-workflow.ts index 68738b026a..5e45068fe3 100644 --- a/packages/eve/src/execution/turn-workflow.ts +++ b/packages/eve/src/execution/turn-workflow.ts @@ -6,7 +6,7 @@ import { } from "#compiled/@workflow/core/index.js"; import type { DeliverHookPayload } from "#channel/types.js"; -import { preserveSerializedSessionDynamicModelSelection } from "#context/serialized-dynamic-model-selection.js"; +import { preserveSerializedSessionPreambleState } from "#context/serialized-session-preamble-state.js"; import { cancelDescendantTurnsStep } from "#execution/cancel-descendant-turns-step.js"; import { sendTurnControlStep, type TurnInboxPayload } from "#execution/turn-control-protocol.js"; import { dispatchRuntimeActionsStep } from "#execution/dispatch-runtime-actions-step.js"; @@ -19,7 +19,11 @@ import { type TurnWorkflowInput, } from "#execution/durable-session-migrations/turn-workflow.js"; import { claimHookOwnership, disposeHook, isHookConflictError } from "#execution/hook-ownership.js"; -import type { NextDriverAction } from "#execution/next-driver-action.js"; +import type { + DurableCommitBarrier, + DurableTransition, + NextDriverAction, +} from "#execution/next-driver-action.js"; import { routeDeliverToChildren } from "#execution/route-child-delivery.js"; import { runProxySubagentEventStep } from "#execution/subagent-event-proxy-step.js"; import { @@ -110,17 +114,8 @@ async function runTurnOwnedWorkflow(input: TurnWorkflowInput): Promise { result.action === "dispatch-workflow-runtime-actions" || result.action === "park" ? result.pendingRuntimeActionKeys : undefined; - const hasBackgroundTasks = (result.backgroundTasks?.length ?? 0) > 0; - - if (hasBackgroundTasks) { - if (result.backgroundTaskState === undefined) { - throw new Error("Background tasks were returned without their committed session state."); - } - await cursor.adopt({ - serializedContext: beforeStep.serializedContext, - sessionState: result.backgroundTaskState, - }); - await acknowledgeDelegatedTasksStep({ tasks: result.backgroundTasks ?? [] }); + if (result.commitBarrier !== undefined) { + await executeCommitBarrier(result.commitBarrier, cursor); } // A cancel observed while the step was returning must still win: the @@ -131,28 +126,27 @@ async function runTurnOwnedWorkflow(input: TurnWorkflowInput): Promise { // the driver epilogue and later turns, plus the accepted user input in // durable history. Adopt those before settling so a steered replacement // keeps the interrupted request without committing partial model output. - await cursor.adopt({ - serializedContext: result.serializedContext, - sessionState: result.backgroundTaskState ?? result.sessionState, - }); + await cursor.adopt(result.cancellationTransition ?? result); await finishCancelledTurn({ bufferedDeliveries, cancellation, cursor }); return; } if ( cancellation?.signal.aborted === true && - (pendingActionKeys === undefined || hasBackgroundTasks) + (pendingActionKeys === undefined || result.commitBarrier !== undefined) ) { // Some worlds cannot interrupt a running step, so it can complete - // normally after the workflow observes cancellation. Roll that result - // back except for a session model selected by its one-time preamble. - await cursor.adopt({ - serializedContext: preserveSerializedSessionDynamicModelSelection( + // normally after the workflow observes cancellation. Prefer the + // step-owned cancellation transition; old results retain the original + // whole-step rollback plus the one-time session model. + const cancellationTransition: DurableTransition = result.cancellationTransition ?? { + serializedContext: preserveSerializedSessionPreambleState( beforeStep.serializedContext, result.serializedContext, ), sessionState: cursor.sessionState, - }); + }; + await cursor.adopt(cancellationTransition); // No `canPark` check here: that gate rejects model-authored waits // (`next: null`) in task mode, whereas every session can resume by // stable ID after a cancelled turn. The epilogue runs in the driver @@ -165,6 +159,7 @@ async function runTurnOwnedWorkflow(input: TurnWorkflowInput): Promise { if (result.sleepDurationMs !== undefined) { const outcome = await waitForTurnSleep(result.sleepDurationMs, cancellation); if (outcome === "cancel") { + await cursor.adopt(result); await finishCancelledTurn({ bufferedDeliveries, cancellation, cursor }); return; } @@ -277,6 +272,17 @@ async function runTurnOwnedWorkflow(input: TurnWorkflowInput): Promise { } } +async function executeCommitBarrier( + barrier: DurableCommitBarrier, + cursor: TurnExecutionCursor, +): Promise { + await cursor.adopt(barrier.transition); + switch (barrier.effect.kind) { + case "release-background-tasks": + await acknowledgeDelegatedTasksStep({ tasks: barrier.effect.tasks }); + } +} + async function finishCancelledTurn(input: { readonly bufferedDeliveries: readonly DeliverHookPayload[]; readonly cancellation: TurnCancellationControl | undefined; diff --git a/packages/eve/src/execution/workflow-steps.test.ts b/packages/eve/src/execution/workflow-steps.test.ts index 52db413b9d..a493e55ec0 100644 --- a/packages/eve/src/execution/workflow-steps.test.ts +++ b/packages/eve/src/execution/workflow-steps.test.ts @@ -12,11 +12,13 @@ import { ContextKey } from "#context/key.js"; import { AuthKey, ContinuationTokenKey, + DynamicSkillManifestKey, DynamicSubagentAgentConfigKey, ModeKey, SessionCallbackKey, SessionDynamicSubagentRuntimeRevisionKey, SessionDynamicSubagentSelectionsKey, + SessionDynamicInstructionsKey, SessionDynamicModelReferenceKey, SessionDynamicToolMetadataKey, SessionDynamicToolRuntimeRevisionKey, @@ -33,7 +35,7 @@ import { TurnCancelledError } from "#harness/turn-cancellation.js"; import { getPendingAuthorization, setPendingAuthorization } from "#harness/authorization.js"; import { getProxyInputRequests, upsertProxyInputRequests } from "#harness/proxy-input-requests.js"; import { appendPendingInputBatch } from "#harness/input-requests.js"; -import type { HarnessSession, StepResult } from "#harness/types.js"; +import type { HarnessSession, StepInput, StepResult } from "#harness/types.js"; import { createEmptyHookRegistry } from "#runtime/hooks/registry.js"; import { createInputRequestedEvent } from "#protocol/message.js"; import { getCompiledRuntimeAgentBundle } from "#runtime/sessions/compiled-agent-cache.js"; @@ -57,6 +59,7 @@ import { recordTaskInputRequestStep } from "#execution/tasks/parent/hitl-proxy-s import { appendTaskAgentAnnouncement } from "#execution/tasks/parent/agent-views.js"; import { emitTerminalSessionFailureStep } from "#execution/terminal-session-failure-step.js"; import { resolveEffectiveOutputSchema } from "#execution/effective-output-schema.js"; +import { writeAgentLoopCheckpoint } from "#execution/agent-loop-checkpoint.js"; import { turnStep } from "#execution/workflow-steps.js"; import { routeProxiedDeliverStep } from "#execution/proxied-deliver-step.js"; import { @@ -65,6 +68,37 @@ import { workflowEntryReference, } from "#execution/workflow-runtime.js"; +const mockAgentLoopCheckpoint = vi.hoisted(() => ({ latest: undefined as unknown })); +vi.mock("./agent-loop-checkpoint.js", () => ({ + readAgentLoopCheckpoint: vi.fn(async () => mockAgentLoopCheckpoint.latest), + resumeAgentLoopCheckpoint: vi.fn(async ({ enabled, rawInput }) => + enabled && mockAgentLoopCheckpoint.latest !== undefined + ? { + checkpoint: mockAgentLoopCheckpoint.latest, + stepInput: { + ...rawInput, + input: undefined, + serializedContext: (mockAgentLoopCheckpoint.latest as { serializedContext: unknown }) + .serializedContext, + sessionState: (mockAgentLoopCheckpoint.latest as { sessionState: unknown }) + .sessionState, + }, + } + : { stepInput: rawInput }, + ), + writeAgentLoopCheckpoint: vi.fn(async (checkpoint) => { + mockAgentLoopCheckpoint.latest = structuredClone({ ...checkpoint, version: 1 }); + return mockAgentLoopCheckpoint.latest; + }), +})); + +const mockAgentLoopConfig = vi.hoisted(() => ({ stepsPerWorkflowStep: 1 })); +vi.mock("./agent-loop-config.js", () => ({ + get AGENT_LOOP_STEPS_PER_WORKFLOW_STEP() { + return mockAgentLoopConfig.stepsPerWorkflowStep; + }, +})); + vi.mock("./durable-session-store.js", async (importOriginal) => { const actual = await importOriginal(); return { @@ -169,6 +203,9 @@ vi.mock("#compiled/@workflow/core/runtime.js", () => ({ })); const ThreadKey = new ContextKey("test.workflow.thread"); +const NestedCheckpointKey = new ContextKey<{ nested: { value: string } }>( + "test.workflow.nestedCheckpoint", +); const TestTurnAgent = { id: "test-agent", instructions: ["You are a test agent."], @@ -203,12 +240,8 @@ function createStubSession(overrides: Partial = {}): HarnessSess }; } -function createSerializedContext( - mode: "conversation" | "task" = "conversation", -): Record { - const ctx = new ContextContainer(); - ctx.set(AuthKey, null); - ctx.set(BundleKey, { +function createTestCompiledBundle() { + return { adapterRegistry: { adaptersByKind: new Map([[threadContextAdapter.kind, threadContextAdapter]]), }, @@ -225,7 +258,15 @@ function createSerializedContext( subagentRegistry: {}, toolRegistry: {}, turnAgent: TestTurnAgent, - } as never); + } as never; +} + +function createSerializedContext( + mode: "conversation" | "task" = "conversation", +): Record { + const ctx = new ContextContainer(); + ctx.set(AuthKey, null); + ctx.set(BundleKey, createTestCompiledBundle()); ctx.set(ChannelKey, threadContextAdapter); ctx.set(ContinuationTokenKey, "http:thread-context"); ctx.set(ModeKey, mode); @@ -234,6 +275,9 @@ function createSerializedContext( } afterEach(() => { + mockAgentLoopCheckpoint.latest = undefined; + mockAgentLoopConfig.stepsPerWorkflowStep = 1; + vi.mocked(writeAgentLoopCheckpoint).mockClear(); getRunMock.mockReset(); resumeHookMock.mockReset(); startMock.mockReset(); @@ -1749,6 +1793,175 @@ describe("turnStep", () => { expect(workflowWritesByNamespace.get(DEFAULT_WORKFLOW_STREAM_NAMESPACE) ?? []).toEqual([]); }); + it("runs agent-loop continuations up to the workflow-step limit", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; + vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(createTestCompiledBundle()); + vi.mocked(createExecutionNodeStep).mockClear(); + const sessions = [0, 1, 2, 3].map((index) => + createStubSession({ history: [{ content: `step ${index}`, role: "assistant" }] }), + ); + installSessionStoreMocks([sessions[0]!]); + const sessionModel = { id: "openai/gpt-5.6-sol", contextWindowTokens: 1_000_000 }; + const run = vi.fn(async (_session: HarnessSession, _input?: StepInput): Promise => { + const index = run.mock.calls.length; + if (index === 3) { + const ctx = loadContext(); + ctx.set(SessionDynamicModelReferenceKey, sessionModel); + ctx.set(DynamicSkillManifestKey, { skills: [{ description: "Skill", name: "skill" }] }); + ctx.set(SessionDynamicInstructionsKey, { + instructions: [{ content: "Session instruction", role: "system" }], + }); + } + return { + next: index < 4 ? run : { done: true, output: "done" }, + session: sessions[index]!, + }; + }); + vi.mocked(createExecutionNodeStep).mockImplementation(() => run); + + const result = await turnStep({ + input: { kind: "deliver", payloads: [{ message: "request" }] }, + parentWritable: createTestWritable(), + serializedContext: createSerializedContext(), + sessionState: createStubSessionState(), + }); + + expect(run).toHaveBeenCalledTimes(3); + expect(createExecutionNodeStep).toHaveBeenCalledTimes(3); + expect(writeAgentLoopCheckpoint).toHaveBeenCalledTimes(2); + expect(run.mock.calls.map((call) => call[1])).toEqual([ + expect.objectContaining({ message: expect.any(String) }), + undefined, + undefined, + ]); + expect(result).toMatchObject({ action: "continue" }); + expect(result.sessionState.snapshot?.session.history).toEqual(sessions[3]!.history); + expect(result.cancellationTransition?.sessionState.snapshot?.session.history).toEqual( + sessions[2]!.history, + ); + expect(result.cancellationTransition?.serializedContext).toMatchObject({ + [DynamicSkillManifestKey.name]: { + skills: [{ description: "Skill", name: "skill" }], + }, + [SessionDynamicInstructionsKey.name]: { + instructions: [{ content: "Session instruction", role: "system" }], + }, + [SessionDynamicModelReferenceKey.name]: sessionModel, + }); + }); + + it("resumes a physical retry from the durable logical checkpoint", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; + const bundle = createTestCompiledBundle(); + vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(bundle); + const checkpointSession = createStubSession({ + history: [{ content: "completed step", role: "assistant" }], + }); + installSessionStoreMocks([checkpointSession]); + mockAgentLoopCheckpoint.latest = { + completedSteps: 1, + serializedContext: createSerializedContext(), + sessionState: createStubSessionState(), + version: 1, + }; + let observedInput: StepInput | undefined; + vi.mocked(createExecutionNodeStep).mockImplementation(() => { + return async (session, input): Promise => { + observedInput = input; + return { next: { done: true, output: "done" }, session }; + }; + }); + + const result = await turnStep({ + input: { kind: "deliver", payloads: [{ message: "must not redeliver" }] }, + parentWritable: createTestWritable(), + serializedContext: createSerializedContext(), + sessionState: createStubSessionState(), + }); + + expect(observedInput).toBeUndefined(); + expect(result.action).toBe("done"); + expect(result.sessionState.snapshot?.session.history).toEqual(checkpointSession.history); + }); + + it("rebuilds channel adapter context for each agent-loop step", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; + const createAdapterContext = vi.fn((base) => base); + const adapter = { ...threadContextAdapter, createAdapterContext } as ChannelAdapter; + const bundle = Object.assign({}, createTestCompiledBundle(), { + adapterRegistry: { adaptersByKind: new Map([[adapter.kind, adapter]]) }, + }) as never; + vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(bundle); + const ctx = new ContextContainer(); + ctx.set(AuthKey, null); + ctx.set(BundleKey, bundle); + ctx.set(ChannelKey, adapter); + ctx.set(ContinuationTokenKey, "http:thread-context"); + ctx.set(ModeKey, "conversation"); + ctx.set(SessionIdKey, "session-1"); + const session = createStubSession(); + installSessionStoreMocks([session]); + vi.mocked(createExecutionNodeStep).mockImplementation(() => { + return async (stepSession): Promise => ({ + next: async () => ({ next: null, session: stepSession }), + session: stepSession, + }); + }); + + await turnStep({ + input: { kind: "deliver", payloads: [{ message: "request" }] }, + parentWritable: createTestWritable(), + serializedContext: serializeContext(ctx), + sessionState: createStubSessionState(), + }); + + expect(createAdapterContext).toHaveBeenCalledTimes(3); + }); + + it("keeps completed agent-loop checkpoints when a later step is cancelled", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; + vi.mocked(getCompiledRuntimeAgentBundle).mockResolvedValue(createTestCompiledBundle()); + const initial = createStubSession(); + const checkpoint = createStubSession({ + history: [ + { content: "request", role: "user" }, + { content: "completed step", role: "assistant" }, + ], + }); + installSessionStoreMocks([initial]); + let stepCount = 0; + vi.mocked(createExecutionNodeStep).mockImplementation(() => { + return async (): Promise => { + stepCount += 1; + if (stepCount === 1) { + loadContext().set(ThreadKey, "completed checkpoint"); + loadContext().set(NestedCheckpointKey, { nested: { value: "completed checkpoint" } }); + return { next: async () => ({ next: null, session: checkpoint }), session: checkpoint }; + } + loadContext().set(ThreadKey, "interrupted mutation"); + loadContext().require(NestedCheckpointKey).nested.value = "interrupted mutation"; + throw new TurnCancelledError(); + }; + }); + + const result = await turnStep({ + input: { kind: "deliver", payloads: [{ message: "request" }] }, + parentWritable: createTestWritable(), + serializedContext: createSerializedContext(), + sessionState: createStubSessionState(), + }); + + expect(result.action).toBe("cancelled"); + expect(result.serializedContext[ThreadKey.name]).toBe("completed checkpoint"); + expect(result.serializedContext[NestedCheckpointKey.name]).toEqual({ + nested: { value: "completed checkpoint" }, + }); + expect(result.sessionState.snapshot?.session.history).toEqual([ + { content: "request", role: "user" }, + { content: "completed step", role: "assistant" }, + ]); + }); + it("keeps a session-scoped dynamic model selection when the first turn is cancelled", async () => { const session = createStubSession(); installSessionStoreMocks([session]); @@ -2457,15 +2670,14 @@ describe("turnStep", () => { }); it("projects a requested sleep onto the durable step result", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; const session = createStubSession(); + const next = vi.fn(async () => ({ next: null, session }) as const); installSessionStoreMocks([session]); vi.mocked(createExecutionNodeStep).mockImplementation(() => { return async (stepSession): Promise => { requestTurnSleep(2_500); - return { - next: async () => ({ next: null, session: stepSession }), - session: stepSession, - }; + return { next, session: stepSession }; }; }); @@ -2483,9 +2695,57 @@ describe("turnStep", () => { action: "continue", sleepDurationMs: 2_500, }); + expect(next).not.toHaveBeenCalled(); expect(result.serializedContext).not.toHaveProperty("eve.pendingTurnSleepDuration"); }); + it("returns after a step starts background tasks", async () => { + mockAgentLoopConfig.stepsPerWorkflowStep = 3; + const session = createStubSession(); + const next = vi.fn(async () => ({ next: null, session }) as const); + const taskSession = createStubSession({ continuationToken: "background-checkpoint" }); + installSessionStoreMocks([session]); + const sessionModel = { id: "anthropic/claude-opus-4.6", contextWindowTokens: 1_000_000 }; + vi.mocked(createExecutionNodeStep).mockImplementation(() => { + return async (stepSession): Promise => { + loadContext().set(SessionDynamicModelReferenceKey, sessionModel); + return { + backgroundTaskSession: taskSession, + backgroundTasks: [ + { taskId: "task-1", taskInboxToken: "task-inbox-1", taskRunId: "task-run-1" }, + ], + next, + session: stepSession, + }; + }; + }); + + const result = await turnStep({ + input: { kind: "deliver", payloads: [{ message: "start work" }] }, + parentWritable: createTestWritable(), + serializedContext: createSerializedContext(), + sessionState: createStubSessionState(), + }); + + expect(next).not.toHaveBeenCalled(); + expect(result).toMatchObject({ + action: "continue", + commitBarrier: { + effect: { + kind: "release-background-tasks", + tasks: [{ taskId: "task-1" }], + }, + }, + }); + expect(result.commitBarrier?.transition.sessionState.snapshot?.session.continuationToken).toBe( + "background-checkpoint", + ); + expect(result.commitBarrier?.transition.serializedContext).toMatchObject({ + [SessionDynamicModelReferenceKey.name]: sessionModel, + }); + expect(result.cancellationTransition).toEqual(result.commitBarrier?.transition); + }); + it("persists onDeliver context into the next durable step", async () => { const seenMessages: string[] = []; const session = createStubSession(); diff --git a/packages/eve/src/execution/workflow-steps.ts b/packages/eve/src/execution/workflow-steps.ts index 91a766ef4c..da74252623 100644 --- a/packages/eve/src/execution/workflow-steps.ts +++ b/packages/eve/src/execution/workflow-steps.ts @@ -1,4 +1,8 @@ import { buildAdapterContext } from "#channel/adapter-context.js"; +import { AgentLoopBatch, preserveCancelledTurnMessage } from "#execution/agent-loop-batch.js"; +import { resumeAgentLoopCheckpoint } from "#execution/agent-loop-checkpoint.js"; +import { AGENT_LOOP_STEPS_PER_WORKFLOW_STEP } from "#execution/agent-loop-config.js"; +import { finalizeTurnStepResult } from "#execution/finalize-turn-step-result.js"; import { callAdapterEventHandler, defaultDeliverResult } from "#channel/adapter.js"; import type { DeliverHookPayload } from "#channel/types.js"; import { contextStorage } from "#context/container.js"; @@ -9,7 +13,7 @@ import { prepareDynamicInstructionPreamble, } from "#context/dynamic-instruction-lifecycle.js"; import { dispatchDynamicModelEvent } from "#context/dynamic-model-lifecycle.js"; -import { preserveSerializedSessionDynamicModelSelection } from "#context/serialized-dynamic-model-selection.js"; +import { preserveSerializedSessionPreambleState } from "#context/serialized-session-preamble-state.js"; import { dispatchDynamicSkillEvent } from "#context/dynamic-skill-lifecycle.js"; import { dispatchDynamicSubagentEvent, @@ -22,7 +26,6 @@ import { import { AuthKey, CapabilitiesKey, - HandleEventKey, ModeKey, SessionDynamicSubagentRuntimeRevisionKey, SessionDynamicToolRuntimeRevisionKey, @@ -50,16 +53,9 @@ import { matchAuthorizationCallbacks } from "#execution/authorization-callback-m import { readTurnSleepDurationMs } from "#harness/turn-sleep.js"; import { isTurnCancellation, throwIfTurnAborted } from "#harness/turn-cancellation.js"; import { setChannelContext } from "#execution/channel-context.js"; -import { hasPendingInputBatch } from "#harness/input-requests.js"; import { activeTurnId } from "#harness/active-turn-id.js"; -import { coalesceTurnInputs, normalizeUserContent } from "#harness/messages.js"; -import { - getRuntimeActionKeysFromWorkflowInterrupt, - isWorkflowRuntimeActionInterrupt, -} from "#harness/workflow-runtime-action-state.js"; -import { getPendingWorkflowInterrupt } from "#harness/workflow-interrupt-state.js"; +import { coalesceTurnInputs } from "#harness/messages.js"; import type { HarnessSession, StepInput, StepResult } from "#harness/types.js"; -import { getTurnUsageState, takeSessionUsageDelta, toUsage } from "#harness/turn-tag-state.js"; import type { DurableStepResult } from "#execution/next-driver-action.js"; import { derivePendingState } from "#execution/pending-turn-state.js"; import { @@ -87,10 +83,7 @@ import { resolveInitiatingTaskContext, resolveTaskDeliveryContext, } from "#tasks/delivery-context.js"; -import { - readRetainedBackgroundToolResult, - runBackgroundStep, -} from "#execution/tasks/parent/tool-execution.js"; +import { readRetainedBackgroundToolResult } from "#execution/tasks/parent/tool-execution.js"; import { isTaskOwnedSerializedContext, TASK_UPDATE_SESSION_INSTRUCTION, @@ -104,10 +97,6 @@ import { createExecutionHistoryView } from "#execution/history-view.js"; import { resolveRuntimeCompiledArtifactsVersionedCacheKey } from "#runtime/cache-key.js"; import { createWorkflowRuntime } from "#execution/workflow-runtime.js"; import { isTaskToolAvailable, TASK_UPDATE_TOOL_NAME } from "#runtime/framework-tools/tasks.js"; -import { stageAttachmentsToSandbox } from "#harness/attachment-staging.js"; - -const TASK_DONE_WITH_PENDING_INPUT_ERROR_MESSAGE = - "Task mode cannot complete while input requests remain pending."; export type { TurnStepInput }; @@ -117,15 +106,19 @@ export type { TurnStepInput }; export async function turnStep(rawInput: TurnStepInput): Promise { "use step"; - let input = rawInput; - + const resumed = await resumeAgentLoopCheckpoint({ + enabled: AGENT_LOOP_STEPS_PER_WORKFLOW_STEP > 1, + rawInput, + }); + const { checkpoint } = resumed; + let input = resumed.stepInput; let durableSession = await readDurableSession(input.sessionState); const ctx = await deserializeContext(input.serializedContext); - if (rawInput.input?.kind === "deliver") { + if (checkpoint === undefined && rawInput.input?.kind === "deliver") { ctx.set(TurnTaskDeliveryKey, "none"); ctx.delete(TurnTaskStateKey); } - const adapter = ctx.require(ChannelKey); + let adapter = ctx.require(ChannelKey); const bundle = ctx.require(BundleKey); const tasksEnabled = bundle.resolvedAgent.config?.experimental?.tasks === true; ctx.set(TasksEnabledKey, tasksEnabled); @@ -196,8 +189,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise instrumentChannelDelivery({ agentName: bundle.turnAgent.id, @@ -225,8 +217,11 @@ export async function turnStep(rawInput: TurnStepInput): Promise { + adapter = ctx.require(ChannelKey); + adapterCtx = buildAdapterContext(adapter, ctx); + }; // Run the adapter's deliver hook for each queued payload and // coalesce the resulting StepInput values. let resolved: StepInput | undefined; @@ -254,7 +249,6 @@ export async function turnStep(rawInput: TurnStepInput): Promise { - ctx.setVirtualContext(HandleEventKey, handleEvent); + const capabilities = ctx.get(CapabilitiesKey); + const runHarnessStep = async ( + lifecycleSession: HarnessSession, + stepInput: StepInput | undefined, + applyCommand: boolean, + ): Promise => { + const refreshedSession = refreshSessionFromTurnAgent({ + compactionOverrides: { + thresholdPercent: effectiveAgent.thresholdPercent, + }, + session: lifecycleSession, + systemPromptAdditions: taskUpdatesEnabled ? [TASK_UPDATE_SESSION_INSTRUCTION] : undefined, + turnAgent: effectiveAgent.turnAgent, + }); + const modelSession = tasksEnabled + ? await appendTaskAgentAnnouncement(refreshedSession, history.messages(refreshedSession)) + : refreshedSession; + const step = createExecutionNodeStep({ + abortSignal: input.abortSignal, + capabilities, + clearOnly: applyCommand && input.input?.kind === "clear", + compactOnly: applyCommand && input.input?.kind === "compact", + createRuntime: createWorkflowRuntime, + handleEvent, + historyProjector: history.projector, + historyView: history.prepare(modelSession), + mode, + modelResolutionScope: { + moduleMap: bundle.moduleMap, + nodeId: bundle.nodeId, + }, + node: effectiveNode, + workflowMaxSubagents: refreshedSession.workflowMaxSubagents, + }); + return step(modelSession, stepInput); + }; + + stepResult = await batch.run(initialSession, handleEvent, async (enrichedSession) => { let schemaSession = resolveEffectiveOutputSchema({ agentOutputSchema: effectiveAgent.turnAgent.outputSchema, input: resolved, @@ -487,46 +522,21 @@ export async function turnStep(rawInput: TurnStepInput): Promise => { - const refreshedSession = refreshSessionFromTurnAgent({ - compactionOverrides: { - thresholdPercent: effectiveAgent.thresholdPercent, - }, - session: lifecycleSession, - systemPromptAdditions: taskUpdatesEnabled ? [TASK_UPDATE_SESSION_INSTRUCTION] : undefined, - turnAgent: effectiveAgent.turnAgent, - }); - const modelSession = tasksEnabled - ? await appendTaskAgentAnnouncement(refreshedSession, history.messages(refreshedSession)) - : refreshedSession; - - const step = createExecutionNodeStep({ - abortSignal: input.abortSignal, - capabilities, - clearOnly: input.input?.kind === "clear", - compactOnly: input.input?.kind === "compact", - createRuntime: createWorkflowRuntime, - handleEvent, - historyProjector: history.projector, - historyView: history.prepare(modelSession), - mode, - modelResolutionScope: { - moduleMap: bundle.moduleMap, - nodeId: bundle.nodeId, - }, - node: effectiveNode, - workflowMaxSubagents: refreshedSession.workflowMaxSubagents, - }); - return step(modelSession, stepInput); - }; - - return runHarnessStep(schemaSession, resolved); + return runHarnessStep(schemaSession, resolved, true); }); + + while ( + batch.completedSteps < AGENT_LOOP_STEPS_PER_WORKFLOW_STEP && + typeof stepResult.next === "function" && + readTurnSleepDurationMs(ctx) === undefined && + stepResult.backgroundTasks === undefined + ) { + await batch.commitContinuation(); + stepResult = await batch.run(stepResult.session, handleEvent, (session) => { + refreshAdapterContext(); + return runHarnessStep(session, undefined, false); + }); + } } catch (error) { if (!isTurnCancellation(error)) { await failChannelDeliveries(error); @@ -539,142 +549,43 @@ export async function turnStep(rawInput: TurnStepInput): Promise { - const message = normalizeUserContent(input?.message); - if (message === undefined) return session; - const content = await stageAttachmentsToSandbox(message); - return { ...session, history: [...session.history, { content, role: "user" }] }; + return finalizeTurnStepResult({ + batch, + ctx, + mode, + stepResult, + tasksEnabled, + writer, + }); } diff --git a/packages/eve/src/internal/testing/durable-session-workflow.ts b/packages/eve/src/internal/testing/durable-session-workflow.ts index 6db0c47286..e41a585306 100644 --- a/packages/eve/src/internal/testing/durable-session-workflow.ts +++ b/packages/eve/src/internal/testing/durable-session-workflow.ts @@ -12,6 +12,10 @@ import { type DurableSessionState, readDurableSession, } from "#execution/durable-session-store.js"; +import { + readAgentLoopCheckpoint, + writeAgentLoopCheckpoint, +} from "#execution/agent-loop-checkpoint.js"; import type { HarnessSession } from "#harness/types.js"; /** Synthetic minimal session for storage-layer round-trips. */ @@ -89,6 +93,58 @@ export async function durableSessionWriteWithRetryStep(input: { return { attempt: meta.attempt, sessionState }; } +export async function agentLoopCheckpointRetryStep(input: { + readonly sessionState: DurableSessionState; + readonly writeBeforeFailure: boolean; +}): Promise<{ + readonly attempt: number; + readonly completedSteps: number; + readonly resumed: boolean; +}> { + "use step"; + + const { attempt } = getStepMetadata(); + let checkpoint = await readAgentLoopCheckpoint(); + const resumed = checkpoint !== undefined; + if (attempt === 1) { + if (input.writeBeforeFailure) { + await writeAgentLoopCheckpoint({ + completedSteps: 3, + serializedContext: { marker: "checkpoint" }, + sessionState: input.sessionState, + }); + } + throw new Error("agent-loop-checkpoint: intentional retry"); + } + checkpoint ??= await writeAgentLoopCheckpoint({ + completedSteps: 0, + serializedContext: { marker: "seed" }, + sessionState: input.sessionState, + }); + return { attempt, completedSteps: checkpoint.completedSteps, resumed }; +} + +export async function agentLoopCheckpointRetryFixtureWorkflow(input: { + readonly writeBeforeFailure: boolean; +}): Promise<{ + readonly attempt: number; + readonly completedSteps: number; + readonly resumed: boolean; +}> { + "use workflow"; + + const { workflowRunId: sessionId } = getWorkflowMetadata(); + const sessionState = await durableSessionWriteStep({ + historyDepth: 0, + marker: "seed", + sessionId, + }); + return agentLoopCheckpointRetryStep({ + sessionState, + writeBeforeFailure: input.writeBeforeFailure, + }); +} + /** Reads the latest snapshot and projects the fields the test asserts on. */ export async function durableSessionReadStep(input: { readonly sessionState: DurableSessionState;