mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
fix(native): refresh per-turn projections from the selected checkpoint (#19744)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Per-turn native sessions save controller checkpoints and suspended Runner backups under separate owners. > - A local controller projection can still name the first turn after the next turn has saved its checkpoint and backup. > - The old per-turn restore path selected that stale local reference instead of the newer canonical source. > - Admission then saw different turn identities and archived the valid backup with the old projection. > - This change uses the canonical selected checkpoint and the existing finished-owner proof before it replaces a stale projection. > - The benefit is that a third turn keeps the valid saved conversation and backup. ## Linked Issues or Issue Description **What happened?** After two successful per-turn remote runs, a third turn could fail before provider execution with unavailable provider resume state. The selected database checkpoint and suspended backup belonged to the second run. The local controller projection still belonged to the first run. **Expected behavior** The selected checkpoint must govern controller rehydration. Preserve the older projection and the selected run's backup. Do not overwrite current, newer, foreign, changed or unsettled ownership evidence. **Steps to reproduce** 1. Save two real controller checkpoints for one per-turn conversation and finish both owners. 2. Leave the first checkpoint's readable projection or external marker beside the second run's suspended backup. 3. Admit the third run through the real canonical selector and backend owner. 4. Before this change, admission archives the active backup with the stale projection and leaves no active controller state. Related: #18018 added per-turn controller restoration. #18718 refreshed warm projections. #18846 added finished-owner retirement. This change keeps the selected per-turn source from entering that retirement path because of an older projection. The open warm-only change #18436 concerns a different restart branch. ## What Changed - Resolve the canonical source before refreshing a prior per-turn controller projection. - Re-read it under the existing sorted coordinator locks. Replacing an older owner reuses finished-owner checks for both runs and verifies their full session scope. A same-source marker only rehydrates its exact selected identity before the normal retained-sandbox verification. - Recheck the exact initial local bytes after asynchronous reads and lock waits. Preserve the old control-plane directory in the existing archive, while keeping the newer backup and provider files active. - Restore a current-run external marker only from its own checkpoint before the existing bootstrap retry proof. A missing current checkpoint cannot select a previous turn. Retain readable current-run, legacy file-backed, backup integrity and provider resume checks. - Stage the existing OpenCode proxy bytes into the controlled fake-sandbox filesystem in the real ingress-retry fixture. Keep remote artifact path and digest checks unchanged. - Add real PostgreSQL/filesystem coverage for Codex and ACPX, readable projections and external markers, missing or corrupt parts, held owners, foreign scope, newer history and a current owner arriving during a real coordinator-lock wait. ## Verification - Baseline RED: the two new readable-projection cases failed against unchanged production code because active control-plane state was absent after admission. The real selector selected the second run. Setup failures were separate and are not counted as RED. - Exact final existing per-turn restore, active-owner refresh and retained-owner wait controls: 19/19 passed in 5.91s. - Exact final local source (`fd8d9d9021`, based on `131b566ca4`): 58/58 real PostgreSQL/filesystem and native Runner tests passed in 31.58s. This includes both existing ingress bootstrap retries through the real Runner and fake Codex/OpenCode providers, plus 56 focused owner cases. Cargo validated and rebuilt current source from a private APFS clone of an unqualified local cache; the original cache was untouched. A private local configuration resolves workspace DB, Shared and Runner sources without replacing behavior. The local CLI allowed a 90s setup hook; no default timeout changed. - Exact final full server and test-tree TypeScript check with Runner source paths: passed. The stock fresh-worktree check first failed on unbuilt Runner declaration entries; no product types were changed to hide that setup gap. - Exact final module-boundary and diff-whitespace checks: passed. - Native-server CI RED on the previous head: both real ingress retry cases failed because a current-run marker was returned without hydration. The aggregate gate failed from that one underlying lane. The correction preserves the own-checkpoint bootstrap path. The first local OpenCode attempt hit a controller-host fixture path guard before its intended failure; staging the same proxy bytes into the fake sandbox made both native cases pass. - Review regression RED: the real retained-sandbox fixture failed at `sandbox_resumed` on the first PR head. GREEN: the corrected owner hydrates the same-source marker and reaches the existing exact retained-lease/suspended-runner read. A mismatched item identity still refuses before any remote read. - Local environment: macOS arm64, Node 26.4.0, PostgreSQL 18.1. Provider execution is deliberately outside this admission fixture. The real three-turn workflow still needs verification on the normally deployed fixed source. ## Risks - Refresh requires settled, committed, released and stopped or fenced owners. Missing or changed evidence remains blocked. - The refresh holds existing coordinator locks while it validates bounded controller state. It performs no provider calls. - The old control-plane bytes remain in the archive. A restore write failure still fails admission; it does not reset provider state. - Final-head full CI is green, including typecheck, build, Rust, native-server, server and browser suites; Greptile is 5/5 with no unresolved threads. Hosted deployment and continuity verification remain separate acceptance steps. A successful replacement conversation is not proof of provider continuity; hosted continuity acceptance is still pending. ## Model Used OpenAI Codex, GPT-6 family, with code execution and repository tools. The host does not expose the exact backend model ID or context-window size. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
24cbea136f
commit
18b6b71d83
4 files changed
+444
-30
No files matched your search
@@ -3811,6 +3811,23 @@ identity, zero-activity and pending-bootstrap-command checks apply before the
|
||||
remote directory claim; saved history and existing directories still prevent
|
||||
an unverified fresh start.
|
||||
|
||||
Per-turn controller recovery uses the newest run selected by the canonical
|
||||
session selector. A readable older projection or an old external marker does
|
||||
not replace that selection. Replacing an older projection holds the existing source coordinator
|
||||
locks and requires settled controller state, committed results, released
|
||||
owners and stopped or fenced controllers for both source runs. The full
|
||||
session scope must match. A current or newer owner, changed local bytes,
|
||||
unreadable checkpoint or unsettled state blocks replacement. Refresh retains
|
||||
the exact old control-plane directory in the existing continuity archive. It
|
||||
keeps the selected run's failover backup and provider files active. The normal
|
||||
backup verification and provider resume checks still run; this refresh does
|
||||
not start a new provider session or declare a continuity break. A marker that
|
||||
already names the exact selected identity only rehydrates its own controller
|
||||
state. The existing retained-sandbox binding and suspended Runner read still
|
||||
govern that same-source resume. A current-run marker rehydrates only that run's
|
||||
own checkpoint before the existing bootstrap retry checks. A missing current
|
||||
checkpoint never selects an older turn instead.
|
||||
|
||||
For the native runner, warm mode requests a reusable sandbox lease **before**
|
||||
lease acquisition. The environment's explicit runner lifecycle overrides the
|
||||
agent default; an inherited lifecycle uses the current agent setting on each
|
||||
|
||||
+2
-1
@@ -98,7 +98,8 @@ describeEmbeddedPostgres("native ingress bootstrap retry through the real Runner
|
||||
`#include <unistd.h>\n#include <stdlib.h>\n#include <stdio.h>\nint main(int argc, char **argv) { FILE *log = fopen(${JSON.stringify(providerCalls)}, "a"); if (!log) return 126; fputs("process-start\\n", log); fclose(log); char **args = calloc(argc + 2, sizeof(char *)); args[0] = ${JSON.stringify(providerNode)}; args[1] = ${JSON.stringify(script)}; for (int i = 1; i < argc; i++) args[i + 1] = argv[i]; execv(args[0], args); return 127; }` });
|
||||
await chmod(opencode, 0o755);
|
||||
}
|
||||
const proxy = resolve(runnerWorkspace, "../dist/cli/opencode-app-server-proxy.cjs");
|
||||
const proxy = join(filesystem, "opencode-app-server-proxy.cjs");
|
||||
if (provider === "opencode") await copyFile(resolve(runnerWorkspace, "../dist/cli/opencode-app-server-proxy.cjs"), proxy);
|
||||
const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId,
|
||||
normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` };
|
||||
const assertSealedSkills = async () => {
|
||||
|
||||
@@ -4,10 +4,10 @@ import { existsSync } from "node:fs";
|
||||
import { cp, mkdir, mkdtemp, readFile, readdir, rename, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations,
|
||||
agents, companies, completionContracts, createDb, environmentLeases, environments, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations,
|
||||
nativeRunnerCheckpoints, nativeRunResults,
|
||||
} from "@paperclipai/db";
|
||||
import { DurablePrpControlPlane, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js";
|
||||
@@ -15,9 +15,10 @@ import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/
|
||||
import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js";
|
||||
import { prepareNativeHeartbeatRun } from "./prepare-native-run.js";
|
||||
import { buildNativeExecutionInput } from "./native-execution-input.js";
|
||||
import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { loadNativeRunnerCheckpoint, nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js";
|
||||
import { createNativeControllerRecovery } from "./native-controller-recovery.js";
|
||||
import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
|
||||
import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
|
||||
import { readProcessStartedAt } from "../hot-restart.js";
|
||||
import {
|
||||
buildNativeExecutionWithCheckpoint, findNativeSessionResumeRun, nativeToolContractFingerprintForTarget,
|
||||
@@ -38,7 +39,7 @@ vi.mock("./current-wake-comments.js", async importOriginal => ({
|
||||
resolveCurrentWakeCommentsBinding: async () => null,
|
||||
assertCurrentWakeCommentsRead: async () => undefined,
|
||||
}));
|
||||
import { createRunnerdBackend } from "./native-session-executor.js";
|
||||
import { buildNativeHarnessBackupManifest, createRunnerdBackend, resolveNativeHarnessPersistenceProfile, scopedRunnerdStateRoot } from "./native-session-executor.js";
|
||||
|
||||
// Durable-root admission is the boundary under test. Reaching the run's tool
|
||||
// authority proves admission passed; no provider or runner process starts.
|
||||
@@ -89,7 +90,7 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
|
||||
* suspended Runner receipt or failover backup ever reached the controller.
|
||||
* Only the database controller checkpoint and recovery pointer survive. */
|
||||
async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled",
|
||||
options: { controller?: { bootId: string; pid: number; processStartedAt: Date } } = {}) {
|
||||
options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode" } = {}) {
|
||||
const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID();
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" });
|
||||
const [row] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native",
|
||||
@@ -97,7 +98,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
|
||||
const native = await prepareNativeHeartbeatRun({ db, run: row, issue: issueFields(issueId), environmentLeaseId: randomUUID() });
|
||||
const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId));
|
||||
const executionFor = (currentRunId: string) => buildNativeExecutionInput({
|
||||
companyId, agentId, runId: currentRunId, provider: "acpx", acpxAgent: "claude", model: "claude-opus-5-5",
|
||||
companyId, agentId, runId: currentRunId, provider: options.provider ?? "acpx", acpxAgent: "claude",
|
||||
model: options.provider === "codex" ? "gpt-6.1-sol"
|
||||
: options.provider === "opencode" ? "openrouter/deepseek/deepseek-v4.1-flash-1017" : "claude-opus-5-5",
|
||||
issue: { id: issueId, identifier: "ROF-1", title: "Continue after recovery", description: null, workMode: "standard" },
|
||||
taskPrompt: "Continue the conversation.",
|
||||
workspace: { id: workspaceId, cwd: "/workspace", repoUrl: null, repoRef: null, branchName: null },
|
||||
@@ -177,19 +180,22 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...current.runnerProfileJson, nativeExecutionInput: execution,
|
||||
nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("remote", execution.provider.kind),
|
||||
sessionCheckpoint: admission.checkpoint } }).where(eq(heartbeatRuns.id, runId));
|
||||
const logs: string[] = [];
|
||||
const start = (outcome: "created" | "replacement" = "created") => createRunnerdBackend({
|
||||
const logs: string[] = [], environmentId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${runId}`;
|
||||
const remoteExecute = vi.fn(async (_request: unknown): Promise<{ exitCode: number; timedOut: boolean; stdout: string; stderr: string }> => {
|
||||
throw new Error("admission must not reach the sandbox");
|
||||
});
|
||||
const start = (outcome: "created" | "replacement" | "resumed" = "created") => createRunnerdBackend({
|
||||
db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); },
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId: randomUUID(),
|
||||
leaseId: randomUUID(), remoteCwd: "/workspace", reusableLeaseConfigured: true,
|
||||
runner: { execute: vi.fn(async () => { throw new Error("admission must not reach the sandbox"); }) },
|
||||
sandboxLeaseAcquisition: { outcome, providerLeaseId: `sandbox-${runId}` } } as never,
|
||||
runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId,
|
||||
leaseId, remoteCwd: "/workspace", reusableLeaseConfigured: true,
|
||||
runner: { execute: remoteExecute },
|
||||
sandboxLeaseAcquisition: { outcome, providerLeaseId } } as never,
|
||||
}).then(() => "constructed", (error: Error) => error.message);
|
||||
// A local Runner keeps its controller root on this host between turns.
|
||||
const startLocal = () => createRunnerdBackend({
|
||||
db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); },
|
||||
}).then(() => "constructed", (error: Error) => error.message);
|
||||
return { runId, logs, start, startLocal, selectedRunId: selected?.id ?? null };
|
||||
return { runId, logs, start, startLocal, native: next, environmentId, leaseId, providerLeaseId, remoteExecute, selectedRunId: selected?.id ?? null };
|
||||
}
|
||||
|
||||
async function quarantined() {
|
||||
@@ -212,6 +218,264 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered
|
||||
}));
|
||||
}
|
||||
|
||||
async function seedSuccessorCheckpoint(prior: Awaited<ReturnType<typeof seedRecoveredPriorRun>>) {
|
||||
const pending = await admitFollowUp(prior), runId = pending.runId, execution = prior.executionFor(runId);
|
||||
const identity = { ...prior.identity, runId, turnId: `turn-${runId}`, itemId: `item-${runId}` };
|
||||
const owner = { companyId, issueId: prior.issueId, runId, authorityRunId: runId,
|
||||
bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
|
||||
await db.update(heartbeatRuns).set({ runnerInstanceId: identity.runnerInstanceId }).where(eq(heartbeatRuns.id, runId));
|
||||
await db.update(nativeRunFinalizations).set({ controllerBootId: owner.bootId, controllerGeneration: 1,
|
||||
controllerPid: 2_147_483_000, controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner: owner.leaseOwner,
|
||||
leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
const scratch = await mkdtemp(join(tmpdir(), "native-successor-checkpoint-"));
|
||||
const controller = new DurablePrpControlPlane({ stateDirectory: scratch, identity,
|
||||
expectedRunnerVersion: "recovered-owner-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
try {
|
||||
controller.persistRunAttachTemplate({ provider: execution.provider, normalizedSessionId: identity.normalizedSessionId });
|
||||
for (const type of ["run.prepare", "session.open", "turn.start", "runner.suspend"]) {
|
||||
Object.assign(controller.queueCommand(type), { status: "completed", result: { status: "completed" } });
|
||||
}
|
||||
controller.store.state.committedEvents.push({ sourceSeq: 2, sourceEventId: `terminal-${runId}`, eventType: "run.terminal",
|
||||
priority: 0, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 });
|
||||
controller.store.state.ackedSourceSeq = 2;
|
||||
controller.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner));
|
||||
await controller.flushCheckpoint();
|
||||
} finally {
|
||||
await controller.stop();
|
||||
await rm(scratch, { recursive: true, force: true });
|
||||
}
|
||||
const [prepared] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const snapshot: PersistedNativeSession = { backendKind: "runner", sessionId: prior.native.normalizedSessionId,
|
||||
providerSessionId: "provider-session-before-recovery", activeTurnId: null, pendingRuntimeRequests: [],
|
||||
identity: { companyId, agentId, issueId: prior.issueId, runId, sessionId: prior.native.normalizedSessionId } };
|
||||
const [result] = await db.insert(nativeRunResults).values({ companyId, issueId: prior.issueId, runId,
|
||||
completionContractId: prepared.completionContractId!, serverFingerprint: runId, schemaStatus: "accepted",
|
||||
resultJson: {}, canonicalSha256: runId }).returning();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date(), runnerProfileJson: {
|
||||
...prepared.runnerProfileJson, nativeExecutionInput: execution, sessionCheckpoint: snapshot,
|
||||
nativeControllerRecovery: createNativeControllerRecovery(snapshot,
|
||||
{ configDigest: `sha256:${"b".repeat(64)}`, processConfigDigest: null },
|
||||
{ runId, runnerInstanceId: identity.runnerInstanceId, normalizedSessionId: identity.normalizedSessionId }),
|
||||
} }).where(eq(heartbeatRuns.id, runId));
|
||||
await db.update(nativeRunFinalizations).set({ phase: "committed", resultId: result.id, leaseOwner: null, leaseExpiresAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, runId));
|
||||
return { runId, execution, identity };
|
||||
}
|
||||
|
||||
async function staleProjection(provider: "codex" | "acpx" = "codex", external = false) {
|
||||
const prior = await seedRecoveredPriorRun("settled", { provider });
|
||||
const latest = await seedSuccessorCheckpoint(prior), followUp = await admitFollowUp(prior);
|
||||
expect(followUp.selectedRunId).toBe(latest.runId);
|
||||
const root = scopedRunnerdStateRoot(latest.execution), backupRoot = join(root, "failover-backups", "current");
|
||||
const oldCheckpoint = await loadNativeRunnerCheckpoint(db, { companyId, issueId: prior.issueId, runId: prior.runId });
|
||||
expect(oldCheckpoint).not.toBeNull();
|
||||
oldCheckpoint!.restoreSessionDirectory(root);
|
||||
const path = join(root, "control-plane", "control-plane-state.json");
|
||||
if (external) await writeFile(path, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: prior.identity }));
|
||||
const originalBytes = await readFile(path, "utf8");
|
||||
const providerFiles = new Map<string, string>();
|
||||
for (const directory of resolveNativeHarnessPersistenceProfile(latest.execution).directories) {
|
||||
await mkdir(join(backupRoot, directory.name), { recursive: true });
|
||||
if (directory.name !== "runner") {
|
||||
const file = join(backupRoot, directory.name, "retained-conversation.txt"), bytes = `saved conversation for ${latest.runId}`;
|
||||
await writeFile(file, bytes);
|
||||
providerFiles.set(file, bytes);
|
||||
}
|
||||
}
|
||||
const runnerBytes = JSON.stringify({ schema: "paperclip.runner.durable.state.v1", ...latest.identity,
|
||||
lifecycle: "suspended", outbox: [], pendingTerminalDelivery: null, pendingProviderCleanup: null });
|
||||
await writeFile(join(backupRoot, "runner", "runner-state.json"), runnerBytes);
|
||||
const manifestBytes = JSON.stringify(buildNativeHarnessBackupManifest({ backupRoot, execution: latest.execution,
|
||||
runnerInstanceId: latest.identity.runnerInstanceId, sourceProviderLeaseId: `sandbox-${latest.runId}`,
|
||||
providerSessionIdentity: { providerSessionId: "provider-session-before-recovery", providerBackendSessionId: null,
|
||||
providerSessionIdentity: null } }));
|
||||
await writeFile(join(backupRoot, "manifest.json"), manifestBytes);
|
||||
const assertBackup = async () => {
|
||||
expect(await readFile(join(backupRoot, "manifest.json"), "utf8")).toBe(manifestBytes);
|
||||
expect(await readFile(join(backupRoot, "runner", "runner-state.json"), "utf8")).toBe(runnerBytes);
|
||||
for (const [file, bytes] of providerFiles) expect(await readFile(file, "utf8")).toBe(bytes);
|
||||
};
|
||||
return { prior, latest, followUp, root, path, backupRoot, originalBytes, assertBackup };
|
||||
}
|
||||
|
||||
it.each([
|
||||
{ provider: "codex" as const, external: false }, { provider: "acpx" as const, external: false },
|
||||
{ provider: "codex" as const, external: true }, { provider: "acpx" as const, external: true },
|
||||
])("keeps the newest $provider checkpoint paired with its suspended backup (external=$external)", async ({ provider, external }) => {
|
||||
const fixture = await staleProjection(provider, external);
|
||||
const { prior, latest, followUp, root, path } = fixture;
|
||||
expect(await followUp.start()).toBe(ADMITTED);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).identity).toEqual(latest.identity);
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(root)).toEqual([prior.runId]);
|
||||
const [archive] = await readdir(join(root, "continuity-breaks"));
|
||||
expect(await readFile(join(root, "continuity-breaks", archive!, "control-plane", "control-plane-state.json"), "utf8"))
|
||||
.toBe(fixture.originalBytes);
|
||||
expect(await readdir(join(root, "continuity-breaks", archive!))).toEqual(["control-plane"]);
|
||||
expect(await quarantined()).toEqual([]);
|
||||
expect(followUp.logs.join("")).not.toContain("starting a new runner for this conversation");
|
||||
}, 60_000);
|
||||
|
||||
it("rehydrates a same-source external marker before reading the suspended resumed sandbox", async () => {
|
||||
const fixture = await staleProjection("codex", true), { latest, followUp, root, path } = fixture;
|
||||
await writeFile(path, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: latest.identity }));
|
||||
await db.insert(environments).values({ id: followUp.environmentId, name: `retained-${followUp.environmentId}`, driver: "sandbox" });
|
||||
await db.insert(environmentLeases).values({ companyId, environmentId: followUp.environmentId,
|
||||
issueId: fixture.prior.issueId, heartbeatRunId: latest.runId, provider: "daytona", providerLeaseId: followUp.providerLeaseId,
|
||||
status: "released", metadata: { remoteCwd: "/workspace" } });
|
||||
const runnerState = { schema: "paperclip.runner.durable.state.v1", ...latest.identity, lifecycle: "suspended", outbox: [],
|
||||
pendingTerminalDelivery: null, pendingProviderCleanup: null, warmTransition: null };
|
||||
followUp.remoteExecute.mockResolvedValueOnce({ exitCode: 0, timedOut: false,
|
||||
stdout: Buffer.from(JSON.stringify(runnerState)).toString("base64"), stderr: "" });
|
||||
expect(await followUp.start("resumed")).toBe(ADMITTED);
|
||||
expect(JSON.parse(await readFile(path, "utf8")).identity).toEqual(latest.identity);
|
||||
expect(JSON.parse(await readFile(join(root, "plugin-runner-state.json"), "utf8"))).toMatchObject(latest.identity);
|
||||
expect(followUp.remoteExecute).toHaveBeenCalledOnce();
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(root)).toEqual([]);
|
||||
expect(await quarantined()).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
it("preserves a same-source marker with a different turn identity before any remote read", async () => {
|
||||
const fixture = await staleProjection("codex", true), { latest, followUp, root, path } = fixture;
|
||||
const marker = JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1",
|
||||
identity: { ...latest.identity, itemId: "foreign-item" } });
|
||||
await writeFile(path, marker);
|
||||
expect(await followUp.start("resumed")).toBe("native_runner_checkpoint_projection_held:prior_identity");
|
||||
expect(await readFile(path, "utf8")).toBe(marker);
|
||||
expect(followUp.remoteExecute).not.toHaveBeenCalled();
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(root)).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
it.each(["codex", "opencode"] as const)("rehydrates the current %s run marker from its own checkpoint instead of an earlier turn", async provider => {
|
||||
const prior = await seedRecoveredPriorRun("settled", { provider }), followUp = await admitFollowUp(prior);
|
||||
const identity = { runId: followUp.runId, runnerInstanceId: followUp.native.runnerInstanceId,
|
||||
environmentLeaseId: followUp.native.environmentLeaseId, normalizedSessionId: followUp.native.normalizedSessionId,
|
||||
turnId: followUp.native.turnId, itemId: followUp.native.itemId };
|
||||
const root = scopedRunnerdStateRoot(prior.executionFor(followUp.runId));
|
||||
const owner = { companyId, issueId: prior.issueId, runId: followUp.runId, authorityRunId: followUp.runId,
|
||||
bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() };
|
||||
await db.update(nativeRunFinalizations).set({ controllerBootId: owner.bootId, controllerGeneration: 1,
|
||||
leaseOwner: owner.leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, followUp.runId));
|
||||
const controller = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), identity,
|
||||
expectedRunnerVersion: "current-marker-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
try {
|
||||
const commands = [controller.queueCommand("run.prepare"), controller.queueCommand("session.open")];
|
||||
controller.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, owner));
|
||||
await controller.flushCheckpoint();
|
||||
const path = join(root, "control-plane", "control-plane-state.json");
|
||||
expect(JSON.parse(await readFile(path, "utf8")).schema).toBe("paperclip.runner.external-control-plane-state.v1");
|
||||
await mkdir(join(root, "runner"), { recursive: true });
|
||||
await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({
|
||||
schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended" }));
|
||||
expect(await followUp.start()).toBe(ADMITTED);
|
||||
const restored = JSON.parse(await readFile(path, "utf8"));
|
||||
expect(restored.identity).toEqual(identity);
|
||||
expect(restored.commands.map((command: { commandId: string }) => command.commandId)).toEqual(commands.map(command => command.commandId));
|
||||
expect(await archivedOwners(root)).toEqual([]);
|
||||
expect(await quarantined()).toEqual([]);
|
||||
} finally { await controller.stop(); }
|
||||
}, 60_000);
|
||||
|
||||
it("never substitutes an older checkpoint when the current run marker has no checkpoint", async () => {
|
||||
const prior = await seedRecoveredPriorRun(), followUp = await admitFollowUp(prior);
|
||||
const root = scopedRunnerdStateRoot(prior.executionFor(followUp.runId)), path = join(root, "control-plane", "control-plane-state.json");
|
||||
await mkdir(join(root, "control-plane"), { recursive: true });
|
||||
const marker = JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: {
|
||||
runId: followUp.runId, runnerInstanceId: followUp.native.runnerInstanceId, environmentLeaseId: followUp.native.environmentLeaseId,
|
||||
normalizedSessionId: followUp.native.normalizedSessionId, turnId: followUp.native.turnId, itemId: followUp.native.itemId,
|
||||
} });
|
||||
await writeFile(path, marker);
|
||||
expect(await followUp.start()).toBe("native_runner_checkpoint_missing");
|
||||
expect(await readFile(path, "utf8")).toBe(marker);
|
||||
expect(await archivedOwners(root)).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
it.each(["unsettled journal", "old owner held", "selected owner held", "foreign scope", "selected parts missing", "selected parts tampered"] as const)(
|
||||
"preserves the stale projection and newer backup with %s", async scenario => {
|
||||
const fixture = await staleProjection(), { prior, latest, followUp, path, root } = fixture;
|
||||
if (scenario === "unsettled journal") {
|
||||
const state = JSON.parse(fixture.originalBytes);
|
||||
state.commands[0].status = "pending";
|
||||
await writeFile(path, JSON.stringify(state));
|
||||
}
|
||||
if (scenario === "old owner held" || scenario === "selected owner held") {
|
||||
await db.update(nativeRunFinalizations).set({ leaseOwner: "another-owner", leaseExpiresAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, scenario === "old owner held" ? prior.runId : latest.runId));
|
||||
}
|
||||
if (scenario === "foreign scope") {
|
||||
const [row] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, prior.runId));
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { ...row.runnerProfileJson, nativeExecutionInput: {
|
||||
...prior.executionFor(prior.runId), binding: { ...prior.executionFor(prior.runId).binding, executionWorkspaceId: randomUUID() },
|
||||
} } }).where(eq(heartbeatRuns.id, prior.runId));
|
||||
}
|
||||
if (scenario === "selected parts missing") {
|
||||
await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, latest.runId));
|
||||
}
|
||||
if (scenario === "selected parts tampered") {
|
||||
await db.update(nativeRunnerCheckpoints).set({ material: { tampered: true } }).where(eq(nativeRunnerCheckpoints.runId, latest.runId));
|
||||
}
|
||||
const before = await readFile(path, "utf8");
|
||||
expect(await followUp.start()).not.toBe(ADMITTED);
|
||||
expect(await readFile(path, "utf8")).toBe(before);
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(root)).toEqual([]);
|
||||
expect(await quarantined()).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
it("never replaces a newer local projection with the earlier selected checkpoint", async () => {
|
||||
const fixture = await staleProjection();
|
||||
await db.update(heartbeatRuns).set({ createdAt: new Date(Date.now() + 60_000) })
|
||||
.where(eq(heartbeatRuns.id, fixture.prior.runId));
|
||||
const before = await readFile(fixture.path, "utf8");
|
||||
expect(await fixture.followUp.start()).toBe("native_controller_recovery_missing_evidence:newer_progress_barrier");
|
||||
expect(await readFile(fixture.path, "utf8")).toBe(before);
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(fixture.root)).toEqual([]);
|
||||
}, 60_000);
|
||||
|
||||
it.each([false, true])("does not overwrite a current owner while waiting for its coordinator lock (external=%s)", async external => {
|
||||
const fixture = await staleProjection("codex", external);
|
||||
let release!: () => void, locked!: () => void;
|
||||
const hold = new Promise<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const blocker = db.transaction(async tx => {
|
||||
await lockSourceRecoveryCoordinators(tx, companyId, [fixture.latest.runId]);
|
||||
locked();
|
||||
await hold;
|
||||
});
|
||||
await acquired;
|
||||
const pending = fixture.followUp.start();
|
||||
try {
|
||||
let waiting = false;
|
||||
for (let attempt = 0; attempt < 200 && !waiting; attempt++) {
|
||||
const rows = await db.execute<{ waiting: boolean }>(sql`select exists (
|
||||
select 1 from pg_stat_activity where datname = current_database() and wait_event_type = 'Lock'
|
||||
and query like '%native_run_finalizations%' and query like '%for update%'
|
||||
) as waiting`);
|
||||
waiting = rows[0]?.waiting === true;
|
||||
if (!waiting) await new Promise(resolve => setTimeout(resolve, 10));
|
||||
}
|
||||
expect(waiting).toBe(true);
|
||||
const state = JSON.parse(fixture.originalBytes);
|
||||
state.identity.runId = fixture.followUp.runId;
|
||||
await writeFile(fixture.path, JSON.stringify(state));
|
||||
const currentBytes = await readFile(fixture.path, "utf8");
|
||||
release();
|
||||
await blocker;
|
||||
expect(await pending).toBe("native_runner_checkpoint_projection_held:journal_changed");
|
||||
expect(await readFile(fixture.path, "utf8")).toBe(currentBytes);
|
||||
await fixture.assertBackup();
|
||||
expect(await archivedOwners(fixture.root)).toEqual([]);
|
||||
} finally {
|
||||
release();
|
||||
await blocker;
|
||||
await pending;
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
it.each(["created", "replacement"] as const)(
|
||||
"admits a fresh runner authority when the recovered owner is provably finished: %s sandbox", async outcome => {
|
||||
const prior = await seedRecoveredPriorRun();
|
||||
|
||||
@@ -139,7 +139,7 @@ import {
|
||||
} from "node:fs";
|
||||
import { hostname, tmpdir, userInfo } from "node:os";
|
||||
import { DatabaseSync } from "node:sqlite";
|
||||
import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES, loadNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader, loadNativeUnclaimedBootstrapCheckpoint, nativeRunnerCheckpointPersistence, nativeRunnerCheckpointRegistration, restoreNativeRunnerCheckpoint } from "./native-runner-checkpoint.js";
|
||||
import { REMOTE_CONTROLLER_LEASE_MS, REMOTE_RUNNER_RECONNECT_GRACE_MS } from "./native-runner-recovery-contract.js";
|
||||
import type { NativeTaskSessionPublicationContext } from "./native-task-session-publication.js";
|
||||
import { basename, dirname, join, posix, resolve } from "node:path";
|
||||
@@ -234,6 +234,7 @@ import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { appendNativeCredentialUpdate } from "./native-credential-updates.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import { lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
|
||||
import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
|
||||
import { createAssignedMcpTools, getAssignedMcpGateway } from "./assigned-mcp-tools.js";
|
||||
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
|
||||
@@ -5377,7 +5378,7 @@ type SettledRunnerdProjection = { bytes: string; refused?: never } | { bytes?: n
|
||||
const LOCAL_RUNNER_LIFECYCLE_COMMANDS = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]);
|
||||
|
||||
function readSettledRunnerdProjection(root: string, identity: Record<string, string>,
|
||||
options: { localRunner?: boolean } = {}): SettledRunnerdProjection {
|
||||
options: { localRunner?: boolean; controlBytes?: string } = {}): SettledRunnerdProjection {
|
||||
const refuse = (refused: string): SettledRunnerdProjection => ({ refused });
|
||||
try {
|
||||
if (!isSafeNativeStateDirectory(root)) return refuse("root_unsafe");
|
||||
@@ -5392,7 +5393,7 @@ function readSettledRunnerdProjection(root: string, identity: Record<string, str
|
||||
const held = unknown.filter(name => NATIVE_SESSION_ROOT_HELD_ENTRIES.includes(name));
|
||||
if (held.length > 0) return refuse(`held_root_entry(${describeUnexpectedNativeSessionRootEntries(held)})`);
|
||||
if (unknown.length > 0) return refuse(`unexpected_root_entry(${describeUnexpectedNativeSessionRootEntries(unknown)})`);
|
||||
const bytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"),
|
||||
const bytes = options.controlBytes ?? readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"),
|
||||
NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8");
|
||||
const control = record(JSON.parse(bytes));
|
||||
const commands = (Array.isArray(control.commands) ? control.commands : []).map(record);
|
||||
@@ -5465,18 +5466,20 @@ function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value:
|
||||
* recorded provider processes are proven stopped by exact PID and start time,
|
||||
* the claim has no owner left and the same retirement applies.
|
||||
*/
|
||||
async function retireFinishedPerTurnRunnerdOwner(input: {
|
||||
async function inspectFinishedPerTurnRunnerdOwner(input: {
|
||||
db: Db;
|
||||
root: string;
|
||||
identity: RunnerdDurableIdentity;
|
||||
execution: NativeExecutionInput;
|
||||
projectedControlBytes?: string;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
}): Promise<{ retired: true } | { retired: false; refused: string }> {
|
||||
}): Promise<{ finished: true; run: typeof heartbeatRuns.$inferSelect; recheck: () => Promise<string | null> }
|
||||
| { finished: false; refused: string }> {
|
||||
const target = input.runnerExecutionTarget;
|
||||
// A resumed sandbox still holds the prior Runner's own lifecycle, which its
|
||||
// existing recovery path reads. Plugin tasks own a separate saved checkpoint.
|
||||
const refuse = (refused: string) => ({ retired: false as const, refused });
|
||||
const refuse = (refused: string) => ({ finished: false as const, refused });
|
||||
if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn");
|
||||
const localRunner = !target || target.kind === "local";
|
||||
if (!localRunner) {
|
||||
@@ -5493,7 +5496,13 @@ async function retireFinishedPerTurnRunnerdOwner(input: {
|
||||
if (typeof value !== "string" || !value) return refuse("durable_identity_incomplete");
|
||||
identity[key] = value;
|
||||
}
|
||||
const projection = readSettledRunnerdProjection(input.root, identity, { localRunner });
|
||||
const statePath = resolve(input.root, "control-plane", "control-plane-state.json");
|
||||
if (!isSafeNativeStateDirectory(resolve(input.root, "control-plane"))) return refuse("journal_unreadable");
|
||||
let originalBytes: string;
|
||||
try { originalBytes = readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8"); }
|
||||
catch { return refuse("journal_unreadable"); }
|
||||
const projectionOptions = { localRunner, controlBytes: input.projectedControlBytes };
|
||||
const projection = readSettledRunnerdProjection(input.root, identity, projectionOptions);
|
||||
if (projection.refused !== undefined) return refuse(projection.refused);
|
||||
const controlBytes = projection.bytes;
|
||||
const { binding } = input.execution;
|
||||
@@ -5560,10 +5569,23 @@ async function retireFinishedPerTurnRunnerdOwner(input: {
|
||||
if (copy.refused !== undefined) return refuse(`quarantine_${copy.refused}`);
|
||||
}
|
||||
}
|
||||
// Database reads can await. Retire only the exact evidence proven above.
|
||||
if (readSettledRunnerdProjection(input.root, identity, { localRunner }).bytes !== controlBytes) return refuse("journal_changed");
|
||||
if (provenFinishedHere && !finishedHere()) return refuse("controller_restarted_here");
|
||||
if (localRunner && !(await proveLocalRunnerStopped()).stopped) return refuse("runner_process_restarted");
|
||||
const recheck = async (): Promise<string | null> => {
|
||||
// Database reads can await. Retire only the exact evidence proven above.
|
||||
if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "runner_durable_identity_too_large").toString("utf8") !== originalBytes
|
||||
|| readSettledRunnerdProjection(input.root, identity, projectionOptions).bytes !== controlBytes) return "journal_changed";
|
||||
if (provenFinishedHere && !finishedHere()) return "controller_restarted_here";
|
||||
if (localRunner && !(await proveLocalRunnerStopped()).stopped) return "runner_process_restarted";
|
||||
return null;
|
||||
};
|
||||
return { finished: true, run: prior.run, recheck };
|
||||
}
|
||||
|
||||
async function retireFinishedPerTurnRunnerdOwner(input: Parameters<typeof inspectFinishedPerTurnRunnerdOwner>[0])
|
||||
: Promise<{ retired: true } | { retired: false; refused: string }> {
|
||||
const proof = await inspectFinishedPerTurnRunnerdOwner(input);
|
||||
if (!proof.finished) return { retired: false, refused: proof.refused };
|
||||
const changed = await proof.recheck();
|
||||
if (changed) return { retired: false, refused: changed };
|
||||
const archivedAt = Date.now();
|
||||
const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`);
|
||||
mkdirSync(archiveRoot, { recursive: true, mode: 0o700 });
|
||||
@@ -5571,7 +5593,7 @@ async function retireFinishedPerTurnRunnerdOwner(input: {
|
||||
moveNativeContinuityState(input.root, archiveRoot);
|
||||
await input.onLog?.(
|
||||
"stdout",
|
||||
`[paperclip-runner] Run ${identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`,
|
||||
`[paperclip-runner] Run ${input.identity.runId} finished without a saved runner checkpoint; starting a new runner for this conversation.\n`,
|
||||
);
|
||||
return { retired: true };
|
||||
}
|
||||
@@ -6543,18 +6565,128 @@ async function restorePerTurnRunnerdController(input: {
|
||||
const identity = readRunnerdDurableIdentity(root);
|
||||
const reference = identity ?? readRunnerdDurableIdentity(root, true);
|
||||
if (!reference) {
|
||||
if (existsSync(resolve(root, "control-plane", "control-plane-state.json"))) return;
|
||||
if (!await restoreNativeControllerProjections(input.db, input.execution, root)) return;
|
||||
} else {
|
||||
if (!durableIdentityMatchesSession(reference, input.execution)) return;
|
||||
if (identity?.runId === input.execution.binding.runId) return;
|
||||
const restored = await restoreNativeRunnerCheckpoint(input.db,
|
||||
{ companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId },
|
||||
root, nativeSessionKey(input.execution));
|
||||
if (!restored && !identity) throw new Error("native_runner_checkpoint_missing");
|
||||
const referenceBytes = readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"),
|
||||
NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8");
|
||||
if (reference.runId === input.execution.binding.runId) {
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(input.db, input.execution.binding);
|
||||
if (!checkpoint) throw new Error("native_runner_checkpoint_missing");
|
||||
const scratch = mkdtempSync(join(tmpdir(), "paperclip-current-controller-"));
|
||||
try {
|
||||
checkpoint.restoreSessionDirectory(scratch);
|
||||
const saved = readRunnerdDurableIdentity(scratch);
|
||||
if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"]
|
||||
.some(key => saved?.[key] !== reference[key])) throw new Error("native_runner_checkpoint_binding_mismatch");
|
||||
if (readBoundedNativeFile(resolve(root, "control-plane", "control-plane-state.json"),
|
||||
NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) {
|
||||
throw new Error("native_runner_checkpoint_projection_held:journal_changed");
|
||||
}
|
||||
checkpoint.restoreSessionDirectory(root);
|
||||
} finally { rmSync(scratch, { recursive: true, force: true }); }
|
||||
return;
|
||||
}
|
||||
const recovered = await resolveNativeControllerRecovery(input.db, {
|
||||
...input.execution.binding, currentRunId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution),
|
||||
});
|
||||
if (recovered.kind === "recoverable") {
|
||||
await refreshFinishedPerTurnRunnerdProjection(input, root, reference, recovered.checkpoint.runId, !identity, referenceBytes);
|
||||
} else {
|
||||
if (recovered.kind === "waiting_for_owner") throw new NativeControllerRecoveryWaitingError(recovered.deadline);
|
||||
if (recovered.kind !== "unsupported" && recovered.kind !== "no_prior_execution") {
|
||||
throw new Error(`native_controller_recovery_${recovered.kind}:${"reason" in recovered ? recovered.reason : "prior_reference"}`);
|
||||
}
|
||||
const restored = await restoreNativeRunnerCheckpoint(input.db,
|
||||
{ companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId },
|
||||
root, nativeSessionKey(input.execution));
|
||||
if (!restored && !identity) throw new Error("native_runner_checkpoint_missing");
|
||||
}
|
||||
}
|
||||
await restoreRemoteSuspendedRunnerCheckpoint(input, root);
|
||||
}
|
||||
|
||||
async function refreshFinishedPerTurnRunnerdProjection(
|
||||
input: Parameters<typeof restorePerTurnRunnerdController>[0], root: string, reference: RunnerdDurableIdentity, sourceRunId: string, external: boolean, referenceBytes: string,
|
||||
): Promise<void> {
|
||||
const refuse = (reason: string): never => { throw new Error(`native_runner_checkpoint_projection_held:${reason}`); };
|
||||
await input.db.transaction(async tx => {
|
||||
await lockSourceRecoveryCoordinators(tx, input.execution.binding.companyId,
|
||||
[reference.runId, sourceRunId, input.execution.binding.runId]);
|
||||
const recovered = await resolveNativeControllerRecovery(tx as unknown as Db, {
|
||||
...input.execution.binding, currentRunId: input.execution.binding.runId, normalizedSessionId: nativeSessionKey(input.execution),
|
||||
});
|
||||
if (recovered.kind !== "recoverable" || recovered.checkpoint.runId !== sourceRunId) return refuse("source_changed");
|
||||
const scratch = mkdtempSync(join(tmpdir(), "paperclip-controller-projection-"));
|
||||
const sourceRoot = resolve(scratch, "source"), oldRoot = resolve(scratch, "prior");
|
||||
try {
|
||||
recovered.checkpoint.restoreSessionDirectory(sourceRoot);
|
||||
const sourceIdentity = readRunnerdDurableIdentity(sourceRoot);
|
||||
if (!durableIdentityMatchesSession(sourceIdentity, input.execution) || sourceIdentity.runId !== sourceRunId) return refuse("source_identity");
|
||||
const statePath = resolve(root, "control-plane", "control-plane-state.json");
|
||||
const sourceBytes = readBoundedNativeFile(resolve(sourceRoot, "control-plane", "control-plane-state.json"),
|
||||
NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8");
|
||||
const originalBytes = readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8");
|
||||
if (originalBytes !== referenceBytes) return refuse("journal_changed");
|
||||
if (!external && sourceRunId === reference.runId && originalBytes === sourceBytes) return;
|
||||
let projectedControlBytes: string | undefined;
|
||||
if (external) {
|
||||
const marker = record(JSON.parse(originalBytes));
|
||||
if (marker.schema !== "paperclip.runner.external-control-plane-state.v1" ||
|
||||
["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"]
|
||||
.some(key => record(marker.identity)[key] !== reference[key])) return refuse("prior_identity");
|
||||
if (sourceRunId === reference.runId) {
|
||||
if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"]
|
||||
.some(key => sourceIdentity[key] !== reference[key])) return refuse("prior_identity");
|
||||
if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) {
|
||||
return refuse("journal_changed");
|
||||
}
|
||||
recovered.checkpoint.restoreSessionDirectory(root);
|
||||
return;
|
||||
}
|
||||
const previous = await loadNativeRunnerCheckpoint(tx as unknown as Db, { ...input.execution.binding, runId: reference.runId });
|
||||
if (!previous) return refuse("prior_checkpoint_missing");
|
||||
previous.restoreSessionDirectory(oldRoot);
|
||||
const previousIdentity = readRunnerdDurableIdentity(oldRoot);
|
||||
if (["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"]
|
||||
.some(key => previousIdentity?.[key] !== reference[key])) return refuse("prior_identity");
|
||||
projectedControlBytes = readBoundedNativeFile(resolve(oldRoot, "control-plane", "control-plane-state.json"),
|
||||
NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8");
|
||||
}
|
||||
const oldProof = await inspectFinishedPerTurnRunnerdOwner({ ...input, db: tx as unknown as Db, root, identity: reference, projectedControlBytes });
|
||||
if (!oldProof.finished) return refuse(oldProof.refused);
|
||||
const sourceProof = await inspectFinishedPerTurnRunnerdOwner({ ...input, db: tx as unknown as Db, root: sourceRoot, identity: sourceIdentity });
|
||||
if (!sourceProof.finished) return refuse(`source_${sourceProof.refused}`);
|
||||
if (reference.runId !== sourceRunId && oldProof.run.createdAt >= sourceProof.run.createdAt) return refuse("source_not_newer");
|
||||
for (const run of [oldProof.run, sourceProof.run]) {
|
||||
let previous: NativeExecutionInput;
|
||||
try { previous = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput); }
|
||||
catch { return refuse("source_scope"); }
|
||||
if (previous.binding.runId !== run.id || nativeSessionScopeKey(previous) !== nativeSessionScopeKey(input.execution)) {
|
||||
return refuse("source_scope");
|
||||
}
|
||||
}
|
||||
const sourceChanged = await sourceProof.recheck(), oldChanged = await oldProof.recheck();
|
||||
if (sourceChanged || oldChanged) return refuse(sourceChanged ?? oldChanged!);
|
||||
if (readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") !== referenceBytes) {
|
||||
return refuse("journal_changed");
|
||||
}
|
||||
const archiveParent = resolve(root, "continuity-breaks");
|
||||
mkdirSync(archiveParent, { recursive: true, mode: 0o700 });
|
||||
if (!isSafeNativeStateDirectory(archiveParent)) return refuse("archive_unsafe");
|
||||
const archivedAt = Date.now(), archiveRoot = resolve(archiveParent, `${archivedAt}-${randomUUID()}`);
|
||||
mkdirSync(archiveRoot, { mode: 0o700 });
|
||||
renameSync(resolve(root, "control-plane"), resolve(archiveRoot, "control-plane"));
|
||||
recovered.checkpoint.restoreSessionDirectory(root);
|
||||
noteNativeRunnerEvidence(archivedAt);
|
||||
} finally {
|
||||
rmSync(scratch, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** The controller checkpoint and Runner lifecycle have separate authorities.
|
||||
* Rehydrate a missing suspension receipt before choosing live warm adoption. */
|
||||
async function restoreRemoteSuspendedRunnerCheckpoint(input: {
|
||||
|
||||
Reference in new issue
Block a user