Reserve bounded Pi provider drain time after idle shutdown

This commit is contained in:
Dotta committed 2026-10-08 19:10:19 -05:00
1 parent 8d758e729e
commit d6ebded575
2 files changed
+46 -1

No files matched your search

@@ -1392,6 +1392,42 @@ it("reserves a bounded suspension window after close preparation", () => {
});
});
it.each([
{ provider: "acpx" as const, acpxAgent: "pi" as const, closeGraceMs: undefined, drained: true },
{ provider: "codex" as const, acpxAgent: undefined, closeGraceMs: undefined, drained: false },
{ provider: "acpx" as const, acpxAgent: "pi" as const, closeGraceMs: 400, drained: false },
])("keeps Pi stop, remote drain and suspension within a finite close budget ($provider, $closeGraceMs)", async ({ drained, ...options }) => {
vi.useFakeTimers();
try {
const { preparationDeadline, closeDeadline } = runnerdRecoveryInternals.runnerCloseDeadlines(
Date.now(), runnerdRecoveryInternals.runnerCloseGraceMs(options),
);
// The retained restart failure spent 5.2s stopping idle Pi before its
// remote drain acknowledgement crossed the next command round trip.
await vi.advanceTimersByTimeAsync(5_200);
const commands: { commandId: string; status: string; result?: unknown }[] = [];
const result = runnerdRecoveryInternals.awaitProviderDrainBarrier({
readProviderState: () => null,
semanticResultsSettled: () => true,
commands: () => commands,
queueDrain: commandId => {
const command = { commandId, status: "pending", result: undefined as unknown };
commands.push(command);
setTimeout(() => {
command.status = "completed";
command.result = { result: { retainedEventsDrained: true } };
}, 2_800);
},
pump: () => undefined,
deadline: Date.now() + Math.min(5_000, Math.max(0, preparationDeadline - Date.now())),
});
await vi.advanceTimersByTimeAsync(5_000);
expect(await result).toBe(drained);
expect(closeDeadline - preparationDeadline).toBeLessThanOrEqual(2_500);
expect(runnerdRecoveryInternals.runnerCloseGraceMs(options)).toBeLessThanOrEqual(15_000);
} finally { vi.useRealTimers(); }
});
it("joins an already-completed suspension without queuing a command to an exited runner", async () => {
const commands = [
{ commandId: "exact-suspend", type: "runner.suspend", status: "completed" },
@@ -838,6 +838,14 @@ async function awaitRunnerSuspensionBarrier(input: {
return false;
}
function runnerCloseGraceMs(options: Pick<CapabilityRunnerdCodexTransportOptions, "provider" | "acpxAgent" | "closeGraceMs">): number {
// Idle Pi retirement may consume its five-second RPC stop window. Keep a
// separate five-second drain round trip and the bounded suspension reserve.
// Explicit caller deadlines remain authoritative; all settlement proofs stay
// mandatory even when the provider already completed its task.
return options.closeGraceMs ?? (options.provider === "acpx" && options.acpxAgent === "pi" ? 15_000 : 10_000);
}
function runnerCloseDeadlines(
startedAtMs: number,
graceMs: number,
@@ -4428,7 +4436,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
// fresh run authority never inherits the prior run's pending events.
const { preparationDeadline, closeDeadline } = runnerCloseDeadlines(
Date.now(),
this.options.closeGraceMs ?? 10_000,
runnerCloseGraceMs(this.options),
);
if (!(await this.#runnerHasExited())) {
// Let an already-admitted tool result reach its original provider
@@ -7040,6 +7048,7 @@ export const runnerdRecoveryInternals = Object.freeze({
recoveredRunAttachment,
releaseRunnerProcessOwnership,
runnerCloseDeadlines,
runnerCloseGraceMs,
rotatedRunAttachPayload,
rotateExternalAuthorityEpoch,
turnStartCommandResultValid,