fix(native): measure resumed run usage from the prior run's total (#19161)

This commit is contained in:
Nicky Leach authored and GitHub committed 2026-11-04 18:55:14 -08:00
1 parent 6452e964a3
commit 737db73275
8 files changed
+176 -5

No files matched your search

+6
View File
@@ -1698,6 +1698,12 @@ future execution until Resume, and missing stop proof still blocks continuation.
### Late provider usage
Each run measures its usage from the total the previous run accounted for, whether
it continues in the same process or resumes from a saved session in a new one.
Usage above that total was not charged, so it stays in the new run's delta. A run
that reports nothing keeps the accounted total for the next run, and a recovery of
the same run keeps its starting point.
A provider can report more usage after its run's charge was posted. That usage
already happened, so it never fails a run or blocks a session. Codex usage totals
are cumulative, so the Runner reports the exact increment above the known total and
@@ -1,5 +1,5 @@
import { readCodexThreadState, readCodexTurnMetadata, readCodexTurnItems } from "./codex-history.js";
import { codexRunUsage, observeCodexUsage } from "./codex-usage-baseline.js";
import { codexRunUsage, codexUsageBaselineForNextRun, observeCodexUsage } from "./codex-usage-baseline.js";
import { randomUUID } from "node:crypto";
import { NativeProviderTerminalFailure } from "../../contracts/native-session-backend.js";
@@ -134,10 +134,7 @@ export class CodexHarnessSession
this.eventQueue.clear();
}
if (this.codexUsageBaseline && input.runId !== this.runId) {
// Keep the accounted total until a run reports usage, so a run with no
// report cannot absorb usage that was carried forward for the next one.
const { accounted, latest } = this.codexUsageBaseline;
this.codexUsageBaseline = { baseline: { ...(accounted ?? latest) }, latest: { ...latest }, ...(accounted ? { accounted: { ...accounted } } : {}) };
this.codexUsageBaseline = codexUsageBaselineForNextRun(this.codexUsageBaseline);
this.usageSnapshot = codexRunUsage(this.codexUsageBaseline);
}
this.runId = input.runId;
@@ -5,8 +5,64 @@ import {
WORKSPACE,
collectUntilTerminal,
} from "./codex-app-server-driver.test-support.js";
import { codexUsageBaselineForNextRun } from "./codex-usage-baseline.js";
describe("Codex resume accounting through the production driver", () => {
it("measures each run from the prior run's total across cold restarts", async () => {
const transports = [new FakeCodexTransport(), new FakeCodexTransport(), new FakeCodexTransport()];
const driver = makeDriver(transports, { conversationMode: "direct" });
const runTurn = async (
session: Awaited<ReturnType<typeof driver.openSession>>,
transport: FakeCodexTransport,
turnId: string,
total: { inputTokens: number; outputTokens: number },
) => {
await session.startTurn({ message: { role: "user", text: turnId } });
transport.push("thread/tokenUsage/updated", {
threadId: "thread-1", turnId, tokenUsage: { total, last: total },
});
transport.push("turn/completed", {
threadId: "thread-1", turn: { id: turnId, status: "completed", items: [] },
});
const events = await collectUntilTerminal(session.events());
const usage = events.filter(event => event.payload.kind === "usage").at(-1);
return (usage?.payload.usage as { runDelta: unknown }).runDelta;
};
const coldRestart = async (
session: Awaited<ReturnType<typeof driver.openSession>>,
transport: FakeCodexTransport,
runId: string,
turnId: string,
) => {
const saved = JSON.parse(JSON.stringify(await session.snapshot!()));
await session.close();
transport.readResponse = {
thread: { id: "thread-1", sessionId: "provider-session-1", cwd: WORKSPACE, turns: [] },
};
transport.turnStartResponse = Promise.resolve({
turn: { id: turnId, status: "inProgress", items: [] },
});
const recovered = await driver.recoverSession({
...saved, runId, activeTurnId: null, terminalTurns: [],
codexUsageBaseline: codexUsageBaselineForNextRun(saved.codexUsageBaseline),
});
expect(recovered.recovered).toBe(true);
return recovered.session!;
};
let session = await driver.openSession({
runId: "first", normalizedSessionId: "cold-restart-usage", workingDirectory: WORKSPACE,
});
await expect(runTurn(session, transports[0]!, "turn-1", { inputTokens: 100, outputTokens: 10 }))
.resolves.toEqual({ inputTokens: 100, outputTokens: 10 });
session = await coldRestart(session, transports[1]!, "second", "turn-2");
await expect(runTurn(session, transports[1]!, "turn-2", { inputTokens: 140, outputTokens: 16 }))
.resolves.toEqual({ inputTokens: 40, outputTokens: 6 });
session = await coldRestart(session, transports[2]!, "third", "turn-3");
await expect(runTurn(session, transports[2]!, "turn-3", { inputTokens: 170, outputTokens: 20 }))
.resolves.toEqual({ inputTokens: 30, outputTokens: 4 });
await session.close();
});
it.each(["thread", "snapshot"] as const)(
"retains the same run delta through an active cold reconciliation: %s",
async (usageLocation) => {
@@ -33,6 +33,14 @@ export function observeCodexUsage(
return { baseline, latest, ...(state?.accounted ? { accounted: state.accounted } : {}) };
}
export function codexUsageBaselineForNextRun(state: CodexUsageBaseline): CodexUsageBaseline {
return {
baseline: { ...(state.accounted ?? state.latest) },
latest: { ...state.latest },
...(state.accounted ? { accounted: { ...state.accounted } } : {}),
};
}
export function codexRunUsage(
state: CodexUsageBaseline,
): Record<string, unknown> {
+1
View File
@@ -60,6 +60,7 @@ export {
export type { DurableRecoveryIdentity } from "./control-plane/prp-transport-types.js";
export * from "./drivers/codex/app-server-transport.js";
export * from "./drivers/codex/codex-app-server-driver.js";
export { codexUsageBaselineForNextRun } from "./drivers/codex/codex-usage-baseline.js";
export { resolveCodexCommand, resolvePinnedCodexCommand } from "./drivers/codex/codex-command.js";
export { runnerBinaryTarget } from "./live/runner-binary.js";
export * from "./drivers/opencode/opencode-server-driver.js";
@@ -10983,6 +10983,7 @@ describe("native warm session supervision", () => {
terminal: null,
terminalTurns: [],
pendingRuntimeRequests: [],
codexUsageBaseline: { baseline: { inputTokens: 40, outputTokens: 4 }, latest: { inputTokens: 100, outputTokens: 10 } },
};
const result = {
result: { summary: "completed" },
@@ -11011,6 +11012,7 @@ describe("native warm session supervision", () => {
driverSessionId: checkpoint.driverSessionId,
providerSessionId: checkpoint.providerSessionId,
activeTurnId: checkpoint.activeTurnId,
codexUsageBaseline: checkpoint.codexUsageBaseline,
}),
);
return result;
@@ -11042,6 +11044,102 @@ describe("native warm session supervision", () => {
}
});
it.each([
{ name: "every report accounted", accounted: { inputTokens: 100, outputTokens: 10 }, latest: { inputTokens: 100, outputTokens: 10 } },
{ name: "late usage not yet charged", accounted: { inputTokens: 100, outputTokens: 10 }, latest: { inputTokens: 130, outputTokens: 12 } },
{ name: "no accounted total", accounted: undefined, latest: { inputTokens: 100, outputTokens: 10 } },
])("starts a new run's usage from the prior run's total when a warm checkpoint resumes: $name", async ({ accounted, latest }) => {
const stateBase = await mkdtemp(join(tmpdir(), "paperclip-warm-next-run-usage-"));
const previousPaperclipHome = process.env.PAPERCLIP_HOME;
process.env.PAPERCLIP_HOME = stateBase;
const firstRun = {
...execution,
binding: {
...execution.binding,
runId: "run-warm-next-run-usage-first",
executionWorkspaceId: "workspace-warm-next-run-usage",
},
session: {
...execution.session,
normalizedSessionId: "session-warm-next-run-usage",
lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 20 },
},
} as NativeExecutionInputV1;
const secondRun = {
...firstRun,
binding: { ...firstRun.binding, runId: "run-warm-next-run-usage-second" },
} as NativeExecutionInputV1;
const checkpoint = {
identity: {
runId: firstRun.binding.runId,
sessionId: firstRun.session.normalizedSessionId,
companyId: firstRun.binding.companyId,
issueId: firstRun.binding.issueId,
agentId: firstRun.binding.agentId,
},
sessionId: firstRun.session.normalizedSessionId,
driverSessionId: "driver-warm-next-run-usage",
providerSessionId: "provider-warm-next-run-usage",
activeTurnId: null,
semanticResult: null,
terminal: null,
terminalTurns: [],
pendingRuntimeRequests: [],
codexUsageBaseline: { baseline: {}, latest, ...(accounted ? { accounted } : {}) },
};
const result = {
result: { summary: "completed" },
terminal: { runTerminalState: "succeeded" },
turnId: "provider-turn-warm-next-run-usage",
normalizedSessionId: firstRun.session.normalizedSessionId,
providerSessionId: checkpoint.providerSessionId,
driverKind: "test",
driverVersion: "1",
nativeEventCount: 1,
highestContiguousSourceSeq: 1,
usage: null,
};
const firstClose = vi.fn(async () => undefined);
let resumedBaseline: unknown;
state.execute
.mockReset()
.mockImplementationOnce(async (options) => {
await options.onCheckpoint?.(checkpoint);
options.onSession?.({ close: firstClose });
return result;
})
.mockImplementationOnce(async (options) => {
resumedBaseline = options.persistedSession?.codexUsageBaseline;
return result;
});
try {
await executePaperclipNativeSession({
db: leaseDb(firstRun),
execution: firstRun,
runnerInstanceId: "runner-warm-next-run-usage",
});
await vi.waitFor(() => expect(firstClose).toHaveBeenCalled(), { timeout: 500 });
await executePaperclipNativeSession({
db: leaseDb(secondRun),
execution: secondRun,
runnerInstanceId: "runner-warm-next-run-usage",
});
expect(resumedBaseline).toEqual({
baseline: accounted ?? latest,
latest,
...(accounted ? { accounted } : {}),
});
} finally {
if (previousPaperclipHome === undefined) {
delete process.env.PAPERCLIP_HOME;
} else {
process.env.PAPERCLIP_HOME = previousPaperclipHome;
}
await rm(stateBase, { recursive: true, force: true });
}
});
it("reuses one session across distinct governed runs and closes it after idle expiry", async () => {
let releaseClose!: () => void;
const closeBarrier = new Promise<void>(resolve => { releaseClose = resolve; });
@@ -57,6 +57,7 @@ import { isCloudManagedInstance } from "../cloud-instance.js";
import { dotRunnerBroker } from "../dot-runner-broker.js";
import { CURSOR_DISTRIBUTION_PINS, KIMI_DISTRIBUTION_CLOSURE_SHA256, QUALIFIED_ACPX_PROFILES, QUALIFIED_ACPX_VERSION } from "../../vendor/paperclip-runner/index.js";
import { isProviderMode } from "../../vendor/paperclip-runner/index.js";
import { codexUsageBaselineForNextRun } from "../../vendor/paperclip-runner/index.js";
import { PROVIDER_PACK_COMPATIBILITY, bundledRemoteProviderPackManifestPath, bundledRemoteRunnerBinary, runnerBinaryTarget } from "../../vendor/paperclip-runner/index.js";
import { nativeRetryCancellationEligible, rethrowNativeCancellationLockConflict, assertCancellationRequest, cancellationIntentId as callerCancellationIntentId, cancellationRequestId } from "./native-cancellation-request.js";
import { readNativePlanWait } from "./native-plan-wait.js";
@@ -7035,6 +7036,9 @@ function loadWarmNativeCheckpoint(
terminalTurns: [],
pendingRuntimeRequests: [],
governedWait: undefined,
...(envelope.snapshot.codexUsageBaseline
? { codexUsageBaseline: codexUsageBaselineForNextRun(envelope.snapshot.codexUsageBaseline) }
: {}),
};
if (path !== scopedPath || envelope.configDigest !== configDigest) {
// Upgrade the validated checkpoint atomically. When moving from a legacy
+1
View File
@@ -164,6 +164,7 @@ export const parseHarnessRuntimeRequestResolution =
export const parseNativeExecutionInput = runner.parseNativeExecutionInput;
export const buildNativeModelEnvelope = runner.buildNativeModelEnvelope;
export const isProviderMode = runner.isProviderMode;
export const codexUsageBaselineForNextRun = runner.codexUsageBaselineForNextRun;
export const parseNativeRuntimeContext = runner.parseNativeRuntimeContext;
export const parsePaperclipQuestionSet = runner.parsePaperclipQuestionSet;
export const parsePaperclipQuestionResponse =