diff --git a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs index abd852e105c..c7ad76fe775 100644 --- a/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs +++ b/packages/paperclip-runner/runner/crates/runner-core/src/durable/runner.rs @@ -99,6 +99,17 @@ impl CommandLifecycle { } } + /// A rejected suspension changed no provider state: the provider still + /// owns its session. Keep the runner serving its controller so it can + /// retry, instead of recording `suspended` over a live provider and + /// exiting. A failed shutdown still stops the runner. + fn after_failure(command: &Command) -> Self { + match command.command_type.as_str() { + "runner.suspend" => Self::Continue, + _ => Self::for_terminal(command), + } + } + fn merge(self, next: Self) -> Self { match (self, next) { (Self::Shutdown, _) | (_, Self::Shutdown) => Self::Shutdown, @@ -1974,6 +1985,23 @@ fn deliver_warm_attachment( } } +/// A replayed terminal command must reproduce the lifecycle its fence +/// recorded. One legacy exception: runners built before a failed +/// `runner.suspend` stopped being terminal persisted a `suspended` fence for +/// it. Its replay now continues; honour that old fence so an upgraded runner +/// can still deliver the journaled result and finish the old transition. +fn replay_satisfies_terminal_fence( + pending: &PendingTerminalDelivery, + result_status: &str, + lifecycle: CommandLifecycle, +) -> bool { + lifecycle.durable_state() == Some(pending.lifecycle.as_str()) + || (pending.command_type == "runner.suspend" + && pending.lifecycle == "suspended" + && matches!(result_status, "failed" | "rejected") + && lifecycle == CommandLifecycle::Continue) +} + fn reconcile_pending_terminal_delivery( state: &mut DurableState, store: &DurableStateStore, @@ -1998,7 +2026,7 @@ fn reconcile_pending_terminal_delivery( )); } let (result, lifecycle) = process_command(state, store, config, executor, command)?; - if lifecycle.durable_state() != Some(pending.lifecycle.as_str()) { + if !replay_satisfies_terminal_fence(&pending, &result.status, lifecycle) { return Err(DurableRunnerError::invalid( "pending terminal command did not replay its durable lifecycle", )); @@ -2293,6 +2321,8 @@ fn process_command( // the old receipt must not replace the newer command cursor // with another terminal-delivery fence at an older sequence. CommandLifecycle::Continue + } else if result.status == "failed" { + CommandLifecycle::after_failure(command) } else { CommandLifecycle::for_terminal(command) }; @@ -2340,7 +2370,7 @@ fn process_command( } let result = state.fail_command(command, failure)?; store.save(state)?; - return Ok((result, CommandLifecycle::for_terminal(command))); + return Ok((result, CommandLifecycle::after_failure(command))); } }; if matches!( @@ -5685,7 +5715,10 @@ mod tests { #[test] fn failed_lifecycle_commands_replay_their_terminal_transition() { for (command_type, expected_lifecycle) in [ - ("runner.suspend", CommandLifecycle::Suspend), + // A rejected suspension leaves the provider's session live, so the + // runner keeps serving its controller (which may retry) rather + // than recording `suspended` over it and exiting. + ("runner.suspend", CommandLifecycle::Continue), ("runner.shutdown", CommandLifecycle::Shutdown), ] { let directory = std::env::temp_dir().join(format!( @@ -5943,6 +5976,87 @@ mod tests { fs::remove_dir_all(directory).unwrap(); } + #[test] + fn legacy_failed_suspend_fence_still_reconciles_after_upgrade() { + let fence = |command_type: &str, lifecycle: &str| PendingTerminalDelivery { + command_id: "command_1".to_owned(), + controller_seq: 1, + command_type: command_type.to_owned(), + lifecycle: lifecycle.to_owned(), + }; + // An old runner persisted `suspended` for a failed suspend; the + // upgraded replay continues, and must still satisfy that fence. + assert!(replay_satisfies_terminal_fence( + &fence("runner.suspend", "suspended"), + "failed", + CommandLifecycle::Continue + )); + assert!(replay_satisfies_terminal_fence( + &fence("runner.suspend", "suspended"), + "completed", + CommandLifecycle::Suspend + )); + assert!(replay_satisfies_terminal_fence( + &fence("runner.shutdown", "stopped"), + "failed", + CommandLifecycle::Shutdown + )); + // A completed suspend or a shutdown must still replay its own lifecycle. + assert!(!replay_satisfies_terminal_fence( + &fence("runner.suspend", "suspended"), + "completed", + CommandLifecycle::Continue + )); + assert!(!replay_satisfies_terminal_fence( + &fence("runner.shutdown", "stopped"), + "failed", + CommandLifecycle::Continue + )); + } + + #[test] + fn rejected_suspend_keeps_the_runner_serving_its_controller() { + let directory = std::env::temp_dir().join(format!( + "paperclip-runner-rejected-suspend-{}", + std::process::id() + )); + let _ = fs::remove_dir_all(&directory); + let config = config(directory.clone()); + let store = DurableStateStore::new(&directory).unwrap(); + let (mut state, _) = store.load_or_create(&config).unwrap(); + let mut executor = FailingExecutor { calls: 0 }; + let command = command("runner.suspend"); + + let (result, lifecycle) = + process_command(&mut state, &store, &config, &mut executor, &command).unwrap(); + let (replayed, replay_lifecycle) = + process_command(&mut state, &store, &config, &mut executor, &command).unwrap(); + + assert_eq!(result.status, "failed"); + assert_eq!(replayed.status, "failed"); + // The provider still owns its session: no `suspended` lifecycle and no + // exit, so a fresh runner.suspend from the controller can be served. + assert_eq!(lifecycle, CommandLifecycle::Continue); + assert_eq!(replay_lifecycle, CommandLifecycle::Continue); + assert_eq!(lifecycle.durable_state(), None); + assert_eq!(executor.calls, 1); + let mut retry = CountingExecutor { calls: 0 }; + let (_, retry_lifecycle) = process_command( + &mut state, + &store, + &config, + &mut retry, + &Command { + command_id: "command-suspend-retry".to_owned(), + controller_seq: 2, + ..command.clone() + }, + ) + .unwrap(); + assert_eq!(retry_lifecycle, CommandLifecycle::Suspend); + fs::remove_dir_all(directory).unwrap(); + } + #[test] fn completed_suspend_replay_remains_restartable() { let directory = std::env::temp_dir().join(format!( 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 0dd2161ca0b..1332df7ddf6 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 @@ -5132,6 +5132,44 @@ describe("DurablePrpControlPlane", () => { }, ); + it("delivers a fresh suspension after a rejected one on the same connection", async () => { + const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-rejected-suspend-")); + const controlPlane = new DurablePrpControlPlane({ + stateDirectory: root, + identity, + expectedRunnerVersion, + expectedRunnerDigest, + }); + try { + await controlPlane.start(); + const rejected = controlPlane.queueCommand("runner.suspend", {}, "command-suspend-rejected"); + const client = await authenticate(controlPlane, controlPlane.issueBootstrapTicket()); + const retry = controlPlane.queueCommand("runner.suspend", {}, "command-suspend-retry", true); + expect(controlPlane.store.state.commandDeliveryCounts[retry.commandId]).toBeUndefined(); + sendSecure(client!, { + protocol: "paperclip.runner", version: 1, kind: "command_result", + payload: { + commandId: rejected.commandId, commandType: rejected.type, controllerSeq: rejected.controllerSeq, + status: "failed", + result: { code: "command_execution_failed", message: "ACPX session is not at a safe suspension point" }, + }, + }); + await expect(receiveSecure(client!)).resolves.toMatchObject({ + kind: "command_result_ack", payload: { commandId: "command-suspend-rejected", status: "failed" }, + }); + // The runner keeps serving after a rejected suspension, so the retry is + // delivered on the same connection instead of waiting behind the fence. + await expect(receiveSecure(client!)).resolves.toMatchObject({ + kind: "command", payload: { commandId: "command-suspend-retry" }, + }); + expect(controlPlane.store.state.commandDeliveryCounts[retry.commandId]).toBe(1); + client?.socket.destroy(); + } finally { + await controlPlane.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + it("acknowledges terminal command results after persisting them", async () => { const root = mkdtempSync(resolve(tmpdir(), "paperclip-prp-terminal-ack-")); const controlPlane = new DurablePrpControlPlane({ 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 147267e0b82..e591d742379 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 @@ -3983,7 +3983,8 @@ export class DurablePrpControlPlane { this.#store.state.duplicateCommandResults += 1; this.#store.save(); this.#ackTerminalCommandResult(connection, command); - if (!this.#isTerminalLifecycleCommand(command)) { + this.#releaseServingLifecycle(connection, command); + if (!this.#isTerminalLifecycleCommand(command) || connection.terminalLifecycleCommandId === null) { this.#sendNextCommand(connection); } return; @@ -4028,11 +4029,31 @@ export class DurablePrpControlPlane { command.result = structuredClone(result); this.#store.save(); this.#ackTerminalCommandResult(connection, command); - if (!this.#isTerminalLifecycleCommand(command)) { + this.#releaseServingLifecycle(connection, command); + if (!this.#isTerminalLifecycleCommand(command) || connection.terminalLifecycleCommandId === null) { this.#sendNextCommand(connection); } } + /** + * A rejected runner.suspend changes no provider state, and the runner keeps + * serving this connection. Lift the terminal-lifecycle fence it installed + * so the controller's next command (a fresh suspend) can be delivered. + * Completed suspensions and every shutdown keep the fence. + */ + #releaseServingLifecycle( + connection: AuthorityConnection, + command: DurableRecoveryCoreCommand, + ): void { + if ( + command.type === "runner.suspend" && + (command.status === "failed" || command.status === "rejected") && + connection.terminalLifecycleCommandId === command.commandId + ) { + connection.terminalLifecycleCommandId = null; + } + } + #ackWarmTransition( connection: AuthorityConnection, receipt: DurableWarmRunTransition, diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts index 6b229b66174..12845bcac4f 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts @@ -2013,6 +2013,56 @@ it("waits for the exact durable suspension command behind prior close work", asy expect(pumpCount).toBeGreaterThanOrEqual(2); }); +it("retries a rejected suspension within the close deadline instead of failing a settled session", async () => { + // An ACPX sidecar can reject session.suspend while the just-completed turn + // is still settling. One rejected command must not leave a successful run's + // own session closed without a proven suspension. + const commands: Array<{ commandId: string; type: string; status: string }> = []; + let lifecycle = "ready"; + const queueSuspend = vi.fn((commandId: string) => { + commands.push({ commandId, type: "runner.suspend", status: "pending" }); + }); + await expect( + runnerdRecoveryInternals.awaitRunnerSuspensionBarrier({ + commands: () => commands, + queueSuspend, + readRunnerState: async () => ({ lifecycle }), + runnerHasExited: async () => false, + pump: () => { + const pending = commands.find(command => command.status === "pending"); + if (!pending) return; + if (commands.length === 1) pending.status = "failed"; + else { pending.status = "completed"; lifecycle = "suspended"; } + }, + deadline: Date.now() + 2_000, + pollIntervalMs: 0, + }), + ).resolves.toBe(true); + expect(queueSuspend).toHaveBeenCalledTimes(2); + expect(new Set(commands.map(command => command.commandId)).size).toBe(2); +}); + +it("stops retrying a rejected suspension at its attempt bound or when the runner exits", async () => { + for (const exited of [false, true]) { + const commands: Array<{ commandId: string; type: string; status: string }> = []; + const queueSuspend = vi.fn((commandId: string) => { + commands.push({ commandId, type: "runner.suspend", status: "failed" }); + }); + await expect( + runnerdRecoveryInternals.awaitRunnerSuspensionBarrier({ + commands: () => commands, + queueSuspend, + readRunnerState: async () => ({ lifecycle: "ready" }), + runnerHasExited: async () => exited, + pump: () => undefined, + deadline: Date.now() + 5_000, + pollIntervalMs: 0, + }), + ).resolves.toBe(false); + expect(queueSuspend).toHaveBeenCalledTimes(exited ? 1 : 3); + } +}); + it("does not accept process exit without durable suspension", async () => { const commands: Array<{ commandId: string; diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 99313961fb3..89b1dd47967 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -929,6 +929,9 @@ async function releaseRunnerProcessOwnership(input: { if (releaseFailure !== undefined) throw releaseFailure; } +const MAX_RUNNER_SUSPEND_ATTEMPTS = 3; +const RUNNER_SUSPEND_RETRY_DELAY_MS = 250; + async function awaitRunnerSuspensionBarrier(input: { commands: () => readonly { commandId: string; @@ -980,10 +983,11 @@ async function awaitRunnerSuspensionBarrier(input: { existing = undefined; } if (Date.now() >= input.deadline) return false; - const commandId = + let commandId = existing?.commandId ?? `command_close_suspend_${randomUUID().replaceAll("-", "")}`; if (!existing) input.queueSuspend(commandId); + let suspendAttempts = 1; while (Date.now() < input.deadline) { input.pump(); @@ -995,7 +999,25 @@ async function awaitRunnerSuspensionBarrier(input: { command.status !== "pending" && command.status !== "completed" ) { - return false; + // A provider can reject suspension while its just-finished turn is + // still settling (an ACPX sidecar reports "not at a safe suspension + // point"). The rejection changed nothing, so ask again with a fresh + // command, a bounded number of times, within the same close deadline. + // An exited runner cannot answer, and an exhausted budget fails closed. + if ( + suspendAttempts >= MAX_RUNNER_SUSPEND_ATTEMPTS || + (await input.runnerHasExited()) + ) { + return false; + } + await new Promise((resolveWait) => + setTimeout(resolveWait, input.pollIntervalMs ?? RUNNER_SUSPEND_RETRY_DELAY_MS), + ); + if (Date.now() >= input.deadline) return false; + suspendAttempts += 1; + commandId = `command_close_suspend_${randomUUID().replaceAll("-", "")}`; + input.queueSuspend(commandId); + continue; } let lifecycle: unknown; try { diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 575bb67bb7a..1daac7d5bab 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -42,6 +42,7 @@ import { completeTerminatedRemoteNativeSessionCleanup, completeTerminatedLocalNativeSessionCleanup, nativeSessionCleanupBlock, + onNativeSessionCleanupQuarantined, type ExecuteNativeSessionOptions, } from "./native-session-runtime.js"; @@ -5006,6 +5007,45 @@ describe("executeNativeSession recovery", () => { completeTerminatedLocalNativeSessionCleanup(binding); }); + it("announces every cleanup quarantine with its cause when it is created", async () => { + const quarantined = vi.fn(); + const unsubscribe = onNativeSessionCleanupQuarantined(quarantined); + const scopedIdentity = { ...identity, companyId: "announce-company", runId: "announce-run", issueId: "announce-issue", sessionId: "announce-session" }; + const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true }; + const settlement = { runnerSuspended: false, suspensionState: { commandStatus: "failed", runnerLifecycle: "suspended" } }; + const session: NativeSession = { + identity: () => scopedIdentity, capabilities: async () => capabilities, + async *events() { throw new Error("announce transport stopped"); }, + startTurn: async () => ({ turnId: "announce-turn" }), result: async () => null, + async snapshot() { return { backendKind: "mock", sessionId: "announce-driver", identity: scopedIdentity, + providerSessionId: "announce-provider", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }; }, + close: vi.fn(async () => { throw new NativeSessionCloseUnrecoverableError(settlement); }), + }; + const backend: NativeSessionBackend = { + descriptor: async () => ({ kind: "runner", name: "announce-backend", version: "1", capabilities }), + openSession: vi.fn(async () => session), + }; + try { + await expect(executeNativeSession({ + input: { ...input, binding: { ...input.binding, companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, issueId: scopedIdentity.issueId }, + session: { ...input.session, normalizedSessionId: scopedIdentity.sessionId } }, + backend, runnerInstanceId: "announce-runner", controlPlaneInstanceId: "control", requireSessionCloseBeforeReturn: true, + controlPlane: { openRun: async () => {}, checkpointSession: async () => {}, + appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }), + replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }), completeRun: async () => {} }, + })).rejects.toThrow("announce transport stopped"); + expect(quarantined).toHaveBeenCalledOnce(); + expect(quarantined).toHaveBeenCalledWith({ + companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, issueId: scopedIdentity.issueId, + sessionId: scopedIdentity.sessionId, operatorRecoveryRequired: true, + cause: expect.stringContaining("native_session_close_unrecoverable"), settlement, + }); + } finally { + unsubscribe(); + completeTerminatedLocalNativeSessionCleanup({ companyId: scopedIdentity.companyId, runId: scopedIdentity.runId, runnerInstanceId: "announce-runner" }); + } + }); + it("keeps an unrecoverable close from quarantining unrelated tasks of the same backend", async () => { const companyId = "unrecoverable-scope-company"; const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true }; diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 5d6743e53fa..913b4aca042 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -601,11 +601,11 @@ function retainUnadmittedSessionCleanup( return; } catch (error) { if (error instanceof NativeSessionCloseUnrecoverableError) { - quarantineSessionCleanup(session, cleanupDomain, true); + quarantineSessionCleanup(session, cleanupDomain, true, error); throw error; } if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) { - quarantineSessionCleanup(session, cleanupDomain); + quarantineSessionCleanup(session, cleanupDomain, false, error); throw error; } retryCount += 1; @@ -763,15 +763,86 @@ function retainFailedSessionCleanupOwner( .catch(() => undefined); } +export interface NativeSessionCleanupQuarantine { + companyId: string; + runId: string; + issueId: string; + sessionId: string; + operatorRecoveryRequired: boolean; + /** Bounded error name/code/message of the close failure that caused it. */ + cause: string; + /** Bounded settlement facts of an unrecoverable close, when it reported them. */ + settlement: Record | null; +} +type NativeSessionCleanupQuarantineListener = ( + quarantine: NativeSessionCleanupQuarantine, +) => void; +const cleanupQuarantineListeners = new Set(); + +/** Observe every cleanup quarantine when it is created or escalates to + * operator recovery, with its cause. Observation grants no authority. */ +export function onNativeSessionCleanupQuarantined( + listener: NativeSessionCleanupQuarantineListener, +): () => void { + cleanupQuarantineListeners.add(listener); + return () => cleanupQuarantineListeners.delete(listener); +} + +function boundedSettlement(cause: unknown): Record | null { + if (!(cause instanceof NativeSessionCloseUnrecoverableError) || !cause.settlement) return null; + try { + const text = JSON.stringify(cause.settlement); + return text.length <= 4_000 ? (JSON.parse(text) as Record) : { truncated: true }; + } catch { + return { unreadable: true }; + } +} + +function announceCleanupQuarantine( + session: NativeSession, + operatorRecoveryRequired: boolean, + cause: unknown, +): void { + if (cleanupQuarantineListeners.size === 0) return; + let identity: ReturnType | null = null; + try { + identity = session.identity(); + } catch { + // Observation must not fail quarantine. + } + const error = cause instanceof Error ? cause : null; + const code = error && "code" in error && typeof error.code === "string" ? error.code : null; + const quarantine: NativeSessionCleanupQuarantine = { + companyId: identity?.companyId ?? "", + runId: identity?.runId ?? "", + issueId: identity?.issueId ?? "", + sessionId: identity?.sessionId ?? "", + operatorRecoveryRequired, + cause: [error?.name, code, error?.message ?? (cause === undefined ? null : String(cause))] + .filter((part): part is string => Boolean(part)).join(": ").slice(0, 1_000) || "unknown", + settlement: boundedSettlement(cause), + }; + for (const listener of cleanupQuarantineListeners) { + try { + listener(structuredClone(quarantine)); + } catch { + // Observers cannot change cleanup ownership. + } + } +} + function quarantineSessionCleanup( session: NativeSession, cleanupDomain: NativeSessionCleanupDomain, operatorRecoveryRequired = false, + cause?: unknown, ): void { const existing = [...quarantinedSessionCleanups].find( (entry) => entry.session === session, ); if (existing) { + if (operatorRecoveryRequired && !existing.operatorRecoveryRequired) + announceCleanupQuarantine(session, true, cause); if (operatorRecoveryRequired) { existing.operatorRecoveryRequired = true; if (existing.timer) clearTimeout(existing.timer); @@ -790,6 +861,7 @@ function quarantineSessionCleanup( operatorRecoveryRequired, }; quarantinedSessionCleanups.add(cleanup); + announceCleanupQuarantine(session, operatorRecoveryRequired, cause); if (operatorRecoveryRequired) return; startQuarantinedSessionCleanupRecovery( cleanup, @@ -835,6 +907,8 @@ function startQuarantinedSessionCleanupRecovery( announceCleanupSettled(cleanup.domain); } catch (error) { if (error instanceof NativeSessionCloseUnrecoverableError) { + if (!cleanup.operatorRecoveryRequired) + announceCleanupQuarantine(cleanup.session, true, error); cleanup.operatorRecoveryRequired = true; } // Retain the quarantine after this finite, sequential retry batch. @@ -2484,11 +2558,11 @@ export async function executeNativeSession( return; } catch (error) { if (error instanceof NativeSessionCloseUnrecoverableError) { - quarantineSessionCleanup(session, cleanupDomain, true); + quarantineSessionCleanup(session, cleanupDomain, true, error); throw error; } if (retryCount >= MAX_FAILED_SESSION_CLOSE_RETRIES) { - quarantineSessionCleanup(session, cleanupDomain); + quarantineSessionCleanup(session, cleanupDomain, false, error); throw error; } retryCount += 1; diff --git a/server/src/index.ts b/server/src/index.ts index 77710bcc477..a962af47685 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -83,6 +83,7 @@ import { createSentryRunLinkResolver } from "./services/sentry-run-links.js"; import { waitForPendingRunFailureReports } from "./services/run-failure-report.js"; import { verifyStoppedNativeSessionForReplacement } from "./services/native-runtime/native-session-executor.js"; import { registerNativeCleanupReleaseWake } from "./services/native-runtime/native-cleanup-release-wake.js"; +import { logNativeCleanupQuarantines } from "./services/native-runtime/native-cleanup-quarantine-log.js"; import { embeddedPostgresOwnerPort } from "./embedded-postgres-owner.js"; import { deliverExecutionStatuses } from "./services/execution-status-delivery.js"; import { deliverReconciledExecutions, settleUnrecoverableExecutions } from "./services/execution-recovery-resolution.js"; @@ -1352,6 +1353,10 @@ async function startServerWithDatabaseTeardown( if (skillVersionRetentionWorker) trackHeartbeatSchedulerWork(skillVersionRetentionWorker.ready); const workspaceRecoveryWork = registerWorkspaceRecoveryWork(db, deliveryWork, workspaceRecoveryRuntime, accountingHeartbeat.resumeRemoteStopComments, () => !heartbeatSchedulerStopped); + // Each native cleanup quarantine is logged with its cause when created. The + // subscription lives with the process: closes that fail during shutdown, + // after HTTP closes, still need their cause logged. + logNativeCleanupQuarantines(logger); const localLoginCleanup = localAiLoginService(db); const localLoginCleanupWork = registerExecutionControlWork(deliveryWork, [ { queue: DELIVERY_QUEUES.localLoginCleanup, run: localLoginCleanup.reapExpired, next: localLoginCleanup.nextCleanupAt }, diff --git a/server/src/services/native-runtime/native-cleanup-quarantine-log.ts b/server/src/services/native-runtime/native-cleanup-quarantine-log.ts new file mode 100644 index 00000000000..41daf290361 --- /dev/null +++ b/server/src/services/native-runtime/native-cleanup-quarantine-log.ts @@ -0,0 +1,24 @@ +import { onNativeSessionCleanupQuarantined } from "../../vendor/paperclip-runner/index.js"; + +type QuarantineLogger = { warn: (details: Record, message: string) => void }; + +let installed: { logger: QuarantineLogger; unsubscribe: () => void } | null = null; + +/** + * Log every native session cleanup quarantine when the runner creates it or + * escalates it to operator recovery, so a later admission refusal can be + * traced to its owning session and close failure. The subscription is + * process-wide and installed once: native closes can still fail during + * shutdown after the HTTP server has closed. + */ +export function logNativeCleanupQuarantines(logger: QuarantineLogger): () => void { + if (installed?.logger === logger) return installed.unsubscribe; + installed?.unsubscribe(); + const unsubscribe = onNativeSessionCleanupQuarantined((quarantine) => { + logger.warn({ ...quarantine }, quarantine.operatorRecoveryRequired + ? "native session cleanup quarantined; operator recovery required" + : "native session cleanup quarantined; automatic close recovery scheduled"); + }); + installed = { logger, unsubscribe: () => { unsubscribe(); installed = null; } }; + return installed.unsubscribe; +} diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index e74290588c0..e78bb85d50b 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -193,6 +193,7 @@ export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTermi export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup; export const nativeSessionCleanupBlock = runner.nativeSessionCleanupBlock; export const onNativeSessionCleanupSettled = runner.onNativeSessionCleanupSettled; +export const onNativeSessionCleanupQuarantined = runner.onNativeSessionCleanupQuarantined; export const probeAcpxClaudeInstallation = runner.probeAcpxClaudeInstallation; export const probeAcpxGrokInstallation = runner.probeAcpxGrokInstallation;