diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 51f4b35daa..c929036915 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -105,6 +105,20 @@ pnpm dev:stop `pnpm dev:once` now tracks backend-relevant file changes and pending migrations. When the current boot is stale, the board UI shows a `Restart required` banner. You can also enable guarded auto-restart in `Instance Settings > Experimental`, which waits for queued/running local agent runs to finish before restarting the dev server. +## Hot-Restart Deploys + +Primary-instance rebuilds that restart `paperclip.service` can request one-shot live-run adoption instead of using the normal graceful shutdown drain. Before restarting the service, write the marker from the newly staged app with the current service PID: + +```sh +old_main_pid="$(systemctl show paperclip.service -p MainPID --value)" +pnpm --filter @paperclipai/server exec tsx ../scripts/request-hot-restart.ts --server-pid "$old_main_pid" +systemctl restart paperclip.service +``` + +Use `--drain-required` only when the deploy intentionally requires the old terminate-and-retry behavior. Without that flag, the old server verifies that the marker targets its own PID, snapshots currently running heartbeat run IDs and child PIDs, and skips the shutdown drain so eligible detached local-agent processes can keep running. On startup the new server writes `$PAPERCLIP_HOME/hot-restart-report.json` with `previousServerPid`, `newServerPid`, `previousServerVersion`, `newServerVersion`, `adoptedRunIds`, `finalizedWhileDownRunIds`, `lostRunIds`, and per-run classifications before the normal orphan reaper runs. + +A healthy guarded deploy must compare the report against `/api/health` (`version` or `serverVersion`) and treat any `lostRunIds` entry as a continuity failure that needs recovery before marking deployment complete. + Tailscale/private-auth dev mode: ```sh diff --git a/scripts/request-hot-restart.ts b/scripts/request-hot-restart.ts new file mode 100644 index 0000000000..1f7ca84cd5 --- /dev/null +++ b/scripts/request-hot-restart.ts @@ -0,0 +1,83 @@ +#!/usr/bin/env -S node --import tsx +import { + resolveHotRestartIntentPath, + writeHotRestartIntent, +} from "../server/src/services/hot-restart.js"; + +function usage(): never { + console.error([ + "Usage: tsx scripts/request-hot-restart.ts --server-pid [--drain-required]", + "", + "Writes a one-shot hot-restart intent marker under PAPERCLIP_HOME.", + ].join("\n")); + process.exit(2); +} + +function readArgs(argv: string[]) { + let serverPid: number | null = null; + let drainRequired = false; + + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + if (arg === "--server-pid") { + const raw = argv[index + 1]; + if (!raw) usage(); + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed <= 0) usage(); + serverPid = parsed; + index += 1; + continue; + } + if (arg === "--drain-required") { + drainRequired = true; + continue; + } + if (arg === "--help" || arg === "-h") usage(); + console.error(`Unknown argument: ${arg}`); + usage(); + } + + if (!serverPid) usage(); + return { serverPid, drainRequired }; +} + +function normalizeApiBase(raw: string | undefined) { + const trimmed = raw?.trim(); + if (!trimmed) return null; + return trimmed.replace(/\/+$/, "").replace(/\/api$/, ""); +} + +async function readPreviousServerVersion() { + const apiBase = normalizeApiBase(process.env.PAPERCLIP_API_URL); + if (!apiBase) return null; + try { + const response = await fetch(`${apiBase}/api/health`, { + signal: AbortSignal.timeout(2_000), + }); + if (!response.ok) return null; + const body = await response.json() as Record; + return typeof body.serverVersion === "string" + ? body.serverVersion + : typeof body.version === "string" + ? body.version + : null; + } catch { + return null; + } +} + +const { serverPid, drainRequired } = readArgs(process.argv.slice(2)); +const intent = await writeHotRestartIntent({ + previousServerPid: serverPid, + previousServerVersion: await readPreviousServerVersion(), + drainRequired, + requestedByRunId: process.env.PAPERCLIP_RUN_ID?.trim() || null, +}); + +console.log(JSON.stringify({ + status: "hot_restart_intent_written", + intentPath: resolveHotRestartIntentPath(), + previousServerPid: intent.previousServerPid, + previousServerVersion: intent.previousServerVersion, + drainRequired: intent.drainRequired, +}, null, 2)); diff --git a/server/src/__tests__/health.test.ts b/server/src/__tests__/health.test.ts index 9fbc47f3b4..69449e1f6e 100644 --- a/server/src/__tests__/health.test.ts +++ b/server/src/__tests__/health.test.ts @@ -73,7 +73,7 @@ describe("GET /health", () => { const app = createApp(); const res = await request(app).get("/health"); expect(res.status).toBe(200); - expect(res.body).toEqual({ status: "ok", version: serverVersion, serverInfo: testServerInfo }); + expect(res.body).toEqual({ status: "ok", version: serverVersion, serverVersion: serverVersion, serverInfo: testServerInfo }); }, 15_000); it("returns 200 when the database probe succeeds", async () => { @@ -105,6 +105,7 @@ describe("GET /health", () => { expect(res.body).toEqual({ status: "unhealthy", version: serverVersion, + serverVersion, error: "database_unreachable", serverInfo: testServerInfo, }); @@ -412,6 +413,7 @@ describe("GET /health", () => { expect(res.body).toMatchObject({ status: "ok", version: serverVersion, + serverVersion, deploymentMode: "authenticated", deploymentExposure: "public", authReady: true, diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index 2464f82fbb..1432021f91 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,5 +1,8 @@ import { randomUUID } from "node:crypto"; import { spawn, type ChildProcess } from "node:child_process"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; import { and, eq, or, inArray, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { @@ -100,6 +103,11 @@ import { heartbeatService, redactDetectedSuccessfulRunProgressSummaryForBoard, } from "../services/heartbeat.ts"; +import { + readHotRestartIntent, + resolveHotRestartReportPath, + writeHotRestartIntent, +} from "../services/hot-restart.ts"; import { secretService } from "../services/secrets.ts"; import { SUCCESSFUL_RUN_HANDOFF_EXHAUSTED_NOTICE_BODY, @@ -1406,6 +1414,200 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(retries).toHaveLength(0); }); + async function withTempPaperclipHome(fn: (home: string) => Promise): Promise { + const home = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-hot-restart-")); + const previousHome = process.env.PAPERCLIP_HOME; + process.env.PAPERCLIP_HOME = home; + try { + return await fn(home); + } finally { + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; + else process.env.PAPERCLIP_HOME = previousHome; + await fs.rm(home, { recursive: true, force: true }); + } + } + + it("captures a hot-restart shutdown snapshot without interrupting running runs", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeGreaterThan(0); + const { runId, wakeupRequestId } = await seedRunFixture({ + agentStatus: "running", + processPid: child.pid ?? null, + processGroupId: null, + }); + + await withTempPaperclipHome(async () => { + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-version", + requestedAt: new Date("2026-03-19T00:05:00.000Z"), + }); + const heartbeat = heartbeatService(db); + + const result = await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-03-19T00:06:00.000Z"), + ); + + expect(result).toEqual({ + mode: "hot_restart", + skipDrain: true, + activeRunIds: [runId], + }); + expect(isPidAlive(child.pid)).toBe(true); + const run = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(run).toMatchObject({ + status: "running", + errorCode: null, + }); + const wakeup = await db + .select() + .from(agentWakeupRequests) + .where(eq(agentWakeupRequests.id, wakeupRequestId)) + .then((rows) => rows[0] ?? null); + expect(wakeup?.status).toBe("claimed"); + const intent = await readHotRestartIntent(); + expect(intent?.shutdownSnapshot).toMatchObject({ + capturedAt: "2026-03-19T00:06:00.000Z", + signal: "SIGTERM", + activeRuns: [ + { + runId, + adapterType: "codex_local", + status: "running", + processPid: child.pid, + }, + ], + }); + }); + }); + + it("reports adopted hot-restart runs before startup reap can mark them process_lost", async () => { + const child = spawnAliveProcess(); + childProcesses.add(child); + expect(child.pid).toBeGreaterThan(0); + const { runId } = await seedRunFixture({ + agentStatus: "running", + processPid: child.pid ?? null, + processGroupId: null, + }); + + await withTempPaperclipHome(async (home) => { + const heartbeat = heartbeatService(db); + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-version", + requestedAt: new Date("2026-03-19T00:05:00.000Z"), + }); + await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-03-19T00:06:00.000Z"), + ); + + const adoption = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-03-19T00:07:00.000Z"), + ); + expect(adoption).toMatchObject({ + mode: "reported", + adoptedRunIds: [runId], + finalizedWhileDownRunIds: [], + lostRunIds: [], + skippedRunIds: [], + }); + + const report = JSON.parse( + await fs.readFile(resolveHotRestartReportPath(home), "utf8"), + ) as Record; + expect(report).toMatchObject({ + previousServerPid: process.pid, + newServerPid: process.pid, + previousServerVersion: "old-version", + adoptedRunIds: [runId], + finalizedWhileDownRunIds: [], + lostRunIds: [], + }); + expect(typeof report.newServerVersion).toBe("string"); + + const reap = await heartbeat.reapOrphanedRuns(); + expect(reap).toEqual({ reaped: 0, runIds: [] }); + const adopted = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(adopted?.status).toBe("running"); + expect(adopted?.errorCode).not.toBe("process_lost"); + expect(adopted?.resultJson).toMatchObject({ + hotRestart: { + adopted: true, + adoptedAt: "2026-03-19T00:07:00.000Z", + previousServerPid: process.pid, + newServerPid: process.pid, + previousServerVersion: "old-version", + processPid: child.pid, + }, + }); + }); + }); + + it.skipIf(process.platform === "win32")("keeps process-group-only hot-restart adoptions out of process_lost reaping", async () => { + const orphan = await spawnOrphanedProcessGroup(); + cleanupPids.add(orphan.descendantPid); + expect(isPidAlive(orphan.descendantPid)).toBe(true); + const { runId } = await seedRunFixture({ + agentStatus: "running", + processPid: orphan.processPid, + processGroupId: orphan.processGroupId, + }); + + await withTempPaperclipHome(async () => { + const heartbeat = heartbeatService(db); + await writeHotRestartIntent({ + previousServerPid: process.pid, + previousServerVersion: "old-version", + requestedAt: new Date("2026-03-19T00:05:00.000Z"), + }); + await heartbeat.prepareHotRestartShutdown( + "SIGTERM", + new Date("2026-03-19T00:06:00.000Z"), + ); + + const adoption = await heartbeat.reconcileHotRestartAdoption( + new Date("2026-03-19T00:07:00.000Z"), + ); + expect(adoption).toMatchObject({ + mode: "reported", + adoptedRunIds: [runId], + finalizedWhileDownRunIds: [], + lostRunIds: [], + skippedRunIds: [], + }); + + const reap = await heartbeat.reapOrphanedRuns(); + expect(reap).toEqual({ reaped: 0, runIds: [] }); + expect(isPidAlive(orphan.descendantPid)).toBe(true); + const adopted = await db + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .then((rows) => rows[0] ?? null); + expect(adopted?.status).toBe("running"); + expect(adopted?.errorCode).not.toBe("process_lost"); + expect(adopted?.resultJson).toMatchObject({ + hotRestart: { + adopted: true, + processPid: orphan.processPid, + processGroupId: orphan.processGroupId, + }, + }); + }); + }); + it("interrupts running runs on graceful shutdown and queues restart recovery without recording a failure", async () => { const { agentId, runId, issueId, wakeupRequestId } = await seedRunFixture({ agentStatus: "running", diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index ef98f7d8bb..91cc9fdd73 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -35,6 +35,7 @@ const { })); const heartbeatServiceMock = { resolveSchedulingSuppression: resolveHeartbeatSchedulingSuppressionMock, + reconcileHotRestartAdoption: vi.fn(async () => ({ mode: "none" })), reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), promoteDueScheduledRetries: vi.fn(async () => ({ promoted: 0, runIds: [] })), resumeQueuedRuns: vi.fn(async () => undefined), @@ -340,6 +341,22 @@ describe("startServer feedback export wiring", () => { } }); + it("does not replay hot-restart adoption when the orphan reaper retries", async () => { + loadConfigMock.mockReturnValue(buildTestConfig({ + heartbeatSchedulerEnabled: true, + heartbeatSchedulerIntervalMs: 30000, + })); + heartbeatServiceMock.reconcileHotRestartAdoption.mockRejectedValueOnce(new Error("partial adoption")); + heartbeatServiceMock.reapOrphanedRuns + .mockRejectedValueOnce(new Error("transient reap failure")) + .mockResolvedValueOnce({ reaped: 0, runIds: [] }); + + await startServer(); + + expect(heartbeatServiceMock.reconcileHotRestartAdoption).toHaveBeenCalledTimes(1); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenCalledTimes(2); + }); + it("refuses authenticated public startup without an external database URL", async () => { loadConfigMock.mockReturnValue(buildTestConfig({ deploymentExposure: "public", diff --git a/server/src/index.ts b/server/src/index.ts index 03e7921ec3..7499477f15 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -822,6 +822,7 @@ export async function startServer(): Promise { } let drainHeartbeatRunsForShutdown: ((signal: "SIGINT" | "SIGTERM") => Promise) | null = null; + let prepareHotRestartShutdown: ((signal: "SIGINT" | "SIGTERM") => Promise<{ skipDrain: boolean }>) | null = null; let heartbeatSchedulerStopped = false; let heartbeatSchedulerInterval: ReturnType | null = null; const heartbeatSchedulerInFlight = new Set>(); @@ -843,6 +844,7 @@ export async function startServer(): Promise { if (config.heartbeatSchedulerEnabled) { const heartbeat = heartbeatService(db as any, { pluginWorkerManager }); drainHeartbeatRunsForShutdown = heartbeat.drainRunningRunsForShutdown; + prepareHotRestartShutdown = heartbeat.prepareHotRestartShutdown; const environmentCustomImages = environmentCustomImageService(db as any, { pluginWorkerManager }); const routines = routineService(db as any, { pluginWorkerManager }); const tools = toolAccessService(db as any, { @@ -873,6 +875,21 @@ export async function startServer(): Promise { ); } else { const startupHeartbeatRecovery = (async () => { + try { + const hotRestart = await heartbeat.reconcileHotRestartAdoption(); + if (hotRestart.mode === "reported") { + logger.info( + hotRestart, + "startup hot-restart adoption reconciliation complete", + ); + } + } catch (err) { + logger.error( + { err }, + "startup hot-restart adoption reconciliation failed - orphan reaper will serve as degraded backstop", + ); + } + for (let attempt = 1; attempt <= 2; attempt++) { try { const result = await heartbeat.reapOrphanedRuns(); @@ -1194,7 +1211,20 @@ export async function startServer(): Promise { await telemetryClient.flush(); } - if (drainHeartbeatRunsForShutdown) { + let skipHeartbeatDrain = false; + if (prepareHotRestartShutdown) { + try { + const hotRestart = await prepareHotRestartShutdown(signal); + skipHeartbeatDrain = hotRestart.skipDrain; + if (skipHeartbeatDrain) { + logger.info({ signal, hotRestart }, "hot-restart shutdown prepared; skipping graceful heartbeat run drain"); + } + } catch (err) { + logger.error({ err, signal }, "hot-restart shutdown preparation failed; falling back to graceful heartbeat run drain"); + } + } + + if (!skipHeartbeatDrain && drainHeartbeatRunsForShutdown) { try { const drain = await drainHeartbeatRunsForShutdown(signal); logger.info({ signal, drain }, "graceful heartbeat run drain complete"); diff --git a/server/src/routes/health.ts b/server/src/routes/health.ts index 60057935fe..fef83f7ad9 100644 --- a/server/src/routes/health.ts +++ b/server/src/routes/health.ts @@ -126,7 +126,7 @@ export function healthRoutes( if (!db) { res.json( exposeFullDetails - ? { status: "ok", version: serverVersion, serverInfo } + ? { status: "ok", version: serverVersion, serverVersion: serverVersion, serverInfo } : { status: "ok", deploymentMode: opts.deploymentMode }, ); return; @@ -139,6 +139,7 @@ export function healthRoutes( res.status(503).json({ status: "unhealthy", version: serverVersion, + serverVersion, error: "database_unreachable", ...(exposeFullDetails ? { serverInfo } : {}), }); @@ -214,6 +215,7 @@ export function healthRoutes( res.json({ status: "ok", version: serverVersion, + serverVersion, deploymentMode: opts.deploymentMode, deploymentExposure: opts.deploymentExposure, authReady: opts.authReady, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index a487f8a914..a11d13783e 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -255,6 +255,15 @@ import { sweepExpiredHeartbeatRunRuntimeStatuses, touchHeartbeatRunRuntimeStatus, } from "./heartbeat-run-runtime-status.js"; +import { + readHotRestartIntent, + removeHotRestartIntent, + shouldHonorHotRestartIntentForProcess, + writeHotRestartReport, + writeHotRestartShutdownSnapshot, + type HotRestartIntentRun, + type HotRestartReportRun, +} from "./hot-restart.js"; import { assertLowTrustRuntimeServicesAllowed, assertLowTrustWorkspaceIsolation, @@ -268,6 +277,7 @@ import { type EffectiveRunConfigSecretManifestEntry, } from "./effective-run-config-fingerprints.js"; import type { PluginWorkerManager } from "./plugin-worker-manager.js"; +import { serverVersion } from "../version.js"; const MAX_LIVE_LOG_CHUNK_BYTES = 8 * 1024; const MAX_PERSISTED_LOG_CHUNK_CHARS = 64 * 1024; @@ -5076,6 +5086,43 @@ function buildProcessLossMessage(run: { return "Process lost -- server may have restarted"; } +function readHotRestartAdoptionMetadata(resultJson: Record | null | undefined) { + const result = parseObject(resultJson); + const hotRestart = parseObject(result.hotRestart); + if (hotRestart.adopted !== true || typeof hotRestart.adoptedAt !== "string") return null; + return hotRestart; +} + +function mergeHotRestartAdoptionResultJson( + resultJson: Record | null | undefined, + input: { + adoptedAt: Date; + previousServerPid: number; + newServerPid: number; + previousServerVersion: string | null; + newServerVersion: string; + processPid: number | null; + processGroupId: number | null; + }, +) { + const result = parseObject(resultJson); + const existing = parseObject(result.hotRestart); + return { + ...result, + hotRestart: { + ...existing, + adopted: true, + adoptedAt: input.adoptedAt.toISOString(), + previousServerPid: input.previousServerPid, + newServerPid: input.newServerPid, + previousServerVersion: input.previousServerVersion, + newServerVersion: input.newServerVersion, + processPid: input.processPid, + processGroupId: input.processGroupId, + }, + }; +} + function truncateDisplayId(value: string | null | undefined, max = 128) { if (!value) return null; return value.length > max ? value.slice(0, max) : value; @@ -8647,6 +8694,287 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) return queued; } + function toHotRestartIntentRun(input: { + run: typeof heartbeatRuns.$inferSelect; + adapterType: string; + }): HotRestartIntentRun { + const context = parseObject(input.run.contextSnapshot); + return { + runId: input.run.id, + companyId: input.run.companyId, + agentId: input.run.agentId, + adapterType: input.adapterType, + status: input.run.status, + processPid: input.run.processPid ?? null, + processGroupId: input.run.processGroupId ?? null, + issueId: readNonEmptyString(context.issueId), + }; + } + + async function prepareHotRestartShutdown(signal: "SIGINT" | "SIGTERM", now = new Date()) { + let intent: Awaited>; + try { + intent = await readHotRestartIntent(); + } catch (err) { + logger.warn({ err }, "failed to read hot-restart intent; falling back to normal shutdown drain"); + return { mode: "read_error" as const, skipDrain: false as const, activeRunIds: [] as string[] }; + } + + if (!intent) return { mode: "not_requested" as const, skipDrain: false as const, activeRunIds: [] as string[] }; + if (intent.drainRequired) return { mode: "drain_required" as const, skipDrain: false as const, activeRunIds: [] as string[] }; + if (!shouldHonorHotRestartIntentForProcess(intent)) { + logger.warn( + { expectedPid: intent.previousServerPid, currentPid: process.pid }, + "hot-restart intent targets a different server pid; falling back to normal shutdown drain", + ); + return { mode: "pid_mismatch" as const, skipDrain: false as const, activeRunIds: [] as string[] }; + } + + const activeRuns = await db + .select({ + run: heartbeatRuns, + adapterType: agents.adapterType, + }) + .from(heartbeatRuns) + .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) + .where(eq(heartbeatRuns.status, "running")); + const snapshotRuns = activeRuns.map(toHotRestartIntentRun); + const intentWithVersion = { + ...intent, + previousServerVersion: intent.previousServerVersion ?? serverVersion, + }; + + await writeHotRestartShutdownSnapshot({ + intent: intentWithVersion, + signal, + activeRuns: snapshotRuns, + capturedAt: now, + }); + + for (const { run } of activeRuns) { + await appendRunEvent(run, await nextRunEventSeq(run.id), { + eventType: "lifecycle", + stream: "system", + level: "info", + message: "Hot restart requested; leaving child process alive for startup adoption", + payload: { + signal, + previousServerPid: intent.previousServerPid, + previousServerVersion: intentWithVersion.previousServerVersion, + processPid: run.processPid ?? null, + processGroupId: run.processGroupId ?? null, + }, + }); + } + + logger.info( + { signal, previousServerPid: intent.previousServerPid, activeRunIds: snapshotRuns.map((run) => run.runId) }, + "hot-restart shutdown snapshot captured; skipping graceful run drain", + ); + + return { + mode: "hot_restart" as const, + skipDrain: true as const, + activeRunIds: snapshotRuns.map((run) => run.runId), + }; + } + + async function reconcileHotRestartAdoption(now = new Date()) { + let intent: Awaited>; + try { + intent = await readHotRestartIntent(); + } catch (err) { + logger.warn({ err }, "failed to read hot-restart intent on startup; skipping adoption"); + return { + mode: "read_error" as const, + adoptedRunIds: [] as string[], + finalizedWhileDownRunIds: [] as string[], + lostRunIds: [] as string[], + skippedRunIds: [] as string[], + }; + } + if (!intent) { + return { + mode: "not_requested" as const, + adoptedRunIds: [] as string[], + finalizedWhileDownRunIds: [] as string[], + lostRunIds: [] as string[], + skippedRunIds: [] as string[], + }; + } + + if (!intent.shutdownSnapshot) { + logger.warn( + { previousServerPid: intent.previousServerPid }, + "hot-restart intent present but shutdown snapshot is missing; no runs can be adopted", + ); + } + const candidates = intent.shutdownSnapshot?.activeRuns ?? []; + const currentRows = candidates.length > 0 + ? await db + .select({ + run: heartbeatRuns, + adapterType: agents.adapterType, + }) + .from(heartbeatRuns) + .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) + .where(inArray(heartbeatRuns.id, candidates.map((run) => run.runId))) + : []; + const currentByRunId = new Map(currentRows.map((row) => [row.run.id, row])); + + const reportRuns: HotRestartReportRun[] = []; + const adoptedRunIds: string[] = []; + const finalizedWhileDownRunIds: string[] = []; + const lostRunIds: string[] = []; + const skippedRunIds: string[] = []; + + const classify = ( + candidate: HotRestartIntentRun, + classification: HotRestartReportRun["classification"], + reason: string, + patch?: Partial, + ) => { + const run = { ...candidate, ...patch, classification, reason } satisfies HotRestartReportRun; + reportRuns.push(run); + if (classification === "adopted") adoptedRunIds.push(candidate.runId); + else if (classification === "finalized_while_down") finalizedWhileDownRunIds.push(candidate.runId); + else if (classification === "lost") lostRunIds.push(candidate.runId); + else skippedRunIds.push(candidate.runId); + }; + + for (const candidate of candidates) { + const current = currentByRunId.get(candidate.runId); + if (!current) { + classify(candidate, "finalized_while_down", "run_row_missing"); + continue; + } + + const { run, adapterType } = current; + const patch = { + adapterType, + status: run.status, + processPid: run.processPid ?? candidate.processPid, + processGroupId: run.processGroupId ?? candidate.processGroupId, + }; + + if (run.status !== "running") { + classify(candidate, "finalized_while_down", `run_status_${run.status}`, patch); + continue; + } + + if (intent.drainRequired) { + classify(candidate, "skipped", "drain_required", patch); + continue; + } + + if (!isTrackedLocalChildProcessAdapter(adapterType)) { + classify(candidate, "skipped", "adapter_not_local_child_process", patch); + continue; + } + + const processPid = run.processPid ?? candidate.processPid; + const processGroupId = run.processGroupId ?? candidate.processGroupId; + const processPidAlive = isProcessAlive(processPid); + const processGroupAlive = isProcessGroupAlive(processGroupId); + if (!processPid && !processGroupId) { + classify(candidate, "lost", "missing_process_metadata", patch); + continue; + } + if (!processPidAlive && !processGroupAlive) { + classify(candidate, "lost", "process_not_alive", patch); + continue; + } + + const resultJson = mergeHotRestartAdoptionResultJson(parseObject(run.resultJson), { + adoptedAt: now, + previousServerPid: intent.previousServerPid, + newServerPid: process.pid, + previousServerVersion: intent.previousServerVersion, + newServerVersion: serverVersion, + processPid, + processGroupId, + }); + const updated = await db + .update(heartbeatRuns) + .set({ + resultJson, + error: run.errorCode === DETACHED_PROCESS_ERROR_CODE ? null : run.error, + errorCode: run.errorCode === DETACHED_PROCESS_ERROR_CODE ? null : run.errorCode, + updatedAt: now, + }) + .where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.status, "running"))) + .returning() + .then((rows) => rows[0] ?? null); + + if (!updated) { + const latest = await db + .select({ status: heartbeatRuns.status }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, run.id)) + .then((rows) => rows[0] ?? null); + if (latest && latest.status !== "running") { + classify(candidate, "finalized_while_down", `run_status_${latest.status}`, patch); + } else { + classify(candidate, "lost", "adoption_update_not_applied", patch); + } + continue; + } + + await appendRunEvent(updated, await nextRunEventSeq(run.id), { + eventType: "lifecycle", + stream: "system", + level: "info", + message: "Adopted live child process after hot restart", + payload: { + previousServerPid: intent.previousServerPid, + newServerPid: process.pid, + previousServerVersion: intent.previousServerVersion, + newServerVersion: serverVersion, + processPid, + processGroupId, + }, + }); + classify(candidate, "adopted", processPidAlive ? "process_pid_alive" : "process_group_alive", patch); + } + + const report = await writeHotRestartReport({ + version: 1, + requestedAt: intent.requestedAt, + completedAt: now.toISOString(), + drainRequired: intent.drainRequired, + previousServerPid: intent.previousServerPid, + newServerPid: process.pid, + previousServerVersion: intent.previousServerVersion, + newServerVersion: serverVersion, + adoptedRunIds, + finalizedWhileDownRunIds, + lostRunIds, + skippedRunIds, + runs: reportRuns, + }); + await removeHotRestartIntent(); + + logger.info( + { + previousServerPid: report.previousServerPid, + newServerPid: report.newServerPid, + adoptedRunIds, + finalizedWhileDownRunIds, + lostRunIds, + skippedRunIds, + }, + "hot-restart adoption report written", + ); + + return { + mode: "reported" as const, + adoptedRunIds, + finalizedWhileDownRunIds, + lostRunIds, + skippedRunIds, + }; + } + async function drainRunningRunsForShutdown(signal: "SIGINT" | "SIGTERM", now = new Date()) { const activeRuns = await db .select({ @@ -10999,6 +11327,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const tracksLocalChild = isTrackedLocalChildProcessAdapter(adapterType); const processPidAlive = tracksLocalChild && run.processPid && isProcessAlive(run.processPid); const processGroupAlive = tracksLocalChild && run.processGroupId && isProcessGroupAlive(run.processGroupId); + if ( + (processPidAlive || processGroupAlive) && + readHotRestartAdoptionMetadata(parseObject(run.resultJson)) + ) { + continue; + } if (processPidAlive) { if (run.errorCode !== DETACHED_PROCESS_ERROR_CODE) { const detachedMessage = `Lost in-memory process handle, but child pid ${run.processPid} is still alive`; @@ -16505,6 +16839,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) reportRunActivity: clearDetachedRunWarning, + prepareHotRestartShutdown, + reconcileHotRestartAdoption, reapOrphanedRuns, // Override-aware scheduling-suppression check (honors the worktree // run-execution experimental setting). Callers outside the service that diff --git a/server/src/services/hot-restart.ts b/server/src/services/hot-restart.ts new file mode 100644 index 0000000000..7c0a4596ce --- /dev/null +++ b/server/src/services/hot-restart.ts @@ -0,0 +1,210 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { resolvePaperclipHomeDir } from "../home-paths.js"; + +export const HOT_RESTART_INTENT_FILENAME = "hot-restart-intent.json"; +export const HOT_RESTART_REPORT_FILENAME = "hot-restart-report.json"; + +export type HotRestartIntentRun = { + runId: string; + companyId: string; + agentId: string; + adapterType: string; + status: string; + processPid: number | null; + processGroupId: number | null; + issueId: string | null; +}; + +export type HotRestartIntent = { + version: 1; + requestedAt: string; + previousServerPid: number; + previousServerVersion: string | null; + drainRequired: boolean; + requestedByRunId: string | null; + shutdownSnapshot?: { + capturedAt: string; + signal: "SIGINT" | "SIGTERM"; + activeRuns: HotRestartIntentRun[]; + }; +}; + +export type HotRestartReportRun = HotRestartIntentRun & { + classification: + | "adopted" + | "finalized_while_down" + | "lost" + | "skipped"; + reason: string; +}; + +export type HotRestartReport = { + version: 1; + requestedAt: string; + completedAt: string; + drainRequired: boolean; + previousServerPid: number; + newServerPid: number; + previousServerVersion: string | null; + newServerVersion: string; + adoptedRunIds: string[]; + finalizedWhileDownRunIds: string[]; + lostRunIds: string[]; + skippedRunIds: string[]; + runs: HotRestartReportRun[]; +}; + +function resolveHotRestartPath(filename: string, homeDir?: string) { + return path.join(resolvePaperclipHomeDir(homeDir), filename); +} + +export function resolveHotRestartIntentPath(homeDir?: string) { + return resolveHotRestartPath(HOT_RESTART_INTENT_FILENAME, homeDir); +} + +export function resolveHotRestartReportPath(homeDir?: string) { + return resolveHotRestartPath(HOT_RESTART_REPORT_FILENAME, homeDir); +} + +async function writeJsonFileAtomic(filePath: string, value: unknown) { + await fs.mkdir(path.dirname(filePath), { recursive: true }); + const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`; + await fs.writeFile(tempPath, `${JSON.stringify(value, null, 2)}\n`, "utf8"); + await fs.rename(tempPath, filePath); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function asString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value : null; +} + +function asNumber(value: unknown): number | null { + return typeof value === "number" && Number.isInteger(value) && value > 0 ? value : null; +} + +function asBoolean(value: unknown): boolean { + return value === true; +} + +function parseRun(value: unknown): HotRestartIntentRun | null { + if (!isRecord(value)) return null; + const runId = asString(value.runId); + const companyId = asString(value.companyId); + const agentId = asString(value.agentId); + const adapterType = asString(value.adapterType); + const status = asString(value.status); + if (!runId || !companyId || !agentId || !adapterType || !status) return null; + return { + runId, + companyId, + agentId, + adapterType, + status, + processPid: asNumber(value.processPid), + processGroupId: asNumber(value.processGroupId), + issueId: asString(value.issueId), + }; +} + +export function parseHotRestartIntent(value: unknown): HotRestartIntent | null { + if (!isRecord(value) || value.version !== 1) return null; + const requestedAt = asString(value.requestedAt); + const previousServerPid = asNumber(value.previousServerPid); + if (!requestedAt || !previousServerPid) return null; + + const intent: HotRestartIntent = { + version: 1, + requestedAt, + previousServerPid, + previousServerVersion: asString(value.previousServerVersion), + drainRequired: asBoolean(value.drainRequired), + requestedByRunId: asString(value.requestedByRunId), + }; + + const snapshot = isRecord(value.shutdownSnapshot) ? value.shutdownSnapshot : null; + const signal = snapshot?.signal === "SIGINT" || snapshot?.signal === "SIGTERM" + ? snapshot.signal + : null; + const capturedAt = asString(snapshot?.capturedAt); + const activeRuns = Array.isArray(snapshot?.activeRuns) + ? snapshot.activeRuns.map(parseRun).filter((run): run is HotRestartIntentRun => run !== null) + : []; + if (signal && capturedAt) { + intent.shutdownSnapshot = { capturedAt, signal, activeRuns }; + } + + return intent; +} + +export async function readHotRestartIntent(homeDir?: string) { + try { + const raw = await fs.readFile(resolveHotRestartIntentPath(homeDir), "utf8"); + return parseHotRestartIntent(JSON.parse(raw)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return null; + throw error; + } +} + +export async function writeHotRestartIntent(input: { + previousServerPid: number; + previousServerVersion?: string | null; + drainRequired?: boolean; + requestedByRunId?: string | null; + requestedAt?: Date; + homeDir?: string; +}) { + const intent: HotRestartIntent = { + version: 1, + requestedAt: (input.requestedAt ?? new Date()).toISOString(), + previousServerPid: input.previousServerPid, + previousServerVersion: input.previousServerVersion ?? null, + drainRequired: input.drainRequired ?? false, + requestedByRunId: input.requestedByRunId ?? null, + }; + await writeJsonFileAtomic(resolveHotRestartIntentPath(input.homeDir), intent); + return intent; +} + +export async function writeHotRestartShutdownSnapshot(input: { + intent: HotRestartIntent; + signal: "SIGINT" | "SIGTERM"; + activeRuns: HotRestartIntentRun[]; + capturedAt?: Date; + homeDir?: string; +}) { + const updated: HotRestartIntent = { + ...input.intent, + shutdownSnapshot: { + capturedAt: (input.capturedAt ?? new Date()).toISOString(), + signal: input.signal, + activeRuns: input.activeRuns, + }, + }; + await writeJsonFileAtomic(resolveHotRestartIntentPath(input.homeDir), updated); + return updated; +} + +export async function writeHotRestartReport(report: HotRestartReport, homeDir?: string) { + await writeJsonFileAtomic(resolveHotRestartReportPath(homeDir), report); + return report; +} + +export async function removeHotRestartIntent(homeDir?: string) { + try { + await fs.unlink(resolveHotRestartIntentPath(homeDir)); + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; + } +} + +export function shouldHonorHotRestartIntentForProcess( + intent: HotRestartIntent, + pid = process.pid, +) { + return !intent.drainRequired && intent.previousServerPid === pid; +}