From 25f1070fbdabd96b26cdf199e624a77905f3ca37 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 28 Oct 2026 09:09:09 -0500 Subject: [PATCH] fix(runner): preserve checkpoint proof across warm preparation (#18011) ## Thinking Path > - Paperclip coordinates agent work and preserves its execution state. > - Native Runner checkpoints preserve authority across controller changes. > - Warm sessions also need checkpoint writes while their successor prepares. > - External persistence replaces the local state file with a marker. > - Two existing checks treated those valid states as lost ownership or prior activity. > - This PR uses the existing preparation lease and canonical controller state to keep those checks accurate. > - The result is safe warm-session retirement and fresh-session startup on reused sandboxes. ## Linked Issues or Issue Description Related: #17439 and #17320. The open #17401 changes the recovery welcome window and does not cover these paths. **What happened?** A configuration change could fail before the next provider turn with `runner did not durably suspend before checkpoint`. The task already pointed to its preparing successor, but that successor did not yet have a native coordinator. The old completed runner's maintenance checkpoint was rejected. A separate fresh session on a reused sandbox could fail with `resumed_unavailable`. Checkpoint registration had replaced the full local state file with an external-persistence marker. The bootstrap check still required the old full-file format. Process supervision then reported unconfirmed termination before launch. **Expected behavior** Retire the completed warm runner under the same process's live preparation lease. Admit a fresh remote session only when the registered controller proves exact identity, no prior activity and pending bootstrap commands, and the remote session directory can be claimed atomically. **Steps to reproduce** 1. Complete a native warm turn with external checkpoint persistence. 2. Change configuration so instruction preparation must retire that session, then submit a follow-up. 3. Separately, admit a fresh native session on a reused sandbox lease after external checkpoint registration. **Paperclip version or commit** Reproduced in the checkpoint implementation introduced by #17439. Fixed against current master. **Deployment mode** Source checkout tests and a hosted Docker/Daytona incident. No deployment or original-task retry is part of this PR. ## What Changed - Accept completed-run maintenance under the exact successor's live, same-process preparation lease only before a native coordinator exists. - Preserve original controller generation, accepted-result, no-new-work, company and task checks. - Pass the registered controller through the existing connection callback and read its current canonical state for untouched-session proof. - Keep marker-only, identity, activity, backup and atomic remote-directory guards. - Add regressions and update execution semantics and session verification recipes. ## Verification - Warm retirement red: the real Postgres writer rejected the preparing successor's `runner.suspend` checkpoint with `native_runner_checkpoint_owner_lost`. - Warm retirement green: 14 Postgres ownership cases pass, including stale/foreign/expired owners, conflicting native authority and new provider work. - Bootstrap red: real controller registration and external persistence caused the resumed and replacement positives to fail. Six external negative cases passed. - Bootstrap green: all 34 bootstrap cases, the bootstrap-state predicate and 13 connection-driver cases pass (48 total). - `node scripts/check-maps.mjs` and `git diff --check` pass. - Full `pnpm -r typecheck` and `pnpm build` pass. - Full local `pnpm test:run` remains partial/interrupted. The default attempt could not initialize embedded Postgres on this Mac. A dedicated Postgres 18 attempt ran for 17 minutes without a completed file result before interruption; the generic helper separately confirmed unsupported embedded Postgres (`initdb` failed after five attempts) and ignores the external database URL. The generic probe skipped one test. Setup and interrupted-run logs are retained; neither is claimed as a behavioral red or a full local pass. - Real source-hosted Codex/Daytona acceptance was attempted at `6ce7fcacc4` with `gpt-6.1-sol`, `xhigh`. The actual task stopped before provider work at `runner_ingress_auth_failed`, with zero authenticated connections and zero source events. A separate environment-qualification handshake reached challenge/response; that does not prove task authentication. The three-turn warm reuse, configuration retirement, useful output and reload workflow remains unverified, so this PR stays draft. - The live attempt used the cached Linux amd64 Runner binary `7437a4e32a099180889be534fcfb2a086046514490c1448aba46c55848983ed8` from the immutable Daytona image. The current staging binary could not be copied because SSH timed out, and the configured current EXE image returned registry 403. No compatible current local or CI-downloadable artifact was available. Earlier setup failures and cleanup receipts are retained; no provider success is claimed. Disposable test resources were cleaned up. - Exact-head Greptile completed at 5/5 with no open feedback: https://github.com/paperclipai/paperclip/pull/18011#issuecomment-6469515952. All current-head CI checks pass: 53 successful checks and two intended Storybook skips. ## Risks The preparation fallback affects maintenance checkpoints only after an accepted result. An existing native coordinator prevents that fallback. Stale controller generations, expired or foreign preparation leases, wrong tasks and new provider work remain denied. The bootstrap check retains the live controller object and reads current state each time. It does not trust a marker or restore local state beneath a live controller. Pending turns, prior events/connections, backups and existing remote directories retain recovery gates. No schema, provider binary, credential, permission or lifecycle-policy change is required. ## Model Used OpenAI GPT-6 through Codex, with reasoning, tool use, code execution and collaborating agents. This session does not expose a more specific 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 --- doc/execution-semantics.md | 15 ++++ maps/agent-access-map/runs-adapters.md | 12 +++ maps/feature-map/runs-adapters.md | 12 +++ .../native-runner-checkpoint.ts | 14 +++- .../native-runner-execution-driver.test.ts | 2 +- .../native-runner-execution-driver.ts | 4 +- ...unner-restart-recovery.integration.test.ts | 75 +++++++++++++++++++ .../native-session-executor.test.ts | 49 ++++++++++-- .../native-runtime/native-session-executor.ts | 37 +++++---- 9 files changed, 189 insertions(+), 31 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index b25ba1962f6..2ceb6be9c74 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2073,6 +2073,14 @@ configuration, and controller-owned credential or instruction resources use verified suspension and saved-history resume. This recovery can restart a provider; it does not suspend every healthy warm turn. +Changing configuration can retire a completed warm runner during instruction +preparation, before the next native coordinator is claimed. Its maintenance +checkpoints may use the next run's live preparation lease from the same +controller process, bound to the same company and task. The original controller +generation and committed result must still match. This permits drain and +suspension; it grants no new provider work. Expired or foreign preparation +owners retain the existing recovery hold. + Plugin sandbox reuse during acquisition uses the configured startup timeout or the provider's declared acquisition default, with the host RPC margin. Resume attempts and transient retries share that deadline. Providers without either @@ -2089,6 +2097,13 @@ parents. Existing session files, prior activity, missing proof or any saved backup retain the verified-restore requirement. Workspace replacement alone never authorizes a provider-session reset. +With external controller checkpoints, the local state file is only a marker. +Fresh-session bootstrap checks use the registered controller's current canonical +state. Marker identity alone cannot prove an untouched session. The same exact +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. + 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 diff --git a/maps/agent-access-map/runs-adapters.md b/maps/agent-access-map/runs-adapters.md index b4dd19ddae9..1d0cfb31992 100644 --- a/maps/agent-access-map/runs-adapters.md +++ b/maps/agent-access-map/runs-adapters.md @@ -271,6 +271,18 @@ session-control access. Deployed diagnostic usefulness remains unverified. ## `session` +Scoped source evidence: completed warm-runner retirement can persist maintenance +under the same process's live, task-bound preparation lease before native +coordinator admission. Original controller generation, committed-result and +no-new-work checks remain required. This changes no actor permission or session +reset authority. Follow the [session recipe](../feature-map/runs-adapters.md#session) +for configuration invalidation and expired or foreign owner rejection. + +Fresh-session admission with external checkpoints reads current canonical state +from the registered controller. A local marker does not grant bootstrap or +session-reset authority. Exact run/session/runner/lease binding, no prior activity, +backup checks and atomic remote directory admission retain their existing gates. + Capability ID: `runs-adapters/session`. Review: `baseline`. diff --git a/maps/feature-map/runs-adapters.md b/maps/feature-map/runs-adapters.md index 25744903964..e6bade34f1d 100644 --- a/maps/feature-map/runs-adapters.md +++ b/maps/feature-map/runs-adapters.md @@ -74,6 +74,13 @@ mutations. Qualify a useful model task separately from saving the account. ### `session` +After a completed warm turn, change an agent setting that invalidates its +instruction copy, then send a follow-up. Verify the old runner suspends during +preparation, saved files remain available, and the follow-up completes once. +Repeat with an expired or foreign preparation owner and verify that execution +stays held. The checkpoint integration regression covers the ownership boundary; +a real provider run remains separate acceptance evidence. + For a repository project that inherits storage-plugin defaults, run a task and send two follow-ups. Start with unset task workspace settings, then repeat with an explicit isolated mode or strategy. Binding must save `reuse_existing` and @@ -95,6 +102,11 @@ reads the retained file and appends one new line. Verify a replacement sandbox, a new provider session, the exact file contents, successful finalization and no stranded task lock. Send another follow-up and verify ordinary session reuse. +With external controller checkpoints enabled, also admit a fresh session on a +reused sandbox lease. Verify startup from canonical controller evidence and +reject a marker without a registered controller, changed identity or prior turn +activity. An existing remote session directory must retain its recovery gate. + Repeat without resetting the session: its replacement must restore verified history or remain blocked. Missing backup state must not silently start an existing session again. The [executor controls](../../server/src/services/native-runtime/native-session-executor.test.ts) diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index 743c2ae8191..2a50fca9fec 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -1,7 +1,8 @@ -import { and, eq, exists, gt, isNull, lte, inArray, or, sql } from "drizzle-orm"; +import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm"; import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; import { decodeDurablePrpCheckpoint, type DurablePrpCheckpoint, type DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js"; +import { legacyControllerBootId } from "../legacy-controller-lease.js"; type Binding = { companyId: string; issueId: string; runId: string }; type Owner = Binding & { bootId: string; leaseOwner: string; generation: number; authorityRunId: string }; @@ -54,7 +55,16 @@ export function nativeRunnerCheckpointWriter(db: Db, owner: Owner) { eq(nativeRunFinalizations.runId, task.executionRunId), eq(nativeRunFinalizations.companyId, owner.companyId), eq(nativeRunFinalizations.issueId, owner.issueId), eq(nativeRunFinalizations.controllerBootId, owner.bootId), gt(nativeRunFinalizations.leaseExpiresAt, sql`clock_timestamp()`))) : []; - if (!current.resultId || newWork || !task || (task.executionRunId && task.executionRunId !== owner.runId && !successor)) { + const [preparingSuccessor] = !successor && task?.executionRunId && task.executionRunId !== owner.runId + ? await tx.select({ runId: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, task.executionRunId), eq(heartbeatRuns.companyId, owner.companyId), + eq(heartbeatRuns.issueId, owner.issueId), eq(heartbeatRuns.runtimeMode, "legacy"), + eq(heartbeatRuns.status, "running"), eq(heartbeatRuns.executionStage, "preparing"), + eq(heartbeatRuns.controllerBootId, legacyControllerBootId), + gt(heartbeatRuns.controllerLeaseExpiresAt, sql`clock_timestamp()`), + notExists(tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, heartbeatRuns.id))))) : []; + if (!current.resultId || newWork || !task || (task.executionRunId && task.executionRunId !== owner.runId && !successor && !preparingSuccessor)) { throw new Error("native_runner_checkpoint_owner_lost"); } } diff --git a/server/src/services/native-runtime/native-runner-execution-driver.test.ts b/server/src/services/native-runtime/native-runner-execution-driver.test.ts index 4958984c22e..28bc0b34db1 100644 --- a/server/src/services/native-runtime/native-runner-execution-driver.test.ts +++ b/server/src/services/native-runtime/native-runner-execution-driver.test.ts @@ -158,7 +158,7 @@ describe("shared Runner execution driver", () => { const identity = { runId: "run-b" } as Parameters["controlPlaneRegistration"]>>[1]; const driver = createNativeRunnerExecutionDriver({ target: { kind: "local" }, binding, beforeConnect }); await driver.controlPlaneRegistration!(authority, identity); - expect(beforeConnect).toHaveBeenCalledWith(identity); + expect(beforeConnect).toHaveBeenCalledWith(identity, authority); expect(transport.register).toHaveBeenCalledWith({ ...binding, runId: "run-b", authority }); expect(beforeConnect.mock.invocationCallOrder[0]).toBeLessThan(transport.register.mock.invocationCallOrder[0]); }); diff --git a/server/src/services/native-runtime/native-runner-execution-driver.ts b/server/src/services/native-runtime/native-runner-execution-driver.ts index 835dc7096e7..6bf3b44ded5 100644 --- a/server/src/services/native-runtime/native-runner-execution-driver.ts +++ b/server/src/services/native-runtime/native-runner-execution-driver.ts @@ -22,7 +22,7 @@ export interface NativeRunnerExecutionDriverInput { mapCaBundle?: (sourcePath: string) => string | Promise; processStarted?: Promise; processIdentityReady?: Promise; - beforeConnect?: (identity?: RunnerIdentity) => Promise; + beforeConnect?: (identity: RunnerIdentity, authority: Parameters[0]) => Promise; controllerPersistence?: { db: Db; bootId: string; authorityRunId?: string | (() => string); onFailure?: (runId: string, leaseOwner: string) => Promise }; checkpoint?: Awaited>["checkpoint"]; process?: Omit; @@ -93,7 +93,7 @@ export function createNativeRunnerExecutionDriver(input: NativeRunnerExecutionDr return { ...input.process, controlPlaneRegistration: async (authority, attachmentIdentity) => { - await input.beforeConnect?.(attachmentIdentity); + await input.beforeConnect?.(attachmentIdentity, authority); if (input.controllerPersistence && input.binding.issueId) { await nativeRunnerCheckpointRegistration({ ...input.controllerPersistence, binding: { companyId: input.binding.companyId, issueId: input.binding.issueId, runId: input.binding.runId }, diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 88b7fcb4610..2dafecf3828 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -72,6 +72,7 @@ import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "./native-runner-recovery-contr import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; import { RUNNER_BRIDGE_READ_OPERATIONS, RUNNER_BRIDGE_SCHEMAS } from "./runner-bridge-tools.js"; import { HttpError } from "../../errors.js"; +import { legacyControllerBootId } from "../legacy-controller-lease.js"; const mockCaptureRunFailure = vi.hoisted(() => vi.fn()); vi.mock("../../sentry.js", async () => { @@ -1255,6 +1256,80 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( } finally { await authority.stop(); } }); + it.each(["idle", "preparing_successor", "unclaimed_successor", "claimed_successor", "expired_preparation", "dispatching_successor", "foreign_preparation", "foreign_native_preparation", "wrong_issue", "new_work", "replaced_controller"] as const)( + "settles a completed warm checkpoint during early retirement without granting work: %s", + async scenario => { + const fixture = await seedRun(`WARM-EARLY-RETIREMENT-${scenario}`); + const binding = { companyId, issueId: fixture.issueId, runId: fixture.runId }; + const identity = transportOptions(fixture, runtimeRoot).prpIdentity; + const authority = new DurablePrpControlPlane({ + stateDirectory: resolve(runtimeRoot, fixture.runId, "early-retirement", "control-plane"), + identity, expectedRunnerVersion: "warm-retirement-fixture", + expectedRunnerDigest: `sha256:${"a".repeat(64)}`, + }); + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: successor.bootId, + controllerGeneration: 1, leaseOwner: "completed-warm-owner", leaseExpiresAt: new Date(Date.now() + 60_000), + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + try { + await registerNativeRunnerCheckpoint({ db: fixture.db, binding, bootId: successor.bootId, authority }); + const [run] = await fixture.db.select({ completionContractId: heartbeatRuns.completionContractId }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + if (!run?.completionContractId) throw new Error("Prepared warm run has no completion contract"); + const resultId = randomUUID(); + await fixture.db.insert(nativeRunResults).values({ + id: resultId, ...binding, turnId: identity.turnId, completionContractId: run.completionContractId, + serverFingerprint: `completed-warm-${scenario}`, schemaStatus: "accepted", + resultJson: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "Completed warm turn." }, + canonicalSha256: "a".repeat(64), + }); + await fixture.db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ phase: "committed", resultId, + leaseOwner: null, leaseExpiresAt: null, + ...(scenario === "replaced_controller" ? { controllerBootId: randomUUID(), controllerGeneration: 2 } : {}), + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + + let nextRunId: string | null = null; + if (scenario !== "idle") { + nextRunId = randomUUID(); + const preparing = scenario !== "claimed_successor" && scenario !== "unclaimed_successor"; + const otherIssue = scenario === "wrong_issue" ? (await seedRun("WARM-OTHER-ISSUE")).issueId : fixture.issueId; + await fixture.db.insert(heartbeatRuns).values({ id: nextRunId, companyId, agentId, + issueId: otherIssue, scopeKind: "issue", contextSnapshot: { issueId: otherIssue }, + ...(preparing ? { runtimeMode: "legacy", executionStage: scenario === "dispatching_successor" ? "dispatching" : "preparing", + controllerBootId: scenario === "foreign_preparation" ? randomUUID() : legacyControllerBootId, + controllerLeaseExpiresAt: scenario === "expired_preparation" ? new Date(0) : new Date(Date.now() + 60_000), + ...(scenario === "foreign_native_preparation" ? { nativeIssueId: fixture.issueId } : {}), + } : { runtimeMode: "native", nativeIssueId: fixture.issueId, nativeSessionId: randomUUID() }), + status: "running", invocationSource: "manual" }); + if (!preparing || scenario === "foreign_native_preparation") await fixture.db.insert(nativeRunFinalizations).values({ runId: nextRunId, companyId, issueId: fixture.issueId, + phase: "observed", ...(scenario === "claimed_successor" || scenario === "foreign_native_preparation" ? { + controllerBootId: scenario === "foreign_native_preparation" ? randomUUID() : successor.bootId, + controllerGeneration: 1, leaseOwner: "next-warm-owner", leaseExpiresAt: new Date(Date.now() + 60_000) } : {}), + }); + } + await fixture.db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, fixture.issueId)); + const before = (await loadNativeRunnerCheckpoint(fixture.db, binding))!; + authority.queueCommand(scenario === "new_work" ? "turn.start" : "runner.suspend", {}, `early-retirement-${scenario}`); + if (!["idle", "preparing_successor", "claimed_successor"].includes(scenario)) { + await expect(authority.flushCheckpoint()).rejects.toThrow("native_runner_checkpoint_owner_lost"); + expect((await loadNativeRunnerCheckpoint(fixture.db, binding))?.records).toEqual(before.records); + } else { + await expect(authority.flushCheckpoint()).resolves.toBeUndefined(); + const saved = (await loadNativeRunnerCheckpoint(fixture.db, binding))!; + const restoredRoot = resolve(runtimeRoot, fixture.runId, "retirement-proof"); + saved.restoreSessionDirectory(restoredRoot); + const restored = JSON.parse(await readFile(resolve(restoredRoot, "control-plane", "control-plane-state.json"), "utf8")); + expect(restored.identity).toEqual(identity); + expect(restored.commands).toContainEqual(expect.objectContaining({ type: "runner.suspend", status: "pending" })); + } + const [task] = await fixture.db.select({ executionRunId: issues.executionRunId }).from(issues) + .where(eq(issues.id, fixture.issueId)); + expect(task.executionRunId).toBe(nextRunId); + } finally { await authority.stop(); } + }, + ); + it("applies checkpoint deltas in Postgres without rewriting authority and removes only evicted events", async () => { const fixture = await seedRun("CHECKPOINT-DELTA"); const root = resolve(runtimeRoot, fixture.runId); diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 5b73257786d..97dd4685495 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -7,7 +7,7 @@ import { environmentTaskOperationSchema } from "@paperclipai/plugin-sdk"; import { trace } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { configuredEnvironment, configuredEnvironmentProjection } from "../../vendor/paperclip-runner/index.js"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; import { readFileSync } from "node:fs"; import { PgDialect } from "drizzle-orm/pg-core"; import type { SQL } from "drizzle-orm"; @@ -14565,7 +14565,7 @@ describe("runnerd provider runtime wiring", () => { }); it.each(["resumed", "replacement"].flatMap((outcome) => - ["fresh", "existing_state", "symlink_parent", "wrong_identity", "wrong_runner", "wrong_lease", "missing_control_plane", "connected", "committed_event", "committed_command", "pending_turn", "remote_probe_failed", "backup_present"] + ["fresh", "external_checkpoint", "external_pending_turn", "external_wrong_identity", "external_unregistered", "existing_state", "symlink_parent", "wrong_identity", "wrong_runner", "wrong_lease", "missing_control_plane", "connected", "committed_event", "committed_command", "pending_turn", "remote_probe_failed", "backup_present"] .map((scenario) => ({ outcome, scenario }))))( "bootstraps only an untouched provider session in a $outcome workspace lease: $scenario", async ({ outcome, scenario }) => { const remoteCwd = join(isolatedStateDirectory, "remote"); @@ -14605,14 +14605,20 @@ describe("runnerd provider runtime wiring", () => { const runtimeContext = nativeRuntimeContextFixture(); const executionWithContext = { ...execution, runtimeContext }; const onTraceEvent = vi.fn(async (_event: AdapterRuntimeEvent) => undefined); + const controllerArtifact = join(isolatedStateDirectory, "bootstrap-runnerd"); + if (scenario.startsWith("external_")) await writeFile(controllerArtifact, "fixture bootstrap artifact"); const backend = await createRunnerdBackend({ db: leaseDb(execution), execution: executionWithContext, runnerInstanceId: "runner-new-in-retained-workspace", + ...(scenario.startsWith("external_") ? { runnerRemoteBinaryPath: controllerArtifact, runnerRemoteCodexPath: controllerArtifact } : {}), trace: createNativeRunTrace({ runId: execution.binding.runId, onEvent: onTraceEvent }), runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd, environmentId: "environment", leaseId: "lease-resumed", providerKey: "daytona", effectiveCapabilities: { runnerWebSocketIngress: true }, + getRunnerIngressEndpoint: async () => ({ kind: "authenticated_websocket", websocketUrl: "wss://sandbox.invalid/runner", + secretHeaders: [{ name: "Authorization", value: "fixture-ingress-secret" }], generation: "1", + refresh: vi.fn(), close: vi.fn(async () => undefined) }), sandboxLeaseAcquisition: { outcome, providerLeaseId: "sandbox-acquired", reason: outcome === "replacement" ? "not_found" : undefined }, runner: { execute: remoteExecute, syncIn }, } as never, @@ -14622,9 +14628,39 @@ describe("runnerd provider runtime wiring", () => { const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { prepareExternalRunnerState: () => Promise; runnerProcessLauncher: ReturnType; + controlPlaneRegistration: NonNullable; }; - await mkdir(join(options.stateDirectory!, "control-plane"), { recursive: true }); - await writeFile(join(options.stateDirectory!, "control-plane", "control-plane-state.json"), JSON.stringify({ + let controller: DurablePrpControlPlane | undefined; + let registration: Awaited> | undefined; + onTestFinished(async () => { + await registration?.release(); + await controller?.stop(); + }); + if (scenario.startsWith("external_")) { + controller = new DurablePrpControlPlane({ stateDirectory: join(options.stateDirectory!, "control-plane"), + identity: { ...options.prpIdentity!, ...(scenario === "external_wrong_identity" ? { runId: "other-run" } : {}) } as ConstructorParameters[0]["identity"], + expectedRunnerVersion: "bootstrap-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + controller.queueCommand("run.prepare", {}); + controller.queueCommand("session.open", { reuse: "same_session" }); + if (scenario !== "external_unregistered") registration = await options.controlPlaneRegistration(controller); + syncIn.mockClear(); + remoteExecute.mockClear(); + const persisted = new Map(); + controller.setCheckpointPersistence(async checkpoint => { + persisted.clear(); + for (const [part, value] of checkpoint.records) persisted.set(part, value); + }); + await controller.flushCheckpoint(); + expect(persisted.has("header")).toBe(true); + expect(JSON.parse(await readFile(join(options.stateDirectory!, "control-plane", "control-plane-state.json"), "utf8")).schema) + .toBe("paperclip.runner.external-control-plane-state.v1"); + if (scenario === "external_pending_turn") { + controller.queueCommand("turn.start", {}); + await controller.flushCheckpoint(); + } + } else { + await mkdir(join(options.stateDirectory!, "control-plane"), { recursive: true }); + await writeFile(join(options.stateDirectory!, "control-plane", "control-plane-state.json"), JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v1", identity: { ...options.prpIdentity, ...(scenario === "wrong_identity" ? { runId: "other-run" } : {}), @@ -14634,7 +14670,8 @@ describe("runnerd provider runtime wiring", () => { committedEvents: scenario === "committed_event" ? [{ type: "session.opened" }] : [], commands: [{ type: "run.prepare", status: scenario === "committed_command" ? "committed" : "pending" }, { type: scenario === "pending_turn" ? "turn.start" : "session.open", status: "pending" }], - })); + })); + } if (scenario === "missing_control_plane") { await rm(join(options.stateDirectory!, "control-plane", "control-plane-state.json")); } @@ -14642,7 +14679,7 @@ describe("runnerd provider runtime wiring", () => { await mkdir(join(options.stateDirectory!, "failover-backups", "current"), { recursive: true }); await writeFile(join(options.stateDirectory!, "failover-backups", "current", "manifest.json"), "{}"); } - if (scenario === "fresh") { + if (scenario === "fresh" || scenario === "external_checkpoint") { await expect(options.prepareExternalRunnerState()).resolves.toBeUndefined(); expect(onTraceEvent).toHaveBeenCalledWith(expect.objectContaining({ payload: expect.objectContaining({ diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6a534ec8c20..5deac9f947c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -97,6 +97,7 @@ import type { } from "../../adapters/index.js"; import { usdToUnits, type NativeFinalizationResult } from "@paperclipai/shared"; import type { + DurablePrpControlPlane, HarnessRuntimeRequestResolution, NativeExecutionInput, NativeRuntimeContextSnapshot, @@ -5470,6 +5471,15 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean { } } +function runnerdStateIsIncompleteBootstrap(state: Record): boolean { + const commands = Array.isArray(state.commands) ? state.commands.map(record) : []; + const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : []; + return state.schema === RUNNERD_CONTROL_PLANE_STATE_SCHEMA && + state.connectionCount === 0 && committedEvents.length === 0 && + commands.every(command => command.status === "pending" && + (command.type === "run.prepare" || command.type === "session.open")); +} + export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { try { const statePath = resolve( @@ -5486,23 +5496,7 @@ export function runnerdStateProvesIncompleteBootstrap(root: string): boolean { ).toString("utf8"), ), ); - const commands = Array.isArray(state.commands) - ? state.commands.map(record) - : []; - const committedEvents = Array.isArray(state.committedEvents) - ? state.committedEvents - : []; - const onlyUnconsumedBootstrapCommands = commands.every( - (command) => - command.status === "pending" && - (command.type === "run.prepare" || command.type === "session.open"), - ); - return ( - state.schema === RUNNERD_CONTROL_PLANE_STATE_SCHEMA && - state.connectionCount === 0 && - committedEvents.length === 0 && - onlyUnconsumedBootstrapCommands - ); + return runnerdStateIsIncompleteBootstrap(state); } catch { return false; } @@ -11938,6 +11932,7 @@ async function createRunnerdBackendWithinSessionClaim( const target = input.runnerExecutionTarget ?? { kind: "local" as const }; const remoteTarget = target.kind === "remote" ? target : null; const pluginTask = remoteTarget?.transport === "sandbox" ? remoteTarget.runnerTask : undefined; + let runnerControlPlane: DurablePrpControlPlane | undefined; if (pluginTask && input.runnerIngressAuthorized !== true) throw new Error("runner_ingress_unavailable"); if (pluginTask && (input.execution.schema === "paperclip.native-execution-input.v6" || input.execution.schema === "paperclip.native-execution-input.v8")) { throw new Error("Plugin Runner tasks require a workspace session"); @@ -12783,11 +12778,12 @@ async function createRunnerdBackendWithinSessionClaim( const claimUntouchedRemoteSession = async (): Promise => { if (!remoteCommandRunner || !remoteRuntimeRoot || !remoteSessionRoot) return false; - const identity = readRunnerdDurableIdentity(root); + const controllerState = runnerControlPlane?.store.state; + const identity = controllerState ? record(controllerState.identity) : readRunnerdDurableIdentity(root); if (!durableIdentityMatchesExecution(identity, input.execution) || identity?.runnerInstanceId !== effectiveRunnerInstanceId || identity?.environmentLeaseId !== effectiveEnvironmentLeaseId || - !runnerdStateProvesIncompleteBootstrap(root)) return false; + !(controllerState ? runnerdStateIsIncompleteBootstrap(record(controllerState)) : runnerdStateProvesIncompleteBootstrap(root))) return false; // Workspace reuse or replacement can acquire a lease for a new session. // Claim this exact new session atomically under readable real directories. // Missing files inside an existing session never authorize a fresh start. @@ -13899,7 +13895,8 @@ async function createRunnerdBackendWithinSessionClaim( remoteCaBundleMapping = { sourcePath, targetPath }; return targetPath; }, - beforeConnect: async () => { + beforeConnect: async (_identity, authority) => { + runnerControlPlane = authority; await input.remoteWarmRecovery?.revalidate(); if (retainedTransition && recoveryPending) { const current = await verifyWarmTransitionRestart(input);