Merge pull request #341 from pikasTech/fix/340-active-steer-resume

修复 backend resume 丢失 durable steer

src/runner/manager-api.ts

@@ -68,7 +68,8 @@ export class RunnerManagerApi {
     if (options.afterSeq !== undefined) listOptions.afterSeq = options.afterSeq;
     if (options.limit !== undefined) listOptions.limit = options.limit;
     const items = await this.listCommands(runId, listOptions);
-    const selected = options.commandId ? items.find((item) => item.id === options.commandId && item.state === "pending" && item.type === "turn") ?? null : items.find((item) => item.state === "pending" && item.type === "turn") ?? null;
+    const receivableTurn = (item: CommandRecord): boolean => item.type === "turn" && (item.state === "pending" || item.state === "acknowledged");
+    const selected = options.commandId ? items.find((item) => item.id === options.commandId && receivableTurn(item)) ?? null : items.find((item) => item.state === "pending" && item.type === "turn") ?? items.find((item) => item.state === "acknowledged" && item.type === "turn") ?? null;
     return { items, selected };
   }
 

src/runner/run-once.ts

@@ -3,7 +3,7 @@ import path from "node:path";
 import { RunnerManagerApi, failureKindFromError, terminalStatusForFailure, errorMessage } from "./manager-api.js";
 import { createBackendSession, runBackendTurn, type BackendActiveTurnControl, type BackendAdapterOptions, type BackendSession } from "../backend/adapter.js";
 import { materializeResourceBundle } from "./resource-bundle.js";
-import type { BackendEvent, BackendProfile, CommandRecord, FailureKind, InitialPromptAssembly, JsonRecord, RunRecord, RunnerRecord, TerminalStatus } from "../common/types.js";
+import type { BackendEvent, BackendProfile, BackendTurnResult, CommandRecord, FailureKind, InitialPromptAssembly, JsonRecord, RunRecord, RunnerRecord, TerminalStatus } from "../common/types.js";
 import { AgentRunError } from "../common/errors.js";
 import { smokeBundledWorkReadyCapabilities, smokeImageWorkReadyCapabilities } from "../common/work-ready.js";
 import { stableHash } from "../common/validation.js";
@@ -254,6 +254,7 @@ function pathDelimiter(): string {
 }
 
 async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions, command: CommandRecord, runner: RunnerRecord, attemptId: string, workspacePath: string | undefined, backendSession: BackendSession | null, runnerLog: RunnerLogSink): Promise<CommandExecutionResult> {
+  const resumeDurableTurn = command.state === "acknowledged";
   await api.ackCommand(command.id);
   const acked = await api.getCommand(options.runId, command.id);
   if (acked.state === "cancelled") return await reportCancelled(api, options.runId, command.id, runner, attemptId, "command cancelled before backend start", runnerLog, options.runnerJobId);
@@ -274,6 +275,10 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
     backendStarted = true;
     await runnerLog.write("command.started", { runId: options.runId, commandId: command.id, runnerId: runner.id, runnerJobId: options.runnerJobId ?? null, attemptId, backendProfile: latestRun.backendProfile, valuesPrinted: false });
     stopCancelWatch = watchCancellation(api, options.runId, command.id, abortController, { attemptId, runnerId: runner.id, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}), runnerLog });
+    let backendAttempt = 1;
+    let durableReplayCommands: CommandRecord[] = [];
+    let activeTurnRegistered = false;
+    let replayOutcome: Promise<unknown | null> = Promise.resolve(null);
     const backendOptions = {
       ...options,
       env: preparedCredential.env,
@@ -299,7 +304,10 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
         await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
       },
       onActiveTurn: (control: BackendActiveTurnControl) => {
-        stopSteerWatch = startActiveTurnCommandWatch(api, options.runId, command.id, attemptId, runner.id, control, options.pollIntervalMs, runnerLog, options.runnerJobId);
+        activeTurnRegistered = true;
+        const watcher = startActiveTurnCommandWatch(api, options.runId, command, attemptId, runner.id, control, options.pollIntervalMs, runnerLog, options.runnerJobId, backendAttempt, resumeDurableTurn, durableReplayCommands);
+        stopSteerWatch = watcher.stop;
+        replayOutcome = watcher.replayBarrier.then(() => null, (error) => error);
         return () => {
           stopSteerWatch?.();
           stopSteerWatch = undefined;
@@ -307,8 +315,23 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
       },
     };
     const retryPolicy = backendRetryPolicy(latestRun.executionPolicy, preparedCredential.env, options);
-    let backendAttempt = 1;
-    let result = backendSession ? await backendSession.runTurn(latestRun, command, backendOptions) : await runBackendTurn(latestRun, command, backendOptions);
+    const runBackendAttempt = async (): Promise<BackendTurnResult> => {
+      const replayRequired = resumeDurableTurn || backendAttempt > 1;
+      durableReplayCommands = replayRequired ? await durableSteersForTurnResume(api, options.runId, command.seq) : [];
+      activeTurnRegistered = false;
+      replayOutcome = Promise.resolve(null);
+      const result = backendSession ? await backendSession.runTurn(latestRun, command, backendOptions) : await runBackendTurn(latestRun, command, backendOptions);
+      if (result.terminalStatus !== "completed" || durableReplayCommands.length === 0) return result;
+      const replayError = activeTurnRegistered
+        ? await replayOutcome
+        : new AgentRunError("backend-protocol-error", "backend completed before durable steer replay control became available", { details: { backendAttempt, replayCommandCount: durableReplayCommands.length, valuesPrinted: false } });
+      if (!replayError) return result;
+      const failureKind = failureKindFromError(replayError);
+      const message = errorMessage(replayError);
+      await appendBestEffort(api, options.runId, { type: "error", payload: { phase: "turn/steer:replay-barrier-failed", commandId: command.id, failureKind, message, backendAttempt, replayCommandCount: durableReplayCommands.length, replayCommandIds: durableReplayCommands.map((item) => item.id), resumeDurableTurn, attemptId, runnerId: runner.id, valuesPrinted: false } });
+      return { terminalStatus: "failed", failureKind, failureMessage: message, events: [], ...(result.threadId ? { threadId: result.threadId } : {}), ...(result.turnId ? { turnId: result.turnId } : {}) };
+    };
+    let result = await runBackendAttempt();
     while (shouldRetryBackendTurn(result, retryPolicy, backendAttempt)) {
       const retryDelayMs = backendRetryDelayMs(retryPolicy, backendAttempt);
       const retryPayload = {
@@ -336,7 +359,7 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
       }
       await sleep(retryDelayMs);
       backendAttempt += 1;
-      result = backendSession ? await backendSession.runTurn(latestRun, command, backendOptions) : await runBackendTurn(latestRun, command, backendOptions);
+      result = await runBackendAttempt();
     }
     for (const event of result.events) {
       if (shouldKeepTerminalOutboxEvent(event)) terminalOutboxEvents.push(event);
@@ -404,21 +427,36 @@ function positiveInteger(value: unknown, fallback: number): number {
   return Math.max(1, Math.floor(parsed));
 }
 
-function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, pollIntervalMs: number | undefined, runnerLog: RunnerLogSink, runnerJobId: string | undefined): () => void {
+function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targetCommand: CommandRecord, attemptId: string, runnerId: string, control: BackendActiveTurnControl, pollIntervalMs: number | undefined, runnerLog: RunnerLogSink, runnerJobId: string | undefined, backendAttempt: number, resumeDurableTurn: boolean, durableReplayCommands: CommandRecord[]): { stop: () => void; replayBarrier: Promise<void> } {
   let stopped = false;
   let polling = false;
+  let replayFailed = false;
   const seen = new Set<string>();
+  const replayBarrier = (async (): Promise<void> => {
+    try {
+      for (const command of durableReplayCommands) {
+        await replayDurableSteerCommand(api, runId, command, targetCommand.id, attemptId, runnerId, control, backendAttempt, resumeDurableTurn, runnerLog, runnerJobId);
+        seen.add(command.id);
+      }
+    } catch (error) {
+      replayFailed = true;
+      throw error;
+    }
+  })();
+  void replayBarrier.catch(() => undefined);
   const intervalMs = normalizePollIntervalMs(pollIntervalMs);
   const poll = async (): Promise<void> => {
-    if (stopped || polling) return;
+    if (stopped || polling || replayFailed) return;
     polling = true;
     try {
-      const commands = await api.listCommands(runId, { afterSeq: 0, limit: 100 });
-      const pendingControlCommands = commands.filter((item) => (item.type === "steer" || item.type === "interrupt") && item.state === "pending" && !seen.has(item.id));
-      for (const controlCommand of pendingControlCommands) {
+      await replayBarrier;
+      if (stopped) return;
+      const commands = await listCurrentTurnControlCommands(api, runId, targetCommand.seq);
+      const controlCommands = commands.filter((item) => !seen.has(item.id) && (item.type === "steer" || item.type === "interrupt") && item.state === "pending");
+      for (const controlCommand of controlCommands) {
         seen.add(controlCommand.id);
-        if (controlCommand.type === "interrupt") await handleInterruptCommand(api, runId, controlCommand, targetCommandId, attemptId, runnerId, control, runnerLog, runnerJobId);
-        else await handleSteerCommand(api, runId, controlCommand, targetCommandId, attemptId, runnerId, control, runnerLog, runnerJobId);
+        if (controlCommand.type === "interrupt") await handleInterruptCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
+        else await handleSteerCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
       }
     } catch {
       // The active backend turn remains authoritative; missed control commands stay pending for the next poll.
@@ -428,12 +466,50 @@ function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targe
   };
   const timer = setInterval(() => { void poll(); }, intervalMs);
   void poll();
-  return () => {
-    stopped = true;
-    clearInterval(timer);
+  return {
+    stop: () => {
+      stopped = true;
+      clearInterval(timer);
+    },
+    replayBarrier,
   };
 }
 
+async function listCurrentTurnControlCommands(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise<CommandRecord[]> {
+  const commands: CommandRecord[] = [];
+  let afterSeq = targetCommandSeq;
+  while (true) {
+    const batch = await api.listCommands(runId, { afterSeq, limit: 100 });
+    if (batch.length === 0) return commands;
+    for (const command of batch) {
+      if (command.type === "turn") return commands;
+      commands.push(command);
+    }
+    if (batch.length < 100) return commands;
+    afterSeq = batch[batch.length - 1]?.seq ?? afterSeq;
+  }
+}
+
+async function durableSteersForTurnResume(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise<CommandRecord[]> {
+  const commands = await listCurrentTurnControlCommands(api, runId, targetCommandSeq);
+  return commands.filter((command) => command.type === "steer" && (command.state === "acknowledged" || command.state === "completed"));
+}
+
+async function replayDurableSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, backendAttempt: number, resumeDurableTurn: boolean, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
+  const prompt = steerPrompt(command.payload);
+  if (!prompt) return;
+  try {
+    await control.steer(prompt);
+    await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "turn/steer:replayed", commandId: command.id, commandType: "steer", commandSeq: command.seq, commandState: command.state, targetCommandId, deliveryState: "replayed-to-resumed-backend", backendAccepted: true, backendAttempt, resumeDurableTurn, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
+    if (command.state === "acknowledged") {
+      await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: control.threadId, turnId: control.turnId }, phase: "runner:steer:replayed", runnerLog });
+    }
+  } catch (error) {
+    await appendBestEffort(api, runId, { type: "error", payload: { phase: "turn/steer:replay-failed", commandId: command.id, commandType: "steer", commandSeq: command.seq, commandState: command.state, targetCommandId, failureKind: failureKindFromError(error), message: errorMessage(error), backendAttempt, resumeDurableTurn, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
+    throw error;
+  }
+}
+
 async function handleSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
   const acked = await api.ackCommand(command.id);
   if (acked.state === "cancelled") {

src/selftest/cases/50-hwlab-manual-dispatch.ts

@@ -301,6 +301,112 @@ process.exit(1);
     assert.ok(steerEvents.some((event) => event.type === "user_message" && event.payload?.commandId === steerCommand.id && event.payload?.traceId === "trc_hwlab_command_steer" && event.payload?.targetTraceId === "trc_hwlab_command_steer_turn"));
     assert.ok(steerEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "steer-command-acknowledged" && event.payload?.commandId === steerCommand.id && event.payload?.targetCommandId === steerRun.commandId));
     assert.ok(steerEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:completed" && event.payload?.commandId === steerCommand.id && event.payload?.targetCommandId === steerRun.commandId && event.payload.deliveryState === "forwarded-to-backend" && event.payload.targetEffect === "not-guaranteed"));
+    assert.equal(steerEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed"), false, "an initial turn must not replay a completed steer");
+
+    const retryBootstrap = await createHwlabRun(client, context, bundle, "hwlab-session-steer-retry", "establish resumable thread", "trc_hwlab_command_steer_retry_bootstrap");
+    const retryBootstrapResult = await runOnce({ managerUrl: server.baseUrl, runId: retryBootstrap.runId, commandId: retryBootstrap.commandId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer-retry") }, oneShot: true });
+    assert.equal(retryBootstrapResult.terminalStatus, "completed");
+    const retryRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer-retry", "start retryable active turn", "trc_hwlab_command_steer_retry_turn", 500);
+    const acknowledgedRetrySteer = await client.post(`/api/v1/runs/${retryRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_RETRY_ACKNOWLEDGED", traceId: "trc_hwlab_command_steer_retry_acknowledged", targetTraceId: "trc_hwlab_command_steer_retry_turn" }, idempotencyKey: "trc_hwlab_command_steer_retry_acknowledged" }) as { id: string };
+    await client.post(`/api/v1/commands/${acknowledgedRetrySteer.id}/ack`, {});
+    const retryStartFile = path.join(context.tmp, "fake-codex-starts-steer-retry.txt");
+    const retrySteerFile = path.join(context.tmp, "fake-codex-steers-retry.txt");
+    const retryRunner = runOnce({
+      managerUrl: server.baseUrl,
+      runId: retryRun.runId,
+      commandId: retryRun.commandId,
+      codexCommand: context.fakeCodexCommand,
+      codexArgs: context.fakeCodexArgs,
+      codexHome: context.codexHome,
+      env: {
+        CODEX_HOME: context.codexHome,
+        AGENTRUN_FAKE_CODEX_MODE: "steer-retry-resume",
+        AGENTRUN_FAKE_CODEX_START_FILE: retryStartFile,
+        AGENTRUN_FAKE_CODEX_STEER_FILE: retrySteerFile,
+        AGENTRUN_FAKE_CODEX_EXPECTED_STEERS: "3",
+        AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer-retry"),
+      },
+      backendRetryMaxAttempts: 2,
+      backendRetryInitialBackoffMs: 100,
+      backendRetryMaxBackoffMs: 100,
+      oneShot: true,
+      pollIntervalMs: 25,
+    });
+    await waitForCommandState(client, retryRun.runId, retryRun.commandId, "acknowledged");
+    const firstRetrySteer = await client.post(`/api/v1/runs/${retryRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_RETRY_FIRST", traceId: "trc_hwlab_command_steer_retry_first", targetTraceId: "trc_hwlab_command_steer_retry_turn" }, idempotencyKey: "trc_hwlab_command_steer_retry_first" }) as { id: string };
+    const secondRetrySteer = await client.post(`/api/v1/runs/${retryRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_RETRY_SECOND", traceId: "trc_hwlab_command_steer_retry_second", targetTraceId: "trc_hwlab_command_steer_retry_turn" }, idempotencyKey: "trc_hwlab_command_steer_retry_second" }) as { id: string };
+    await waitForCommandState(client, retryRun.runId, firstRetrySteer.id, "completed");
+    await waitForCommandState(client, retryRun.runId, secondRetrySteer.id, "completed");
+    const retryRunnerResult = await retryRunner as JsonRecord;
+    assert.equal(retryRunnerResult.terminalStatus, "completed");
+    const retrySteerDeliveries = (await readFile(retrySteerFile, "utf8")).trim().split("\n").filter(Boolean).map((line) => JSON.parse(line) as { attempt: number; text: string });
+    assert.deepEqual(retrySteerDeliveries, [
+      { attempt: 1, text: "STEER_RETRY_FIRST" },
+      { attempt: 1, text: "STEER_RETRY_SECOND" },
+      { attempt: 2, text: "STEER_RETRY_ACKNOWLEDGED" },
+      { attempt: 2, text: "STEER_RETRY_FIRST" },
+      { attempt: 2, text: "STEER_RETRY_SECOND" },
+    ], "durable steers must be replayed exactly once per resumed backend attempt and keep command seq order");
+    const retryEventsResponse = await client.get(`/api/v1/runs/${retryRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
+    const retryEvents = retryEventsResponse.items ?? [];
+    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "runner:backend-retry").length, 1);
+    assert.equal(retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "thread/resume:completed").length, 2);
+    const replayedSteers = retryEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed");
+    assert.deepEqual(replayedSteers.map((event) => event.payload?.commandId), [acknowledgedRetrySteer.id, firstRetrySteer.id, secondRetrySteer.id]);
+    assert.deepEqual(replayedSteers.map((event) => event.payload?.commandState), ["acknowledged", "completed", "completed"]);
+    assert.deepEqual(replayedSteers.map((event) => event.payload?.backendAttempt), [2, 2, 2]);
+    await waitForCommandState(client, retryRun.runId, acknowledgedRetrySteer.id, "completed");
+    const retryEnvelope = await client.get(`/api/v1/runs/${retryRun.runId}/commands/${retryRun.commandId}/result`) as JsonRecord;
+    assert.equal(retryEnvelope.terminalStatus, "completed");
+    assert.equal(retryEnvelope.reply, "replayed:STEER_RETRY_ACKNOWLEDGED|STEER_RETRY_FIRST|STEER_RETRY_SECOND");
+
+    const fastTerminalRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer-retry", "fast terminal must wait for replay barrier", "trc_hwlab_command_steer_fast_terminal", 500);
+    const fastTerminalSteer = await client.post(`/api/v1/runs/${fastTerminalRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_FAST_TERMINAL", traceId: "trc_hwlab_command_steer_fast_terminal_control" }, idempotencyKey: "trc_hwlab_command_steer_fast_terminal_control" }) as { id: string };
+    await client.post(`/api/v1/commands/${fastTerminalSteer.id}/ack`, {});
+    const fastTerminalResult = await runOnce({ managerUrl: server.baseUrl, runId: fastTerminalRun.runId, commandId: fastTerminalRun.commandId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "steer-retry-fast-terminal", AGENTRUN_FAKE_CODEX_START_FILE: path.join(context.tmp, "fake-codex-starts-steer-fast-terminal.txt"), AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer-retry") }, backendRetryMaxAttempts: 2, backendRetryInitialBackoffMs: 100, backendRetryMaxBackoffMs: 100, oneShot: true, pollIntervalMs: 25 }) as JsonRecord;
+    assert.equal(fastTerminalResult.terminalStatus, "failed", "backend success before replay barrier must not terminalize the command as completed");
+    const fastTerminalEventsResponse = await client.get(`/api/v1/runs/${fastTerminalRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
+    const fastTerminalEvents = fastTerminalEventsResponse.items ?? [];
+    assert.ok(fastTerminalEvents.some((event) => event.type === "error" && event.payload?.phase === "turn/steer:replay-failed" && event.payload?.backendAttempt === 2));
+    assert.ok(fastTerminalEvents.some((event) => event.type === "error" && event.payload?.phase === "turn/steer:replay-barrier-failed" && event.payload?.backendAttempt === 2));
+    assert.equal(fastTerminalEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed"), false);
+
+    const replayRpcRetryRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer-retry", "retry replay after first RPC failure", "trc_hwlab_command_steer_rpc_retry", 500);
+    const replayRpcRetrySteer = await client.post(`/api/v1/runs/${replayRpcRetryRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_RPC_RETRY", traceId: "trc_hwlab_command_steer_rpc_retry_control" }, idempotencyKey: "trc_hwlab_command_steer_rpc_retry_control" }) as { id: string };
+    await client.post(`/api/v1/commands/${replayRpcRetrySteer.id}/ack`, {});
+    const replayRpcRetrySteerFile = path.join(context.tmp, "fake-codex-steers-rpc-retry.txt");
+    const replayRpcRetryResult = await runOnce({ managerUrl: server.baseUrl, runId: replayRpcRetryRun.runId, commandId: replayRpcRetryRun.commandId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "steer-retry-rpc-fails-once", AGENTRUN_FAKE_CODEX_START_FILE: path.join(context.tmp, "fake-codex-starts-steer-rpc-retry.txt"), AGENTRUN_FAKE_CODEX_STEER_FILE: replayRpcRetrySteerFile, AGENTRUN_FAKE_CODEX_EXPECTED_STEERS: "1", AGENTRUN_FAKE_CODEX_COMPLETE_FROM_ATTEMPT: "3", AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer-retry") }, backendRetryMaxAttempts: 3, backendRetryInitialBackoffMs: 100, backendRetryMaxBackoffMs: 100, oneShot: true, pollIntervalMs: 25 }) as JsonRecord;
+    assert.equal(replayRpcRetryResult.terminalStatus, "completed");
+    const replayRpcDeliveries = (await readFile(replayRpcRetrySteerFile, "utf8")).trim().split("\n").filter(Boolean).map((line) => JSON.parse(line) as { attempt: number; text: string });
+    assert.deepEqual(replayRpcDeliveries, [{ attempt: 2, text: "STEER_RPC_RETRY" }, { attempt: 3, text: "STEER_RPC_RETRY" }], "a failed replay RPC must not be marked seen, and the next backend attempt must retry it once");
+    const replayRpcEventsResponse = await client.get(`/api/v1/runs/${replayRpcRetryRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
+    const replayRpcEvents = replayRpcEventsResponse.items ?? [];
+    assert.ok(replayRpcEvents.some((event) => event.type === "error" && event.payload?.phase === "turn/steer:replay-failed" && event.payload?.backendAttempt === 2));
+    assert.deepEqual(replayRpcEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed").map((event) => event.payload?.backendAttempt), [3]);
+
+    const rebuiltRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer-retry", "resume acknowledged target on rebuilt runner", "trc_hwlab_command_rebuilt_runner", 2_000);
+    await client.post("/api/v1/runners/register", { id: "runner_stale_before_rebuild", runId: rebuiltRun.runId, attemptId: "attempt_stale_before_rebuild", backendProfile: "codex", placement: "host-process", sourceCommit: "self-test" });
+    await client.post(`/api/v1/runs/${rebuiltRun.runId}/claim`, { runnerId: "runner_stale_before_rebuild", leaseMs: 1 });
+    await client.post(`/api/v1/commands/${rebuiltRun.commandId}/ack`, {});
+    const rebuiltAckSteer = await client.post(`/api/v1/runs/${rebuiltRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_REBUILT_ACKNOWLEDGED", traceId: "trc_hwlab_command_rebuilt_ack" }, idempotencyKey: "trc_hwlab_command_rebuilt_ack" }) as { id: string };
+    await client.post(`/api/v1/commands/${rebuiltAckSteer.id}/ack`, {});
+    const rebuiltCompletedSteer = await client.post(`/api/v1/runs/${rebuiltRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_REBUILT_COMPLETED", traceId: "trc_hwlab_command_rebuilt_completed" }, idempotencyKey: "trc_hwlab_command_rebuilt_completed" }) as { id: string };
+    await client.post(`/api/v1/commands/${rebuiltCompletedSteer.id}/ack`, {});
+    await client.patch(`/api/v1/commands/${rebuiltCompletedSteer.id}/status`, { terminalStatus: "completed", failureKind: null, failureMessage: null });
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    const rebuiltSteerFile = path.join(context.tmp, "fake-codex-steers-runner-rebuild.txt");
+    const rebuiltResult = await runOnce({ managerUrl: server.baseUrl, runId: rebuiltRun.runId, commandId: rebuiltRun.commandId, runnerId: "runner_after_rebuild", attemptId: "attempt_after_rebuild", leaseMs: 1_000, claimRetryTimeoutMs: 500, claimRetryIntervalMs: 25, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "steer-runner-rebuild", AGENTRUN_FAKE_CODEX_START_FILE: path.join(context.tmp, "fake-codex-starts-runner-rebuild.txt"), AGENTRUN_FAKE_CODEX_STEER_FILE: rebuiltSteerFile, AGENTRUN_FAKE_CODEX_EXPECTED_STEERS: "2", AGENTRUN_FAKE_CODEX_COMPLETE_FROM_ATTEMPT: "1", AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer-retry") }, oneShot: true, pollIntervalMs: 25 }) as JsonRecord;
+    assert.equal(rebuiltResult.terminalStatus, "completed");
+    const rebuiltDeliveries = (await readFile(rebuiltSteerFile, "utf8")).trim().split("\n").filter(Boolean).map((line) => JSON.parse(line) as { attempt: number; text: string });
+    assert.deepEqual(rebuiltDeliveries, [{ attempt: 1, text: "STEER_REBUILT_ACKNOWLEDGED" }, { attempt: 1, text: "STEER_REBUILT_COMPLETED" }]);
+    const rebuiltEventsResponse = await client.get(`/api/v1/runs/${rebuiltRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
+    const rebuiltEvents = rebuiltEventsResponse.items ?? [];
+    const rebuiltReplayEvents = rebuiltEvents.filter((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:replayed");
+    assert.deepEqual(rebuiltReplayEvents.map((event) => event.payload?.backendAttempt), [1, 1]);
+    assert.deepEqual(rebuiltReplayEvents.map((event) => event.payload?.resumeDurableTurn), [true, true]);
+    assert.equal(rebuiltEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "runner:backend-retry"), false);
+    await waitForCommandState(client, rebuiltRun.runId, rebuiltAckSteer.id, "completed");
+    await waitForCommandState(client, rebuiltRun.runId, rebuiltRun.commandId, "completed");
 
     const noEventWatchdog = await createHwlabRun(client, context, bundle, "hwlab-session-no-event-watchdog", "complete a tool and then produce no more events", "hwlab-command-no-event-watchdog", 10_000);
     const noEventWatchdogRunner = runOnce({ managerUrl: server.baseUrl, runId: noEventWatchdog.runId, commandId: noEventWatchdog.commandId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "tool-completes-without-terminal", AGENTRUN_RUNNER_IDLE_TIMEOUT_MS: "300", AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-no-event-watchdog") }, oneShot: true, pollIntervalMs: 50 });
@@ -335,7 +441,7 @@ process.exit(1);
     const runningResult = await running;
     assert.equal(runningResult.terminalStatus, "cancelled");
 
-    return { name: "hwlab-manual-dispatch", tests: ["runner-job-idempotency", "pending-cancel", "result-envelope", "session-ref-resume", "resource-gitbundle-materialization", "gitbundle-ref-resolution", "gitbundle-tools-path", "gitbundle-skill-dir-assembly", "resource-prompt-required-blocker", "resource-required-skill-blocker", "same-run-runner-multiturn", "running-steer", "running-steer-user-message-target-trace", "no-event-watchdog-after-tool", "tool-output-hard-timeout", "running-cancel"] };
+    return { name: "hwlab-manual-dispatch", tests: ["runner-job-idempotency", "pending-cancel", "result-envelope", "session-ref-resume", "resource-gitbundle-materialization", "gitbundle-ref-resolution", "gitbundle-tools-path", "gitbundle-skill-dir-assembly", "resource-prompt-required-blocker", "resource-required-skill-blocker", "same-run-runner-multiturn", "running-steer", "running-steer-user-message-target-trace", "backend-retry-replays-durable-steers", "backend-replay-barrier-fast-terminal", "backend-replay-rpc-failure-retries", "rebuilt-runner-resumes-acknowledged-turn", "no-event-watchdog-after-tool", "tool-output-hard-timeout", "running-cancel"] };
   } finally {
     await new Promise<void>((resolve) => server.server.close(() => resolve()));
   }

src/selftest/fake-codex-app-server.ts

@@ -1,10 +1,11 @@
 import * as readline from "node:readline";
 import { spawnSync } from "node:child_process";
-import { appendFileSync, writeFileSync } from "node:fs";
+import { appendFileSync, readFileSync, writeFileSync } from "node:fs";
 
 const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity });
 const mode = process.env.AGENTRUN_FAKE_CODEX_MODE ?? "success";
 if (process.env.AGENTRUN_FAKE_CODEX_START_FILE) appendFileSync(process.env.AGENTRUN_FAKE_CODEX_START_FILE, `${process.pid}\n`);
+const fakeProcessAttempt = process.env.AGENTRUN_FAKE_CODEX_START_FILE ? lineCount(process.env.AGENTRUN_FAKE_CODEX_START_FILE) : 1;
 let threadCounter = 0;
 let turnCounter = 0;
 let observedThreadModel = false;
@@ -438,6 +439,19 @@ for await (const line of rl) {
       respond(message.id, { turn });
       continue;
     }
+    if (["steer-retry-resume", "steer-retry-fast-terminal", "steer-retry-rpc-fails-once", "steer-runner-rebuild"].includes(mode)) {
+      turnCounter += 1;
+      const turn = { id: `turn_selftest_${turnCounter}`, status: "running" };
+      notify("turn/started", { turn });
+      activeSteerTurn = {
+        id: turn.id,
+        completed: false,
+        timer: setTimeout(() => undefined, 60_000),
+      };
+      respond(message.id, { turn });
+      if (mode === "steer-retry-fast-terminal" && fakeProcessAttempt > 1) completeActiveSteerTurn("fast-terminal-before-replay");
+      continue;
+    }
     if (mode === "noisy-reasoning-events") {
       turnCounter += 1;
       const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" };
@@ -467,15 +481,31 @@ for await (const line of rl) {
     continue;
   }
   if (message.method === "turn/steer") {
-    if (mode !== "steer-waits" || !activeSteerTurn) {
+    const durableSteerMode = ["steer-retry-resume", "steer-retry-fast-terminal", "steer-retry-rpc-fails-once", "steer-runner-rebuild"].includes(mode);
+    if ((mode !== "steer-waits" && !durableSteerMode) || !activeSteerTurn || activeSteerTurn.completed) {
       respond(message.id, null, { code: -32000, message: "no active fake turn for steer" });
       continue;
     }
     const text = steerText(message.params?.input);
+    if (process.env.AGENTRUN_FAKE_CODEX_STEER_FILE) appendFileSync(process.env.AGENTRUN_FAKE_CODEX_STEER_FILE, `${JSON.stringify({ attempt: fakeProcessAttempt, text })}\n`);
+    const delivered = process.env.AGENTRUN_FAKE_CODEX_STEER_FILE ? steerAttempts(process.env.AGENTRUN_FAKE_CODEX_STEER_FILE, fakeProcessAttempt) : [];
+    if (mode === "steer-retry-rpc-fails-once" && fakeProcessAttempt === 2 && delivered.length === 1) {
+      respond(message.id, null, { code: -32000, message: "turn/steer replay timeout before acknowledgement" });
+      continue;
+    }
     notify("item/agentMessage/delta", { itemId: "msg_steer", delta: `steered:${text}` });
     notify("item/completed", { item: { id: "msg_steer", type: "agentMessage", text: `steered:${text}` } });
     respond(message.id, { accepted: true });
-    setTimeout(() => completeActiveSteerTurn("steer-applied"), 20);
+    if (durableSteerMode) {
+      const expectedSteers = Number(process.env.AGENTRUN_FAKE_CODEX_EXPECTED_STEERS ?? "2");
+      const completeFromAttempt = Number(process.env.AGENTRUN_FAKE_CODEX_COMPLETE_FROM_ATTEMPT ?? "2");
+      if (delivered.length >= expectedSteers) {
+        if (fakeProcessAttempt >= completeFromAttempt) {
+          notify("item/completed", { item: { id: "msg_steer_replayed", type: "agentMessage", text: `replayed:${delivered.join("|")}` } });
+          setTimeout(() => completeActiveSteerTurn("steers-replayed"), 20);
+        }
+      }
+    } else setTimeout(() => completeActiveSteerTurn("steer-applied"), 20);
     continue;
   }
   if (message.method === "turn/interrupt") {
@@ -508,6 +538,25 @@ function completeActiveSteerTurn(reason: string, status = "completed"): void {
   notify("turn/completed", { turn });
 }
 
+function lineCount(file: string): number {
+  try {
+    return readFileSync(file, "utf8").split("\n").filter(Boolean).length;
+  } catch {
+    return 1;
+  }
+}
+
+function steerAttempts(file: string, attempt: number): string[] {
+  try {
+    return readFileSync(file, "utf8").split("\n").filter(Boolean).flatMap((line) => {
+      const record = JSON.parse(line) as { attempt?: unknown; text?: unknown };
+      return record.attempt === attempt && typeof record.text === "string" ? [record.text] : [];
+    });
+  } catch {
+    return [];
+  }
+}
+
 function steerText(input: unknown): string {
   if (!Array.isArray(input)) return "";
   return input.flatMap((item) => {