refactor(eve): configure agent steps per workflow step

.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.

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);
+  });
+});

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<string, unknown>,
+  interrupted: Record<string, unknown>,
+): Record<string, unknown> {
+  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;
+}

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<HarnessSession> {
+  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<string, unknown>;
+  private latestSession: HarnessSession;
+  private checkpoint: AgentLoopCheckpoint | undefined;
+  completedSteps: number;
+
+  constructor(
+    abortSignal: AbortSignal | undefined,
+    ctx: ContextContainer,
+    initialSession: HarnessSession,
+    initialSessionState: AgentLoopCheckpoint["sessionState"],
+    initialSerializedContext: Record<string, unknown>,
+    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<string, unknown> {
+    return this.checkpoint?.serializedContext ?? this.initialSerializedContext;
+  }
+
+  async commitContinuation(): Promise<void> {
+    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<StepResult>,
+  ): Promise<StepResult> {
+    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;
+  }
+}

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<string, unknown>;
+  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<AgentLoopCheckpoint | undefined> {
+  const { attempt, stepId } = getStepMetadata();
+  if (attempt === 1) return undefined;
+  return readCheckpointTail(checkpointNamespace(stepId));
+}
+
+export async function writeAgentLoopCheckpoint(
+  checkpoint: Omit<AgentLoopCheckpoint, "version">,
+): Promise<AgentLoopCheckpoint> {
+  const namespace = checkpointNamespace(getStepMetadata().stepId);
+  const writer = getWritable<AgentLoopCheckpoint>({ 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<AgentLoopCheckpoint | undefined> {
+  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<unknown>(runId);
+  const tail = run.getReadable<AgentLoopCheckpoint>({ namespace });
+  if ((await tail.getTailIndex()) === -1) return undefined;
+  const reader = run.getReadable<unknown>({ 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}`;
+}

packages/eve/src/execution/agent-loop-config.ts

@@ -0,0 +1 @@
+export const AGENT_LOOP_STEPS_PER_WORKFLOW_STEP = 1;

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" } });