diff --git a/server/src/services/native-runtime/native-controller-migration.integration.test.ts b/server/src/services/native-runtime/native-controller-migration.integration.test.ts index a58ee298adb..69005e82bb3 100644 --- a/server/src/services/native-runtime/native-controller-migration.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-migration.integration.test.ts @@ -1,15 +1,17 @@ import { createHash, randomUUID } from "node:crypto"; import { mkdir, open, readFile, rm, stat, writeFile } from "node:fs/promises"; import { dirname, join } from "node:path"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { afterEach, expect, it, vi } from "vitest"; -import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { agents, agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; import { captureLegacyDurablePrpCheckpoint, DurablePrpControlPlane, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; import { nativeToolContractFingerprintForTarget } from "./native-session-resume.js"; +import * as sessionResume from "./native-session-resume.js"; import { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; import { hasLocalNativeControllerObligations, migrateNativeControllers, prestageNativeControllers } from "./native-controller-migration.js"; -import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; +import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; +import { currentNativeControllerIdentity } from "./native-restart-recovery.js"; import { createNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; import { canonicalProviderPackJson } from "./remote-provider-pack-verification.js"; @@ -23,7 +25,7 @@ import { legacyNativeControllerArchive } from "./native-controller-archive.js"; import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archive_online_changed", "archive_online_commit", "archive_online_active", "archive_online_foreign", - "online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", "durable_inline_failed", + "active_warm", "source_scan", "active_durable", "online_success", "online_owned", "online_same_source", "online_busy", "online_changed", "online_evidence", "import", "archive", "success", "batch", "durable_failed", "durable_inline_failed", "stale_confirmed", "stale_foreign", "stale_changed", "stale_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable", "stale_durable_absent", "stale_durable_runner", "stale_durable_session", "stale_durable_snapshot", "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable", @@ -32,11 +34,11 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv "controller_observed_handoff", "controller_observed_handoff_stopped", "controller_observed_handoff_broken", "controller_observed_handoff_fork"])("offline native controller migration (%s)", scenario => { const environment = nativeRestartFixture(); afterEach(() => { vi.unstubAllEnvs(); vi.restoreAllMocks(); }); - async function fixture(commandCount = 1, warm = false) { + async function fixture(commandCount = 1, warm = false, modernWorkspace = false) { vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "73".repeat(32)); const f = await environment.seedRun(randomUUID()); const { companyId, agentId, paperclipHome, runtimeRoot } = environment.current; - const binding = { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: f.native.environmentLeaseId }; + const binding = { companyId, agentId, issueId: f.issueId, runId: f.runId, executionWorkspaceId: modernWorkspace ? randomUUID() : f.native.environmentLeaseId }; const execution = parseNativeExecutionInput({ schema: "paperclip.native-execution-input.v1", provider: { kind: "codex", model: null }, binding, task: { identifier: "MIG-1", title: "Retained work", description: null, prompt: "Continue the work.", workMode: "standard" }, @@ -50,7 +52,7 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv const fingerprint = nativeToolContractFingerprintForTarget("remote", "codex"); const paths = nativeControllerMigrationDescriptor(execution, fingerprint); const identity = { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId, - environmentLeaseId: f.native.environmentLeaseId, turnId: f.native.turnId, itemId: f.native.itemId }; + environmentLeaseId: binding.executionWorkspaceId, turnId: f.native.turnId, itemId: f.native.itemId }; const core = new DurablePrpControlPlane({ stateDirectory: join(paths.root, "control-plane"), identity, expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); core.issueBootstrapTicket(); @@ -91,6 +93,200 @@ describeEmbeddedPostgres.each(["archive_online", "archive_online_owned", "archiv taskKey: f.issueId, lastRunId: f.runId }).returning(); return { ...f, companyId, agentId, paperclipHome, runtimeRoot, paths, file, source, profile, session, queuedId, identity, snapshot }; } + if (scenario === "active_warm") it("preserves an active warm rotation only with its exact older committed authority", async () => { + const f = await fixture(0, true, true), controller = await currentNativeControllerIdentity(); + const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const owner = { controllerBootId: controller.bootId, controllerGeneration: 1, leaseOwner: "warm-checkpoint-writer", + leaseExpiresAt: new Date(Date.now() + 60_000), phase: "observed", resultId: null, failureCode: null, recoveryState: null } as const; + await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId)); + const core = new DurablePrpControlPlane({ stateDirectory: join(f.paths.root, "control-plane"), identity: f.identity, + expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + let writer = nativeRunnerCheckpointWriter(f.db, { ...binding, bootId: controller.bootId, leaseOwner: owner.leaseOwner, + generation: 1, authorityRunId: f.runId }); + core.setCheckpointPersistence(value => writer(value)); + try { + await core.flushCheckpoint(); + const marker = await readFile(f.file, "utf8"), [prior] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [result] = await f.db.insert(nativeRunResults).values({ ...binding, completionContractId: prior.completionContractId!, + serverFingerprint: "warm-source-result", schemaStatus: "accepted", resultJson: {}, canonicalSha256: nativeSha256({}) }).returning(); + const finishedAt = new Date(); + await f.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, + leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + const next = { ...f.identity, runId: randomUUID(), turnId: randomUUID(), itemId: randomUUID() }; + const execution = parseNativeExecutionInput({ ...f.profile.nativeExecutionInput, + binding: { ...f.profile.nativeExecutionInput.binding, runId: next.runId } }); + const profile = { ...f.profile, nativeExecutionInput: execution, sessionCheckpoint: null }; + expect(nativeControllerMigrationDescriptor(execution, f.profile.nativeToolContractFingerprint).root).toBe(f.paths.root); + await f.db.insert(heartbeatRuns).values({ id: next.runId, companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, + nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId, + runtimeMode: "native", status: "running", nativePhase: "observed", processPid: process.pid, processStartedAt: new Date(), + createdAt: new Date(finishedAt.getTime() + 1), startedAt: new Date(finishedAt.getTime() + 2), runnerProfileJson: profile, + completionContractId: prior.completionContractId }); + await f.db.insert(nativeRunFinalizations).values({ ...owner, ...binding, runId: next.runId }); + await f.db.update(issues).set({ executionRunId: next.runId, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + await f.db.update(agentTaskSessions).set({ lastRunId: next.runId }).where(eq(agentTaskSessions.id, f.session.id)); + writer = nativeRunnerCheckpointWriter(f.db, { ...binding, runId: next.runId, authorityRunId: next.runId, + bootId: controller.bootId, leaseOwner: owner.leaseOwner, generation: 1 }); + core.rotateRunIdentity(next); + await core.flushCheckpoint(); + expect(await readFile(f.file, "utf8")).toBe(marker); + expect((await loadNativeRunnerCheckpoint(f.db, { ...binding, runId: next.runId }))?.runId).toBe(next.runId); + const oldRows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + const currentRows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, next.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + for (const value of [{ ...f.identity, runId: randomUUID() }, { ...f.identity, runnerInstanceId: randomUUID() }, + { ...f.identity, normalizedSessionId: randomUUID() }, { ...f.identity, environmentLeaseId: randomUUID() }, + { ...f.identity, turnId: randomUUID() }]) { + await writeFile(f.file, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: value })); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + } + await writeFile(f.file, marker); + await f.db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.runId, f.runId), eq(nativeRunnerCheckpoints.part, "cursor"))); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + await f.db.insert(nativeRunnerCheckpoints).values(oldRows); + await f.db.delete(nativeRunnerCheckpoints).where(and(eq(nativeRunnerCheckpoints.runId, next.runId), eq(nativeRunnerCheckpoints.part, "cursor"))); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, next.runId)); + await f.db.insert(nativeRunnerCheckpoints).values(currentRows); + for (const patch of [{ status: "failed" }, { createdAt: new Date(Date.now() + 60_000) }, + { finishedAt: new Date(Date.now() + 60_000) }, { runnerInstanceId: randomUUID() }]) { + await f.db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(heartbeatRuns).set({ status: "succeeded", createdAt: prior.createdAt, finishedAt, + runnerInstanceId: prior.runnerInstanceId }).where(eq(heartbeatRuns.id, f.runId)); + } + await f.db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, result.id)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(nativeRunResults).set({ schemaStatus: "accepted" }).where(eq(nativeRunResults.id, result.id)); + const [foreign] = await f.db.insert(agents).values({ companyId: f.companyId, name: "Foreign owner", role: "engineer", adapterType: "paperclip_runner" }).returning(); + await f.db.update(heartbeatRuns).set({ agentId: foreign.id }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(heartbeatRuns).set({ agentId: f.agentId }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, nativeExecutionInput: { + ...execution, session: { ...execution.session, lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } } } } }) + .where(eq(heartbeatRuns.id, next.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, next.runId)); + await f.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, next.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, next.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + expect(await readFile(f.file, "utf8")).toBe(marker); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(oldRows); + } finally { await core.stop(); } + }); + if (scenario === "source_scan") it("bounds retained source readers at eight and joins every started reader before a failure", async () => { + const f = await fixture(), binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, f.identity, + { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }); + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, binding, checkpoint)); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...f.profile, + nativeControllerRecovery: createNativeControllerRecovery(f.snapshot, { configDigest: f.paths.configDigest, + processConfigDigest: null }, { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, + normalizedSessionId: f.native.normalizedSessionId }) } }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.insert(agentTaskSessions).values(Array.from({ length: 15 }, () => ({ companyId: f.companyId, + agentId: f.agentId, adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId }))); + const original = sessionResume.findNativeSessionResumeRun; + let started = 0, finished = 0, inFlight = 0, peak = 0, fail = false; + const reason = new Error("retained source reader failed"); + const spy = vi.spyOn(sessionResume, "findNativeSessionResumeRun").mockImplementation(async (...args) => { + const first = ++started === 1; peak = Math.max(peak, ++inFlight); + try { + await new Promise(resolve => setTimeout(resolve, 20)); + if (fail && first) throw reason; + return await original(...args); + } finally { inFlight--; finished++; } + }); + try { + const before = performance.now(); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + console.info(JSON.stringify({ state: "retained_source_readers", elapsedMs: performance.now() - before, peak, started, finished })); + expect(started).toBe(16); expect(finished).toBe(16); expect(peak).toBeGreaterThan(1); expect(peak).toBeLessThanOrEqual(8); + started = finished = peak = 0; fail = true; + await expect(hasLocalNativeControllerObligations(f.db)).rejects.toBe(reason); + expect(started).toBe(8); expect(finished).toBe(8); expect(inFlight).toBe(0); + expect(await readFile(f.file, "utf8")).toBe(f.source); + } finally { spy.mockRestore(); } + }); + if (scenario === "active_durable") it("accepts a complete active database authority without terminal metadata and refuses stale or incomplete handoff proof", async () => { + const f = await fixture(1, false, true), controller = await currentNativeControllerIdentity(); + const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const { nativeWorkspaceSync, ...baseProfile } = f.profile; + const remoteWorkspace = { schema: "paperclip.remote-workspace-run.v1", workspaceId: nativeWorkspaceSync.workspaceId, + leaseId: nativeWorkspaceSync.leaseId, providerLeaseId: nativeWorkspaceSync.providerLeaseId, + remoteCwd: nativeWorkspaceSync.remoteCwd, epoch: 1, state: "prepared", snapshotId: null }; + const profile = { ...baseProfile, sessionCheckpoint: null, remoteWorkspace }; + await f.db.update(heartbeatRuns).set({ status: "running", finishedAt: null, nativePhase: "observed", + processPid: process.pid, processStartedAt: new Date(), runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + const owner = { controllerBootId: controller.bootId, controllerGeneration: 1, leaseOwner: "live-checkpoint-writer", + leaseExpiresAt: new Date(Date.now() + 60_000), phase: "observed", resultId: null, failureCode: null, recoveryState: null } as const; + await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(issues).set({ executionRunId: f.runId, checkoutRunId: null }).where(eq(issues.id, f.issueId)); + const core = new DurablePrpControlPlane({ stateDirectory: join(f.paths.root, "control-plane"), identity: f.identity, + expectedRunnerVersion: "migration-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(f.db, { ...binding, bootId: controller.bootId, + leaseOwner: owner.leaseOwner, generation: 1, authorityRunId: f.runId })); + try { + await core.flushCheckpoint(); + const marker = await readFile(f.file, "utf8"); + const rows = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + expect(rows.length).toBeGreaterThan(2); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + for (const value of [f.source, JSON.stringify({ schema: "unknown", identity: f.identity }), + JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: { ...f.identity, runId: randomUUID() } })]) { + await writeFile(f.file, value); expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + } + await writeFile(f.file, marker); + for (const part of ["cursor", "command:1"]) { + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.part, part)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + const savedProfile = { ...profile, sessionCheckpoint: f.snapshot, + nativeControllerRecovery: createNativeControllerRecovery(f.snapshot, { configDigest: f.paths.configDigest, + processConfigDigest: null }, { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, + normalizedSessionId: f.native.normalizedSessionId }) }; + await f.db.update(heartbeatRuns).set({ runnerProfileJson: savedProfile }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + await f.db.insert(nativeRunnerCheckpoints).values(rows); + await f.db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing", leaseOwner: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId)); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + } + const cipher = createLocalEncryptedCipher("write"), header = JSON.parse((await loadNativeRunnerCheckpoint(f.db, binding))!.records.get("header")!); + for (const field of ["runId", "runnerInstanceId", "normalizedSessionId", "environmentLeaseId"]) { + const contents = JSON.stringify({ ...header, state: { ...header.state, identity: { ...header.state.identity, [field]: randomUUID() } } }); + await f.db.update(nativeRunnerCheckpoints).set({ material: cipher.encrypt(JSON.stringify({ ...binding, part: "header", contents })) }) + .where(eq(nativeRunnerCheckpoints.part, "header")); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + } + await f.db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + await f.db.insert(nativeRunnerCheckpoints).values(rows); + for (const patch of [{ leaseExpiresAt: new Date(0) }, { controllerGeneration: 0 }, { controllerBootId: null }, + { leaseOwner: null }, { failureCode: "uncertain" }, { phase: "retryable_failure" }]) { + await f.db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(nativeRunFinalizations).set(owner).where(eq(nativeRunFinalizations.runId, f.runId)); + } + await f.db.update(issues).set({ executionRunId: f.queuedId }).where(eq(issues.id, f.issueId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + for (const value of [null, {}, { ...remoteWorkspace, workspaceId: randomUUID() }]) { + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...profile, remoteWorkspace: value, + nativeWorkspaceSync } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(true); + } + await f.db.update(heartbeatRuns).set({ runnerProfileJson: { ...baseProfile, sessionCheckpoint: null, nativeWorkspaceSync } }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await hasLocalNativeControllerObligations(f.db)).toBe(false); + await f.db.update(heartbeatRuns).set({ runnerProfileJson: profile }).where(eq(heartbeatRuns.id, f.runId)); + expect(await readFile(f.file, "utf8")).toBe(marker); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(rows); + } finally { await core.stop(); } + }); if (scenario.startsWith("archive_online")) it("preserves a failed source online without changing its execution or unrelated work", async () => { const f = await fixture(); await f.db.update(issues).set({ executionRunId: null, checkoutRunId: null }).where(eq(issues.id, f.issueId)); diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index c4997488c63..4c74b2e3608 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -3,9 +3,10 @@ import { lstat, open, readdir } from "node:fs/promises"; import { join } from "node:path"; import { and, eq, inArray, or, sql } from "drizzle-orm"; import { heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; import { captureLegacyDurablePrpCheckpoint, inspectDurablePrpCheckpointHeader, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; -import { findNativeSessionResumeRun } from "./native-session-resume.js"; +import { findNativeSessionResumeRun, isNativeSessionId } from "./native-session-resume.js"; import { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; import { importNativeControllerRecovery, readNativeControllerRecovery } from "./native-controller-recovery.js"; @@ -76,7 +77,7 @@ async function hasConfirmedOlderController(db: Db, selected: Selected, } catch { return false; } } -async function* retainedSources(db: Db, signal?: AbortSignal) { +async function* retainedSources(db: Db, signal?: AbortSignal, concurrency = 1) { let after: string | undefined; for (;;) { signal?.throwIfAborted(); @@ -86,12 +87,18 @@ async function* retainedSources(db: Db, signal?: AbortSignal) { and r.company_id = s.company_id and r.agent_id = s.agent_id where s.adapter_type = 'paperclip_runner' and r.native_issue_id is not null and r.native_session_id is not null ${after ? sql`and s.id > ${after}::uuid` : sql``} order by s.id limit 100`); - for (const selected of rows) { + for (let offset = 0; offset < rows.length; offset += concurrency) { signal?.throwIfAborted(); - const source = await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId, - issueId: selected.issueId, normalizedSessionId: selected.sessionId, - beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true }); - if (source) yield { selected, source }; + const batch = await Promise.allSettled(rows.slice(offset, offset + concurrency).map(async selected => ({ + selected, source: await findNativeSessionResumeRun(db, { companyId: selected.companyId, agentId: selected.agentId, + issueId: selected.issueId, normalizedSessionId: selected.sessionId, + beforeCreatedAt: new Date(selected.createdAt), includeProgressBarrier: true }), + }))); + for (const result of batch) { + signal?.throwIfAborted(); + if (result.status === "rejected") throw result.reason; + if (result.value.source) yield { selected: result.value.selected, source: result.value.source }; + } } if (rows.length < 100) return; after = rows.at(-1)!.id; @@ -175,18 +182,89 @@ async function preservedDatabaseOnlyFailedCheckpoint(selected: Selected, return true; } +async function preservedWarmMarker(db: Db, selected: Selected, + source: NonNullable>>, markerIdentity: Record) { + if (!isNativeSessionId(markerIdentity.runId) || markerIdentity.runId === source.id) return false; + const [prior] = await db.execute<{ id: string }>(sql` + select r.id from heartbeat_runs r join heartbeat_runs current on current.id=${source.id}::uuid and current.company_id=r.company_id + join native_run_finalizations f on f.run_id=r.id and f.company_id=r.company_id and f.issue_id=r.native_issue_id + join native_run_results result on result.id=f.result_id and result.run_id=r.id and result.company_id=r.company_id + and result.issue_id=r.native_issue_id and result.completion_contract_id=r.completion_contract_id and result.schema_status='accepted' + where r.id=${markerIdentity.runId}::uuid and r.company_id=${selected.companyId}::uuid and r.agent_id=${selected.agentId}::uuid + and r.native_issue_id=${selected.issueId}::uuid and r.native_session_id=${selected.sessionId}::uuid + and r.runner_instance_id=${source.runnerInstanceId} and r.runtime_mode='native' and r.status='succeeded' + and r.created_at null); + const header = checkpoint && inspectDurablePrpCheckpointHeader(checkpoint.records.get("header") ?? ""); + return Boolean(header && nativeSha256(header) === nativeSha256(markerIdentity)); +} + +async function preservedActiveCheckpoint(db: Db, selected: Selected, + source: NonNullable>>, paths: Awaited>) { + if (source.status !== "running" || source.runtimeMode !== "native") return false; + let marker: Record, bytes: string | undefined; + try { + bytes = await readPrivateSource(paths.controlFile, 4096); + marker = record(JSON.parse(bytes ?? "null")); + } catch { return false; } + if (marker.schema !== "paperclip.runner.external-control-plane-state.v1" + || Object.keys(marker).sort().join(",") !== "identity,schema") return false; + const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id }; + const checkpoint = await loadNativeRunnerCheckpoint(db, binding).catch(() => null); + const header = checkpoint && inspectDurablePrpCheckpointHeader(checkpoint.records.get("header") ?? ""); + if (!header || header.runId !== source.id || header.runnerInstanceId !== source.runnerInstanceId + || header.normalizedSessionId !== selected.sessionId) return false; + const exact = nativeSha256(header) === nativeSha256(marker.identity), markerIdentity = record(marker.identity); + if (!exact && (paths.execution.session.lifecyclePolicy.mode !== "warm" + || ["runnerInstanceId", "normalizedSessionId", "environmentLeaseId"].some(field => markerIdentity[field] !== header[field as keyof typeof header]) + || !await preservedWarmMarker(db, selected, source, markerIdentity))) return false; + const [current] = await db.execute<{ workspace: unknown; remoteWorkspace: unknown; hasRemoteWorkspace: boolean; pluginLeaseId: string | null; liveOwner: boolean }>(sql` + select r.runner_profile_json->'nativeWorkspaceSync' as workspace, + r.runner_profile_json->'remoteWorkspace' as "remoteWorkspace", + r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId", + (r.native_phase='observed' and r.process_pid is not null and r.process_started_at is not null + and i.execution_run_id=r.id and i.assignee_agent_id=r.agent_id + and f.controller_boot_id is not null and f.controller_boot_id<>'' and f.controller_generation>0 + and f.lease_owner is not null and f.lease_owner<>'' and f.lease_expires_at>clock_timestamp() + and f.phase='observed' and f.result_id is null and f.failure_code is null) is true as "liveOwner" + from heartbeat_runs r left join native_run_finalizations f on f.run_id=r.id and f.company_id=r.company_id and f.issue_id=r.native_issue_id + left join issues i on i.id=r.native_issue_id and i.company_id=r.company_id + left join lateral (select l.id from environment_leases l where l.company_id=r.company_id and l.heartbeat_run_id=r.id + and l.issue_id=r.native_issue_id and l.status='active' and l.released_at is null and l.cleanup_status is null + and (l.expires_at is null or l.expires_at>clock_timestamp()) and l.metadata->>'agentId'=r.agent_id::text + and l.metadata->>'runnerTaskLease'='true' and l.metadata->>'runnerTaskDispatched'='true' + and coalesce(l.metadata->>'runnerTaskExecutionStopped','false')!='true' + and coalesce(l.metadata->>'runnerTaskRetiring','false')!='true' order by l.created_at desc limit 1) p on true + where r.id=${source.id}::uuid and r.company_id=${selected.companyId}::uuid and r.agent_id=${selected.agentId}::uuid + and r.native_issue_id=${selected.issueId}::uuid and r.native_session_id=${selected.sessionId}::uuid + and r.runner_instance_id=${source.runnerInstanceId} and r.runtime_mode='native' and r.status='running'`); + if (!current) return false; + const remote = remoteWorkspaceRunReferenceSchema.safeParse(current.remoteWorkspace); + const workspaceId = current.hasRemoteWorkspace + ? remote.success && !current.pluginLeaseId ? remote.data.workspaceId : null + : current.pluginLeaseId ?? readNativeWorkspaceSyncReference(current.workspace)?.workspaceId; + return workspaceId === header.environmentLeaseId + && (current.pluginLeaseId !== null || paths.execution.binding.executionWorkspaceId === workspaceId) + && (current.liveOwner || exact && hasMatchingControllerRecovery(selected, source, checkpoint)) + && bytes === await readPrivateSource(paths.controlFile, 4096); +} + export async function hasLocalNativeControllerObligations(db: Db, storage?: StorageService | null): Promise { const roots = await readdir(nativeHarnessStateBase()).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return [] as string[]; throw error; }); if (!roots.length) return false; - for await (const { selected, source } of retainedSources(db)) { + for await (const { selected, source } of retainedSources(db, undefined, 8)) { const binding = { companyId: selected.companyId, issueId: selected.issueId, runId: source.id }; - if (readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint) + if (source.status !== "running" + && readNativeControllerRecovery(source.runnerProfileJson.nativeControllerRecovery, source.runnerProfileJson.sessionCheckpoint) && await hasNativeRunnerCheckpoint(db, binding)) continue; if (source.status === "failed" && source.runnerProfileJson.sessionCheckpoint == null && preservedFailedCheckpoint(source, selected.sessionId, await loadNativeRunnerCheckpoint(db, binding))) continue; const paths = await localPaths(source); + if (await preservedActiveCheckpoint(db, selected, source, paths)) continue; const present = await lstat(paths.controlFile).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return undefined; throw error; });