diff --git a/ci/source-architecture-budget.json b/ci/source-architecture-budget.json index 25fd5118b8a..f29f248e4fa 100644 --- a/ci/source-architecture-budget.json +++ b/ci/source-architecture-budget.json @@ -18,7 +18,7 @@ "src/lib/core/ports.ts": 89, "src/lib/core/shell-quote.ts": 28, "src/lib/core/url-utils.ts": 29, - "src/lib/core/wait.ts": 35, + "src/lib/core/wait.ts": 36, "src/lib/credentials/store.ts": 46, "src/lib/inference/config.ts": 30, "src/lib/inference/web-search.ts": 21, @@ -37,15 +37,17 @@ "defaultMax": 20, "maxByFile": { "src/lib/actions/inference-set.ts": 32, - "src/lib/actions/sandbox/connect.ts": 40, + "src/lib/actions/sandbox/connect.ts": 41, "src/lib/actions/sandbox/destroy.ts": 29, "src/lib/actions/sandbox/doctor.ts": 30, - "src/lib/actions/sandbox/status-snapshot.ts": 20, + "src/lib/actions/sandbox/status-snapshot.ts": 21, "src/lib/actions/sandbox/policy-channel.ts": 30, "src/lib/actions/sandbox/process-recovery.ts": 21, "src/lib/actions/sandbox/rebuild-pipeline.ts": 28, "src/lib/actions/sandbox/snapshot.ts": 40, "src/lib/actions/uninstall/run-plan.ts": 26, + + "src/lib/inference/local.ts": 21, "src/lib/inference/onboard-probes.ts": 21, "src/lib/inference/vllm.ts": 21, "src/lib/onboard.ts": 202, diff --git a/src/lib/actions/dns/index.test.ts b/src/lib/actions/dns/index.test.ts index 3a290de9e11..27a2be5b0dc 100644 --- a/src/lib/actions/dns/index.test.ts +++ b/src/lib/actions/dns/index.test.ts @@ -162,6 +162,10 @@ describe("runSetupDnsProxy", () => { it("configures the DNS proxy through kubectl-in-docker argv calls", () => { const calls: string[][] = []; const log = vi.fn(); + let dnsReadyCalls = 0; + let pidReads = 0; + let getentCalls = 0; + const sleep = vi.fn(); const runDocker = vi.fn((args: string[]) => { calls.push(args); const cmd = args.join(" "); @@ -170,14 +174,23 @@ describe("runSetupDnsProxy", () => { if (cmd.includes("get endpoints kube-dns")) return ok("10.42.0.15"); if (cmd.includes("get pods -n openshell -o name")) return ok("pod/box[1]-abc\n"); if (cmd.includes("ip addr show")) return ok("10.200.0.1\n"); - if (cmd.includes("cat /tmp/dns-proxy.pid")) return ok("12345\n"); + if (cmd.includes("cat /tmp/dns-proxy.pid")) { + pidReads += 1; + return pidReads === 1 ? ok("") : ok("12345\n"); + } if (cmd.includes("cat /tmp/dns-proxy.log")) return ok("dns-proxy: 10.200.0.1:53 -> 10.43.0.10:53 pid=12345\n"); - if (cmd.includes("python3 -c")) return ok("ok"); + if (cmd.includes("python3 -c")) { + dnsReadyCalls += 1; + return dnsReadyCalls === 3 ? ok("ok") : ok(""); + } if (cmd.includes("ls /run/netns/")) return ok("sandbox-ns\n"); if (cmd.includes("test -x")) return ok(); if (cmd.includes("cat /etc/resolv.conf")) return ok("nameserver 10.200.0.1\n"); - if (cmd.includes("getent hosts github.com")) return ok("140.82.112.4 github.com\n"); + if (cmd.includes("getent hosts github.com")) { + getentCalls += 1; + return getentCalls === 3 ? ok("140.82.112.4 github.com\n") : ok(""); + } return ok(); }); @@ -187,7 +200,7 @@ describe("runSetupDnsProxy", () => { env: { DOCKER_HOST: "unix:///tmp/fake-docker.sock" }, log, runDocker, - sleep: vi.fn(), + sleep, }, ); @@ -204,6 +217,10 @@ describe("runSetupDnsProxy", () => { ), ).toBe(true); expect(log).toHaveBeenCalledWith(" DNS verification: 4 passed, 0 failed"); + expect(getentCalls).toBe(3); + expect(sleep.mock.calls.filter(([milliseconds]) => milliseconds === 2_000)).toHaveLength(2); + expect(dnsReadyCalls).toBe(3); + expect(sleep.mock.calls).toEqual([[1_000], [1_000], [2_000], [2_000]]); }); it("falls back to the CoreDNS pod endpoint when the kube-dns service IP is unavailable", () => { diff --git a/src/lib/actions/dns/index.ts b/src/lib/actions/dns/index.ts index 931c085ac35..e592b80acfb 100644 --- a/src/lib/actions/dns/index.ts +++ b/src/lib/actions/dns/index.ts @@ -7,6 +7,7 @@ import os from "node:os"; import path from "node:path"; import { dockerSpawnSync } from "../../adapters/docker/exec"; +import { retryUntil } from "../../core/retry"; import { buildCoreDnsPatchJson, dockerHostRuntime, @@ -462,30 +463,33 @@ export function runSetupDnsProxy( dockerEnv, ); - let dnsReady = false; - for (let attempt = 0; attempt < 10; attempt += 1) { - const probe = kctl( - runDocker, - cluster, - [ - "exec", - "-n", - "openshell", - pod, - "--", - "python3", - "-c", - buildDnsReadyProbePython(vethGateway), - ], - dockerEnv, + const dnsReady = retryUntil( + () => + kctl( + runDocker, + cluster, + [ + "exec", + "-n", + "openshell", + pod, + "--", + "python3", + "-c", + buildDnsReadyProbePython(vethGateway), + ], + dockerEnv, + ).stdout.includes("ok"), + { + accept: Boolean, + retryDelaysMs: Array.from({ length: 9 }, () => 1_000), + sleep, + }, + ); + if (!dnsReady) + log( + "WARNING: DNS forwarder did not respond after 10 attempts. The following DNS checks can report failures.", ); - if (probe.stdout.includes("ok")) { - dnsReady = true; - break; - } - sleep(1000); - } - if (!dnsReady) log("WARNING: DNS forwarder not responding after 10s — verification may fail"); const sandboxNamespace = selectSandboxNamespace( commandOutput( @@ -686,14 +690,17 @@ export function runSetupDnsProxy( verificationFail += 1; } - let dnsResult = ""; - for (let attempt = 1; attempt <= 3; attempt += 1) { - dnsResult = commandOutput( - sbExec(["getent", "hosts", "github.com"]) ?? { status: 1, stdout: "", stderr: "" }, - ).trim(); - if (dnsResult) break; - if (attempt < 3) sleep(2000); - } + const dnsResult = retryUntil( + () => + commandOutput( + sbExec(["getent", "hosts", "github.com"]) ?? { status: 1, stdout: "", stderr: "" }, + ).trim(), + { + accept: Boolean, + retryDelaysMs: [2_000, 2_000], + sleep, + }, + ); if (dnsResult) { log(` [PASS] getent hosts github.com -> ${dnsResult}`); verificationPass += 1; diff --git a/src/lib/actions/sandbox/connect-route-repair.test.ts b/src/lib/actions/sandbox/connect-route-repair.test.ts index 4f5224ce3df..609e0f6a07f 100644 --- a/src/lib/actions/sandbox/connect-route-repair.test.ts +++ b/src/lib/actions/sandbox/connect-route-repair.test.ts @@ -4,6 +4,8 @@ import { describe, expect, it, vi } from "vitest"; import type { SandboxEntry } from "../../state/registry"; +import { captureOpenshell } from "../../adapters/openshell/runtime"; + vi.mock("../../adapters/openshell/runtime", () => ({ captureOpenshell: vi.fn(() => ({ status: 0, output: "" })), getOpenshellBinary: vi.fn(() => "openshell"), @@ -37,6 +39,8 @@ vi.mock("./gateway-state", () => ({ import { type ManagedInferenceRouteResetDeps, + + probeSandboxInferenceRoute, repairSandboxInferenceRouteWithDeps, resetManagedInferenceRouteWithDeps, type SandboxInferenceRouteProbe, @@ -391,3 +395,35 @@ describe("managed inference route reset unit flow", () => { expect(calls.unrecoverable).toEqual([{ sandboxName: "demo", detail: "BROKEN 503 still down" }]); }); }); + + +describe("connect inference route retries", () => { + it("returns the third healthy probe result after two unhealthy probe results (#9218)", () => { + vi.mocked(captureOpenshell) + .mockReturnValueOnce({ status: 0, output: "BROKEN 503", stderr: "" }) + .mockReturnValueOnce({ status: 0, output: "BROKEN 503", stderr: "" }) + .mockReturnValueOnce({ status: 0, output: "OK 200", stderr: "" }); + + const result = probeSandboxInferenceRoute( + "alpha", + { name: "hermes" }, + { attempts: 3, delayMs: 2_000 }, + ); + + expect(result).toMatchObject({ healthy: true, broken: false, httpStatus: 200 }); + expect(captureOpenshell).toHaveBeenCalledTimes(3); + }); + + it("returns the final unhealthy probe result after exhausting attempts (#9218)", () => { + vi.mocked(captureOpenshell).mockReturnValue({ status: 0, output: "BROKEN 503", stderr: "" }); + + const result = probeSandboxInferenceRoute( + "alpha", + { name: "hermes" }, + { attempts: 2, delayMs: 500 }, + ); + + expect(result).toMatchObject({ healthy: false, broken: true, httpStatus: 503 }); + expect(captureOpenshell).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/lib/actions/sandbox/connect.ts b/src/lib/actions/sandbox/connect.ts index eb3078cef32..8692356a3a9 100644 --- a/src/lib/actions/sandbox/connect.ts +++ b/src/lib/actions/sandbox/connect.ts @@ -3,12 +3,9 @@ import { spawnSync } from "node:child_process"; import { resolveOpenshell } from "../../adapters/openshell/resolve"; +import { captureOpenshell, getOpenshellBinary, runOpenshell } from "../../adapters/openshell/runtime"; import { - captureOpenshell, - getOpenshellBinary, - runOpenshell, -} from "../../adapters/openshell/runtime"; -import { + OPENSHELL_INFERENCE_ROUTE_PROBE_TIMEOUT_MS, OPENSHELL_OPERATION_TIMEOUT_MS, OPENSHELL_PROBE_TIMEOUT_MS, @@ -17,6 +14,8 @@ import type { AgentDefinition } from "../../agent/defs"; import * as agentRuntime from "../../agent/runtime"; import { CLI_NAME } from "../../cli/branding"; import { D, G, R, YW } from "../../cli/terminal-style"; +import { retryUntil } from "../../core/retry"; + import { spawnExitCode } from "../../core/process-exit"; import { shellQuote } from "../../core/shell-quote"; import { getNamedGatewayLifecycleState } from "../../gateway-runtime-action"; @@ -128,6 +127,7 @@ type SandboxListProbe = { output: string; }; + export type SandboxInferenceRouteProbe = { healthy: boolean; broken: boolean; @@ -135,16 +135,16 @@ export type SandboxInferenceRouteProbe = { detail: string; }; -type SandboxInferenceRouteEnsureResult = { - sandbox: SandboxEntry | null; - routeHealthy: boolean | null; -}; - type InferenceRouteProbeOptions = { attempts?: number; delayMs?: number; }; +type SandboxInferenceRouteEnsureResult = { + sandbox: SandboxEntry | null; + routeHealthy: boolean | null; +}; + export type SandboxInferenceRouteRepairResult = { healthy: boolean; repairAttempted: boolean; @@ -352,15 +352,6 @@ async function runSandboxConnectProbe(sandboxName: string): Promise { process.exit(1); } -function sleepSync(ms: number): void { - if (ms <= 0) return; - if (process.env.VITEST === "true" || process.env.NEMOCLAW_TEST_NO_SLEEP === "1") return; - spawnSync(process.execPath, ["-e", `setTimeout(() => {}, ${ms})`], { - stdio: "ignore", - timeout: ms + 1_000, - }); -} - const GATEWAY_UNAVAILABLE_RE = /No gateway configured|No active gateway|Connection refused|client error \(Connect\)|tcp connect error|Status:\s*Disconnected/i; @@ -412,43 +403,49 @@ function failIfGatewayBlocksConnectReadiness(sandboxName: string): void { } } -function probeSandboxInferenceRoute( + +function sleepSync(milliseconds: number): void { + if (milliseconds <= 0) return; + if (process.env.VITEST === "true" || process.env.NEMOCLAW_TEST_NO_SLEEP === "1") return; + spawnSync(process.execPath, ["-e", `setTimeout(() => {}, ${milliseconds})`], { + stdio: "ignore", + timeout: milliseconds + 1_000, + }); +} + +export function probeSandboxInferenceRoute( sandboxName: string, agent: InferenceRouteProbeAgent, { attempts = 1, delayMs = 0 }: InferenceRouteProbeOptions = {}, ): SandboxInferenceRouteProbe { - let lastProbe: SandboxInferenceRouteProbe | null = null; - const boundedAttempts = Math.max(1, attempts); - - for (let attempt = 1; attempt <= boundedAttempts; attempt += 1) { - // Keep the shell string inside the sandbox: curl write-out, body capture, - // and status classification must run as one bounded probe. sandboxName - // remains an argv value, so no user input is interpolated into the script. - const probe = captureOpenshell(buildSandboxInferenceRouteProbeArgs(sandboxName, agent), { - ignoreError: true, - includeStreams: true, - timeout: OPENSHELL_INFERENCE_ROUTE_PROBE_TIMEOUT_MS, - }); - const parsed = parseSandboxInferenceRouteProbeResult(probe); - lastProbe = { - healthy: parsed.healthy, - broken: parsed.broken, - httpStatus: parsed.httpStatus, - detail: parsed.detail, - }; - if (lastProbe.healthy || attempt === boundedAttempts) return lastProbe; - sleepSync(delayMs); - } - - return ( - lastProbe ?? { - healthy: false, - broken: false, - detail: "inference route probe did not run", - } + const attemptCount = Math.max(1, Math.floor(attempts)); + return retryUntil( + () => { + // Keep the shell string inside the sandbox: curl write-out, body capture, + // and status classification must run as one bounded probe. sandboxName + // remains an argv value, so no user input is interpolated into the script. + const probe = captureOpenshell(buildSandboxInferenceRouteProbeArgs(sandboxName, agent), { + ignoreError: true, + includeStreams: true, + timeout: OPENSHELL_INFERENCE_ROUTE_PROBE_TIMEOUT_MS, + }); + const parsed = parseSandboxInferenceRouteProbeResult(probe); + return { + healthy: parsed.healthy, + broken: parsed.broken, + httpStatus: parsed.httpStatus, + detail: parsed.detail, + }; + }, + { + accept: (result) => result.healthy, + retryDelaysMs: Array.from({ length: attemptCount - 1 }, () => delayMs), + sleep: sleepSync, + }, ); } + function shouldUseLegacyDnsProxyRepair(sb: SandboxEntry | null): boolean { // The legacy repair patches CoreDNS inside an `openshell-cluster-` // container, which only the k3s/kubernetes gateway runs. The docker driver diff --git a/src/lib/actions/sandbox/inference-route-health.ts b/src/lib/actions/sandbox/inference-route-health.ts index f99d5275577..e0e1de332f6 100644 --- a/src/lib/actions/sandbox/inference-route-health.ts +++ b/src/lib/actions/sandbox/inference-route-health.ts @@ -21,6 +21,7 @@ import { export type { SandboxInferenceInvocationResult } from "./inference-invocation-probe"; export type ProbeSandboxInferenceInvocation = typeof probeSandboxInferenceInvocation; + export type SandboxInferenceRouteHealth = { ok: boolean; endpoint: string; diff --git a/src/lib/actions/sandbox/status-snapshot.ts b/src/lib/actions/sandbox/status-snapshot.ts index 301cf7a5103..7a213c84a50 100644 --- a/src/lib/actions/sandbox/status-snapshot.ts +++ b/src/lib/actions/sandbox/status-snapshot.ts @@ -2,12 +2,15 @@ // SPDX-License-Identifier: Apache-2.0 import { setTimeout as sleep } from "node:timers/promises"; + import { detectOpenShellStateRpcResultIssue, type OpenShellStateRpcIssue, } from "../../adapters/openshell/gateway-drift"; import { captureOpenshellForStatus, isCommandTimeout } from "../../adapters/openshell/runtime"; import { type AgentDefinition, getAgentRuntimeKind, loadAgent } from "../../agent/defs"; +import { retryUntilAsync } from "../../core/retry"; + import { withStdoutRedirectedToStderr } from "../../cli/stdout-guard"; import type { CuaAppliedPolicyIdentity } from "../../cua/contract"; import { @@ -582,11 +585,14 @@ export async function collectSandboxStatusSnapshot( const probe = opts.deps?.probeSandboxInferenceGatewayHealthImpl ?? probeSandboxInferenceGatewayHealth; const attempts = recoveredManagedGateway ? RECOVERED_INFERENCE_PROBE_ATTEMPTS : 1; - for (let attempt = 1; attempt <= attempts; attempt += 1) { - gatewayChain = await probe(sandboxName); - if (gatewayChain?.ok || attempt === attempts) break; - await (opts.deps?.delayInferenceRecoveryProbe ?? sleep)(RECOVERED_INFERENCE_PROBE_DELAY_MS); - } + gatewayChain = await retryUntilAsync(() => probe(sandboxName), { + accept: (result) => Boolean(result?.ok), + retryDelaysMs: Array.from( + { length: attempts - 1 }, + () => RECOVERED_INFERENCE_PROBE_DELAY_MS, + ), + sleep: opts.deps?.delayInferenceRecoveryProbe ?? sleep, + }); } catch (error) { // This is a permanent fail-closed runtime boundary, but unexpected // OpenShell/transport exceptions must remain observable for diagnosis. diff --git a/src/lib/actions/sandbox/stopped-sandbox-backup.ts b/src/lib/actions/sandbox/stopped-sandbox-backup.ts index 291dc597daf..67be9fb8931 100644 --- a/src/lib/actions/sandbox/stopped-sandbox-backup.ts +++ b/src/lib/actions/sandbox/stopped-sandbox-backup.ts @@ -3,6 +3,7 @@ import { dockerContainerInspectFormat } from "../../adapters/docker/inspect"; import { dockerCapture, dockerRun } from "../../adapters/docker/run"; +import { retryUntilAsync } from "../../core/retry"; import { resolveSandboxContainerOwner } from "../../domain/sandbox/container-owner"; import { findLabeledSandboxContainers, @@ -208,14 +209,12 @@ export async function backupStartedSandboxState( depsOverride: Partial = {}, ): Promise { const deps: BackupRetryDeps = { ...defaultBackupRetryDeps, ...depsOverride }; - let result = deps.backup(sandboxName); - for ( - let attempt = 1; - attempt < deps.attempts && !result.success && result.unreachable; - attempt++ - ) { - await deps.sleep(deps.delayMs); - result = deps.backup(sandboxName); - } - return result; + return retryUntilAsync(() => deps.backup(sandboxName), { + accept: (result) => result.success || !result.unreachable, + retryDelaysMs: Array.from( + { length: Math.max(0, Math.ceil(deps.attempts) - 1) }, + () => deps.delayMs, + ), + sleep: deps.sleep, + }); } diff --git a/src/lib/adapters/docker/runtime.test.ts b/src/lib/adapters/docker/runtime.test.ts index 6356ea2124e..b6800741b00 100644 --- a/src/lib/adapters/docker/runtime.test.ts +++ b/src/lib/adapters/docker/runtime.test.ts @@ -11,6 +11,7 @@ vi.mock("../../runner", () => ({ import { DOCKER_INFO_RUNTIME_PROBE_ATTEMPTS, + DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS, DOCKER_INFO_RUNTIME_PROBE_TIMEOUT_MS, detectContainerRuntimeFromDockerInfo, } from "./runtime"; @@ -20,11 +21,13 @@ describe("docker runtime detection", () => { const calls: unknown[] = []; const outputs = ["", "", "Operating System: Docker Desktop"]; + const sleep = vi.fn(); const runtime = detectContainerRuntimeFromDockerInfo({ dockerInfoImpl: (opts) => { calls.push(opts); return outputs.shift() ?? ""; }, + sleep, }); expect(runtime).toBe("docker-desktop"); @@ -35,11 +38,17 @@ describe("docker runtime detection", () => { timeout: DOCKER_INFO_RUNTIME_PROBE_TIMEOUT_MS, })), ); + + expect(sleep.mock.calls).toEqual([ + [DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS], + [DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS], + ]); }); it("returns unknown after all attempts are indeterminate", () => { const calls: unknown[] = []; + const sleep = vi.fn(); const runtime = detectContainerRuntimeFromDockerInfo({ attempts: 2, dockerInfoImpl: (opts) => { @@ -47,6 +56,7 @@ describe("docker runtime detection", () => { return ""; }, timeoutMs: 1234, + sleep, }); expect(runtime).toBe("unknown"); @@ -54,5 +64,8 @@ describe("docker runtime detection", () => { { ignoreError: true, timeout: 1234 }, { ignoreError: true, timeout: 1234 }, ]); + + expect(sleep).toHaveBeenCalledOnce(); + expect(sleep).toHaveBeenCalledWith(DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS); }); }); diff --git a/src/lib/adapters/docker/runtime.ts b/src/lib/adapters/docker/runtime.ts index eca9cb29e29..12a8b137788 100644 --- a/src/lib/adapters/docker/runtime.ts +++ b/src/lib/adapters/docker/runtime.ts @@ -1,12 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +import { retryUntil } from "../../core/retry"; +import { sleepMs } from "../../core/wait"; import type { ContainerRuntime } from "../../platform"; import { inferContainerRuntime } from "../../platform"; import { dockerInfo } from "./info"; import type { DockerCaptureOptions } from "./run"; export const DOCKER_INFO_RUNTIME_PROBE_ATTEMPTS = 3; +export const DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS = 250; export const DOCKER_INFO_RUNTIME_PROBE_TIMEOUT_MS = 5000; type DockerInfoProbe = (opts: DockerCaptureOptions) => string; @@ -15,6 +18,7 @@ export interface DetectContainerRuntimeOptions { attempts?: number; dockerInfoImpl?: DockerInfoProbe; timeoutMs?: number; + sleep?: (milliseconds: number) => void; } export function detectContainerRuntimeFromDockerInfo( @@ -24,10 +28,15 @@ export function detectContainerRuntimeFromDockerInfo( const timeout = Math.max(1, Math.floor(opts.timeoutMs ?? DOCKER_INFO_RUNTIME_PROBE_TIMEOUT_MS)); const probe = opts.dockerInfoImpl ?? dockerInfo; - for (let attempt = 0; attempt < attempts; attempt++) { - const runtime = inferContainerRuntime(probe({ ignoreError: true, timeout })); - if (runtime !== "unknown") return runtime; - } - - return "unknown"; + return retryUntil( + () => inferContainerRuntime(probe({ ignoreError: true, timeout })), + { + accept: (runtime) => runtime !== "unknown", + retryDelaysMs: Array.from( + { length: attempts - 1 }, + () => DOCKER_INFO_RUNTIME_PROBE_RETRY_DELAY_MS, + ), + sleep: opts.sleep ?? sleepMs, + }, + ); } diff --git a/src/lib/core/retry.ts b/src/lib/core/retry.ts new file mode 100644 index 00000000000..4356ec2e40e --- /dev/null +++ b/src/lib/core/retry.ts @@ -0,0 +1,66 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +interface RetryUntilBaseOptions { + /** Return true to stop retrying after this result. */ + accept: (result: T, attempt: number) => boolean; + /** Delay in milliseconds before each additional attempt. */ + retryDelaysMs: readonly number[]; +} + +export type RetryUntilOptions = RetryUntilBaseOptions & { + /** Run after `accept` returns false and before the scheduled sleep. */ + onRetry?: (result: T, delayMs: number, attempt: number) => void; + /** Sleep for the specified number of milliseconds between attempts. */ + sleep: (ms: number) => void; +}; + +export type RetryUntilAsyncOptions = RetryUntilBaseOptions & { + /** Run after `accept` returns false and before the scheduled sleep. */ + onRetry?: (result: T, delayMs: number, attempt: number) => void | Promise; + /** Sleep for the specified number of milliseconds between attempts. */ + sleep: (ms: number) => Promise; +}; + +/** + * Retry a synchronous operation until its result is accepted or the delay + * schedule is exhausted. Returns the final operation result. + */ +export function retryUntil(operation: (attempt: number) => T, options: RetryUntilOptions): T { + let attempt = 1; + let result = operation(attempt); + if (options.accept(result, attempt)) return result; + + for (const delayMs of options.retryDelaysMs) { + options.onRetry?.(result, delayMs, attempt); + + options.sleep(delayMs); + attempt += 1; + result = operation(attempt); + if (options.accept(result, attempt)) return result; + } + return result; +} + +/** + * Retry an asynchronous operation until its result is accepted or the delay + * schedule is exhausted. Returns the final operation result. + */ +export async function retryUntilAsync( + operation: (attempt: number) => T | Promise, + options: RetryUntilAsyncOptions, +): Promise { + let attempt = 1; + let result = await operation(attempt); + if (options.accept(result, attempt)) return result; + + for (const delayMs of options.retryDelaysMs) { + await options.onRetry?.(result, delayMs, attempt); + + await options.sleep(delayMs); + attempt += 1; + result = await operation(attempt); + if (options.accept(result, attempt)) return result; + } + return result; +} diff --git a/src/lib/inference/https-pin-runtime-adapter.test.ts b/src/lib/inference/https-pin-runtime-adapter.test.ts index 82c23f29781..c6f6d77e090 100644 --- a/src/lib/inference/https-pin-runtime-adapter.test.ts +++ b/src/lib/inference/https-pin-runtime-adapter.test.ts @@ -1221,6 +1221,23 @@ describe("adapter recovery lock (#6141)", () => { expect(sleep).toHaveBeenCalledWith(25); }); + it("does not probe or sleep when the adapter exit attempt budget is zero (#9218)", async () => { + const isRunning = vi.fn(() => true); + const sleep = vi.fn(async () => {}); + + await expect( + lockModule.__test.waitForAdapterProcessExit(12345, { + isRunning, + sleep, + attempts: 0, + intervalMs: 25, + }), + ).resolves.toBe(false); + expect(isRunning).not.toHaveBeenCalled(); + expect(sleep).not.toHaveBeenCalled(); + }); + + it("refuses replacement when the old adapter never exits within the bounded wait", async () => { const sleep = vi.fn(async () => {}); diff --git a/src/lib/inference/https-pin-runtime-adapter.ts b/src/lib/inference/https-pin-runtime-adapter.ts index 6c12d96c126..9a8f3721190 100644 --- a/src/lib/inference/https-pin-runtime-adapter.ts +++ b/src/lib/inference/https-pin-runtime-adapter.ts @@ -55,6 +55,7 @@ import { VLLM_PORT, validateHttpsPinRuntimeAdapterPort, } from "../core/ports"; +import { retryUntilAsync } from "../core/retry"; import { getVersion } from "../core/version"; import { ROOT, run, runCapture } from "../runner"; import { buildMinimalCredentialAdapterEnv } from "../subprocess-env"; @@ -843,13 +844,15 @@ async function waitForAdapterProcessExit( ): Promise { const isRunning = options.isRunning || ((candidatePid: number) => isAdapterProcess(candidatePid)); const sleep = options.sleep || sleepMs; - const attempts = options.attempts || PROCESS_EXIT_WAIT_ATTEMPTS; + const attempts = options.attempts ?? PROCESS_EXIT_WAIT_ATTEMPTS; const intervalMs = options.intervalMs || PROCESS_EXIT_WAIT_MS; - for (let attempt = 0; attempt < attempts; attempt++) { - if (!isRunning(pid)) return true; - if (attempt + 1 < attempts) await sleep(intervalMs); - } - return false; + if (attempts <= 0) return false; + + return retryUntilAsync(() => !isRunning(pid), { + accept: (exited) => exited, + retryDelaysMs: Array.from({ length: Math.ceil(attempts) - 1 }, () => intervalMs), + sleep, + }); } async function killStaleAdapter(): Promise { @@ -1380,11 +1383,11 @@ async function revokeRouteLocked( const allowedSourceCidrs = deps.readAllowedSourceCidrs(); const authenticatedLiveAdapter = Boolean( controlToken && - allowedSourceCidrs && - (await deps.probeHealth({ - controlToken: controlToken as string, - expectedSourceCidrs: allowedSourceCidrs, - })), + allowedSourceCidrs && + (await deps.probeHealth({ + controlToken: controlToken as string, + expectedSourceCidrs: allowedSourceCidrs, + })), ); if (authenticatedLiveAdapter && controlToken) { await deps.deleteRoute(controlToken, routeId); diff --git a/src/lib/inference/local.ts b/src/lib/inference/local.ts index 4ef4047f608..b680519e1d7 100644 --- a/src/lib/inference/local.ts +++ b/src/lib/inference/local.ts @@ -16,10 +16,13 @@ import { buildValidatedCurlCommandArgs } from "../adapters/http/curl-args"; import type { CurlProbeOptions, CurlProbeResult } from "../adapters/http/probe"; import { runCurlProbe } from "../adapters/http/probe"; import { OLLAMA_PORT, OLLAMA_PROXY_PORT, VLLM_PORT } from "../core/ports"; + +import { retryUntil } from "../core/retry"; import { sleepSeconds } from "../core/wait"; import { containerCanReachHostLoopback, isWsl, type WslDetectionOptions } from "../platform"; import { type CaptureResult, runCapture, runCaptureEx, shellQuote } from "../runner"; import { buildSubprocessEnv } from "../subprocess-env"; + import { resolveSharedLocalAdapterStateRoot } from "./local-adapter-lifecycle"; import { detectNvidiaPlatform } from "./nim"; import { @@ -1083,9 +1086,9 @@ export function getLocalProviderContainerReachabilityCheck(provider: string): st } } + const CONTAINER_CHECK_MAX_ATTEMPTS = 3; const CONTAINER_CHECK_RETRY_DELAY_SECS = 2; - export function validateLocalProvider( provider: string, runCaptureImpl?: RunCaptureFn, @@ -1099,7 +1102,6 @@ export function validateLocalProvider( } const capture = runCaptureImpl ?? runCapture; - const sleep = sleepFn ?? sleepSeconds; const command = getLocalProviderHealthCheck(provider); if (!command) { if (provider === "vllm-local") { @@ -1142,15 +1144,21 @@ export function validateLocalProvider( return { ok: true }; } - // Retry container reachability check with backoff - for (let attempt = 1; attempt <= CONTAINER_CHECK_MAX_ATTEMPTS; attempt++) { - const containerOutput = capture(containerCommand, { ignoreError: true }); - if (isLocalProviderProbeOutputHealthy(containerCommand.at(-1) ?? "", containerOutput)) { - return { ok: true }; - } - if (attempt < CONTAINER_CHECK_MAX_ATTEMPTS) { - sleep(CONTAINER_CHECK_RETRY_DELAY_SECS); - } + const sleep = sleepFn ?? sleepSeconds; + const containerOutput = retryUntil( + () => capture(containerCommand, { ignoreError: true }), + { + accept: (output) => + isLocalProviderProbeOutputHealthy(containerCommand.at(-1) ?? "", output), + retryDelaysMs: Array.from( + { length: CONTAINER_CHECK_MAX_ATTEMPTS - 1 }, + () => CONTAINER_CHECK_RETRY_DELAY_SECS * 1_000, + ), + sleep: (milliseconds) => sleep(milliseconds / 1_000), + }, + ); + if (isLocalProviderProbeOutputHealthy(containerCommand.at(-1) ?? "", containerOutput)) { + return { ok: true }; } // All retries exhausted — collect diagnostics diff --git a/src/lib/inference/openai-validation-session.ts b/src/lib/inference/openai-validation-session.ts index 2ba2c35fd8a..c4c11be98e0 100644 --- a/src/lib/inference/openai-validation-session.ts +++ b/src/lib/inference/openai-validation-session.ts @@ -7,6 +7,7 @@ import { createValidationSession, type ValidationSessionOptions, } from "../adapters/http/validation-session"; +import { retryUntilAsync } from "../core/retry"; import { addTraceEvent, withTraceSpan } from "../trace"; import type { TrustedPrivateEndpointCapability } from "./endpoint-ssrf-preflight"; import { @@ -192,36 +193,31 @@ async function requestWithHttpRetry( request: () => Promise, retryTransientHttp = true, ): Promise { - let result = await request(); - let attempt = 1; - addTraceEvent("probe_result", { - attempt, - ok: result.ok, - http_status: result.httpStatus, - curl_status: result.curlStatus, - }); - for (const delayMs of RETRY_DELAYS_MS) { - if ( - !retryTransientHttp || - result.curlStatus !== 0 || - !RETRIABLE_HTTP_STATUSES.has(result.httpStatus) - ) { - break; - } - console.log( - ` ${name} validation returned HTTP ${result.httpStatus}; retrying in ${Math.round(delayMs / 1000)}s...`, - ); - await waitForRetry(delayMs); - attempt += 1; - result = await request(); - addTraceEvent("probe_result", { - attempt, - ok: result.ok, - http_status: result.httpStatus, - curl_status: result.curlStatus, - }); - } - return result; + return retryUntilAsync( + async (attempt) => { + const result = await request(); + addTraceEvent("probe_result", { + attempt, + ok: result.ok, + http_status: result.httpStatus, + curl_status: result.curlStatus, + }); + return result; + }, + { + accept: (result) => + !retryTransientHttp || + result.curlStatus !== 0 || + !RETRIABLE_HTTP_STATUSES.has(result.httpStatus), + retryDelaysMs: RETRY_DELAYS_MS, + onRetry: (result, delayMs) => { + console.log( + ` ${name} validation returned HTTP ${result.httpStatus}; retrying in ${Math.round(delayMs / 1000)}s...`, + ); + }, + sleep: waitForRetry, + }, + ); } function shouldUseLegacyForModel(model: string): boolean { diff --git a/src/lib/inference/probe-retry.ts b/src/lib/inference/probe-retry.ts index db9b55fe232..cc1e153a301 100644 --- a/src/lib/inference/probe-retry.ts +++ b/src/lib/inference/probe-retry.ts @@ -13,6 +13,7 @@ // onboard-probes.test.ts. const trace = require("../trace"); +const { retryUntil } = require("../core/retry"); const CURL_TIMEOUT_STATUS = 28; const NODE_SPAWN_TIMEOUT_STATUS = -110; @@ -73,38 +74,38 @@ function executeProbeWithHttpRetry(probe) { "nemoclaw.inference.validation_probe", { probe_name: probe.name, api: probe.api || null }, () => { - let attempt = 1; - let result = probe.execute(); - trace.addTraceEvent("probe_result", { - attempt, - ok: result.ok, - http_status: result.httpStatus, - curl_status: result.curlStatus, - }); - for (const delayMs of HTTP_PROBE_RETRY_DELAYS_MS) { - const customReason = probe.retryReason?.(result); - const httpRetry = shouldRetryHttpProbe(result); - if (!httpRetry && !customReason) break; - const reason = customReason || `returned HTTP ${result.httpStatus}`; - console.log( - ` ${probe.name} validation ${reason}; retrying in ${Math.round(delayMs / 1000)}s...`, - ); - trace.addTraceEvent("probe_retry_sleep", { - delay_ms: delayMs, - http_status: result.httpStatus, - retry_reason: customReason ? "semantic_readiness" : "http_status", - }); - sleepSync(delayMs); - attempt += 1; - result = probe.execute(); - trace.addTraceEvent("probe_result", { - attempt, - ok: result.ok, - http_status: result.httpStatus, - curl_status: result.curlStatus, - }); - } - return result; + let customRetryReason; + return retryUntil( + (attempt) => { + const result = probe.execute(); + trace.addTraceEvent("probe_result", { + attempt, + ok: result.ok, + http_status: result.httpStatus, + curl_status: result.curlStatus, + }); + return result; + }, + { + accept: (result) => { + customRetryReason = probe.retryReason?.(result); + return !shouldRetryHttpProbe(result) && !customRetryReason; + }, + retryDelaysMs: HTTP_PROBE_RETRY_DELAYS_MS, + onRetry: (result, delayMs) => { + const reason = customRetryReason || `returned HTTP ${result.httpStatus}`; + console.log( + ` ${probe.name} validation ${reason}; retrying in ${Math.round(delayMs / 1000)}s...`, + ); + trace.addTraceEvent("probe_retry_sleep", { + delay_ms: delayMs, + http_status: result.httpStatus, + retry_reason: customRetryReason ? "semantic_readiness" : "http_status", + }); + }, + sleep: sleepSync, + }, + ); }, ); } @@ -114,21 +115,19 @@ function executeProbeWithHttpRetry(probe) { // final probe result. Logs the same "retrying in Xs" notice the responses // retry loop emits so users on slow links see consistent progress messages. function runChatCompletionsRetryLoop(runProbe) { - let result = runProbe(); - if (result.ok) return result; - for (const delayMs of HTTP_PROBE_RETRY_DELAYS_MS) { - if (!isRetriableProbeResult(result)) break; - const reason = isTimeoutOrConnFailureStatus(result.curlStatus) - ? "timed out" - : `returned HTTP ${result.httpStatus}`; - console.log( - ` Chat Completions API validation ${reason}; retrying in ${Math.round(delayMs / 1000)}s...`, - ); - sleepSync(delayMs); - result = runProbe(); - if (result.ok) return result; - } - return result; + return retryUntil(runProbe, { + accept: (result) => result.ok || !isRetriableProbeResult(result), + retryDelaysMs: HTTP_PROBE_RETRY_DELAYS_MS, + onRetry: (result, delayMs) => { + const reason = isTimeoutOrConnFailureStatus(result.curlStatus) + ? "timed out" + : `returned HTTP ${result.httpStatus}`; + console.log( + ` Chat Completions API validation ${reason}; retrying in ${Math.round(delayMs / 1000)}s...`, + ); + }, + sleep: sleepSync, + }); } module.exports = { diff --git a/src/lib/onboard/docker-gpu-local-inference.test.ts b/src/lib/onboard/docker-gpu-local-inference.test.ts index 7e0bd73b992..6ed6f3a5a7c 100644 --- a/src/lib/onboard/docker-gpu-local-inference.test.ts +++ b/src/lib/onboard/docker-gpu-local-inference.test.ts @@ -238,6 +238,9 @@ describe("verifyDockerGpuSandboxLocalInference", () => { } expect(execInSandbox).toHaveBeenCalledTimes(3); expect(sleep).toHaveBeenCalledTimes(2); + + expect(sleep).toHaveBeenNthCalledWith(1, 2_000); + expect(sleep).toHaveBeenNthCalledWith(2, 2_000); }); it("fails when the inference route is up but the local backend errors (HTTP 502)", () => { diff --git a/src/lib/onboard/docker-gpu-local-inference.ts b/src/lib/onboard/docker-gpu-local-inference.ts index f0999099539..a51ad8aa213 100644 --- a/src/lib/onboard/docker-gpu-local-inference.ts +++ b/src/lib/onboard/docker-gpu-local-inference.ts @@ -1,6 +1,8 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +import { retryUntil } from "../core/retry"; + import { getLocalProviderLabel } from "../inference/local"; import type { SandboxGpuProofResult } from "../state/registry"; import { @@ -21,7 +23,7 @@ const { const DOCKER_GPU_INFERENCE_PROBE_CONNECT_TIMEOUT_SECS = 5; const DOCKER_GPU_INFERENCE_PROBE_MAX_TIME_SECS = 10; const DOCKER_GPU_INFERENCE_PROBE_MAX_ATTEMPTS = 3; -const DOCKER_GPU_INFERENCE_PROBE_RETRY_DELAY_SECS = 2; +const DOCKER_GPU_INFERENCE_PROBE_RETRY_DELAY_MS = 2_000; // The OpenShell inference route OpenClaw's LLM client actually uses inside the // sandbox. It is served by the OpenShell L7 proxy/router and routes to the @@ -160,7 +162,7 @@ export type SandboxExecResult = { export type DockerGpuSandboxInferenceVerifyDeps = { execInSandbox?: (sandboxName: string, script: string) => SandboxExecResult; - sleep?: (seconds: number) => void; + sleep?: (milliseconds: number) => void; }; export type DockerGpuSandboxInferenceVerification = @@ -221,54 +223,55 @@ function probeSandboxRuntimeInference( `--connect-timeout ${DOCKER_GPU_INFERENCE_PROBE_CONNECT_TIMEOUT_SECS} ` + `--max-time ${DOCKER_GPU_INFERENCE_PROBE_MAX_TIME_SECS} ${safeEndpoint} 2>/dev/null || echo 000); ` + `echo "HTTP_$code"`; - let last: RuntimeProbeOutcome = { - kind: "exec-failed", - detail: "openshell sandbox exec did not run (sandbox unreachable or exec timed out)", - }; - for (let attempt = 1; attempt <= DOCKER_GPU_INFERENCE_PROBE_MAX_ATTEMPTS; attempt++) { + const probe = (): RuntimeProbeOutcome => { const result = deps.execInSandbox(sandboxName, script); if (result === null) { - last = { + return { kind: "exec-failed", detail: "openshell sandbox exec did not run (sandbox unreachable or exec timed out)", }; - } else { - const out = (result.stdout || "").trim(); - if (out === "NO_CURL") return { kind: "no-curl" }; - const match = out.match(/HTTP_(\d{3})/); - if (match) { - const httpCode = match[1]; - // This gate is the #4509 runtime proof, so only a 2xx — the model list - // actually returned through the proxy with injected auth — counts as - // success. `000` is the reported failure (DNS / connection refused). A - // 4xx (wrong provider route, or auth the proxy failed to inject) or 5xx - // (local Ollama/vLLM backend down) means OpenClaw's real request would - // fail too, so do NOT report it as reachable. - if (/^2\d\d$/.test(httpCode)) return { kind: "ok", httpCode }; - last = { - kind: "unreachable", - detail: - httpCode === "000" - ? `${endpoint} returned HTTP 000 (DNS failure or connection refused)` - : `${endpoint} returned HTTP ${httpCode} (inference route reached but not usable — provider route/auth misconfigured or the local backend is failing)`, - }; - } else { - // The exec ran but produced no sentinel — the sandbox runtime exec - // path itself is broken (e.g. sandbox in Error, exec denied). Treat as - // an exec failure, NOT a missing-curl soft-skip, so we never declare - // success without actually exercising the runtime (#4509 review). - const noise = (out || result.stderr || "").slice(0, 160); - last = { - kind: "exec-failed", - detail: `unexpected sandbox exec output: ${noise}`, - }; - } } - if (attempt < DOCKER_GPU_INFERENCE_PROBE_MAX_ATTEMPTS) { - deps.sleep(DOCKER_GPU_INFERENCE_PROBE_RETRY_DELAY_SECS); + + const out = (result.stdout || "").trim(); + if (out === "NO_CURL") return { kind: "no-curl" }; + const match = out.match(/HTTP_(\d{3})/); + if (match) { + const httpCode = match[1]; + // This gate is the #4509 runtime proof, so only a 2xx — the model list + // actually returned through the proxy with injected auth — counts as + // success. `000` is the reported failure (DNS / connection refused). A + // 4xx (wrong provider route, or auth the proxy failed to inject) or 5xx + // (local Ollama/vLLM backend down) means OpenClaw's real request would + // fail too, so do NOT report it as reachable. + if (/^2\d\d$/.test(httpCode)) return { kind: "ok", httpCode }; + return { + kind: "unreachable", + detail: + httpCode === "000" + ? `${endpoint} returned HTTP 000 (DNS failure or connection refused)` + : `${endpoint} returned HTTP ${httpCode} (inference route reached but not usable — provider route/auth misconfigured or the local backend is failing)`, + }; } - } - return last; + + // The exec ran but produced no sentinel — the sandbox runtime exec + // path itself is broken (e.g. sandbox in Error, exec denied). Treat as + // an exec failure, NOT a missing-curl soft-skip, so we never declare + // success without actually exercising the runtime (#4509 review). + const noise = (out || result.stderr || "").slice(0, 160); + return { + kind: "exec-failed", + detail: `unexpected sandbox exec output: ${noise}`, + }; + }; + + return retryUntil(probe, { + accept: (result) => result.kind === "ok" || result.kind === "no-curl", + retryDelaysMs: Array.from( + { length: DOCKER_GPU_INFERENCE_PROBE_MAX_ATTEMPTS - 1 }, + () => DOCKER_GPU_INFERENCE_PROBE_RETRY_DELAY_MS, + ), + sleep: deps.sleep, + }); } /** @@ -308,8 +311,8 @@ export function verifyDockerGpuSandboxLocalInference( const execInSandbox = deps.execInSandbox ?? executeSandboxCommandForVerification; const sleep = deps.sleep ?? - ((seconds: number) => { - Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, Math.max(0, seconds) * 1000); + ((milliseconds: number) => { + Atomics.wait(new Int32Array(new SharedArrayBuffer(4)), 0, 0, Math.max(0, milliseconds)); }); const outcome = probeSandboxRuntimeInference(options.sandboxName, endpoint, { diff --git a/src/lib/onboard/docker-gpu-patch-rollback.ts b/src/lib/onboard/docker-gpu-patch-rollback.ts index 79be9a39fe0..9422f21899a 100644 --- a/src/lib/onboard/docker-gpu-patch-rollback.ts +++ b/src/lib/onboard/docker-gpu-patch-rollback.ts @@ -8,6 +8,7 @@ import { dockerStart as defaultDockerStart, dockerStop as defaultDockerStop, } from "../adapters/docker"; +import { retryUntil } from "../core/retry"; import { hasZeroDockerExitStatus } from "./docker-command-result"; import { fullDockerContainerId } from "./docker-gpu-patch-clone"; import { DOCKER_GPU_PATCH_TIMEOUT_MS } from "./docker-gpu-patch-constants"; @@ -29,7 +30,7 @@ type DockerRenameFn = ( type DockerRunFn = (args: readonly string[], opts?: DockerRunOptions) => DockerRunResult; const REPLACEMENT_PRESENCE_ATTEMPTS = 3; -const REPLACEMENT_PRESENCE_RETRY_SECONDS = 0.5; +const REPLACEMENT_PRESENCE_RETRY_MS = 500; function sleepBeforeReplacementPresenceRetry(seconds: number): void { if (seconds <= 0 || !Number.isFinite(seconds)) return; @@ -77,23 +78,28 @@ function observeReplacementPresence( ): DockerGpuPatchRollbackOutcome["replacementPresence"] { const exactId = fullDockerContainerId(containerId); if (!exactId) return "unknown"; - for (let attempt = 0; attempt < REPLACEMENT_PRESENCE_ATTEMPTS; attempt += 1) { - const result = deps.dockerRun( - ["ps", "-a", "--no-trunc", "--filter", `id=${exactId}`, "--format", "{{.ID}}"], - options, - ); - if (hasZeroDockerExitStatus(result)) { - const ids = outputText(result.stdout) - .split(/\r?\n/u) - .map((value) => value.trim()) - .filter(Boolean); - return ids.some((id) => fullDockerContainerId(id) === exactId) ? "present" : "absent"; - } - if (attempt + 1 < REPLACEMENT_PRESENCE_ATTEMPTS) { - deps.sleep(REPLACEMENT_PRESENCE_RETRY_SECONDS); - } - } - return "unknown"; + const result = retryUntil( + () => + deps.dockerRun( + ["ps", "-a", "--no-trunc", "--filter", `id=${exactId}`, "--format", "{{.ID}}"], + options, + ), + { + accept: hasZeroDockerExitStatus, + retryDelaysMs: Array.from( + { length: REPLACEMENT_PRESENCE_ATTEMPTS - 1 }, + () => REPLACEMENT_PRESENCE_RETRY_MS, + ), + sleep: (milliseconds) => deps.sleep(milliseconds / 1000), + }, + ); + if (!hasZeroDockerExitStatus(result)) return "unknown"; + + const ids = outputText(result.stdout) + .split(/\r?\n/u) + .map((value) => value.trim()) + .filter(Boolean); + return ids.some((id) => fullDockerContainerId(id) === exactId) ? "present" : "absent"; } export function rollbackToBackupContainer( diff --git a/src/lib/shields/inference-convergence.ts b/src/lib/shields/inference-convergence.ts index dabc017b48c..7e4bf37639f 100644 --- a/src/lib/shields/inference-convergence.ts +++ b/src/lib/shields/inference-convergence.ts @@ -7,6 +7,8 @@ import { } from "../actions/sandbox/connect-inference-route-probe"; import { buildOpenshellCommand } from "../adapters/openshell/command-argv"; +import { retryUntil } from "../core/retry"; + const DEFAULT_MAX_ATTEMPTS = 4; const DEFAULT_RETRY_DELAY_MS = 500; const INFERENCE_ROUTE_PROBE_TIMEOUT_MS = 10_000; @@ -65,28 +67,34 @@ export function waitForHermesInferenceRouteConvergence( ? Math.max(0, Math.trunc(configuredRetryDelayMs)) : DEFAULT_RETRY_DELAY_MS; const buildCommand = options.buildOpenshellCommand ?? buildOpenshellCommand; - const sleep = options.sleep ?? sleepMs; - let httpStatus = 0; - - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { - const probe = options.run( - buildCommand(buildSandboxInferenceRouteProbeArgs(sandboxName, { name: "hermes" })), - { - ignoreError: true, - suppressOutput: true, - timeout: INFERENCE_ROUTE_PROBE_TIMEOUT_MS, - }, - ); - const parsed = parseSandboxInferenceRouteProbeResult({ - status: probe.status, - output: String(probe.stdout ?? ""), - stderr: String(probe.stderr ?? ""), - }); - httpStatus = parsed.httpStatus; - const usable = parsed.healthy && httpStatus >= 200 && httpStatus < 300; - if (usable) return { ok: true, attempts: attempt, httpStatus }; - if (attempt < maxAttempts) sleep(retryDelayMs); - } + const retryDelaysMs = Array.from({ length: maxAttempts - 1 }, () => retryDelayMs); - return { ok: false, attempts: maxAttempts, httpStatus }; + return retryUntil( + (attempt) => { + const probe = options.run( + buildCommand(buildSandboxInferenceRouteProbeArgs(sandboxName, { name: "hermes" })), + { + ignoreError: true, + suppressOutput: true, + timeout: INFERENCE_ROUTE_PROBE_TIMEOUT_MS, + }, + ); + const parsed = parseSandboxInferenceRouteProbeResult({ + status: probe.status, + output: String(probe.stdout ?? ""), + stderr: String(probe.stderr ?? ""), + }); + const httpStatus = parsed.httpStatus; + return { + ok: parsed.healthy && httpStatus >= 200 && httpStatus < 300, + attempts: attempt, + httpStatus, + }; + }, + { + accept: (result) => result.ok, + retryDelaysMs, + sleep: options.sleep ?? sleepMs, + }, + ); } diff --git a/src/lib/verify-deployment.ts b/src/lib/verify-deployment.ts index 1918e762b69..2eeb752cd1a 100644 --- a/src/lib/verify-deployment.ts +++ b/src/lib/verify-deployment.ts @@ -21,6 +21,8 @@ import { parseVersionFromText } from "./adapters/openshell/client"; import { compareChannelSets, type RuntimeChannelStatus } from "./channel-runtime-status"; import type { DashboardDeliveryChain } from "./dashboard/contract"; import { listMessagingChannelsWithoutCredentials } from "./messaging/channels"; + +import { retryUntilAsync } from "./core/retry"; import { buildCustomOpenClawRuntimeFailureHints, classifyOpenClawRuntimeFailure, @@ -198,14 +200,11 @@ async function verifyGatewayInSandbox( retryDelaysMs: readonly number[], sleep: (ms: number) => Promise, ): Promise<{ reachable: boolean; httpCode: number; detail: string }> { - let last = probeGatewayInSandboxOnce(sandboxName, chain, deps); - if (last.reachable) return last; - for (const delayMs of retryDelaysMs) { - await sleep(delayMs); - last = probeGatewayInSandboxOnce(sandboxName, chain, deps); - if (last.reachable) return last; - } - return last; + return retryUntilAsync(() => probeGatewayInSandboxOnce(sandboxName, chain, deps), { + accept: (result) => result.reachable, + retryDelaysMs, + sleep, + }); } /** @@ -253,14 +252,11 @@ async function verifyInferenceRoute( retryDelaysMs: readonly number[], sleep: (ms: number) => Promise, ): Promise<{ status: InferenceRouteStatus; detail: string }> { - let last = probeInferenceRouteOnce(sandboxName, deps); - if (last.status === "ok") return last; - for (const delayMs of retryDelaysMs) { - await sleep(delayMs); - last = probeInferenceRouteOnce(sandboxName, deps); - if (last.status === "ok") return last; - } - return last; + return retryUntilAsync(() => probeInferenceRouteOnce(sandboxName, deps), { + accept: (result) => result.status === "ok", + retryDelaysMs, + sleep, + }); } /** @@ -289,14 +285,11 @@ async function verifyDashboardFromHost( retryDelaysMs: readonly number[], sleep: (ms: number) => Promise, ): Promise<{ reachable: boolean; detail: string }> { - let last = probeDashboardFromHostOnce(chain, deps); - if (last.reachable) return last; - for (const delayMs of retryDelaysMs) { - await sleep(delayMs); - last = probeDashboardFromHostOnce(chain, deps); - if (last.reachable) return last; - } - return last; + return retryUntilAsync(() => probeDashboardFromHostOnce(chain, deps), { + accept: (result) => result.reachable, + retryDelaysMs, + sleep, + }); } /** diff --git a/test/onboard-prepared-build-context.test.ts b/test/onboard-prepared-build-context.test.ts index c7265a9163a..9718bf71255 100644 --- a/test/onboard-prepared-build-context.test.ts +++ b/test/onboard-prepared-build-context.test.ts @@ -17,6 +17,7 @@ type PreparedContextResult = { commands: string[]; errorMessage: string | null; patchCalls: number; + patchSleepUsesSeconds: boolean | null; planFromRefs: string[]; registerCalls: Array<{ imageTag?: string | null }>; resolvedBuildIds: string[]; @@ -69,6 +70,7 @@ function runPreparedContextScenario(scenario: PreparedContextScenario): Prepared const dockerGpuSandboxCreatePath = JSON.stringify( path.join(repoRoot, "src", "lib", "onboard", "docker-gpu-sandbox-create.ts"), ); + const waitPath = JSON.stringify(path.join(repoRoot, "src", "lib", "core", "wait.ts")); const script = String.raw` const fs = require("node:fs"); @@ -83,6 +85,7 @@ const dockerfilePatchFlow = require(${dockerfilePatchFlowPath}); const sandboxCreatePlanMaterialization = require(${sandboxCreatePlanPath}); const imageTag = require(${imageTagPath}); const dockerGpuSandboxCreate = require(${dockerGpuSandboxCreatePath}); +const wait = require(${waitPath}); const { loadAgent } = require(${agentDefsPath}); const scenario = ${JSON.stringify(scenario)}; @@ -95,9 +98,12 @@ const planFromRefs = []; const resolvedBuildIds = []; let cleanupCalls = 0; let patchCalls = 0; +let patchSleepUsesSeconds = null; let stageCalls = 0; -dockerGpuSandboxCreate.createDockerGpuSandboxCreatePatch = () => ({ +dockerGpuSandboxCreate.createDockerGpuSandboxCreatePatch = (options) => { + patchSleepUsesSeconds = options.deps.sleep === wait.sleepSeconds; + return { maybeApplyDuringCreate: () => {}, createFailureMessage: () => null, exitOnPatchError: async () => {}, @@ -109,7 +115,8 @@ dockerGpuSandboxCreate.createDockerGpuSandboxCreatePatch = () => ({ selectedMode: () => null, printReadinessFailureIfEnabled: () => {}, verifyGpuOrExit: async (verify) => verify(sandboxName), -}); + }; +}; buildContextStage.stageCreateSandboxBuildContext = () => { stageCalls += 1; @@ -233,6 +240,7 @@ const { createSandbox } = require(${onboardPath}); commands, errorMessage, patchCalls, + patchSleepUsesSeconds, planFromRefs, registerCalls, resolvedBuildIds, @@ -294,6 +302,16 @@ describe("onboard prepared DCode build context", () => { ); }); + + it("passes the seconds-based sleep helper to the Docker GPU patch during prepared-context onboarding (#9218)", { + timeout: 90_000, + }, () => { + const result = runPreparedContextScenario("create"); + + assert.equal(result.errorMessage, null); + assert.equal(result.patchSleepUsesSeconds, true); + }); + it("rejects a prepared context combined with a custom Dockerfile (#6195)", { timeout: 90_000, }, () => { diff --git a/test/wait.test.ts b/test/wait.test.ts index c382dc4034d..8e59bedae03 100644 --- a/test/wait.test.ts +++ b/test/wait.test.ts @@ -4,6 +4,8 @@ import assert from "node:assert"; import { createServer, type AddressInfo } from "node:net"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { retryUntil, retryUntilAsync } from "../src/lib/core/retry.js"; + import { buildLoopbackProbeEnv, sleepMs, @@ -47,6 +49,113 @@ describe("wait utility", () => { assert.ok(duration < 50, `duration ${duration}ms > 50ms`); }); + const throwWhenSelected = (selected: boolean, error: Error): void => + selected ? (() => { + throw error; + })() : undefined; + + + const retryCases = [ + { label: "accepts the first result", acceptAt: 1, delays: [10, 20], attempt: 1 }, + { label: "accepts the third result", acceptAt: 3, delays: [10, 20, 30], attempt: 3 }, + { label: "returns the exhausted result", acceptAt: 0, delays: [10, 20], attempt: 3 }, + { label: "runs once without retries", acceptAt: 0, delays: [], attempt: 1 }, + ] as const; + + it.each(retryCases)("retryUntil $label (#9218)", ({ acceptAt, delays, attempt }) => { + const operation = vi.fn((currentAttempt: number) => `result-${currentAttempt}`); + const onRetry = vi.fn(); + const sleep = vi.fn(); + + const result = retryUntil(operation, { + accept: (_value, currentAttempt) => currentAttempt === acceptAt, + retryDelaysMs: delays, + onRetry, + sleep, + }); + + expect(result).toBe(`result-${attempt}`); + expect(operation).toHaveBeenCalledTimes(attempt); + expect(sleep.mock.calls).toEqual(delays.slice(0, attempt - 1).map((delay) => [delay])); + expect(onRetry).toHaveBeenCalledTimes(attempt - 1); + }); + + it.each(["operation", "onRetry", "sleep"] as const)( + "retryUntil propagates an error from %s before the next attempt (#9218)", + (failure) => { + const error = new Error(`${failure} failed`); + const operation = vi.fn(() => { + throwWhenSelected(failure === "operation", error); + return "retry"; + }); + const onRetry = vi.fn(() => { + throwWhenSelected(failure === "onRetry", error); + }); + const sleep = vi.fn(() => { + throwWhenSelected(failure === "sleep", error); + }); + + expect(() => + retryUntil(operation, { + accept: () => false, + retryDelaysMs: [10], + onRetry, + sleep, + }), + ).toThrow(error); + expect(operation).toHaveBeenCalledOnce(); + expect(onRetry).toHaveBeenCalledTimes(failure === "operation" ? 0 : 1); + expect(sleep).toHaveBeenCalledTimes(failure === "sleep" ? 1 : 0); + }, + ); + + it.each(retryCases)("retryUntilAsync $label (#9218)", async ({ acceptAt, delays, attempt }) => { + const operation = vi.fn(async (currentAttempt: number) => `result-${currentAttempt}`); + const onRetry = vi.fn(async () => {}); + const sleep = vi.fn(async () => {}); + + const result = await retryUntilAsync(operation, { + accept: (_value, currentAttempt) => currentAttempt === acceptAt, + retryDelaysMs: delays, + onRetry, + sleep, + }); + + expect(result).toBe(`result-${attempt}`); + expect(operation).toHaveBeenCalledTimes(attempt); + expect(sleep.mock.calls).toEqual(delays.slice(0, attempt - 1).map((delay) => [delay])); + expect(onRetry).toHaveBeenCalledTimes(attempt - 1); + }); + + it.each(["operation", "onRetry", "sleep"] as const)( + "retryUntilAsync propagates an error from %s before the next attempt (#9218)", + async (failure) => { + const error = new Error(`${failure} failed`); + const operation = vi.fn(async () => { + throwWhenSelected(failure === "operation", error); + return "retry"; + }); + const onRetry = vi.fn(async () => { + throwWhenSelected(failure === "onRetry", error); + }); + const sleep = vi.fn(async () => { + throwWhenSelected(failure === "sleep", error); + }); + + await expect( + retryUntilAsync(operation, { + accept: () => false, + retryDelaysMs: [10], + onRetry, + sleep, + }), + ).rejects.toBe(error); + expect(operation).toHaveBeenCalledOnce(); + expect(onRetry).toHaveBeenCalledTimes(failure === "operation" ? 0 : 1); + expect(sleep).toHaveBeenCalledTimes(failure === "sleep" ? 1 : 0); + }, + ); + it("waitUntil returns immediately when the condition is already true", () => { const sleeps: number[] = []; let attempts = 0;