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) => {