diff --git a/src/cli/gateway-cli/register-restart-handoff.test.ts b/src/cli/gateway-cli/register-restart-handoff.test.ts new file mode 100644 index 000000000000..17a1848e9246 --- /dev/null +++ b/src/cli/gateway-cli/register-restart-handoff.test.ts @@ -0,0 +1,170 @@ +// Gateway restart-handoff command registration and machine JSON contract tests. +import { Command } from "commander"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { runRegisteredCli } from "../../test-utils/command-runner.js"; +import { addGatewayRestartHandoffCommands } from "./register-restart-handoff.js"; + +const mocks = vi.hoisted(() => ({ + consumeGatewayRestartHandoffSync: vi.fn(), + defaultRuntime: { + error: vi.fn(), + writeJson: vi.fn(), + exit: vi.fn(), + }, +})); + +vi.mock("../../infra/restart-handoff.js", () => ({ + consumeGatewayRestartHandoffSync: (opts: unknown) => mocks.consumeGatewayRestartHandoffSync(opts), +})); + +vi.mock("../../runtime.js", () => ({ + defaultRuntime: mocks.defaultRuntime, +})); + +function registerGatewayRestartHandoffCli(program: Command): void { + const gateway = program.command("gateway"); + addGatewayRestartHandoffCommands(gateway); +} + +describe("gateway restart-handoff commands", () => { + beforeEach(() => { + mocks.consumeGatewayRestartHandoffSync.mockReset(); + mocks.defaultRuntime.error.mockReset(); + mocks.defaultRuntime.writeJson.mockReset(); + mocks.defaultRuntime.exit.mockReset(); + }); + + it("keeps the machine command group out of normal gateway help", () => { + const program = new Command(); + const gateway = program.command("gateway"); + addGatewayRestartHandoffCommands(gateway); + + expect(gateway.helpInformation()).not.toContain("restart-handoff"); + }); + + it("reports protocol version 1 capabilities", async () => { + await runRegisteredCli({ + register: registerGatewayRestartHandoffCli, + argv: ["gateway", "restart-handoff", "capabilities", "--json"], + }); + + expect(mocks.defaultRuntime.writeJson).toHaveBeenCalledWith({ + ok: true, + protocol: "openclaw.gateway.restart-handoff", + protocolVersion: 1, + operations: ["consume"], + }); + }); + + it("forwards a validated expected PID and wraps the consume result", async () => { + mocks.consumeGatewayRestartHandoffSync.mockReturnValue({ + status: "none", + reason: "missing", + }); + + await runRegisteredCli({ + register: registerGatewayRestartHandoffCli, + argv: ["gateway", "restart-handoff", "consume", "--expected-pid", "4242", "--json"], + }); + + expect(mocks.consumeGatewayRestartHandoffSync).toHaveBeenCalledWith({ + expectedPid: 4242, + }); + expect(mocks.defaultRuntime.writeJson).toHaveBeenCalledWith({ + ok: true, + protocol: "openclaw.gateway.restart-handoff", + protocolVersion: 1, + status: "none", + reason: "missing", + }); + }); + + it.each([ + { + name: "missing", + argv: ["gateway", "restart-handoff", "consume", "--json"], + }, + { + name: "valueless", + argv: ["gateway", "restart-handoff", "consume", "--expected-pid", "--json"], + }, + { + name: "unsafe", + argv: [ + "gateway", + "restart-handoff", + "consume", + "--expected-pid", + "9007199254740992", + "--json", + ], + }, + { + name: "dash-prefixed malformed", + argv: ["gateway", "restart-handoff", "consume", "--expected-pid", "-invalid", "--json"], + }, + { + name: "valid with an unknown option", + argv: [ + "gateway", + "restart-handoff", + "consume", + "--expected-pid", + "4242", + "--unknown", + "--json", + ], + }, + { + name: "repeated", + argv: [ + "gateway", + "restart-handoff", + "consume", + "--expected-pid", + "111", + "--expected-pid", + "222", + "--json", + ], + }, + ])("rejects $name expected PID values before touching the handoff store", async ({ argv }) => { + await runRegisteredCli({ + register: registerGatewayRestartHandoffCli, + argv, + }); + + expect(mocks.consumeGatewayRestartHandoffSync).not.toHaveBeenCalled(); + expect(mocks.defaultRuntime.writeJson).toHaveBeenCalledWith({ + ok: false, + protocol: "openclaw.gateway.restart-handoff", + protocolVersion: 1, + status: "error", + reason: "invalid-expected-pid", + }); + expect(mocks.defaultRuntime.exit).toHaveBeenCalledWith(2); + }); + + it("returns a stable machine error when the handoff store is unavailable", async () => { + mocks.consumeGatewayRestartHandoffSync.mockImplementation(() => { + throw new Error("database locked"); + }); + + await runRegisteredCli({ + register: registerGatewayRestartHandoffCli, + argv: ["gateway", "restart-handoff", "consume", "--expected-pid", "4242", "--json"], + }); + + expect(mocks.defaultRuntime.error).toHaveBeenCalledWith( + "Gateway restart handoff consume failed: Error: database locked", + ); + expect(mocks.defaultRuntime.writeJson).toHaveBeenCalledWith({ + ok: false, + protocol: "openclaw.gateway.restart-handoff", + protocolVersion: 1, + status: "error", + reason: "store-unavailable", + }); + expect(mocks.defaultRuntime.exit).toHaveBeenCalledWith(1); + }); +}); diff --git a/src/cli/gateway-cli/register-restart-handoff.ts b/src/cli/gateway-cli/register-restart-handoff.ts new file mode 100644 index 000000000000..f739057831e8 --- /dev/null +++ b/src/cli/gateway-cli/register-restart-handoff.ts @@ -0,0 +1,73 @@ +// Hidden machine-facing gateway restart-handoff commands for external supervisors. +import type { Command } from "commander"; +import { parseStrictPositiveInteger } from "../../infra/parse-finite-number.js"; +import { + createGatewayRestartHandoffCapabilities, + GATEWAY_RESTART_HANDOFF_PROTOCOL, + GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION, +} from "../../infra/restart-handoff-contract.js"; +import { defaultRuntime } from "../../runtime.js"; + +function writeRestartHandoffError(reason: "invalid-expected-pid" | "store-unavailable") { + defaultRuntime.writeJson({ + ok: false, + protocol: GATEWAY_RESTART_HANDOFF_PROTOCOL, + protocolVersion: GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION, + status: "error", + reason, + }); +} + +function collectExpectedPid(value: string, previous: unknown): unknown[] { + return Array.isArray(previous) ? [...previous, value] : [previous, value]; +} + +export function addGatewayRestartHandoffCommands(gateway: Command): void { + const restartHandoff = gateway.command("restart-handoff", { hidden: true }); + + restartHandoff + .command("capabilities") + .description("Report the gateway restart-handoff machine contract") + .option("--json", "Output JSON", false) + .action(() => { + defaultRuntime.writeJson({ + ok: true, + ...createGatewayRestartHandoffCapabilities(), + }); + }); + + restartHandoff + .command("consume") + .description("Atomically consume a gateway restart handoff") + .helpOption(false) + .allowUnknownOption() + .allowExcessArguments() + .option("--expected-pid [pid]", "PID of the exited gateway process", collectExpectedPid, []) + .option("--json", "Output JSON", false) + .action(async (opts, command: Command) => { + const expectedPidValues = Array.isArray(opts.expectedPid) ? opts.expectedPid : []; + const expectedPid = + command.args.length === 0 && expectedPidValues.length === 1 + ? parseStrictPositiveInteger(expectedPidValues[0]) + : undefined; + if (expectedPid === undefined || !Number.isSafeInteger(expectedPid)) { + writeRestartHandoffError("invalid-expected-pid"); + defaultRuntime.exit(2); + return; + } + try { + const { consumeGatewayRestartHandoffSync } = await import("../../infra/restart-handoff.js"); + const result = consumeGatewayRestartHandoffSync({ expectedPid }); + defaultRuntime.writeJson({ + ok: true, + protocol: GATEWAY_RESTART_HANDOFF_PROTOCOL, + protocolVersion: GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION, + ...result, + }); + } catch (err) { + defaultRuntime.error(`Gateway restart handoff consume failed: ${String(err)}`); + writeRestartHandoffError("store-unavailable"); + defaultRuntime.exit(1); + } + }); +} diff --git a/src/cli/gateway-cli/register.ts b/src/cli/gateway-cli/register.ts index afcbd22816ce..1690fd61a92d 100644 --- a/src/cli/gateway-cli/register.ts +++ b/src/cli/gateway-cli/register.ts @@ -25,6 +25,7 @@ import { formatHelpExamples } from "../help-format.js"; import { parseTimeoutMsWithFallback } from "../parse-timeout.js"; import type { GatewayRpcOpts } from "./call.js"; import type { GatewayDiscoverOpts } from "./discover.js"; +import { addGatewayRestartHandoffCommands } from "./register-restart-handoff.js"; import { addGatewayRunCommand } from "./run-command.js"; const configModuleLoader = createLazyImportLoader( @@ -573,6 +574,7 @@ export function registerGatewayCli(program: Command) { addGatewayServiceCommands(gateway, { statusDescription: "Show gateway service status + probe connectivity/capability", }); + addGatewayRestartHandoffCommands(gateway); gatewayCallOpts( gateway diff --git a/src/infra/restart-handoff-contract.ts b/src/infra/restart-handoff-contract.ts new file mode 100644 index 000000000000..1e6e2da86d6b --- /dev/null +++ b/src/infra/restart-handoff-contract.ts @@ -0,0 +1,11 @@ +// Stable machine contract for external supervisors consuming gateway restart handoffs. +export const GATEWAY_RESTART_HANDOFF_PROTOCOL = "openclaw.gateway.restart-handoff"; +export const GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION = 1; + +export function createGatewayRestartHandoffCapabilities() { + return { + protocol: GATEWAY_RESTART_HANDOFF_PROTOCOL, + protocolVersion: GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION, + operations: ["consume"] as const, + }; +} diff --git a/src/infra/restart-handoff.test.ts b/src/infra/restart-handoff.test.ts index 2f8d2d2c4b01..31e47640cf3c 100644 --- a/src/infra/restart-handoff.test.ts +++ b/src/infra/restart-handoff.test.ts @@ -1,8 +1,9 @@ // Covers gateway restart handoff persistence and diagnostics. +import { spawn } from "node:child_process"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; import { closeOpenClawStateDatabaseForTest, @@ -14,6 +15,7 @@ import { getNodeSqliteKysely, } from "./kysely-sync.js"; import { + consumeGatewayRestartHandoffSync, formatGatewayRestartHandoffDiagnostic, readGatewayRestartHandoffSync, writeGatewayRestartHandoffSync, @@ -115,6 +117,88 @@ function expectWrittenHandoff( return handoff; } +function spawnHandoffConsumer(params: { + env: NodeJS.ProcessEnv; + expectedPid: number; + now: number; + startFile: string; +}): Promise { + const moduleUrl = new URL("./restart-handoff.ts", import.meta.url).href; + const script = ` + import fs from "node:fs"; + while (!fs.existsSync(process.env.OPENCLAW_HANDOFF_TEST_START_FILE)) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + const mod = await import(process.env.OPENCLAW_HANDOFF_TEST_MODULE_URL); + const result = mod.consumeGatewayRestartHandoffSync({ + expectedPid: Number(process.env.OPENCLAW_HANDOFF_TEST_EXPECTED_PID), + now: Number(process.env.OPENCLAW_HANDOFF_TEST_NOW), + env: process.env, + }); + process.stdout.write(JSON.stringify(result)); + `; + return new Promise((resolve, reject) => { + let settled = false; + const child = spawn( + process.execPath, + ["--import", "tsx", "--input-type=module", "--eval", script], + { + cwd: process.cwd(), + env: { + ...params.env, + OPENCLAW_HANDOFF_TEST_EXPECTED_PID: String(params.expectedPid), + OPENCLAW_HANDOFF_TEST_MODULE_URL: moduleUrl, + OPENCLAW_HANDOFF_TEST_NOW: String(params.now), + OPENCLAW_HANDOFF_TEST_START_FILE: params.startFile, + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + const timeout = setTimeout(() => { + if (settled) { + return; + } + settled = true; + child.kill(); + reject(new Error("handoff consumer timed out")); + }, 10_000); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + child.stdout.on("data", (chunk: string) => { + stdout += chunk; + }); + child.stderr.on("data", (chunk: string) => { + stderr += chunk; + }); + child.once("error", (err) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + reject(err); + }); + child.once("exit", (code) => { + if (settled) { + return; + } + settled = true; + clearTimeout(timeout); + if (code !== 0) { + reject(new Error(`handoff consumer exited ${code}: ${stderr}`)); + return; + } + try { + resolve(JSON.parse(stdout)); + } catch (err) { + reject(new Error(`invalid handoff consumer output: ${stdout}`, { cause: err })); + } + }); + }); +} + describe("gateway restart handoff", () => { afterEach(() => { closeOpenClawStateDatabaseForTest(); @@ -244,7 +328,7 @@ describe("gateway restart handoff", () => { createdAt: 1_000, ttlMs: 1_000, }); - expect(readGatewayRestartHandoffSync(env, 2_001)).toBeNull(); + expect(readGatewayRestartHandoffSync(env, 2_000)).toBeNull(); }); it("rejects persisted handoffs with a ttl longer than the supported window", () => { @@ -283,4 +367,211 @@ describe("gateway restart handoff", () => { expect(readGatewayRestartHandoffSync(env)?.pid).toBe(67_890); expect(fs.existsSync(legacyHandoffPath(env))).toBe(false); }); + + it("atomically accepts and removes a matching handoff", () => { + const env = createHandoffEnv(); + const handoff = expectWrittenHandoff({ + env, + pid: 12_345, + reason: "gateway.restart", + restartKind: "full-process", + supervisorMode: "external", + createdAt: 1_000, + }); + + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 12_345, + now: 1_500, + }), + ).toStrictEqual({ + status: "accepted", + handoff, + }); + expect(readHandoffRow(env)).toBeUndefined(); + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 12_345, + now: 1_500, + }), + ).toStrictEqual({ + status: "none", + reason: "missing", + }); + }); + + it("retains a PID-mismatched handoff for the matching consumer", () => { + const env = createHandoffEnv(); + expectWrittenHandoff({ + env, + pid: 12_345, + restartKind: "full-process", + supervisorMode: "external", + createdAt: 1_000, + }); + + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 54_321, + now: 1_500, + }), + ).toStrictEqual({ + status: "rejected", + reason: "pid-mismatch", + handoffPid: 12_345, + }); + expect(readHandoffRow(env)?.pid).toBe(12_345); + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 12_345, + now: 1_500, + }), + ).toMatchObject({ + status: "accepted", + handoff: { pid: 12_345 }, + }); + }); + + it.each([ + { + name: "expired", + insert: (env: NodeJS.ProcessEnv) => + expectWrittenHandoff({ + env, + pid: 12_345, + restartKind: "full-process", + supervisorMode: "external", + createdAt: 1_000, + ttlMs: 1_000, + }), + now: 2_000, + expected: { + status: "rejected", + reason: "expired", + handoffPid: 12_345, + }, + }, + { + name: "malformed", + insert: (env: NodeJS.ProcessEnv) => + insertHandoffRow(env, { + pid: 12_345, + source: "invalid-source", + createdAt: 1_000, + expiresAt: 61_000, + }), + now: 1_500, + expected: { + status: "rejected", + reason: "invalid", + }, + }, + { + name: "future-dated", + insert: (env: NodeJS.ProcessEnv) => + expectWrittenHandoff({ + env, + pid: 12_345, + restartKind: "full-process", + supervisorMode: "external", + createdAt: 2_000, + }), + now: 1_500, + expected: { + status: "rejected", + reason: "invalid", + }, + }, + ])("removes a $name handoff after rejecting it", ({ insert, now, expected }) => { + const env = createHandoffEnv(); + insert(env); + + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 12_345, + now, + }), + ).toStrictEqual(expected); + expect(readHandoffRow(env)).toBeUndefined(); + }); + + it("accepts a handoff exactly once across concurrent consumers", async () => { + const env = createHandoffEnv(); + expectWrittenHandoff({ + env, + pid: 12_345, + restartKind: "full-process", + supervisorMode: "external", + createdAt: 1_000, + }); + closeOpenClawStateDatabaseForTest(); + const startFile = path.join(env.OPENCLAW_STATE_DIR ?? "", "start-consumers"); + + const first = spawnHandoffConsumer({ + env, + expectedPid: 12_345, + now: 1_500, + startFile, + }); + const second = spawnHandoffConsumer({ + env, + expectedPid: 12_345, + now: 1_500, + startFile, + }); + fs.writeFileSync(startFile, "start", "utf8"); + + const results = await Promise.all([first, second]); + expect( + results.map((result) => (result as { status?: string }).status).toSorted(), + ).toStrictEqual(["accepted", "none"]); + expect(readHandoffRow(env)).toBeUndefined(); + }); + + it("samples the default time after beginning the write transaction", () => { + const env = createHandoffEnv(); + expectWrittenHandoff({ + env, + pid: 12_345, + restartKind: "full-process", + supervisorMode: "external", + createdAt: 1_000, + ttlMs: 1_000, + }); + const { db } = openOpenClawStateDatabase({ env }); + const originalExec = db.exec.bind(db); + let transactionBegan = false; + const execSpy = vi.spyOn(db, "exec").mockImplementation((sql) => { + if (sql === "BEGIN IMMEDIATE") { + transactionBegan = true; + } + return originalExec(sql); + }); + const nowSpy = vi + .spyOn(Date, "now") + .mockImplementation(() => (transactionBegan ? 2_000 : 1_500)); + + try { + expect( + consumeGatewayRestartHandoffSync({ + env, + expectedPid: 12_345, + }), + ).toStrictEqual({ + status: "rejected", + reason: "expired", + handoffPid: 12_345, + }); + } finally { + nowSpy.mockRestore(); + execSpy.mockRestore(); + } + + expect(readHandoffRow(env)).toBeUndefined(); + }); }); diff --git a/src/infra/restart-handoff.ts b/src/infra/restart-handoff.ts index 4e0c87c69959..61082ffdc458 100644 --- a/src/infra/restart-handoff.ts +++ b/src/infra/restart-handoff.ts @@ -1,5 +1,6 @@ // Persists short-lived gateway restart handoff metadata. import { randomUUID } from "node:crypto"; +import type { DatabaseSync } from "node:sqlite"; import { truncateUtf16Safe } from "@openclaw/normalization-core/utf16-slice"; import { createSubsystemLogger } from "../logging/subsystem.js"; import type { DB as OpenClawStateKyselyDatabase } from "../state/openclaw-state-db.generated.js"; @@ -13,10 +14,17 @@ import { getNodeSqliteKysely, } from "./kysely-sync.js"; +export { + createGatewayRestartHandoffCapabilities, + GATEWAY_RESTART_HANDOFF_PROTOCOL, + GATEWAY_RESTART_HANDOFF_PROTOCOL_VERSION, +} from "./restart-handoff-contract.js"; + // Restart handoff rows let a supervisor explain a recent gateway restart after // the old process exits. The row is short-lived, bounded, and replaced on write. const GATEWAY_SUPERVISOR_RESTART_HANDOFF_KIND = "gateway-supervisor-restart-handoff"; const GATEWAY_SUPERVISOR_RESTART_HANDOFF_KEY = "current"; +const GATEWAY_RESTART_HANDOFF_SCHEMA_VERSION = 1; const GATEWAY_RESTART_HANDOFF_TTL_MS = 60_000; const GATEWAY_RESTART_TRACE_HANDOFF_MAX_DURATION_MS = 10 * 60_000; const MAX_INTENT_ID_LENGTH = 120; @@ -25,6 +33,21 @@ const MAX_REASON_LENGTH = 200; const handoffLog = createSubsystemLogger("restart-handoff"); type GatewayRestartHandoffDatabase = Pick; +type GatewayRestartHandoffRow = { + kind: string; + version: number; + intent_id: string; + pid: number; + process_instance_id: string | null; + created_at: number; + expires_at: number; + reason: string | null; + restart_trace_started_at: number | null; + restart_trace_last_at: number | null; + source: string; + restart_kind: string; + supervisor_mode: string; +}; type GatewayRestartHandoffRestartKind = "full-process" | "update-process"; type GatewayRestartHandoffSource = @@ -38,7 +61,7 @@ type GatewayRestartHandoffSupervisorMode = "launchd" | "systemd" | "schtasks" | export type GatewayRestartHandoff = { kind: typeof GATEWAY_SUPERVISOR_RESTART_HANDOFF_KIND; - version: 1; + version: typeof GATEWAY_RESTART_HANDOFF_SCHEMA_VERSION; intentId: string; pid: number; processInstanceId?: string; @@ -54,6 +77,21 @@ export type GatewayRestartHandoff = { }; }; +export type GatewayRestartHandoffConsumeResult = + | { + status: "accepted"; + handoff: GatewayRestartHandoff; + } + | { + status: "none"; + reason: "missing"; + } + | { + status: "rejected"; + reason: "expired" | "invalid" | "pid-mismatch"; + handoffPid?: number; + }; + function formatShortDuration(ms: number): string { const clamped = Math.max(0, Math.floor(ms)); if (clamped < 1000) { @@ -197,25 +235,13 @@ function isSupervisorMode(value: unknown): value is GatewayRestartHandoffSupervi return value === "launchd" || value === "systemd" || value === "schtasks" || value === "external"; } -function normalizeGatewayRestartHandoffRow(row: { - kind: string; - version: number; - intent_id: string; - pid: number; - process_instance_id: string | null; - created_at: number; - expires_at: number; - reason: string | null; - restart_trace_started_at: number | null; - restart_trace_last_at: number | null; - source: string; - restart_kind: string; - supervisor_mode: string; -}): GatewayRestartHandoff | null { +function normalizeGatewayRestartHandoffRow( + row: GatewayRestartHandoffRow, +): GatewayRestartHandoff | null { const intentId = normalizeText(row.intent_id, MAX_INTENT_ID_LENGTH); if ( row.kind !== GATEWAY_SUPERVISOR_RESTART_HANDOFF_KIND || - row.version !== 1 || + row.version !== GATEWAY_RESTART_HANDOFF_SCHEMA_VERSION || !intentId || typeof row.pid !== "number" || !Number.isSafeInteger(row.pid) || @@ -242,7 +268,7 @@ function normalizeGatewayRestartHandoffRow(row: { const reason = normalizeText(row.reason, MAX_REASON_LENGTH); return { kind: GATEWAY_SUPERVISOR_RESTART_HANDOFF_KIND, - version: 1, + version: GATEWAY_RESTART_HANDOFF_SCHEMA_VERSION, intentId, pid: row.pid, ...(processInstanceId ? { processInstanceId } : {}), @@ -256,31 +282,37 @@ function normalizeGatewayRestartHandoffRow(row: { }; } +function selectGatewayRestartHandoffRowSync( + db: DatabaseSync, +): GatewayRestartHandoffRow | undefined { + const stateDb = getNodeSqliteKysely(db); + return executeSqliteQueryTakeFirstSync( + db, + stateDb + .selectFrom("gateway_restart_handoff") + .select([ + "kind", + "version", + "intent_id", + "pid", + "process_instance_id", + "created_at", + "expires_at", + "reason", + "restart_trace_started_at", + "restart_trace_last_at", + "source", + "restart_kind", + "supervisor_mode", + ]) + .where("handoff_key", "=", GATEWAY_SUPERVISOR_RESTART_HANDOFF_KEY), + ); +} + function readGatewayRestartHandoffRowSync(env: NodeJS.ProcessEnv) { try { const { db } = openOpenClawStateDatabase({ env }); - const stateDb = getNodeSqliteKysely(db); - return executeSqliteQueryTakeFirstSync( - db, - stateDb - .selectFrom("gateway_restart_handoff") - .select([ - "kind", - "version", - "intent_id", - "pid", - "process_instance_id", - "created_at", - "expires_at", - "reason", - "restart_trace_started_at", - "restart_trace_last_at", - "source", - "restart_kind", - "supervisor_mode", - ]) - .where("handoff_key", "=", GATEWAY_SUPERVISOR_RESTART_HANDOFF_KEY), - ); + return selectGatewayRestartHandoffRowSync(db); } catch { return null; } @@ -319,7 +351,7 @@ export function writeGatewayRestartHandoffSync(opts: { const restartTrace = normalizeRestartTraceHandoff(opts.restartTrace); const payload: GatewayRestartHandoff = { kind: GATEWAY_SUPERVISOR_RESTART_HANDOFF_KIND, - version: 1, + version: GATEWAY_RESTART_HANDOFF_SCHEMA_VERSION, intentId: randomUUID(), pid, ...(processInstanceId ? { processInstanceId } : {}), @@ -393,8 +425,82 @@ export function readGatewayRestartHandoffSync( ): GatewayRestartHandoff | null { const row = readGatewayRestartHandoffRowSync(env); const payload = row ? normalizeGatewayRestartHandoffRow(row) : null; - if (!payload || now < payload.createdAt || now > payload.expiresAt) { + if (!payload || now < payload.createdAt || now >= payload.expiresAt) { return null; } return payload; } + +/** + * Atomically validate and consume the current restart handoff for one exited process. + * + * PID mismatches are retained for the matching supervisor. Accepted, expired, and + * malformed rows are removed while the immediate transaction still owns the write lock. + */ +export function consumeGatewayRestartHandoffSync(opts: { + env?: NodeJS.ProcessEnv; + expectedPid: number; + now?: number; +}): GatewayRestartHandoffConsumeResult { + const expectedPid = normalizePid(opts.expectedPid); + if (expectedPid === null) { + throw new Error("expectedPid must be a positive safe integer"); + } + const fixedNow = + typeof opts.now === "number" && Number.isFinite(opts.now) && opts.now >= 0 + ? Math.floor(opts.now) + : undefined; + + return runOpenClawStateWriteTransaction( + ({ db }) => { + const now = fixedNow ?? Date.now(); + const stateDb = getNodeSqliteKysely(db); + const row = selectGatewayRestartHandoffRowSync(db); + if (!row) { + return { + status: "none", + reason: "missing", + }; + } + + const removeCurrent = () => { + executeSqliteQuerySync( + db, + stateDb + .deleteFrom("gateway_restart_handoff") + .where("handoff_key", "=", GATEWAY_SUPERVISOR_RESTART_HANDOFF_KEY), + ); + }; + const handoff = normalizeGatewayRestartHandoffRow(row); + if (!handoff || now < handoff.createdAt) { + removeCurrent(); + return { + status: "rejected", + reason: "invalid", + }; + } + if (now >= handoff.expiresAt) { + removeCurrent(); + return { + status: "rejected", + reason: "expired", + handoffPid: handoff.pid, + }; + } + if (handoff.pid !== expectedPid) { + return { + status: "rejected", + reason: "pid-mismatch", + handoffPid: handoff.pid, + }; + } + + removeCurrent(); + return { + status: "accepted", + handoff, + }; + }, + { env: opts.env ?? process.env }, + ); +}