fix(runner): keep Codex thread usage reported before a run's first turn as history (#19783)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A new Codex run can attach to a Codex thread that earlier runs used
> - In restart storm 9, a comment-wake run attached to that thread after
a restart. Before its first turn, it received the thread's usage from
the earlier run
> - The runnerd transport gave that usage an empty turn owner. The
driver rejected it as an unowned authoritative notification and failed
the run
> - The task then stopped with `native_event_replay_conflict` and needed
the next run to finish
> - This pull request keeps usage that arrives before a run's first turn
as history
> - The benefit is that a run is not lost to earlier work's usage, and
it is not billed for that usage

## Linked Issues or Issue Description

Refs #19718, #19722

**What happened?**

Restart storm 9 (seed 7111), task CHA-5 (Codex). Run A succeeded. A K2
restart fired. The comment-wake run B then started on a fresh runnerd
and attached to the same durable Codex session. Before run B started a
turn, runnerd committed a `usage.reported` event for the thread's
earlier usage. The transport rehydrated it as
`thread/tokenUsage/updated` with `turnId: ""`, because no turn was bound
yet. `classifyCodexNotification` classifies an empty id as
`invalid_authority`. The driver failed the run with "Provider
authoritative notification thread/tokenUsage/updated did not name its
execution owner." Recovery then stopped with
`native_event_replay_conflict`.

**Expected behavior**

Usage that arrives before a run's first turn does not fail the run. If
the session has no usage baseline yet, that usage is history, so the run
is not billed for it.

**Steps to reproduce**

1. Use fake Codex with `--usage-before-first-turn`. It reports thread
usage right after `thread/start`.
2. Open a driver session over the real runnerd transport. Wait until
runnerd commits the usage, then start a turn.
3. Before this change, the run fails with "did not name its execution
owner".

## What Changed

- `runnerd-codex-transport.ts` (`rehydrateRunnerdUsageNotification`):
the transport omits `turnId` when no turn is bound. It no longer sends
an empty owner.
- `codex-session-notifications.ts`: usage reported before the session's
first turn, with no baseline yet, sets the baseline (history). Usage
during or after a turn keeps its existing accounting.
- Fake Codex: new `--usage-before-first-turn` flag.
- New test `src/live/runnerd-codex-pre-turn-usage.test.ts` with real
runnerd, the real Codex driver and fake Codex.

## Verification

- Red, run twice before the fix: the test failed with
`NativeProviderTerminalFailure: Provider authoritative notification
thread/tokenUsage/updated did not name its execution owner.` This is the
live message.
- Green, run three times after the fix: the turn completes. The run's
usage `runDelta` is 7/2 (thread total 12/3 minus the 5/1 reported before
the turn).
- `npx vitest run src/drivers/codex src/live/runnerd-codex-transport
src/native-session-runtime`: 991 of 992 tests pass. One settlement test
timed out under parallel load. It passes alone (19/19), and this change
does not touch it.
- `pnpm run typecheck` in `packages/paperclip-runner` passes. `cargo
fmt` is clean.
- Not run: the live restart-chaos replay. It needs provider credentials,
which this session does not load.

## Risks

- Low risk. The accounting change applies only when no turn is active,
no turn is pending, the session has no settled turns and there is no
baseline. In that state, the usage cannot belong to this run's work.
- The test reproduces the failing shape (usage committed before the
first turn) with a fresh session. It does not reproduce the exact
restart and run-attach sequence from the storm.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, used through Claude
Code with tool use (shell, file editing). Default context window and
reasoning mode.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <[email protected]>

packages/paperclip-runner/runner/crates/runner-core/src/bin/fake-codex-app-server.rs

@@ -1334,6 +1334,14 @@ fn run() -> Result<(), Box<dyn std::error::Error>> {
                         "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 {

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 };

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<ReturnType<typeof driver.openSession>> | 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<string, number> } }).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);

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),