diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b400dc6bd3a..b8aa995d37f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3832,6 +3832,18 @@ A successor claims its existing native coordinator and starts renewal before retiring a superseded warm session. An unclaimed preparation cannot authorize the predecessor's checkpoint writes. Shared attachment staging follows verified retirement; a failed claim or Stop cannot close that predecessor or stage files. + +A completed native turn keeps its exact coordinator claim and the existing +execution lease renewal until its required warm checkpoint, session close and +instruction collection have joined. It then stops and joins renewal and releases +only that claim. A checkpoint or collection failure keeps its initiating error +while the same fenced release runs; a changed controller or owner is never +cleared. Restart detachment retains its existing recovery owner and its typed +outcome takes precedence over checkpoint, collection or renewal-stop errors. +Only closed cleanup-failure flags accompany that outcome; raw secondary errors +are not attached. Renewal exists +only for this active obligation and stops on normal completion or unwinding; it +does not create an idle worker or another recovery budget. Recovery also preserves the implicit isolated-home default when binding admitted configuration, so an unchanged home does not invalidate the retained session. Explicit home-mode changes retain their normal configuration-change checks. diff --git a/scripts/__tests__/run-vitest-stable-shard.test.mjs b/scripts/__tests__/run-vitest-stable-shard.test.mjs index 0d42c06cb98..1494fb0762c 100644 --- a/scripts/__tests__/run-vitest-stable-shard.test.mjs +++ b/scripts/__tests__/run-vitest-stable-shard.test.mjs @@ -322,6 +322,7 @@ const nativeRunnerSuitePath = "server/src/services/native-runtime/native-codex-runner.integration.test.ts"; const ingressRetrySuitePath = "server/src/services/native-runtime/native-ingress-bootstrap-retry.integration.test.ts"; const usageDeliverySuitePath = "server/src/services/native-runtime/native-usage-delivery.integration.test.ts"; +const warmCheckpointOwnershipSuitePath = "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts"; const dotRunnerSuitePath = "server/src/__tests__/dot-runner.test.ts"; const restartRecoverySuitePath = "server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts"; @@ -369,7 +370,8 @@ test("12 PR without-chat shards plus the dedicated chat and native-runner lanes assert.ok(!files.includes(dotRunnerSuitePath)); assert.ok(!files.includes(restartRecoverySuitePath)); assert.ok(!files.includes(usageDeliverySuitePath)); - assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); + assert.ok(!files.includes(warmCheckpointOwnershipSuitePath)); + assert.deepEqual([...files, chatSuitePath, nativeRunnerSuitePath, ingressRetrySuitePath, dotRunnerSuitePath, restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, warmCheckpointOwnershipSuitePath, "server/src/services/native-runtime/plugin-runner-probe.test.ts"].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); const defaultRun = dryRunJson([], prEnv); assert.ok(defaultRun.generalServerSuiteCount === full.generalServerSuiteCount); @@ -389,6 +391,7 @@ for (const [caller, envOverrides] of [["Release", { GITHUB_WORKFLOW: "Release" } assert.ok(files.includes(nativeRunnerSuitePath)); assert.ok(files.includes(dotRunnerSuitePath)); assert.ok(files.includes(restartRecoverySuitePath)); + assert.ok(files.includes(warmCheckpointOwnershipSuitePath)); assert.deepEqual([...files, chatSuitePath].sort(), full.selectedGeneralServerSuites.sort()); assert.equal(new Set(files).size, files.length); }); @@ -421,6 +424,7 @@ test("the native-runner lane runs exactly the cargo-dependent suites", () => { restartRecoverySuitePath, ...restartWaitSuites, usageDeliverySuitePath, + warmCheckpointOwnershipSuitePath, ]); }); diff --git a/scripts/run-vitest-stable.mjs b/scripts/run-vitest-stable.mjs index fe5fdb4c22a..fa74e5eef99 100644 --- a/scripts/run-vitest-stable.mjs +++ b/scripts/run-vitest-stable.mjs @@ -103,6 +103,7 @@ const nativeRunnerSuites = [ "server/src/services/native-runtime/native-runner-restart-checkpoint-cost.integration.test.ts", "server/src/services/native-runtime/native-runner-restart-diskless.integration.test.ts", "server/src/services/native-runtime/native-usage-delivery.integration.test.ts", + "server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts", ]; // In the PR workflow (pr.yml, the caller of pr-trusted.yml — reusable // workflows inherit the caller's GITHUB_WORKFLOW), the last Verify Paperclip diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index ef3ebc8b855..509213ca403 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -392,17 +392,18 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( onExecutionLeaseAcquired: () => preparation!.stop(), instructionWorkingCopy: { hasChanges: async () => { - // Completion has released the coordinator, but the startup still - // owns its session. A leftover preparation timer would detach it. const [settled] = await fixture.db.select().from(nativeRunFinalizations) .where(eq(nativeRunFinalizations.runId, fixture.runId)); - expect(settled.leaseOwner).toBeNull(); + expect(settled).toMatchObject({ leaseOwner: claim.leaseOwner, attempt: claim.providerAttempt }); await new Promise(resolve => setTimeout(resolve, 21_000)); return false; }, collectStopped: async () => undefined, }, }).finally(() => preparation?.stop().catch(() => undefined)); + const [released] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(released.leaseOwner).toBeNull(); expect(logs.filter(line => line.includes("controller authority detached for restart"))).toHaveLength(1); const recoveredCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] }); expect(recoveredCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(1); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c0b38645082..2432068f8ec 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -328,9 +328,12 @@ class NativePostResultFailureError extends Error { } export class NativeControllerDetachedForRestartError extends Error { - constructor() { + readonly diagnostic?: Readonly<{ stage: "post_result_cleanup"; postExecutionCleanupFailed: boolean; renewalStopFailed: boolean }>; + + constructor(diagnostic?: { stage: "post_result_cleanup"; postExecutionCleanupFailed: boolean; renewalStopFailed: boolean }) { super("native_controller_detached_for_restart"); this.name = "NativeControllerDetachedForRestartError"; + if (diagnostic) this.diagnostic = Object.freeze(diagnostic); } } @@ -10506,7 +10509,6 @@ async function executePaperclipNativeSessionWithinScope( expectedCurrentWakeComments, ); } - await leaseRenewal.stop(); await trace.record({ name: "native.result.finalize", parentName: "task.settle", @@ -11141,62 +11143,117 @@ async function executePaperclipNativeSessionWithinScope( await timeController("github_stop", () => githubAccess!.stop()); } } - if ( - planSynchronizations.length === 0 && - "executionMode" in input.execution && - input.execution.executionMode === "plan" - ) { - const markdown = semanticProviderPlanMarkdown( - native.result as unknown as Record, - ); - if (markdown) { - const digest = createHash("sha256").update(markdown).digest("hex"); - await recordPlanSynchronization({ - sourceEventId: `semantic-plan:${input.execution.binding.runId}:${digest}`, - ...(native.turnId ? { turnId: native.turnId } : {}), - eventType: "plan.updated", - payload: { - schema: "paperclip.plan.updated.v1", - planId: `semantic:${native.turnId ?? input.execution.binding.runId}`, - revision: 1, - complete: true, - markdown, - source: "semantic_result_artifact", - }, - }); + let postExecutionFailed = false; + let postExecutionCleanupFailed = false; + try { + if (warmSessionId !== null && lifecyclePolicy.mode === "warm") { + const retainSession = native.terminal.runTerminalState === "succeeded" && native.result.continuation?.kind !== "case_handoff"; + const instructionCopy = input.instructionWorkingCopy; + const ownedSession = warmNativeSessions.get(warmSessionId); + let collectInstructions: boolean; + try { + collectInstructions = Boolean(instructionCopy && ownedSession?.ownerToken === warmSessionOwnerToken && + (instructionCopy.checkpointWarm ? !await timeController("instruction_checkpoint", () => instructionCopy.checkpointWarm!()) : await timeController("instruction_changes", () => instructionCopy.hasChanges()))); + } catch (error) { + await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true) + .catch(() => undefined); + throw error; + } + const collectedByOwner = ownedSession?.instructionCopy?.collectStopped === instructionCopy?.collectStopped + || ownedSession?.instructionWorkingCopy?.collectStopped === instructionCopy?.collectStopped; + if (collectInstructions && ownedSession) { + ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection"; + } + await timeController("warm_release", () => releaseWarmNativeSession( + warmSessionId, + warmSessionOwnerToken, + lifecyclePolicy.idleTimeoutMs, + !retainSession, + )); + if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped()); + } + assertControllerActive(); + if ( + planSynchronizations.length === 0 && + "executionMode" in input.execution && + input.execution.executionMode === "plan" + ) { + const markdown = semanticProviderPlanMarkdown( + native.result as unknown as Record, + ); + if (markdown) { + const digest = createHash("sha256").update(markdown).digest("hex"); + await recordPlanSynchronization({ + sourceEventId: `semantic-plan:${input.execution.binding.runId}:${digest}`, + ...(native.turnId ? { turnId: native.turnId } : {}), + eventType: "plan.updated", + payload: { + schema: "paperclip.plan.updated.v1", + planId: `semantic:${native.turnId ?? input.execution.binding.runId}`, + revision: 1, + complete: true, + markdown, + source: "semantic_result_artifact", + }, + }); + } + } + } catch (error) { + postExecutionFailed = true; + postExecutionCleanupFailed = !(error instanceof NativeControllerDetachedForRestartError); + throw error; + } finally { + let renewalStopFailed = false; + try { + try { + await leaseRenewal.stop(); + } catch (error) { + renewalStopFailed = true; + throw error; + } + if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { + throw new NativeControllerDetachedForRestartError(); + } + const releaseNow = new Date(); + const released = await timeController("coordinator_release", () => input.db.transaction(async tx => { + await notifyNativeFinalizationWork(tx, input.execution.binding); + return tx.update(nativeRunFinalizations) + .set({ + leaseOwner: null, + leaseExpiresAt: null, + recoveryState: null, + recoveryRequestId: null, + updatedAt: releaseNow, + }) + .where( + and( + eq(nativeRunFinalizations.runId, input.execution.binding.runId), + eq(nativeRunFinalizations.companyId, input.execution.binding.companyId), + eq(nativeRunFinalizations.issueId, input.execution.binding.issueId), + eq(nativeRunFinalizations.leaseOwner, leaseOwner), + eq(nativeRunFinalizations.attempt, attempt), + eq(nativeRunFinalizations.controllerBootId, controller.bootId), + eq(nativeRunFinalizations.controllerPid, controller.pid), + eq( + nativeRunFinalizations.controllerProcessStartedAt, + controller.processStartedAt, + ), + gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), + ), + ) + .returning({ runId: nativeRunFinalizations.runId }) + .then((rows) => rows[0] ?? null); + })); + if (!released) throw new Error("native_session_lease_lost"); + } catch (error) { + if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { + throw new NativeControllerDetachedForRestartError({ stage: "post_result_cleanup", + postExecutionCleanupFailed, renewalStopFailed }); + } + if (!postExecutionFailed) throw error; + logger.warn({ runId: input.execution.binding.runId }, "Native finalization claim release failed after an earlier error"); } } - const releaseNow = new Date(); - const released = await timeController("coordinator_release", () => input.db.transaction(async tx => { - await notifyNativeFinalizationWork(tx, input.execution.binding); - return tx.update(nativeRunFinalizations) - .set({ - leaseOwner: null, - leaseExpiresAt: null, - recoveryState: null, - recoveryRequestId: null, - updatedAt: releaseNow, - }) - .where( - and( - eq(nativeRunFinalizations.runId, input.execution.binding.runId), - eq(nativeRunFinalizations.companyId, input.execution.binding.companyId), - eq(nativeRunFinalizations.issueId, input.execution.binding.issueId), - eq(nativeRunFinalizations.leaseOwner, leaseOwner), - eq(nativeRunFinalizations.attempt, attempt), - eq(nativeRunFinalizations.controllerBootId, controller.bootId), - eq(nativeRunFinalizations.controllerPid, controller.pid), - eq( - nativeRunFinalizations.controllerProcessStartedAt, - controller.processStartedAt, - ), - gt(nativeRunFinalizations.leaseExpiresAt, sql`now()`), - ), - ) - .returning({ runId: nativeRunFinalizations.runId }) - .then((rows) => rows[0] ?? null); - })); - if (!released) throw new Error("native_session_lease_lost"); const finalization: NativeFinalizationResult = { schema: "paperclip.native-finalization.v1", runtimeMode: "native", @@ -11215,45 +11272,6 @@ async function executePaperclipNativeSessionWithinScope( highestContiguousSourceSeq: native.highestContiguousSourceSeq, workspaceFinalizeStatus: "pending", }; - // A following run cannot attach until the prior run's durable finalization - // is committed. Provider completion alone is not an authority boundary. - if (warmSessionId !== null && lifecyclePolicy.mode === "warm") { - // A structured failed/cancelled result completes the protocol without - // throwing. Heartbeat still stops its sandbox, so retire the provider now - // while its transport can collect files and checkpoint the suspended runner. - // Leaving it in the warm map makes the next turn retire a stopped transport. - // A different agent may next use this same task workspace. Persist this - // agent's session checkpoint, then close its provider process before the - // heartbeat release proof. Returning later resumes its own saved session. - const retainSession = native.terminal.runTerminalState === "succeeded" && native.result.continuation?.kind !== "case_handoff"; - const instructionCopy = input.instructionWorkingCopy; - const ownedSession = warmNativeSessions.get(warmSessionId); - let collectInstructions: boolean; - try { - collectInstructions = Boolean(instructionCopy && ownedSession?.ownerToken === warmSessionOwnerToken && - (instructionCopy.checkpointWarm ? !await timeController("instruction_checkpoint", () => instructionCopy.checkpointWarm!()) : await timeController("instruction_changes", () => instructionCopy.hasChanges()))); - } catch (error) { - // A checkpoint or its receipt callback can reject. Retire this owner - // before heartbeat stops the sandbox, preserving the initiating error. - await releaseWarmNativeSession(warmSessionId, warmSessionOwnerToken, lifecyclePolicy.idleTimeoutMs, true) - .catch(() => undefined); - throw error; - } - const collectedByOwner = ownedSession?.instructionCopy?.collectStopped === instructionCopy?.collectStopped - || ownedSession?.instructionWorkingCopy?.collectStopped === instructionCopy?.collectStopped; - if (collectInstructions && ownedSession) { - // Keep the unchanged warm path intact. A changed private instruction copy - // requires the existing checkpoint-and-close boundary before collection. - ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection"; - } - await timeController("warm_release", () => releaseWarmNativeSession( - warmSessionId, - warmSessionOwnerToken, - lifecyclePolicy.idleTimeoutMs, - !retainSession, - )); - if (collectInstructions && !collectedByOwner) await timeController("instruction_collection", () => instructionCopy!.collectStopped()); - } const providerFailure = providerFailureObservation.forTerminal(native.turnId, native.terminal); const providerExplanation = providerFailureObservation.providerFailureForTerminal(native.turnId, native.terminal); // Keep the generic failure behavior, but record why the turn failed. diff --git a/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts new file mode 100644 index 00000000000..95f0c955fe0 --- /dev/null +++ b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts @@ -0,0 +1,253 @@ +import { randomUUID } from "node:crypto"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, expect, it, vi } from "vitest"; +import { completionContracts, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; +import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; +import { createStorageService } from "../../storage/service.js"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js"; +import { defaultCapabilityRunnerdBinary } from "../../vendor/paperclip-runner/index.js"; + +type ProviderTransport = ReturnType[1]>["codexTransportFactory"]>>; +const { FakeCodexTransport } = await import(new URL("../../../../packages/paperclip-runner/src/drivers/codex/codex-app-server-driver.test-support.ts", import.meta.url).href) as { FakeCodexTransport: new () => ProviderTransport }; +const { runnerdRecoveryInternals } = await import(new URL("../../../../packages/paperclip-runner/src/live/runnerd-codex-transport.ts", import.meta.url).href) as { + runnerdRecoveryInternals: { releaseRunnerProcessOwnership(input: { runnerSettled: boolean; checkpoint: typeof state.checkpoint; + release: () => Promise; forceKill: () => Promise }): Promise }; +}; + +const state = vi.hoisted(() => ({ execute: vi.fn(), checkpoint: null as null | ((settlement: "settled" | "unsettled") => Promise), storage: null as unknown, transport: null as unknown })); +vi.mock("./native-harness-backup-storage.js", async original => ({ + ...await original(), + durableNativeHarnessStorage: () => state.storage, +})); +vi.mock("./native-runner-execution-driver.js", async original => { + const actual = await original(); + return { ...actual, createNativeRunnerExecutionDriver: (input: Parameters[0]) => { + if (input.checkpoint) state.checkpoint = input.checkpoint as typeof state.checkpoint; + return actual.createNativeRunnerExecutionDriver(input); + } }; +}); +vi.mock("../../vendor/paperclip-runner/index.js", async original => { + const actual = await original(); + return { ...actual, executeNativeSession: state.execute, + createNativeSessionBackend: (input: Parameters[0], options: Parameters[1]) => { + options?.codexTransportFactory?.(); + return actual.createNativeSessionBackend(input, { ...options, codexTransportFactory: () => state.transport as ProviderTransport }); + } }; +}); +import { closeWarmNativeSessionsForEnvironment, detachNativeSessionsForRestart, executePaperclipNativeSession, NativeControllerDetachedForRestartError } from "./native-session-executor.js"; +import * as executionLease from "./native-execution-lease.js"; +import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; +import { createPostgresWakeQueueAdapter } from "../../modules/wake-queue/adapters/postgres.js"; +import { createReleaseIssueExecution } from "../../modules/wake-queue/application/use-cases.js"; + +describeEmbeddedPostgres("managed warm checkpoint ownership", () => { + const environment = nativeRestartFixture(); + let root: string; + beforeAll(async () => { root = await mkdtemp(join(tmpdir(), "warm-checkpoint-owner-")); }); + afterAll(async () => { state.storage = null; await rm(root, { recursive: true, force: true }); }); + + it.each(["normal", "delayed", "instruction_error", "fresh_successor", "wrong_boot", "superseded_owner", "detach_during_close", "detach_during_instruction_error", "detach_during_instruction_error_and_renewal_loss"])("keeps the producing coordinator claim until the managed failed-turn close joins: %s", async scenario => { + let fixture: Omit>, "runId"> & { runId: string } = await environment.seedRun(`WARM-CHECKPOINT-OWNER-${scenario}`); + const { companyId, agentId } = environment.current; + const [contract] = await fixture.db.select().from(completionContracts).where(eq(completionContracts.issueId, fixture.issueId)); + let execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, executionWorkspaceId: randomUUID() }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: "WARM-1", title: "Retain checkpoint owner", description: null, prompt: "Return a failed turn.", workMode: "standard" }, + workspace: { cwd: root, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: fixture.native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 } }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: { ...contract.contractJson, revision: String(contract.revision) } } as NativeExecutionInputV1["completionContract"], + interactionResponses: [], credentialBindings: [], + }; + state.storage = createStorageService(createLocalDiskStorageProvider(join(root, "objects"))); + const transport = new FakeCodexTransport(); + const request = transport.request.bind(transport); + transport.request = async (method, params) => { + const response = await request(method, params); + return method === "thread/start" || method === "thread/resume" + ? { ...response, cwd: root, thread: { ...response.thread as Record, cwd: root } } + : response; + }; + state.transport = transport; + const closeFailures: string[] = []; + const ownersAtClose: Array<{ phase: string; leaseOwner: string | null; resultId: string | null; leaseExpiresAt: Date | null }> = []; + const initiatingError = new Error("fixture instruction checkpoint failed private-canary-N45"); + const invalidated = scenario === "wrong_boot" || scenario === "superseded_owner"; + const renewalLoss = scenario === "detach_during_instruction_error_and_renewal_loss"; + const instructionErrorWithDetach = scenario.startsWith("detach_during_instruction_error"); + let originalBoot: string | null = null; + let replacementOwner: string | null = null; + let enteredClose!: () => void, releaseClose!: () => void; + const closeEntered = new Promise(resolve => { enteredClose = resolve; }); + const closeGate = new Promise(resolve => { releaseClose = resolve; }); + const detached = vi.fn(async () => {}); + Object.assign(transport, { detachControllerForRestart: detached }); + const renew = executionLease.startNativeSessionExecutionLeaseRenewal; + let renewalStopped = false; + const renewal = vi.spyOn(executionLease, "startNativeSessionExecutionLeaseRenewal").mockImplementation(input => { + const owned = renew(input); + return { ...owned, stop: async () => { renewalStopped = true; await owned.stop(); } }; + }); + transport.close = async () => { + const [owner] = await fixture.db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner, + resultId: nativeRunFinalizations.resultId, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + ownersAtClose.push(owner); + expect(renewalStopped).toBe(false); + if (scenario === "detach_during_close") { enteredClose(); await closeGate; } + if (scenario === "delayed") { + await expect.poll(async () => { + const [current] = await fixture.db.select({ expires: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + return current.expires!.getTime(); + }, { timeout: 25_000, interval: 250 }).toBeGreaterThan(owner.leaseExpiresAt!.getTime() + 10_000); + expect(renewalStopped).toBe(false); + } + try { + await runnerdRecoveryInternals.releaseRunnerProcessOwnership({ runnerSettled: false, + checkpoint: state.checkpoint, release: async () => {}, forceKill: async () => {} }); + } catch (error) { closeFailures.push(String(error)); throw error; } + }; + state.execute.mockReset().mockImplementation(async options => { + const session = await options.backend.openSession({ identity: { companyId, agentId, issueId: fixture.issueId, + runId: fixture.runId, sessionId: execution.session.normalizedSessionId }, workingDirectory: root }); + await options.onSession(session); + await options.controlPlane.openRun({ identity: { companyId, agentId, issueId: fixture.issueId, runId: fixture.runId, + sessionId: execution.session.normalizedSessionId }, backendKind: "mock", sourceInstanceId: fixture.native.runnerInstanceId }); + await options.controlPlane.appendEvent({ schema: "paperclip.prp.event.v1", runId: fixture.runId, + normalizedSessionId: execution.session.normalizedSessionId, turnId: fixture.native.turnId, + sourceInstanceId: fixture.native.runnerInstanceId, sourceEventId: "provider-overload", sourceSeq: 1, + sourceKind: "runner", eventType: "turn.failed", schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), + payload: { status: "failed", error: { codexErrorInfo: "serverOverloaded" } } }); + const terminal = { schema: "paperclip.prp.terminal.v1", runTerminalState: "failed", turnTerminalState: "failed", reportedWorkDisposition: "needs_review" }; + const result = { ...CONTROL_PLANE_CONFORMANCE_RESULT, reportedWorkDisposition: "needs_review", + completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: execution.completionContract.contract.revision } }; + await options.controlPlane.completeRun({ result, terminal, turnId: fixture.native.turnId }); + return { result, terminal, turnId: fixture.native.turnId, normalizedSessionId: execution.session.normalizedSessionId, + providerSessionId: "fixture-session", driverKind: "codex_app_server", driverVersion: "fixture", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }; + }); + const environmentId = randomUUID(); + let acquisition: "created" | "resumed" = "created"; + const providerLeaseId = randomUUID(); + const runCurrent = async () => { + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.preparedProfile, nativeExecutionInput: execution } }) + .where(eq(heartbeatRuns.id, fixture.runId)); + return executePaperclipNativeSession({ db: fixture.db, execution, runnerInstanceId: fixture.native.runnerInstanceId, + useRunnerd: true, runnerIngressAuthorized: true, + runnerRemoteBinaryPath: defaultCapabilityRunnerdBinary(), runnerRemoteCodexPath: defaultCapabilityRunnerdBinary(), + runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: root, environmentId, + leaseId: randomUUID(), providerKey: "fixture", effectiveCapabilities: { runnerWebSocketIngress: true }, + sandboxLeaseAcquisition: { outcome: acquisition, providerLeaseId }, + runner: { execute: async (command: { args?: string[] }) => command.args?.[1] === "uname -s; uname -m" + ? { exitCode: 0, timedOut: false, stdout: `${process.platform === "darwin" ? "Darwin" : "Linux"}\n${process.arch === "arm64" ? "arm64" : "x86_64"}\n`, stderr: "" } + : { exitCode: 1, timedOut: false, stdout: "", stderr: "" }, syncIn: async () => {} } } as never, + instructionWorkingCopy: { runId: fixture.runId, root, checkpointWarm: async () => { + if (instructionErrorWithDetach) { enteredClose(); await closeGate; throw initiatingError; } + if (scenario === "instruction_error") throw initiatingError; + if (invalidated) { + const [current] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + originalBoot = current.controllerBootId; + replacementOwner = scenario === "superseded_owner" ? randomUUID() : current.leaseOwner; + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), leaseOwner: replacementOwner, + controllerGeneration: current.controllerGeneration + 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + } + return true; + }, + collectStopped: async () => {}, hasChanges: async () => false }, + }); + }; + let running: ReturnType | undefined; + try { + running = runCurrent(); + if (scenario.startsWith("detach_during_")) { + await Promise.race([closeEntered, running.then(() => { throw new Error("fixture checkpoint was not reached"); })]); + expect(renewalStopped).toBe(false); + const [beforeDetach] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const detach = await detachNativeSessionsForRestart([fixture.runId]); + expect(detach).toEqual({ detachedRunIds: [fixture.runId], inactiveRunIds: [], unsupportedRunIds: [], pendingRunIds: [] }); + expect(detached).toHaveBeenCalledOnce(); + const [held] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(held).toMatchObject({ phase: "workspace_finalizing", leaseOwner: beforeDetach.leaseOwner, resultId: beforeDetach.resultId }); + if (renewalLoss) { + originalBoot = held.controllerBootId; + replacementOwner = randomUUID(); + await fixture.db.update(nativeRunFinalizations).set({ leaseOwner: replacementOwner, + controllerGeneration: held.controllerGeneration + 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await expect.poll(() => renewal.mock.results[0].value.remainingMs(), { timeout: 25_000, interval: 250 }).toBe(0); + } + releaseClose(); + await expect(running).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError); + const failure = await running.catch(error => error); + expect(failure).toMatchObject({ diagnostic: { stage: "post_result_cleanup", + postExecutionCleanupFailed: instructionErrorWithDetach, renewalStopFailed: renewalLoss } }); + expect(failure).not.toHaveProperty("cause"); + expect(JSON.stringify(failure)).not.toContain("private-canary-N45"); + } else if (invalidated) await expect(running).rejects.toThrow("native_session_lease_lost"); + else if (scenario === "instruction_error") await expect(running).rejects.toBe(initiatingError); + else { + const result = await running; + expect(result.exitCode).toBe(1); + expect(result.errorCode).toBe("native_provider_overloaded"); + } + expect(state.checkpoint).toBeTypeOf("function"); + if (invalidated) expect(closeFailures).toEqual(["Error: native_harness_storage_owner_missing"]); + else expect(closeFailures).toEqual([]); + expect(ownersAtClose).toHaveLength(1); + expect(ownersAtClose[0].leaseOwner).not.toBeNull(); + expect(renewalStopped).toBe(true); + const [released] = await fixture.db.select({ leaseOwner: nativeRunFinalizations.leaseOwner }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(released.leaseOwner).toBe(invalidated || renewalLoss ? replacementOwner : scenario.startsWith("detach_during_") ? ownersAtClose[0].leaseOwner : null); + expect(state.execute).toHaveBeenCalledTimes(1); + if (scenario.startsWith("detach_during_")) { + const rows = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.nativeIssueId, fixture.issueId)); + expect(rows).toHaveLength(1); + expect(rows[0].resultJson).not.toHaveProperty("nativeCancellation"); + } + if (scenario === "fresh_successor") { + await fixture.db.update(heartbeatRuns).set({ responsibleUserId: "local-capacity-owner" }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(issues).set({ responsibleUserId: "local-capacity-owner" }).where(eq(issues.id, fixture.issueId)); + await finalizeNativeRun({ db: fixture.db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + const [retry] = await fixture.db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.retryOfRunId, fixture.runId))); + expect(retry).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "native_provider_overloaded", contextSnapshot: { forceFreshSession: true } }); + const unexpected = async () => { throw new Error("unexpected fixture escalation"); }; + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(fixture.db, { + resolveResponsibleUserId: unexpected, getRoutineEnv: unexpected, resolveSessionBeforeForWakeup: unexpected }), + recovery: { escalateStrandedAssignedIssue: unexpected, escalateStrandedRecoveryIssueInPlace: unexpected } }); + expect((await release({ companyId, runId: fixture.runId, now: new Date() })).outcome.kind).toBe("released"); + const [ready] = await fixture.db.update(heartbeatRuns).set({ status: "running", runtimeMode: "native", nativeIssueId: fixture.issueId }) + .where(eq(heartbeatRuns.id, retry.id)).returning(); + await fixture.db.update(issues).set({ executionRunId: retry.id }).where(eq(issues.id, fixture.issueId)); + const [task] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)); + const native = await prepareNativeHeartbeatRun({ db: fixture.db, run: ready, issue: task, environmentLeaseId: randomUUID() }); + await fixture.db.insert(nativeRunFinalizations).values({ runId: retry.id, companyId, issueId: fixture.issueId, phase: "observed" }); + const previousSession = execution.session.normalizedSessionId; + execution = { ...execution, binding: { ...execution.binding, runId: retry.id }, session: { ...execution.session, normalizedSessionId: native.normalizedSessionId } }; + expect(execution.session.normalizedSessionId).not.toBe(previousSession); + fixture = { ...fixture, runId: retry.id, native }; + acquisition = "resumed"; + renewalStopped = false; + const second = await runCurrent(); + expect(second.errorCode).toBe("native_provider_overloaded"); + expect(state.execute).toHaveBeenCalledTimes(2); + expect(ownersAtClose).toHaveLength(2); + expect(closeFailures).toEqual([]); + } + } finally { + releaseClose(); + await running?.catch(() => undefined); + if (invalidated || renewalLoss) await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: originalBoot, phase: "committed", + leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await closeWarmNativeSessionsForEnvironment({ environmentId, reason: "fixture cleanup" }); + renewal.mockRestore(); + } + }, 45_000); +});