diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 4402393073c..27362c345a2 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1678,6 +1678,23 @@ resumed sandbox, a warm session, a plugin task, a restart claim, or any missing proof keeps the existing `prior_owner_terminal_state_indeterminate` rejection. +Resume source selection skips a failed follow-up that never started provider +work. Such a run holds only the provider checkpoint that it inherited. Without +the skip, a retry selects it, and controller recovery fails with +`native_controller_recovery_unsupported:metadata_format` because the run has no +controller checkpoint. The skip requires all of these: + +- The run failed with a bootstrap receipt that says provider work did not start. +- Its coordinator is `terminal_failure`, with no result, lease owner, or + controller process. +- It has no process identity, provider event, controller checkpoint, run + result, or controller recovery pointer. +- Its checkpoint is bound to this run and session and has no cursor. +- Its checkpoint names the same provider session as the next selected source. + +Otherwise, the run stays a progress barrier. Runtime selection and controller +recovery use the same selector, so both then use the last committed owner. + A repository-free fresh plugin task directory remains the task's directory on ordinary follow-ups. Reuse requires its exact company, task, workspace, provider owner, fresh-attempt identity and path. Mode-only isolation settings do not turn diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 43c14c251d8..bd4fb833711 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -51,7 +51,9 @@ They do not qualify distinct application releases or complete webserver disk los database discovery, one-source import, idempotency, live-owner waiting, corrupt bindings and paginated inventory. `native-recovered-owner-followup.test.ts` admits a per-turn follow-up after recovery committed an owner that never saved -a suspension receipt, and keeps unsettled or ambiguous owners held. +a suspension receipt, and keeps unsettled or ambiguous owners held. It also +retries from that owner after an earlier follow-up failed before provider work, +and keeps any follow-up with provider or controller evidence as the source. `native-session-resume-projection.test.ts` retains the 32-candidate and newer-progress barrier checks. Snapshot transaction counts and stale-generation rejection are exercised by the control-plane port diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index 907fdfef91a..2f91c43a862 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -6,7 +6,8 @@ import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { - agents, companies, completionContracts, createDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, + agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, + nativeRunnerCheckpoints, nativeRunResults, } from "@paperclipai/db"; import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js"; @@ -156,11 +157,10 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered nativeIssueId: prior.issueId, nativeSessionId: prior.native.normalizedSessionId, contextSnapshot: { issueId: prior.issueId } }).returning(); const next = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(prior.issueId), environmentLeaseId: randomUUID() }); await db.insert(nativeRunFinalizations).values({ companyId, issueId: prior.issueId, runId, phase: "observed" }); - await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, prior.issueId)); + await db.update(issues).set({ executionRunId: runId, status: "in_progress" }).where(eq(issues.id, prior.issueId)); const execution = prior.executionFor(runId); const selected = await findNativeSessionResumeRun(db, { companyId, agentId, issueId: prior.issueId, normalizedSessionId: prior.native.normalizedSessionId, currentRunId: runId, beforeCreatedAt: row.createdAt }); - expect(selected?.id).toBe(prior.runId); const admission = buildNativeExecutionWithCheckpoint({ previousRun: selected, normalizedSessionId: prior.native.normalizedSessionId, executionTargetKind: "remote", buildExecution: () => execution }); expect(admission.checkpoint).not.toBeNull(); @@ -176,7 +176,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered runner: { execute: vi.fn(async () => { throw new Error("admission must not reach the sandbox"); }) }, sandboxLeaseAcquisition: { outcome, providerLeaseId: `sandbox-${runId}` } } as never, }).then(() => "constructed", (error: Error) => error.message); - return { runId, logs, start }; + return { runId, logs, start, selectedRunId: selected?.id ?? null }; } async function quarantined() { @@ -203,6 +203,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered "admits a fresh runner authority when the recovered owner is provably finished: %s sandbox", async outcome => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); + expect(followUp.selectedRunId).toBe(prior.runId); expect(await followUp.start(outcome)).toBe(ADMITTED); expect(await quarantined()).toEqual([]); const root = await sessionRoot(); @@ -212,6 +213,91 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); }, 60_000); + /** The state left by a follow-up admitted before retirement existed. Its + * admission quarantined the rebuilt projection and failed before provider + * work; it keeps only the provider checkpoint inherited from the owner. */ + async function failUnstartedFollowUp(prior: Awaited>) { + const followUp = await admitFollowUp(prior); + const error = await (async () => { + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + try { + return await followUp.start(); + } finally { + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); + } + })(); + expect(error).toBe(STRANDED); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed", error, finishedAt: new Date(), + nativePhase: "observed", resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }) + .where(eq(heartbeatRuns.id, followUp.runId)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_continuation_requires_reconciliation" }) + .where(eq(nativeRunFinalizations.runId, followUp.runId)); + await db.update(issues).set({ executionRunId: null, status: "blocked" }).where(eq(issues.id, prior.issueId)); + return followUp.runId; + } + + it("retries from the committed owner after a follow-up failed before provider work", async () => { + const prior = await seedRecoveredPriorRun(); + await failUnstartedFollowUp(prior); + const retry = await admitFollowUp(prior); + expect(await retry.start()).toBe(ADMITTED); + expect(retry.selectedRunId).toBe(prior.runId); + expect(await archivedOwners(await sessionRoot())).toEqual([prior.runId]); + }, 60_000); + + it("skips every unstarted follow-up after repeated failed retries", async () => { + const prior = await seedRecoveredPriorRun(); + const failed = await failUnstartedFollowUp(prior); + const retry = await admitFollowUp(prior); + await db.update(heartbeatRuns).set({ status: "failed", errorCode: "adapter_failed", finishedAt: new Date(), + error: "native_controller_recovery_unsupported:metadata_format", + resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } } }).where(eq(heartbeatRuns.id, retry.runId)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure" }).where(eq(nativeRunFinalizations.runId, retry.runId)); + const next = await admitFollowUp(prior); + expect([failed, retry.runId]).not.toContain(next.selectedRunId); + expect(next.selectedRunId).toBe(prior.runId); + expect(await next.start()).toBe(ADMITTED); + }, 60_000); + + it.each(["process identity", "provider event", "controller checkpoint", "provider work receipt", "claimed coordinator", + "run result", "different provider session", "own progress cursor"] as const)( + "keeps a failed follow-up as the source with %s", async scenario => { + const prior = await seedRecoveredPriorRun(); + const failed = await failUnstartedFollowUp(prior); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, failed)); + const checkpoint = run.runnerProfileJson!.sessionCheckpoint as Record; + if (scenario === "process identity") { + await db.update(heartbeatRuns).set({ processPid: 4_242 }).where(eq(heartbeatRuns.id, failed)); + } + if (scenario === "provider event") { + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: failed, seq: 1, eventType: "session.started", + stream: "system", payload: {} }); + } + if (scenario === "controller checkpoint") { + await db.insert(nativeRunnerCheckpoints).values({ companyId, issueId: prior.issueId, runId: failed, part: "header", material: {} }); + } + if (scenario === "provider work receipt") { + await db.update(heartbeatRuns).set({ resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: true } } }) + .where(eq(heartbeatRuns.id, failed)); + } + if (scenario === "claimed coordinator") { + await db.update(nativeRunFinalizations).set({ controllerPid: 4_242 }).where(eq(nativeRunFinalizations.runId, failed)); + } + if (scenario === "run result") { + await db.insert(nativeRunResults).values({ companyId, issueId: prior.issueId, runId: failed, + completionContractId: run.completionContractId!, serverFingerprint: "fixture", schemaStatus: "rejected", + resultJson: {}, canonicalSha256: "fixture" }); + } + if (scenario === "different provider session" || scenario === "own progress cursor") { + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson, sessionCheckpoint: { + ...checkpoint, ...(scenario === "own progress cursor" ? { cursor: { sourceSeq: 1 } } : { providerSessionId: "another-provider-session" }) } } }) + .where(eq(heartbeatRuns.id, failed)); + } + const retry = await admitFollowUp(prior); + expect(retry.selectedRunId).toBe(failed); + expect(await retry.start()).toBe("native_controller_recovery_unsupported:metadata_format"); + }, 60_000); + it("admits again when a retry rebuilds the same projection from the database", async () => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); diff --git a/server/src/services/native-runtime/native-session-resume.ts b/server/src/services/native-runtime/native-session-resume.ts index f297fa95057..135eb9c27b2 100644 --- a/server/src/services/native-runtime/native-session-resume.ts +++ b/server/src/services/native-runtime/native-session-resume.ts @@ -1,7 +1,9 @@ import { createHash, randomUUID } from "node:crypto"; import { isDeepStrictEqual } from "node:util"; -import { and, desc, eq, exists, inArray, isNotNull, ne, not, or, sql } from "drizzle-orm"; -import { heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db"; +import { and, desc, eq, exists, inArray, isNotNull, ne, not, notInArray, or, sql } from "drizzle-orm"; +import { + heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, nativeRunnerCheckpoints, nativeRunResults, type Db, +} from "@paperclipai/db"; import type { NativeExecutionInput, PersistedNativeSession, @@ -214,20 +216,78 @@ export async function nativeSessionProviderEvidence( return new Set(rows.map((row) => row.runId)); } -export async function findNativeSessionResumeRun( - db: Db, - input: { - companyId: string; - agentId: string; - issueId: string; - normalizedSessionId: string; - currentRunId: string; - beforeCreatedAt: Date; - /** Recovery discovery must retain an ineligible newest progress barrier. */ - includeProgressBarrier?: boolean; - }, -) { +type NativeSessionResumeSelection = { + companyId: string; + agentId: string; + issueId: string; + normalizedSessionId: string; + currentRunId: string; + beforeCreatedAt: Date; + /** Recovery discovery must retain an ineligible newest progress barrier. */ + includeProgressBarrier?: boolean; +}; + +const NATIVE_RESUME_CANDIDATE_WINDOW = 32; +const INHERITED_CHECKPOINT_SOURCE_FIELDS = ["backendKind", "driverKind", "sessionId", "providerSessionId"] as const; + +export async function findNativeSessionResumeRun(db: Db, input: NativeSessionResumeSelection) { if (!isNativeSessionId(input.normalizedSessionId)) return null; + const barrier = await selectNativeSessionResumeBarrier(db, input, []); + // A follow-up that failed before any provider or controller work holds only + // the checkpoint it inherited. Select its source instead, but only when + // every skipped run proves that and inherited exactly that source session. + const skipped: NonNullable[] = []; + let selected = barrier; + while (selected && skipped.length < NATIVE_RESUME_CANDIDATE_WINDOW && + await isUnstartedInheritedFollowUp(db, input, selected)) { + skipped.push(selected); + selected = await selectNativeSessionResumeBarrier(db, input, skipped.map(run => run.id)); + } + if (!skipped.length) return selected; + const source = record(selected?.runnerProfileJson.sessionCheckpoint); + const inherited = selected !== null && typeof source.sessionId === "string" && skipped.every(run => { + const checkpoint = record(run.runnerProfileJson.sessionCheckpoint); + return INHERITED_CHECKPOINT_SOURCE_FIELDS.every(field => checkpoint[field] === source[field]); + }); + return inherited ? selected : barrier; +} + +/** The run row, its coordinator, and the absence of controller checkpoints, + * results and provider events together prove no provider work started. */ +async function isUnstartedInheritedFollowUp( + db: Db, + input: NativeSessionResumeSelection, + run: { id: string; runtimeMode: string | null; status: string; processPid: number | null; + processGroupId: number | null; processStartedAt: Date | null; runnerProfileJson: { sessionCheckpoint: unknown; nativeControllerRecovery: unknown } }, +): Promise { + const checkpoint = record(run.runnerProfileJson.sessionCheckpoint); + const identity = record(checkpoint.identity); + if (run.runtimeMode !== "native" || run.status !== "failed" || run.processPid !== null || + run.processGroupId !== null || run.processStartedAt !== null || run.runnerProfileJson.nativeControllerRecovery != null || + identity.runId !== run.id || identity.sessionId !== input.normalizedSessionId || checkpoint.cursor !== null) return false; + const [evidence] = await db.select({ + executionRecovery: sql`${heartbeatRuns.resultJson}->'executionRecovery'`, + phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, + leaseOwner: nativeRunFinalizations.leaseOwner, + controllerPid: nativeRunFinalizations.controllerPid, + controllerCheckpoint: sql`exists (select 1 from ${nativeRunnerCheckpoints} + where ${nativeRunnerCheckpoints.companyId} = ${heartbeatRuns.companyId} and ${nativeRunnerCheckpoints.runId} = ${heartbeatRuns.id})`, + result: sql`exists (select 1 from ${nativeRunResults} + where ${nativeRunResults.companyId} = ${heartbeatRuns.companyId} and ${nativeRunResults.runId} = ${heartbeatRuns.id})`, + }).from(heartbeatRuns).innerJoin(nativeRunFinalizations, and( + eq(nativeRunFinalizations.runId, heartbeatRuns.id), eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), + )).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId))); + const recovery = record(evidence?.executionRecovery); + return Boolean(evidence && recovery.kind === "bootstrap" && recovery.providerWorkStarted === false && + evidence.phase === "terminal_failure" && evidence.resultId === null && evidence.leaseOwner === null && + evidence.controllerPid === null && !evidence.controllerCheckpoint && !evidence.result && + !(await nativeSessionProviderEvidence(db, [run.id])).has(run.id)); +} + +async function selectNativeSessionResumeBarrier(db: Db, input: NativeSessionResumeSelection, skippedRunIds: string[]) { const candidates = db.$with("native_resume_candidates").as(db .select({ id: heartbeatRuns.id, @@ -250,12 +310,12 @@ export async function findNativeSessionResumeRun( ), ) .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) - .limit(32)); + .limit(NATIVE_RESUME_CANDIDATE_WINDOW)); const terminalStatuses = [...LEGACY_RETRY_SOURCE_TERMINAL_STATUSES]; const barrier = db.$with("native_resume_barrier").as(db .select() .from(candidates) - .where(or( + .where(and(skippedRunIds.length ? notInArray(candidates.id, skippedRunIds) : undefined, or( ne(candidates.runtimeMode, "native"), not(inArray(candidates.status, terminalStatuses)), candidates.hasCheckpoint, @@ -268,7 +328,7 @@ export async function findNativeSessionResumeRun( eq(heartbeatRunEvents.runId, candidates.id), inArray(heartbeatRunEvents.eventType, NATIVE_SESSION_PROVIDER_EVENT_TYPES), ))), - )) + ))) .orderBy(desc(candidates.createdAt), desc(candidates.id)) .limit(1)); const [run] = await db.with(candidates, barrier)