diff --git a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md b/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md index 8b36618d840..81a26008517 100644 --- a/packages/paperclip-runner/runner/DURABLE_TRANSPORT.md +++ b/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 diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts index e81f64320c7..cbd15ea79ac 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-attachment-recovery.test.ts +++ b/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([ 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([ 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([ 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([ (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([ }), ); 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"]); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-warm-attachment.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-warm-attachment.test.ts index 083290d5fae..f6861fb0cec 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport-warm-attachment.test.ts +++ b/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 (label: string, promise: Promise) => + await Promise.race([ + promise, + new Promise((_, 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"); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 05351de090f..bf1fb16079b 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/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); 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 864d9f9903c..6df279028a4 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/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>[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 }> = []; state.execute.mockReset().mockImplementationOnce(async () => { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index ef0d14306c4..c965fa59cbc 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/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>[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, () => {