mirror of
https://github.com/getpaseo/paseo.git
synced 2026-10-28 20:22:45 +00:00
Wake a waiting steer only once the starting turn has an id or its run has ended
A failed start used to wake the waiting steer before its run was cleared, leaving the later send to replace a run that had already failed.
This commit is contained in:
@@ -957,6 +957,64 @@ test("steers a send that arrives while the previous turn is still starting", asy
|
||||
}
|
||||
});
|
||||
|
||||
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 {
|
||||
override async startTurn(prompt: AgentPromptInput): Promise<{ turnId: string }> {
|
||||
if (prompt === "FIRST") {
|
||||
this.startPrompts.push(prompt);
|
||||
await startHeld.promise;
|
||||
throw new Error("first start failed");
|
||||
}
|
||||
return await super.startTurn(prompt);
|
||||
}
|
||||
}
|
||||
const session = new FailingFirstStartSession({
|
||||
provider: "codex",
|
||||
cwd: process.cwd(),
|
||||
});
|
||||
const workdir = mkdtempSync(join(tmpdir(), "agent-manager-steer-failed-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;
|
||||
const steerOptions = (clientMessageId: string) => ({
|
||||
replaceRunning: true,
|
||||
activeTurnBehavior: "steer" as const,
|
||||
runOptions: { clientMessageId },
|
||||
});
|
||||
await startAgentRun(manager, agent.id, "FIRST", logger, steerOptions("first-client"));
|
||||
const second = startAgentRun(
|
||||
manager,
|
||||
agent.id,
|
||||
"SECOND",
|
||||
logger,
|
||||
steerOptions("second-client"),
|
||||
);
|
||||
|
||||
startHeld.resolve();
|
||||
await expect(second).resolves.toEqual({ disposition: "turn_started" });
|
||||
await vi.waitFor(() => expect(session.startPrompts).toEqual(["FIRST", "SECOND"]));
|
||||
expect(session.interruptCount).toBe(0);
|
||||
} finally {
|
||||
startHeld.resolve();
|
||||
if (agentId) await manager.closeAgent(agentId).catch(() => undefined);
|
||||
rmSync(workdir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
test("orders an accepted steer before output emitted while acknowledgement is pending", async () => {
|
||||
const entered = deferred<void>();
|
||||
const release = deferred<void>();
|
||||
|
||||
@@ -20,8 +20,8 @@ export interface PendingForegroundRun {
|
||||
kind: "foreground";
|
||||
stagedEvents: AgentStreamEvent[];
|
||||
start: ForegroundRunStart;
|
||||
startSettledPromise: Promise<void>;
|
||||
resolveStartSettled: () => void;
|
||||
turnStartedPromise: Promise<void>;
|
||||
resolveTurnStarted: () => void;
|
||||
settled: boolean;
|
||||
settledPromise: Promise<void>;
|
||||
resolveSettled: () => void;
|
||||
@@ -63,16 +63,21 @@ export class AgentRunState {
|
||||
start: Exclude<ForegroundRunStart, { status: "pending" }>,
|
||||
): void {
|
||||
run.start = start;
|
||||
run.resolveStartSettled();
|
||||
if (start.status === "started") {
|
||||
run.resolveTurnStarted();
|
||||
}
|
||||
}
|
||||
|
||||
/** Resolves once the agent's foreground run has a turn id, failed to start, or ended. */
|
||||
/**
|
||||
* Resolves once the agent's foreground run has a turn id or has ended. A failed start
|
||||
* resolves only once its run is cleared, so callers never see a failed run still tracked.
|
||||
*/
|
||||
async waitForForegroundStart(agentId: string): Promise<void> {
|
||||
const run = this.getPendingRun(agentId);
|
||||
if (run?.start.status !== "pending") {
|
||||
return;
|
||||
}
|
||||
await Promise.race([run.startSettledPromise, run.settledPromise]);
|
||||
await Promise.race([run.turnStartedPromise, run.settledPromise]);
|
||||
}
|
||||
|
||||
hasPendingRun(agentId: string): boolean {
|
||||
@@ -298,16 +303,16 @@ export class ForegroundTurnStream {
|
||||
}
|
||||
|
||||
function createPendingForegroundRun(): PendingForegroundRun {
|
||||
let resolveStartSettled!: () => void;
|
||||
const startSettledPromise = new Promise<void>((resolvePromise) => {
|
||||
resolveStartSettled = resolvePromise;
|
||||
let resolveTurnStarted!: () => void;
|
||||
const turnStartedPromise = new Promise<void>((resolvePromise) => {
|
||||
resolveTurnStarted = resolvePromise;
|
||||
});
|
||||
return {
|
||||
...createTrackedRunState(),
|
||||
kind: "foreground",
|
||||
start: { status: "pending" },
|
||||
startSettledPromise,
|
||||
resolveStartSettled,
|
||||
turnStartedPromise,
|
||||
resolveTurnStarted,
|
||||
stagedEvents: [],
|
||||
};
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user