fix: await queued Pi worker suspension

This commit is contained in:
2026-08-23 14:07:00 +08:00
parent 1727b75f30
commit 30c8bfd4d8
3 changed files with 68 additions and 19 deletions

View File

@@ -258,12 +258,22 @@ describe('Pi worker pool', () => {
it('suspends and later resumes the oldest queued parent to make child capacity', async () => {
const processBudget = new PiProcessBudget(8);
const workers = new Map<string, FakeWorker[]>();
const queuedStopGate = deferred();
let stoppingQueuedWorkers = 0;
const pool = new PiWorkerPool({
processBudget,
maxRunning: 4,
maxIdle: 8,
openWorker: async ({ conversation: input, generation, existingSession }) => {
const worker = new FakeWorker(`worker-${input.conversationId}-${generation}`);
if (generation === 1 && Number(input.conversationId.split('-').at(-1)) > 4) {
worker.stop = async () => {
worker.stopped = true;
stoppingQueuedWorkers += 1;
await queuedStopGate.promise;
return { mode: 'stdin-close' as const, code: 0, signal: null };
};
}
workers.set(input.conversationId, [...(workers.get(input.conversationId) ?? []), worker]);
return {
worker,
@@ -297,7 +307,7 @@ describe('Pi worker pool', () => {
async stop() {},
}),
});
await expect(scheduler.dispatch({
const children = scheduler.dispatch({
conversationId: ids[0]!,
workerGeneration: 1,
runId: 'run-child',
@@ -310,27 +320,47 @@ describe('Pi worker pool', () => {
toolProfile: 'read-only',
})),
},
})).resolves.toMatchObject({
details: {
tasks: Array.from({ length: 4 }, () => ({ status: 'complete' })),
},
});
await expect.poll(() => stoppingQueuedWorkers).toBe(4);
expect(processBudget.activeCount).toBe(8);
expect(processBudget.waitingCount).toBe(4);
const firstQueuedId = ids[4]!;
expect(ids.slice(4).map((id) => workers.get(id)?.[0]?.stopped))
.toEqual([true, true, true, true]);
let firstQueuedAccepted = false;
void tickets[4]!.accepted.then(
() => { firstQueuedAccepted = true; },
() => undefined,
);
workers.get(ids[0]!)?.[0]?.emit({ type: 'agent_settled' });
await new Promise((resolvePromise) => setImmediate(resolvePromise));
expect(firstQueuedAccepted).toBe(false);
expect(workers.get(ids[4]!)?.length).toBe(1);
expect(workers.get(ids[4]!)?.[0]?.requests).toHaveLength(0);
expect(pool.getState(firstQueuedId)).toMatchObject({
state: 'queued',
state: 'running',
generation: 1,
session: {
piSessionId: `session-${firstQueuedId}`,
sessionKey: `key-${firstQueuedId}`,
},
});
expect(processBudget.activeCount).toBe(4);
queuedStopGate.resolve();
await expect(children).resolves.toMatchObject({
details: {
tasks: Array.from({ length: 4 }, () => ({ status: 'complete' })),
},
});
expect(ids.slice(4).map((id) => workers.get(id)?.[0]?.stopped))
.toEqual([true, true, true, true]);
await expect.poll(() => workers.get(firstQueuedId)?.length).toBe(2);
await expect.poll(() => workers.get(firstQueuedId)?.[1]?.requests.length).toBe(1);
await expect(tickets[4]!.accepted).resolves.toMatchObject({ success: true });
expect(processBudget.activeCount).toBe(5);
expect(processBudget.waitingCount).toBe(0);
for (let index = 0; index < 4; index += 1) {
for (let index = 1; index < 4; index += 1) {
const queuedId = ids[index + 4]!;
workers.get(ids[index]!)?.[0]?.emit({ type: 'agent_settled' });
await expect.poll(() => workers.get(queuedId)?.length).toBe(2);