diff --git a/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts b/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts new file mode 100644 index 00000000000..275a67ff316 --- /dev/null +++ b/server/src/__tests__/heartbeat-native-predispatch-retry.test.ts @@ -0,0 +1,167 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { agents, authUsers, companyMemberships, createDb, heartbeatRuns, issueComments, issueRecoveryActions, issues } from "@paperclipai/db"; +import type { NativeRunIdentity, NativeSession, NativeSessionBackend, PrpEvent } from "@paperclipai/paperclip-runner"; +import { describe, expect, it, vi } from "vitest"; +import { heartbeatService, type HeartbeatServiceOptions } from "../services/heartbeat.js"; +import { instanceSettingsService } from "../services/instance-settings.js"; +import { closeIdleWarmNativeSessionsForRestart } from "../services/native-runtime/native-session-executor.js"; +import { nativeRestartFixture, waitForCondition } from "./helpers/native-restart-fixture.js"; +import { getRecoveryPostgresSupport } from "./helpers/recovery-postgres-fixture.js"; + +// Budget admission is the last database write before the dispatch gate enters +// the native executor. Exhausted serialization retries there fail the run with +// no provider work started. +const admissionFaults = vi.hoisted(() => ({ remaining: 0 })); +vi.mock("../services/budget-reservations.js", async importOriginal => { + const actual = await importOriginal(); + const { AccountingContentionError } = await import("../services/accounting-transaction.js"); + return { + ...actual, + reserveRunBudget: async (...args: Parameters) => { + if (admissionFaults.remaining > 0) { + admissionFaults.remaining -= 1; + throw new AccountingContentionError(Object.assign(new Error("could not serialize access due to concurrent update"), + { name: "PostgresError", code: "40001" }), 4); + } + return actual.reserveRunBudget(...args); + }, + }; +}); + +const support = await getRecoveryPostgresSupport({}); + +function nativeBackend(db: ReturnType, options: { failDescriptorOnce?: { failed: boolean } } = {}): + NonNullable { + return execution => { + let identity: NativeRunIdentity; + let sourceInstanceId = ""; + let turn = Promise.withResolvers(); + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; + const session: NativeSession = { + identity: () => identity, + capabilities: async () => capabilities, + attachRun: async input => { identity = input.identity; }, + startTurn: async () => { + const turnId = randomUUID(); + turn.resolve(turnId); + return { turnId }; + }, + async *events() { + const turnId = await turn.promise; + const event = (sourceSeq: number, eventType: PrpEvent["eventType"], payload: Record): PrpEvent => ({ + schema: "paperclip.prp.event.v1", sourceKind: "runner", sourceInstanceId, + sourceEventId: `${sourceInstanceId}:${identity.runId}:${sourceSeq}`, sourceSeq, + runId: identity.runId, normalizedSessionId: identity.sessionId, turnId, + schemaVersion: 1, priority: 0, emittedAt: new Date().toISOString(), eventType, payload, + }); + yield event(1, "turn.started", {}); + yield event(2, "item.completed", { kind: "agentMessage", channel: "final", text: "Done." }); + turn = Promise.withResolvers(); + yield event(3, "turn.completed", { status: "completed" }); + }, + result: async () => null, + snapshot: async () => ({ backendKind: "mock", sessionId: "predispatch-session", identity, + providerSessionId: "predispatch-provider-session", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }), + close: async () => {}, + }; + const backend: NativeSessionBackend = { + descriptor: async () => { + if (options.failDescriptorOnce && !options.failDescriptorOnce.failed) { + options.failDescriptorOnce.failed = true; + throw Object.assign(new Error("could not serialize access due to concurrent update"), { name: "PostgresError", code: "40001" }); + } + const [run] = await db.select({ runnerInstanceId: heartbeatRuns.runnerInstanceId }).from(heartbeatRuns) + .where(eq(heartbeatRuns.id, execution.binding.runId)); + if (!run?.runnerInstanceId) throw new Error("Normal native admission did not persist a runner identity"); + sourceInstanceId = run.runnerInstanceId; + return { kind: "mock", name: "native-predispatch-retry", version: "1", capabilities, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; + }, + openSession: async input => { identity = input.identity; return session; }, + recoverSession: async snapshot => { identity = snapshot.identity; return { recovered: true, session }; }, + }; + return backend; + }; +} + +describe.skipIf(!support.supported)("heartbeat native pre-dispatch failure retry", () => { + const fixture = nativeRestartFixture({}); + + async function startTask(db: ReturnType, heartbeat: ReturnType) { + const { companyId, agentId } = fixture.current; + const userId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Task owner", email: `${userId}@example.invalid`, + createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner" }); + await db.update(agents).set({ status: "idle", adapterConfig: { provider: "codex" } }).where(eq(agents.id, agentId)); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true, enableNativeRunner: true }); + const [issue] = await db.insert(issues).values({ companyId, title: "Survive a database fault", status: "in_progress", + assigneeAgentId: agentId, responsibleUserId: userId, conversationAgentId: agentId, conversationUserId: userId, + conversationState: "active" }).returning(); + const [message] = await db.insert(issueComments).values({ companyId, issueId: issue!.id, + authorType: "user", authorUserId: userId, body: "Start." }).returning(); + const run = await heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + requestedByActorType: "user", requestedByActorId: userId, manualUserWake: true, + payload: { issueId: issue!.id, commentId: message!.id }, + contextSnapshot: { issueId: issue!.id, taskKey: issue!.id, wakeCommentId: message!.id } }); + expect(run).not.toBeNull(); + await waitForCondition("the first run to settle", () => heartbeat.getTaskDrainStatus().quiescent); + return { issueId: issue!.id, runId: run!.id }; + } + + it("re-queues a native run that failed at budget admission and completes it without a Retry", async () => { + const { companyId } = fixture.current; + const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 }); + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: nativeBackend(db) }); + try { + admissionFaults.remaining = 1; + const { issueId, runId } = await startTask(db, heartbeat); + const [failed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(failed).toMatchObject({ status: "failed", runtimeMode: "native", errorCode: "adapter_failed", + resultJson: expect.objectContaining({ failureCause: "40001", + nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }) }); + const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.retryOfRunId, runId))); + expect(successors).toEqual([expect.objectContaining({ scheduledRetryReason: "interrupted_preparation", + scheduledRetryAttempt: 1 })]); + expect(successors[0]!.contextSnapshot?.executionRetryAccounting).toMatchObject({ failureRetries: 0 }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]?.status).not.toBe("blocked"); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + + // The first attempt is due at once; the scheduler promotes it and the + // queue starts it, exactly once. + expect(await heartbeat.promoteDueScheduledRetries()).toMatchObject({ promoted: 1 }); + await heartbeat.resumeQueuedRuns(); + await waitForCondition("the retry to succeed", async () => + (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successors[0]!.id)))[0]?.status === "succeeded"); + await waitForCondition("the retry to settle", () => heartbeat.getTaskDrainStatus().quiescent); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.retryOfRunId, runId)))).toHaveLength(1); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + } finally { + admissionFaults.remaining = 0; + await heartbeat.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await db.$client.end(); + } + }, 90_000); + + it("keeps the hold when the failure follows entry into the native executor", async () => { + const { companyId } = fixture.current; + const db = createDb(fixture.current.temporary.connectionString, { maxConnections: 4 }); + const heartbeat = heartbeatService(db, { nativeSessionBackendFactory: nativeBackend(db, { failDescriptorOnce: { failed: false } }) }); + try { + const { runId } = await startTask(db, heartbeat); + const [failed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + expect(failed?.status).toBe("failed"); + expect(failed?.resultJson ?? {}).not.toHaveProperty("nativePreDispatchFailure"); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), + eq(heartbeatRuns.retryOfRunId, runId), eq(heartbeatRuns.scheduledRetryReason, "interrupted_preparation")))).toHaveLength(0); + } finally { + await heartbeat.drainActiveRunExecutions(); + await closeIdleWarmNativeSessionsForRestart(); + await db.$client.end(); + } + }, 90_000); +}); diff --git a/server/src/__tests__/heartbeat-retry-scheduling.test.ts b/server/src/__tests__/heartbeat-retry-scheduling.test.ts index 77a397ac7db..1a42e69572c 100644 --- a/server/src/__tests__/heartbeat-retry-scheduling.test.ts +++ b/server/src/__tests__/heartbeat-retry-scheduling.test.ts @@ -30,6 +30,7 @@ import { import { drainHeartbeatRunsToQuiescence } from "./helpers/drain-heartbeat-runs.js"; import { registerServerAdapter, unregisterServerAdapter } from "../adapters/index.ts"; import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js"; +import { INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS } from "../services/execution-recovery-attempt.js"; const mockTelemetryClient = vi.hoisted(() => ({ track: vi.fn() })); const mockTrackAgentTaskRun = vi.hoisted(() => vi.fn()); @@ -536,6 +537,40 @@ describeEmbeddedPostgres("heartbeat bounded retry scheduling", () => { expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(3); }); + it("re-queues a native pre-dispatch database failure without charging an attempt, until its lane is spent", async () => { + const { agentId, issueId, runId, now } = await seedMaxTurnFixture(); + await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "gpt-6.1-sol" } }) + .where(eq(agents.id, agentId)); + const resultJson = { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }; + await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "adapter_failed", resultJson }) + .where(eq(heartbeatRuns.id, runId)); + let predecessor = runId; + for (const attempt of [1, 2]) { + const outcome = await heartbeatService(db).scheduleBoundedRetry(predecessor, + { now, random: () => 0, retryReason: "interrupted_preparation" }); + expect(outcome.outcome).toBe("scheduled"); + const [child] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, predecessor)); + expect(child).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "interrupted_preparation", + scheduledRetryAttempt: attempt }); + // Uncharged: the failure budget stays untouched; the first attempt is due at once. + expect(child!.contextSnapshot?.executionRetryAccounting).toMatchObject({ failureRetries: 0, interruptedPreparations: attempt }); + if (attempt === 1) expect(child!.scheduledRetryAt?.getTime()).toBe(now.getTime()); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]?.executionRunId).toBe(child!.id); + predecessor = child!.id; + await db.update(heartbeatRuns).set({ status: "failed", runtimeMode: "native", errorCode: "adapter_failed", resultJson, finishedAt: now }) + .where(eq(heartbeatRuns.id, predecessor)); + } + // A failure that keeps recurring ends visibly failed once the lane is spent. + const [last] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, predecessor)); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...last!.contextSnapshot, executionRetryAccounting: { + version: 1, failureRetries: 0, maxTurnContinuations: 0, interruptedPreparations: INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS } } }) + .where(eq(heartbeatRuns.id, predecessor)); + expect(await heartbeatService(db).scheduleBoundedRetry(predecessor, { now, retryReason: "interrupted_preparation" })) + .toMatchObject({ outcome: "retry_exhausted", maxAttempts: INTERRUPTED_PREPARATION_RETRY_MAX_ATTEMPTS }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, predecessor))).toHaveLength(0); + expect((await heartbeat.getRun(predecessor))?.status).toBe("failed"); + }); + it("retains the reviewed chat admission on a durable retry without approving the task", async () => { const { companyId, agentId, issueId, runId, now } = await seedMaxTurnFixture({ issueStatus: "in_review" }); const wakeCommentIds = [randomUUID(), randomUUID()]; diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 761624744fa..01a6c659415 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -868,6 +868,35 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].statusVersion).toBe(blockedIssue.statusVersion); }); + it.each([true, false])("holds a native pre-dispatch failure only when no automatic retry owns it (retry scheduled: %s)", async (retryScheduled) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "failed", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "adapter_failed", + resultJson: { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + if (retryScheduled) { + await db.insert(heartbeatRuns).values({ companyId, agentId, invocationSource: "automation", status: "scheduled_retry", + retryOfRunId: runId, scheduledRetryReason: "interrupted_preparation", scheduledRetryAttempt: 1, + contextSnapshot: { issueId, retryReason: "interrupted_preparation" } }); + } + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + const issue = (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; + if (retryScheduled) { + // The uncharged retry owns the next attempt: no incident, no blocked task, no Retry click. + expect(issue.status).toBe("in_progress"); + expect(actions).toHaveLength(0); + } else { + // An exhausted lane keeps today's hold, whose task offers a manual Retry. + expect(issue.status).toBe("blocked"); + expect(actions[0]).toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + } + }); + it("does not open a reconciliation incident for a native run whose model the provider rejected", async () => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 16b9897657d..ce6d1e564de 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -2044,6 +2044,11 @@ export function heartbeatService( }; let instructionWorkingCopy: ReturnType | null = null; let nativeDispatchStarted = false; + // Set once the dispatch gate enters the adapter or native executor. Unlike + // nativeDispatchStarted it also covers a gate that fails after entering. + let dispatchGateEntered = false; + const nativeDispatchMayHaveStarted = () => nativeDispatchStarted || dispatchGateEntered || nativeOwnershipHeld + || Boolean(runOptions.nativeRestartRecovery || runOptions.nativeLeaseOwner); const leaveToNextController = async (error: unknown) => { if (!shutdownHandedOff() || legacyAdapterEntered || nativeDispatchStarted) return false; nativeSessionResumeScheduled = true; @@ -3004,6 +3009,7 @@ export function heartbeatService( try { return await dispatchResolvedInteractionContinuationAfterAdmission((markDispatchStarted) => { entered = true; + dispatchGateEntered = true; return dispatch(markDispatchStarted); }, reservation); } catch (error) { @@ -3837,6 +3843,7 @@ export function heartbeatService( session: { runtimeForAdapter, previousSessionParams, taskKey, configuredModel, sessionConfigMetadata }, readFailureReportSecrets, runId, err, identityRedactor, requiredWorkspaceRestoreEvidence, legacyAdapterEntered, + nativeDispatchMayHaveStarted: nativeDispatchMayHaveStarted(), goalCheckpointSession, previousSessionDisplayId, taskSession, }); } @@ -3893,6 +3900,7 @@ export function heartbeatService( await runCompletion.failRunSetup({ run, runId, outerErr, signal: executionControl.controller.signal, providerDispatchStarted: legacyAdapterEntered || nativeDispatchStarted, + nativeDispatchMayHaveStarted: nativeDispatchMayHaveStarted(), identityRedactor, readFailureReportSecrets, }); legacyPreparationFailedBeforeDispatch = !legacyAdapterEntered && !nativeDispatchStarted; diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 0242cafc0a8..77458f2a094 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -1466,6 +1466,110 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( expect(failureLog?.[0]).toMatchObject({ runId: f.run.id, errorCode: "adapter_failed", failureCause: "40001" }); }); + describe("a native run that fails before provider work", () => { + const contention = () => new AccountingContentionError(Object.assign(new Error("deadlock detected"), + { name: "PostgresError", code: "40P01" }), 4); + const nativeFixture = async () => { + const f = await fixture({ runtimeMode: "native" }, `NPD${randomUUID().slice(0, 6)}`); + f.deps.scheduleBoundedRetryForRun.mockResolvedValue({ outcome: "scheduled", run: f.run } as never); + return { ...f, preDispatch: (err: unknown, nativeDispatchMayHaveStarted?: boolean) => + ({ ...f.failure(err), legacyAdapterEntered: false, nativeDispatchMayHaveStarted }) }; + }; + + it("records that no provider work started and re-queues itself in the uncharged preparation lane", async () => { + const f = await nativeFixture(); + await f.completion.failRun(f.preDispatch(contention(), false)); + const failed = await read(f.run.id); + expect(failed).toMatchObject({ status: "failed", errorCode: "adapter_failed", + resultJson: { failureCause: "40P01", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent, + { retryReason: "interrupted_preparation" }); + expect(f.deps.releaseIssueExecutionAndPromote).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), + expect.objectContaining({ suppressImmediateRecovery: true })); + }); + + it("re-queues a setup failure on a transient database fault", async () => { + const f = await nativeFixture(); + await f.completion.failRunSetup({ ...f.setup(Object.assign(new Error("could not serialize access"), + { name: "PostgresError", code: "40001" })), nativeDispatchMayHaveStarted: false }); + expect(await read(f.run.id)).toMatchObject({ status: "failed", errorCode: "setup_failed", + resultJson: { failureCause: "40001", nativePreDispatchFailure: { version: 1, providerWorkStarted: false } } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledWith(expect.objectContaining({ id: f.run.id }), f.agent, + { retryReason: "interrupted_preparation" }); + }); + + it("keeps the hold once the run's history shows a native process start", async () => { + const f = await nativeFixture(); + await appendHeartbeatRunEvent(db, { companyId: f.run.companyId, runId: f.run.id, agentId: f.run.agentId, + eventType: "native.process_start_requested", stream: "system", level: "info", message: "Native execution requested", + payload: { providerAttempt: 1 } }); + await f.completion.failRun(f.preDispatch(contention(), false)); + const failed = await read(f.run.id); + expect(failed.resultJson).not.toHaveProperty("nativePreDispatchFailure"); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it.each([ + ["a dispatch that may have started", true], + ["a caller that does not report dispatch", undefined], + ] as const)("keeps the hold for %s", async (_label, mayHaveStarted) => { + const f = await nativeFixture(); + await f.completion.failRun(f.preDispatch(contention(), mayHaveStarted)); + expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure"); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it("keeps the hold when a native coordinator was claimed", async () => { + const f = await nativeFixture(); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Claimed native run" }).returning(); + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.run.id)); + await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, issueId: issue.id, phase: "observed", attempt: 1, leaseOwner: "executor-lease" }); + await f.completion.failRun(f.preDispatch(contention(), false)); + expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure"); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it("re-queues when runtime selection created the coordinator but no executor claimed it", async () => { + const f = await nativeFixture(); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Unclaimed native run" }).returning(); + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id }).where(eq(heartbeatRuns.id, f.run.id)); + await db.insert(nativeRunFinalizations).values({ runId: f.run.id, companyId: f.run.companyId, issueId: issue.id, phase: "observed" }); + await f.completion.failRun(f.preDispatch(contention(), false)); + expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce(); + // The unclaimed coordinator is closed so it cannot hold the successor as pending cleanup. + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.run.id)))[0]) + .toMatchObject({ phase: "terminal_failure", failureCode: "native_pre_dispatch_failure" }); + }); + + it.each([ + ["a recorded process", { processPid: 4242 }], + ["a provider session", { sessionIdAfter: "provider-session" }], + ] as const)("keeps the hold when the run carries %s", async (_label, patch) => { + const f = await nativeFixture(); + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.run.id)); + await f.completion.failRun(f.preDispatch(contention(), false)); + expect((await read(f.run.id)).resultJson).not.toHaveProperty("nativePreDispatchFailure"); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + + it("re-queues a follow-up whose process identity was inherited from its warm predecessor", async () => { + const f = await nativeFixture(); + await db.update(heartbeatRuns).set({ processPid: 4242, runnerProfileJson: { inheritedProcessIdentityRunId: randomUUID() } }) + .where(eq(heartbeatRuns.id, f.run.id)); + await f.completion.failRun(f.preDispatch(contention(), false)); + expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }); + expect(f.deps.scheduleBoundedRetryForRun).toHaveBeenCalledOnce(); + }); + + it("records the proof but does not re-queue a failure that is not a transient database fault", async () => { + const f = await nativeFixture(); + await f.completion.failRun(f.preDispatch(new Error("adapter crashed"), false)); + expect((await read(f.run.id)).resultJson).toMatchObject({ nativePreDispatchFailure: { version: 1, providerWorkStarted: false } }); + expect(f.deps.scheduleBoundedRetryForRun).not.toHaveBeenCalled(); + }); + }); + it("records a workspace provider's reason on the run and in the setup failure log without retrying", async () => { const f = await fixture(); const logged = vi.spyOn(logger, "error"); diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 5759739cd5b..847d63bda94 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1,7 +1,10 @@ import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js"; import { isContextCompactionRun } from "../context-compaction.js"; import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js"; -import { GENERIC_RUN_FAILURE_CODES, failureCauseFromAdapterResult, failureCauseFromError } from "./run-failure-cause.js"; +import { GENERIC_RUN_FAILURE_CODES, NATIVE_PRE_DISPATCH_FAILURE_CODE, NATIVE_PRE_DISPATCH_FAILURE_KEY, NATIVE_PRE_DISPATCH_FAILURE_RECORD, failureCauseFromAdapterResult, failureCauseFromError, + isNativePreDispatchTransientFailure } from "./run-failure-cause.js"; +import { INTERRUPTED_PREPARATION_RETRY_REASON } from "../execution-recovery-attempt.js"; +import { hasNativeExecutionStartEvidence } from "../native-runtime/native-restart-recovery.js"; import { withoutSqlStatementText } from "@paperclipai/shared"; import { nativeTaskSessionPublicationHandled } from "../native-runtime/native-task-session-publication.js"; import { boundOriginalMessage, describeRefusedExactResume, interruptedExactResumeCause } from "../native-runtime/native-exact-resume-failure.js"; @@ -35,8 +38,8 @@ import { accountRunCost } from "../run-cost-accounting.js"; import { isConversation, settleConversationTurn } from "../agent-conversations.js"; import { legacyExecutionNeedsReconciliationWithEvidence } from "../legacy-execution-recovery.js"; import { isWorkspaceGitScanError } from "../workspace-git-operation-scheduler.js"; -import { and, eq, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { and, eq, isNull, sql } from "drizzle-orm"; +import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; import { HttpError } from "../../errors.js"; import { logger } from "../../middleware/logger.js"; import { normalizeResponsibleUserDenialCode } from "../responsible-user-denial-run-outcomes.js"; @@ -152,6 +155,12 @@ export interface FailHeartbeatRunInput extends RunCompletionInput { identityRedactor: Pick, "redact">; requiredWorkspaceRestoreEvidence: Record | null; legacyAdapterEntered: boolean; + /** + * False only when this controller never entered the native dispatch gate and + * is not resuming a restart claim. Omitted means provider work may have + * started, so the failure keeps the native hold. + */ + nativeDispatchMayHaveStarted?: boolean; goalCheckpointSession: { current: { params: Record; displayId: string } | null }; previousSessionDisplayId: string | null; taskSession: TaskSession; @@ -164,6 +173,8 @@ export interface FailHeartbeatRunSetupInput { signal: AbortSignal; /** True once a legacy adapter or native dispatch began provider work. */ providerDispatchStarted: boolean; + /** See FailHeartbeatRunInput.nativeDispatchMayHaveStarted. */ + nativeDispatchMayHaveStarted?: boolean; identityRedactor: Pick, "redact">; readFailureReportSecrets: () => string[]; } @@ -1098,9 +1109,89 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC await settleAgent(); } + /** + * The pre-dispatch record for a native run that failed before provider work + * could start, else nothing. Only a failure path calls it, so the normal + * path gains no I/O. Missing or unreadable evidence keeps the native hold. + */ + async function nativePreDispatchFailureResult(runId: string, dispatchMayHaveStarted: boolean | undefined) { + if (dispatchMayHaveStarted !== false) return null; + try { + const [row] = await db.select({ companyId: heartbeatRuns.companyId, runtimeMode: heartbeatRuns.runtimeMode, + processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + processStartedAt: heartbeatRuns.processStartedAt, sessionIdAfter: heartbeatRuns.sessionIdAfter, + inheritedProcessIdentityRunId: sql`${heartbeatRuns.runnerProfileJson}->>'inheritedProcessIdentityRunId'`, + coordinator: { runId: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, + attempt: nativeRunFinalizations.attempt, leaseOwner: nativeRunFinalizations.leaseOwner, + controllerBootId: nativeRunFinalizations.controllerBootId, resultId: nativeRunFinalizations.resultId } }) + .from(heartbeatRuns) + .leftJoin(nativeRunFinalizations, and(eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId))) + .where(eq(heartbeatRuns.id, runId)).limit(1); + if (!row || row.runtimeMode !== "native") return null; + // A recorded process or provider session also means work may have run. + // Runtime selection copies a warm predecessor's process identity onto a + // follow-up and names that run; such an inherited identity is not this + // run's process. Its own process would still record its start events. + const inheritedProcessIdentity = typeof row.inheritedProcessIdentityRunId === "string"; + if ((!inheritedProcessIdentity && (row.processPid !== null || row.processGroupId !== null + || row.processStartedAt !== null)) || row.sessionIdAfter !== null) return null; + // Runtime selection creates the coordinator unclaimed ("observed", + // attempt 0). Only the executor, after entering dispatch, claims it with + // a lease owner and controller; any claim, in any generation, disqualifies. + const coordinator = row.coordinator; + if (coordinator?.runId && (coordinator.phase !== "observed" || coordinator.attempt !== 0 + || coordinator.leaseOwner !== null || coordinator.controllerBootId !== null || coordinator.resultId !== null)) return null; + // The executor records native.process_start_requested before any backend + // may spawn a runner; runner events carry a source event ID. Any of them, + // in this or an earlier controller generation, means work may have run. + // Restart recovery uses the same canonical proof. + if (await hasNativeExecutionStartEvidence(db, { companyId: row.companyId, id: runId })) return null; + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults) + .where(and(eq(nativeRunResults.companyId, row.companyId), eq(nativeRunResults.runId, runId))).limit(1); + if (result) return null; + return { [NATIVE_PRE_DISPATCH_FAILURE_KEY]: NATIVE_PRE_DISPATCH_FAILURE_RECORD }; + } catch (error) { + logger.warn({ err: error, runId }, "could not prove a native failure started no provider work; keeping the native hold"); + return null; + } + } + + /** + * Runtime selection leaves an unclaimed coordinator ("observed", attempt 0). + * It counts as pending cleanup and would hold the re-queued successor, so a + * failure proven to start no provider work closes it as a terminal failure. + * The guard matches only an unclaimed row; a claimed one keeps its owner. + */ + async function settleUnclaimedNativeCoordinator(run: Pick) { + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: NATIVE_PRE_DISPATCH_FAILURE_CODE, + updatedAt: new Date() }).where(and(eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.runId, run.id), eq(nativeRunFinalizations.phase, "observed"), eq(nativeRunFinalizations.attempt, 0), + isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.controllerBootId), isNull(nativeRunFinalizations.resultId))); + } + + /** + * A native run that failed on a transient database fault before provider + * work re-queues itself in the uncharged, bounded interrupted-preparation + * lane. An answered interaction keeps its own continuation retry. + */ + async function scheduleNativePreDispatchRetry(run: Run, agent: Agent) { + if (!isNativePreDispatchTransientFailure(run) || isResolvedInteractionContinuationWakeContext(run.contextSnapshot)) return null; + try { + await settleUnclaimedNativeCoordinator(run); + } catch (error) { + logger.warn({ err: error, runId: run.id }, "could not close the unclaimed native coordinator; keeping the native hold"); + return null; + } + return scheduleBoundedRetryForRun(run, agent, { retryReason: INTERRUPTED_PREPARATION_RETRY_REASON }).catch((retryError) => { + logger.warn({ err: retryError, runId: run.id }, "failed to re-queue a native run that failed before provider work"); + return null; + }); + } + async function failRun(input: FailHeartbeatRunInput) { const { run, agent, issueId, issueRef, signal, output, session, readFailureReportSecrets, runId, err, identityRedactor, requiredWorkspaceRestoreEvidence, - legacyAdapterEntered, goalCheckpointSession, previousSessionDisplayId, taskSession } = input; + legacyAdapterEntered, nativeDispatchMayHaveStarted, goalCheckpointSession, previousSessionDisplayId, taskSession } = input; const { runtimeForAdapter, previousSessionParams, taskKey, configuredModel, sessionConfigMetadata } = session; const { handle, outputProgressState, flushOutputProgress } = output; // A process adapter may throw while its owned Stop is joining the @@ -1208,6 +1299,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC const stoppedDuringFailure = signal.aborted; const stopSnapshot = stoppedDuringFailure ? await getRun(run.id) : null; const failureOutcome = stoppedDuringFailure ? "cancelled" : "failed"; + const nativePreDispatchFailure = !stoppedDuringFailure && nativeTerminalFailureCode === null + ? await nativePreDispatchFailureResult(run.id, nativeDispatchMayHaveStarted) : null; const failedRunWrite = await measureRunPhase("completion.terminal_write", () => setRunStatusIfRunning(run.id, failureOutcome, { error: message, errorCode: stopSnapshot?.errorCode ?? failureErrorCode, @@ -1231,6 +1324,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC configurationIncompleteFailure?.resultJson ?? {}), ...(stoppedDuringFailure ? controllerLeaseLossResultJson(signal) : {}), + ...nativePreDispatchFailure, ...(!legacyAdapterEntered && run.runtimeMode !== "native" ? { executionRecovery: { @@ -1318,7 +1412,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC }) : null; if (storageWait?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun); - if (storageWait?.outcome !== "scheduled" && storageWait?.outcome !== "retry_exhausted") { + const nativePreDispatchRetry = storageWait === null ? await scheduleNativePreDispatchRetry(livenessRun, agent) : null; + if (storageWait?.outcome !== "scheduled" && storageWait?.outcome !== "retry_exhausted" && nativePreDispatchRetry === null) { await scheduleInteractionContinuationInfrastructureRetryIfEligible( livenessRun, agent, @@ -1330,7 +1425,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC // terminal failure, generic issue recovery must not create a // replacement retryOfRunId chain for the same provider work. suppressImmediateRecovery: nativeTerminalFailureCode !== null || storageWait?.outcome === "scheduled" - || storageWait?.outcome === "retry_exhausted", + || storageWait?.outcome === "retry_exhausted" || nativePreDispatchRetry?.outcome === "scheduled", }); await handleIssueReviewPathDisposition(livenessRun); @@ -1392,7 +1487,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC } async function failRunSetup(input: FailHeartbeatRunSetupInput) { - const { run, runId, outerErr, signal, providerDispatchStarted, identityRedactor, readFailureReportSecrets } = input; + const { run, runId, outerErr, signal, providerDispatchStarted, nativeDispatchMayHaveStarted, identityRedactor, readFailureReportSecrets } = input; // The lease watchdog stopped preparation because this host no longer owns // the run. That is host infrastructure, not a defect in the agent or setup. // A post-dispatch error can also land here (for example a failed usage @@ -1558,6 +1653,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC ...(!providerDispatchStarted ? { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } : {}), + ...(!providerDispatchStarted && !signal.aborted + ? await nativePreDispatchFailureResult(runId, nativeDispatchMayHaveStarted) : null), }; const setupFailureWrite = await setRunStatusIfRunning(runId, "failed", { error: message, @@ -1612,6 +1709,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC failedRun, ).catch(() => failedRun); let storageWaitScheduled = false; + let nativePreDispatchRetryScheduled = false; const setupFailureIssueId = readNonEmptyString( parseObject(livenessRun.contextSnapshot).issueId, ); @@ -1675,7 +1773,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC : livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE ? scheduleBoundedRetryForRun(livenessRun, failedAgent, STORAGE_UNAVAILABLE_RETRY) : scheduleBoundedRetryForRun(livenessRun, failedAgent) - : scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent) + : scheduleNativePreDispatchRetry(livenessRun, failedAgent).then(nativeRetry => nativeRetry + ?? scheduleInteractionContinuationInfrastructureRetryIfEligible(livenessRun, failedAgent)) ).catch((retryError) => { logger.warn( { err: retryError, runId: livenessRun.id }, @@ -1687,10 +1786,11 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC if (livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE && retry?.outcome === "retry_exhausted") await markStorageWaitExhausted(livenessRun); storageWaitScheduled = livenessRun.errorCode === STORAGE_UNAVAILABLE_CODE && (retry?.outcome === "scheduled" || retry?.outcome === "retry_exhausted"); + nativePreDispatchRetryScheduled = retry?.outcome === "scheduled" && isNativePreDispatchTransientFailure(livenessRun); } } await releaseIssueExecutionAndPromote(livenessRun, { - suppressImmediateRecovery: storageWaitScheduled || + suppressImmediateRecovery: storageWaitScheduled || nativePreDispatchRetryScheduled || readNonEmptyString( parseObject(livenessRun.contextSnapshot).goalControlRequestId, ) !== null || diff --git a/server/src/services/heartbeat/run-failure-cause.test.ts b/server/src/services/heartbeat/run-failure-cause.test.ts index 6b5ff341efe..7d69ba3e7e4 100644 --- a/server/src/services/heartbeat/run-failure-cause.test.ts +++ b/server/src/services/heartbeat/run-failure-cause.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "vitest"; import { normalizeRunFailureCause } from "@paperclipai/shared"; -import { failureCauseFromAdapterResult, failureCauseFromError } from "./run-failure-cause.js"; +import { failureCauseFromAdapterResult, failureCauseFromError, isNativePreDispatchTransientFailure } from "./run-failure-cause.js"; describe("run failure cause", () => { it("reads the first safe code from the error and at most three nested causes", () => { @@ -35,4 +35,20 @@ describe("run failure cause", () => { expect(failureCauseFromAdapterResult({ ...base, errorMessage: "select * from secrets where id = $1" })).toBe("adapter_reported_error"); expect(failureCauseFromAdapterResult(base)).toBe("unknown_error"); }); + + it("re-queues only a recorded native pre-dispatch failure with a transient database cause", () => { + const record = { version: 1, providerWorkStarted: false }; + const run = { runtimeMode: "native", status: "failed", errorCode: "adapter_failed", + resultJson: { failureCause: "40001", nativePreDispatchFailure: record } }; + expect(isNativePreDispatchTransientFailure(run)).toBe(true); + expect(isNativePreDispatchTransientFailure({ ...run, errorCode: "setup_failed", resultJson: { failureCause: "57P01", nativePreDispatchFailure: record } })).toBe(true); + // A row saved by an older server has no record and keeps the native hold. + expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "40001" } })).toBe(false); + expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "40001", + nativePreDispatchFailure: { version: 1, providerWorkStarted: true } } })).toBe(false); + expect(isNativePreDispatchTransientFailure({ ...run, resultJson: { failureCause: "Error", nativePreDispatchFailure: record } })).toBe(false); + expect(isNativePreDispatchTransientFailure({ ...run, errorCode: "native_session_interrupted" })).toBe(false); + expect(isNativePreDispatchTransientFailure({ ...run, runtimeMode: "legacy" })).toBe(false); + expect(isNativePreDispatchTransientFailure({ ...run, status: "cancelled" })).toBe(false); + }); }); diff --git a/server/src/services/heartbeat/run-failure-cause.ts b/server/src/services/heartbeat/run-failure-cause.ts index 257f5763560..49f96f4704f 100644 --- a/server/src/services/heartbeat/run-failure-cause.ts +++ b/server/src/services/heartbeat/run-failure-cause.ts @@ -46,3 +46,48 @@ export function failureCauseFromAdapterResult( if (result.errorMessage) return "adapter_reported_error"; return UNKNOWN_RUN_FAILURE_CAUSE; } + +/** + * Durable record on a failed native run that no provider work could have + * started: the controller never entered the dispatch gate, the run was not a + * resumed restart claim, and the run's whole history has no native coordinator + * and no execution-start evidence (`native.process_start_requested` among them). + * Rows saved before this record existed have none and keep the native hold. + */ +export const NATIVE_PRE_DISPATCH_FAILURE_KEY = "nativePreDispatchFailure"; +/** Failure code on the unclaimed native coordinator that such a failure closes. */ +export const NATIVE_PRE_DISPATCH_FAILURE_CODE = "native_pre_dispatch_failure"; +export const NATIVE_PRE_DISPATCH_FAILURE_RECORD = Object.freeze({ version: 1, providerWorkStarted: false } as const); + +export function hasNativePreDispatchFailureRecord(resultJson: unknown): boolean { + if (typeof resultJson !== "object" || resultJson === null || Array.isArray(resultJson)) return false; + const record = (resultJson as Record)[NATIVE_PRE_DISPATCH_FAILURE_KEY]; + if (typeof record !== "object" || record === null) return false; + const fields = record as Record; + return fields.version === 1 && fields.providerWorkStarted === false; +} + +/** + * Database faults that clear on their own: serialization and deadlock aborts, + * a lock timeout, an administrator or crash shutdown, too many connections and + * a dropped connection (SQLSTATEs and postgres.js codes). Storage outages keep + * their own wait lane (`storage_unavailable`). + */ +const TRANSIENT_PRE_DISPATCH_FAILURE_CAUSES: ReadonlySet = new Set([ + "40001", "40P01", "55P03", "57P01", "57P02", "57P03", "53300", "08000", "08001", "08003", "08006", + "CONNECTION_CLOSED", "CONNECTION_ENDED", "CONNECTION_DESTROYED", "CONNECT_TIMEOUT", +]); + +/** + * A failed native run that may re-queue itself without a person: it carries the + * pre-dispatch record, a generic failure code and a transient database cause. + * Anything else keeps the native hold and its manual Retry. + */ +export function isNativePreDispatchTransientFailure(run: { + runtimeMode: string | null; status: string; errorCode: string | null; resultJson: unknown; +}): boolean { + if (run.runtimeMode !== "native" || run.status !== "failed" || !run.errorCode + || !GENERIC_RUN_FAILURE_CODES.has(run.errorCode) || !hasNativePreDispatchFailureRecord(run.resultJson)) return false; + const cause = normalizeRunFailureCause((run.resultJson as Record).failureCause); + return cause !== null && TRANSIENT_PRE_DISPATCH_FAILURE_CAUSES.has(cause); +}