diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 3784b3bee2f..39bbe8ecdad 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -752,6 +752,17 @@ and current-user redaction. The task page uses one history read on mount/reconne and the existing live transcript stream, then makes a bounded plugin readiness check. No panel discovery timer scans runs or provider state while connected. +## Post-result failure diagnostics + +`native.post_result_failure` records `stage: post_result_execution`, a closed +`category` (`io_not_found`, `io_permission_denied`, `io_failure`, or +`execution_failure`), and `resultAlreadyAccepted: true`. It requires an accepted +selected result and exact current, unexpired controller authority. A busy diagnostic +row lock is skipped. Reporting is best effort under the existing 60-second control +bound; the sink can finish later and emit a late event. The `native_session_lease_lost` +control classification stays unchanged. The labels grant no recovery, result, +Stop, provider-work or payment authority and do not establish a hosted failure cause. + ## Payload storage Structured events in this guide remain in `heartbeat_run_events`. When the diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index f8e1e833b3e..7c32ea45ca0 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -4,12 +4,12 @@ import { createNativeUsageDelivery } from "../native-runtime/native-usage-delive import { flushBudgetEvaluationWork } from "../budgets.js"; import type { PrpEvent } from "../../vendor/paperclip-runner/index.js"; import { randomUUID } from "node:crypto"; -import { mkdtemp, rm } from "node:fs/promises"; +import { mkdtemp, rm, readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { agents, budgetPolicies, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, executionWorkspaces, nativeRunFinalizations, runUsageReceipts, workspaceOperations, type Db } from "@paperclipai/db"; +import { agents, budgetPolicies, companies, completionContracts, nativeRunResults, costEvents, agentWakeupRequests, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, executionWorkspaces, nativeRunFinalizations, runUsageReceipts, workspaceOperations, type Db } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; -import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; import { trace } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { createRunPreparationTiming, measureRunPhase, measureWorkspaceSyncPhase, withRunExecutionTiming } from "../execution-preparation-timing.js"; @@ -21,7 +21,7 @@ import { createAdapterExecutionControl } from "../adapter-execution-control.js"; import { executionWorkspaceService } from "../execution-workspaces.js"; import { workspaceOperationService } from "../workspace-operations.js"; import { instanceSettingsService } from "../instance-settings.js"; -import { executePaperclipNativeSession, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError } from "../native-runtime/native-session-executor.js"; +import { executePaperclipNativeSession, NativeCancellationPendingRecoveryError, NativeControllerDetachedForRestartError, nativeSessionFailureSourceCode, nativeSessionFailureDisposition } from "../native-runtime/native-session-executor.js"; import { NativeRunnerOwnershipUnverifiedError } from "../native-runtime/native-runner-ownership.js"; import { NativeWorkspaceFinalizationBusyError, NativeWorkspaceFinalizationOwnershipLostError } from "../native-runtime/native-workspace-finalization-ownership.js"; import { resolveHeartbeatNativeRuntimeMode } from "../native-runtime/runtime-mode.js"; @@ -32,6 +32,16 @@ import { buildEffectiveRunSessionConfigMetadata } from "./workspaces.js"; import { revokeHeartbeatRunGatewayTokens } from "./run-preparation.js"; import { executeHeartbeatRuntime, type HeartbeatRuntimeExecutionInput } from "./runtime-execution.js"; +import { prepareNativeHeartbeatRun } from "../native-runtime/prepare-native-run.js"; +import { buildNativeCompletionContract } from "../native-runtime/completion-contracts.js"; +import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; +import { CONTROL_PLANE_CONFORMANCE_RESULT } from "../../vendor/paperclip-runner/testing.js"; +import type { ControlPlanePort, NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; +const postResultProvider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async original => ({ + ...await original(), executeNativeSession: postResultProvider.execute, +})); + // Keep real accounting, workspace barriers, and task-session persistence. Providers // and gateway delivery are external boundaries; browser acceptance covers their wiring. vi.mock("../native-runtime/native-session-executor.js", async original => ({ @@ -115,6 +125,169 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = }; return { input, adapter, run }; } + it.each(["current", "expired", "replaced-owner", "generation", "attempt", "boot", "pid", "process-start", "stopped", "unknown-code", "permission", "io", "diagnostic-sink-failed", "diagnostic-lock-busy"] as const)( + "preserves post-result diagnostics without changing recovery authority (%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" }); + const execution: NativeExecutionInputV1 = { + 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 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 } }, + }; + input.runtime.nativeExecution = execution; + input.runtime.nativeRunnerInstanceId = prepared.runnerInstanceId; + input.workspace.nativeWorkspaceSync = { mode: "host_current", reference: { + schema: "paperclip.native-workspace-sync/v3", state: "prepared", descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null, + workspaceId: run.id, leaseId: randomUUID(), providerLeaseId: "controlled-local", remoteCwd: home, resourceDisposition: null }, + restoreWorkspace: vi.fn(async () => {}), cleanup: vi.fn(async () => {}) }; + const diagnosticExpected = ["current", "stopped", "unknown-code", "permission", "io", "diagnostic-sink-failed"].includes(scenario); + let faultInjected = false, lockHeld = false, settledWhileLocked = false; + let releaseLock: () => void = () => {}; + let lockTask: Promise | undefined; + let diagnosticLockError: { code?: string; message?: string } | undefined; + if (scenario === "diagnostic-lock-busy") { + const transaction = db.transaction.bind(db); + let armed = false; + const transactionSpy = vi.spyOn(db, "transaction").mockImplementation(async (body, config) => { + const result = await transaction(async tx => { + // Observe real query errors; the executor still executes its actual + // diagnostic SELECT against PostgreSQL and the competing row lock. + if (faultInjected) { + const select = tx.select.bind(tx); + vi.spyOn(tx, "select").mockImplementation(fields => { + const selection = fields === undefined ? select() : select(fields); + const from = selection.from.bind(selection); + selection.from = ((table: Parameters[0]) => { + const query = from(table); + if (table === nativeRunFinalizations && fields?.runId === nativeRunFinalizations.runId) { + const lock = query.for.bind(query); + query.for = ((...args: Parameters) => lock(...args).catch(error => { + diagnosticLockError = error.cause ?? error; + throw error; + })) as typeof query.for; + } + return query; + }) as typeof selection.from; + return selection as ReturnType; + }); + } + return body(tx); + }, config); + // The first transaction after the injected I/O failure publishes the + // control deadline. Hold the row only after that transaction commits. + if (faultInjected && !armed) { + armed = true; + let ready!: () => void; + const acquired = new Promise(resolve => { ready = resolve; }); + const release = new Promise(resolve => { releaseLock = resolve; }); + lockTask = transaction(async locker => { + await locker.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, run.id)).for("update"); + lockHeld = true; + ready(); + await release; + lockHeld = false; + }); + await acquired; + } + return result; + }); + onTestFinished(async () => { releaseLock(); await lockTask; transactionSpy.mockRestore(); }); + } + 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 }); + const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + const controller = await currentNativeControllerIdentity(); + expect(owner).toMatchObject({ phase: "workspace_finalizing", attempt: 1, controllerGeneration: 1, controllerBootId: controller.bootId, controllerPid: controller.pid, controllerProcessStartedAt: controller.processStartedAt }); + expect(owner.leaseOwner).toBeTruthy(); + expect(owner.leaseExpiresAt!.getTime()).toBeGreaterThan(Date.now()); + const changed = scenario === "expired" ? { leaseExpiresAt: new Date(0) } + : scenario === "replaced-owner" ? { leaseOwner: "replacement-owner" } + : scenario === "generation" ? { controllerGeneration: owner.controllerGeneration + 1 } + : scenario === "attempt" ? { attempt: owner.attempt + 1 } + : scenario === "boot" ? { controllerBootId: randomUUID() } + : scenario === "pid" ? { controllerPid: controller.pid + 1 } + : scenario === "process-start" ? { controllerProcessStartedAt: new Date(controller.processStartedAt.getTime() - 1000) } : null; + if (changed) await db.update(nativeRunFinalizations).set(changed).where(eq(nativeRunFinalizations.runId, run.id)); + if (scenario === "stopped") await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, run.id)); + try { await readFile(join(home, `missing-${run.id}`, "harmless-checkpoint")); } + catch (error) { + faultInjected = true; + if (scenario === "unknown-code") (error as NodeJS.ErrnoException).code = "private-unrecognized-code"; + if (scenario === "permission") (error as NodeJS.ErrnoException).code = "EACCES"; + if (scenario === "io") (error as NodeJS.ErrnoException).code = "EIO"; + throw error; + } + throw new Error("fixture must fail with ENOENT"); + }); + const events: unknown[] = []; + let controlError: unknown; + const actual = await vi.importActual("../native-runtime/native-session-executor.js"); + vi.mocked(executePaperclipNativeSession).mockImplementationOnce(async options => { + try { return await actual.executePaperclipNativeSession({ ...options, useRunnerd: false, + backend: { descriptor: async () => { throw new Error("Unexpected provider call"); }, openSession: async () => { throw new Error("Unexpected provider call"); } }, + onEvent: async event => { + events.push(event); + if (event.eventType === "native.post_result_failure" && scenario === "diagnostic-sink-failed") throw new Error("private sink failure"); + await options.onEvent?.(event); + } }); } + catch (error) { + controlError = error; + settledWhileLocked = lockHeld; + releaseLock(); + await lockTask; + throw error; + } + }); + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ name: "NativeWorkspaceFinalizeScheduledError", reasonCode: "workspace_sync_out_failed", terminalFailure: false }); + expect((controlError as Error).message).toBe("native_session_lease_lost"); + if (scenario === "diagnostic-lock-busy") { + expect(settledWhileLocked).toBe(true); + expect(diagnosticLockError?.code).toBe("55P03"); + expect(diagnosticLockError?.message).toContain("could not obtain lock"); + } + const source = nativeSessionFailureSourceCode(controlError); + expect(source).toBe(nativeSessionFailureSourceCode(new Error("native_session_lease_lost"))); + expect(nativeSessionFailureDisposition(1, new Date(0), source)).toEqual(nativeSessionFailureDisposition(1, new Date(0), nativeSessionFailureSourceCode(new Error("native_session_lease_lost")))); + expect((controlError as { diagnostic?: unknown }).diagnostic).toEqual(diagnosticExpected + ? { stage: "post_result_execution", category: scenario === "unknown-code" ? "execution_failure" : scenario === "permission" ? "io_permission_denied" : scenario === "io" ? "io_failure" : "io_not_found", resultAlreadyAccepted: true } + : undefined); + const diagnostics = events.filter(event => (event as { eventType: string }).eventType === "native.post_result_failure") as Array<{ payload: unknown }>; + expect(diagnostics).toHaveLength(diagnosticExpected ? 1 : 0); + if (diagnosticExpected) expect(diagnostics[0].payload).toEqual({ stage: "post_result_execution", category: scenario === "unknown-code" ? "execution_failure" : scenario === "permission" ? "io_permission_denied" : scenario === "io" ? "io_failure" : "io_not_found", resultAlreadyAccepted: true }); + expect(JSON.stringify(diagnostics)).not.toContain(home); + expect(JSON.stringify(diagnostics)).not.toContain("private-"); + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + const results = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, run.id)); + expect(results).toHaveLength(1); + expect(after.resultId).toBe(results[0].id); + expect(after.phase).toBe("retryable_failure"); + expect(after.failureCode).toBe("native_workspace_sync_out_failed"); + expect(after.nextAttemptAt).toBeInstanceOf(Date); + const saved = await read(run.id); + expect(saved.costAccountingPending).toBe(true); + expect(saved.costAccountedAt).toBeNull(); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, run.id))).toHaveLength(0); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, run.companyId))).toHaveLength(0); + if (scenario === "stopped") expect(saved.status).toBe("cancelled"); + expect(postResultProvider.execute).toHaveBeenCalledOnce(); + expect(input.effects.onNativeWorkspaceFinalizeScheduled).toHaveBeenCalledOnce(); + expect(input.services.enqueueWakeup).not.toHaveBeenCalled(); + expect(input.workspace.nativeWorkspaceSync.restoreWorkspace).not.toHaveBeenCalled(); + }, + ); + 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/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 275f7a96b16..d36f7395977 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -190,6 +190,7 @@ import { gt, inArray, isNull, + isNotNull, like, notInArray, or, @@ -307,6 +308,27 @@ export class NativeCancellationPendingRecoveryError extends Error { } } +type NativePostResultFailureCategory = "io_not_found" | "io_permission_denied" | "io_failure" | "execution_failure"; + +class NativePostResultFailureError extends Error { + readonly diagnostic: Readonly<{ + stage: "post_result_execution"; + category: NativePostResultFailureCategory; + resultAlreadyAccepted: true; + }>; + + constructor(error: unknown) { + // Keep the existing control classification. These labels explain why its + // failure write was rejected; they grant no finalization or retry authority. + super("native_session_lease_lost"); + const code = record(error).code; + const category: NativePostResultFailureCategory = code === "ENOENT" ? "io_not_found" + : code === "EACCES" || code === "EPERM" ? "io_permission_denied" + : code === "EIO" ? "io_failure" : "execution_failure"; + this.diagnostic = Object.freeze({ stage: "post_result_execution", category, resultAlreadyAccepted: true }); + } +} + export class NativeControllerDetachedForRestartError extends Error { constructor() { super("native_controller_detached_for_restart"); @@ -10738,7 +10760,39 @@ async function executePaperclipNativeSessionWithinScope( ) .returning({ runId: nativeRunFinalizations.runId, resultId: nativeRunFinalizations.resultId }) .then((rows) => rows[0] ?? null); - if (!updated) throw new Error("native_session_lease_lost"); + if (!updated) { + // Result acceptance intentionally prevents the failure transition. + // Diagnose that case only under the same still-current authority; + // stale or expired owners must retain the ordinary fencing error. + const [acceptedUnderCurrentAuthority] = await tx + .select({ runId: nativeRunFinalizations.runId }) + .from(nativeRunFinalizations) + .innerJoin(nativeRunResults, and( + eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, input.execution.binding.runId), + eq(nativeRunResults.companyId, input.execution.binding.companyId), + eq(nativeRunResults.issueId, input.execution.binding.issueId), + eq(nativeRunResults.schemaStatus, "accepted"), + )) + .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), + )) + .for("update", { of: nativeRunFinalizations, noWait: true }) + // A failed diagnostic read must retain the existing control error. + .catch(() => []); + if (acceptedUnderCurrentAuthority) throw new NativePostResultFailureError(error); + throw new Error("native_session_lease_lost"); + } const [runBeforeWrite] = await tx .select({ status: heartbeatRuns.status, resultJson: heartbeatRuns.resultJson, runnerInstanceId: heartbeatRuns.runnerInstanceId, nativeSessionId: heartbeatRuns.nativeSessionId, @@ -10899,6 +10953,17 @@ async function executePaperclipNativeSessionWithinScope( supersedeOnIdentityChange: recoveryProjection.supersedeOnIdentityChange, }); + }).catch(async (failure: unknown) => { + // Emit after rollback so the diagnostic sink cannot deadlock on the + // coordinator transaction. Reporting failure cannot change fencing. + if (failure instanceof NativePostResultFailureError) { + await boundedExecutionCleanup(async () => { + await input.onEvent?.({ eventType: "native.post_result_failure", level: "warn", + message: "Native execution failed after its result was accepted.", + payload: { ...failure.diagnostic } }); + }); + } + throw failure; }); // Pass the raw error only for closed settlement labels: its text bypasses // the identity redactor applied to the persisted message above.