fix: 以 replay barrier 恢复 active turn

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);
@@ -275,6 +276,9 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
     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,
@@ -300,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, attemptId, runner.id, control, options.pollIntervalMs, runnerLog, options.runnerJobId, backendAttempt);
+        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;
@@ -308,7 +315,23 @@ async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions,
       },
     };
     const retryPolicy = backendRetryPolicy(latestRun.executionPolicy, preparedCredential.env, options);
-    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,27 +427,36 @@ function positiveInteger(value: unknown, fallback: number): number {
   return Math.max(1, Math.floor(parsed));
 }
 
-function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targetCommand: CommandRecord, attemptId: string, runnerId: string, control: BackendActiveTurnControl, pollIntervalMs: number | undefined, runnerLog: RunnerLogSink, runnerJobId: string | undefined, backendAttempt: number): () => 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 replayDurableSteers = backendAttempt > 1;
+  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 {
+      await replayBarrier;
+      if (stopped) return;
       const commands = await listCurrentTurnControlCommands(api, runId, targetCommand.seq);
-      const controlCommands = commands.filter((item) => {
-        if (seen.has(item.id)) return false;
-        if ((item.type === "steer" || item.type === "interrupt") && item.state === "pending") return true;
-        return replayDurableSteers && item.type === "steer" && (item.state === "acknowledged" || item.state === "completed");
-      });
+      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, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
-        else if (controlCommand.state === "pending") await handleSteerCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
-        else await replayDurableSteerCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, backendAttempt, 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.
@@ -434,9 +466,12 @@ 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,
   };
 }
 
@@ -455,17 +490,23 @@ async function listCurrentTurnControlCommands(api: RunnerManagerApi, runId: stri
   }
 }
 
-async function replayDurableSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, backendAttempt: number, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
+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, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
+    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, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
+    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;
   }
 }
 

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

@@ -360,6 +360,54 @@ process.exit(1);
     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 });
     await waitForCommandState(client, noEventWatchdog.runId, noEventWatchdog.commandId, "acknowledged");
@@ -393,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", "backend-retry-replays-durable-steers", "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

@@ -439,7 +439,7 @@ for await (const line of rl) {
       respond(message.id, { turn });
       continue;
     }
-    if (mode === "steer-retry-resume") {
+    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 });
@@ -449,6 +449,7 @@ for await (const line of rl) {
         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") {
@@ -480,20 +481,26 @@ for await (const line of rl) {
     continue;
   }
   if (message.method === "turn/steer") {
-    if ((mode !== "steer-waits" && mode !== "steer-retry-resume") || !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 });
-    if (mode === "steer-retry-resume") {
-      const delivered = process.env.AGENTRUN_FAKE_CODEX_STEER_FILE ? steerAttempts(process.env.AGENTRUN_FAKE_CODEX_STEER_FILE, fakeProcessAttempt) : [];
+    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 > 1) {
+        if (fakeProcessAttempt >= completeFromAttempt) {
           notify("item/completed", { item: { id: "msg_steer_replayed", type: "agentMessage", text: `replayed:${delivered.join("|")}` } });
           setTimeout(() => completeActiveSteerTurn("steers-replayed"), 20);
         }