fix(runtime): honor durable authority and retired lease handoffs

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-11-04 15:10:54 -06:00
1 parent d71dc913f7
commit 8ff3a22109
5 files changed
+82 -12

No files matched your search

@@ -156,3 +156,17 @@ 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.
@@ -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, environmentLeases, 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";
@@ -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 { 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";
@@ -22,10 +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_queued", "stale_overlap", "stale_success", "stale_busy", "stale_durable",
"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"])("offline native controller migration (%s)", scenario => {
"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) {
@@ -269,17 +272,47 @@ 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_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);
@@ -360,6 +393,27 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
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));
@@ -376,7 +430,7 @@ describeEmbeddedPostgres.each(["import", "archive", "success", "batch", "durable
}
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"].includes(scenario)) {
} 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") {
@@ -201,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<ReturnType<typeof importNativeControllerRecovery>>;
@@ -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" };
@@ -16,7 +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 { hasRemoteTerminationReceipt, resolveRemoteTerminationLease } from "../remote-execution-termination.js";
import { offlineMigrationExecutionWhere } from "../offline-migration-execution.js";
const SCHEMA = "paperclip.native-harness-object.v1";
@@ -121,7 +121,8 @@ async function fence(tx: Parameters<Parameters<Db["transaction"]>[0]>[0], owner:
|| 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 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)