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 9a21d3e3e95..97b22d1da5b 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -144,3 +144,10 @@ 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 the failed finish. 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. 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 43515e99312..7e280a12fc6 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,7 +3,7 @@ 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, 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"; @@ -13,6 +13,7 @@ import { insertMigratedNativeRunnerCheckpoint, loadNativeRunnerCheckpoint } from 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 { 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"; @@ -22,7 +23,9 @@ import { restoreDurableNativeHarnessBackup } from "./native-harness-backup-stora describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable_failed", "stale_confirmed", "stale_foreign", "stale_changed", "stale_success", "stale_busy", "stale_durable", - "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable"])("offline native controller migration (%s)", scenario => { + "controller_confirmed", "controller_foreign", "controller_thread", "controller_runner", "controller_success", "controller_busy", "controller_invalid", "controller_durable", + "controller_observed_stopped", "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"])("offline native controller migration (%s)", scenario => { const environment = nativeRestartFixture(); afterEach(() => vi.unstubAllEnvs()); async function fixture(commandCount = 1, warm = false) { @@ -331,15 +334,44 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable { 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_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 } }); + } + } 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 (scenario !== "controller_confirmed") { + } else if (!["controller_confirmed", "controller_observed_stopped"].includes(scenario)) { await expect(migrateNativeControllers(f.db, options)).rejects.toThrow(); - expect(await f.db.select().from(nativeHarnessBackups)).toHaveLength(0); + 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"); @@ -360,7 +392,8 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable 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(f.queuedId); + 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(); 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..e6b704dbc66 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 } from "../remote-execution-termination.js"; import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js"; const SCHEMA = "paperclip.native-harness-object.v1"; @@ -102,14 +103,32 @@ async function fence(tx: Parameters[0]>[0], owner: 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 the failed finish. + 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"); + if (!leases.length || leases.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.finishedAt!.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`);