fix(vibe): distinguish wait abort from timeout
This commit is contained in:
@@ -2,6 +2,10 @@
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Fixed
|
||||
|
||||
- Fixed interrupted `vibe_wait` calls being reported as elapsed timeout windows while preserving the per-call wait timeout.
|
||||
|
||||
## [17.3.3] - 2026-08-14
|
||||
|
||||
### Fixed
|
||||
|
||||
@@ -421,11 +421,17 @@ export class VibeSessionRegistry {
|
||||
}
|
||||
|
||||
/**
|
||||
* Insert a bare worker record without the spawn/job machinery. Test-only —
|
||||
* lets {@link aggregateVibeWorkerTokensPerSecond} be exercised against a
|
||||
* fake roster + AgentRegistry session without driving a real turn.
|
||||
* Insert a bare worker record without the spawn machinery. Test-only —
|
||||
* lets focused runtime tests attach an optional synthetic in-flight job.
|
||||
*/
|
||||
registerRecordForTests(record: { id: string; cli?: VibeCli; ownerId: string; state?: VibeSessionState }): void {
|
||||
registerRecordForTests(record: {
|
||||
id: string;
|
||||
cli?: VibeCli;
|
||||
ownerId: string;
|
||||
state?: VibeSessionState;
|
||||
jobId?: string;
|
||||
}): void {
|
||||
const now = Date.now();
|
||||
this.#records.set(record.id, {
|
||||
id: record.id,
|
||||
cli: record.cli ?? "fast",
|
||||
@@ -434,8 +440,11 @@ export class VibeSessionRegistry {
|
||||
parentSessionFile: null,
|
||||
agent: getBundledAgent("sonic")!,
|
||||
state: record.state ?? "running",
|
||||
createdAt: Date.now(),
|
||||
lastActivityAt: Date.now(),
|
||||
createdAt: now,
|
||||
lastActivityAt: now,
|
||||
turn: record.jobId
|
||||
? { jobId: record.jobId, message: "test turn", startedAt: now, trace: [], toolCount: 0 }
|
||||
: undefined,
|
||||
queue: [],
|
||||
turnCount: 0,
|
||||
killed: false,
|
||||
@@ -1113,25 +1122,31 @@ export class VibeSessionRegistry {
|
||||
if (job?.status === "running") runningJobs.push(job);
|
||||
}
|
||||
|
||||
let waited = false;
|
||||
let waitEndedByTimeout = false;
|
||||
if (runningJobs.length > 0 && collectSettled().length === 0) {
|
||||
waited = true;
|
||||
const timeoutMs = Math.max(1, Math.trunc(args.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS));
|
||||
const watchedJobIds = runningJobs.map(job => job.id);
|
||||
manager.watchJobs(watchedJobIds);
|
||||
const { promise: timeoutPromise, resolve: timeoutResolve } = Promise.withResolvers<void>();
|
||||
const timeoutHandle = setTimeout(() => timeoutResolve(), timeoutMs);
|
||||
const racePromises: Promise<unknown>[] = [...runningJobs.map(job => job.promise), timeoutPromise];
|
||||
const { promise: timeoutPromise, resolve: timeoutResolve } = Promise.withResolvers<"timeout">();
|
||||
const timeoutHandle = setTimeout(() => timeoutResolve("timeout"), timeoutMs);
|
||||
const racePromises: Array<Promise<"settled" | "timeout" | "aborted">> = [
|
||||
...runningJobs.map(job => job.promise.then(() => "settled" as const)),
|
||||
timeoutPromise,
|
||||
];
|
||||
let abortCleanup: (() => void) | undefined;
|
||||
if (args.signal) {
|
||||
const { promise: abortPromise, resolve: abortResolve } = Promise.withResolvers<void>();
|
||||
const onAbort = () => abortResolve();
|
||||
args.signal.addEventListener("abort", onAbort, { once: true });
|
||||
abortCleanup = () => args.signal?.removeEventListener("abort", onAbort);
|
||||
const { promise: abortPromise, resolve: abortResolve } = Promise.withResolvers<"aborted">();
|
||||
const onAbort = () => abortResolve("aborted");
|
||||
if (args.signal.aborted) {
|
||||
onAbort();
|
||||
} else {
|
||||
args.signal.addEventListener("abort", onAbort, { once: true });
|
||||
abortCleanup = () => args.signal?.removeEventListener("abort", onAbort);
|
||||
}
|
||||
racePromises.push(abortPromise);
|
||||
}
|
||||
try {
|
||||
await Promise.race(racePromises);
|
||||
waitEndedByTimeout = (await Promise.race(racePromises)) === "timeout";
|
||||
} finally {
|
||||
manager.unwatchJobs(watchedJobIds);
|
||||
clearTimeout(timeoutHandle);
|
||||
@@ -1144,7 +1159,7 @@ export class VibeSessionRegistry {
|
||||
// Current in-flight state, independent of the snapshot: a session whose
|
||||
// watched turn settled may already be mid queued follow-up.
|
||||
const stillRunning = watched.filter(record => record.turn !== undefined).map(record => record.id);
|
||||
return { settled, stillRunning, timedOut: waited && settled.length === 0 };
|
||||
return { settled, stillRunning, timedOut: waitEndedByTimeout && settled.length === 0 };
|
||||
}
|
||||
|
||||
/** Detach one parent's process-local workers without tombstoning their persisted conversations. */
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "bun:test";
|
||||
import { AsyncJobManager } from "../../src/async/job-manager";
|
||||
import type { ToolSession } from "../../src/tools";
|
||||
import { VibeWaitTool } from "../../src/tools/vibe";
|
||||
import { VibeSessionRegistry } from "../../src/vibe/runtime";
|
||||
|
||||
const OWNER = "test-owner";
|
||||
const WORKER = "test-worker";
|
||||
|
||||
interface TestTurn {
|
||||
jobId: string;
|
||||
complete: (text: string) => void;
|
||||
}
|
||||
|
||||
let manager: AsyncJobManager;
|
||||
let session: ToolSession;
|
||||
|
||||
function startTurn(options?: { onDelivery?: (jobId: string, text: string) => void }): TestTurn {
|
||||
const completion = Promise.withResolvers<string>();
|
||||
if (options?.onDelivery) {
|
||||
manager.registerDeliverySink(OWNER, options.onDelivery);
|
||||
}
|
||||
const jobId = manager.register(
|
||||
"task",
|
||||
"test vibe turn",
|
||||
async ({ signal }) => {
|
||||
const aborted = Promise.withResolvers<never>();
|
||||
const onAbort = () => aborted.reject(new Error("cancelled"));
|
||||
if (signal.aborted) onAbort();
|
||||
else signal.addEventListener("abort", onAbort, { once: true });
|
||||
try {
|
||||
return await Promise.race([completion.promise, aborted.promise]);
|
||||
} finally {
|
||||
signal.removeEventListener("abort", onAbort);
|
||||
}
|
||||
},
|
||||
{ ownerId: OWNER },
|
||||
);
|
||||
VibeSessionRegistry.global().registerRecordForTests({ id: WORKER, ownerId: OWNER, jobId });
|
||||
return { jobId, complete: completion.resolve };
|
||||
}
|
||||
|
||||
beforeEach(() => {
|
||||
manager = new AsyncJobManager({});
|
||||
session = {
|
||||
getAgentId: () => OWNER,
|
||||
getSessionId: () => "test-parent-session",
|
||||
getSessionFile: () => null,
|
||||
asyncJobManager: manager,
|
||||
} as unknown as ToolSession;
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
vi.useRealTimers();
|
||||
await manager.dispose({ timeoutMs: 100 });
|
||||
VibeSessionRegistry.resetGlobalForTests();
|
||||
});
|
||||
|
||||
describe("vibe wait completion classification", () => {
|
||||
it("reports a true timer expiry as timed out", async () => {
|
||||
vi.useFakeTimers();
|
||||
startTurn();
|
||||
const pending = VibeSessionRegistry.global().wait(session, { timeoutMs: 10 });
|
||||
vi.advanceTimersByTime(10);
|
||||
|
||||
const outcome = await pending;
|
||||
|
||||
expect(outcome.timedOut).toBe(true);
|
||||
expect(outcome.settled).toEqual([]);
|
||||
expect(outcome.stillRunning).toEqual([WORKER]);
|
||||
});
|
||||
|
||||
it("returns a settled worker result instead of timing out", async () => {
|
||||
const turn = startTurn();
|
||||
const pending = VibeSessionRegistry.global().wait(session, { timeoutMs: 1_000 });
|
||||
turn.complete("worker result");
|
||||
|
||||
const outcome = await pending;
|
||||
|
||||
expect(outcome.timedOut).toBe(false);
|
||||
expect(outcome.settled).toEqual([
|
||||
{ id: WORKER, jobId: turn.jobId, status: "completed", resultText: "worker result" },
|
||||
]);
|
||||
});
|
||||
|
||||
it("does not render an abort as an elapsed wait window, even with a long timeout", async () => {
|
||||
startTurn();
|
||||
const controller = new AbortController();
|
||||
controller.abort();
|
||||
|
||||
const result = await new VibeWaitTool(session).execute("wait-call", { timeout: 900 }, controller.signal);
|
||||
const text = result.content[0]?.type === "text" ? result.content[0].text : "";
|
||||
|
||||
expect(result.details?.wait?.timedOut).toBe(false);
|
||||
expect(text).toContain("Still running");
|
||||
expect(text).not.toContain("Wait window elapsed");
|
||||
expect(text).not.toContain("re-issue vibe_wait");
|
||||
});
|
||||
|
||||
it("restores async self-delivery after an interrupted wait", async () => {
|
||||
const deliveries: Array<{ jobId: string; text: string }> = [];
|
||||
const turn = startTurn({ onDelivery: (jobId, text) => deliveries.push({ jobId, text }) });
|
||||
const controller = new AbortController();
|
||||
const pending = VibeSessionRegistry.global().wait(session, {
|
||||
timeoutMs: 1_000,
|
||||
signal: controller.signal,
|
||||
});
|
||||
controller.abort();
|
||||
|
||||
const outcome = await pending;
|
||||
expect(outcome.timedOut).toBe(false);
|
||||
turn.complete("delivered later");
|
||||
await manager.getJob(turn.jobId)?.promise;
|
||||
await manager.drainDeliveries({ timeoutMs: 1_000 });
|
||||
|
||||
expect(deliveries).toEqual([{ jobId: turn.jobId, text: "delivered later" }]);
|
||||
});
|
||||
|
||||
it("returns a cancelled worker settlement without classifying it as timeout", async () => {
|
||||
const turn = startTurn();
|
||||
const pending = VibeSessionRegistry.global().wait(session, { timeoutMs: 1_000 });
|
||||
manager.cancel(turn.jobId, { ownerId: OWNER });
|
||||
|
||||
const outcome = await pending;
|
||||
|
||||
expect(outcome.timedOut).toBe(false);
|
||||
expect(outcome.settled[0]).toMatchObject({ id: WORKER, jobId: turn.jobId, status: "cancelled" });
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user