fix(runner): keep warm run.attach events on ordinary accounting (#19707)

packages/paperclip-runner/runner/DURABLE_TRANSPORT.md

@@ -115,6 +115,12 @@ run, normalized session, turn and item. History reads require A’s thread. A
 resume request returns saved state without asking Codex to resume twice. Retained events and usage
 belong to A. The authenticated `authorityRunId` identifies the current owner
 (A for same-run recovery). Losing that owner blocks new commands and dispatch.
+Only a transport that starts under a later owner sends A's events to recovery
+accounting, and only while its event identity is still A. After recovery moves
+the transport to B, a later warm attachment treats B like any live owner. A
+live warm `run.attach` also moves `authorityRunId` to B before
+the Runner flushes A's outbox. Those events use ordinary accounting, which
+drains A's usage under B.
 
 If Core loses its cached preparation, the preparer uses A’s frozen
 `recoveryLaunch.requirements`, exact identity and authenticated owner. It

packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts

@@ -11,6 +11,7 @@ import { createServer } from "node:http";
 import { tmpdir } from "node:os";
 import { join, resolve } from "node:path";
 import { expect, it, vi } from "vitest";
+import type { PrpEvent } from "../protocol/replay-contract.js";
 import { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js";
 import * as durableControlPlane from "../control-plane/durable-prp-control-plane.js";
 import { createCapabilityRunnerdCodexTransport, readRunnerdArtifactBinding } from "./runnerd-codex-transport.js";
@@ -81,12 +82,12 @@ it.each<WarmAttachRecoveryCase>([
     recoveryFault: "none",
     externalArtifact: true,
   },
-  {
-    lossPoint: "after-confirmation",
+  ...["before-result", "after-result", "after-confirmation"].map((lossPoint) => ({
+    lossPoint,
     routed: true,
     recoveryFault: "none",
     ordinaryFollowup: true,
-  },
+  })),
   ...[
     "missing_snapshot",
     "rejected_snapshot",
@@ -128,6 +129,8 @@ it.each<WarmAttachRecoveryCase>([
     let recoveryClaimCurrent = true;
     let recoveryFenceActive = true;
     let recoveredOwner: string;
+    const accountingRoutes: Array<{ route: "ordinary" | "recovery"; runId: string; eventType: string }> = [];
+    let followupAttachStarted = false;
     const completionSnapshotIds: string[] = [];
     let rejectHeldCompletion: ((error: Error) => void) | undefined;
     let markHeldCompletionEntered!: () => void;
@@ -731,6 +734,20 @@ it.each<WarmAttachRecoveryCase>([
       resumed = createCapabilityRunnerdCodexTransport({
         ...options,
         get authorityRunId() { return ordinaryFollowup ? recoveredOwner : undefined; },
+        ...(ordinaryFollowup === true
+          ? {
+              onAccountingEvent: async (event: PrpEvent) => {
+                accountingRoutes.push({ route: "ordinary", runId: event.runId, eventType: event.eventType });
+              },
+              // After recovery, the recovered owner is this transport's own
+              // run. A later warm attachment must not report it as a source;
+              // the executor rejects source accounting without a source.
+              onRecoveryAccountingEvent: async (event: PrpEvent) => {
+                accountingRoutes.push({ route: "recovery", runId: event.runId, eventType: event.eventType });
+                if (followupAttachStarted) throw new Error("native_recovery_usage_source_unproven");
+              },
+            }
+          : {}),
         ...(externalArtifact ? {
           runnerBinary: "/controller-missing-runner",
           runnerArtifact: readRunnerdArtifactBinding(options.runnerBinary),
@@ -1056,6 +1073,8 @@ it.each<WarmAttachRecoveryCase>([
           (await within(resumed.transport.request("thread/read", {}))).thread,
         ).toMatchObject(thread);
         const thirdRunId = randomUUID();
+        const routesBeforeFollowup = accountingRoutes.length;
+        followupAttachStarted = true;
         recoveredOwner = thirdRunId;
         await within(
           resumed.transport.attachRun!({
@@ -1065,6 +1084,11 @@ it.each<WarmAttachRecoveryCase>([
           }),
         );
         expect(resumedCore.store.state.identity.runId).toBe(thirdRunId);
+        if (ordinaryFollowup === true) {
+          const followupRoutes = accountingRoutes.slice(routesBeforeFollowup);
+          expect(followupRoutes.filter((entry) => entry.route === "recovery")).toEqual([]);
+          expect(followupRoutes).toContainEqual({ route: "ordinary", runId: desired.runId, eventType: "run.attached" });
+        }
         expect(
           (await readFile(callsPath, "utf8")).trim().split(/\r?\n/),
         ).toEqual([...calls, "thread/goal/get"]);

packages/paperclip-runner/src/live/runnerd-codex-transport-warm-attachment.test.ts

@@ -709,6 +709,106 @@ it("rotates PRP authority in place for a warm cross-run attachment", async () =>
   }
 }, 30_000);
 
+it("keeps prior-run events on ordinary accounting when the owner moves ahead for a warm attachment", async () => {
+  const stateDirectory = await mkdtemp(join(tmpdir(), "runnerd-warm-owner-"));
+  const first = {
+    runnerInstanceId: "runner-warm-owner",
+    environmentLeaseId: "lease-warm-owner",
+    runId: "run-warm-owner-first",
+    normalizedSessionId: "session-warm-owner",
+    turnId: "turn-warm-owner-first",
+    itemId: "item-warm-owner-first",
+  };
+  // A plugin Runner owner follows the rebound execution before the live
+  // transport attaches it, so the owner is ahead during run.attach.
+  let owner = first.runId;
+  const ordinary: PrpEvent[] = [];
+  const recovery: PrpEvent[] = [];
+  const bundle = createCapabilityRunnerdCodexTransport({
+    runnerBinary: defaultCapabilityRunnerdBinary(),
+    codexCommand: fakeCodex,
+    codexArgs: fakeCodexArgs(stateDirectory),
+    stateDirectory,
+    prpIdentity: first,
+    lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 },
+    get authorityRunId() { return owner; },
+    onAccountingEvent: async (event) => { ordinary.push(structuredClone(event)); },
+    // A live warm transport has no recovery source. The executor rejects
+    // source accounting for it, and the rejection closes the connection.
+    onRecoveryAccountingEvent: async (event) => {
+      recovery.push(structuredClone(event));
+      throw new Error("native_recovery_usage_source_unproven");
+    },
+  });
+  bundle.transport.setServerRequestHandler(async () => ({
+    success: true,
+    contentItems: [],
+  }));
+  const within = async <T>(label: string, promise: Promise<T>) =>
+    await Promise.race([
+      promise,
+      new Promise<never>((_, reject) =>
+        setTimeout(() => reject(new Error(`${label} timeout`)), 10_000),
+      ),
+    ]);
+  let passed = false;
+  try {
+    await within("initialize", bundle.transport.request("initialize", {}));
+    await within(
+      "thread start",
+      bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] }),
+    );
+    const notifications = bundle.transport
+      .notifications()
+      [Symbol.asyncIterator]();
+    const complete = async (label: string) => {
+      await within(
+        `${label} turn start`,
+        bundle.transport.request("turn/start", {
+          input: [{ type: "text", text: label }],
+        }),
+      );
+      const deadline = Date.now() + 5_000;
+      while (Date.now() < deadline) {
+        const next = await within(`${label} notification`, notifications.next());
+        if (next.value?.method === "turn/completed") return;
+      }
+      throw new Error(`${label} completion timeout`);
+    };
+    await complete("first run");
+    const runnerPid = bundle.evidence().runnerPid;
+
+    for (const runId of ["run-warm-owner-second", "run-warm-owner-third"]) {
+      const prior = owner;
+      const seen = ordinary.length;
+      owner = runId;
+      await within(
+        `attach ${runId}`,
+        bundle.transport.attachRun!({
+          runId,
+          turnId: `turn-${runId}`,
+          itemId: `item-${runId}`,
+        }),
+      );
+      // The Runner flushes the prior run's outbox before it returns the
+      // run.attach result. Those events stay ordinary accounting.
+      expect(ordinary.slice(seen).filter((event) => event.runId === prior)
+        .map((event) => event.eventType)).toContain("run.attached");
+      await complete(runId);
+    }
+
+    expect(recovery).toEqual([]);
+    expect(bundle.evidence()).toMatchObject({ runnerPid, runnerExited: false });
+    passed = true;
+  } finally {
+    // Keep the assertion failure visible when a stuck attachment also
+    // prevents a clean close.
+    const closing = bundle.transport.close();
+    await (passed ? closing : closing.catch(() => undefined));
+    await rm(stateDirectory, { recursive: true, force: true });
+  }
+}, 60_000);
+
 it("recovers an active second warm business run after deleting its controller context assets", async () => {
   const root = await mkdtemp(join(tmpdir(), "runnerd-warm-second-active-"));
   const firstRoot = join(root, "first-controller"), nextRoot = join(root, "next-controller");

packages/paperclip-runner/src/live/runnerd-codex-transport.ts

@@ -3834,6 +3834,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
   #eventSourceSeq = 0;
   #eventIdentity: DurableRecoveryIdentity | null = null;
   #sourceRecoveryIdentity: DurableRecoveryIdentity | null = null;
+  // The event identity this transport's core started under while a later run
+  // already owned the session (ordinary source recovery). The owner also moves
+  // ahead during a live warm run attachment; that alone never makes the prior
+  // run a source. Any later move of #eventIdentity ends source accounting.
+  #retainedSourceIdentity: DurableRecoveryIdentity | null = null;
   #sourceAttachmentDeadline: number | null = null;
   #pendingWarmRecoveryCompletion: {
     transitionId: string;
@@ -4087,6 +4092,22 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       && this.#authorityRunId() !== identity.runId;
   }
 
+  /** The source run whose committed events need recovery accounting, if any. */
+  #sourceAccountingIdentity(): DurableRecoveryIdentity | undefined {
+    if (this.#sourceRecoveryIdentity) return this.#sourceRecoveryIdentity;
+    // A live warm attachment keeps this transport's prior run on ordinary
+    // accounting. The executor drains that run's usage under the new owner.
+    const retained = this.#retainedSourceIdentity;
+    const identity = this.#eventIdentity ?? this.options.prpIdentity;
+    if (!retained || !identity || !recoveryIdentityMatches(identity, retained) || !this.#retainsSourceInputs()) return undefined;
+    return identity;
+  }
+
+  #captureRetainedSourceIdentity(): void {
+    const identity = this.#eventIdentity ?? this.options.prpIdentity;
+    this.#retainedSourceIdentity = identity && this.#retainsSourceInputs() ? structuredClone(identity) as DurableRecoveryIdentity : null;
+  }
+
   evidence(): CapabilityRunnerdProcessEvidence {
     return structuredClone(this.#evidence);
   }
@@ -4799,6 +4820,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       this.#assertStartupOpen();
       if (retainingSource) await core.flushCheckpoint();
       this.#eventIdentity = structuredClone(desired);
+      this.#retainedSourceIdentity = null;
       this.#eventSourceSeq = 0;
       this.#deferredTurnStartEvents = [];
       this.#durableTurnId = desired.turnId;
@@ -5782,6 +5804,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       ? params.dynamicTools.map(record)
       : [];
     const authorityRunId = this.#authorityRunId;
+    this.#captureRetainedSourceIdentity();
     const core = new DurablePrpControlPlane({
       stateDirectory: resolve(this.#root, "control-plane"),
       identity,
@@ -5793,7 +5816,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
       recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call),
       onCommittedEvent: async (event) => {
-        const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined);
+        const source = this.#sourceAccountingIdentity();
         if (source && event.runId === source.runId) {
           if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable");
           await this.options.onRecoveryAccountingEvent?.(event, source);
@@ -6293,6 +6316,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
           )
         : undefined;
     const authorityRunId = this.#authorityRunId;
+    this.#captureRetainedSourceIdentity();
     const core = new DurablePrpControlPlane({
       stateDirectory: controlPlaneDirectory,
       identity,
@@ -6304,7 +6328,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
       onSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : (call) => this.#handleSemanticToolInput(call),
       recoverSemanticToolInput: this.options.recoveryPurpose === "stop_cleanup" ? undefined : call => this.#recoverSemanticToolInput(call),
       onCommittedEvent: async (event) => {
-        const source = this.#sourceRecoveryIdentity ?? (this.#retainsSourceInputs() ? this.#eventIdentity ?? this.options.prpIdentity : undefined);
+        const source = this.#sourceAccountingIdentity();
         if (source && event.runId === source.runId) {
           if (!this.options.onRecoveryAccountingEvent && this.options.onAccountingEvent) throw new Error("native_runner_source_accounting_unavailable");
           await this.options.onRecoveryAccountingEvent?.(event, source);

server/src/services/native-runtime/native-session-executor.test.ts

@@ -8586,6 +8586,46 @@ describe("native terminal-turn accounting", () => {
     }
   });
 
+  it("acknowledges source lifecycle events without a proven recovery usage source", async () => {
+    const directory = await mkdtemp(join(tmpdir(), "native-plugin-recovery-lifecycle-"));
+    const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
+    process.env.PAPERCLIP_RUNNER_STATE_DIR = directory;
+    const current = { ...execution, provider: { kind: "codex", model: "gpt-6-astra" },
+      binding: { ...execution.binding, runId: "plugin-lifecycle-run", executionWorkspaceId: "plugin-lifecycle-workspace" },
+      session: { ...execution.session, normalizedSessionId: "plugin-lifecycle-session" } } as NativeExecutionInput;
+    const launcher = vi.fn(), registration = vi.fn();
+    const driver = { authorityRunId: current.binding.runId, prepareLaunch: vi.fn(), runnerProcessLauncher: launcher, controlPlaneRegistration: registration };
+    const owner = vi.spyOn(pluginTaskRuntime, "createPluginRunnerTaskOwner").mockReturnValue(
+      { launcher, registration, driver, readWorkspaceFile: vi.fn(), isRetired: () => false, rebind: vi.fn() } as never);
+    const task = vi.fn(async (operation: unknown) => {
+      environmentTaskOperationSchema.parse(operation);
+      return { kind: "description", launchPreparationVersion: 1, projectAuthCheckVersion: 1, taskId: "task",
+        runner: { version: "fixture", digest: `sha256:${"a".repeat(64)}` } };
+    });
+    const capture = vi.fn(async (_event: PrpEvent) => {});
+    state.createBackend.mockClear(); state.createTransport.mockClear();
+    try {
+      // A fresh session has no durable source identity, so it has no recovery usage delivery.
+      await createRunnerdBackend({ db: leaseDb(current), execution: current, runnerInstanceId: "plugin-producer", runnerIngressAuthorized: true,
+        onAccountingEvent: capture, runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "plugin",
+          leaseId: "plugin-lifecycle-lease", remoteCwd: "/unused", runnerTask: task } });
+      state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!();
+      const recover = state.createTransport.mock.calls.at(-1)![0].onRecoveryAccountingEvent!;
+      const source: Parameters<NonNullable<RunnerTransportOptions["onRecoveryAccountingEvent"]>>[1] = { runnerInstanceId: "plugin-producer", environmentLeaseId: "plugin-lifecycle-lease", runId: "plugin-lifecycle-source",
+        normalizedSessionId: "plugin-lifecycle-session", turnId: "turn-source", itemId: "item-source" };
+      for (const eventType of ["session.resumed", "session.capabilities.updated", "session.goal.snapshot", "run.attached"]) {
+        await expect(recover({ eventType, runId: source.runId, sourceInstanceId: "plugin-producer" } as PrpEvent, source)).resolves.toBeUndefined();
+      }
+      await expect(recover({ eventType: "usage.reported", runId: source.runId, sourceInstanceId: "plugin-producer" } as PrpEvent, source))
+        .rejects.toThrow("native_recovery_usage_source_unproven");
+      expect(capture).not.toHaveBeenCalled();
+    } finally {
+      owner.mockRestore();
+      if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
+      await rm(directory, { recursive: true, force: true });
+    }
+  });
+
   it("uses the canonical recorder for partial and completed usage when no onUsage callback is supplied", async () => {
     const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
     state.execute.mockReset().mockImplementationOnce(async () => {

server/src/services/native-runtime/native-session-executor.ts

@@ -14322,6 +14322,10 @@ async function createRunnerdBackendWithinSessionClaim(
       await current.acceptAccountingEvent(event);
     } : undefined,
     onRecoveryAccountingEvent: input.onAccountingEvent ? async (event: PrpEvent, source: Parameters<NonNullable<typeof recoveryUsage>>[1]) => {
+      // Only usage needs a proven recovery source. A lifecycle event that
+      // fails here closes the Runner connection without an ACK, and the
+      // Runner then replays that event without end.
+      if (!isNativeUsageAccountingEvent(event)) return;
       const current = sessionToolAuthorityEpochs.get(sessionScopeId);
       if (!current) throw new Error("native_recovery_usage_source_unproven");
       await current.acceptRecoveryAccountingEvent(event, () => {