From 737db732755e76f41b88e6b377397c2b2ea5a917 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 4 Nov 2026 18:55:14 -0800 Subject: [PATCH] fix(native): measure resumed run usage from the prior run's total (#19161) --- doc/execution-semantics.md | 6 ++ .../drivers/codex/codex-harness-session.ts | 7 +- .../drivers/codex/codex-resume-usage.test.ts | 56 +++++++++++ .../src/drivers/codex/codex-usage-baseline.ts | 8 ++ packages/paperclip-runner/src/index.ts | 1 + .../native-session-executor.test.ts | 98 +++++++++++++++++++ .../native-runtime/native-session-executor.ts | 4 + server/src/vendor/paperclip-runner/index.ts | 1 + 8 files changed, 176 insertions(+), 5 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index ea73364ec3c..31a0b82d3e5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -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 diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index e2a3a9e8557..1c60274c6af 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -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; diff --git a/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts b/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts index f49d899b6ef..7e986c46baf 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-resume-usage.test.ts @@ -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>, + 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>, + 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) => { diff --git a/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts b/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts index b6d6266e58c..2bef2c5c396 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-usage-baseline.ts @@ -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 { diff --git a/packages/paperclip-runner/src/index.ts b/packages/paperclip-runner/src/index.ts index c6c5de0ddea..5f40128c54a 100644 --- a/packages/paperclip-runner/src/index.ts +++ b/packages/paperclip-runner/src/index.ts @@ -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"; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index cfad7a837cb..e69c2b0876d 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -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(resolve => { releaseClose = resolve; }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7b5335794a0..54ed773ea8b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -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 diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index de94cf68272..f72c65e333d 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -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 =