diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 8bc8886b075..b4d63b5687f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1683,6 +1683,18 @@ controller must recover it through the authenticated ownership checks. This preserves active work and queued messages without treating a server restart as user cancellation. +During controller shutdown, already registered provider startups share one +30-second sweep deadline to finish read-only run admission and publish their exact +sessions for detachment. Its company, run, session, runner and completion-contract binding +must still match persisted authority. The draining controller cannot checkpoint, +publish events, submit a turn or settle the run. A new startup after the shutdown +fence, or admission after the window expires, remains rejected. A Stop pending +before or after session publication keeps cancellation and cleanup with the +current owner. Shutdown waits for that owner only until the same sweep deadline; +it does not detach the session or submit a prompt while Stop is pending. An +unsettled owner remains explicitly pending for reconciliation. Its timeout does +not abort detachment of healthy sessions or trigger a global graceful drain. + Recovery restores a structured result proposal from the durable run events when its append survived but the active turn's session checkpoint did not. Advancing the replay cursor must preserve that proposal for the exact run, session, source, 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 ad754e8e84e..bfb1f4f309f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -33,6 +33,7 @@ import { dirname, join } from "node:path"; import { heartbeatRuns, heartbeatRunEvents, + issueThreadInteractions, issues, nativeRunFinalizations, nativeRunResults, @@ -49,6 +50,7 @@ import { validatePrpStructuredRunResult, validatePrpEvent, parseNativeExecutionInput, + executeNativeSession, type NativeExecutionInputV1, type NativeExecutionInputV3, type NativeExecutionInput, @@ -343,6 +345,7 @@ import { executePaperclipNativeSession, detachNativeSessionsForRestart, NativeControllerDetachedForRestartError, + NativeCancellationPendingRecoveryError, getNativeSessionSteeringState, NativeSessionSteeringError, assertRemoteRunnerBuildMetadata, @@ -5695,6 +5698,10 @@ function leaseDb( agentId: boundExecution.binding.agentId, companyId: boundExecution.binding.companyId, nativeIssueId: boundExecution.binding.issueId, + nativeSessionId: boundExecution.session.normalizedSessionId ?? `session-${boundExecution.binding.runId}`, + runnerInstanceId: "runner", + completionContractId: boundExecution.completionContract.id, + completionContractSha256: boundExecution.completionContract.sha256, resultJson: runResultJson, runnerProfileJson, runtimeMode: "native", @@ -5892,6 +5899,306 @@ describe("native startup cancellation fence", () => { }); describe("native startup restart detachment", () => { + it("uses one admission deadline for the whole restart sweep", async () => { + const inputs = ["first", "second"].map(label => { + const input = structuredClone(execution); + input.binding.runId = `restart-shared-deadline-${label}`; + input.binding.agentId = `restart-shared-deadline-agent-${label}`; + return input; + }); + const releases: Array<() => void> = []; + const ready: Array> = []; + state.execute.mockReset(); + for (const _input of inputs) { + let started!: () => void; + ready.push(new Promise(resolve => { started = resolve; })); + const opening = new Promise(resolve => { releases.push(resolve); }); + state.execute.mockImplementationOnce(async () => { + started(); await opening; + throw new Error("startup fixture released"); + }); + } + const outcomes = inputs.map(input => executePaperclipNativeSession({ + db: leaseDb(input), execution: input, runnerInstanceId: "runner", + }).catch(error => error)); + await Promise.all(ready); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const finished = vi.fn(); + const runIds = inputs.map(input => input.binding.runId); + const shutdown = detachNativeSessionsForRestart(runIds).then(finished); + try { + await vi.advanceTimersByTimeAsync(30_000); + expect(finished).toHaveBeenCalledWith({ detachedRunIds: [], inactiveRunIds: [], + unsupportedRunIds: [], pendingRunIds: runIds }); + } finally { + vi.useRealTimers(); releases.forEach(release => release()); + await Promise.all(outcomes); await shutdown; + } + }); + + it.each(["authority", "questions"])("keeps a post-publication Stop owner through %s attachment and restart", async (seam) => { + const restarting = structuredClone(execution); + restarting.binding.runId = `restart-stop-after-publication-${seam}`; + let published!: () => void, releaseRead!: () => void, acknowledgeStop!: () => void; + let cleaning!: () => void, releaseCleanup!: () => void; + const publication = new Promise(resolve => { published = resolve; }); + const read = new Promise(resolve => { releaseRead = resolve; }); + const receipt = new Promise(resolve => { acknowledgeStop = resolve; }); + const cleanupStarted = new Promise(resolve => { cleaning = resolve; }); + const cleanup = new Promise(resolve => { releaseCleanup = resolve; }); + const detach = vi.fn(async () => undefined); + const cancel = vi.fn(() => ({ cleanup: receipt })); + const submit = vi.fn(); + const db = leaseDb(restarting); + const select = db.select.bind(db); + db.select = ((fields?: Record) => { + if (seam === "authority" && fields && Object.keys(fields).length === 2 && "status" in fields && "resultJson" in fields) { + return { from: () => ({ where: () => ({ limit: () => { + published(); + return read.then(() => [{ status: "running", resultJson: {} }]); + } }) }) }; + } + if (seam === "questions" && !fields) { + const query = select(); + return { from: (table: unknown) => table === issueThreadInteractions + ? { where: () => ({ orderBy: () => { published(); return read.then(() => []); } }) } + : query.from(table as never) }; + } + return select(fields as never); + }) as typeof db.select; + state.execute.mockReset().mockImplementationOnce(async options => { + try { + await options.onSession({ detachControllerForRestart: detach, cancel, resolveRuntimeRequest: vi.fn() }); + submit(); + } finally { + cleaning(); + await cleanup; + await options.onSession(null); + } + }); + const outcome = executePaperclipNativeSession({ db, execution: restarting, runnerInstanceId: "runner" }).catch(error => error); + await publication; + const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop after publication"); + const completed = vi.fn(); + const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).then(completed); + try { + releaseRead(); + await new Promise(resolve => setImmediate(resolve)); + expect(cancel).toHaveBeenCalledOnce(); + expect(detach).not.toHaveBeenCalled(); + expect(completed).not.toHaveBeenCalled(); + // Admission must not unwind into generic close while Stop awaits its ACK. + const cleanupObserved = vi.fn(); + void cleanupStarted.then(cleanupObserved); + await new Promise(resolve => setImmediate(resolve)); + expect(cleanupObserved).not.toHaveBeenCalled(); + acknowledgeStop(); + expect(await stopping).toBe(true); + await cleanupStarted; + expect(completed).not.toHaveBeenCalled(); + releaseCleanup(); + await outcome; + await shutdown; + expect(completed).toHaveBeenCalledWith(expect.objectContaining({ + detachedRunIds: [], inactiveRunIds: [restarting.binding.runId], pendingRunIds: [], + })); + expect(detach).not.toHaveBeenCalled(); + expect(submit).not.toHaveBeenCalled(); + } finally { + releaseRead(); acknowledgeStop(); releaseCleanup(); + await stopping; await outcome; await shutdown; + } + }); + + it.each(["startup", "active"])("keeps healthy hot-restart adoption when another %s Stop cleanup exceeds the sweep deadline", async (mode) => { + const { coordinateHeartbeatSchedulerShutdown } = await import("../../shutdown.js"); + const { createHeartbeatRecovery } = await import("../heartbeat/recovery.js"); + const { writeHotRestartIntent, readHotRestartIntent } = await import("../hot-restart.js"); + const home = await mkdtemp(join(tmpdir(), "native-restart-stop-sweep-")); + const previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = home; + await writeHotRestartIntent({ previousServerPid: process.pid, previousServerIdentity: "test-controller" }); + const stoppingExecution = structuredClone(execution); + stoppingExecution.binding.runId = `restart-${mode}-stop-timeout`; + stoppingExecution.session.normalizedSessionId = "d371f9bb-6c46-44c6-9e3a-026fa9be0153"; + const healthyExecution = structuredClone(execution); + healthyExecution.binding.runId = `restart-healthy-with-${mode}-stopping-peer`; + healthyExecution.session.normalizedSessionId = "a8fa0cb8-9cf1-46b0-96f0-78518b15847e"; + healthyExecution.binding.agentId = "healthy-restart-agent"; + let stoppedReady!: () => void, healthyReady!: () => void; + let releaseStopped!: () => void, releaseHealthy!: () => void, opening!: () => void, open!: () => void; + const openingStarted = new Promise(resolve => { opening = resolve; }); + const openingGate = new Promise(resolve => { open = resolve; }); + const stopPublished = new Promise(resolve => { stoppedReady = resolve; }); + const healthyPublished = new Promise(resolve => { healthyReady = resolve; }); + const stoppedLifetime = new Promise(resolve => { releaseStopped = resolve; }); + const healthyLifetime = new Promise(resolve => { releaseHealthy = resolve; }); + const stoppedDetach = vi.fn(async () => undefined); + const healthyDetach = vi.fn(async () => undefined); + state.execute.mockReset() + .mockImplementationOnce(async options => { + opening(); await openingGate; + try { + await options.onSession({ detachControllerForRestart: stoppedDetach, cancel: () => ({ cleanup: Promise.resolve() }) }); + } finally { + stoppedReady(); await stoppedLifetime; + await options.onSession(null); + } + throw new Error("stopped owner settled"); + }) + .mockImplementationOnce(async options => { + await options.onSession({ detachControllerForRestart: healthyDetach }); + healthyReady(); await healthyLifetime; + await options.onSession(null); + throw new Error("healthy controller detached"); + }); + const stoppedOutcome = executePaperclipNativeSession({ db: leaseDb(stoppingExecution), execution: stoppingExecution, runnerInstanceId: "runner" }).catch(error => error); + await openingStarted; + const stopping = mode === "startup" ? cancelNativeSession(stoppingExecution.binding.runId, "Stop while opening") : null; + open(); + await stopPublished; + await stopping; + const healthyOutcome = executePaperclipNativeSession({ db: leaseDb(healthyExecution), execution: healthyExecution, runnerInstanceId: "runner" }).catch(error => error); + await healthyPublished; + if (mode === "active") await cancelNativeSession(stoppingExecution.binding.runId, "operator Stop after active admission"); + const snapshotRows = [stoppingExecution, healthyExecution].map(item => ({ + adapterType: "paperclip_runner", adapterConfig: {}, + run: { id: item.binding.runId, companyId: item.binding.companyId, agentId: item.binding.agentId, + runtimeMode: "native", nativeSessionId: item.session.normalizedSessionId, status: "running", contextSnapshot: {} }, + })); + const snapshotDb = { select: () => ({ from: () => ({ innerJoin: () => ({ where: async () => snapshotRows }) }) }) } as unknown as Db; + const { logger } = await import("../../middleware/logger.js"); + const logged = vi.spyOn(logger, "info").mockImplementation(() => undefined); + const recovery = createHeartbeatRecovery(snapshotDb, { + enterShutdown: vi.fn(), appendRunEvent: vi.fn(async () => undefined), + } as unknown as import("../heartbeat/recovery.js").HeartbeatRecoveryDependencies); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] }); + const shutdown = coordinateHeartbeatSchedulerShutdown({ signal: "SIGTERM", + waitForHeartbeatSchedulerIdle: async () => undefined, + prepareHotRestartShutdown: recovery.prepareHotRestartShutdown, + }); + try { + // Wait for the real snapshot I/O before advancing its bounded sweep. + await vi.waitFor(() => expect(vi.getTimerCount()).toBeGreaterThan(0)); + await vi.advanceTimersByTimeAsync(30_000); + expect(await shutdown).toMatchObject({ hotRestart: { mode: "hot_restart", skipDrain: true }, preparationError: null }); + expect(logged).toHaveBeenCalledWith(expect.objectContaining({ detachedNativeSessions: { + detachedRunIds: [healthyExecution.binding.runId], inactiveRunIds: [], unsupportedRunIds: [], + pendingRunIds: [stoppingExecution.binding.runId], + } }), expect.any(String)); + expect((await readHotRestartIntent())?.shutdownSnapshot?.activeRuns.map(run => run.runId)) + .toEqual([stoppingExecution.binding.runId, healthyExecution.binding.runId]); + expect(stoppedDetach).not.toHaveBeenCalled(); + expect(healthyDetach).toHaveBeenCalledOnce(); + } finally { + vi.useRealTimers(); releaseStopped(); releaseHealthy(); + await stoppedOutcome; await healthyOutcome; await shutdown; + logged.mockRestore(); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousHome; + await rm(home, { recursive: true, force: true }); + } + }); + + it.each(["detach", "expired", "stop"])("bounds read-only startup admission during drain: %s", async (mode) => { + const expired = mode === "expired"; + const stopped = mode === "stop"; + const restarting = structuredClone(execution); + restarting.binding.runId = `restart-real-admission-${mode}`; + restarting.session.normalizedSessionId = "restart-real-admission-session"; + const identity = { ...restarting.binding, sessionId: restarting.session.normalizedSessionId }; + let release!: () => void, started!: () => void; + const opening = new Promise(resolve => { release = resolve; }); + const pending = new Promise(resolve => { started = resolve; }); + const detach = vi.fn(async () => undefined); + const close = vi.fn(async () => undefined); + const startTurn = vi.fn(); + let acknowledgeStop!: () => void; + const stopReceipt = new Promise(resolve => { acknowledgeStop = resolve; }); + const cancel = vi.fn(() => ({ cleanup: stopReceipt })); + const backend = { + descriptor: async () => ({ kind: "mock" as const, name: "startup-drain", version: "1", + capabilities: { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }), + openSession: vi.fn(async () => { + started(); await opening; + return { identity: () => identity, detachControllerForRestart: detach, close, startTurn, cancel, + snapshot: async () => ({ sessionId: identity.sessionId, identity, backendKind: "mock" as const }), + }; + }), + }; + let runtimeError: unknown; + state.execute.mockReset().mockImplementationOnce(async options => { + try { return await executeNativeSession({ ...options, onSessionAdmission: undefined }); } catch (error) { runtimeError = error; throw error; } + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, + runnerInstanceId: "runner", backend: backend as never }).catch(error => error); + await Promise.race([pending, outcome.then(error => { throw runtimeError ?? error; })]); + const stopping = stopped ? cancelNativeSession(restarting.binding.runId, "operator Stop during startup") : null; + if (expired) vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error); + try { + if (expired) { + await vi.advanceTimersByTimeAsync(30_000); + expect(await shutdown).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] }); + vi.useRealTimers(); + } else { + await new Promise(resolve => setImmediate(resolve)); + } + release(); + if (stopped) { + await vi.waitFor(() => expect(cancel).toHaveBeenCalledOnce()); + expect(detach).not.toHaveBeenCalled(); + acknowledgeStop(); + expect(await stopping).toBe(true); + } + expect(await outcome).toBeInstanceOf(NativeControllerDetachedForRestartError); + if (stopped) { + expect(await shutdown).toMatchObject({ detachedRunIds: [], inactiveRunIds: [restarting.binding.runId] }); + expect(detach).not.toHaveBeenCalled(); + } else if (expired) { + expect(detach).not.toHaveBeenCalled(); + expect(close).toHaveBeenCalledWith({ reason: "native control-plane run admission failed" }); + } else { + expect(await shutdown).toMatchObject({ detachedRunIds: [restarting.binding.runId], inactiveRunIds: [] }); + expect(detach).toHaveBeenCalledOnce(); + expect(close).not.toHaveBeenCalledWith({ reason: "native control-plane run admission failed" }); + } + expect(backend.openSession).toHaveBeenCalledOnce(); + expect(startTurn).not.toHaveBeenCalled(); + } finally { vi.useRealTimers(); release(); acknowledgeStop(); await stopping; await outcome; await shutdown; } + }); + it("bounds a pending startup Stop without detaching its cleanup owner", async () => { + const restarting = structuredClone(execution); + restarting.binding.runId = "restart-startup-stop-cleanup-timeout"; + let open!: () => void, started!: () => void, releaseCleanup!: () => void, cleaning!: () => void; + const opening = new Promise(resolve => { open = resolve; }); + const pending = new Promise(resolve => { started = resolve; }); + const cleanup = new Promise(resolve => { releaseCleanup = resolve; }); + const cleanupStarted = new Promise(resolve => { cleaning = resolve; }); + const detach = vi.fn(async () => undefined); + const cancel = vi.fn(() => ({ cleanup: Promise.resolve() })); + state.execute.mockReset().mockImplementationOnce(async options => { + started(); await opening; + try { await options.onSession({ detachControllerForRestart: detach, cancel }); } + finally { cleaning(); await cleanup; } + }); + const outcome = executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" }).catch(error => error); + await pending; + const stopping = cancelNativeSession(restarting.binding.runId, "operator Stop before restart"); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const shutdown = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error); + try { + open(); + await cleanupStarted; + expect(await stopping).toBe(true); + await vi.advanceTimersByTimeAsync(30_000); + expect(await shutdown).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] }); + expect(detach).not.toHaveBeenCalled(); + expect(cancel).toHaveBeenCalledOnce(); + } finally { vi.useRealTimers(); open(); releaseCleanup(); await stopping; await outcome; await shutdown; } + }); + it("waits for a checkpointed governed settlement before relinquishing its controller", async () => { const restarting = structuredClone(execution); restarting.binding.runId = "restart-governed-settlement"; @@ -6033,7 +6340,7 @@ describe("native startup restart detachment", () => { const detached = detachNativeSessionsForRestart([restarting.binding.runId]).catch(error => error); if (exceedDeadline) { await vi.advanceTimersByTimeAsync(30_000); - expect(await detached).toMatchObject({ message: "native_restart_startup_not_ready" }); + expect(await detached).toMatchObject({ pendingRunIds: [restarting.binding.runId], detachedRunIds: [] }); vi.useRealTimers(); } else { release(); @@ -6049,18 +6356,13 @@ describe("native startup restart detachment", () => { } }); - it("remembers shutdown while the session is still opening and detaches its late publication", async () => { + it("rejects an entirely new startup after the run has been fenced for shutdown", async () => { const restarting = structuredClone(execution); - restarting.binding.runId = "restart-during-session-open"; - const detach = vi.fn(async () => undefined); + restarting.binding.runId = "restart-new-startup-after-drain"; await expect(detachNativeSessionsForRestart([restarting.binding.runId])).resolves.toMatchObject({ inactiveRunIds: [restarting.binding.runId] }); - state.execute.mockReset().mockImplementationOnce(async (options) => { - await options.onSession({ detachControllerForRestart: detach }); - expect(detach).toHaveBeenCalledOnce(); - await options.onSession(null); - throw new Error("detachment closed the old event stream"); - }); + state.execute.mockReset(); await expect(executePaperclipNativeSession({ db: leaseDb(restarting), execution: restarting, runnerInstanceId: "runner" })).rejects.toBeInstanceOf(NativeControllerDetachedForRestartError); + expect(state.execute).not.toHaveBeenCalled(); }); }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index cc03fcbfa56..ccf020f577b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -227,6 +227,7 @@ import { type ActiveNativeSession = { session: NativeSession; cancelRequested: boolean; + cancellationSettled?: Promise; restartDetach?: Promise; }; @@ -264,6 +265,7 @@ type NativeSessionStartup = { promise: Promise; resolve: (session: ActiveNativeSession | null) => void; stopRequested?: boolean; + restartAdmissionPending?: boolean; cancellationSettled?: Promise; governedWait?: boolean; settled: Promise; @@ -272,13 +274,14 @@ type NativeSessionStartup = { const nativeSessionStartups = new Map(); function detachActiveNativeSessionForRestart(active: ActiveNativeSession) { - active.restartDetach ??= Promise.resolve().then(() => active.session.detachControllerForRestart!()); + active.restartDetach ??= active.session.detachControllerForRestart!(); return active.restartDetach; } async function waitForNativeSessionStartup( runId: string, timeoutError: () => Error = () => new Error("native_restart_startup_not_ready"), + timeoutMs = 30_000, ) { const startup = nativeSessionStartups.get(runId); if (!startup) return null; @@ -287,7 +290,7 @@ async function waitForNativeSessionStartup( return await Promise.race([ startup.promise, new Promise((_resolve, reject) => { - timer = setTimeout(() => reject(timeoutError()), 30_000); + timer = setTimeout(() => reject(timeoutError()), timeoutMs); }), ]); } finally { @@ -301,10 +304,13 @@ export async function detachNativeSessionsForRestart( detachedRunIds: string[]; inactiveRunIds: string[]; unsupportedRunIds: string[]; + pendingRunIds: string[]; }> { const detachedRunIds: string[] = []; const inactiveRunIds: string[] = []; const unsupportedRunIds: string[] = []; + const pendingRunIds: string[] = []; + const restartDeadline = Date.now() + 30_000; const settlementDeadline = Date.now() + 20_000; for (const runId of new Set(runIds)) { // A governed stop already revoked new work and is collecting terminal @@ -320,11 +326,48 @@ export async function detachNativeSessionsForRestart( } finally { clearTimeout(timer); } } nativeRunsDetachingForRestart.add(runId); - const active = activeNativeSessions.get(runId) ?? await waitForNativeSessionStartup(runId); + let active = activeNativeSessions.get(runId); + if (!active) { + // Only a startup registered before this fence may complete read-only + // admission. It must publish its exact session for detachment; writes and + // new turns remain revoked, including after the bounded wait expires. + const startup = nativeSessionStartups.get(runId); + if (startup && Date.now() >= restartDeadline) { + pendingRunIds.push(runId); + continue; + } + if (startup) startup.restartAdmissionPending = true; + try { + active = await waitForNativeSessionStartup(runId, undefined, Math.max(0, restartDeadline - Date.now())) ?? undefined; + } catch (error) { + if (!(error instanceof Error) || error.message !== "native_restart_startup_not_ready") throw error; + pendingRunIds.push(runId); + continue; + } finally { + if (startup) startup.restartAdmissionPending = false; + } + } if (!active) { inactiveRunIds.push(runId); continue; } + const stopping = nativeSessionStartups.get(runId); + if (stopping?.stopRequested || active.cancelRequested) { + // Stop owns cancellation and final cleanup, including after publication. + // A deadline leaves this run pending; it must not abort healthy adoption + // or claim that the transport/cleanup owner has settled. + let timer: ReturnType | undefined; + try { + const settled = stopping && await Promise.race([ + stopping.settled.then(() => true), + new Promise(resolve => { + timer = setTimeout(() => resolve(false), Math.max(0, restartDeadline - Date.now())); + }), + ]); + (settled ? inactiveRunIds : pendingRunIds).push(runId); + } finally { clearTimeout(timer); } + continue; + } if (active.session.detachControllerForRestart === undefined) { unsupportedRunIds.push(runId); continue; @@ -332,7 +375,7 @@ export async function detachNativeSessionsForRestart( await detachActiveNativeSessionForRestart(active); detachedRunIds.push(runId); } - return { detachedRunIds, inactiveRunIds, unsupportedRunIds }; + return { detachedRunIds, inactiveRunIds, unsupportedRunIds, pendingRunIds }; } const MAX_REMOTE_CHECKPOINT_ARCHIVE_BYTES = 64 * 1024 * 1024; const MAX_REMOTE_CHECKPOINT_EXPANDED_BYTES = 64 * 1024 * 1024; @@ -7230,6 +7273,7 @@ export async function cancelNativeSession( // Do not acknowledge a no-op Stop and let that owner submit a turn afterwards. const startup = runStop && !activeNativeSessions.has(runId) ? nativeSessionStartups.get(runId) : undefined; let settleStartupCancellation: (() => void) | undefined; + let settleActiveCancellation: (() => void) | undefined; if (startup) { startup.stopRequested = true; startup.cancellationSettled = new Promise(resolve => { settleStartupCancellation = resolve; }); @@ -7242,6 +7286,7 @@ export async function cancelNativeSession( dispatched = true; if (!active.cancelRequested) { active.cancelRequested = true; + active.cancellationSettled = new Promise(resolve => { settleActiveCancellation = resolve; }); try { if (active.session.cancel) { const cancellationAbort = new AbortController(); @@ -7451,6 +7496,7 @@ export async function cancelNativeSession( return { dispatched, decision, decisionId, auditId }; } finally { settleStartupCancellation?.(); + settleActiveCancellation?.(); } } @@ -7674,6 +7720,9 @@ export async function executePaperclipNativeSession(input: { } } const runId = input.execution.binding.runId; + if (nativeRunsDetachingForRestart.has(runId)) { + throw new NativeControllerDetachedForRestartError(); + } if (nativeSessionStartups.has(runId)) { throw new Error("native_session_supervisor_busy"); } @@ -8362,6 +8411,12 @@ async function executePaperclipNativeSessionWithinScope( throw new NativeControllerDetachedForRestartError(); } }; + const startupForAdmission = nativeSessionStartups.get(input.execution.binding.runId); + const assertRunAdmissionActive = () => { + if (startupForAdmission?.restartAdmissionPending + && nativeSessionStartups.get(input.execution.binding.runId) === startupForAdmission) return; + assertControllerActive(); + }; const controlPlane = new PaperclipControlPlanePort( input.db, { @@ -8378,6 +8433,7 @@ async function executePaperclipNativeSessionWithinScope( { privateKeyPem: input.runnerEnvironment?.PAPERCLIP_AGENT_PRIVATE_KEY, assertControllerActive, + assertRunAdmissionActive, onCommittedEvent: async (event) => { const textKind = !firstTokenRecorded ? observeDeliverableText(event) : null; if (textKind) { @@ -9220,6 +9276,13 @@ async function executePaperclipNativeSessionWithinScope( eq(heartbeatRuns.agentId, input.execution.binding.agentId), )).limit(1); const cancellation = record(currentRun?.resultJson?.nativeCancellation); + if (active.cancelRequested) { + // A Stop can acquire the published handle while the authority + // read above is in flight. Keep its ACK/cleanup owner before + // unwinding admission; neither detach nor generic close wins. + await active.cancellationSettled; + throw new NativeCancellationPendingRecoveryError(); + } if (!currentRun || currentRun.status !== "running" || currentRun.resultJson?.startupCancellation || (cancellation.scope === "run" && @@ -9229,11 +9292,15 @@ async function executePaperclipNativeSessionWithinScope( // callback fails; no provider turn may follow publication. throw new NativeCancellationPendingRecoveryError(); } - if (session.resolveRuntimeRequest) await liveQuestions.attach(); if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { if (session.detachControllerForRestart) await detachActiveNativeSessionForRestart(active); + throw new NativeControllerDetachedForRestartError(); + } + if (session.resolveRuntimeRequest) await liveQuestions.attach(); + if (active.cancelRequested) { + await active.cancellationSettled; + throw new NativeCancellationPendingRecoveryError(); } - if (active.cancelRequested) throw new NativeCancellationPendingRecoveryError(); } else { liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId); diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts index 5321a764f8b..f8b9eb1d022 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.test.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.test.ts @@ -300,6 +300,38 @@ describe("PaperclipControlPlanePort conformance", () => { } }); + it("permits only read-only admission while an existing startup drains", async () => { + const identity = { ...CONTROL_PLANE_CONFORMANCE_OPEN.identity, runId: randomUUID(), sessionId: randomUUID() }; + const runnerId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: identity.runId, companyId: identity.companyId, agentId: identity.agentId, + status: "running", runtimeMode: "native", nativeIssueId: identity.issueId, + nativeSessionId: identity.sessionId, runnerInstanceId: runnerId, + completionContractId: contractId, completionContractSha256: contractSha, + }); + let awaitingStartup = true; + const port = new PaperclipControlPlanePort(db, { + ...identity, completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: runnerId, controlPlaneSourceInstanceId: `draining-${identity.runId}`, + }, { + assertControllerActive: () => { throw new Error("controller_detached"); }, + assertRunAdmissionActive: () => { if (!awaitingStartup) throw new Error("controller_detached"); }, + }); + const open = { identity, backendKind: "mock" as const, sourceInstanceId: runnerId }; + await expect(port.openRun({ ...open, sourceInstanceId: randomUUID() })).rejects.toThrow("native_open_run_binding_mismatch"); + await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, identity.runId)); + await expect(port.openRun(open)).rejects.toThrow("native_open_run_not_authorized"); + await db.update(heartbeatRuns).set({ runnerInstanceId: runnerId }).where(eq(heartbeatRuns.id, identity.runId)); + await expect(port.openRun(open)).resolves.toBeUndefined(); + await expect(port.checkpointSession({ backendKind: "mock", sessionId: identity.sessionId, identity })).rejects.toThrow("controller_detached"); + await expect(port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL })).rejects.toThrow("controller_detached"); + awaitingStartup = false; + await expect(port.openRun(open)).rejects.toThrow("controller_detached"); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, identity.runId)); + expect(run.runnerProfileJson?.sessionCheckpoint).toBeUndefined(); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, identity.runId))).toEqual([]); + }); + it.each([ ["checkpoint", "detach"], ["result", "detach"], ["event", "detach"], ["checkpoint", "abort"], ["result", "abort"], ["event", "abort"], diff --git a/server/src/services/native-runtime/paperclip-control-plane-port.ts b/server/src/services/native-runtime/paperclip-control-plane-port.ts index acb02e19b63..d6c6413d060 100644 --- a/server/src/services/native-runtime/paperclip-control-plane-port.ts +++ b/server/src/services/native-runtime/paperclip-control-plane-port.ts @@ -83,6 +83,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { readonly #binding: PaperclipControlPlaneBinding; #sessionId: string | null = null; readonly #assertControllerActive?: () => void; + readonly #assertRunAdmissionActive?: () => void; readonly #onCommittedEvent?: (event: PrpEvent) => Promise; readonly #onDuplicateEvent?: (event: PrpEvent) => Promise; @@ -94,12 +95,15 @@ export class PaperclipControlPlanePort implements ControlPlanePort { privateKeyPem?: string; /** Synchronous revocation fence for a controller handing off on restart. */ assertControllerActive?: () => void; + /** Read-only admission may finish an existing startup solely for restart detachment. */ + assertRunAdmissionActive?: () => void; onCommittedEvent?: (event: PrpEvent) => Promise; onDuplicateEvent?: (event: PrpEvent) => Promise; } = {}, ) { this.#db = db; this.#assertControllerActive = options.assertControllerActive; + this.#assertRunAdmissionActive = options.assertRunAdmissionActive ?? options.assertControllerActive; this.#identityRedactor = createAgentIdentityRedactor(options.privateKeyPem); this.#binding = structuredClone(binding); this.#onCommittedEvent = options.onCommittedEvent; @@ -123,7 +127,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { } async openRun(input: OpenControlPlaneRunInput): Promise { - this.#assertControllerActive?.(); + this.#assertRunAdmissionActive?.(); const identity = input.identity; if ( identity.companyId !== this.#binding.companyId @@ -146,7 +150,7 @@ export class PaperclipControlPlanePort implements ControlPlanePort { if (!run || !this.#matchesPersistedBinding(run)) { throw new Error("native_open_run_not_authorized"); } - this.#assertControllerActive?.(); + this.#assertRunAdmissionActive?.(); this.#sessionId = identity.sessionId; }