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);