diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs index d34e98fe922..929ae1f74e7 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs @@ -1334,6 +1334,14 @@ fn run() -> Result<(), Box> { "params": {"threadId": state.thread_id, "goal": state.goal} }))?; } + // Thread usage that predates this run's first turn, as a + // shared thread reports after an earlier run's work. + if args.iter().any(|arg| arg == "--usage-before-first-turn") { + send(json!({"method": "thread/tokenUsage/updated", "params": { + "threadId": state.thread_id, + "tokenUsage": {"total": {"inputTokens": 5, "outputTokens": 1}, "last": {"inputTokens": 5, "outputTokens": 1}} + }}))?; + } } "thread/resume" => { if reject_missing_resume_state { diff --git a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts index 3e35bef3d68..cfac70f602d 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-session-notifications.ts @@ -655,8 +655,12 @@ async function mapNotificationBody(state: CodexSessionState, notification: Codex const runDeltaComplete = validCount(reported.inputTokens) && validCount(reported.outputTokens) && Object.entries(reported).every(([key, value]) => !key.endsWith("Tokens") || validCount(value)) && Object.keys(state.codexUsageBaseline?.latest ?? {}).every(key => !key.endsWith("Tokens") || validCount(reported[key])); + // Usage reported before this session's first turn, with no baseline, + // is the thread's earlier work. Record it as history, not this run's. + const beforeFirstTurn = state.activeTurnId === null && state.codexUsageBaseline === null + && !state.turnStartPending && state.terminalTurns.size === 0; if (Object.keys(reported).length > 0) { - state.codexUsageBaseline = observeCodexUsage(state.codexUsageBaseline, reported, false); + state.codexUsageBaseline = observeCodexUsage(state.codexUsageBaseline, reported, beforeFirstTurn); state.usageSnapshot = { ...state.usageSnapshot, ...codexRunUsage(state.codexUsageBaseline) }; } state.usageSnapshot = { ...state.usageSnapshot, runDeltaComplete }; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-pre-turn-usage.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-pre-turn-usage.test.ts new file mode 100644 index 00000000000..d9e2bfa9e47 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-pre-turn-usage.test.ts @@ -0,0 +1,62 @@ +import { mkdir, mkdtemp, realpath, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import { CodexAppServerDriver } from "../drivers/codex/codex-app-server-driver.js"; +import { collectUntilTerminal } from "../drivers/codex/codex-app-server-driver.test-support.js"; +import { createCodexTaskEnvelope } from "../contracts/codex.js"; +import { createRunnerdCodexTransport, defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js"; + +it("keeps thread usage reported before a run's first turn as history, not a binding failure", async () => { + // RC-61: a run attached to a shared Codex thread received the thread's usage + // from earlier work before its own first turn. The transport bound it to an + // empty turn, and the driver failed the run as an unowned authoritative + // notification. That usage belongs to earlier work: keep it as the baseline. + const root = await realpath(await mkdtemp(join(tmpdir(), "runnerd-pre-turn-usage-"))); + const workspace = join(root, "workspace"); + await Promise.all([mkdir(workspace), mkdir(join(root, "home"))]); + const environment = { PATH: process.env.PATH, HOME: join(root, "home"), CODEX_HOME: join(root, "codex-home") }; + const bundle = createRunnerdCodexTransport({ + runnerBinary: process.env.PAPERCLIP_ATTACH_TRANSITION_RUNNER ?? defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "provider.json"), "--usage-before-first-turn"], + stateDirectory: root, environment, + prpIdentity: { runnerInstanceId: "runner-pre-turn-usage", environmentLeaseId: "lease-pre-turn-usage", + runId: "pre-turn-usage", normalizedSessionId: "session-pre-turn-usage", turnId: "turn-pre-turn-usage", + itemId: "item-pre-turn-usage" }, + }); + const driver = new CodexAppServerDriver({ + approvalPolicy: "never", includeCollaborationModeInstructions: false, environment, + taskEnvelope: createCodexTaskEnvelope({ objective: "Finish.", contractRevision: "1", + criteria: [{ id: "objective", requirement: "Finish." }], constraints: [] }), + transportFactory: () => bundle.transport, + }); + let session: Awaited> | undefined; + try { + session = await driver.openSession({ + runId: "pre-turn-usage", normalizedSessionId: "session-pre-turn-usage", workingDirectory: workspace, + }); + // The driver consumes the thread usage while no turn is active, and keeps + // it as history: the thread total, with nothing billed to this run. + const opened = session; + await vi.waitFor(async () => { + expect(await opened.usage!()).toMatchObject({ + total: { inputTokens: 5, outputTokens: 1 }, + runDelta: { inputTokens: 0, outputTokens: 0 }, + }); + }, { timeout: 15_000, interval: 50 }); + await session.startTurn({ message: { role: "user", text: "Finish." } }); + const events = await collectUntilTerminal(session.events()); + expect(events.at(-1)?.eventType).toBe("turn.completed"); + const usage = events.filter(event => event.eventType === "item.completed" && event.payload.kind === "usage") + .map(event => (event.payload as { usage: { runDelta: Record } }).usage.runDelta); + expect(usage.length).toBeGreaterThan(0); + // The turn's thread total is 12/3. The 5/1 reported before the turn is + // history, so this run is billed only for the difference. + expect(usage.at(-1)).toMatchObject({ inputTokens: 7, outputTokens: 2 }); + } finally { + await session?.close({ reason: "test complete" }).catch(() => undefined); + await bundle.transport.close().catch(() => undefined); + await rm(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 8b7031df883..c5cdae6cb56 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -2114,7 +2114,10 @@ export function rehydrateRunnerdUsageNotification( // providerSessionId can be a distinct backend/agent session (ACPX is one // example), so it must remain metadata rather than become a thread bind. threadId: openedThreadId, - turnId: activeTurnId, + // Before this run's first turn there is no turn to own the usage. A + // shared thread can report usage from earlier work then; an empty owner + // would make the driver reject it as unowned authority. + ...(activeTurnId ? { turnId: activeTurnId } : {}), tokenUsage: { total: record(rawParams.cumulative), runDelta: record(rawParams.runDelta),