mirror of
https://github.com/getpaseo/paseo.git
synced 2026-11-10 18:35:17 +00:00
fix(server): let an aborted wait end while the provider is starting the turn
10c443a32 made waitForAgentEvent wait for a starting run before reading its
state, but that wait came before the abort listener. A provider start has no
deadline of its own, so a stalled startTurn kept every timed wait (MCP wait,
CLI wait, schedules, finish subscriptions) past its timeout. The start wait
now rejects as soon as the caller's signal aborts.
This commit is contained in:
@@ -1257,6 +1257,44 @@ test("steers a send that arrives while the previous turn is still starting", asy
|
||||
}
|
||||
});
|
||||
|
||||
test("an aborted wait ends while the provider is still starting the turn", async () => {
|
||||
const startHeld = deferred<void>();
|
||||
class StalledStartSession extends TestAgentSession {
|
||||
override async startTurn(): Promise<{ turnId: string }> {
|
||||
await startHeld.promise;
|
||||
throw new Error("start released by the test");
|
||||
}
|
||||
}
|
||||
const session = new StalledStartSession({ provider: "codex", cwd: process.cwd() });
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-wait-stalled-start-"));
|
||||
const manager = new AgentManager({
|
||||
clients: {
|
||||
codex: new (class extends TestAgentClient {
|
||||
override async createSession(): Promise<AgentSession> {
|
||||
return session;
|
||||
}
|
||||
})(),
|
||||
},
|
||||
logger,
|
||||
});
|
||||
let agentId: string | null = null;
|
||||
try {
|
||||
const agent = await manager.createAgent({ provider: "codex", cwd: workdir }, undefined, {
|
||||
workspaceId: undefined,
|
||||
});
|
||||
agentId = agent.id;
|
||||
await startAgentRun(manager, agent.id, "stalls", logger, {});
|
||||
const controller = new AbortController();
|
||||
const wait = manager.waitForAgentEvent(agent.id, { signal: controller.signal });
|
||||
controller.abort(new Error("wait timeout"));
|
||||
await expect(wait).rejects.toThrow("wait timeout");
|
||||
} finally {
|
||||
startHeld.resolve();
|
||||
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("starts its own turn when the previous turn fails to start under a steer send", async () => {
|
||||
const startHeld = deferred<void>();
|
||||
class FailingFirstStartSession extends SteeringTestSession {
|
||||
|
||||
@@ -4261,6 +4261,25 @@ export class AgentManager {
|
||||
return chunks.length ? chunks.toReversed().join("") : null;
|
||||
}
|
||||
|
||||
/** A provider start has no deadline of its own, so the caller's abort ends the wait for it. */
|
||||
private async waitForForegroundStartUnlessAborted(
|
||||
agentId: string,
|
||||
signal: AbortSignal | undefined,
|
||||
): Promise<void> {
|
||||
if (!signal) return await this.runs.waitForForegroundStart(agentId);
|
||||
if (signal.aborted) throw createAbortError(signal, "wait_for_agent aborted");
|
||||
let onAbort: () => void = () => undefined;
|
||||
const aborted = new Promise<never>((_resolve, reject) => {
|
||||
onAbort = () => reject(createAbortError(signal, "wait_for_agent aborted"));
|
||||
signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
try {
|
||||
await Promise.race([this.runs.waitForForegroundStart(agentId), aborted]);
|
||||
} finally {
|
||||
signal.removeEventListener("abort", onAbort);
|
||||
}
|
||||
}
|
||||
|
||||
async waitForAgentEvent(
|
||||
agentId: string,
|
||||
options?: WaitForAgentOptions,
|
||||
@@ -4270,7 +4289,7 @@ export class AgentManager {
|
||||
}
|
||||
// A starting run publishes its running state only once its turn starts. Until then the
|
||||
// replayed state below is the idle state before the run, which would end the wait early.
|
||||
await this.runs.waitForForegroundStart(agentId);
|
||||
await this.waitForForegroundStartUnlessAborted(agentId, options?.signal);
|
||||
const snapshot = this.getAgent(agentId);
|
||||
if (!snapshot) {
|
||||
throw new Error(`Agent ${agentId} not found`);
|
||||
|
||||
Reference in New Issue
Block a user