mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 在 backend resume 时重放 durable steer

Browse Source
This commit is contained in:
Lyon
2026-07-13 02:15:05 +02:00
parent 2fc565e2e8
commit e502758d7b
3 changed files with 147 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/runner/run-once.ts
+43 -8
View File
@@ -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<string>();
const replayDurableSteers = backendAttempt > 1;
const intervalMs = normalizePollIntervalMs(pollIntervalMs);
const poll = async (): Promise<void> => {
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<CommandRecord[]> {
const commands: CommandRecord[] = [];
let afterSeq = targetCommandSeq;
while (true) {
const batch = await api.listCommands(runId, { afterSeq, limit: 100 });
if (batch.length === 0) return commands;
for (const command of batch) {
if (command.type === "turn") return commands;
commands.push(command);
}
if (batch.length < 100) return commands;
afterSeq = batch[batch.length - 1]?.seq ?? afterSeq;
}
}
async function replayDurableSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, backendAttempt: number, 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 } });
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<void> {
const acked = await api.ackCommand(command.id);
if (acked.state === "cancelled") {
src/selftest/cases/50-hwlab-manual-dispatch.ts
+59 -1
View File
@@ -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<void>((resolve) => server.server.close(() => resolve()));
}
src/selftest/fake-codex-app-server.ts
+45 -3
View File
@@ -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) => {