import { randomBytes } from "node:crypto"; import type { GatewaySuspendPrepareResult, GatewaySuspendResumeResult, } from "../../../packages/gateway-protocol/src/index.js"; import { colorize, isRich, theme } from "../../../packages/terminal-core/src/theme.js"; import type { OutputRuntimeEnv } from "../../runtime.js"; import { formatCliCommand } from "../command-format.js"; import type { callGatewayFromCliWithTransport } from "../gateway-rpc.js"; type SuspendRpcOpts = Parameters[1]; type SuspendRpcCall = (method: string, opts: SuspendRpcOpts, params?: unknown) => Promise; type SuspendCliDeps = { callGateway: SuspendRpcCall; runtime: OutputRuntimeEnv; nowMs?: () => number; sleep?: (delayMs: number) => Promise; }; const MIN_SUSPEND_POLL_DELAY_MS = 50; function parseWaitMs(value: string | number | undefined): number | undefined { if (value === undefined) { return undefined; } const seconds = typeof value === "number" ? value : Number(value.trim() || Number.NaN); if (!Number.isFinite(seconds) || seconds < 0) { throw new Error("--wait must be a non-negative number of seconds"); } const milliseconds = Math.floor(seconds * 1_000); if (!Number.isSafeInteger(milliseconds)) { throw new Error("--wait is too large"); } return milliseconds; } function resolveRequestId(value: string | undefined): string { if (value === undefined) { return `cli-${randomBytes(4).toString("hex")}`; } const requestId = value.trim(); if (!requestId || requestId.length > 128) { throw new Error("--request-id must contain 1 to 128 characters"); } return requestId; } function formatBusyResult( result: Extract, ): string { const blockers = result.blockers.map((blocker) => `- ${blocker.message}`); return [ `Gateway suspension is busy (${result.reason}; ${result.activeCount} active).`, ...(blockers.length > 0 ? ["Blockers:", ...blockers] : []), ].join("\n"); } export async function runGatewaySuspend( options: { rpcOpts: SuspendRpcOpts; requestId?: string; waitSeconds?: string | number; json?: boolean; }, deps: SuspendCliDeps, ): Promise { const nowMs = deps.nowMs ?? Date.now; const sleep = deps.sleep ?? (async (delayMs: number) => await new Promise((resolve) => { setTimeout(resolve, delayMs); })); const requestId = resolveRequestId(options.requestId); const waitMs = parseWaitMs(options.waitSeconds); const deadlineMs = waitMs === undefined ? undefined : nowMs() + waitMs; const maxAttempts = waitMs === undefined ? 1 : Math.ceil(waitMs / MIN_SUSPEND_POLL_DELAY_MS) + 1; let latest: GatewaySuspendPrepareResult | undefined; for (let attempt = 0; attempt < maxAttempts; attempt += 1) { // A sleep can overshoot the deadline; never issue a prepare that could // suspend the Gateway after the operator's advertised --wait window. if (attempt > 0 && deadlineMs !== undefined && nowMs() >= deadlineMs) { break; } latest = (await deps.callGateway("gateway.suspend.prepare", options.rpcOpts, { requestId, })) as GatewaySuspendPrepareResult; if (latest.status === "ready") { if (options.json) { deps.runtime.writeJson({ ...latest, requestId }); return; } const rich = isRich(); deps.runtime.log(colorize(rich, theme.success, "Gateway suspension prepared.")); deps.runtime.log(`${colorize(rich, theme.muted, "Suspension ID:")} ${latest.suspensionId}`); deps.runtime.log( `${colorize(rich, theme.muted, "Expires:")} ${new Date(latest.expiresAtMs).toISOString()} (${latest.expiresAtMs} ms)`, ); const port = options.rpcOpts.localPortOverride; const command = `openclaw gateway resume ${latest.suspensionId}`; deps.runtime.log( `Resume with: ${formatCliCommand(port === undefined ? command : `${command} --port ${port}`)}`, ); return; } if (latest.status === "draining") { throw new Error("Gateway suspension unexpectedly entered drain mode"); } if (deadlineMs === undefined) { if (options.json) { deps.runtime.writeJson({ ...latest, requestId }); deps.runtime.exit(1); return; } throw new Error(`${formatBusyResult(latest)}\nRetry later or use --wait .`); } const remainingMs = deadlineMs - nowMs(); if (remainingMs <= 0) { break; } const delayMs = Math.min(remainingMs, Math.max(MIN_SUSPEND_POLL_DELAY_MS, latest.retryAfterMs)); await sleep(delayMs); } if (!latest || latest.status !== "busy") { throw new Error("Gateway suspension polling ended without a result"); } if (options.json) { deps.runtime.writeJson({ ...latest, requestId }); deps.runtime.exit(1); return; } throw new Error(`${formatBusyResult(latest)}\nTimed out waiting for the Gateway to become idle.`); } export async function runGatewayResume( options: { rpcOpts: SuspendRpcOpts; suspensionId: string; json?: boolean }, deps: Pick, ): Promise { const result = (await deps.callGateway("gateway.suspend.resume", options.rpcOpts, { suspensionId: options.suspensionId, })) as GatewaySuspendResumeResult; if (options.json) { deps.runtime.writeJson(result); return; } deps.runtime.log( result.resumed ? "Gateway resumed." : "No matching suspension was held (lease already expired or resumed); gateway is running.", ); }