diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index 18fdcd5bced..60d821e9f0f 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -14,6 +14,14 @@ Recovery can restart a provider to refresh its tool contract; process reuse is required for ordinary healthy follow-ups. Each recovery case includes another healthy turn after the recovered turn and asserts that it does not start another runner or provider process. +The same-server case also starts preparation lease renewal during recovery and +holds instruction collection open past a renewal tick after coordinator release. +The executor must acquire renewal before preparation releases its reference, then +stop renewal before releasing the coordinator. Completion must not detach the +retained runner or strand the next turn. +Buffered transport messages and close notifications can arrive synchronously on +subscription. The controller must register the connection before subscribing so +completed message processing can drain and the suspended session can retire. It does not qualify a live model or a sandbox provider API. Per-turn plugin continuation uses a stopped checkpoint on a replacement lease. diff --git a/maps/agent-access-map/runs-adapters.md b/maps/agent-access-map/runs-adapters.md index 84b1e806d77..10d603fe63f 100644 --- a/maps/agent-access-map/runs-adapters.md +++ b/maps/agent-access-map/runs-adapters.md @@ -286,6 +286,13 @@ session-control access. Deployed diagnostic usefulness remains unverified. ## `session` +Scoped source evidence: restart preparation transfers its lease renewal to the +native executor after the executor acquires its own reference. Completion stops +renewal before coordinator release, while real ownership loss retains the detach +guard. This changes no actor permission or session-reset authority. Follow the +[session recipe](../feature-map/runs-adapters.md#session) for recovery followed by +ordinary turns; deterministic runner checks do not establish live provider access. + Plugin suspension receipts remain required through recovery and admission, then are removed when the admitted session becomes live. A later warm run therefore does not inherit an obsolete stopped-run identity. Repository-free fresh plugin diff --git a/maps/feature-map/runs-adapters.md b/maps/feature-map/runs-adapters.md index 21470054436..ac99cd084ca 100644 --- a/maps/feature-map/runs-adapters.md +++ b/maps/feature-map/runs-adapters.md @@ -85,6 +85,15 @@ mutations. Qualify a useful model task separately from saving the account. ### `session` +Recover an in-flight remote runner after controller restart, then let it finish +with slow instruction collection. Send two follow-ups without resetting its +session. Verify that completion does not detach the retained runner and that +provider history survives with one turn per wake. Preparation must retain its +lease until execution owns renewal; a genuine ownership loss must still detach. +The [remote continuity regression](../../doc/testing-native-remote-continuity.md) +uses a real Linux runner and deterministic provider. Live provider acceptance is +a separate check. + For a plugin computer, complete a task, let its warm session suspend, then send two follow-ups within the next warm window. Verify the first resumes the saved session and the second keeps the same provider history without an identity diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts index cf85e3bcde7..b9f982401d2 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.test.ts @@ -2438,6 +2438,31 @@ it.each([false, true])( }, ); +it.each(["message", "close"] as const)("settles a wire that delivers buffered %s during subscription", async buffered => { + const root = mkdtempSync(resolve(tmpdir(), "runner-buffered-subscription-")); + const core = new DurablePrpControlPlane({ stateDirectory: root, identity, expectedRunnerVersion, expectedRunnerDigest }); + let closed = () => {}; + try { + core.attachWireConnection({ + onJson: listener => { if (buffered === "message") listener({ invalid: "buffered frame" }); }, + onClose: listener => { + closed = () => listener({ message: "closed" }); + if (buffered === "close") closed(); + }, + close: () => closed(), + sendJson: () => {}, + }); + await new Promise(resolveTurn => setImmediate(resolveTurn)); + // The invalid frame closes its wire. A completed callback must not leave a + // ghost connection or a settled promise that makes drain spin forever. + expect(() => core.retireSemanticToolCallbacks()).not.toThrow(); + await core.drainPendingConnectionProcessing(); + } finally { + await core.stop(); + rmSync(root, { recursive: true, force: true }); + } +}); + it("refuses a processing-drain absence claim while local or remote ingress remains admitted", async () => { const root = mkdtempSync( resolve(tmpdir(), "runner-processing-drain-ingress-"), diff --git a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts index 0903e3c1207..89dd0138fbd 100644 --- a/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts +++ b/packages/paperclip-runner/src/control-plane/durable-prp-control-plane.ts @@ -1715,6 +1715,7 @@ class AuthorityConnection { readonly wire: PrpWireConnection; #closed = false; #onClose: () => void; + #onJson: (value: unknown) => void; constructor(input: { wire: PrpWireConnection; @@ -1723,8 +1724,12 @@ class AuthorityConnection { }) { this.wire = input.wire; this.#onClose = input.onClose; - this.wire.onJson(input.onJson); + this.#onJson = input.onJson; + } + + start(): void { this.wire.onClose(() => this.#markClosed()); + if (!this.#closed) this.wire.onJson(this.#onJson); } sendJson(value: unknown): void { @@ -2558,6 +2563,9 @@ export class DurablePrpControlPlane { }, }); this.#connections.add(connection); + // A wire can synchronously deliver buffered data or close on subscription. + // Both callbacks must see the registered connection and its exact identity. + connection.start(); return { isAuthenticated: () => connection.secureChannel !== null, }; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index c40c6215a57..34aae3b54a3 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1791,6 +1791,7 @@ export function heartbeatService( runOptions: { nativeLeaseOwner?: string; nativeRestartRecovery?: NativeRestartRecoveryClaim; + onNativeExecutionLeaseAcquired?: () => Promise; } = {}, ) { return withRunExecutionTiming(runId, () => executeRunAttempt(runId, runOptions)); @@ -1798,7 +1799,7 @@ export function heartbeatService( async function executeRunAttempt( runId: string, - runOptions: { nativeLeaseOwner?: string; nativeRestartRecovery?: NativeRestartRecoveryClaim }, + runOptions: { nativeLeaseOwner?: string; nativeRestartRecovery?: NativeRestartRecoveryClaim; onNativeExecutionLeaseAcquired?: () => Promise }, ) { const attemptStartedAtMs = Date.now(); let attestedQuestionResponseAtMs: number | null = null; diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index a98a413ef94..c3a11b83666 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -11,7 +11,8 @@ import { createHeartbeatRunState } from "./run-state.js"; import type { HotRestartIntent } from "../hot-restart.js"; import type { NativeRestartRecoveryClaim, NativeRestartRecoveryDisposition } from "../native-runtime/index.js"; import { NATIVE_OWNERSHIP_UNVERIFIED_ERROR_CODE } from "../native-runtime/native-runner-ownership.js"; -import { claimNativeRestartRecoveries } from "../native-runtime/native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; +import { startNativeSessionExecutionLeaseRenewal } from "../native-runtime/native-execution-lease.js"; import { allowLegacyShutdownWorkspaceCleanup, beginLegacyShutdownWorkspaceSettlement, finishLegacyShutdownWorkspaceSettlement, maintainLegacyShutdownWorkspaceSettlement, legacyShutdownWorkspaceResourceProtected, sweepLegacyShutdownWorkspaceSettlements } from "../legacy-shutdown-workspace-settlement.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "../adapter-execution-control.js"; import { logger } from "../../middleware/logger.js"; @@ -682,6 +683,67 @@ describePostgres("heartbeat recovery module ownership", () => { await vi.waitFor(() => expect(deps.activeRunExecutionPromises.size).toBe(0)); }); + it.each(["completion", "failure_before_dispatch"] as const)("stops preparation renewal after recovered %s", async outcome => { + const run = await seedRun(); + const deps = callbacks(db); + const controller = await currentNativeControllerIdentity(); + const claim: NativeRestartRecoveryClaim = { kind: "reattach_remote_runner", runId: run.id, + leaseOwner: randomUUID(), controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null, + remote: { providerLeaseId: randomUUID(), remoteCwd: "/workspace" } }; + const scope = { companyId: run.companyId, issueId: run.issueId!, controller, remote: true }; + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId }).where(eq(heartbeatRuns.id, run.id)); + const expiresAt = new Date(Date.now() + 60_000); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "observed", attempt: 1, + controllerBootId: controller.bootId, controllerPid: controller.pid, + controllerProcessStartedAt: controller.processStartedAt, + leaseOwner: claim.leaseOwner, leaseExpiresAt: expiresAt }); + vi.useFakeTimers({ toFake: ["setInterval", "clearInterval"] }); + native.claims.mockImplementationOnce(async input => { + await input.onClaim?.(claim, scope); + return [claim]; + }); + let released!: () => void, finish!: () => void; + const leaseReleased = new Promise(resolve => { released = resolve; }); + const finalization = new Promise(resolve => { finish = resolve; }); + deps.executeRun.mockImplementationOnce(async (_runId, options) => { + if (outcome === "failure_before_dispatch") { + released(); + throw new Error("workspace preparation failed"); + } + const execution = startNativeSessionExecutionLeaseRenewal({ db, ...scope, runId: run.id, + leaseOwner: claim.leaseOwner, attempt: claim.providerAttempt, + onLost: () => native.detach() }); + try { + // The executor takes its reference before preparation releases its own. + await options?.onNativeExecutionLeaseAcquired?.(); + await execution.stop(); + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, run.id)); + released(); + // Workspace/result settlement may still be pending after lease release. + await finalization; + } finally { await execution.stop(); } + }); + try { + await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(new Date(), [run.id]); + await leaseReleased; + if (outcome === "failure_before_dispatch") await Promise.all([...deps.activeRunExecutionPromises]); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(21_000); + expect(deps.activeRunExecutionPromises.size).toBe(outcome === "completion" ? 1 : 0); + expect(native.detach).not.toHaveBeenCalled(); + expect(deps.executeRun).toHaveBeenCalledOnce(); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + expect(coordinator.leaseExpiresAt).toEqual(outcome === "completion" ? null : expiresAt); + } finally { + finish(); + await Promise.all([...deps.activeRunExecutionPromises]); + vi.useRealTimers(); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + } + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + it.each(["bootstrap_incomplete", "blocked"] as const)("retries durable remote takeover and preserves its %s disposition", async kind => { const run = await seedRun(); const deps = callbacks(db); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index c5285332d9d..3b180d2679d 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -121,6 +121,7 @@ export interface HeartbeatRecoveryDependencies extends Pick Promise; }) => Promise; scheduleNativeSessionResumeDispatch: (runId: string, nextAttemptAt: Date) => void; cancelHeartbeatNativeRun: (input: { @@ -1052,13 +1053,16 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD const execution = executeRun(claim.runId, { nativeLeaseOwner: claim.leaseOwner, nativeRestartRecovery: claim, + ...(preparationRenewal ? { onNativeExecutionLeaseAcquired: () => preparationRenewal.stop() } : {}), }).catch((error) => { logger.error( { err: error, runId: claim.runId, disposition: claim.kind }, "native restart recovery execution failed", ); }).finally(() => preparationRenewal?.stop().catch(() => undefined)); - preparationRenewals.delete(claim.runId); // Execution now owns this reference. + // The executor releases preparation only after acquiring its own renewal. + // The finalizer above also covers exits before execution can acquire it. + preparationRenewals.delete(claim.runId); activeRunExecutionPromises.add(execution); void execution.finally(() => activeRunExecutionPromises.delete(execution), diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index c080ba515ff..f40166ebd45 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -118,6 +118,16 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = expect(input.instructions.releaseInstructionCopy).toHaveBeenCalledOnce(); } + it("passes preparation renewal ownership to native dispatch", async () => { + const { input } = await fixture(true); + const acquired = vi.fn(async () => undefined); + input.runtime.runOptions.onNativeExecutionLeaseAcquired = acquired; + await executeHeartbeatRuntime(db, input); + expect(executePaperclipNativeSession).toHaveBeenCalledWith(expect.objectContaining({ onExecutionLeaseAcquired: acquired })); + // Preparation is still protected until the native executor claims its lease. + expect(acquired).not.toHaveBeenCalled(); + }); + it.each([false, true])("preserves the dispatch gate's early return and cleanup (native: %s)", async native => { const { input, adapter } = await fixture(native); input.control.dispatchResolvedInteractionContinuationWithAtomicGate = async () => ({ dispatched: false }); diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index c080a24cae8..00430730ba3 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -593,6 +593,7 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe runnerInstanceId: nativeRunnerInstanceId, leaseOwner: runOptions.nativeLeaseOwner, restartRecovery: runOptions.nativeRestartRecovery, + onExecutionLeaseAcquired: runOptions.onNativeExecutionLeaseAcquired, backend: options.nativeSessionBackendFactory?.(nativeExecution), useRunnerd: agent.adapterType === "paperclip_runner", diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index 72e3ec01264..fafaf38af07 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -137,7 +137,7 @@ export interface HeartbeatRuntimeSelectionInput { resolvedConfig: Record; managedAiRuntime: Awaited> | undefined; }; - recovery: { nativeLeaseOwner?: string; nativeRestartRecovery?: NativeRestartRecoveryClaim }; + recovery: { nativeLeaseOwner?: string; nativeRestartRecovery?: NativeRestartRecoveryClaim; onNativeExecutionLeaseAcquired?: () => Promise }; trace: { providerTraceRequested: boolean; providerTraceCapture: { metadata: { id: string } } | null }; issuesSvc: Pick, "addComment">; runtimeEnv: Record; 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 2dafecf3828..3dbcd10ab35 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 @@ -421,12 +421,33 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( const priorProcess = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] }); await fixture.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1) }) .where(eq(nativeRunFinalizations.runId, fixture.runId)); + let preparation: NativeExecutionLeaseRenewal | undefined; const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, - controller: await currentNativeControllerIdentity(), restartKind: "hard", runIds: [fixture.runId] }); + controller: await currentNativeControllerIdentity(), restartKind: "hard", runIds: [fixture.runId], + onClaim: (claim, scope) => { + preparation = startNativeSessionExecutionLeaseRenewal({ db: fixture.db, ...scope, + runId: claim.runId, leaseOwner: claim.leaseOwner, attempt: claim.providerAttempt, + onLost: () => detachNativeSessionsForRestart([claim.runId], claim.leaseOwner) }); + } }); if (!claim || claim.kind !== "reattach_remote_runner") { throw new Error(`same-server claim missing: ${claim && "reason" in claim ? claim.reason : claim?.kind}`); } - const result = await executePaperclipNativeSession({ ...firstOptions, leaseOwner: claim.leaseOwner, restartRecovery: claim }); + const result = await executePaperclipNativeSession({ ...firstOptions, leaseOwner: claim.leaseOwner, restartRecovery: claim, + onExecutionLeaseAcquired: () => preparation!.stop(), + instructionWorkingCopy: { + hasChanges: async () => { + // Completion has released the coordinator, but the startup still + // owns its session. A leftover preparation timer would detach it. + const [settled] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(settled.leaseOwner).toBeNull(); + await new Promise(resolve => setTimeout(resolve, 21_000)); + return false; + }, + collectStopped: async () => undefined, + }, + }).finally(() => preparation?.stop().catch(() => undefined)); + expect(logs.filter(line => line.includes("controller authority detached for restart"))).toHaveLength(1); const recoveredCalls = await runner.execute({ command: "cat", args: [`${remoteCwd}/provider-calls.log`] }); expect(recoveredCalls.stdout.split("\n").filter(line => line === "turn/start")).toHaveLength(1); const nextProcess = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/runner/runner-process.identity`] }); 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 eab9db6547b..db399718a65 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -10373,10 +10373,15 @@ describe("native session bounded recovery", () => { }); const stop = vi.fn(async () => undefined); const renew = vi.spyOn(nativeLeaseRenewal, "startNativeSessionExecutionLeaseRenewal").mockReturnValue({ stop }); + const releasePreparation = vi.fn(async () => { + expect(renew).toHaveBeenCalledOnce(); + expect(stop).not.toHaveBeenCalled(); + }); let reached!: () => void, release!: () => void; const claiming = new Promise(resolve => { reached = resolve; }); const preparation = new Promise(resolve => { release = resolve; }); const running = executePaperclipNativeSession({ db, execution: fresh, runnerInstanceId: "fresh-runner", + onExecutionLeaseAcquired: releasePreparation, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", environmentId: "fresh-environment", leaseId: "fresh-lease" }, onEvent: async event => { if (event.payload?.span === "native.coordinator.claim") { reached(); await preparation; } @@ -10388,6 +10393,7 @@ describe("native session bounded recovery", () => { db, runId: fresh.binding.runId, attempt: 1, remote: true, })); expect(stop).not.toHaveBeenCalled(); + expect(releasePreparation).toHaveBeenCalledOnce(); release(); expect(await running).toBe(failure); expect(stop).toHaveBeenCalledOnce(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 21cc7420294..924d0cf0375 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -7791,6 +7791,8 @@ export async function executePaperclipNativeSession(input: { turnTimeoutMs?: number; leaseOwner?: string; restartRecovery?: NativeRestartRecoveryClaim; + /** Release recovery preparation's renewal after execution owns its reference. */ + onExecutionLeaseAcquired?: () => Promise; onSpawn?: (meta: { pid: number; processGroupId: number | null; @@ -8395,7 +8397,7 @@ async function executePaperclipNativeSessionWithinScope( }) .where(eq(heartbeatRuns.id, coordinator.runId)); return nextAttempt; - }).then(claimedAttempt => { + }).then(async claimedAttempt => { const startup = nativeSessionStartups.get(input.execution.binding.runId); if (!startup) throw new Error("native_session_startup_owner_missing"); startup.leaseOwner = leaseOwner; @@ -8410,6 +8412,7 @@ async function executePaperclipNativeSessionWithinScope( ? detachNativeSessionsForRestart([input.execution.binding.runId], leaseOwner) : cancelNativeSession(input.execution.binding.runId, "native session execution lease lost"), }); + await input.onExecutionLeaseAcquired?.(); return claimedAttempt; }), { parentName: "task.prepare" },