diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index a2eebcada7c..96f0a88b8b4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2328,6 +2328,17 @@ predecessor/successor receipts for every earlier owner. Unrelated, live, or incompletely linked owners remain ambiguous. The run then fails closed with `runner_harness_state_mismatch: backup_without_reusable_lease`. +A Runner close can also fail to prove suspension after the control plane +accepted the run's result (`native.post_result_failure` with category +`runner_settlement_incomplete`). The provider work is complete, so this is not +a run failure. The executor first proves that it still holds the exact, +unexpired finalization lease and that the result is accepted. It then releases +that lease to finalization. A run on a local workspace has nothing to copy +back. Paperclip finalizes its accepted result and completes the run from that +result through the ordinary completion path. Load or a slow suspend command +cannot turn an accepted result into `adapter_failed`. A run with a synced or +remote workspace keeps its copy-back retry path. + A per-turn sandbox turn can finish through recovery without a Runner suspension receipt or failover backup on the controller. For example, its post-turn checkpoint copy can fail on a full sandbox disk. The controller diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index 7c32ea45ca0..37cb96940b8 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -2,7 +2,7 @@ import { NativeControllerRecoveryWaitingError } from "../native-runtime/native-c import { createNativeUsageDelivery } from "../native-runtime/native-usage-delivery.js"; import { flushBudgetEvaluationWork } from "../budgets.js"; -import type { PrpEvent } from "../../vendor/paperclip-runner/index.js"; +import { NativeSessionCloseUnrecoverableError, type PrpEvent } from "../../vendor/paperclip-runner/index.js"; import { randomUUID } from "node:crypto"; import { mkdtemp, rm, readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; @@ -288,6 +288,118 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = }, ); + it.each(["settled", "workspace-check-failed", "terminal-unreadable"] as const)("settles an accepted local result when runner close cannot prove suspension (%s)", async scenario => { + const { input } = await fixture(true); + const issue = input.task.issueRef!; + const [run] = await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, contextSnapshot: { issueId: issue.id }, costAccountingPending: true }).where(eq(heartbeatRuns.id, input.run.id)).returning(); + input.run = run; + const prepared = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issue.id)); + const completion = buildNativeCompletionContract(issue, { revision: contract.revision }); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: issue.id, phase: "observed" }); + input.runtime.nativeExecution = { + schema: "paperclip.native-execution-input.v1", interactionResponses: [], credentialBindings: [], binding: { companyId: run.companyId, agentId: run.agentId, issueId: issue.id, runId: run.id, executionWorkspaceId: run.id }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", contract: completion }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issue.id, title: issue.title, description: null, prompt: "Local runner settlement fault", workMode: "standard" }, + workspace: { cwd: home, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: prepared.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + } satisfies NativeExecutionInputV1; + input.runtime.nativeRunnerInstanceId = prepared.runnerInstanceId; + postResultProvider.execute.mockReset().mockImplementationOnce(async (options: { controlPlane: ControlPlanePort }) => { + await options.controlPlane.openRun({ identity: { companyId: run.companyId, issueId: issue.id, runId: run.id, agentId: run.agentId, sessionId: prepared.normalizedSessionId }, backendKind: "mock", sourceInstanceId: prepared.runnerInstanceId }); + await options.controlPlane.completeRun({ result: { ...CONTROL_PLANE_CONFORMANCE_RESULT, completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: completion.revision } }, + terminal: { schema: "paperclip.prp.terminal.v1", turnTerminalState: "completed", runTerminalState: "succeeded", reportedWorkDisposition: "done" }, turnId: prepared.turnId }); + if (scenario === "terminal-unreadable") { + await db.update(nativeRunResults).set({ resultJson: sql`${nativeRunResults.resultJson} - 'terminal'` }).where(eq(nativeRunResults.runId, run.id)); + } + // The runner retired after the accepted result, but its suspend command + // did not complete before the close deadline (for example, under load). + throw new NativeSessionCloseUnrecoverableError({ runnerSuspended: false, providerDrained: true }, true); + }); + const actual = await vi.importActual("../native-runtime/native-session-executor.js"); + vi.mocked(executePaperclipNativeSession).mockImplementationOnce(async options => actual.executePaperclipNativeSession({ ...options, useRunnerd: false, + backend: { descriptor: async () => { throw new Error("Unexpected provider call"); }, openSession: async () => { throw new Error("Unexpected provider call"); } } })); + + if (scenario === "terminal-unreadable") { + // Without a rebuildable accepted result, no succeeded finalize row is written. + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ message: "native_session_lease_lost" }); + const operations = await finalizations(); + expect(operations.map(operation => operation.status)).toEqual(["failed"]); + return; + } + if (scenario === "workspace-check-failed") { + // A failed workspace finalization check keeps the existing failure path. + const recorder = input.workspace.workspaceOperationRecorder; + const record = recorder.recordOperation.bind(recorder); + vi.spyOn(recorder, "recordOperation").mockImplementation(async operation => { + if (operation.phase === "workspace_finalize") throw new Error("workspace check failed"); + return record(operation); + }); + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ message: "native_session_lease_lost", diagnostic: { category: "runner_settlement_incomplete" } }); + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + expect(after).toMatchObject({ phase: "retryable_failure", leaseOwner: null, failureCode: "workspace_finalization_failed" }); + return; + } + // Runner bookkeeping after an accepted result is not an adapter failure: + // the result returns to ordinary completion like any successful turn. + const execution = await executeHeartbeatRuntime(db, input); + if (!execution.dispatched) throw new Error("expected a dispatched native execution"); + expect(execution.adapterResult).toMatchObject({ exitCode: 0, errorMessage: null, timedOut: false, + resultJson: { nativePostResultFailure: { stage: "post_result_execution", category: "runner_settlement_incomplete", resultAlreadyAccepted: true } }, + nativeFinalization: { runId: run.id, terminal: { runTerminalState: "succeeded" } } }); + + // The executor handed over its lease, so the accepted result was finalized + // now rather than after the lease TTL. + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + expect(after).toMatchObject({ phase: "committed", leaseOwner: null, leaseExpiresAt: null }); + const saved = await read(run.id); + expect(saved.status).toBe("running"); + expect(saved.errorCode).toBeNull(); + const [operation] = await finalizations(); + expect(operation.status).toBe("succeeded"); + expect(postResultProvider.execute).toHaveBeenCalledOnce(); + expect(input.services.enqueueWakeup).not.toHaveBeenCalled(); + }); + + it("keeps workspace failure handling for other accepted-result failures on a local workspace", async () => { + const { input } = await fixture(true); + const issue = input.task.issueRef!; + const [run] = await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, contextSnapshot: { issueId: issue.id }, costAccountingPending: true }).where(eq(heartbeatRuns.id, input.run.id)).returning(); + input.run = run; + const prepared = await prepareNativeHeartbeatRun({ db, run, issue, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issue.id)); + const completion = buildNativeCompletionContract(issue, { revision: contract.revision }); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: issue.id, phase: "observed" }); + input.runtime.nativeExecution = { + schema: "paperclip.native-execution-input.v1", interactionResponses: [], credentialBindings: [], binding: { companyId: run.companyId, agentId: run.agentId, issueId: issue.id, runId: run.id, executionWorkspaceId: run.id }, + completionContract: { id: contract.id, sha256: contract.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", contract: completion }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issue.id, title: issue.title, description: null, prompt: "Local post-result I/O fault", workMode: "standard" }, + workspace: { cwd: home, repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: prepared.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + } satisfies NativeExecutionInputV1; + input.runtime.nativeRunnerInstanceId = prepared.runnerInstanceId; + postResultProvider.execute.mockReset().mockImplementationOnce(async (options: { controlPlane: ControlPlanePort }) => { + await options.controlPlane.openRun({ identity: { companyId: run.companyId, issueId: issue.id, runId: run.id, agentId: run.agentId, sessionId: prepared.normalizedSessionId }, backendKind: "mock", sourceInstanceId: prepared.runnerInstanceId }); + await options.controlPlane.completeRun({ result: { ...CONTROL_PLANE_CONFORMANCE_RESULT, completionClaim: { ...CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim, contractRevision: completion.revision } }, + terminal: { schema: "paperclip.prp.terminal.v1", turnTerminalState: "completed", runTerminalState: "succeeded", reportedWorkDisposition: "done" }, turnId: prepared.turnId }); + await readFile(join(home, `missing-${run.id}`, "harmless-checkpoint")); + throw new Error("fixture must fail with ENOENT"); + }); + const actual = await vi.importActual("../native-runtime/native-session-executor.js"); + vi.mocked(executePaperclipNativeSession).mockImplementationOnce(async options => actual.executePaperclipNativeSession({ ...options, useRunnerd: false, + backend: { descriptor: async () => { throw new Error("Unexpected provider call"); }, openSession: async () => { throw new Error("Unexpected provider call"); } } })); + + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ message: "native_session_lease_lost", diagnostic: { category: "io_not_found" } }); + + // The executor hands its exact lease to finalization, which records the + // workspace failure for repair instead of waiting out the lease TTL. + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + expect(after).toMatchObject({ phase: "retryable_failure", leaseOwner: null, failureCode: "workspace_finalization_failed" }); + expect((await read(run.id)).status).toBe("running"); + }); + const read = (id: string) => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)).then(rows => rows[0]!); const finalizations = () => db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "workspace_finalize")); function expectCleanup(input: HeartbeatRuntimeExecutionInput) { diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 396814abde9..6ffd6205b6a 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -2,6 +2,7 @@ import { readClaimedNativeStopRecovery } from "../native-runtime/native-cancella import { readContextCompactionRequest } from "../context-compaction.js"; import { remoteWorkspaceFailureCause } from "../remote-workspace-transport.js"; import { readNativePostResultFailureDiagnostic } from "../native-runtime/native-post-result-failure.js"; +import { acceptedNativeAdapterResult } from "../native-runtime/native-accepted-result-settlement.js"; import { NativeStartupCleanupPendingError } from "../native-runtime/native-startup-cleanup.js"; import { NativeControllerRecoveryWaitingError } from "../native-runtime/native-controller-recovery.js"; import { describeFailedAttachment } from "../native-runtime/remote-warm-session-recovery.js"; @@ -1160,14 +1161,39 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe // The control code can be a fencing label; the closed diagnostic says // what actually failed after an accepted result (no provider text). const postResultFailure = readNativePostResultFailureDiagnostic(adapterErr); + // A runner that could not prove suspension after an accepted result left a + // local workspace untouched: there is nothing to copy back. Settle the + // accepted result instead of failing the run over runner bookkeeping. + // Rebuild the accepted result before writing any finalize status: only a + // proven, complete accepted result may record workspace finalization as + // succeeded. + let acceptedResult = nativeRuntimeResolution.kind === "native" + && postResultFailure?.category === "runner_settlement_incomplete" + && !nativeWorkspaceSync && !remoteWorkspaceAuthority + ? await (async () => { + const resultId = await db.select({ resultId: nativeRunFinalizations.resultId }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)).limit(1).then(rows => rows[0]?.resultId ?? null); + return resultId ? acceptedNativeAdapterResult(db, { + runId: run.id, companyId: run.companyId, resultId, + model: nativeExecution?.provider.model ?? null, driverKind: nativeExecution?.session.driverKind ?? "unknown", + postResultFailure: postResultFailure!, + }) : null; + })().catch((readErr: unknown) => { + logger.warn({ err: readErr, runId: run.id }, "accepted native result could not be read; keeping the failure path"); + return null; + }) + : null; + let settleAcceptedResult = acceptedResult !== null; try { - await recordWorkspaceFinalize("failed", { - errorMessage: - adapterErr instanceof Error - ? adapterErr.message - : String(adapterErr), - ...(postResultFailure ? { failureDiagnostic: postResultFailure } : {}), - }); + await recordWorkspaceFinalize(settleAcceptedResult ? "succeeded" : "failed", settleAcceptedResult + ? { runnerSettlementDiagnostic: postResultFailure } + : { + errorMessage: + adapterErr instanceof Error + ? adapterErr.message + : String(adapterErr), + ...(postResultFailure ? { failureDiagnostic: postResultFailure } : {}), + }); } catch (recordErr) { logger.warn( { @@ -1178,6 +1204,9 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe "failed to record workspace_finalize=failed operation; dependents may remain gated", ); } + // Settle only when the workspace finalization record itself succeeded. A + // failed branch or plugin workspace check keeps the failure path. + if (adapterFinalizeOutcome !== "succeeded") { settleAcceptedResult = false; acceptedResult = null; } if (nativeRuntimeResolution.kind === "native") { const proposedResult = await db .select({ resultId: nativeRunFinalizations.resultId }) @@ -1216,6 +1245,24 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe : "workspace_sync_out_failed", ); } + if (settleAcceptedResult && acceptedResult) { + // Finalize exactly as a returned result would, then let the ordinary + // completion path settle the run, session, accounting and agent. + await onLog("stderr", "[paperclip] runner settlement was incomplete after the result was accepted; finalizing the accepted result\n") + .catch(() => undefined); + try { + await measureRunPhase("execution.native_finalize", () => finalizeNativeRun({ + db, runId: run.id, workspaceFinalizeStatus: "succeeded", + })); + await dispatchPendingNativeStatusWakeups({ companyId: run.companyId }); + } catch (finalizeErr) { + logger.warn( + { err: finalizeErr, runId: run.id }, + "native result persisted but finalization did not apply; the reconciliation loop will retry", + ); + } + return { dispatched: true, adapterResult: acceptedResult }; + } try { await measureRunPhase("execution.native_finalize", () => finalizeNativeRun({ db, diff --git a/server/src/services/native-runtime/native-accepted-result-settlement.ts b/server/src/services/native-runtime/native-accepted-result-settlement.ts new file mode 100644 index 00000000000..423a63f6887 --- /dev/null +++ b/server/src/services/native-runtime/native-accepted-result-settlement.ts @@ -0,0 +1,88 @@ +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns, nativeRunResults, type Db } from "@paperclipai/db"; +import { NATIVE_FINALIZATION_SCHEMA, type NativeFinalizationResult } from "@paperclipai/shared"; +import type { AdapterExecutionResult } from "../../adapters/index.js"; +import { readAcceptedRunUsageCheckpoint } from "../usage-receipts.js"; +import type { NativePostResultFailureDiagnostic } from "./native-post-result-failure.js"; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +const RUN_TERMINAL_STATES = new Set(["succeeded", "failed", "cancelled"]); + +/** + * Rebuild the execution result of an accepted native run whose runner could not + * prove suspension when it closed. The provider turn, its result and terminal + * state are durable; only runner bookkeeping failed. Ordinary completion can + * then settle the run from the result instead of reporting an adapter failure. + * Returns null unless the exact accepted result and its terminal are present. + */ +export async function acceptedNativeAdapterResult(db: Db, input: { + runId: string; + companyId: string; + resultId: string; + model: string | null; + driverKind: string; + postResultFailure: NativePostResultFailureDiagnostic; +}): Promise { + const [row] = await db.select({ + issueId: nativeRunResults.issueId, + turnId: nativeRunResults.turnId, + resultJson: nativeRunResults.resultJson, + nativeSessionId: heartbeatRuns.nativeSessionId, + runnerInstanceId: heartbeatRuns.runnerInstanceId, + }).from(nativeRunResults) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunResults.runId), eq(heartbeatRuns.companyId, nativeRunResults.companyId))) + .where(and( + eq(nativeRunResults.id, input.resultId), + eq(nativeRunResults.runId, input.runId), + eq(nativeRunResults.companyId, input.companyId), + eq(nativeRunResults.schemaStatus, "accepted"), + )).limit(1); + if (!row) return null; + const result = record(row.resultJson.result); + const terminal = record(row.resultJson.terminal); + if (terminal.schema !== "paperclip.prp.terminal.v1" || !RUN_TERMINAL_STATES.has(String(terminal.runTerminalState))) return null; + const nativeTerminal = terminal as unknown as NativeFinalizationResult["terminal"]; + const succeeded = nativeTerminal.runTerminalState === "succeeded"; + const usage = await readAcceptedRunUsageCheckpoint(db, { companyId: input.companyId, runId: input.runId }); + const normalizedSessionId = row.nativeSessionId ?? null; + return { + exitCode: succeeded ? 0 : 1, + signal: null, + timedOut: false, + errorMessage: succeeded ? null : `Native session ${nativeTerminal.runTerminalState}`, + resultJson: { + nativeResult: result, + nativeTerminal: terminal, + nativePostResultFailure: { ...input.postResultFailure }, + }, + summary: typeof result.summary === "string" ? result.summary : null, + ...(normalizedSessionId ? { sessionId: normalizedSessionId, sessionDisplayId: normalizedSessionId } : {}), + model: input.model ?? undefined, + ...(usage ?? {}), + usageComplete: usage?.complete ?? false, + ...(!usage ? { costStatus: "unpriced" as const } : {}), + ...(normalizedSessionId && row.runnerInstanceId ? { + nativeFinalization: { + schema: NATIVE_FINALIZATION_SCHEMA, + runtimeMode: "native", + runId: input.runId, + issueId: row.issueId, + companyId: input.companyId, + result, + terminal: nativeTerminal, + turnId: row.turnId, + sourceInstanceId: row.runnerInstanceId, + normalizedSessionId, + providerSessionId: null, + driverKind: input.driverKind, + driverVersion: "unknown", + nativeEventCount: 0, + highestContiguousSourceSeq: 0, + workspaceFinalizeStatus: "succeeded", + }, + } : {}), + }; +} diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 9a9382c8363..6e1833712bb 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -11249,6 +11249,31 @@ async function executePaperclipNativeSessionWithinScope( // Emit after rollback so the diagnostic sink cannot deadlock on the // coordinator transaction. Reporting failure cannot change fencing. if (failure instanceof NativePostResultFailureError) { + // The rejected write proved this exact, unexpired owner holds an + // accepted result. Provider work is over and renewal is stopping, so + // hand the lease to finalization now, as the success path does. A + // retained lease would make the finalizer wait out its whole TTL. + await stoppedLeaseRenewal; + await boundedExecutionCleanup(async () => { + await input.db.transaction(async tx => { + await notifyNativeFinalizationWork(tx, input.execution.binding); + await tx.update(nativeRunFinalizations) + .set({ leaseOwner: null, leaseExpiresAt: null, recoveryState: null, recoveryRequestId: null, updatedAt: new Date() }) + .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.controllerGeneration, controllerGeneration), + eq(nativeRunFinalizations.controllerBootId, controller.bootId), + eq(nativeRunFinalizations.controllerPid, controller.pid), + eq(nativeRunFinalizations.controllerProcessStartedAt, controller.processStartedAt), + gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`), + isNotNull(nativeRunFinalizations.resultId), + )); + }); + }); await boundedExecutionCleanup(async () => { await input.onEvent?.({ eventType: "native.post_result_failure", level: "warn", message: "Native execution failed after its result was accepted.",