fix(gateway): keep setup.detect off the event loop so health stays live (#110625)

This commit is contained in:
Peter Steinberger
2026-07-18 12:19:59 +01:00
committed by GitHub
parent 01a9e1d398
commit 0bbb862c55
15 changed files with 404 additions and 14 deletions
+2
View File
@@ -113,6 +113,8 @@ const rootEntries = [
"src/audit/audit-event-writer.worker.ts!",
"src/state/openclaw-database-verify.worker.ts!",
"src/agents/model-provider-auth.worker.ts!",
// Loaded by URL from setup-inference-detection.ts; no static import edge exists.
"src/system-agent/setup-inference-detection.worker.ts!",
// Split runtime loaded through a path assembled in subagent-registry.ts.
"src/agents/subagent-registry.runtime.ts!",
// Loaded lazily by the registry; its callbacks form the orphan-recovery runtime contract.
+1
View File
@@ -114,6 +114,7 @@ const requiredPathGroups = [
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/state/openclaw-database-verify.worker.js",
"dist/system-agent/setup-inference-detection.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
"dist/build-info.json",
+20
View File
@@ -36,6 +36,26 @@ describe("detectInferenceBackends", () => {
expect(candidates).toEqual([]);
});
it("does not offer external CLIs whose version probes time out", async () => {
const candidates = await detectInferenceBackends({
env: {},
platform: "linux",
deps: {
probeLocalCommand: async (command) => ({
command,
found: true,
timedOut: true,
error: "timed out after 1500ms",
}),
readClaudeCliCredentials: () => ({ type: "oauth" }),
readCodexCliCredentials: () => ({ type: "oauth" }),
readGeminiCliCredentials: () => ({ type: "oauth" }),
},
});
expect(candidates).toEqual([]);
});
it("orders the ladder: existing model, env keys, then CLI logins", async () => {
const candidates = await detectInferenceBackends({
config: { agents: { defaults: { model: "zai/glm-5.2" } } },
+3 -3
View File
@@ -237,7 +237,7 @@ export async function detectInferenceBackends(
probe("gemini"),
]);
const cliCandidates: InferenceBackendCandidate[] = [];
if (claudeProbe.found) {
if (claudeProbe.found && !claudeProbe.timedOut) {
const credentials = detectCliCredentialState({
probe: claudeProbe,
hasStoredCredentials: readClaude() !== null,
@@ -251,7 +251,7 @@ export async function detectInferenceBackends(
...(credentials === undefined ? {} : { credentials }),
});
}
if (codexProbe.found) {
if (codexProbe.found && !codexProbe.timedOut) {
const credentials = options.deps?.readCodexCliCredentials
? detectCliCredentialState({
probe: codexProbe,
@@ -267,7 +267,7 @@ export async function detectInferenceBackends(
...(credentials === undefined ? {} : { credentials }),
});
}
if (geminiProbe.found) {
if (geminiProbe.found && !geminiProbe.timedOut) {
// Gemini CLI stores its OAuth login in a plain file on every platform (no
// keychain), so a missing credential file is a definitive logout signal.
const credentials = readGemini() !== null;
@@ -29,6 +29,9 @@ const setupInferenceMocks = vi.hoisted(() => ({
detectSetupInference: vi.fn(),
verifySetupInference: vi.fn(),
}));
const setupInferenceDetectionMocks = vi.hoisted(() => ({
detectSetupInferenceIsolated: vi.fn(),
}));
const providerAuthChoiceMocks = vi.hoisted(() => ({
applyAuthChoiceLoadedPluginProvider: vi.fn(),
}));
@@ -42,6 +45,9 @@ vi.mock("../../system-agent/setup-inference.js", () => ({
detectSetupInference: setupInferenceMocks.detectSetupInference,
verifySetupInference: setupInferenceMocks.verifySetupInference,
}));
vi.mock("../../system-agent/setup-inference-detection.js", () => ({
detectSetupInferenceIsolated: setupInferenceDetectionMocks.detectSetupInferenceIsolated,
}));
vi.mock("../../plugins/provider-auth-choice.js", () => ({
applyAuthChoiceLoadedPluginProvider: providerAuthChoiceMocks.applyAuthChoiceLoadedPluginProvider,
}));
@@ -168,6 +174,7 @@ afterEach(() => {
vi.restoreAllMocks();
setupInferenceMocks.activateSetupInference.mockReset();
setupInferenceMocks.detectSetupInference.mockReset();
setupInferenceDetectionMocks.detectSetupInferenceIsolated.mockReset();
setupInferenceMocks.verifySetupInference.mockReset();
providerAuthChoiceMocks.applyAuthChoiceLoadedPluginProvider.mockReset();
setupSharedMocks.readSetupConfigFileSnapshot.mockReset();
@@ -443,10 +450,10 @@ describe("openclaw.chat", () => {
expect(secondCall.ok).toBe(true);
});
it("tracks setup detection until its RPC response is sent", async () => {
it("keeps read-only setup detection outside the serialized system-agent lane", async () => {
const started = createDeferred();
const release = createDeferred();
setupInferenceMocks.detectSetupInference.mockImplementation(async () => {
setupInferenceDetectionMocks.detectSetupInferenceIsolated.mockImplementation(async () => {
started.resolve();
await release.promise;
return {
@@ -472,11 +479,11 @@ describe("openclaw.chat", () => {
} as never);
await started.promise;
expect(getCommandLaneSnapshot(CommandLane.SystemAgent).activeCount).toBe(1);
expect(getCommandLaneSnapshot(CommandLane.SystemAgent).activeCount).toBe(0);
release.resolve();
await pending;
expect(activeAtResponse).toEqual([1]);
expect(activeAtResponse).toEqual([0]);
expect(getCommandLaneSnapshot(CommandLane.SystemAgent).activeCount).toBe(0);
});
+5 -4
View File
@@ -213,10 +213,11 @@ export const systemAgentHandlers: GatewayRequestHandlers = {
) {
return;
}
await runSystemAgentGatewayTask(async () => {
const { detectSetupInference } = await import("../../system-agent/setup-inference.js");
respond(true, await detectSetupInference(), undefined);
});
// Detection is read-only and may load native provider code. Keep it outside
// the mutation lane and off the Gateway event loop so health stays live.
const { detectSetupInferenceIsolated } =
await import("../../system-agent/setup-inference-detection.js");
respond(true, await detectSetupInferenceIsolated(), undefined);
},
/** Re-run the exact current default-agent inference route without mutating setup. */
"openclaw.setup.verify": async ({ params, respond }) => {
+1
View File
@@ -103,6 +103,7 @@ describe("tsdown config", () => {
"agents/compaction-planning.worker",
"agents/model-provider-auth.worker",
"state/openclaw-database-verify.worker",
"system-agent/setup-inference-detection.worker",
"plugins/memory-state",
"subagent-registry.runtime",
"task-registry-control.runtime",
+1
View File
@@ -34,6 +34,7 @@ describe("openclaw probes", () => {
command: process.execPath,
error: "timed out after 25ms",
found: true,
timedOut: true,
});
expect(Date.now() - startedAt).toBeLessThan(2_000);
},
+2
View File
@@ -14,6 +14,7 @@ export type LocalCommandProbe = {
found: boolean;
version?: string;
error?: string;
timedOut?: boolean;
};
const LOCAL_COMMAND_PROBE_OUTPUT_MAX_CHARS = 16 * 1024;
@@ -36,6 +37,7 @@ export async function probeLocalCommand(
command,
found: true,
error: `timed out after ${timeoutMs}ms`,
timedOut: true,
};
}
// Version output can arrive on stdout or stderr depending on the CLI.
@@ -0,0 +1,125 @@
import { createServer, get } from "node:http";
import type { AddressInfo } from "node:net";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { SetupInferenceDetection } from "./setup-inference.js";
const blockingWorkerUrl = new URL(
`data:text/javascript,${encodeURIComponent(`
import { parentPort, workerData } from "node:worker_threads";
parentPort.postMessage({ type: "partial", detection: workerData.partialDetection });
const deadline = Date.now() + workerData.blockMs;
while (Date.now() < deadline) {}
parentPort.postMessage({ type: "result", detection: workerData.detection });
parentPort.close();
`)}`,
);
function emptyDetection(): SetupInferenceDetection {
return {
candidates: [],
unavailableCandidates: [],
manualProviders: [],
authOptions: [],
recommendedInstalls: [],
workspace: "/tmp/work",
setupComplete: false,
};
}
const servers = new Set<ReturnType<typeof createServer>>();
beforeEach(() => {
vi.resetModules();
});
async function loadDetectionModule() {
return await import("./setup-inference-detection.js");
}
async function requestHealth(url: string): Promise<{ body: string; statusCode: number }> {
return await new Promise((resolve, reject) => {
const request = get(url, { agent: false, headers: { connection: "close" } }, (response) => {
let body = "";
response.setEncoding("utf8");
response.on("data", (chunk) => {
body += chunk;
});
response.on("end", () => resolve({ body, statusCode: response.statusCode ?? 0 }));
});
request.on("error", reject);
});
}
afterEach(async () => {
await Promise.all(
[...servers].map(
(server) =>
new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
}),
),
);
servers.clear();
});
describe("isolated setup inference detection", () => {
it("keeps HTTP responsive while a detection worker is synchronously blocked", async () => {
const { detectSetupInferenceIsolated } = await loadDetectionModule();
const server = createServer((_request, response) => {
response.writeHead(200, { "content-type": "application/json" });
response.end('{"ok":true,"status":"live"}');
});
servers.add(server);
await new Promise<void>((resolve) => {
server.listen(0, "127.0.0.1", () => {
resolve();
});
});
const address = server.address() as AddressInfo;
const fallback = emptyDetection();
const pendingStartedAt = performance.now();
const pending = detectSetupInferenceIsolated({
workerUrl: blockingWorkerUrl,
workerData: {
blockMs: 10_000,
detection: emptyDetection(),
partialDetection: fallback,
},
timeoutMs: 100,
});
const startedAt = performance.now();
const response = await requestHealth(`http://127.0.0.1:${address.port}/health`);
const elapsedMs = performance.now() - startedAt;
expect(response.statusCode).toBe(200);
expect(JSON.parse(response.body)).toEqual({ ok: true, status: "live" });
expect(elapsedMs).toBeLessThan(500);
await expect(pending).resolves.toEqual(fallback);
expect(performance.now() - pendingStartedAt).toBeLessThan(1_000);
});
it("coalesces concurrent detections behind one bounded worker", async () => {
const { detectSetupInferenceIsolated } = await loadDetectionModule();
const fallback = vi.fn(async () => emptyDetection());
const options = {
workerUrl: blockingWorkerUrl,
workerData: {
blockMs: 10_000,
detection: emptyDetection(),
partialDetection: emptyDetection(),
},
timeoutMs: 50,
fallback,
};
const [first, second] = await Promise.all([
detectSetupInferenceIsolated(options),
detectSetupInferenceIsolated(options),
]);
expect(first).toEqual(emptyDetection());
expect(second).toEqual(emptyDetection());
expect(fallback).toHaveBeenCalledOnce();
});
});
@@ -0,0 +1,190 @@
import path from "node:path";
import { fileURLToPath, pathToFileURL } from "node:url";
import { Worker, type WorkerOptions } from "node:worker_threads";
import { DEFAULT_AGENT_WORKSPACE_DIR } from "../agents/workspace-default.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
import { listRecommendedToolInstalls } from "../plugins/recommended-tool-installs.js";
import type { SetupInferenceDetection } from "./setup-inference.js";
const SETUP_INFERENCE_DETECTION_TIMEOUT_MS = 10_000;
const log = createSubsystemLogger("system-agent/setup-inference-detection");
type DetectionWorkerMessage =
| { type: "partial"; detection: SetupInferenceDetection }
| { type: "result"; detection: SetupInferenceDetection }
| { ok: false; error: string };
type DetectionWorkerOptions = {
timeoutMs?: number;
workerUrl?: URL;
workerData?: WorkerOptions["workerData"];
fallback?: () => Promise<SetupInferenceDetection>;
};
let inFlightDetection: Promise<SetupInferenceDetection> | undefined;
let workerShutdown: Promise<void> | undefined;
let workerShutdownResult: SetupInferenceDetection | undefined;
function trackWorkerShutdown(worker: Worker): void {
const current = worker.terminate().then(
() => undefined,
(error: unknown) => {
log.warn(`Setup inference detection worker termination failed: ${String(error)}`);
},
);
workerShutdown = current;
void current.finally(() => {
if (workerShutdown === current) {
workerShutdown = undefined;
workerShutdownResult = undefined;
}
});
}
function resolveDetectionWorkerUrl(currentModuleUrl = import.meta.url): URL {
const currentPath = fileURLToPath(currentModuleUrl);
const normalized = currentPath.replaceAll(path.sep, "/");
const distMarker = "/dist/";
const distIndex = normalized.lastIndexOf(distMarker);
if (distIndex >= 0) {
const distRoot = currentPath.slice(0, distIndex + distMarker.length);
return pathToFileURL(
path.join(distRoot, "system-agent", "setup-inference-detection.worker.js"),
);
}
const extension = path.extname(currentPath) || ".js";
return new URL(`./setup-inference-detection.worker${extension}`, currentModuleUrl);
}
function parseDetectionWorkerMessage(value: unknown): DetectionWorkerMessage | undefined {
if (!value || typeof value !== "object" || Array.isArray(value)) {
return undefined;
}
const message = value as Record<string, unknown>;
if (
(message.type === "partial" || message.type === "result") &&
message.detection &&
typeof message.detection === "object"
) {
return message as DetectionWorkerMessage;
}
if (message.ok === false && typeof message.error === "string") {
return message as DetectionWorkerMessage;
}
return undefined;
}
function createUndetectedFallback(): SetupInferenceDetection {
// This fallback must stay independent of the detection/plugin graph. The worker
// supplies richer partial data when that graph loads before the deadline.
return {
candidates: [],
unavailableCandidates: [],
manualProviders: [],
authOptions: [],
recommendedInstalls: listRecommendedToolInstalls(),
workspace: DEFAULT_AGENT_WORKSPACE_DIR,
setupComplete: false,
};
}
async function runDetectionWorker(
options: DetectionWorkerOptions = {},
): Promise<SetupInferenceDetection> {
const workerUrl = options.workerUrl ?? resolveDetectionWorkerUrl();
const execArgv = workerUrl.pathname.endsWith(".ts") ? ["--import", "tsx"] : undefined;
const worker = new Worker(workerUrl, {
execArgv,
...(options.workerData === undefined ? {} : { workerData: options.workerData }),
});
const timeoutMs = options.timeoutMs ?? SETUP_INFERENCE_DETECTION_TIMEOUT_MS;
return await new Promise<SetupInferenceDetection>((resolve, reject) => {
let settled = false;
let partialDetection: SetupInferenceDetection | undefined;
const settle = (finish: () => void) => {
if (settled) {
return;
}
settled = true;
clearTimeout(timer);
worker.removeAllListeners();
// terminate() is asynchronous; keep teardown errors from becoming uncaught events.
worker.on("error", () => undefined);
trackWorkerShutdown(worker);
finish();
};
worker.on("message", (value: unknown) => {
const message = parseDetectionWorkerMessage(value);
if (message && "type" in message && message.type === "partial") {
partialDetection = message.detection;
return;
}
settle(() => {
if (!message) {
reject(new Error("setup inference detection worker returned an invalid result"));
return;
}
if ("ok" in message) {
reject(new Error(message.error));
return;
}
workerShutdownResult = message.detection;
resolve(message.detection);
});
});
worker.once("error", (error) =>
settle(() => reject(error instanceof Error ? error : new Error(String(error)))),
);
worker.once("exit", (code) => {
if (code !== 0) {
settle(() =>
reject(new Error(`setup inference detection worker exited with code ${code}`)),
);
} else {
settle(() => reject(new Error("setup inference detection worker exited without results")));
}
});
const timer = setTimeout(() => {
settle(() => {
log.warn(
`Setup inference detection timed out after ${timeoutMs}ms; returning partial detection.`,
);
if (options.fallback) {
void options.fallback().then(resolve, reject);
return;
}
const detection = partialDetection ?? createUndetectedFallback();
workerShutdownResult = detection;
resolve(detection);
});
}, timeoutMs);
// Installing a message listener references the underlying MessagePort.
// Unref only after all listeners exist so timed-out workers cannot pin shutdown.
worker.unref();
});
}
/** Coalesce read-only detection and isolate native/plugin discovery from Gateway liveness. */
export async function detectSetupInferenceIsolated(
options: DetectionWorkerOptions = {},
): Promise<SetupInferenceDetection> {
if (inFlightDetection) {
return await inFlightDetection;
}
// A native provider probe can delay Worker termination. Reuse the bounded
// result until exit instead of allowing repeat UI requests to stack threads.
if (workerShutdown) {
return workerShutdownResult ?? createUndetectedFallback();
}
const current = runDetectionWorker(options);
inFlightDetection = current;
try {
return await current;
} finally {
if (inFlightDetection === current) {
inFlightDetection = undefined;
}
}
}
@@ -0,0 +1,36 @@
import { parentPort } from "node:worker_threads";
import { listRecommendedToolInstalls } from "../plugins/recommended-tool-installs.js";
import {
detectSetupInference,
listManualSetupInferenceOptions,
type SetupInferenceDetection,
} from "./setup-inference.js";
if (!parentPort) {
throw new Error("setup inference detection worker requires a parent port");
}
const port = parentPort;
try {
const manual = await listManualSetupInferenceOptions();
const partial: SetupInferenceDetection = {
candidates: [],
unavailableCandidates: [],
recommendedInstalls: listRecommendedToolInstalls(),
...manual,
};
port.postMessage({
type: "partial",
detection: partial,
});
const detection = await detectSetupInference();
port.postMessage({ type: "result", detection });
} catch (error) {
port.postMessage({
ok: false,
error: error instanceof Error ? error.message : String(error),
});
} finally {
port.close();
}
+3 -3
View File
@@ -441,7 +441,7 @@ export async function detectSetupInference(
probe("pi"),
probe("opencode"),
]);
if (antigravity.found) {
if (antigravity.found && !antigravity.timedOut) {
unavailableCandidates.push({
id: "antigravity-cli",
label: "Antigravity CLI",
@@ -450,7 +450,7 @@ export async function detectSetupInference(
"Can't be auto-tested safely here. Sign in with a provider or use an API key instead.",
});
}
if (pi.found) {
if (pi.found && !pi.timedOut) {
unavailableCandidates.push({
id: "pi-cli",
label: "Pi CLI",
@@ -459,7 +459,7 @@ export async function detectSetupInference(
"Pi CLI is installed, but its whole-agent sessions require separate setup and are not a reusable guided-setup inference route.",
});
}
if (opencode.found) {
if (opencode.found && !opencode.timedOut) {
unavailableCandidates.push({
id: "opencode-cli",
label: "OpenCode CLI",
+2
View File
@@ -712,6 +712,7 @@ describe("collectMissingPackPaths", () => {
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/state/openclaw-database-verify.worker.js",
"dist/system-agent/setup-inference-detection.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
bundledDistPluginFile("telegram", "runtime-api.js"),
@@ -750,6 +751,7 @@ describe("collectMissingPackPaths", () => {
"dist/audit/audit-event-writer.worker.js",
"dist/config/sessions/session-transcript-reconcile.worker.js",
"dist/state/openclaw-database-verify.worker.js",
"dist/system-agent/setup-inference-detection.worker.js",
"dist/task-registry-control.runtime.js",
"dist/telegram-ingress-worker.runtime.js",
"dist/build-info.json",
+2
View File
@@ -278,6 +278,8 @@ function buildCoreDistEntries(): Record<string, string> {
"config/sessions/session-transcript-reconcile.worker":
"src/config/sessions/session-transcript-reconcile.worker.ts",
"state/openclaw-database-verify.worker": "src/state/openclaw-database-verify.worker.ts",
"system-agent/setup-inference-detection.worker":
"src/system-agent/setup-inference-detection.worker.ts",
"acp/control-plane/manager": "src/acp/control-plane/manager.ts",
"cli/gateway-lifecycle.runtime": "src/cli/gateway-cli/lifecycle.runtime.ts",
"provider-dispatcher.runtime": "src/auto-reply/reply/provider-dispatcher.runtime.ts",