From e502758d7bf69ea60595d323487fc4119d553842 Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Mon, 13 Jul 2026 02:15:05 +0200 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E5=9C=A8=20backend=20resume=20?= =?UTF-8?q?=E6=97=B6=E9=87=8D=E6=94=BE=20durable=20steer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/runner/run-once.ts | 51 +++++++++++++--- .../cases/50-hwlab-manual-dispatch.ts | 60 ++++++++++++++++++- src/selftest/fake-codex-app-server.ts | 48 ++++++++++++++- 3 files changed, 147 insertions(+), 12 deletions(-) diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index 145529e..46953c5 100644 --- a/src/runner/run-once.ts +++ b/src/runner/run-once.ts @@ -274,6 +274,7 @@ 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; const backendOptions = { ...options, env: preparedCredential.env, @@ -299,7 +300,7 @@ 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); + stopSteerWatch = startActiveTurnCommandWatch(api, options.runId, command, attemptId, runner.id, control, options.pollIntervalMs, runnerLog, options.runnerJobId, backendAttempt); return () => { stopSteerWatch?.(); stopSteerWatch = undefined; @@ -307,7 +308,6 @@ 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); while (shouldRetryBackendTurn(result, retryPolicy, backendAttempt)) { const retryDelayMs = backendRetryDelayMs(retryPolicy, backendAttempt); @@ -404,21 +404,27 @@ 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): () => void { let stopped = false; let polling = false; const seen = new Set(); + const replayDurableSteers = backendAttempt > 1; const intervalMs = normalizePollIntervalMs(pollIntervalMs); const poll = async (): Promise => { if (stopped || polling) 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) { + 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"); + }); + 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 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); } } catch { // The active backend turn remains authoritative; missed control commands stay pending for the next poll. @@ -434,6 +440,35 @@ function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targe }; } +async function listCurrentTurnControlCommands(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise { + 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 replayDurableSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, backendAttempt: number, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise { + 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 } }); + 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 } }); + } +} + async function handleSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise { const acked = await api.ackCommand(command.id); if (acked.state === "cancelled") { diff --git a/src/selftest/cases/50-hwlab-manual-dispatch.ts b/src/selftest/cases/50-hwlab-manual-dispatch.ts index 2e6a02b..01a3ae1 100644 --- a/src/selftest/cases/50-hwlab-manual-dispatch.ts +++ b/src/selftest/cases/50-hwlab-manual-dispatch.ts @@ -301,6 +301,64 @@ 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 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 +393,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", "no-event-watchdog-after-tool", "tool-output-hard-timeout", "running-cancel"] }; } finally { await new Promise((resolve) => server.server.close(() => resolve())); } diff --git a/src/selftest/fake-codex-app-server.ts b/src/selftest/fake-codex-app-server.ts index 0ce61b3..2110630 100644 --- a/src/selftest/fake-codex-app-server.ts +++ b/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,18 @@ for await (const line of rl) { respond(message.id, { turn }); continue; } + if (mode === "steer-retry-resume") { + 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 }); + continue; + } if (mode === "noisy-reasoning-events") { turnCounter += 1; const turn = { id: `turn_selftest_${turnCounter}`, status: "completed" }; @@ -467,15 +480,25 @@ for await (const line of rl) { continue; } if (message.method === "turn/steer") { - if (mode !== "steer-waits" || !activeSteerTurn) { + if ((mode !== "steer-waits" && mode !== "steer-retry-resume") || !activeSteerTurn) { 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`); 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 (mode === "steer-retry-resume") { + const delivered = process.env.AGENTRUN_FAKE_CODEX_STEER_FILE ? steerAttempts(process.env.AGENTRUN_FAKE_CODEX_STEER_FILE, fakeProcessAttempt) : []; + const expectedSteers = Number(process.env.AGENTRUN_FAKE_CODEX_EXPECTED_STEERS ?? "2"); + if (delivered.length >= expectedSteers) { + if (fakeProcessAttempt > 1) { + 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 +531,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) => { From 21aa4e00d70ae3258131db1315692f33ea3e0a1a Mon Sep 17 00:00:00 2001 From: Lyon <88232613+pikasTech@users.noreply.github.com> Date: Mon, 13 Jul 2026 02:29:56 +0200 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20=E4=BB=A5=20replay=20barrier=20?= =?UTF-8?q?=E6=81=A2=E5=A4=8D=20active=20turn?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/runner/manager-api.ts | 3 +- src/runner/run-once.ts | 81 ++++++++++++++----- .../cases/50-hwlab-manual-dispatch.ts | 50 +++++++++++- src/selftest/fake-codex-app-server.ts | 17 ++-- 4 files changed, 124 insertions(+), 27 deletions(-) diff --git a/src/runner/manager-api.ts b/src/runner/manager-api.ts index 175f2d0..818f9f4 100644 --- a/src/runner/manager-api.ts +++ b/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 }; } diff --git a/src/runner/run-once.ts b/src/runner/run-once.ts index 46953c5..d19a1f3 100644 --- a/src/runner/run-once.ts +++ b/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 { + 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 = 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 => { + 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 } { let stopped = false; let polling = false; + let replayFailed = false; const seen = new Set(); - const replayDurableSteers = backendAttempt > 1; + const replayBarrier = (async (): Promise => { + 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 => { - 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 { +async function durableSteersForTurnResume(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise { + 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 { 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; } } diff --git a/src/selftest/cases/50-hwlab-manual-dispatch.ts b/src/selftest/cases/50-hwlab-manual-dispatch.ts index 01a3ae1..e5f554f 100644 --- a/src/selftest/cases/50-hwlab-manual-dispatch.ts +++ b/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((resolve) => server.server.close(() => resolve())); } diff --git a/src/selftest/fake-codex-app-server.ts b/src/selftest/fake-codex-app-server.ts index 2110630..2229ea5 100644 --- a/src/selftest/fake-codex-app-server.ts +++ b/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); }