diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 29e7669f97e..9ac6842d450 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -347,6 +347,14 @@ For a prepared remote workspace interrupted before its first executor claim, that ledger records `remote_execution_never_claimed` with a `bootstrap_incomplete` disposition. The provider attempt remains zero during classification; the executor increments it before starting provider work. +A claimed remote bootstrap (the sandbox runner was launched and may have +connected) records `remote_bootstrap_proven_incomplete` with the same +disposition. Recovery admits it only after the previous controller lease has +expired and the Postgres controller checkpoint holds just `run.prepare` and an +unfinished `session.open`, with no committed or acknowledged event. The +executor restores that checkpoint and checks the same proof before it +bootstraps again. After three provider attempts it records +`execution_recovery_budget_exhausted` instead. For a local Codex, OpenCode, or ACPX turn lost while runnerd was stopped, the canonical `turn.failed` event retains `error.code: "provider_turn_lost_on_restore"` and diff --git a/server/src/__tests__/execution-workspaces-service.test.ts b/server/src/__tests__/execution-workspaces-service.test.ts index cb1dde02123..9e114d6ee68 100644 --- a/server/src/__tests__/execution-workspaces-service.test.ts +++ b/server/src/__tests__/execution-workspaces-service.test.ts @@ -2419,6 +2419,129 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => { ])); }); + async function seedSharedLocalWorkspace(cwd: string, options: { metadata?: Record } = {}) { + const companyId = randomUUID(); + const projectId = randomUUID(); + const projectWorkspaceId = randomUUID(); + const executionWorkspaceId = randomUUID(); + await db.insert(companies).values({ + id: companyId, + name: "Paperclip", + issuePrefix: "PAP", + requireBoardApprovalForNewAgents: false, + }); + await db.insert(projects).values({ + id: projectId, + companyId, + name: "Workspaces", + status: "in_progress", + executionWorkspacePolicy: { enabled: true }, + }); + await db.insert(projectWorkspaces).values({ + id: projectWorkspaceId, + companyId, + projectId, + name: "Primary", + sourceType: "local_path", + isPrimary: true, + cwd, + }); + await db.insert(executionWorkspaces).values({ + id: executionWorkspaceId, + companyId, + projectId, + projectWorkspaceId, + mode: "shared_workspace", + strategyType: "project_primary", + name: "Shared workspace", + status: "active", + providerType: "local_fs", + cwd, + metadata: options.metadata ?? null, + }); + return { executionWorkspaceId }; + } + + it("treats an existing workspace directory that is not a git repository as ready", async () => { + const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-non-git-workspace-")); + tempDirs.add(workspaceDir); + await fs.writeFile(path.join(workspaceDir, "notes.txt"), "plain files\n", "utf8"); + const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir); + + const readiness = await svc.getCloseReadiness(executionWorkspaceId); + + expect(readiness).toMatchObject({ + state: "ready_with_warnings", + isDestructiveCloseAllowed: true, + git: { repoRoot: null, hasDirtyTrackedFiles: false, hasUntrackedFiles: false }, + }); + expect(readiness?.blockingReasons).toEqual([]); + expect(readiness?.warnings.some((warning) => warning.startsWith("Could not inspect git status"))).toBe(false); + }); + + it("plans removal of a runtime-created plain workspace directory", async () => { + const parentDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-non-git-runtime-")); + tempDirs.add(parentDir); + const workspaceDir = path.join(parentDir, "task"); + await fs.mkdir(workspaceDir); + const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir, { + metadata: { createdByRuntime: true }, + }); + await db.update(projectWorkspaces).set({ cwd: path.join(parentDir, "primary") }) + .where(eq(projectWorkspaces.cwd, workspaceDir)); + + const readiness = await svc.getCloseReadiness(executionWorkspaceId); + + expect(readiness?.blockingReasons).toEqual([]); + expect(readiness?.git).toMatchObject({ repoRoot: null, createdByRuntime: true }); + expect(readiness?.plannedActions.map((action) => action.kind)).toContain("remove_local_directory"); + }); + + it("still blocks close when an ancestor directory holds an unreadable .git marker", async () => { + const parentDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ancestor-git-workspace-")); + tempDirs.add(parentDir); + await fs.writeFile(path.join(parentDir, ".git"), `gitdir: ${path.join(parentDir, "missing-git-dir")}\n`, "utf8"); + const workspaceDir = path.join(parentDir, "nested", "task"); + await fs.mkdir(workspaceDir, { recursive: true }); + const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir); + + const readiness = await svc.getCloseReadiness(executionWorkspaceId); + + expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false }); + expect(readiness?.blockingReasons).toContain( + "Paperclip could not verify the workspace git status. Retry before destructive cleanup.", + ); + }); + + it("still blocks close when the workspace .git marker is a dangling symlink", async () => { + const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-dangling-git-workspace-")); + tempDirs.add(workspaceDir); + await fs.symlink(path.join(workspaceDir, "unmounted-repo", ".git"), path.join(workspaceDir, ".git")); + const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir); + + const readiness = await svc.getCloseReadiness(executionWorkspaceId); + + expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false }); + expect(readiness?.blockingReasons).toContain( + "Paperclip could not verify the workspace git status. Retry before destructive cleanup.", + ); + }); + + it("still blocks close when a workspace git repository cannot be inspected", async () => { + const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-broken-git-workspace-")); + tempDirs.add(workspaceDir); + // A worktree-style .git pointer whose git dir is missing: git refuses to read it. + await fs.writeFile(path.join(workspaceDir, ".git"), `gitdir: ${path.join(workspaceDir, "missing-git-dir")}\n`, "utf8"); + const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir); + + const readiness = await svc.getCloseReadiness(executionWorkspaceId); + + expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false }); + expect(readiness?.blockingReasons).toContain( + "Paperclip could not verify the workspace git status. Retry before destructive cleanup.", + ); + }); + it("clears matching environment selections transactionally without touching other workspaces", async () => { const companyId = randomUUID(); const projectId = randomUUID(); diff --git a/server/src/__tests__/task-interruption-query.test.ts b/server/src/__tests__/task-interruption-query.test.ts index df854e62152..f0352841235 100644 --- a/server/src/__tests__/task-interruption-query.test.ts +++ b/server/src/__tests__/task-interruption-query.test.ts @@ -4,6 +4,7 @@ import { createRunSecretRedactionRegistry } from "../services/run-secret-redacti import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, agentWakeupRequests, companies, createDb, applyPendingMigrations, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db"; import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js"; +import { readNativeSessionResumptionAgent } from "../services/native-runtime/native-session-resumption-admission.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js"; @@ -39,6 +40,49 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => await taskInterruptionPresentation(db, issue)]) { expect(presentation).toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() }); } + // The coordinator also pushes nextAttemptAt while it defers an ineligible + // candidate. Those deferrals, or one far past the failure, never hide Retry. + const otherAgentId = randomUUID(); + await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }); + const admission = async () => { + const [currentIssue] = await db.select().from(issues).where(eq(issues.id, issue.id)); + const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + return readNativeSessionResumptionAgent(db, { issue: currentIssue!, run: currentRun!, coordinator: coordinator! }); + }; + expect(await admission()).not.toBeNull(); + const otherRunId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: otherRunId, companyId, agentId: otherAgentId, status: "cancelled", + finishedAt: new Date(Date.now() - 60_000), createdAt: new Date(Date.now() - 60 * 60_000) }); + const terminatedManagerId = randomUUID(); + await db.insert(agents).values({ id: terminatedManagerId, companyId, name: "Former manager", + adapterType: "paperclip_runner", status: "terminated" }); + for (const [table, change, restore] of [ + ["issue", { assigneeAgentId: otherAgentId }, { assigneeAgentId: agentId }], + ["issue", { status: "blocked" }, { status: "in_progress" }], + ["issue", { executionRunId: otherRunId }, { executionRunId: null }], + ["issue", { checkoutRunId: otherRunId }, { checkoutRunId: null }], + ["run", { contextSnapshot: { nativeQuestionCancellation: { requestId: "q" } } }, { contextSnapshot: {} }], + // Invokability is more than status: an idle agent under a terminated manager is refused. + ["agent", { reportsTo: terminatedManagerId }, { reportsTo: null }], + ] as const) { + if (table === "issue") await db.update(issues).set(change).where(eq(issues.id, issue.id)); + else if (table === "run") await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, runId)); + else await db.update(agents).set(change).where(eq(agents.id, agentId)); + // The scheduler would defer this candidate, so the task offers Retry. + if (table !== "run") expect(await admission()).toBeNull(); + expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id)) + .toMatchObject({ state: "failed", canRetry: true }); + if (table === "issue") await db.update(issues).set(restore).where(eq(issues.id, issue.id)); + else if (table === "run") await db.update(heartbeatRuns).set(restore).where(eq(heartbeatRuns.id, runId)); + else await db.update(agents).set(restore).where(eq(agents.id, agentId)); + } + expect(await admission()).not.toBeNull(); + // A provider cooldown can schedule the resume an hour out; it stays visible. + const cooldown = new Date(Date.now() + 60 * 60_000); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: cooldown }).where(eq(nativeRunFinalizations.runId, runId)); + expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id)) + .toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: cooldown.toISOString() }); // Exhausted recovery is terminal: exactly one Retry returns. await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null }) .where(eq(nativeRunFinalizations.runId, runId)); @@ -46,6 +90,29 @@ describe.skipIf(!support.supported)("task interruption latest run query", () => .toMatchObject({ state: "failed", canRetry: true }); }); + it("decides several pending native resumes in one list read", async () => { + const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Resuming list", issuePrefix: `L${companyId.slice(0, 7)}` }); + await db.insert(agents).values([{ id: agentId, companyId, name: "Resuming agent", adapterType: "paperclip_runner" }, + { id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }]); + const resumeAt = new Date(Date.now() + 30_000); + const list = []; + for (let index = 0; index < 6; index++) { + const [issue] = await db.insert(issues).values({ companyId, title: `Resume ${index}`, status: "in_progress", + assigneeAgentId: index % 2 ? otherAgentId : agentId }).returning(); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue.id, scopeKind: "issue", + status: "failed", runtimeMode: "native", errorCode: "native_session_interrupted", finishedAt: new Date() }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "retryable_failure", + attempt: 1, nextAttemptAt: resumeAt }); + list.push(issue); + } + const result = await taskInterruptionPresentations(db, companyId, list); + list.forEach((issue, index) => expect(result.get(issue.id)).toMatchObject(index % 2 + ? { state: "failed", canRetry: true } + : { state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() })); + }); + it("redacts workspace preparation errors on task detail and list reads", async () => { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64)); try { diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 4a4601844bf..3b29a35aa28 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -521,6 +521,31 @@ async function pathExists(value: string | null | undefined) { } } +/** + * Returns false only when no `.git` entry exists on the path or any ancestor. + * Uses lstat so a dangling `.git` symlink still counts, and treats any error + * other than ENOENT as "may exist" so unreadable locations fail closed. + */ +async function hasGitMarkerInPathOrAncestors(start: string) { + let current: string; + try { + current = await fs.realpath(start); + } catch { + return true; + } + while (true) { + try { + await fs.lstat(path.join(current, ".git")); + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException | null)?.code !== "ENOENT") return true; + } + const parent = path.dirname(current); + if (parent === current) return false; + current = parent; + } +} + async function runGit(args: string[], cwd: string) { return await execFileAsync("git", ["-C", cwd, ...args], { cwd }); } @@ -963,6 +988,29 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise< try { repoRoot = (await runGit(["rev-parse", "--show-toplevel"], workspacePath)).stdout.trim() || null; } catch (error) { + // A plain directory (e.g. a local_fs shared workspace) has no git state to + // lose. Only treat it that way when no `.git` exists on the path, so a + // broken or unreadable repository still fails closed. + if (workspace.providerType !== "git_worktree" && !(await hasGitMarkerInPathOrAncestors(workspacePath))) { + return { + git: { + repoRoot: null, + workspacePath, + branchName: workspace.branchName, + baseRef: workspace.baseRef, + hasDirtyTrackedFiles: false, + hasUntrackedFiles: false, + dirtyEntryCount: 0, + untrackedEntryCount: 0, + aheadCount: null, + behindCount: null, + isMergedIntoBase: null, + createdByRuntime, + }, + warnings, + statusInspectionSucceeded: true, + }; + } warnings.push( `Could not inspect git status for "${workspacePath}": ${error instanceof Error ? error.message : String(error)}`, ); diff --git a/server/src/services/native-runtime/native-bootstrap-proof.ts b/server/src/services/native-runtime/native-bootstrap-proof.ts new file mode 100644 index 00000000000..786db4d3034 --- /dev/null +++ b/server/src/services/native-runtime/native-bootstrap-proof.ts @@ -0,0 +1,68 @@ +/** + * Pure proofs over durable runner state that a bootstrap never reached + * provider work. Restart recovery and the executor share them, so the claim + * and the executor's own check accept exactly the same shapes. + */ + +export const RUNNERD_CONTROL_PLANE_STATE_SCHEMAS: ReadonlySet = new Set([ + "paperclip.runner.durable.control-plane-state.v1", + "paperclip.runner.durable.control-plane-state.v2", +]); + +const RUNNERD_BOOTSTRAP_COMMAND_TYPES = new Set(["run.prepare", "session.open"]); + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; +} + +/** The runner never connected, so every bootstrap command is still undelivered. */ +export function runnerdStateIsUnconnectedBootstrap(state: Record): boolean { + const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; + const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : []; + return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) && + state.connectionCount === 0 && committedEvents.length === 0 && + commands.every(command => command.status === "pending" && + RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type))); +} + +/** + * A provider turn can start only after session.open completes and the + * controller issues a turn command. A root whose controller journal stops + * before that point (run.prepare may be acknowledged, session.open never is, + * no event was committed or acknowledged) and whose runner journal emitted no + * event either holds no provider work, so restart recovery may quarantine it + * and bootstrap again. Any other shape stays fail-closed. + * + * `runnerJournal` is `undefined` when the runner journal is not on this host + * (never written locally, or a remote runner), `null` when it is unreadable. + */ +export function runnerdStateIsIncompleteBootstrap( + state: Record, + runnerJournal?: Record | null, +): boolean { + if (runnerdStateIsUnconnectedBootstrap(state)) return true; + if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false; + if (typeof state.connectionCount !== "number" || !Number.isSafeInteger(state.connectionCount) || + state.connectionCount < 0) return false; + if (!Array.isArray(state.committedEvents) || state.committedEvents.length !== 0) return false; + if (state.ackedSourceSeq !== 0) return false; + if (state.warmTransition != null || state.completedWarmTransition != null || + state.stagedRunAttachment != null) return false; + if (record(state.bootstrapHistory).adopted === true) return false; + if (!Array.isArray(state.commands)) return false; + const commands = state.commands.map(record); + if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) && + (command.status === "pending" || + (command.type === "run.prepare" && command.status === "completed")))) return false; + // The controller cannot see runner events that were emitted but never + // committed. When the runner journal exists it must agree that none were. + if (runnerJournal === undefined) return true; + if (runnerJournal === null) return false; + const processed = Object.values(record(runnerJournal.processedCommands)).map(record); + return Array.isArray(runnerJournal.outbox) && runnerJournal.outbox.length === 0 && + runnerJournal.nextSourceSeq === 1 && runnerJournal.ackedSourceSeq === 0 && + processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) && + (command.commandType === "run.prepare" || command.status !== "completed")); +} diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index ba43627916a..5d97112d267 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,4 +1,5 @@ import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; +import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { pendingNativeRunProjectionCondition } from "./native-run-projection.js"; import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js"; @@ -158,7 +159,7 @@ export async function claimNativeSessionResumptions(input: { if (await retireSupersededNativeRetry(tx as unknown as Db, { run, coordinator, issueExecutionRunId: issue.executionRunId, now, })) return false; - if (!["todo", "in_progress", "in_review"].includes(issue.status)) return false; + if (!(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status)) return false; const row = { run, coordinator }; if ( row.run.runtimeMode !== "native" || diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 1baceb4e514..d750a83a141 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -11,7 +11,8 @@ import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-pre import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js"; -import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader } from "./native-runner-checkpoint.js"; +import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader, + nativeRunnerCheckpointProvesIncompleteBootstrap } from "./native-runner-checkpoint.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js"; import { activeNativeExecutionRunIds, NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js"; @@ -340,6 +341,8 @@ export function classifyNativeRunnerRecoveryEvidence(input: { runnerGroupAlive: boolean; remoteSandbox?: boolean; unclaimedBootstrap?: boolean; + /** The durable controller checkpoint of a claimed remote bootstrap proves no provider work. */ + remoteBootstrapProvenIncomplete?: boolean; processStartMatches: boolean; knownProviderProcessAlive?: boolean; knownProviderProcessIdentityAmbiguous?: boolean; @@ -368,6 +371,12 @@ export function classifyNativeRunnerRecoveryEvidence(input: { if (input.unclaimedBootstrap === true && !input.hasCheckpoint && !input.hasProviderEvidence) { return { claimKind: "bootstrap_incomplete", reason: "remote_execution_never_claimed" }; } + if (input.remoteBootstrapProvenIncomplete === true && !input.hasCheckpoint && !input.hasProviderEvidence) { + // Each same-run bootstrap is another provider attempt; keep restarts bounded. + if ((input.providerAttempt ?? 0) >= 3) + return { claimKind: null, reason: "execution_recovery_budget_exhausted" }; + return { claimKind: "bootstrap_incomplete", reason: "remote_bootstrap_proven_incomplete" }; + } return input.hasCheckpoint && input.checkpointIdentityMatches === true && input.hasProviderEvidence @@ -1017,9 +1026,23 @@ export async function claimNativeRestartRecoveries(input: { unclaimedBootstrap = !executionEvent; } + // A claimed remote bootstrap (runner launched, perhaps connected) has no + // provider evidence when its controller dies before session.open + // completes. The owner's lease has expired above, and the controller + // writes every command to Postgres before sending it, so that checkpoint + // is the complete record of what reached the sandbox runner. When it + // proves no provider work, bootstrap again instead of holding the task. + const remoteBootstrapProvenIncomplete = !unclaimedBootstrap && remoteWorkspace !== null && !pluginTask + && Boolean(durableAuthority) && stagedAttachment === null && !hasCheckpoint && !hasProviderEvidence + && Boolean(row.run.runnerInstanceId && row.run.nativeSessionId) + && await nativeRunnerCheckpointProvesIncompleteBootstrap(tx, { + companyId: row.run.companyId, issueId: row.coordinator.issueId, runId: row.run.id, + }, { runnerInstanceId: row.run.runnerInstanceId!, normalizedSessionId: row.run.nativeSessionId! }); + const classification = classifyNativeRunnerRecoveryEvidence({ remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, + remoteBootstrapProvenIncomplete, runnerPidAlive, runnerGroupAlive, processStartMatches: exactRunnerIdentity, diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 9e7340553d5..f72e9123c6c 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -5,6 +5,7 @@ import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint, import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; import { assertNativeSourceCheckpointConsumer, lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js"; +import { runnerdStateIsIncompleteBootstrap } from "./native-bootstrap-proof.js"; type Binding = { companyId: string; issueId: string; runId: string }; export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024; @@ -248,6 +249,53 @@ export async function loadNativeUnclaimedBootstrapCheckpoint( } catch { return null; } } +const BOOTSTRAP_CHECKPOINT_PARTS = new Set(["header", "cursor", "command:1", "command:2"]); + +/** + * A remote runner's controller state lives only in this checkpoint, and every + * command is written here before the controller sends it. Once the previous + * controller's lease has expired, the checkpoint is therefore a complete record + * of what reached the sandbox runner. Returns true only when that record holds + * the bootstrap commands alone (run.prepare, an unfinished session.open) with + * no committed or acknowledged event, so no provider turn can exist. The + * sandbox runner journal is not on this host; the proof uses the controller + * record alone, as for a local root whose runner never journaled. + */ +export async function nativeRunnerCheckpointProvesIncompleteBootstrap( + db: Pick, binding: Binding, expected: { runnerInstanceId: string; normalizedSessionId: string }, +): Promise { + try { + const parts = await db.select({ part: nativeRunnerCheckpoints.part }).from(nativeRunnerCheckpoints).where(and( + eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId), + eq(nativeRunnerCheckpoints.runId, binding.runId))).limit(BOOTSTRAP_CHECKPOINT_PARTS.size + 1); + if (!parts.length || parts.length > BOOTSTRAP_CHECKPOINT_PARTS.size + || parts.some(row => !BOOTSTRAP_CHECKPOINT_PARTS.has(row.part))) return false; + const checkpoint = await loadNativeRunnerCheckpoint(db, binding); + if (!checkpoint || checkpoint.predecessor !== null + || checkpoint.recoveryEvidence.runnerInstanceId !== expected.runnerInstanceId + || checkpoint.recoveryEvidence.normalizedSessionId !== expected.normalizedSessionId) return false; + // The decoder above validated these canonical v2 records. Rebuild the + // controller state in memory; the part check admits no event records. + const read = (part: string) => { + const value = checkpoint.records.get(part); + return value === undefined ? undefined : JSON.parse(value) as Record; + }; + const header = read("header"); + const cursor = read("cursor"); + if (header?.schema !== "paperclip.runner.controller-checkpoint.v2" || !cursor) return false; + const state: Record = { + ...(header.state as Record), + ackedSourceSeq: cursor.ackedSourceSeq, + commands: ["command:1", "command:2"].map(read).filter(command => command !== undefined), + committedEvents: [], + }; + return (state.identity as { runId?: unknown } | undefined)?.runId === binding.runId + && runnerdStateIsIncompleteBootstrap(state); + } catch { + return false; + } +} + export async function restoreNativeRunnerCheckpoint(db: Db, binding: Binding, root: string, normalizedSessionId?: string) { const checkpoint = await loadNativeRunnerCheckpoint(db, binding); if (!checkpoint) return null; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 5241fe2b5b5..a9cc8f325cc 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -60,7 +60,10 @@ import { detachNativeSessionsForRestart, executePaperclipNativeSession, NativeControllerDetachedForRestartError, + restoreRemoteBootstrapCheckpointForRecovery, + runnerdStateProvesIncompleteBootstrap, } from "./native-session-executor.js"; +import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import { listExpiredNativeRunnerAuthorities, loadNativeRunnerCheckpoint, @@ -2220,6 +2223,140 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }, ); + // RC-29: a restart while a Daytona runner was connecting. The executor had + // claimed the run and launched the sandbox runner, which authenticated, but + // session.open never completed. The controller's Postgres checkpoint is the + // only record of what the runner received. + async function seedClaimedRemoteBootstrap(label: string, options: { + attempt?: number; + mutate?: (core: DurablePrpControlPlane) => void; + checkpoint?: boolean; + } = {}) { + const fixture = await seedRun(`REMOTE-CLAIMED-${label}`); + const { runId, issueId, native } = fixture; + const remote = remotePreparedWorkspace(); + await fixture.db.update(heartbeatRuns).set({ + runnerProfileJson: { ...fixture.preparedProfile, nativeWorkspaceSync: remote }, + processPid: 443, processStartedAt: new Date(Date.now() - 30_000), + }).where(eq(heartbeatRuns.id, runId)); + await fixture.db.insert(heartbeatRunEvents).values([ + { companyId, runId, agentId, seq: 900, stream: "system", level: "info", eventType: "native.process_start_requested" }, + { companyId, runId, agentId, seq: 901, stream: "system", level: "info", eventType: "native.process_identity_recorded", + payload: { targetKind: "remote", processPid: 443 } }, + ]); + const bootId = randomUUID(); + const leaseOwner = `${bootId}:1:${randomUUID()}`; + await fixture.db.update(nativeRunFinalizations).set({ + attempt: options.attempt ?? 1, controllerGeneration: 1, controllerBootId: bootId, controllerPid: 2_000_000_001, + controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000), + }).where(eq(nativeRunFinalizations.runId, runId)); + if (options.checkpoint !== false) { + const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId, + normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` }; + const core = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId, "control-plane"), identity, + expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + try { + const template = { provider: { provider: "acpx", command: "fixture" }, authorizedTools: {} }; + core.persistRunAttachTemplate(template); + core.queueCommand("run.prepare", template); + core.queueCommand("session.open", { reuse: "same_session" }); + core.issueBootstrapTicket(); + // The runner authenticated once and received run.prepare. + core.store.state.connectionCount = 1; + core.store.state.freshBootstraps = 1; + core.store.state.commandDeliveryCounts = { [core.store.state.commands[0]!.commandId]: 1 }; + options.mutate?.(core); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(fixture.db, { + companyId, issueId, runId, authorityRunId: runId, bootId, leaseOwner, generation: 1, + })); + await core.flushCheckpoint(); + } finally { + await core.stop(); + } + } + // The killed controller can no longer renew its lease. + await fixture.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1_000) }) + .where(eq(nativeRunFinalizations.runId, runId)); + return fixture; + } + + it("restarts a claimed remote bootstrap whose durable checkpoint proves no provider work", async () => { + const fixture = await seedClaimedRemoteBootstrap("PROVEN"); + const dispositions = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + expect(dispositions).toEqual([expect.objectContaining({ kind: "bootstrap_incomplete", runId: fixture.runId, + providerAttempt: 1, controllerGeneration: 2 })]); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator).toMatchObject({ phase: "observed", recoveryState: "bootstrap_incomplete", failureCode: null }); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ disposition: "bootstrap_incomplete", + reason: "remote_bootstrap_proven_incomplete" }); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run).toMatchObject({ status: "running", errorCode: null, processPid: null }); + expect(await fixture.db.select().from(issueRecoveryActions) + .where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(0); + }); + + it("lets the executor prove the remote bootstrap from its Postgres checkpoint, not the local marker", async () => { + const fixture = await seedClaimedRemoteBootstrap("EXECUTOR"); + const execution = buildNativeExecutionInput({ + companyId, agentId, runId: fixture.runId, + issue: { id: fixture.issueId, identifier: null, title: "Restart bootstrap", description: null, workMode: "standard" }, + taskPrompt: "Restart bootstrap", provider: "codex", + workspace: { id: randomUUID(), cwd: runtimeRoot, repoUrl: null, repoRef: null, branchName: null }, + normalizedSessionId: fixture.native.normalizedSessionId, + runtimeContext: nativeRuntimeContextFixture(), + completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1", + contract: { revision: "1", objective: "Restart bootstrap", criteria: [{ id: "result", requirement: "Complete work" }] } }, + }); + const root = resolve(runtimeRoot, `marker-${fixture.runId}`); + await mkdir(resolve(root, "control-plane"), { recursive: true, mode: 0o700 }); + const writeMarker = (runId: string) => writeFile(resolve(root, "control-plane", "control-plane-state.json"), JSON.stringify({ + schema: "paperclip.runner.external-control-plane-state.v1", + identity: { runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.runId, runId, + normalizedSessionId: fixture.native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` }, + })); + const claim = { kind: "bootstrap_incomplete" as const, runId: fixture.runId, leaseOwner: "successor", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard" as const, recoveryRequestId: null }; + const target = { kind: "remote", transport: "sandbox", remoteCwd: "/workspace" } as AdapterExecutionTarget; + try { + await writeMarker(fixture.runId); + expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false); + // Only a bootstrap recovery on a classic sandbox restores the checkpoint. + for (const other of [{ restartRecovery: undefined, runnerExecutionTarget: target }, + { restartRecovery: claim, runnerExecutionTarget: { ...target, runnerTask: vi.fn() } as AdapterExecutionTarget }]) { + await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root, ...other }); + expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false); + } + // A marker for another run never receives this run's checkpoint. + await writeMarker(randomUUID()); + await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root, + restartRecovery: claim, runnerExecutionTarget: target }); + expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false); + await writeMarker(fixture.runId); + await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root, + restartRecovery: claim, runnerExecutionTarget: target }); + expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(true); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + it.each<[string, Parameters[1], string]>([ + ["a turn was issued", { mutate: core => { core.queueCommand("turn.start", { prompt: "Prior work must stay held." }); } }, + "ambiguous_remote_recovery_evidence"], + ["an event was acknowledged", { mutate: core => { core.store.state.ackedSourceSeq = 1; } }, + "ambiguous_remote_recovery_evidence"], + ["the state was adopted", { mutate: core => { core.store.state.bootstrapHistory!.adopted = true; } }, + "ambiguous_remote_recovery_evidence"], + ["no checkpoint exists", { checkpoint: false }, "ambiguous_remote_recovery_evidence"], + ["the bootstrap budget is spent", { attempt: 3 }, "execution_recovery_budget_exhausted"], + ])("keeps a claimed remote bootstrap held when %s", async (label, options, reason) => { + const fixture = await seedClaimedRemoteBootstrap(label.replace(/\W+/g, "-"), options); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] })).toEqual([{ kind: "blocked", runId: fixture.runId, reason }]); + }); + it.each([ { attempt: 2, driverKind: "codex_app_server" }, { attempt: 3, driverKind: "codex_app_server" }, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 5d2321c14e6..c7fddc8388b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -85,6 +85,8 @@ import { createNativeToolTrace, type NativeToolTrace } from "./native-tool-trace import { createNativeProviderFailureObservation } from "./native-provider-failure.js"; import { nativeStartupRetryExhaustedAction, nativeStartupRetryWait, planNativeStartupRetry } from "./native-startup-retry.js"; import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js"; +import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap, + runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js"; import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js"; import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js"; import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js"; @@ -522,10 +524,6 @@ const ACPX_RUNTIME_CHECKPOINT_EXCLUDED_ENTRIES = [ "home/.npm", "data/pnpm/store", ] as const; -const RUNNERD_CONTROL_PLANE_STATE_SCHEMAS = new Set([ - "paperclip.runner.durable.control-plane-state.v1", - "paperclip.runner.durable.control-plane-state.v2", -]); const RUNNERD_STATE_SCHEMA = "paperclip.runner.durable.state.v1"; const CODEX_PROVIDER_STATE_SCHEMA = "paperclip.runner.codex-provider-state.v1"; const ACPX_PROVIDER_STATE_SCHEMA = "paperclip.runner.acpx-provider-state.v3"; @@ -5486,6 +5484,7 @@ async function migrateRunnerdStateRootForExecution(input: { if (input.restartRecovery?.kind === "reattach_remote_runner" && !existsSync(scoped)) { throw new Error("runner_state_identity_mismatch: remote_reattach_root_missing"); } + await restoreRemoteBootstrapCheckpointForRecovery({ ...input, root: scoped }); if (existsSync(scoped)) { if (!isSafeNativeStateDirectory(scoped)) { throw new Error("runner_state_directory_unsafe"); @@ -5952,55 +5951,6 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean { } } -const RUNNERD_BOOTSTRAP_COMMAND_TYPES = new Set(["run.prepare", "session.open"]); - -/** The runner never connected, so every bootstrap command is still undelivered. */ -function runnerdStateIsUnconnectedBootstrap(state: Record): boolean { - const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; - const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : []; - return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) && - state.connectionCount === 0 && committedEvents.length === 0 && - commands.every(command => command.status === "pending" && - RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type))); -} - -/** - * A provider turn can start only after session.open completes and the - * controller issues a turn command. A root whose controller journal stops - * before that point (run.prepare may be acknowledged, session.open never is, - * no event was committed or acknowledged) and whose runner journal emitted no - * event either holds no provider work, so restart recovery may quarantine it - * and bootstrap again. Any other shape stays fail-closed. - */ -function runnerdStateIsIncompleteBootstrap( - state: Record, - runnerJournal?: Record | null, -): boolean { - if (runnerdStateIsUnconnectedBootstrap(state)) return true; - if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false; - if (typeof state.connectionCount !== "number" || !Number.isSafeInteger(state.connectionCount) || - state.connectionCount < 0) return false; - if (!Array.isArray(state.committedEvents) || state.committedEvents.length !== 0) return false; - if (state.ackedSourceSeq !== 0) return false; - if (state.warmTransition != null || state.completedWarmTransition != null || - state.stagedRunAttachment != null) return false; - if (record(state.bootstrapHistory).adopted === true) return false; - if (!Array.isArray(state.commands)) return false; - const commands = state.commands.map(record); - if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) && - (command.status === "pending" || - (command.type === "run.prepare" && command.status === "completed")))) return false; - // The controller cannot see runner events that were emitted but never - // committed. When the runner journal exists it must agree that none were. - if (runnerJournal === undefined) return true; - if (runnerJournal === null) return false; - const processed = Object.values(record(runnerJournal.processedCommands)).map(record); - return Array.isArray(runnerJournal.outbox) && runnerJournal.outbox.length === 0 && - runnerJournal.nextSourceSeq === 1 && runnerJournal.ackedSourceSeq === 0 && - processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) && - (command.commandType === "run.prepare" || command.status !== "completed")); -} - /** `undefined` when the runner never journaled, `null` when unreadable. */ function readRunnerdBootstrapJournal(root: string): Record | null | undefined { const runnerRoot = resolve(root, "runner"); @@ -6029,6 +5979,26 @@ function readRunnerdControlPlaneState(root: string): Record { ); } +/** + * A remote runner's controller state lives in Postgres; its local root holds + * only a fail-closed marker. Before a `bootstrap_incomplete` recovery checks + * that root, restore the exact checkpoint so the same incomplete-bootstrap + * proof applies. A missing checkpoint leaves the marker, which stays fail-closed. + */ +export async function restoreRemoteBootstrapCheckpointForRecovery(input: { + db: Db; execution: NativeExecutionInput; root: string; + restartRecovery?: NativeRestartRecoveryClaim; + runnerExecutionTarget?: AdapterExecutionTarget | null; +}): Promise { + const target = input.runnerExecutionTarget; + if (input.restartRecovery?.kind !== "bootstrap_incomplete" || target?.kind !== "remote" + || target.transport !== "sandbox" || target.runnerTask || !existsSync(input.root) + || readRunnerdDurableIdentity(input.root) !== null) return; + const marker = readRunnerdDurableIdentity(input.root, true); + if (!marker || marker.runId !== input.execution.binding.runId) return; + await restoreNativeRunnerCheckpoint(input.db, input.execution.binding, input.root, nativeSessionKey(input.execution)); +} + export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { try { return runnerdStateIsIncompleteBootstrap(readRunnerdControlPlaneState(root), readRunnerdBootstrapJournal(root)); diff --git a/server/src/services/native-runtime/native-session-resumption-admission.ts b/server/src/services/native-runtime/native-session-resumption-admission.ts index 6d27fff6f82..e768328c05a 100644 --- a/server/src/services/native-runtime/native-session-resumption-admission.ts +++ b/server/src/services/native-runtime/native-session-resumption-admission.ts @@ -8,6 +8,7 @@ import { getNativeReviewAssignment } from "./native-review-participant.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { notifyStatusCardIssueChange } from "../status-card-notifications.js"; import { normalizeAgentNameKey } from "../agent-name-key.js"; +import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js"; export async function restoreNativeSessionExecution(db: Db, input: { issue: Pick; @@ -48,7 +49,7 @@ export async function readNativeSessionResumptionAgent(db: Db, input: { const { issue, run, coordinator } = input; if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId || coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id || - !["todo", "in_progress", "in_review"].includes(issue.status) || + !(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status) || (issue.executionRunId && issue.executionRunId !== run.id) || (issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null; const [agent] = await db.select().from(agents).where(and( diff --git a/server/src/services/native-runtime/native-session-resumption-statuses.ts b/server/src/services/native-runtime/native-session-resumption-statuses.ts new file mode 100644 index 00000000000..9c19a2a9dc6 --- /dev/null +++ b/server/src/services/native-runtime/native-session-resumption-statuses.ts @@ -0,0 +1,4 @@ +/** Task statuses from which a native same-run resume may be claimed. The task + * interruption read model shares this set with resumption admission so it + * never advertises a resume the scheduler would refuse. */ +export const NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES = ["todo", "in_progress", "in_review"] as const; diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index c24db5150b4..73c7682f9be 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -216,10 +216,15 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i or ${heartbeatRuns.errorCode} = 'workspace_validation_failed' then ${heartbeatRuns.error} else null end`, errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt, scheduledRetryReason: heartbeatRuns.scheduledRetryReason, + // Candidate only: the coordinator also pushes nextAttemptAt forward while + // it defers an ineligible run. The scheduler's own admission check below + // decides whether this resume is real. nativeResumeAt: sql`(select ${nativeRunFinalizations.nextAttemptAt} from ${nativeRunFinalizations} where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed' - and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure')`.mapWith(value => value == null ? null : new Date(value)), + and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure' + and ${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null + and ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null)`.mapWith(value => value == null ? null : new Date(value)), // The run whose resultJson supplied providerFailure. A scheduled retry // shows its predecessor's reason, so redaction must use that run's registry. providerFailureRunId: sql`case when ${heartbeatRuns.resultJson}->'providerFailure' is null @@ -279,6 +284,38 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i inArray(approvals.status, ["pending", "revision_requested"]))), db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)), ]); + // Advertise a pending resume only when the scheduler would admit it, using + // its own check, so the task never hides Retry behind a deferred candidate + // and never offers Retry beside a resume that will run. Rare rows only. + // Candidates are failed native runs inside an active coordinator retry, so + // there are few. Their rows load in three batched reads; admission itself + // runs a few bounded reads per candidate, four at a time. + const resumeCandidates = runs.filter(run => run.nativeResumeAt); + if (resumeCandidates.length) { + const { readNativeSessionResumptionAgent } = await import("./native-runtime/native-session-resumption-admission.js"); + const candidateIds = resumeCandidates.map(run => run.id); + const [fullRuns, coordinators] = await Promise.all([ + db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, candidateIds))), + db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId), + inArray(nativeRunFinalizations.runId, candidateIds))), + ]); + const issueIds = [...new Set(fullRuns.flatMap(run => run.nativeIssueId ? [run.nativeIssueId] : []))]; + const issueRows = issueIds.length ? await db.select().from(issues).where(and(eq(issues.companyId, companyId), + inArray(issues.id, issueIds))) : []; + const runById = new Map(fullRuns.map(run => [run.id, run])); + const coordinatorByRun = new Map(coordinators.map(row => [row.runId, row])); + const issueById = new Map(issueRows.map(row => [row.id, row])); + for (let start = 0; start < resumeCandidates.length; start += 4) { + await Promise.all(resumeCandidates.slice(start, start + 4).map(async candidate => { + const run = runById.get(candidate.id); + const issue = run?.nativeIssueId ? issueById.get(run.nativeIssueId) : undefined; + const coordinator = coordinatorByRun.get(candidate.id); + const admitted = run && issue && coordinator + && await readNativeSessionResumptionAgent(db, { issue, run, coordinator }).catch(() => null); + if (!admitted) candidate.nativeResumeAt = null; + })); + } + } const runIds = runs.map(run => run.id); const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids), diff --git a/tests/runner-e2e/reliability-testops-registration.mjs b/tests/runner-e2e/reliability-testops-registration.mjs index c9da1096cec..42d1c9706a2 100644 --- a/tests/runner-e2e/reliability-testops-registration.mjs +++ b/tests/runner-e2e/reliability-testops-registration.mjs @@ -94,8 +94,7 @@ export async function inspectReliability(api) { if (candidates.length !== 1 || !positive(candidates[0].job.id)) fail('standing_job'); const job = await api(`/api/job/${candidates[0].job.id}`); if (job?.id !== candidates[0].job.id || job.type !== 'github' || !workflowUrls.has(job.url)) fail('standing_job'); - const assignments = plan ? await api(`/api/testplan/${plan.id}/job`) : []; - if (!Array.isArray(assignments)) fail('plan_job'); + const assignments = plan ? jobAssignments(await api(`/api/testplan/${plan.id}/job`)) : []; return { caseIds: Object.fromEntries(found), plan, jobId: job.id, standingParameters: Array.isArray(candidates[0].job.parameters) ? candidates[0].job.parameters : [], assignments }; } @@ -134,8 +133,8 @@ export async function registerReliability(api) { plan = await api(`/api/testplan/${plan.id}`); } assertPlan(plan, ids); - let assignments = await api(`/api/testplan/${plan.id}/job`); - if (!Array.isArray(assignments) || assignments.length > 1 || assignments.some(row => row?.job?.id !== before.jobId)) fail('plan_job_conflict'); + let assignments = jobAssignments(await api(`/api/testplan/${plan.id}/job`)); + if (assignments.length > 1 || assignments.some(row => row?.job?.id !== before.jobId)) fail('plan_job_conflict'); if (!assignments.length) { await api(`/api/testplan/${plan.id}/assign`, 'POST', { jobId: before.jobId, selection: { inverted: false, manual: false, leafsInclude: ids } }); created.assignment = true; @@ -145,13 +144,21 @@ export async function registerReliability(api) { await api(`/api/testplan/${plan.id}/jobparameter`, 'POST', { jobId: before.jobId, parameters: before.standingParameters }); created.parameters = true; } - assignments = await api(`/api/testplan/${plan.id}/job`); - if (!Array.isArray(assignments) || assignments.length !== 1 || assignments[0]?.job?.id !== before.jobId + assignments = jobAssignments(await api(`/api/testplan/${plan.id}/job`)); + if (assignments.length !== 1 || assignments[0]?.job?.id !== before.jobId || assignments[0].testCasesCount !== ids.length || (before.standingParameters.length && !same(assignments[0].job.parameters ?? [], before.standingParameters))) fail('plan_job'); return { caseIds, plan, jobId: before.jobId, created }; } +// TestOps lists a plan's cases that no job covers as a job-less row, e.g. +// `[{ testCasesCount: 7 }]` for a freshly created plan. That row is not a job +// assignment; only rows naming a job are. +function jobAssignments(rows) { + if (!Array.isArray(rows)) fail('plan_job'); + return rows.filter(row => row?.job != null); +} + export function registrationReceipt(operation, state, env) { const caseIds = Object.fromEntries(reliabilityDefinitions.filter(row => state.caseIds[row.name]).map(row => [row.name, state.caseIds[row.name]])); const complete = Object.keys(caseIds).length === reliabilityDefinitions.length && positive(state.plan?.id) diff --git a/tests/runner-e2e/reliability-testops.test.mjs b/tests/runner-e2e/reliability-testops.test.mjs index c10727bd54c..4cda542b8c3 100644 --- a/tests/runner-e2e/reliability-testops.test.mjs +++ b/tests/runner-e2e/reliability-testops.test.mjs @@ -196,6 +196,8 @@ function fakeTestOps({ existing = [], plan = null, assignments = [], standing = if (method === 'GET' && url.pathname === '/api/testplan') return { content: state.plan ? [{ id: state.plan.id, name: state.plan.name }] : [], totalPages: 1 }; if (method === 'POST' && url.pathname === '/api/testplan') { state.plan = { id: 77, projectId: 1, name: body.name, treeSelection: body.treeSelection, testCasesCount: body.treeSelection.leafsInclude.length }; + // Live TestOps lists a new plan's uncovered cases as one job-less row. + state.assignments = [{ testCasesCount: body.treeSelection.leafsInclude.length }]; return { id: 77 }; } if (method === 'GET' && url.pathname === '/api/testplan/77') return state.plan; @@ -293,6 +295,20 @@ test('registration refuses a standing plan without exactly one trusted job or a await assert.rejects(registerReliability(api), /plan_job_conflict/); }); +test('a plan TestOps reports only as a job-less row is unassigned, so registration assigns the standing job', async () => { + const { api, state } = fakeTestOps(); + await registerReliability(api); + // Reproduces staging plan 135: created, then the run stopped before assignment. + state.assignments = [{ testCasesCount: 7 }]; + const inspected = await inspectReliability(api); + assert.deepEqual(inspected.assignments, []); + assert.equal(registrationReceipt('inspect', inspected, env).complete, false); + const repaired = await registerReliability(api); + assert.equal(repaired.created.assignment, true); + assert.deepEqual(state.assignments.map(row => row.job?.id), [50]); + assert.equal(registrationReceipt('register', repaired, { GITHUB_RUN_ID: '1', GITHUB_SHA: workflowSha }).complete, true); +}); + test('the workflows expose the reliability plan without new provider or reporting authority', async () => { const full = parse(await readFile(new URL('../../.github/workflows/runner-full-stack-e2e.yml', import.meta.url), 'utf8')); assert.deepEqual(full.on.workflow_dispatch.inputs.testops_plan.options, ['product-regression', 'reliability']); diff --git a/tests/runner-e2e/workflow-target.test.ts b/tests/runner-e2e/workflow-target.test.ts index a4e1c57998b..d02c01d647b 100644 --- a/tests/runner-e2e/workflow-target.test.ts +++ b/tests/runner-e2e/workflow-target.test.ts @@ -16,6 +16,7 @@ async function resolveTarget(event: string, cadence: string, branch = "master") const output = path.join(directory, "output"); const result = spawnSync("bash", ["-c", `gh() { printf '%s\\n' "$MOCK_BRANCH_SHA"; }\n${script}`], { encoding: "utf8", env: { ...process.env, EVENT_NAME: event, EVAL_CADENCE: cadence, + IMAGE_ONLY: "false", TESTOPS_REQUESTED: "false", TARGET_BRANCH: branch, DEFAULT_BRANCH: "master", GITHUB_SHA: queuedSha, MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip" }, });