fix(server): resume agents cleanly after failed turns

Track each foreground run start as pending, started, or failed so a previous turn error cannot reject a newly accepted prompt. Includes a regression for resuming after a failed turn.
This commit is contained in:
Mohamed Boudra
2026-08-31 12:10:34 +02:00
committed by GitHub
parent 4a3e6dea63
commit 11e76f6655
3 changed files with 117 additions and 15 deletions
@@ -5543,6 +5543,88 @@ test("waitForAgentRunStart resolves while a foreground run is still only pending
expect(manager.getAgent(snapshot.id)?.lifecycle).toBe("idle");
});
test("waitForAgentRunStart ignores a prior turn error while the next run starts", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-error-resume-"));
const storage = new AgentStorage(join(workdir, "agents"), logger);
const retryStartEntered = deferred<void>();
const releaseRetryStart = deferred<void>();
class ErrorThenResumeSession extends TestAgentSession {
override async startTurn(): Promise<{ turnId: string }> {
const turnId = `turn-${++this.turnIdCounter}`;
if (this.turnIdCounter === 1) {
void (async () => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
this.pushEvent({
type: "turn_failed",
provider: this.provider,
turnId,
error: "model at capacity",
});
})();
return { turnId };
}
retryStartEntered.resolve();
await releaseRetryStart.promise;
void (async () => {
this.pushEvent({ type: "turn_started", provider: this.provider, turnId });
this.pushEvent({ type: "turn_completed", provider: this.provider, turnId });
})();
return { turnId };
}
}
class ErrorThenResumeClient extends TestAgentClient {
override async createSession(config: AgentSessionConfig): Promise<AgentSession> {
return new ErrorThenResumeSession(config);
}
}
const manager = new AgentManager({
clients: { codex: new ErrorThenResumeClient() },
registry: storage,
logger,
});
let agentId: string | null = null;
try {
const snapshot = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
workspaceId: undefined,
});
agentId = snapshot.id;
await manager.runAgent(agentId, "fail").catch(() => undefined);
expect(manager.getAgent(agentId)?.lifecycle).toBe("error");
const dispatch = await startAgentRun(manager, agentId, "resume", logger);
expect(dispatch.disposition).toBe("turn_started");
const wait = manager.waitForAgentRunStart(agentId);
let earlyResult: "pending" | "resolved" | "rejected" = "pending";
void wait.then(
() => {
earlyResult = "resolved";
return earlyResult;
},
() => {
earlyResult = "rejected";
return earlyResult;
},
);
await retryStartEntered.promise;
await Promise.resolve();
expect(earlyResult).toBe("pending");
releaseRetryStart.resolve();
await expect(wait).resolves.toBeUndefined();
await manager.waitForAgentEvent(agentId);
} finally {
releaseRetryStart.resolve();
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
rmSync(workdir, { recursive: true, force: true });
}
});
test("replaceAgentRun does not emit idle or resolve waiters between interrupted and replacement runs", async () => {
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-replace-run-"));
const storagePath = join(workdir, "agents");
@@ -2165,6 +2165,7 @@ export class AgentManager {
}
agent.pendingReplacement = false;
const errorMsg = error instanceof Error ? error.message : "Failed to start turn";
pendingRun.start = { status: "failed", error: errorMsg };
await this.handleStreamEvent(agent, {
type: "turn_failed",
provider: agent.provider,
@@ -2232,8 +2233,7 @@ export class AgentManager {
agent.pendingReplacement = false;
}
const turnStartedAt = new Date();
pendingRun.started = true;
pendingRun.turnId = turnId;
pendingRun.start = { status: "started", turnId };
agent.activeForegroundTurnId = turnId;
this.openActiveTurn(agent, turnId, turnStartedAt);
agent.lifecycle = "running";
@@ -2575,7 +2575,10 @@ export class AgentManager {
}
const pendingRun = this.runs.getPendingRun(agentId);
if ((snapshot.lifecycle === "running" || pendingRun?.started) && !snapshot.pendingReplacement) {
if (
(snapshot.lifecycle === "running" || pendingRun?.start.status === "started") &&
!snapshot.pendingReplacement
) {
return;
}
@@ -2640,14 +2643,19 @@ export class AgentManager {
const currentPendingRun = this.runs.getPendingRun(agentId);
if (
(current.lifecycle === "running" || currentPendingRun?.started) &&
(current.lifecycle === "running" || currentPendingRun?.start.status === "started") &&
!current.pendingReplacement
) {
finishOk();
return true;
}
if (current.lifecycle === "error" && !currentPendingRun?.started) {
if (currentPendingRun?.start.status === "failed") {
finishErr(new Error(currentPendingRun.start.error));
return true;
}
if (current.lifecycle === "error" && !currentPendingRun) {
finishErr(new Error(current.lastError ?? `Agent ${agentId} failed to start`));
return true;
}
@@ -2734,16 +2742,17 @@ export class AgentManager {
return { status: settlement === "completed" ? "settled" : "refused" };
}
if (settlement === "timed_out" && run.turnId) {
const runTurnId = this.runs.getTurnId(agentId);
if (settlement === "timed_out" && runTurnId) {
this.logger.warn(
{ agentId, turnId: run.turnId, kind: run.kind },
{ agentId, turnId: runTurnId, kind: run.kind },
"cancelAgentRun: acknowledged turn still active after timeout, force-canceling",
);
await this.dispatchSessionEvent(agent, {
type: "turn_canceled",
provider: agent.provider,
reason: "interrupted",
turnId: run.turnId,
turnId: runTurnId,
});
await run.settledPromise;
} else if (settlement === "timed_out" && run.kind === "foreground") {
@@ -3023,7 +3032,7 @@ export class AgentManager {
let hasStarted =
isAgentBusy(initialStatus) ||
Boolean(snapshot.activeForegroundTurnId) ||
Boolean(pendingForegroundRun?.started);
pendingForegroundRun?.start.status === "started";
let terminalStatusOverride: AgentLifecycleStatus | null = null;
let finished = false;
@@ -3429,7 +3438,7 @@ export class AgentManager {
return;
}
const pendingRun = this.runs.getPendingRun(agentId);
if (pendingRun && !pendingRun.started) {
if (pendingRun?.start.status === "pending") {
pendingRun.stagedEvents.push(event);
return;
}
@@ -14,8 +14,10 @@ export interface PendingForegroundRun {
token: string;
kind: "foreground";
stagedEvents: AgentStreamEvent[];
turnId: string | null;
started: boolean;
start:
| { status: "pending" }
| { status: "started"; turnId: string }
| { status: "failed"; error: string };
settled: boolean;
settledPromise: Promise<void>;
resolveSettled: () => void;
@@ -64,6 +66,13 @@ export class AgentRunState {
return this.runs.has(agentId);
}
getTurnId(agentId: string): string | null {
const run = this.runs.get(agentId);
if (!run) return null;
if (run.kind === "autonomous") return run.turnId;
return run.start.status === "started" ? run.start.turnId : null;
}
trackAutonomousRun(agentId: string, turnId: string | null): TrackedAgentRun {
const current = this.runs.get(agentId);
if (current) {
@@ -85,7 +94,10 @@ export class AgentRunState {
if (!run) {
return;
}
if (run.kind === "foreground" && (run.turnId === null || run.turnId !== turnId)) {
if (
run.kind === "foreground" &&
(run.start.status !== "started" || run.start.turnId !== turnId)
) {
return;
}
if (
@@ -268,8 +280,7 @@ function createPendingForegroundRun(): PendingForegroundRun {
return {
...createTrackedRunState(),
kind: "foreground",
turnId: null,
started: false,
start: { status: "pending" },
stagedEvents: [],
};
}