vultr/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-24 09:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add restart handoff consume protocol

Browse Source
This commit is contained in:
Shakker
2026-07-16 15:21:08 +01:00
committed by Shakker
parent e123606b46
commit 333c4f9a61
6 changed files with 697 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
src/cli/gateway-cli/register-restart-handoff.test.ts Normal file
View File

@@ -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);
});
});

73
src/cli/gateway-cli/register-restart-handoff.ts Normal file
View File

@@ -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);
}
});
}

2
src/cli/gateway-cli/register.ts
View File

@@ -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

11
src/infra/restart-handoff-contract.ts Normal file
View File

@@ -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,
};
}

295
src/infra/restart-handoff.test.ts
View File

@@ -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<unknown> {
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();
});
});

190
src/infra/restart-handoff.ts
View File

@@ -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<OpenClawStateKyselyDatabase, "gateway_restart_handoff">;
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<GatewayRestartHandoffDatabase>(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<GatewayRestartHandoffDatabase>(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<GatewayRestartHandoffDatabase>(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 },
);
}