diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index 2b1f6ddcbe7..098f1ecb34f 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -137,3 +137,36 @@ test resumes that same queued run after deleting the old disk, verifies saved workspace bytes and instructions at provider dispatch, and checks one result and one wake consumption even after a second queue sweep. Its sandbox transport and provider session are scripted; database, S3, admission and restoration are real. + +A failed follow-up can also have no warm supervisor and retain the preceding +successful run in its controller file. Archive this source only after the Runner +parser validates that older authority and scoped history confirms the same +Runner, session, provider thread and an earlier successful finish. Keep current +snapshots, ownership and typed checkpoints unchanged. The archive cannot become +executable recovery authority. Foreign, invalid and active sources remain blocked. + +For a finished failed source with a stale observed coordinator, evidence archival +also requires an expired controller lease, no accepted result or retry deadline, +and no current task ownership. Every remote lease must carry the exact host-owned +termination receipt, confirmed after execution started. Recheck those receipts +before publication. Preserve the coordinator as recorded; this proof permits +evidence storage, not execution or a claim that the coordinator has settled. + +Compare retained turn completion to the follow-up execution start, not its queue +creation time. A queued follow-up can predate the prior turn completion. A remote +stop can also predate a later logical failure caused by recovery timeout. These +proofs use the recorded execution start and never rewrite lifecycle timestamps. + +### Existing typed checkpoint authority + +When a retained source already has a valid typed database checkpoint, derive its +small recovery metadata from that authority and the current admitted input. Do +not consult a stale or absent local warm supervisor. Validate the same company, +agent, task, normalized session and Runner bindings and retain the checkpoint, +snapshot, owner and session pointer unchanged. Offline warm import without a +supervisor still requires this existing typed authority; normal runtime import +keeps its supervisor and ownership requirements. + +For a retired reusable sandbox lease, the existing remote-termination resolver +may prove destruction through its exact successor handoff chain. A stopped +successor, broken chain or competing tip does not grant archival authority. 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 4e489090f6b..ff5a26e692e 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 @@ -3,16 +3,18 @@ import { mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { dirname, join } from "node:path"; import { eq } from "drizzle-orm"; import { afterEach, expect, it, vi } from "vitest"; -import { agentTaskSessions, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { agentTaskSessions, environmentLeases, environments, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, 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 { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; import { hasLocalNativeControllerObligations, migrateNativeControllers } from "./native-controller-migration.js"; import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; -import { resolveNativeControllerRecovery } from "./native-controller-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"; +import { createReusableLeaseHandoff } from "../environment-lease-handoff.js"; +import { remoteTerminationReceipt } from "../remote-execution-termination.js"; import { nativeSha256 } from "./canonical.js"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import { createLocalDiskStorageProvider } from "../../storage/local-disk-provider.js"; @@ -21,7 +23,12 @@ import { legacyNativeControllerArchive } from "./native-controller-archive.js"; import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-storage.js"; describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable_failed", - "stale_confirmed", "stale_foreign", "stale_changed", "stale_success", "stale_busy", "stale_durable"])("offline native controller migration (%s)", scenario => { + "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", + "controller_observed_stopped", "controller_observed_stop_before_finish", "controller_observed_missing", "controller_observed_foreign", + "controller_observed_live", "controller_observed_result", "controller_observed_retry", "controller_observed_task", "controller_observed_old_receipt", "controller_observed_revoked", + "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()); async function fixture(commandCount = 1, warm = false) { @@ -68,7 +75,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable digest: `sha256:${createHash("sha256").update(canonicalProviderPackJson(payload)).digest("hex")}` }), runnerArtifact: () => ({ version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }) }); pack.verified(); - const snapshot = { backendKind: "mock", sessionId: f.native.normalizedSessionId, + const snapshot = { backendKind: "mock" as const, sessionId: f.native.normalizedSessionId, identity: { companyId, agentId, issueId: f.issueId, runId: f.runId, sessionId: f.native.normalizedSessionId } }; const profile = { ...f.preparedProfile, nativeExecutionInput: execution, nativeToolContractFingerprint: fingerprint, sessionCheckpoint: snapshot, nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared", descriptorSha256: "a".repeat(64), @@ -249,6 +256,12 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runnerInstanceId: f.native.runnerInstanceId, runtimeMode: "native", status: "succeeded", createdAt: new Date(Date.now() - 60_000), finishedAt: new Date(Date.now() - 50_000), runnerProfileJson: priorProfile }); + if (scenario === "stale_queued" || scenario === "stale_overlap") { + const now = Date.now(); + await f.db.update(heartbeatRuns).set({ createdAt: new Date(now - 45_000), + startedAt: new Date(now - (scenario === "stale_queued" ? 30_000 : 43_000)) }).where(eq(heartbeatRuns.id, f.runId)); + await f.db.update(heartbeatRuns).set({ finishedAt: new Date(now - 40_000) }).where(eq(heartbeatRuns.id, priorId)); + } const supervisor = JSON.stringify({ schema: "paperclip.native-session-supervisor.v1", configDigest: f.paths.configDigest, snapshot: priorSnapshot }); await mkdir(dirname(f.paths.supervisorPath), { recursive: true }); @@ -259,18 +272,48 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable if (scenario === "stale_success") await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); if (scenario === "stale_busy") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-owner", leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); - if (scenario === "stale_durable") { - const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, { runId: f.runId, - runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId }, + if (scenario.startsWith("stale_durable")) { + const state = JSON.parse(f.source); + if (scenario === "stale_durable_runner") state.identity.runnerInstanceId = randomUUID(); + if (scenario === "stale_durable_session") state.identity.normalizedSessionId = randomUUID(); + const checkpoint = captureLegacyDurablePrpCheckpoint(JSON.stringify(state), { runId: f.runId, + runnerInstanceId: state.identity.runnerInstanceId, normalizedSessionId: state.identity.normalizedSessionId }, { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }); await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, { companyId: f.companyId, issueId: f.issueId, runId: f.runId }, checkpoint)); + if (scenario === "stale_durable_absent") await rm(f.paths.supervisorPath); + if (scenario === "stale_durable_snapshot") await f.db.update(heartbeatRuns).set({ runnerProfileJson: { + ...f.profile, sessionCheckpoint: { ...f.snapshot, identity: { ...f.snapshot.identity, agentId: randomUUID() } }, + } }).where(eq(heartbeatRuns.id, f.runId)); } const [originalRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); const [originalOwner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects"))); const options = { storage, writersStopped: true as const, signal: AbortSignal.timeout(30_000) }; - if (scenario !== "stale_confirmed") { + if (scenario === "stale_durable" || scenario === "stale_durable_absent") { + const binding = { companyId: f.companyId, issueId: f.issueId, runId: f.runId }; + const before = await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId)); + expect(await migrateNativeControllers(f.db, options)).toMatchObject({ imported: 1, archivedFailed: 0 }); + expect(await migrateNativeControllers(f.db, options)).toMatchObject({ unchanged: 1, archivedFailed: 0 }); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toEqual(before); + expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0); + const [afterRun] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(afterRun.runnerProfileJson?.sessionCheckpoint).toEqual(originalRun.runnerProfileJson?.sessionCheckpoint); + expect(afterRun).toEqual({ ...originalRun, runnerProfileJson: afterRun.runnerProfileJson, updatedAt: afterRun.updatedAt }); + expect(afterRun.runnerProfileJson).toEqual({ ...originalRun.runnerProfileJson, + nativeControllerRecovery: expect.objectContaining({ snapshotSha256: nativeSha256(f.snapshot), + checkpoint: { runId: f.runId, runnerInstanceId: f.native.runnerInstanceId, normalizedSessionId: f.native.normalizedSessionId } }) }); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(originalOwner); + await rm(f.paperclipHome, { recursive: true, force: true }); + expect((await loadNativeRunnerCheckpoint(f.db, binding))?.records.size).toBe(before.length); + await f.db.update(heartbeatRuns).set({ nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, + status: "running", runtimeMode: "native" }).where(eq(heartbeatRuns.id, f.queuedId)); + expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).toBe("recoverable"); + return; + } + if (!["stale_confirmed", "stale_queued"].includes(scenario)) { await expect(migrateNativeControllers(f.db, options)).rejects.toThrow(); expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0); expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(true); @@ -299,6 +342,126 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).not.toBe("recoverable"); }); + if (scenario.startsWith("controller_")) it("preserves failed legacy files only with proven older controller lineage", async () => { + const f = await fixture(), priorId = randomUUID(); + const priorSnapshot = { ...f.snapshot, identity: { ...f.snapshot.identity, runId: priorId } }; + if (scenario === "controller_foreign") priorSnapshot.identity.agentId = randomUUID(); + const priorProfile = { ...f.profile, nativeExecutionInput: { ...f.profile.nativeExecutionInput, + binding: { ...f.profile.nativeExecutionInput.binding, runId: priorId } }, sessionCheckpoint: priorSnapshot }; + await f.db.insert(heartbeatRuns).values({ id: priorId, companyId: f.companyId, + agentId: f.agentId, + nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, + runnerInstanceId: scenario === "controller_runner" ? randomUUID() : f.native.runnerInstanceId, + runtimeMode: "native", status: "succeeded", createdAt: new Date(Date.now() - 60_000), + finishedAt: new Date(Date.now() - 50_000), runnerProfileJson: scenario === "controller_thread" + ? { ...priorProfile, sessionCheckpoint: { ...priorSnapshot, sessionId: "foreign-provider-thread" } } : priorProfile }); + const state = JSON.parse(f.source); + state.identity = { ...state.identity, runId: priorId, turnId: `turn-${priorId}`, itemId: `item-${priorId}` }; + if (scenario === "controller_invalid") state.schema = "unrecognized"; + const source = JSON.stringify(state); + await writeFile(f.file, source, { mode: 0o600 }); + if (scenario === "controller_success") await f.db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId)); + if (scenario === "controller_busy") await f.db.update(nativeRunFinalizations).set({ leaseOwner: "live-owner", + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "controller_durable") { + const checkpoint = captureLegacyDurablePrpCheckpoint(f.source, f.identity, + { version: "migration-fixture", digest: `sha256:${"a".repeat(64)}` }); + await f.db.transaction(tx => insertMigratedNativeRunnerCheckpoint(tx, + { companyId: f.companyId, issueId: f.issueId, runId: f.runId }, 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)); + } + if (scenario.startsWith("controller_observed_")) { + await f.db.update(nativeRunFinalizations).set({ phase: scenario === "controller_observed_result" ? "result_accepted" : "observed", + leaseOwner: "retired-controller", leaseExpiresAt: new Date(Date.now() + (scenario === "controller_observed_live" ? 60_000 : -60_000)), + nextAttemptAt: scenario === "controller_observed_retry" ? new Date(Date.now() + 60_000) : null, + }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (scenario === "controller_observed_task") await f.db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + if (scenario !== "controller_observed_missing") { + const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.runId, issueId: f.issueId, + provider: "daytona", providerLeaseId: randomUUID(), status: "released", cleanupStatus: "success", releasedAt: new Date() }; + const receipt = remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" })!; + if (scenario === "controller_observed_stop_before_finish") { + const now = Date.now(); + await f.db.update(heartbeatRuns).set({ createdAt: new Date(now - 40_000), startedAt: new Date(now - 30_000) }) + .where(eq(heartbeatRuns.id, f.runId)); + receipt.confirmedAt = new Date(now - 10_000).toISOString(); + lease.releasedAt = new Date(now - 10_000); + } + if (scenario === "controller_observed_foreign") receipt.runId = randomUUID(); + if (scenario === "controller_observed_old_receipt") receipt.confirmedAt = new Date(0).toISOString(); + await f.db.insert(environmentLeases).values({ ...lease, metadata: { remoteExecutionTermination: receipt } }); + if (scenario.startsWith("controller_observed_handoff")) { + const environmentId = randomUUID(), successorRunId = randomUUID(), handoffAt = new Date(Date.now() - 1000); + await f.db.insert(environments).values({ id: environmentId, name: environmentId, driver: "sandbox" }); + await f.db.insert(heartbeatRuns).values({ id: successorRunId, companyId: f.companyId, agentId: f.agentId, + nativeIssueId: f.issueId, nativeSessionId: f.native.normalizedSessionId, runtimeMode: "native", status: "succeeded", finishedAt: new Date() }); + const [previous] = await f.db.update(environmentLeases).set({ environmentId, leasePolicy: "reuse_by_environment", + status: "expired", acquiredAt: new Date(handoffAt.getTime() - 1000), releasedAt: handoffAt, + metadata: { agentId: f.agentId } }).where(eq(environmentLeases.id, lease.id)).returning(); + const successor = { ...previous, id: randomUUID(), heartbeatRunId: successorRunId, acquiredAt: handoffAt, + status: "released", releasedAt: new Date(), metadata: { agentId: f.agentId } }; + const handoff = createReusableLeaseHandoff(previous, successor); + const termination = remoteTerminationReceipt(successor, { providerLeaseId: successor.providerLeaseId, + state: scenario === "controller_observed_handoff_stopped" ? "stopped" : "destroyed" }); + await f.db.update(environmentLeases).set({ metadata: { agentId: f.agentId, reusableLeaseSuccessor: handoff } }) + .where(eq(environmentLeases.id, previous.id)); + await f.db.insert(environmentLeases).values({ ...successor, metadata: { agentId: f.agentId, + reusableLeasePredecessor: scenario === "controller_observed_handoff_broken" ? { ...handoff, handoffId: randomUUID() } : handoff, + remoteExecutionTermination: termination } }); + if (scenario === "controller_observed_handoff_fork") await f.db.insert(environmentLeases).values({ ...previous, + id: randomUUID(), heartbeatRunId: f.queuedId, status: "active", releasedAt: null, metadata: { agentId: f.agentId } }); + } + } + } + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const storage = createStorageService(createLocalDiskStorageProvider(join(f.runtimeRoot, "durable-objects"))); + const options = { storage, writersStopped: true as const, signal: AbortSignal.timeout(30_000) }; + if (scenario === "controller_observed_revoked") { + const put = storage.putFile.bind(storage); + vi.spyOn(storage, "putFile").mockImplementation(async args => { + const result = await put(args); + await f.db.update(environmentLeases).set({ metadata: {} }).where(eq(environmentLeases.heartbeatRunId, f.runId)); + return result; + }); + } + if (scenario === "controller_durable") { + expect(await migrateNativeControllers(f.db, options)).toMatchObject({ unchanged: 1, archivedFailed: 0 }); + } else if (!["controller_confirmed", "controller_observed_stopped", "controller_observed_stop_before_finish", "controller_observed_handoff"].includes(scenario)) { + await expect(migrateNativeControllers(f.db, options)).rejects.toThrow(); + const archives = await f.db.select().from(nativeHarnessBackups); + if (scenario === "controller_observed_revoked") { + expect(storage.putFile).toHaveBeenCalledOnce(); + expect(archives).toHaveLength(1); + expect(archives[0].status).toBe("pending"); + } + else expect(archives).toHaveLength(0); + } else { + const put = vi.spyOn(storage, "putFile").mockRejectedValue(new Error("storage unavailable")); + await expect(migrateNativeControllers(f.db, options)).rejects.toThrow("storage unavailable"); + put.mockRestore(); + expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 }); + expect(await migrateNativeControllers(f.db, options)).toMatchObject({ archivedFailed: 1, imported: 0 }); + expect(await hasLocalNativeControllerObligations(f.db, storage)).toBe(false); + expect(await f.db.select().from(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, f.runId))).toHaveLength(0); + await rm(f.paperclipHome, { recursive: true, force: true }); + const descriptor = legacyNativeControllerArchive({ companyId: f.companyId, issueId: f.issueId, runId: f.runId, controller: source }); + const replacement = join(f.runtimeRoot, "archived-older-controller"); + expect(await restoreDurableNativeHarnessBackup({ db: f.db, storage, companyId: f.companyId, + scopeDigest: descriptor.scopeDigest, backupRoot: replacement })).toBe(true); + expect(await readFile(join(replacement, "current", "control-plane-state.json"), "utf8")).toBe(source); + expect((await resolveNativeControllerRecovery(f.db, { companyId: f.companyId, agentId: f.agentId, + issueId: f.issueId, normalizedSessionId: f.native.normalizedSessionId, currentRunId: f.queuedId })).kind).not.toBe("recoverable"); + } + expect((await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toEqual(run); + expect((await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(owner); + expect((await f.db.select().from(agentTaskSessions).where(eq(agentTaskSessions.id, f.session.id)))[0].lastRunId).toBe(f.runId); + expect((await f.db.select().from(issues).where(eq(issues.id, f.issueId)))[0].executionRunId) + .toBe(scenario === "controller_observed_task" ? f.runId : f.queuedId); + }); if (scenario === "success") it("does not substitute a failure archive for a successful retained session", async () => { const f = await fixture(); await rm(join(f.paths.root, "provider-pack-authority.json")); diff --git a/server/src/services/native-runtime/native-controller-migration.ts b/server/src/services/native-runtime/native-controller-migration.ts index 93626245b40..443ffb73bed 100644 --- a/server/src/services/native-runtime/native-controller-migration.ts +++ b/server/src/services/native-runtime/native-controller-migration.ts @@ -3,7 +3,7 @@ import { lstat, open, readdir } from "node:fs/promises"; import { join } from "node:path"; import { and, eq, sql } from "drizzle-orm"; import { heartbeatRuns, type Db } from "@paperclipai/db"; -import { parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; +import { captureLegacyDurablePrpCheckpoint, parseNativeExecutionInput } from "../../vendor/paperclip-runner/index.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; import { findNativeSessionResumeRun } from "./native-session-resume.js"; import { nativeControllerMigrationDescriptor } from "./native-session-executor.js"; @@ -24,34 +24,54 @@ function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } -async function hasConfirmedOlderSupervisor(db: Db, selected: Selected, +async function confirmedOlderSuccessfulSnapshot(db: Db, selected: Selected, source: NonNullable>>, - paths: Awaited>, supervisor: unknown) { + paths: Awaited>, olderRunId: unknown) { const snapshot = record(source.runnerProfileJson.sessionCheckpoint), identity = record(snapshot.identity); - const olderSnapshot = record(record(supervisor).snapshot), olderIdentity = record(olderSnapshot.identity); const scope = { companyId: selected.companyId, agentId: selected.agentId, issueId: selected.issueId, sessionId: selected.sessionId }; if (source.status !== "failed" || identity.runId !== source.id - || Object.entries(scope).some(([key, value]) => identity[key] !== value || olderIdentity[key] !== value) + || Object.entries(scope).some(([key, value]) => identity[key] !== value) || paths.execution.binding.runId !== source.id || paths.execution.binding.companyId !== selected.companyId || paths.execution.binding.agentId !== selected.agentId || paths.execution.binding.issueId !== selected.issueId || paths.execution.session.normalizedSessionId !== selected.sessionId || typeof snapshot.sessionId !== "string" || !snapshot.sessionId.trim() - || snapshot.sessionId !== olderSnapshot.sessionId || snapshot.backendKind !== olderSnapshot.backendKind - || typeof olderIdentity.runId !== "string" || olderIdentity.runId === source.id - || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(olderIdentity.runId)) return false; + || typeof olderRunId !== "string" || olderRunId === source.id + || !/^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(olderRunId)) return null; const [prior] = await db.select({ createdAt: heartbeatRuns.createdAt, finishedAt: heartbeatRuns.finishedAt, runnerInstanceId: heartbeatRuns.runnerInstanceId, snapshot: sql`${heartbeatRuns.runnerProfileJson}->'sessionCheckpoint'` }) - .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, olderIdentity.runId), eq(heartbeatRuns.companyId, selected.companyId), + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, olderRunId), eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId), eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"))); - if (!prior?.finishedAt || prior.createdAt >= source.createdAt || prior.finishedAt > source.createdAt + const priorSnapshot = record(prior?.snapshot), priorIdentity = record(priorSnapshot.identity); + if (!prior?.finishedAt || prior.createdAt >= source.createdAt || !source.runnerInstanceId || prior.runnerInstanceId !== source.runnerInstanceId - || nativeSha256(prior.snapshot) !== nativeSha256(redactRunnerApiPayloads(olderSnapshot))) return false; - const [failed] = await db.select({ finishedAt: heartbeatRuns.finishedAt }).from(heartbeatRuns).where(and( + || priorIdentity.runId !== olderRunId || Object.entries(scope).some(([key, value]) => priorIdentity[key] !== value) + || priorSnapshot.sessionId !== snapshot.sessionId || priorSnapshot.backendKind !== snapshot.backendKind) return null; + const [failed] = await db.select({ finishedAt: heartbeatRuns.finishedAt, startedAt: heartbeatRuns.startedAt }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, selected.companyId), eq(heartbeatRuns.agentId, selected.agentId), eq(heartbeatRuns.nativeIssueId, selected.issueId), eq(heartbeatRuns.nativeSessionId, selected.sessionId), eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "failed"))); - return Boolean(failed?.finishedAt); + return failed?.finishedAt && prior.finishedAt <= (failed.startedAt ?? source.createdAt) ? priorSnapshot : null; +} + +async function hasConfirmedOlderSupervisor(db: Db, selected: Selected, + source: NonNullable>>, + paths: Awaited>, supervisor: unknown) { + const olderSnapshot = record(record(supervisor).snapshot); + const prior = await confirmedOlderSuccessfulSnapshot(db, selected, source, paths, record(olderSnapshot.identity).runId); + return prior !== null && nativeSha256(prior) === nativeSha256(redactRunnerApiPayloads(olderSnapshot)); +} + +async function hasConfirmedOlderController(db: Db, selected: Selected, + source: NonNullable>>, + paths: Awaited>, controller: string, artifact?: { version: string; digest: string }) { + const identity = record(record(JSON.parse(controller)).identity); + if (!await confirmedOlderSuccessfulSnapshot(db, selected, source, paths, identity.runId)) return false; + try { + captureLegacyDurablePrpCheckpoint(controller, { runId: String(identity.runId), + runnerInstanceId: source.runnerInstanceId!, normalizedSessionId: selected.sessionId }, artifact); + return true; + } catch { return false; } } async function* retainedSources(db: Db, signal?: AbortSignal) { @@ -181,7 +201,7 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped: providerDigest: nativeSha256(execution.provider) }, current: unavailable, runnerArtifact: unavailable }).runnerArtifact; } - const supervisor = paths.execution.session.lifecyclePolicy.mode === "warm" + const supervisor = !checkpoint && paths.execution.session.lifecyclePolicy.mode === "warm" ? await readPrivateSource(paths.supervisorPath, 4 * 1024 * 1024) : undefined; input.signal.throwIfAborted(); let result: Awaited>; @@ -192,8 +212,11 @@ export async function migrateNativeControllers(db: Db, input: { writersStopped: offline: { writersStopped: true, legacyControllerState, legacyRunnerArtifact } }); } catch (error) { const storage = input.storage ?? durableNativeHarnessStorage(); - if (!(error instanceof Error) || error.message !== "native_runner_checkpoint_artifact_unproven" - || source.status !== "failed" || legacyControllerState === undefined || !storage) throw error; + if (!(error instanceof Error) || source.status !== "failed" || legacyControllerState === undefined || !storage) throw error; + const archivable = error.message === "native_runner_checkpoint_artifact_unproven" + || error.message === "native_runner_checkpoint_binding_mismatch" + && await hasConfirmedOlderController(db, selected, source, paths, legacyControllerState, legacyRunnerArtifact); + if (!archivable) throw error; await archiveFailedLegacyNativeController(db, { ...binding, storage, controller: legacyControllerState, supervisor, writersStopped: true, signal: input.signal }); counts.archivedFailed++; diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index 64e9f6d1d23..e78beedfbe3 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -262,17 +262,18 @@ export async function importNativeControllerRecovery(db: Db, input: { || input.supervisor !== undefined && nativeSha256(snapshot) !== nativeSha256(redactRunnerApiPayloads(supervisor.snapshot))) { return { kind: "missing_evidence" as const, reason: "stale_supervisor_snapshot" }; } + let checkpoint = await loadNativeRunnerCheckpoint(tx, input); if (input.supervisor === undefined) { try { const admitted = parseNativeExecutionInput(profile.nativeExecutionInput); - if (!input.offline || admitted.session.lifecyclePolicy.mode !== "per_turn" + if (!input.offline || (!checkpoint && admitted.session.lifecyclePolicy.mode !== "per_turn") + || admitted.session.normalizedSessionId !== input.normalizedSessionId || admitted.binding.runId !== input.runId || admitted.binding.companyId !== input.companyId || admitted.binding.agentId !== input.agentId || admitted.binding.issueId !== input.issueId) { return { kind: "missing_evidence" as const, reason: "admitted_per_turn_binding" }; } } catch { return { kind: "unsupported" as const, reason: "admitted_input_format" }; } } - let checkpoint = await loadNativeRunnerCheckpoint(tx, input); const importing = !checkpoint && input.offline?.legacyControllerState !== undefined; if (importing) { if (!run.runnerInstanceId) return { kind: "missing_evidence" as const, reason: "checkpoint_binding" }; diff --git a/server/src/services/native-runtime/native-harness-backup-storage.ts b/server/src/services/native-runtime/native-harness-backup-storage.ts index 47ab7e74dc9..b6de9ffa865 100644 --- a/server/src/services/native-runtime/native-harness-backup-storage.ts +++ b/server/src/services/native-runtime/native-harness-backup-storage.ts @@ -7,7 +7,7 @@ import os from "node:os"; import { pipeline } from "node:stream/promises"; import { Transform } from "node:stream"; import { and, eq, inArray, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, nativeRunFinalizations, type Db } from "@paperclipai/db"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; import type { StorageService } from "../../storage/types.js"; import { lockNativeRunnerCheckpointOwner, validateNativeRunnerCheckpointOwner, type NativeRunnerCheckpointFenceOwner } from "./native-runner-checkpoint.js"; @@ -16,6 +16,7 @@ import { getStorageService } from "../../storage/index.js"; import { readNativeHarnessBackupStamp, verifyNativeHarnessBackupStamp } from "./native-harness-backup-stamp.js"; import { captureNativeHarnessArchive, restoreNativeHarnessArchive, MAX_NATIVE_HARNESS_OBJECT_BYTES } from "./native-harness-backup-archive.js"; import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; +import { hasRemoteTerminationReceipt, resolveRemoteTerminationLease } from "../remote-execution-termination.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; const SCHEMA = "paperclip.native-harness-object.v1"; @@ -98,18 +99,37 @@ async function fence(tx: Parameters[0]>[0], owner: if (!owner.writersStopped) fail("offline_writers_required"); const [active] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(offlineMigrationExecutionWhere()).limit(1); if (active) fail("offline_writers_required"); - const [run] = await tx.select({ finishedAt: heartbeatRuns.finishedAt, status: heartbeatRuns.status }).from(heartbeatRuns).where(and( + const [run] = await tx.select({ finishedAt: heartbeatRuns.finishedAt, startedAt: heartbeatRuns.startedAt, createdAt: heartbeatRuns.createdAt, status: heartbeatRuns.status }).from(heartbeatRuns).where(and( eq(heartbeatRuns.id, owner.runId), eq(heartbeatRuns.companyId, owner.companyId), eq(heartbeatRuns.nativeIssueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "native"))).for("share"); const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner, - leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase }) + leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, phase: nativeRunFinalizations.phase, + resultId: nativeRunFinalizations.resultId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt }) .from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, owner.runId)).for("share"); const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`); if (!run?.finishedAt || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status) - || !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now)) - || owner.terminalFailureOnly && (run.status !== "failed" || coordinator.phase !== "terminal_failure")) { + || !coordinator || coordinator.leaseOwner && (!coordinator.leaseExpiresAt || coordinator.leaseExpiresAt > new Date(clock.now))) { fail("offline_owner_unsettled"); } + if (owner.terminalFailureOnly) { + if (run.status !== "failed") fail("offline_owner_unsettled"); + if (coordinator.phase !== "terminal_failure") { + // Evidence archival does not settle or resume this stale coordinator. + // Lease expiry alone is insufficient: every remote writer needs its + // exact host-owned stop receipt, recorded after execution began. + if (coordinator.phase !== "observed" || coordinator.resultId || coordinator.nextAttemptAt + || task.executionRunId === owner.runId) fail("offline_owner_unsettled"); + const leases = await tx.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, owner.companyId), eq(environmentLeases.heartbeatRunId, owner.runId))).for("share"); + const stopped = await Promise.all(leases.map(lease => resolveRemoteTerminationLease(tx as unknown as Db, lease))); + if (!stopped.length || stopped.some(lease => { + const receipt = lease.metadata?.remoteExecutionTermination as { confirmedAt?: unknown } | undefined; + const confirmedAt = Date.parse(String(receipt?.confirmedAt)); + return lease.issueId !== owner.issueId || !hasRemoteTerminationReceipt(lease) + || !Number.isFinite(confirmedAt) || confirmedAt < (run.startedAt ?? run.createdAt).getTime() || confirmedAt > new Date(clock.now).getTime(); + })) fail("offline_owner_unsettled"); + } + } return; } const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`);