From 95cfbc3c7d8bb5b09c71ecdc8d10902017fb22db Mon Sep 17 00:00:00 2001 From: Tony Luo Date: Mon, 6 Jul 2026 14:16:35 +0800 Subject: [PATCH 01/12] fix(inference): validate custom Anthropic endpoint streaming during onboarding (#6289) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Custom Anthropic-compatible endpoints were validated with a non-streaming /v1/messages probe only. Gateways whose non-streaming responses are valid but whose SSE streaming layer is malformed (e.g. duplicate message_start events with the same message id, observed on the Inference Hub route) passed onboarding and then failed at runtime inside the sandbox with Hermes' cryptic 'no final response was produced' (exit 1) — agent runtimes only use the streaming path. Add runAnthropicStreamingEventProbe: a stream:true /v1/messages probe that validates the SSE event sequence (exactly one message_start, at least one content_block_delta, one message_stop). Wire it into probeAnthropicEndpoint behind {probeStreaming} and enable it for the custom Anthropic-compatible onboarding path, mirroring the existing /v1/responses streaming validation (#1833). Reasoning mode skips the streaming probe, matching the custom OpenAI-compatible path. The official Anthropic provider path is unchanged. Malformed streams now fail validation at onboarding with an actionable message naming the duplicate/missing events instead of surfacing later as an empty final response at runtime. The Anthropic credential-retry integration tests move to a focused test/onboard-selection-anthropic-retry.test.ts (with the fake curl now serving SSE for stream:true probes), shrinking onboard-selection.test.ts below its legacy size budget, which is ratcheted down accordingly. Signed-off-by: Tony Luo --- ci/test-file-size-budget.json | 2 +- docs/inference/inference-options.mdx | 9 +- docs/reference/troubleshooting.mdx | 20 ++ src/lib/adapters/http/probe.test.ts | 219 ++++++++++++++++ src/lib/adapters/http/probe.ts | 203 +++++++++++++-- src/lib/inference/probe-anthropic.test.ts | 113 ++++++++ src/lib/inference/probe-anthropic.ts | 106 ++++++-- .../inference-selection-validation.test.ts | 113 ++++++++ .../onboard/inference-selection-validation.ts | 12 +- .../onboard-selection-anthropic-retry.test.ts | 246 ++++++++++++++++++ test/onboard-selection.test.ts | 211 --------------- 11 files changed, 998 insertions(+), 256 deletions(-) create mode 100644 test/onboard-selection-anthropic-retry.test.ts diff --git a/ci/test-file-size-budget.json b/ci/test-file-size-budget.json index efa6090688e..19db4ba6fad 100644 --- a/ci/test-file-size-budget.json +++ b/ci/test-file-size-budget.json @@ -10,7 +10,7 @@ "test/install-preflight.test.ts": 3934, "test/nemoclaw-start.test.ts": 4827, "test/onboard-messaging.test.ts": 2062, - "test/onboard-selection.test.ts": 6146, + "test/onboard-selection.test.ts": 5935, "test/onboard.test.ts": 4057, "test/policies.test.ts": 2332 } diff --git a/docs/inference/inference-options.mdx b/docs/inference/inference-options.mdx index 4970e5f9a94..bddc1a4a9a4 100644 --- a/docs/inference/inference-options.mdx +++ b/docs/inference/inference-options.mdx @@ -243,7 +243,7 @@ Other provider credentials, such as `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMI | NVIDIA Endpoints | Validates through `/v1/chat/completions` only; NemoClaw skips the `/v1/responses` probe because NVIDIA Build does not expose `/v1/responses` (returns 404 for every model). | | Google Gemini | Validates through Gemini's OpenAI-compatible chat-completions path only; NemoClaw skips the `/v1/responses` probe because Gemini does not support the Responses API. | | Other OpenAI-compatible endpoint | Tries `/v1/responses` first with a tool-calling probe; falls back to `/v1/chat/completions`. Selected runtime API defaults to `/v1/chat/completions`; set `NEMOCLAW_PREFERRED_API=openai-responses` to allow `/v1/responses` at runtime when validation succeeds. | -| Anthropic-compatible | Tries `/v1/messages`. | +| Anthropic-compatible | Tries `/v1/messages` with a non-streaming request, then repeats the request with `stream: true` and validates the SSE event sequence. Set `NEMOCLAW_REASONING=true` to skip the streaming check for reasoning-only endpoints. | | NVIDIA Endpoints (manual model entry) | Validates the model name against the catalog API. | | Compatible endpoints | Sends a real inference request because many proxies do not expose a `/models` endpoint. For OpenAI-compatible endpoints, the probe tries `/v1/responses` first then falls back to `/v1/chat/completions`; the selected runtime API defaults to `/v1/chat/completions`. Set `NEMOCLAW_PREFERRED_API=openai-responses` to allow `/v1/responses` at runtime when validation succeeds. | | Local NVIDIA NIM | Validates through `/v1/chat/completions` only; NemoClaw skips the `/v1/responses` probe (same as NVIDIA Endpoints). | @@ -348,6 +348,13 @@ If your local server implements the Anthropic Messages API (`/v1/messages`), cho $$nemoclaw onboard ``` +NemoClaw validates the endpoint by sending a non-streaming `/v1/messages` request, then a `stream: true` request to the same path. +The streaming check requires a well-formed SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). +Endpoints and gateways whose non-streaming responses work but whose streaming layer is malformed fail validation during onboarding instead of failing later at runtime inside the sandbox. +Refer to [Onboarding fails with "Anthropic Messages streaming on this endpoint emits duplicate message_start"](../reference/troubleshooting#onboarding-fails-with-anthropic-messages-streaming-on-this-endpoint-emits-duplicate-message_start) if the streaming check fails. +Set `NEMOCLAW_REASONING=true` to skip the streaming check when the endpoint serves a reasoning-only model. +Agent runs still use the streaming path, so skipping the check moves any streaming defect to runtime. + For non-interactive setup, use `NEMOCLAW_PROVIDER=anthropicCompatible` and set `COMPATIBLE_ANTHROPIC_API_KEY`. ```bash diff --git a/docs/reference/troubleshooting.mdx b/docs/reference/troubleshooting.mdx index 861ffc077ee..73c847d66d7 100644 --- a/docs/reference/troubleshooting.mdx +++ b/docs/reference/troubleshooting.mdx @@ -1212,6 +1212,26 @@ Do not rely on `NEMOCLAW_INFERENCE_API_OVERRIDE` alone. It patches the config at container startup but does not update the Dockerfile ARG baked into the image. A fresh `$$nemoclaw onboard` is the reliable fix. +### Onboarding fails with "Anthropic Messages streaming on this endpoint emits duplicate message_start" + +Validation for the **Other Anthropic-compatible endpoint** provider ends with an error like: + +```text +Anthropic Messages API (streaming): Anthropic Messages streaming on this endpoint emits duplicate message_start (2 events for one request). Agent runs use the streaming path and would fail with an empty final response. +``` + +During onboarding, NemoClaw sends a `stream: true` request to `/v1/messages` and validates the SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). +This error means the streaming layer on the endpoint or gateway is malformed even though its non-streaming responses are valid. +A working non-streaming response does not imply that streaming works. +Some inference gateways proxy plain requests correctly but corrupt the SSE stream, for example by emitting `message_start` twice for one request. +Agent runs always use the streaming path, so without this check the defect would first surface inside the sandbox as a runtime failure, such as Hermes reporting that no final response was produced. + +Fix the streaming layer on the endpoint or gateway, or onboard with a different Anthropic-compatible endpoint. +The official Anthropic provider does not run this check and is not affected. +If a sandbox created by an older release fails at runtime with an empty final response on an Anthropic-compatible endpoint, re-run `$$nemoclaw onboard` so the streaming check can diagnose the endpoint. +If the endpoint serves a reasoning-only model, set `NEMOCLAW_REASONING=true` to skip the streaming check. +Streaming defects then surface at runtime instead of during onboarding. + ### `NEMOCLAW_DISABLE_DEVICE_AUTH=1` does not change an existing sandbox This is expected behavior. diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index c7b8b2d2a4c..e9ef219c321 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -10,6 +10,7 @@ import { restoreEnvBulk } from "../../../../test/helpers/env-test-helpers"; import { flushTrace, resetTraceForTests, TRACE_FILE_ENV, type TraceArtifact } from "../../trace"; import { getCurlTimingArgs, + runAnthropicStreamingEventProbe, runChatCompletionsStreamingProbe, runCurlProbe, runStreamingEventProbe, @@ -800,3 +801,221 @@ describe("runStreamingEventProbe", () => { }); }); }); + +describe("runAnthropicStreamingEventProbe", () => { + /** Helper to build a spawnSyncImpl that writes SSE content to the -o file. */ + function mockStreaming(sseBody: string, exitCode = 0) { + return (_command: string, args: readonly string[]) => { + const oIdx = args.indexOf("-o"); + if (oIdx !== -1) { + const outputPath = args[oIdx + 1]; + if (typeof outputPath === "string") { + fs.writeFileSync(outputPath, sseBody); + } + } + return { + pid: 1, + output: [], + stdout: "", + stderr: "", + status: exitCode, + signal: null, + }; + }; + } + + const healthyStream = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_1"}}', + "", + "event: content_block_start", + 'data: {"type":"content_block_start","index":0}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"OK"}}', + "", + "event: content_block_stop", + 'data: {"type":"content_block_stop","index":0}', + "", + "event: message_delta", + 'data: {"type":"message_delta","delta":{"stop_reason":"end_turn"}}', + "", + "event: message_stop", + 'data: {"type":"message_stop"}', + "", + ].join("\n"); + + it("passes when the Anthropic Messages event sequence is well formed", () => { + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(healthyStream) }, + ); + + expect(result.ok).toBe(true); + expect(result.missingEvents).toEqual([]); + expect(result.duplicateEvents).toEqual([]); + }); + + it("fails when message_start is emitted twice for one request (#6289)", () => { + const duplicatedStart = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_4963f1e3"}}', + "", + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_4963f1e3"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}', + "", + "event: message_stop", + 'data: {"type":"message_stop"}', + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(duplicatedStart) }, + ); + + expect(result.ok).toBe(false); + expect(result.duplicateEvents).toEqual(["message_start"]); + expect(result.missingEvents).toEqual([]); + expect(result.message).toContain("duplicate message_start (2 events for one request)"); + }); + + it("fails when the stream carries no content deltas", () => { + const emptyStream = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_1"}}', + "", + "event: message_stop", + 'data: {"type":"message_stop"}', + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(emptyStream) }, + ); + + expect(result.ok).toBe(false); + expect(result.missingEvents).toEqual(["content_block_delta"]); + expect(result.message).toContain("missing required events: content_block_delta"); + }); + + it("fails when the stream never terminates with message_stop", () => { + const unterminatedStream = [ + "event: message_start", + 'data: {"type":"message_start","message":{"id":"msg_1"}}', + "", + "event: content_block_delta", + 'data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"OK"}}', + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(unterminatedStream, 28) }, + ); + + expect(result.ok).toBe(false); + expect(result.missingEvents).toEqual(["message_stop"]); + }); + + it("still passes if curl exits with 28 (timeout) but the full sequence was captured", () => { + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(healthyStream, 28) }, + ); + + expect(result.ok).toBe(true); + }); + + it("fails on spawn error", () => { + const result = runAnthropicStreamingEventProbe(["-sS", "https://example.test/v1/messages"], { + spawnSyncImpl: () => { + const error = Object.assign(new Error("spawn ENOENT"), { code: "ENOENT" }); + return { + pid: 1, + output: [], + stdout: "", + stderr: "", + status: null, + signal: null, + error, + }; + }, + }); + + expect(result.ok).toBe(false); + expect(result.message).toContain("Streaming probe failed"); + }); + + it("records curl_result metadata including duplicate counts", () => { + withTraceFile((traceFile) => { + const duplicatedStart = [ + "event: message_start", + "data: {}", + "", + "event: message_start", + "data: {}", + "", + "event: content_block_delta", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(duplicatedStart) }, + ); + + expect(result.ok).toBe(false); + flushTrace(); + const artifact = JSON.parse(fs.readFileSync(traceFile, "utf8")) as TraceArtifact; + const span = artifact.resource_spans[0].scope_spans[0].spans.find( + (entry) => entry.name === "nemoclaw.inference.curl_anthropic_streaming_probe", + ); + expect(span?.events[0].attributes).toMatchObject({ + ok: false, + missing_events_count: 0, + duplicate_events_count: 1, + curl_status: 0, + }); + }); + }); + + it("cleans up temp files after probe", () => { + let outputPath = ""; + runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { + spawnSyncImpl: (_command, args) => { + const oIdx = args.indexOf("-o"); + if (oIdx !== -1) { + const nextArg = args[oIdx + 1]; + if (typeof nextArg === "string") { + outputPath = nextArg; + fs.writeFileSync(outputPath, healthyStream); + } + } + return { + pid: 1, + output: [], + stdout: "", + stderr: "", + status: 0, + signal: null, + }; + }, + }, + ); + + expect(outputPath).not.toBe(""); + expect(fs.existsSync(outputPath)).toBe(false); + expect(fs.existsSync(path.dirname(outputPath))).toBe(false); + }); +}); diff --git a/src/lib/adapters/http/probe.ts b/src/lib/adapters/http/probe.ts index 0f951f67d8f..c7f5a82cb51 100644 --- a/src/lib/adapters/http/probe.ts +++ b/src/lib/adapters/http/probe.ts @@ -445,11 +445,26 @@ export function runStreamingEventProbe( ); } -function runStreamingEventProbeImpl( +interface SseEventCaptureResult { + ok: boolean; + curlStatus: number; + /** Transport/execution error detail when `ok` is false. */ + detail: string; + /** Occurrence count per SSE `event:` type parsed from the response body. */ + eventCounts: Map; +} + +/** + * Run a streaming curl probe and count the SSE `event:` types in the + * response body. Shared by the Responses API and Anthropic Messages + * streaming validators, which apply protocol-specific rules to the counts. + */ +function captureSseEventCounts( argv: string[], - opts: CurlProbeOptions = {}, -): StreamingProbeResult { - const bodyFile = secureTempFile("nemoclaw-streaming-probe", ".sse"); + opts: CurlProbeOptions, + tempPrefix: string, +): SseEventCaptureResult { + const bodyFile = secureTempFile(tempPrefix, ".sse"); try { const { args, url } = validateCurlProbeArgs(argv, opts); const spawnSyncImpl = opts.spawnSyncImpl ?? spawnSync; @@ -478,34 +493,52 @@ function runStreamingEventProbeImpl( const detail = result.error ? String(result.error.message || result.error) : String(result.stderr || ""); - emitCurlResultTraceEvent({ - ok: false, - missing_events_count: REQUIRED_STREAMING_EVENTS.length, - curl_status: curlStatus, - }); - return { - ok: false, - missingEvents: REQUIRED_STREAMING_EVENTS, - message: `Streaming probe failed: ${compactText(detail).slice(0, 200)}`, - }; + return { ok: false, curlStatus, detail, eventCounts: new Map() }; } // Parse SSE event types from the raw output. // Each event line looks like: "event: response.output_text.delta" - const eventTypes = new Set(); + const eventCounts = new Map(); for (const line of body.split("\n")) { const match = /^event:\s*(.+)$/i.exec(line.trim()); if (match) { - eventTypes.add(match[1].trim()); + const eventType = match[1].trim(); + eventCounts.set(eventType, (eventCounts.get(eventType) ?? 0) + 1); } } + return { ok: true, curlStatus: result.status ?? 0, detail: "", eventCounts }; + } finally { + cleanupTempDir(bodyFile, tempPrefix); + } +} + +function runStreamingEventProbeImpl( + argv: string[], + opts: CurlProbeOptions = {}, +): StreamingProbeResult { + try { + const capture = captureSseEventCounts(argv, opts, "nemoclaw-streaming-probe"); + if (!capture.ok) { + emitCurlResultTraceEvent({ + ok: false, + missing_events_count: REQUIRED_STREAMING_EVENTS.length, + curl_status: capture.curlStatus, + }); + return { + ok: false, + missingEvents: REQUIRED_STREAMING_EVENTS, + message: `Streaming probe failed: ${compactText(capture.detail).slice(0, 200)}`, + }; + } - const missing = REQUIRED_STREAMING_EVENTS.filter((e) => !eventTypes.has(e)); + const missing = REQUIRED_STREAMING_EVENTS.filter( + (e) => (capture.eventCounts.get(e) ?? 0) === 0, + ); if (missing.length > 0) { emitCurlResultTraceEvent({ ok: false, missing_events_count: missing.length, - curl_status: result.status ?? 0, + curl_status: capture.curlStatus, }); return { ok: false, @@ -519,7 +552,7 @@ function runStreamingEventProbeImpl( emitCurlResultTraceEvent({ ok: true, missing_events_count: 0, - curl_status: result.status ?? 0, + curl_status: capture.curlStatus, }); return { ok: true, missingEvents: [], message: "" }; } catch (error) { @@ -536,7 +569,135 @@ function runStreamingEventProbeImpl( missingEvents: REQUIRED_STREAMING_EVENTS, message: `Streaming probe error: ${detail}`, }; - } finally { - cleanupTempDir(bodyFile, "nemoclaw-streaming-probe"); + } +} + +/** + * The Anthropic Messages streaming event sequence that agent runtimes + * (Hermes `api_mode=anthropic_messages`, OpenClaw Anthropic routes) require + * from a `/v1/messages` endpoint: one `message_start`, at least one + * `content_block_delta` carrying incremental content, and a terminal + * `message_stop`. + */ +const REQUIRED_ANTHROPIC_STREAMING_EVENTS = [ + "message_start", + "content_block_delta", + "message_stop", +]; + +/** + * Anthropic Messages events that must appear exactly once per stream. + * Anthropic-compatible gateways with broken streaming layers have been + * observed emitting `message_start` twice with the same message id, which + * corrupts streaming-client state machines: the agent run then ends with an + * empty final response even though the non-streaming path works (#6289). + */ +const SINGLETON_ANTHROPIC_STREAMING_EVENTS = ["message_start"]; + +export interface AnthropicStreamingProbeResult { + ok: boolean; + missingEvents: string[]; + duplicateEvents: string[]; + message: string; +} + +/** + * Send a streaming request to an Anthropic-compatible `/v1/messages` + * endpoint and verify the SSE event stream is well formed: the required + * event sequence is present and no singleton event is duplicated. + * + * This catches gateways whose non-streaming responses are valid but whose + * streaming layer is broken — runtime agents only use the streaming path, + * so without this probe the defect first surfaces as a cryptic + * "no final response was produced" failure inside the sandbox. + */ +export function runAnthropicStreamingEventProbe( + argv: string[], + opts: CurlProbeOptions = {}, +): AnthropicStreamingProbeResult { + return withTraceSpan( + "nemoclaw.inference.curl_anthropic_streaming_probe", + getCurlProbeTraceAttributes(argv, opts), + () => runAnthropicStreamingEventProbeImpl(argv, opts), + ); +} + +function runAnthropicStreamingEventProbeImpl( + argv: string[], + opts: CurlProbeOptions = {}, +): AnthropicStreamingProbeResult { + try { + const capture = captureSseEventCounts(argv, opts, "nemoclaw-anthropic-streaming-probe"); + if (!capture.ok) { + emitCurlResultTraceEvent({ + ok: false, + missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, + duplicate_events_count: 0, + curl_status: capture.curlStatus, + }); + return { + ok: false, + missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, + duplicateEvents: [], + message: `Streaming probe failed: ${compactText(capture.detail).slice(0, 200)}`, + }; + } + + const missing = REQUIRED_ANTHROPIC_STREAMING_EVENTS.filter( + (e) => (capture.eventCounts.get(e) ?? 0) === 0, + ); + const duplicates = SINGLETON_ANTHROPIC_STREAMING_EVENTS.filter( + (e) => (capture.eventCounts.get(e) ?? 0) > 1, + ); + if (missing.length > 0 || duplicates.length > 0) { + const problems: string[] = []; + if (duplicates.length > 0) { + const detail = duplicates + .map((e) => `${e} (${capture.eventCounts.get(e)} events for one request)`) + .join(", "); + problems.push(`emits duplicate ${detail}`); + } + if (missing.length > 0) { + problems.push(`is missing required events: ${missing.join(", ")}`); + } + emitCurlResultTraceEvent({ + ok: false, + missing_events_count: missing.length, + duplicate_events_count: duplicates.length, + curl_status: capture.curlStatus, + }); + return { + ok: false, + missingEvents: missing, + duplicateEvents: duplicates, + message: + `Anthropic Messages streaming on this endpoint ${problems.join(" and ")}. ` + + "Agent runs use the streaming path and would fail with an empty final response.", + }; + } + + emitCurlResultTraceEvent({ + ok: true, + missing_events_count: 0, + duplicate_events_count: 0, + curl_status: capture.curlStatus, + }); + return { ok: true, missingEvents: [], duplicateEvents: [], message: "" }; + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + const curlStatus = + typeof error === "object" && error && "status" in error ? Number(error.status) || 1 : 1; + emitCurlResultTraceEvent({ + ok: false, + missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, + duplicate_events_count: 0, + curl_status: curlStatus, + }); + return { + ok: false, + missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, + duplicateEvents: [], + message: `Streaming probe error: ${detail}`, + }; } } diff --git a/src/lib/inference/probe-anthropic.test.ts b/src/lib/inference/probe-anthropic.test.ts index a672312d0b7..20ca5b45b21 100644 --- a/src/lib/inference/probe-anthropic.test.ts +++ b/src/lib/inference/probe-anthropic.test.ts @@ -77,6 +77,119 @@ describe("probeAnthropicEndpoint", () => { expect(result.message).toContain("HTTP 401"); }); + it("does not run the streaming probe unless probeStreaming is requested", () => { + vi.spyOn(probe, "runCurlProbe").mockReturnValue({ + ok: true, + httpStatus: 200, + curlStatus: 0, + body: "{}", + stderr: "", + message: "HTTP 200", + }); + const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe"); + + const result = probeAnthropicEndpoint( + "https://api.anthropic.com", + "claude-test", + "sk-ant-secret", + ); + + expect(result.ok).toBe(true); + expect(streamSpy).not.toHaveBeenCalled(); + }); + + it("validates the streaming event sequence when probeStreaming is set (#6289)", () => { + vi.spyOn(probe, "runCurlProbe").mockReturnValue({ + ok: true, + httpStatus: 200, + curlStatus: 0, + body: "{}", + stderr: "", + message: "HTTP 200", + }); + let streamingArgv: readonly string[] = []; + const streamSpy = vi + .spyOn(probe, "runAnthropicStreamingEventProbe") + .mockImplementation((argv) => { + streamingArgv = argv; + return { ok: true, missingEvents: [], duplicateEvents: [], message: "" }; + }); + + const result = probeAnthropicEndpoint( + "https://custom.endpoint.test", + "nvidia/nemotron-3-super-v3", + "sk-custom-secret", + { probeStreaming: true }, + ); + + expect(result).toEqual({ + ok: true, + api: "anthropic-messages", + label: "Anthropic Messages API", + }); + expect(streamSpy).toHaveBeenCalledOnce(); + expect(streamingArgv.at(-1)).toBe("https://custom.endpoint.test/v1/messages"); + expect(streamingArgv.join(" ")).toContain('"stream":true'); + expect(streamingArgv.join(" ")).not.toContain("sk-custom-secret"); + }); + + it("fails validation when the streaming event sequence is malformed", () => { + vi.spyOn(probe, "runCurlProbe").mockReturnValue({ + ok: true, + httpStatus: 200, + curlStatus: 0, + body: "{}", + stderr: "", + message: "HTTP 200", + }); + vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ + ok: false, + missingEvents: [], + duplicateEvents: ["message_start"], + message: + "Anthropic Messages streaming on this endpoint emits duplicate message_start " + + "(2 events for one request). Agent runs use the streaming path and would fail " + + "with an empty final response.", + }); + + const result = probeAnthropicEndpoint( + "https://custom.endpoint.test", + "nvidia/nemotron-3-super-v3", + "sk-custom-secret", + { probeStreaming: true }, + ); + + expect(result.ok).toBe(false); + expect(result.failures?.[0]).toMatchObject({ + name: "Anthropic Messages API (streaming)", + httpStatus: 0, + curlStatus: 0, + }); + expect(result.message).toContain("duplicate message_start"); + }); + + it("skips the streaming probe when the non-streaming probe already failed", () => { + vi.spyOn(probe, "runCurlProbe").mockReturnValue({ + ok: false, + httpStatus: 401, + curlStatus: 0, + body: "{}", + stderr: "", + message: "HTTP 401", + }); + const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe"); + + const result = probeAnthropicEndpoint( + "https://custom.endpoint.test", + "claude-test", + "sk-ant-bad", + { probeStreaming: true }, + ); + + expect(result.ok).toBe(false); + expect(streamSpy).not.toHaveBeenCalled(); + }); + it("converts an auth-config setup failure into the same structured probe-failure shape", () => { const spy = vi.spyOn(probe, "runCurlProbe"); // Force createXApiKeyAuthConfig to throw by stubbing the os.tmpdir lookup diff --git a/src/lib/inference/probe-anthropic.ts b/src/lib/inference/probe-anthropic.ts index 05f5a3ad30b..fb689956a60 100644 --- a/src/lib/inference/probe-anthropic.ts +++ b/src/lib/inference/probe-anthropic.ts @@ -7,7 +7,11 @@ // specific probes. import { createXApiKeyAuthConfig } from "../adapters/http/auth-config"; -import { getCurlTimingArgs, runCurlProbe } from "../adapters/http/probe"; +import { + getCurlTimingArgs, + runAnthropicStreamingEventProbe, + runCurlProbe, +} from "../adapters/http/probe"; import { normalizeCredentialValue } from "../credentials/store"; export interface AnthropicProbeFailureDetail { @@ -25,6 +29,26 @@ export interface AnthropicProbeResult { failures?: AnthropicProbeFailureDetail[]; } +export interface AnthropicProbeOptions { + /** + * Also validate the `/v1/messages` SSE event sequence with a + * `stream: true` request. Catches Anthropic-compatible gateways whose + * non-streaming responses are valid but whose streaming layer is malformed + * (duplicate `message_start` events, missing content deltas) — agent + * runtimes only use the streaming path, so the defect otherwise first + * surfaces in-sandbox as "no final response was produced" (#6289). + */ + probeStreaming?: boolean; +} + +// Streaming validation must not hang the onboarding wizard on an endpoint +// that keeps the SSE connection open: mirror the tighter per-validation +// timing used for /v1/responses streaming checks (issue #1601) instead of +// the 60s default in getCurlTimingArgs(). curl exit 28 (timeout) is +// tolerated by the streaming probe when the required events were already +// collected before the cap. +const STREAMING_PROBE_TIMING_ARGS = ["--connect-timeout", "10", "--max-time", "15"]; + function anthropicFailureFromError(error: unknown): AnthropicProbeResult { const message = error instanceof Error ? error.message : String(error); return { @@ -34,14 +58,25 @@ function anthropicFailureFromError(error: unknown): AnthropicProbeResult { }; } +function anthropicMessagesPayload(model: string, stream: boolean): string { + return JSON.stringify({ + model, + max_tokens: 16, + ...(stream ? { stream: true } : {}), + messages: [{ role: "user", content: "Reply with exactly: OK" }], + }); +} + export function probeAnthropicEndpoint( endpointUrl: string, model: string, apiKey: string, + options: AnthropicProbeOptions = {}, ): AnthropicProbeResult { let authConfig: ReturnType | undefined; try { authConfig = createXApiKeyAuthConfig(normalizeCredentialValue(apiKey)); + const messagesUrl = `${String(endpointUrl).replace(/\/+$/, "")}/v1/messages`; const result = runCurlProbe( [ "-sS", @@ -52,30 +87,59 @@ export function probeAnthropicEndpoint( "-H", "content-type: application/json", "-d", - JSON.stringify({ - model, - max_tokens: 16, - messages: [{ role: "user", content: "Reply with exactly: OK" }], - }), - `${String(endpointUrl).replace(/\/+$/, "")}/v1/messages`, + anthropicMessagesPayload(model, false), + messagesUrl, ], { trustedConfigFiles: authConfig.trustedConfigFiles }, ); - if (result.ok) { - return { ok: true, api: "anthropic-messages", label: "Anthropic Messages API" }; + if (!result.ok) { + return { + ok: false, + message: result.message, + failures: [ + { + name: "Anthropic Messages API", + httpStatus: result.httpStatus, + curlStatus: result.curlStatus, + message: result.message, + }, + ], + }; } - return { - ok: false, - message: result.message, - failures: [ - { - name: "Anthropic Messages API", - httpStatus: result.httpStatus, - curlStatus: result.curlStatus, - message: result.message, - }, - ], - }; + + if (options.probeStreaming === true) { + const streamResult = runAnthropicStreamingEventProbe( + [ + "-sS", + ...STREAMING_PROBE_TIMING_ARGS, + ...authConfig.args, + "-H", + "anthropic-version: 2023-06-01", + "-H", + "content-type: application/json", + "-d", + anthropicMessagesPayload(model, true), + messagesUrl, + ], + { trustedConfigFiles: authConfig.trustedConfigFiles }, + ); + if (!streamResult.ok) { + return { + ok: false, + message: `Anthropic Messages API (streaming): ${streamResult.message}`, + failures: [ + { + name: "Anthropic Messages API (streaming)", + httpStatus: 0, + curlStatus: 0, + message: streamResult.message, + }, + ], + }; + } + } + + return { ok: true, api: "anthropic-messages", label: "Anthropic Messages API" }; } catch (error) { return anthropicFailureFromError(error); } finally { diff --git a/src/lib/onboard/inference-selection-validation.test.ts b/src/lib/onboard/inference-selection-validation.test.ts index 46fcfe3deab..f8406e03ef9 100644 --- a/src/lib/onboard/inference-selection-validation.test.ts +++ b/src/lib/onboard/inference-selection-validation.test.ts @@ -86,4 +86,117 @@ describe("inference selection validation", () => { vi.unstubAllEnvs(); } }); + + it("requests streaming validation for custom Anthropic-compatible endpoints (#6289)", async () => { + const probeAnthropicEndpoint = vi.fn(() => ({ + ok: true, + api: "anthropic-messages", + label: "Anthropic Messages API", + })); + const log = vi.spyOn(console, "log").mockImplementation(() => {}); + const helpers = createInferenceSelectionValidationHelpers({ + isNonInteractive: () => false, + agentProductName: () => "Hermes", + getCredential: () => "test-key", + probeAnthropicEndpoint, + promptValidationRecovery: vi.fn(async () => "selection" as const), + }); + + try { + await expect( + helpers.validateCustomAnthropicSelection( + "Custom Anthropic endpoint", + "https://compatible.example", + "nvidia/nemotron-3-super-v3", + "COMPATIBLE_ANTHROPIC_API_KEY", + ), + ).resolves.toEqual({ ok: true, api: "anthropic-messages" }); + expect(probeAnthropicEndpoint).toHaveBeenCalledWith( + "https://compatible.example", + "nvidia/nemotron-3-super-v3", + "test-key", + { probeStreaming: true }, + ); + } finally { + log.mockRestore(); + } + }); + + it("skips Anthropic streaming validation in reasoning mode", async () => { + vi.stubEnv("NEMOCLAW_REASONING", "yes"); + const probeAnthropicEndpoint = vi.fn(() => ({ + ok: true, + api: "anthropic-messages", + label: "Anthropic Messages API", + })); + const log = vi.spyOn(console, "log").mockImplementation(() => {}); + const helpers = createInferenceSelectionValidationHelpers({ + isNonInteractive: () => false, + agentProductName: () => "Hermes", + getCredential: () => "test-key", + probeAnthropicEndpoint, + promptValidationRecovery: vi.fn(async () => "selection" as const), + }); + + try { + await helpers.validateCustomAnthropicSelection( + "Custom Anthropic endpoint", + "https://compatible.example", + "reasoning-model", + "COMPATIBLE_ANTHROPIC_API_KEY", + ); + expect(probeAnthropicEndpoint).toHaveBeenCalledWith( + "https://compatible.example", + "reasoning-model", + "test-key", + { probeStreaming: false }, + ); + } finally { + log.mockRestore(); + vi.unstubAllEnvs(); + } + }); + + it("routes a malformed-streaming probe failure through validation recovery (#6289)", async () => { + const probeAnthropicEndpoint = vi.fn(() => ({ + ok: false, + message: + "Anthropic Messages API (streaming): Anthropic Messages streaming on this endpoint " + + "emits duplicate message_start (2 events for one request).", + failures: [ + { + name: "Anthropic Messages API (streaming)", + httpStatus: 0, + curlStatus: 0, + message: "duplicate message_start", + }, + ], + })); + const promptValidationRecovery = vi.fn(async () => "model" as const); + const error = vi.spyOn(console, "error").mockImplementation(() => {}); + const helpers = createInferenceSelectionValidationHelpers({ + isNonInteractive: () => false, + agentProductName: () => "Hermes", + getCredential: () => "test-key", + probeAnthropicEndpoint, + promptValidationRecovery, + }); + + try { + await expect( + helpers.validateCustomAnthropicSelection( + "Custom Anthropic endpoint", + "https://compatible.example", + "nvidia/nemotron-3-super-v3", + "COMPATIBLE_ANTHROPIC_API_KEY", + ), + ).resolves.toEqual({ ok: false, retry: "model" }); + expect(promptValidationRecovery).toHaveBeenCalledOnce(); + expect(error.mock.calls.map((args) => args.join(" ")).join("\n")).toContain( + "Custom Anthropic endpoint endpoint validation failed.", + ); + } finally { + error.mockRestore(); + } + }); }); diff --git a/src/lib/onboard/inference-selection-validation.ts b/src/lib/onboard/inference-selection-validation.ts index b65f4d2a90d..9a226e85de4 100644 --- a/src/lib/onboard/inference-selection-validation.ts +++ b/src/lib/onboard/inference-selection-validation.ts @@ -9,6 +9,7 @@ const { probeAnthropicEndpoint, probeOpenAiLikeEndpoint } = endpointUrl: string, model: string, apiKey: string | null | undefined, + options?: { probeStreaming?: boolean }, ): any; probeOpenAiLikeEndpoint( endpointUrl: string, @@ -229,7 +230,16 @@ export function createInferenceSelectionValidationHelpers( helpUrl: string | null = null, ): Promise { const apiKey = resolveCredential(credentialEnv); - const probe = runAnthropicProbe(endpointUrl, model, apiKey); + const reasoningEnabled = normalizeReasoningFlag(process.env.NEMOCLAW_REASONING) === "true"; + // Streaming validation catches Anthropic-compatible gateways whose + // non-streaming responses are valid but whose SSE streams are malformed + // (duplicate message_start, missing content deltas) — the agent runtime + // only uses the streaming path, so onboarding must exercise it (#6289). + // Reasoning-only compatible endpoints often reject streaming probes, so + // mirror the custom OpenAI-compatible path and skip streaming for them. + const probe = runAnthropicProbe(endpointUrl, model, apiKey, { + probeStreaming: !reasoningEnabled, + }); if (probe.ok) { console.log(` ${probe.label} available — ${deps.agentProductName()} will use ${probe.api}.`); return { ok: true, api: probe.api }; diff --git a/test/onboard-selection-anthropic-retry.test.ts b/test/onboard-selection-anthropic-retry.test.ts new file mode 100644 index 00000000000..926492c2768 --- /dev/null +++ b/test/onboard-selection-anthropic-retry.test.ts @@ -0,0 +1,246 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { describe, it } from "vitest"; + +import { testTimeout } from "./helpers/timeouts"; + +const CREDENTIAL_RETRY_PROMPT_RE = + /Options: retry \(re-enter key\), back \(change provider\), exit \[retry\]: /; + +const PROVIDER_SELECTION_TEST_TIMEOUT_MS = testTimeout(60_000); + +function writeAnthropicStyleAuthRetryCurl( + fakeBin: string, + goodToken: string, + models = ["claude-sonnet-4-6"], +) { + fs.writeFileSync( + path.join(fakeBin, "curl"), + `#!/usr/bin/env bash +body='{"error":{"message":"forbidden"}}' +status="403" +outfile="" +auth="" +url="" +data="" +while [ "$#" -gt 0 ]; do + case "$1" in + -o) outfile="$2"; shift 2 ;; + -d) data="$2"; shift 2 ;; + -H) + if echo "$2" | grep -q '^x-api-key: '; then + auth="$2" + fi + shift 2 + ;; + --config) auth="$(cat "$2" 2>/dev/null)"; shift 2 ;; *) url="$1"; shift ;; + esac +done +if echo "$url" | grep -q '/v1/models$'; then + body='{"data":[${models.map((model) => `{"id":"${model}"}`).join(",")}]}' + status="200" +elif echo "$auth" | grep -q '${goodToken}' && echo "$url" | grep -q '/v1/messages$'; then + if echo "$data" | grep -q '"stream":true'; then + # Streaming validation probe: serve a well-formed Anthropic SSE sequence. + body='event: message_start +data: {"type":"message_start","message":{"id":"msg_123"}} + +event: content_block_delta +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"OK"}} + +event: message_stop +data: {"type":"message_stop"} +' + else + body='{"id":"msg_123","content":[{"type":"text","text":"OK"}]}' + fi + status="200" +fi +printf '%s' "$body" > "$outfile" +printf '%s' "$status" +`, + { mode: 0o755 }, + ); +} + +describe("onboard Anthropic credential retry UX", { + timeout: PROVIDER_SELECTION_TEST_TIMEOUT_MS, +}, () => { + it("lets users re-enter an Anthropic API key after authorization failure", () => { + const repoRoot = path.join(import.meta.dirname, ".."); + const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-onboard-anthropic-auth-retry-")); + const fakeBin = path.join(tmpDir, "bin"); + const scriptPath = path.join(tmpDir, "anthropic-auth-retry-check.js"); + const onboardPath = JSON.stringify(path.join(repoRoot, "src", "lib", "onboard.ts")); + const credentialsPath = JSON.stringify( + path.join(repoRoot, "src", "lib", "credentials", "store.ts"), + ); + const runnerPath = JSON.stringify(path.join(repoRoot, "src", "lib", "runner.ts")); + + fs.mkdirSync(fakeBin, { recursive: true }); + writeAnthropicStyleAuthRetryCurl(fakeBin, "anthropic-good", ["claude-sonnet-4-6"]); + + const script = String.raw` +const credentials = require(${credentialsPath}); +const runner = require(${runnerPath}); + +const answers = ["4", "", "retry", "anthropic-good", ""]; +const messages = []; + +credentials.prompt = async (message) => { + messages.push(message); + return answers.shift() || ""; +}; +runner.runCapture = () => ""; + +const { setupNim } = require(${onboardPath}); + +(async () => { + process.env.ANTHROPIC_API_KEY = "anthropic-bad"; + const originalLog = console.log; + const originalError = console.error; + const lines = []; + console.log = (...args) => lines.push(args.join(" ")); + console.error = (...args) => lines.push(args.join(" ")); + try { + const result = await setupNim(null); + originalLog(JSON.stringify({ result, messages, lines, key: process.env.ANTHROPIC_API_KEY })); + } finally { + console.log = originalLog; + console.error = originalError; + } +})().catch((error) => { + console.error(error); + process.exit(1); +}); +`; + fs.writeFileSync(scriptPath, script); + + const result = spawnSync(process.execPath, [scriptPath], { + cwd: repoRoot, + encoding: "utf-8", + env: { + ...process.env, + HOME: tmpDir, + PATH: `${fakeBin}:${process.env.PATH || ""}`, + }, + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout.trim()); + assert.equal(payload.result.provider, "anthropic-prod"); + assert.equal(payload.result.model, "claude-sonnet-4-6"); + assert.equal(payload.result.preferredInferenceApi, "anthropic-messages"); + assert.equal(payload.key, "anthropic-good"); + assert.ok( + payload.lines.some((line: string) => line.includes("Anthropic authorization failed")), + ); + assert.ok(payload.messages.some((message: string) => CREDENTIAL_RETRY_PROMPT_RE.test(message))); + assert.ok(payload.messages.some((message: string) => /Anthropic API key: /.test(message))); + assert.equal(payload.messages.filter((message: string) => /Choose \[/.test(message)).length, 1); + assert.equal( + payload.messages.filter((message: string) => /Choose model \[1\]/.test(message)).length, + 2, + ); + }); + + it("lets users re-enter a custom Anthropic-compatible API key without re-entering the endpoint URL", () => { + const repoRoot = path.join(import.meta.dirname, ".."); + const tmpDir = fs.mkdtempSync( + path.join(os.tmpdir(), "nemoclaw-onboard-custom-anthropic-auth-retry-"), + ); + const fakeBin = path.join(tmpDir, "bin"); + const scriptPath = path.join(tmpDir, "custom-anthropic-auth-retry-check.js"); + const onboardPath = JSON.stringify(path.join(repoRoot, "src", "lib", "onboard.ts")); + const credentialsPath = JSON.stringify( + path.join(repoRoot, "src", "lib", "credentials", "store.ts"), + ); + const runnerPath = JSON.stringify(path.join(repoRoot, "src", "lib", "runner.ts")); + + fs.mkdirSync(fakeBin, { recursive: true }); + writeAnthropicStyleAuthRetryCurl(fakeBin, "anthropic-proxy-good", ["claude-proxy"]); + + const script = String.raw` +const credentials = require(${credentialsPath}); +const runner = require(${runnerPath}); + +const answers = ["5", "https://proxy.example.com/v1/messages?token=secret#frag", "claude-proxy", "retry", "anthropic-proxy-good", "claude-proxy"]; +const messages = []; + +credentials.prompt = async (message) => { + messages.push(message); + return answers.shift() || ""; +}; +runner.runCapture = () => ""; + +const { setupNim } = require(${onboardPath}); + +(async () => { + process.env.COMPATIBLE_ANTHROPIC_API_KEY = "anthropic-proxy-bad"; + const originalLog = console.log; + const originalError = console.error; + const lines = []; + console.log = (...args) => lines.push(args.join(" ")); + console.error = (...args) => lines.push(args.join(" ")); + try { + const result = await setupNim(null); + originalLog(JSON.stringify({ result, messages, lines, key: process.env.COMPATIBLE_ANTHROPIC_API_KEY })); + } finally { + console.log = originalLog; + console.error = originalError; + } +})().catch((error) => { + console.error(error); + process.exit(1); +}); +`; + fs.writeFileSync(scriptPath, script); + + const result = spawnSync(process.execPath, [scriptPath], { + cwd: repoRoot, + encoding: "utf-8", + env: { + ...process.env, + HOME: tmpDir, + PATH: `${fakeBin}:${process.env.PATH || ""}`, + }, + }); + + assert.equal(result.status, 0, result.stderr); + const payload = JSON.parse(result.stdout.trim()); + assert.equal(payload.result.provider, "compatible-anthropic-endpoint"); + assert.equal(payload.result.model, "claude-proxy"); + assert.equal(payload.result.endpointUrl, "https://proxy.example.com"); + assert.equal(payload.result.preferredInferenceApi, "anthropic-messages"); + assert.equal(payload.key, "anthropic-proxy-good"); + assert.ok( + payload.lines.some((line: string) => + line.includes("Other Anthropic-compatible endpoint authorization failed"), + ), + ); + assert.ok(payload.messages.some((message: string) => CREDENTIAL_RETRY_PROMPT_RE.test(message))); + assert.ok( + payload.messages.some((message: string) => + /Other Anthropic-compatible endpoint API key: /.test(message), + ), + ); + assert.equal( + payload.messages.filter((message: string) => /Anthropic-compatible base URL/.test(message)) + .length, + 1, + ); + assert.equal( + payload.messages.filter((message: string) => + /Other Anthropic-compatible endpoint model/.test(message), + ).length, + 2, + ); + assert.equal(payload.messages.filter((message: string) => /Choose \[/.test(message)).length, 1); + }); +}); diff --git a/test/onboard-selection.test.ts b/test/onboard-selection.test.ts index 7a9bdca0c7e..a6b04c0ac8e 100644 --- a/test/onboard-selection.test.ts +++ b/test/onboard-selection.test.ts @@ -314,45 +314,6 @@ printf '%s' "$status" ); } -function writeAnthropicStyleAuthRetryCurl( - fakeBin: string, - goodToken: string, - models = ["claude-sonnet-4-6"], -) { - fs.writeFileSync( - path.join(fakeBin, "curl"), - `#!/usr/bin/env bash -body='{"error":{"message":"forbidden"}}' -status="403" -outfile="" -auth="" -url="" -while [ "$#" -gt 0 ]; do - case "$1" in - -o) outfile="$2"; shift 2 ;; - -H) - if echo "$2" | grep -q '^x-api-key: '; then - auth="$2" - fi - shift 2 - ;; - --config) auth="$(cat "$2" 2>/dev/null)"; shift 2 ;; *) url="$1"; shift ;; - esac -done -if echo "$url" | grep -q '/v1/models$'; then - body='{"data":[${models.map((model) => `{"id":"${model}"}`).join(",")}]}' - status="200" -elif echo "$auth" | grep -q '${goodToken}' && echo "$url" | grep -q '/v1/messages$'; then - body='{"id":"msg_123","content":[{"type":"text","text":"OK"}]}' - status="200" -fi -printf '%s' "$body" > "$outfile" -printf '%s' "$status" -`, - { mode: 0o755 }, - ); -} - type CredentialBackScenario = { name: string; answers: string[]; @@ -4270,84 +4231,6 @@ const { setupNim } = require(${onboardPath}); ); }); - it("lets users re-enter an Anthropic API key after authorization failure", () => { - const repoRoot = path.join(import.meta.dirname, ".."); - const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-onboard-anthropic-auth-retry-")); - const fakeBin = path.join(tmpDir, "bin"); - const scriptPath = path.join(tmpDir, "anthropic-auth-retry-check.js"); - const onboardPath = JSON.stringify(path.join(repoRoot, "src", "lib", "onboard.ts")); - const credentialsPath = JSON.stringify( - path.join(repoRoot, "src", "lib", "credentials", "store.ts"), - ); - const runnerPath = JSON.stringify(path.join(repoRoot, "src", "lib", "runner.ts")); - - fs.mkdirSync(fakeBin, { recursive: true }); - writeAnthropicStyleAuthRetryCurl(fakeBin, "anthropic-good", ["claude-sonnet-4-6"]); - - const script = String.raw` -const credentials = require(${credentialsPath}); -const runner = require(${runnerPath}); - -const answers = ["4", "", "retry", "anthropic-good", ""]; -const messages = []; - -credentials.prompt = async (message) => { - messages.push(message); - return answers.shift() || ""; -}; -runner.runCapture = () => ""; - -const { setupNim } = require(${onboardPath}); - -(async () => { - process.env.ANTHROPIC_API_KEY = "anthropic-bad"; - const originalLog = console.log; - const originalError = console.error; - const lines = []; - console.log = (...args) => lines.push(args.join(" ")); - console.error = (...args) => lines.push(args.join(" ")); - try { - const result = await setupNim(null); - originalLog(JSON.stringify({ result, messages, lines, key: process.env.ANTHROPIC_API_KEY })); - } finally { - console.log = originalLog; - console.error = originalError; - } -})().catch((error) => { - console.error(error); - process.exit(1); -}); -`; - fs.writeFileSync(scriptPath, script); - - const result = spawnSync(process.execPath, [scriptPath], { - cwd: repoRoot, - encoding: "utf-8", - env: { - ...process.env, - HOME: tmpDir, - PATH: `${fakeBin}:${process.env.PATH || ""}`, - }, - }); - - assert.equal(result.status, 0, result.stderr); - const payload = JSON.parse(result.stdout.trim()); - assert.equal(payload.result.provider, "anthropic-prod"); - assert.equal(payload.result.model, "claude-sonnet-4-6"); - assert.equal(payload.result.preferredInferenceApi, "anthropic-messages"); - assert.equal(payload.key, "anthropic-good"); - assert.ok( - payload.lines.some((line: string) => line.includes("Anthropic authorization failed")), - ); - assert.ok(payload.messages.some((message: string) => CREDENTIAL_RETRY_PROMPT_RE.test(message))); - assert.ok(payload.messages.some((message: string) => /Anthropic API key: /.test(message))); - assert.equal(payload.messages.filter((message: string) => /Choose \[/.test(message)).length, 1); - assert.equal( - payload.messages.filter((message: string) => /Choose model \[1\]/.test(message)).length, - 2, - ); - }); - it("lets users re-enter a Gemini API key after authorization failure", () => { const repoRoot = path.join(import.meta.dirname, ".."); const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-onboard-gemini-auth-retry-")); @@ -4520,100 +4403,6 @@ const { setupNim } = require(${onboardPath}); assert.equal(payload.messages.filter((message: string) => /Choose \[/.test(message)).length, 1); }); - it("lets users re-enter a custom Anthropic-compatible API key without re-entering the endpoint URL", () => { - const repoRoot = path.join(import.meta.dirname, ".."); - const tmpDir = fs.mkdtempSync( - path.join(os.tmpdir(), "nemoclaw-onboard-custom-anthropic-auth-retry-"), - ); - const fakeBin = path.join(tmpDir, "bin"); - const scriptPath = path.join(tmpDir, "custom-anthropic-auth-retry-check.js"); - const onboardPath = JSON.stringify(path.join(repoRoot, "src", "lib", "onboard.ts")); - const credentialsPath = JSON.stringify( - path.join(repoRoot, "src", "lib", "credentials", "store.ts"), - ); - const runnerPath = JSON.stringify(path.join(repoRoot, "src", "lib", "runner.ts")); - - fs.mkdirSync(fakeBin, { recursive: true }); - writeAnthropicStyleAuthRetryCurl(fakeBin, "anthropic-proxy-good", ["claude-proxy"]); - - const script = String.raw` -const credentials = require(${credentialsPath}); -const runner = require(${runnerPath}); - -const answers = ["5", "https://proxy.example.com/v1/messages?token=secret#frag", "claude-proxy", "retry", "anthropic-proxy-good", "claude-proxy"]; -const messages = []; - -credentials.prompt = async (message) => { - messages.push(message); - return answers.shift() || ""; -}; -runner.runCapture = () => ""; - -const { setupNim } = require(${onboardPath}); - -(async () => { - process.env.COMPATIBLE_ANTHROPIC_API_KEY = "anthropic-proxy-bad"; - const originalLog = console.log; - const originalError = console.error; - const lines = []; - console.log = (...args) => lines.push(args.join(" ")); - console.error = (...args) => lines.push(args.join(" ")); - try { - const result = await setupNim(null); - originalLog(JSON.stringify({ result, messages, lines, key: process.env.COMPATIBLE_ANTHROPIC_API_KEY })); - } finally { - console.log = originalLog; - console.error = originalError; - } -})().catch((error) => { - console.error(error); - process.exit(1); -}); -`; - fs.writeFileSync(scriptPath, script); - - const result = spawnSync(process.execPath, [scriptPath], { - cwd: repoRoot, - encoding: "utf-8", - env: { - ...process.env, - HOME: tmpDir, - PATH: `${fakeBin}:${process.env.PATH || ""}`, - }, - }); - - assert.equal(result.status, 0, result.stderr); - const payload = JSON.parse(result.stdout.trim()); - assert.equal(payload.result.provider, "compatible-anthropic-endpoint"); - assert.equal(payload.result.model, "claude-proxy"); - assert.equal(payload.result.endpointUrl, "https://proxy.example.com"); - assert.equal(payload.result.preferredInferenceApi, "anthropic-messages"); - assert.equal(payload.key, "anthropic-proxy-good"); - assert.ok( - payload.lines.some((line: string) => - line.includes("Other Anthropic-compatible endpoint authorization failed"), - ), - ); - assert.ok(payload.messages.some((message: string) => CREDENTIAL_RETRY_PROMPT_RE.test(message))); - assert.ok( - payload.messages.some((message: string) => - /Other Anthropic-compatible endpoint API key: /.test(message), - ), - ); - assert.equal( - payload.messages.filter((message: string) => /Anthropic-compatible base URL/.test(message)) - .length, - 1, - ); - assert.equal( - payload.messages.filter((message: string) => - /Other Anthropic-compatible endpoint model/.test(message), - ).length, - 2, - ); - assert.equal(payload.messages.filter((message: string) => /Choose \[/.test(message)).length, 1); - }); - it("forces openai-completions for vLLM even when probe detects openai-responses", () => { const repoRoot = path.join(import.meta.dirname, ".."); const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-onboard-vllm-override-")); From a25a10919f32badee2edd133ce30f58d1d04afb7 Mon Sep 17 00:00:00 2001 From: Tony Luo Date: Mon, 6 Jul 2026 14:24:27 +0800 Subject: [PATCH 02/12] test(inference): reuse writeCurlOutputBody in anthropic streaming probe mocks The duplicated mockStreaming helper and inline -o writer added four if statements to probe.test.ts, tripping the changed-test-file if-count guardrail. Route both through the existing writeCurlOutputBody helper, which exists precisely to keep that budget steady. Signed-off-by: Tony Luo --- src/lib/adapters/http/probe.test.ts | 18 +++--------------- 1 file changed, 3 insertions(+), 15 deletions(-) diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index e9ef219c321..e4f3293f5d3 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -806,13 +806,7 @@ describe("runAnthropicStreamingEventProbe", () => { /** Helper to build a spawnSyncImpl that writes SSE content to the -o file. */ function mockStreaming(sseBody: string, exitCode = 0) { return (_command: string, args: readonly string[]) => { - const oIdx = args.indexOf("-o"); - if (oIdx !== -1) { - const outputPath = args[oIdx + 1]; - if (typeof outputPath === "string") { - fs.writeFileSync(outputPath, sseBody); - } - } + writeCurlOutputBody(args, sseBody); return { pid: 1, output: [], @@ -994,14 +988,8 @@ describe("runAnthropicStreamingEventProbe", () => { ["-sS", "--max-time", "15", "https://example.test/v1/messages"], { spawnSyncImpl: (_command, args) => { - const oIdx = args.indexOf("-o"); - if (oIdx !== -1) { - const nextArg = args[oIdx + 1]; - if (typeof nextArg === "string") { - outputPath = nextArg; - fs.writeFileSync(outputPath, healthyStream); - } - } + outputPath = String(args[args.indexOf("-o") + 1]); + writeCurlOutputBody(args, healthyStream); return { pid: 1, output: [], From bbe74000bd3bc1f4e669b094f279493939ef01b0 Mon Sep 17 00:00:00 2001 From: Tony Luo Date: Mon, 6 Jul 2026 15:02:53 +0800 Subject: [PATCH 03/12] fix(inference): enforce event order in Anthropic streaming validation Address CodeRabbit review on #6297: - Validate the Anthropic stream as a sequence, not just counts: keep the ordered event list from the SSE capture, require message_start before any content_block_delta and message_stop after the last one, and add message_stop to the singleton set so duplicate terminal events fail too. Interleaved unknown events (ping) stay tolerated. - Stub the negative-call streaming spies in probe-anthropic tests so a regression can never invoke the real curl-backed probe. - Add the (#6289) issue-ref suffix to the moved integration test titles. Signed-off-by: Tony Luo --- src/lib/adapters/http/probe.test.ts | 98 +++++++++++++++++++ src/lib/adapters/http/probe.ts | 53 ++++++++-- src/lib/inference/probe-anthropic.test.ts | 25 ++++- .../onboard-selection-anthropic-retry.test.ts | 4 +- 4 files changed, 169 insertions(+), 11 deletions(-) diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index e4f3293f5d3..5493aeff27a 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -848,6 +848,104 @@ describe("runAnthropicStreamingEventProbe", () => { expect(result.ok).toBe(true); expect(result.missingEvents).toEqual([]); expect(result.duplicateEvents).toEqual([]); + expect(result.sequenceErrors).toEqual([]); + }); + + it("fails when message_stop is emitted twice for one request", () => { + const duplicatedStop = [ + "event: message_start", + "data: {}", + "", + "event: content_block_delta", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(duplicatedStop) }, + ); + + expect(result.ok).toBe(false); + expect(result.duplicateEvents).toEqual(["message_stop"]); + }); + + it("fails when content deltas arrive before message_start", () => { + const startAfterDelta = [ + "event: content_block_delta", + "data: {}", + "", + "event: message_start", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(startAfterDelta) }, + ); + + expect(result.ok).toBe(false); + expect(result.sequenceErrors).toEqual(["content_block_delta before message_start"]); + expect(result.message).toContain("out of order"); + }); + + it("fails when content deltas continue after message_stop", () => { + const deltaAfterStop = [ + "event: message_start", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + "event: content_block_delta", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(deltaAfterStop) }, + ); + + expect(result.ok).toBe(false); + expect(result.sequenceErrors).toEqual(["content_block_delta after message_stop"]); + }); + + it("tolerates interleaved unknown events like ping in a well-formed stream", () => { + const withPing = [ + "event: message_start", + "data: {}", + "", + "event: ping", + "data: {}", + "", + "event: content_block_delta", + "data: {}", + "", + "event: ping", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(withPing) }, + ); + + expect(result.ok).toBe(true); }); it("fails when message_start is emitted twice for one request (#6289)", () => { diff --git a/src/lib/adapters/http/probe.ts b/src/lib/adapters/http/probe.ts index c7f5a82cb51..5cb27b8c6a9 100644 --- a/src/lib/adapters/http/probe.ts +++ b/src/lib/adapters/http/probe.ts @@ -452,6 +452,8 @@ interface SseEventCaptureResult { detail: string; /** Occurrence count per SSE `event:` type parsed from the response body. */ eventCounts: Map; + /** SSE `event:` types in stream order, for sequence validation. */ + eventSequence: string[]; } /** @@ -493,20 +495,22 @@ function captureSseEventCounts( const detail = result.error ? String(result.error.message || result.error) : String(result.stderr || ""); - return { ok: false, curlStatus, detail, eventCounts: new Map() }; + return { ok: false, curlStatus, detail, eventCounts: new Map(), eventSequence: [] }; } // Parse SSE event types from the raw output. // Each event line looks like: "event: response.output_text.delta" const eventCounts = new Map(); + const eventSequence: string[] = []; for (const line of body.split("\n")) { const match = /^event:\s*(.+)$/i.exec(line.trim()); if (match) { const eventType = match[1].trim(); eventCounts.set(eventType, (eventCounts.get(eventType) ?? 0) + 1); + eventSequence.push(eventType); } } - return { ok: true, curlStatus: result.status ?? 0, detail: "", eventCounts }; + return { ok: true, curlStatus: result.status ?? 0, detail: "", eventCounts, eventSequence }; } finally { cleanupTempDir(bodyFile, tempPrefix); } @@ -591,20 +595,45 @@ const REQUIRED_ANTHROPIC_STREAMING_EVENTS = [ * observed emitting `message_start` twice with the same message id, which * corrupts streaming-client state machines: the agent run then ends with an * empty final response even though the non-streaming path works (#6289). + * `message_stop` is the single terminal event of the same contract. */ -const SINGLETON_ANTHROPIC_STREAMING_EVENTS = ["message_start"]; +const SINGLETON_ANTHROPIC_STREAMING_EVENTS = ["message_start", "message_stop"]; export interface AnthropicStreamingProbeResult { ok: boolean; missingEvents: string[]; duplicateEvents: string[]; + /** Order violations, e.g. content deltas before message_start or after message_stop. */ + sequenceErrors: string[]; message: string; } +/** + * Order rules for a well-formed Anthropic Messages stream: `message_start` + * opens the stream before any content delta, and `message_stop` terminates + * it after the last content delta. Only evaluated once all required events + * are present; interleaved unknown events (e.g. `ping`) are ignored. + */ +function anthropicSequenceErrors(eventSequence: string[]): string[] { + const errors: string[] = []; + const firstStart = eventSequence.indexOf("message_start"); + const firstDelta = eventSequence.indexOf("content_block_delta"); + const lastDelta = eventSequence.lastIndexOf("content_block_delta"); + const lastStop = eventSequence.lastIndexOf("message_stop"); + if (firstDelta < firstStart) { + errors.push("content_block_delta before message_start"); + } + if (lastStop < lastDelta) { + errors.push("content_block_delta after message_stop"); + } + return errors; +} + /** * Send a streaming request to an Anthropic-compatible `/v1/messages` * endpoint and verify the SSE event stream is well formed: the required - * event sequence is present and no singleton event is duplicated. + * events are present, no singleton event is duplicated, and the events + * arrive in protocol order (message_start → content deltas → message_stop). * * This catches gateways whose non-streaming responses are valid but whose * streaming layer is broken — runtime agents only use the streaming path, @@ -633,12 +662,14 @@ function runAnthropicStreamingEventProbeImpl( ok: false, missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, duplicate_events_count: 0, + sequence_errors_count: 0, curl_status: capture.curlStatus, }); return { ok: false, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], + sequenceErrors: [], message: `Streaming probe failed: ${compactText(capture.detail).slice(0, 200)}`, }; } @@ -649,7 +680,9 @@ function runAnthropicStreamingEventProbeImpl( const duplicates = SINGLETON_ANTHROPIC_STREAMING_EVENTS.filter( (e) => (capture.eventCounts.get(e) ?? 0) > 1, ); - if (missing.length > 0 || duplicates.length > 0) { + const sequenceErrors = + missing.length === 0 ? anthropicSequenceErrors(capture.eventSequence) : []; + if (missing.length > 0 || duplicates.length > 0 || sequenceErrors.length > 0) { const problems: string[] = []; if (duplicates.length > 0) { const detail = duplicates @@ -660,16 +693,21 @@ function runAnthropicStreamingEventProbeImpl( if (missing.length > 0) { problems.push(`is missing required events: ${missing.join(", ")}`); } + if (sequenceErrors.length > 0) { + problems.push(`emits events out of order (${sequenceErrors.join("; ")})`); + } emitCurlResultTraceEvent({ ok: false, missing_events_count: missing.length, duplicate_events_count: duplicates.length, + sequence_errors_count: sequenceErrors.length, curl_status: capture.curlStatus, }); return { ok: false, missingEvents: missing, duplicateEvents: duplicates, + sequenceErrors, message: `Anthropic Messages streaming on this endpoint ${problems.join(" and ")}. ` + "Agent runs use the streaming path and would fail with an empty final response.", @@ -680,9 +718,10 @@ function runAnthropicStreamingEventProbeImpl( ok: true, missing_events_count: 0, duplicate_events_count: 0, + sequence_errors_count: 0, curl_status: capture.curlStatus, }); - return { ok: true, missingEvents: [], duplicateEvents: [], message: "" }; + return { ok: true, missingEvents: [], duplicateEvents: [], sequenceErrors: [], message: "" }; } catch (error) { const detail = error instanceof Error ? error.message : String(error); const curlStatus = @@ -691,12 +730,14 @@ function runAnthropicStreamingEventProbeImpl( ok: false, missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, duplicate_events_count: 0, + sequence_errors_count: 0, curl_status: curlStatus, }); return { ok: false, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], + sequenceErrors: [], message: `Streaming probe error: ${detail}`, }; } diff --git a/src/lib/inference/probe-anthropic.test.ts b/src/lib/inference/probe-anthropic.test.ts index 20ca5b45b21..f0d0a0022d8 100644 --- a/src/lib/inference/probe-anthropic.test.ts +++ b/src/lib/inference/probe-anthropic.test.ts @@ -86,7 +86,13 @@ describe("probeAnthropicEndpoint", () => { stderr: "", message: "HTTP 200", }); - const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe"); + const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ + ok: true, + missingEvents: [], + duplicateEvents: [], + sequenceErrors: [], + message: "", + }); const result = probeAnthropicEndpoint( "https://api.anthropic.com", @@ -112,7 +118,13 @@ describe("probeAnthropicEndpoint", () => { .spyOn(probe, "runAnthropicStreamingEventProbe") .mockImplementation((argv) => { streamingArgv = argv; - return { ok: true, missingEvents: [], duplicateEvents: [], message: "" }; + return { + ok: true, + missingEvents: [], + duplicateEvents: [], + sequenceErrors: [], + message: "", + }; }); const result = probeAnthropicEndpoint( @@ -146,6 +158,7 @@ describe("probeAnthropicEndpoint", () => { ok: false, missingEvents: [], duplicateEvents: ["message_start"], + sequenceErrors: [], message: "Anthropic Messages streaming on this endpoint emits duplicate message_start " + "(2 events for one request). Agent runs use the streaming path and would fail " + @@ -177,7 +190,13 @@ describe("probeAnthropicEndpoint", () => { stderr: "", message: "HTTP 401", }); - const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe"); + const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ + ok: true, + missingEvents: [], + duplicateEvents: [], + sequenceErrors: [], + message: "", + }); const result = probeAnthropicEndpoint( "https://custom.endpoint.test", diff --git a/test/onboard-selection-anthropic-retry.test.ts b/test/onboard-selection-anthropic-retry.test.ts index 926492c2768..cd8fa4ed031 100644 --- a/test/onboard-selection-anthropic-retry.test.ts +++ b/test/onboard-selection-anthropic-retry.test.ts @@ -72,7 +72,7 @@ printf '%s' "$status" describe("onboard Anthropic credential retry UX", { timeout: PROVIDER_SELECTION_TEST_TIMEOUT_MS, }, () => { - it("lets users re-enter an Anthropic API key after authorization failure", () => { + it("lets users re-enter an Anthropic API key after authorization failure (#6289)", () => { const repoRoot = path.join(import.meta.dirname, ".."); const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), "nemoclaw-onboard-anthropic-auth-retry-")); const fakeBin = path.join(tmpDir, "bin"); @@ -150,7 +150,7 @@ const { setupNim } = require(${onboardPath}); ); }); - it("lets users re-enter a custom Anthropic-compatible API key without re-entering the endpoint URL", () => { + it("lets users re-enter a custom Anthropic-compatible API key without re-entering the endpoint URL (#6289)", () => { const repoRoot = path.join(import.meta.dirname, ".."); const tmpDir = fs.mkdtempSync( path.join(os.tmpdir(), "nemoclaw-onboard-custom-anthropic-auth-retry-"), From e15819126afd51ff857f54ba0e2976a1e1d9c7ed Mon Sep 17 00:00:00 2001 From: Tony Luo Date: Mon, 6 Jul 2026 15:15:04 +0800 Subject: [PATCH 04/12] fix(inference): treat all Anthropic content events in stream order checks Address CodeRabbit follow-up on #6297: message_stop terminality was only checked against content_block_delta, so a stream ending message_stop -> content_block_stop still passed. Compare against the full known payload-event set (content_block_start/delta/stop, message_delta) while continuing to ignore unknown interleavings like ping. Signed-off-by: Tony Luo --- src/lib/adapters/http/probe.test.ts | 29 ++++++++++++++++++++++++-- src/lib/adapters/http/probe.ts | 32 +++++++++++++++++++++-------- 2 files changed, 50 insertions(+), 11 deletions(-) diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index 5493aeff27a..969e1763556 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -895,7 +895,7 @@ describe("runAnthropicStreamingEventProbe", () => { ); expect(result.ok).toBe(false); - expect(result.sequenceErrors).toEqual(["content_block_delta before message_start"]); + expect(result.sequenceErrors).toEqual(["content events before message_start"]); expect(result.message).toContain("out of order"); }); @@ -918,7 +918,32 @@ describe("runAnthropicStreamingEventProbe", () => { ); expect(result.ok).toBe(false); - expect(result.sequenceErrors).toEqual(["content_block_delta after message_stop"]); + expect(result.sequenceErrors).toEqual(["content events after message_stop"]); + }); + + it("fails when non-delta content events trail message_stop", () => { + const stopNotTerminal = [ + "event: message_start", + "data: {}", + "", + "event: content_block_delta", + "data: {}", + "", + "event: message_stop", + "data: {}", + "", + "event: content_block_stop", + "data: {}", + "", + ].join("\n"); + + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(stopNotTerminal) }, + ); + + expect(result.ok).toBe(false); + expect(result.sequenceErrors).toEqual(["content events after message_stop"]); }); it("tolerates interleaved unknown events like ping in a well-formed stream", () => { diff --git a/src/lib/adapters/http/probe.ts b/src/lib/adapters/http/probe.ts index 5cb27b8c6a9..bbd1e78626a 100644 --- a/src/lib/adapters/http/probe.ts +++ b/src/lib/adapters/http/probe.ts @@ -608,23 +608,37 @@ export interface AnthropicStreamingProbeResult { message: string; } +/** + * Known Anthropic Messages payload events that must sit between + * `message_start` and `message_stop` in a well-formed stream. + */ +const ANTHROPIC_CONTENT_STREAMING_EVENTS = new Set([ + "content_block_start", + "content_block_delta", + "content_block_stop", + "message_delta", +]); + /** * Order rules for a well-formed Anthropic Messages stream: `message_start` - * opens the stream before any content delta, and `message_stop` terminates - * it after the last content delta. Only evaluated once all required events - * are present; interleaved unknown events (e.g. `ping`) are ignored. + * opens the stream before any content event, and `message_stop` terminates + * it after the last one. Only evaluated once all required events are + * present; interleaved unknown events (e.g. `ping`) are ignored. */ function anthropicSequenceErrors(eventSequence: string[]): string[] { const errors: string[] = []; const firstStart = eventSequence.indexOf("message_start"); - const firstDelta = eventSequence.indexOf("content_block_delta"); - const lastDelta = eventSequence.lastIndexOf("content_block_delta"); const lastStop = eventSequence.lastIndexOf("message_stop"); - if (firstDelta < firstStart) { - errors.push("content_block_delta before message_start"); + const contentIndexes = eventSequence + .map((event, index) => (ANTHROPIC_CONTENT_STREAMING_EVENTS.has(event) ? index : -1)) + .filter((index) => index >= 0); + const firstContent = contentIndexes[0] ?? -1; + const lastContent = contentIndexes[contentIndexes.length - 1] ?? -1; + if (firstContent >= 0 && firstContent < firstStart) { + errors.push("content events before message_start"); } - if (lastStop < lastDelta) { - errors.push("content_block_delta after message_stop"); + if (lastContent >= 0 && lastStop < lastContent) { + errors.push("content events after message_stop"); } return errors; } From 31ca7a84a17ae978f8a5f384296436064e57de41 Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 11:30:50 -0700 Subject: [PATCH 05/12] fix(inference): preserve streaming timeout recovery Signed-off-by: Carlos Villela --- src/lib/adapters/http/probe.test.ts | 3 ++ src/lib/adapters/http/probe.ts | 14 ++++++- src/lib/inference/probe-anthropic.test.ts | 50 ++++++++++++++++++++++- src/lib/inference/probe-anthropic.ts | 2 +- 4 files changed, 66 insertions(+), 3 deletions(-) diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index 969e1763556..3871c7c66d4 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -846,6 +846,7 @@ describe("runAnthropicStreamingEventProbe", () => { ); expect(result.ok).toBe(true); + expect(result.curlStatus).toBe(0); expect(result.missingEvents).toEqual([]); expect(result.duplicateEvents).toEqual([]); expect(result.sequenceErrors).toEqual([]); @@ -1036,6 +1037,7 @@ describe("runAnthropicStreamingEventProbe", () => { ); expect(result.ok).toBe(false); + expect(result.curlStatus).toBe(28); expect(result.missingEvents).toEqual(["message_stop"]); }); @@ -1046,6 +1048,7 @@ describe("runAnthropicStreamingEventProbe", () => { ); expect(result.ok).toBe(true); + expect(result.curlStatus).toBe(28); }); it("fails on spawn error", () => { diff --git a/src/lib/adapters/http/probe.ts b/src/lib/adapters/http/probe.ts index bbd1e78626a..c3e1045b139 100644 --- a/src/lib/adapters/http/probe.ts +++ b/src/lib/adapters/http/probe.ts @@ -601,6 +601,8 @@ const SINGLETON_ANTHROPIC_STREAMING_EVENTS = ["message_start", "message_stop"]; export interface AnthropicStreamingProbeResult { ok: boolean; + /** curl exit status, including 28 when a bounded stream timed out. */ + curlStatus: number; missingEvents: string[]; duplicateEvents: string[]; /** Order violations, e.g. content deltas before message_start or after message_stop. */ @@ -681,6 +683,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + curlStatus: capture.curlStatus, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], sequenceErrors: [], @@ -719,6 +722,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + curlStatus: capture.curlStatus, missingEvents: missing, duplicateEvents: duplicates, sequenceErrors, @@ -735,7 +739,14 @@ function runAnthropicStreamingEventProbeImpl( sequence_errors_count: 0, curl_status: capture.curlStatus, }); - return { ok: true, missingEvents: [], duplicateEvents: [], sequenceErrors: [], message: "" }; + return { + ok: true, + curlStatus: capture.curlStatus, + missingEvents: [], + duplicateEvents: [], + sequenceErrors: [], + message: "", + }; } catch (error) { const detail = error instanceof Error ? error.message : String(error); const curlStatus = @@ -749,6 +760,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + curlStatus, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], sequenceErrors: [], diff --git a/src/lib/inference/probe-anthropic.test.ts b/src/lib/inference/probe-anthropic.test.ts index f0d0a0022d8..9600ec114ce 100644 --- a/src/lib/inference/probe-anthropic.test.ts +++ b/src/lib/inference/probe-anthropic.test.ts @@ -6,6 +6,7 @@ import fs from "node:fs"; import { afterEach, describe, expect, it, vi } from "vitest"; import * as probe from "../adapters/http/probe"; +import { getProbeRecovery } from "../validation-recovery"; import { probeAnthropicEndpoint } from "./probe-anthropic"; describe("probeAnthropicEndpoint", () => { @@ -88,6 +89,7 @@ describe("probeAnthropicEndpoint", () => { }); const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: true, + curlStatus: 0, missingEvents: [], duplicateEvents: [], sequenceErrors: [], @@ -114,12 +116,15 @@ describe("probeAnthropicEndpoint", () => { message: "HTTP 200", }); let streamingArgv: readonly string[] = []; + let streamingOpts: probe.CurlProbeOptions | undefined; const streamSpy = vi .spyOn(probe, "runAnthropicStreamingEventProbe") - .mockImplementation((argv) => { + .mockImplementation((argv, opts) => { streamingArgv = argv; + streamingOpts = opts; return { ok: true, + curlStatus: 0, missingEvents: [], duplicateEvents: [], sequenceErrors: [], @@ -143,6 +148,10 @@ describe("probeAnthropicEndpoint", () => { expect(streamingArgv.at(-1)).toBe("https://custom.endpoint.test/v1/messages"); expect(streamingArgv.join(" ")).toContain('"stream":true'); expect(streamingArgv.join(" ")).not.toContain("sk-custom-secret"); + const configIndex = streamingArgv.indexOf("--config"); + const configPath = configIndex >= 0 ? streamingArgv[configIndex + 1] : ""; + expect(streamingOpts?.trustedConfigFiles).toEqual([configPath]); + expect(fs.existsSync(configPath)).toBe(false); }); it("fails validation when the streaming event sequence is malformed", () => { @@ -156,6 +165,7 @@ describe("probeAnthropicEndpoint", () => { }); vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: false, + curlStatus: 0, missingEvents: [], duplicateEvents: ["message_start"], sequenceErrors: [], @@ -181,6 +191,43 @@ describe("probeAnthropicEndpoint", () => { expect(result.message).toContain("duplicate message_start"); }); + it("preserves streaming timeouts for transport recovery", () => { + vi.spyOn(probe, "runCurlProbe").mockReturnValue({ + ok: true, + httpStatus: 200, + curlStatus: 0, + body: "{}", + stderr: "", + message: "HTTP 200", + }); + vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ + ok: false, + curlStatus: 28, + missingEvents: ["message_stop"], + duplicateEvents: [], + sequenceErrors: [], + message: "Anthropic Messages streaming is missing required events: message_stop.", + }); + + const result = probeAnthropicEndpoint( + "https://custom.endpoint.test", + "nvidia/nemotron-3-super-v3", + "sk-custom-secret", + { probeStreaming: true }, + ); + + expect(result.failures?.[0]).toMatchObject({ + name: "Anthropic Messages API (streaming)", + httpStatus: 0, + curlStatus: 28, + }); + expect(getProbeRecovery(result)).toMatchObject({ + kind: "transport", + retry: "retry", + failure: { curlStatus: 28 }, + }); + }); + it("skips the streaming probe when the non-streaming probe already failed", () => { vi.spyOn(probe, "runCurlProbe").mockReturnValue({ ok: false, @@ -192,6 +239,7 @@ describe("probeAnthropicEndpoint", () => { }); const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: true, + curlStatus: 0, missingEvents: [], duplicateEvents: [], sequenceErrors: [], diff --git a/src/lib/inference/probe-anthropic.ts b/src/lib/inference/probe-anthropic.ts index fb689956a60..2e5f0d60d89 100644 --- a/src/lib/inference/probe-anthropic.ts +++ b/src/lib/inference/probe-anthropic.ts @@ -131,7 +131,7 @@ export function probeAnthropicEndpoint( { name: "Anthropic Messages API (streaming)", httpStatus: 0, - curlStatus: 0, + curlStatus: streamResult.curlStatus, message: streamResult.message, }, ], From 939498aaf35eeadf64f55aeb38af1c30fc58396a Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 11:45:10 -0700 Subject: [PATCH 06/12] fix(onboard): surface safe Anthropic streaming diagnostics Map malformed Anthropic SSE results to fixed diagnostic codes. Render only allowlisted messages while preserving raw-provider suppression. Keep HTTP and curl precedence so timeout recovery remains unchanged. Signed-off-by: Carlos Villela --- src/lib/inference/probe-anthropic.test.ts | 1 + src/lib/inference/probe-anthropic.ts | 47 +++++++++++++++++++ .../inference-selection-validation.test.ts | 4 ++ src/lib/onboard/probe-diagnostics.test.ts | 23 +++++++++ src/lib/onboard/probe-diagnostics.ts | 23 +++++++++ 5 files changed, 98 insertions(+) diff --git a/src/lib/inference/probe-anthropic.test.ts b/src/lib/inference/probe-anthropic.test.ts index 9600ec114ce..46902c84acd 100644 --- a/src/lib/inference/probe-anthropic.test.ts +++ b/src/lib/inference/probe-anthropic.test.ts @@ -187,6 +187,7 @@ describe("probeAnthropicEndpoint", () => { name: "Anthropic Messages API (streaming)", httpStatus: 0, curlStatus: 0, + diagnosticCodes: ["anthropic-streaming-duplicate-message-start"], }); expect(result.message).toContain("duplicate message_start"); }); diff --git a/src/lib/inference/probe-anthropic.ts b/src/lib/inference/probe-anthropic.ts index 2e5f0d60d89..9b455277295 100644 --- a/src/lib/inference/probe-anthropic.ts +++ b/src/lib/inference/probe-anthropic.ts @@ -8,17 +8,28 @@ import { createXApiKeyAuthConfig } from "../adapters/http/auth-config"; import { + type AnthropicStreamingProbeResult, getCurlTimingArgs, runAnthropicStreamingEventProbe, runCurlProbe, } from "../adapters/http/probe"; import { normalizeCredentialValue } from "../credentials/store"; +export type AnthropicStreamingDiagnosticCode = + | "anthropic-streaming-content-after-message-stop" + | "anthropic-streaming-content-before-message-start" + | "anthropic-streaming-duplicate-message-start" + | "anthropic-streaming-duplicate-message-stop" + | "anthropic-streaming-missing-content-block-delta" + | "anthropic-streaming-missing-message-start" + | "anthropic-streaming-missing-message-stop"; + export interface AnthropicProbeFailureDetail { name: string; httpStatus: number; curlStatus: number; message: string; + diagnosticCodes?: AnthropicStreamingDiagnosticCode[]; } export interface AnthropicProbeResult { @@ -67,6 +78,41 @@ function anthropicMessagesPayload(model: string, stream: boolean): string { }); } +const DUPLICATE_EVENT_DIAGNOSTICS: Record = { + message_start: "anthropic-streaming-duplicate-message-start", + message_stop: "anthropic-streaming-duplicate-message-stop", +}; + +const MISSING_EVENT_DIAGNOSTICS: Record = { + message_start: "anthropic-streaming-missing-message-start", + content_block_delta: "anthropic-streaming-missing-content-block-delta", + message_stop: "anthropic-streaming-missing-message-stop", +}; + +const SEQUENCE_ERROR_DIAGNOSTICS: Record = { + "content events before message_start": "anthropic-streaming-content-before-message-start", + "content events after message_stop": "anthropic-streaming-content-after-message-stop", +}; + +function anthropicStreamingDiagnosticCodes( + result: Pick< + AnthropicStreamingProbeResult, + "duplicateEvents" | "missingEvents" | "sequenceErrors" + >, +): AnthropicStreamingDiagnosticCode[] { + return [ + ...result.duplicateEvents.flatMap((event) => + DUPLICATE_EVENT_DIAGNOSTICS[event] ? [DUPLICATE_EVENT_DIAGNOSTICS[event]] : [], + ), + ...result.missingEvents.flatMap((event) => + MISSING_EVENT_DIAGNOSTICS[event] ? [MISSING_EVENT_DIAGNOSTICS[event]] : [], + ), + ...result.sequenceErrors.flatMap((error) => + SEQUENCE_ERROR_DIAGNOSTICS[error] ? [SEQUENCE_ERROR_DIAGNOSTICS[error]] : [], + ), + ]; +} + export function probeAnthropicEndpoint( endpointUrl: string, model: string, @@ -133,6 +179,7 @@ export function probeAnthropicEndpoint( httpStatus: 0, curlStatus: streamResult.curlStatus, message: streamResult.message, + diagnosticCodes: anthropicStreamingDiagnosticCodes(streamResult), }, ], }; diff --git a/src/lib/onboard/inference-selection-validation.test.ts b/src/lib/onboard/inference-selection-validation.test.ts index f8406e03ef9..40ef0c7dba6 100644 --- a/src/lib/onboard/inference-selection-validation.test.ts +++ b/src/lib/onboard/inference-selection-validation.test.ts @@ -169,6 +169,7 @@ describe("inference selection validation", () => { httpStatus: 0, curlStatus: 0, message: "duplicate message_start", + diagnosticCodes: ["anthropic-streaming-duplicate-message-start"], }, ], })); @@ -195,6 +196,9 @@ describe("inference selection validation", () => { expect(error.mock.calls.map((args) => args.join(" ")).join("\n")).toContain( "Custom Anthropic endpoint endpoint validation failed.", ); + expect(error.mock.calls.map((args) => args.join(" ")).join("\n")).toContain( + "Anthropic Messages API (streaming): duplicate message_start", + ); } finally { error.mockRestore(); } diff --git a/src/lib/onboard/probe-diagnostics.test.ts b/src/lib/onboard/probe-diagnostics.test.ts index 974884afc93..c864c9cae39 100644 --- a/src/lib/onboard/probe-diagnostics.test.ts +++ b/src/lib/onboard/probe-diagnostics.test.ts @@ -33,6 +33,7 @@ describe("summarizeProbeForDisplay", () => { httpStatus: 0, curlStatus: 28, message: "curl failed (exit 28): operation timed out with token secret-key", + diagnosticCodes: ["anthropic-streaming-missing-message-stop"], }, ], }); @@ -42,6 +43,28 @@ describe("summarizeProbeForDisplay", () => { expect(summary).not.toContain("operation timed out with token"); }); + it("surfaces allowlisted streaming diagnostics without raw provider text (#6289)", () => { + const summary = summarizeProbeForDisplay({ + message: "raw provider response with secret-key", + failures: [ + { + name: "Anthropic Messages API (streaming)", + httpStatus: 0, + curlStatus: 0, + message: "raw provider response with secret-key", + diagnosticCodes: [ + "anthropic-streaming-duplicate-message-start", + "provider-controlled-diagnostic", + ], + }, + ], + }); + + expect(summary).toBe("Anthropic Messages API (streaming): duplicate message_start"); + expect(summary).not.toContain("secret-key"); + expect(summary).not.toContain("provider-controlled-diagnostic"); + }); + it("falls back to coarse message classification", () => { expect(summarizeProbeForDisplay({ message: "HTTP 404: not found for secret-key" })).toBe( "HTTP 404", diff --git a/src/lib/onboard/probe-diagnostics.ts b/src/lib/onboard/probe-diagnostics.ts index 6759f266221..a5202ab36e6 100644 --- a/src/lib/onboard/probe-diagnostics.ts +++ b/src/lib/onboard/probe-diagnostics.ts @@ -1,12 +1,35 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 +const SAFE_PROBE_DIAGNOSTICS = new Map([ + ["anthropic-streaming-content-after-message-stop", "content events after message_stop"], + ["anthropic-streaming-content-before-message-start", "content events before message_start"], + ["anthropic-streaming-duplicate-message-start", "duplicate message_start"], + ["anthropic-streaming-duplicate-message-stop", "duplicate message_stop"], + ["anthropic-streaming-missing-content-block-delta", "missing content_block_delta"], + ["anthropic-streaming-missing-message-start", "missing message_start"], + ["anthropic-streaming-missing-message-stop", "missing message_stop"], +]); + +function summarizeSafeProbeDiagnostics(value: unknown): string[] { + if (!Array.isArray(value)) return []; + const summaries = new Set(); + for (const code of value) { + if (typeof code !== "string") continue; + const summary = SAFE_PROBE_DIAGNOSTICS.get(code); + if (summary) summaries.add(summary); + } + return [...summaries]; +} + function summarizeProbeFailureForDisplay(failure: Record): string { const name = typeof failure.name === "string" ? failure.name : "probe"; const httpStatus = typeof failure.httpStatus === "number" ? failure.httpStatus : 0; const curlStatus = typeof failure.curlStatus === "number" ? failure.curlStatus : 0; if (httpStatus > 0) return `${name}: HTTP ${httpStatus}`; if (curlStatus !== 0) return `${name}: curl exit ${curlStatus}`; + const diagnostics = summarizeSafeProbeDiagnostics(failure.diagnosticCodes); + if (diagnostics.length > 0) return `${name}: ${diagnostics.join("; ")}`; return `${name}: no HTTP response`; } From 7d1203d48bd129c436e1478eec24de97e6fbd23f Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 11:59:04 -0700 Subject: [PATCH 07/12] docs(inference): align Anthropic stream diagnostic Signed-off-by: Carlos Villela --- docs/inference/inference-options.mdx | 2 +- docs/reference/troubleshooting.mdx | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/inference/inference-options.mdx b/docs/inference/inference-options.mdx index bddc1a4a9a4..607e874c9c1 100644 --- a/docs/inference/inference-options.mdx +++ b/docs/inference/inference-options.mdx @@ -351,7 +351,7 @@ $$nemoclaw onboard NemoClaw validates the endpoint by sending a non-streaming `/v1/messages` request, then a `stream: true` request to the same path. The streaming check requires a well-formed SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). Endpoints and gateways whose non-streaming responses work but whose streaming layer is malformed fail validation during onboarding instead of failing later at runtime inside the sandbox. -Refer to [Onboarding fails with "Anthropic Messages streaming on this endpoint emits duplicate message_start"](../reference/troubleshooting#onboarding-fails-with-anthropic-messages-streaming-on-this-endpoint-emits-duplicate-message_start) if the streaming check fails. +Refer to [Onboarding fails with duplicate Anthropic message_start events](../reference/troubleshooting#onboarding-fails-with-duplicate-anthropic-message_start-events) if the streaming check fails. Set `NEMOCLAW_REASONING=true` to skip the streaming check when the endpoint serves a reasoning-only model. Agent runs still use the streaming path, so skipping the check moves any streaming defect to runtime. diff --git a/docs/reference/troubleshooting.mdx b/docs/reference/troubleshooting.mdx index 73c847d66d7..c93386b188c 100644 --- a/docs/reference/troubleshooting.mdx +++ b/docs/reference/troubleshooting.mdx @@ -1212,12 +1212,12 @@ Do not rely on `NEMOCLAW_INFERENCE_API_OVERRIDE` alone. It patches the config at container startup but does not update the Dockerfile ARG baked into the image. A fresh `$$nemoclaw onboard` is the reliable fix. -### Onboarding fails with "Anthropic Messages streaming on this endpoint emits duplicate message_start" +### Onboarding fails with duplicate Anthropic message_start events Validation for the **Other Anthropic-compatible endpoint** provider ends with an error like: ```text -Anthropic Messages API (streaming): Anthropic Messages streaming on this endpoint emits duplicate message_start (2 events for one request). Agent runs use the streaming path and would fail with an empty final response. +Anthropic Messages API (streaming): duplicate message_start ``` During onboarding, NemoClaw sends a `stream: true` request to `/v1/messages` and validates the SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). From 5aea46f1d4812e3d6075cd10a96755daecb5355a Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 12:02:16 -0700 Subject: [PATCH 08/12] test(sandbox): accept notice in rebuild recovery harness Signed-off-by: Carlos Villela --- .../sandbox/rebuild-prepared-recovery.test.ts | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/src/lib/actions/sandbox/rebuild-prepared-recovery.test.ts b/src/lib/actions/sandbox/rebuild-prepared-recovery.test.ts index 3f0935d67de..fd5851e2449 100644 --- a/src/lib/actions/sandbox/rebuild-prepared-recovery.test.ts +++ b/src/lib/actions/sandbox/rebuild-prepared-recovery.test.ts @@ -1,27 +1,24 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -import { createRequire } from "node:module"; - -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { createRebuildFlowHarness, makePreparedRecoveryManifest, + resetRebuildFlowTestEnvironment, + restoreRebuildFlowTestEnvironment, snapshotEnv, } from "../../../../test/helpers/rebuild-flow-harness"; -const requireDist = createRequire(import.meta.url); -const rebuildModulePath = "./rebuild.js"; const restoreSandboxEnv = snapshotEnv(["NEMOCLAW_SANDBOX_NAME"]); describe("prepared rebuild recovery", () => { beforeEach(() => { - delete process.env.NEMOCLAW_SANDBOX_NAME; + resetRebuildFlowTestEnvironment(); }); afterEach(() => { - vi.restoreAllMocks(); - delete require.cache[requireDist.resolve(rebuildModulePath)]; + restoreRebuildFlowTestEnvironment(); restoreSandboxEnv(); }); From 4c5fba2dc14e816d3976501140210b7dc34e8316 Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 12:44:21 -0700 Subject: [PATCH 09/12] fix(onboard): validate selected inference surface Signed-off-by: Carlos Villela --- docs/inference/inference-options.mdx | 11 ++-- docs/reference/troubleshooting.mdx | 12 ++-- src/lib/adapters/http/curl-args.ts | 5 +- src/lib/adapters/http/probe.test.ts | 17 ++++- src/lib/adapters/http/probe.ts | 57 +++++++++++++++-- src/lib/inference/config.test.ts | 31 +++++++++ src/lib/inference/config.ts | 27 ++++++++ src/lib/inference/probe-anthropic.test.ts | 9 ++- src/lib/inference/probe-anthropic.ts | 2 +- src/lib/onboard.ts | 7 ++- src/lib/onboard/inference-providers/remote.ts | 7 +-- .../inference-selection-validation.test.ts | 63 +++++++++++++++++-- .../onboard/inference-selection-validation.ts | 44 +++++++++---- .../machine/handlers/provider-inference.ts | 10 +-- src/lib/onboard/probe-diagnostics.test.ts | 21 ++++++- src/lib/onboard/probe-diagnostics.ts | 5 +- src/lib/onboard/setup-nim-flow.test.ts | 32 ++++++++++ src/lib/onboard/setup-nim-flow.ts | 29 ++++++++- src/lib/onboard/setup-nim-selection.test.ts | 56 +++++++++++++++++ src/lib/onboard/setup-nim-selection.ts | 14 +++++ 20 files changed, 408 insertions(+), 51 deletions(-) diff --git a/docs/inference/inference-options.mdx b/docs/inference/inference-options.mdx index 5f200b5ecf9..4061a73b2e9 100644 --- a/docs/inference/inference-options.mdx +++ b/docs/inference/inference-options.mdx @@ -75,7 +75,7 @@ NemoClaw neither displays nor accepts an unsafe `NEMOCLAW_MODEL` value as the ma | OpenAI | Routes to the OpenAI API. Set `OPENAI_API_KEY`. | `gpt-5.4`, `gpt-5.4-mini`, `gpt-5.4-nano`, `gpt-5.4-pro-2026-03-05` | | Other OpenAI-compatible endpoint | Routes to any server that implements `/v1/chat/completions`. NemoClaw uses `/v1/chat/completions` at runtime by default; set `NEMOCLAW_PREFERRED_API=openai-responses` to allow `/v1/responses` for proxies that implement it, such as some llama.cpp builds. The wizard prompts for a base URL and model name. The adapter is validated against OpenRouter (refer to the status table above); behavior on other OpenAI-compatible proxies, gateways, and self-hosted implementations such as LocalAI or llama.cpp may vary. When you enable Telegram messaging, onboarding also runs a bounded sandbox-side smoke check through `https://inference.local/v1/chat/completions`. Set `COMPATIBLE_API_KEY`. | You provide the model name. | | Anthropic | Routes to the Anthropic Messages API. Set `ANTHROPIC_API_KEY`. | `claude-sonnet-4-6`, `claude-haiku-4-5`, `claude-opus-4-6` | -| Other Anthropic-compatible endpoint | Routes to any server that implements the Anthropic Messages API (`/v1/messages`). The adapter is validated against AWS Bedrock (refer to the status table above); behavior on other Anthropic-compatible proxies and gateways may vary. The wizard prompts for a base URL and model name. Set `COMPATIBLE_ANTHROPIC_API_KEY`. | You provide the model name. | +| Other Anthropic-compatible endpoint | Routes agents that support Anthropic Messages, including OpenClaw, to `/v1/messages`. For Hermes and agents that only support OpenAI-compatible inference, NemoClaw instead requires `/v1/chat/completions`, which is the surface it validates and uses at runtime. The adapter is validated against AWS Bedrock (refer to the status table above); behavior on other Anthropic-compatible proxies and gateways may vary. The wizard prompts for a base URL and model name. Set `COMPATIBLE_ANTHROPIC_API_KEY`. | You provide the model name. | | Google Gemini | Routes to Google's OpenAI-compatible chat-completions endpoint. NemoClaw skips the Responses-API probe because Gemini does not support `/v1/responses`. Set `GEMINI_API_KEY`. | `gemini-3.1-pro-preview`, `gemini-3.1-flash-lite-preview`, `gemini-3-flash-preview`, `gemini-2.5-pro`, `gemini-2.5-flash`, `gemini-2.5-flash-lite` | | Hermes Provider | Routes Hermes Agent through the host OpenShell provider registered by NemoClaw when onboarding Hermes Agent. | Curated Hermes Provider models such as `moonshotai/kimi-k2.6`, `openai/gpt-5.4-mini`, and `z-ai/glm-5.1`. | | Local Ollama | Routes to a local Ollama instance on `localhost:11434`. NemoClaw detects installed models, offers starter models if none are present, pulls and warms the selected model, and validates it. | Selected during onboarding. For more information, refer to [Use a Local Inference Server](use-local-inference). | @@ -243,7 +243,7 @@ Other provider credentials, such as `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `GEMI | NVIDIA Endpoints | Validates through `/v1/chat/completions` only; NemoClaw skips the `/v1/responses` probe because NVIDIA Build does not expose `/v1/responses` (returns 404 for every model). | | Google Gemini | Validates through Gemini's OpenAI-compatible chat-completions path only; NemoClaw skips the `/v1/responses` probe because Gemini does not support the Responses API. | | Other OpenAI-compatible endpoint | Tries `/v1/responses` first with a tool-calling probe; falls back to `/v1/chat/completions`. Selected runtime API defaults to `/v1/chat/completions`; set `NEMOCLAW_PREFERRED_API=openai-responses` to allow `/v1/responses` at runtime when validation succeeds. | -| Anthropic-compatible | Tries `/v1/messages` with a non-streaming request, then repeats the request with `stream: true` and validates the SSE event sequence. Set `NEMOCLAW_REASONING=true` to skip the streaming check for reasoning-only endpoints. | +| Other Anthropic-compatible endpoint | For agents that support Anthropic Messages, including OpenClaw, tries `/v1/messages` with a non-streaming request, then repeats the request with `stream: true` and validates the SSE event sequence. Set `NEMOCLAW_REASONING=true` to skip the streaming check for reasoning-only endpoints. For Hermes and OpenAI-compatible-only agents, validates `/v1/chat/completions`, the surface used by the managed OpenAI frontend. | | NVIDIA Endpoints (manual model entry) | Validates the model name against the catalog API. | | Compatible endpoints | Sends a real inference request because many proxies do not expose a `/models` endpoint. For OpenAI-compatible endpoints, the probe tries `/v1/responses` first then falls back to `/v1/chat/completions`; the selected runtime API defaults to `/v1/chat/completions`. Set `NEMOCLAW_PREFERRED_API=openai-responses` to allow `/v1/responses` at runtime when validation succeeds. | | Local NVIDIA NIM | Validates through `/v1/chat/completions` only; NemoClaw skips the `/v1/responses` probe (same as NVIDIA Endpoints). | @@ -342,22 +342,25 @@ Refer to [Switch Inference Models](switch-inference-providers) for more informat ## Anthropic-Compatible Server -If your local server implements the Anthropic Messages API (`/v1/messages`), choose **Other Anthropic-compatible endpoint** during onboarding instead. +Choose **Other Anthropic-compatible endpoint** during onboarding to configure a custom base URL with `COMPATIBLE_ANTHROPIC_API_KEY`. ```bash $$nemoclaw onboard ``` + NemoClaw validates the endpoint by sending a non-streaming `/v1/messages` request, then a `stream: true` request to the same path. The streaming check requires a well-formed SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). Endpoints and gateways whose non-streaming responses work but whose streaming layer is malformed fail validation during onboarding instead of failing later at runtime inside the sandbox. Refer to [Onboarding fails with duplicate Anthropic message_start events](../reference/troubleshooting#onboarding-fails-with-duplicate-anthropic-message_start-events) if the streaming check fails. Set `NEMOCLAW_REASONING=true` to skip the streaming check when the endpoint serves a reasoning-only model. Agent runs still use the streaming path, so skipping the check moves any streaming defect to runtime. + For `compatible-anthropic-endpoint`, Hermes uses the managed OpenAI Chat Completions frontend at `https://inference.local/v1`. -During onboarding, NemoClaw verifies that the endpoint also serves `/v1/chat/completions`, then registers that surface with OpenShell as `type=openai` using `OPENAI_BASE_URL`. +During provider selection, NemoClaw validates `/v1/chat/completions` instead of probing the unused native Anthropic SSE path. +Inference setup verifies the same path again, then registers that surface with OpenShell as `type=openai` using `OPENAI_BASE_URL`. The route retains `COMPATIBLE_ANTHROPIC_API_KEY` as its credential binding. This avoids duplicate Anthropic SSE `message_start` events. If the endpoint only serves Anthropic Messages, onboarding stops with guidance instead of creating a Hermes sandbox with an unroutable or broken streaming path. diff --git a/docs/reference/troubleshooting.mdx b/docs/reference/troubleshooting.mdx index c93386b188c..8f49a56bf37 100644 --- a/docs/reference/troubleshooting.mdx +++ b/docs/reference/troubleshooting.mdx @@ -1214,21 +1214,25 @@ A fresh `$$nemoclaw onboard` is the reliable fix. ### Onboarding fails with duplicate Anthropic message_start events -Validation for the **Other Anthropic-compatible endpoint** provider ends with an error like: +Validation for an OpenClaw **Other Anthropic-compatible endpoint** selection ends with an error like: ```text Anthropic Messages API (streaming): duplicate message_start ``` -During onboarding, NemoClaw sends a `stream: true` request to `/v1/messages` and validates the SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). +For OpenClaw custom Anthropic routes, NemoClaw sends a `stream: true` request to `/v1/messages` and validates the SSE event sequence (exactly one `message_start`, at least one `content_block_delta`, and a `message_stop`). This error means the streaming layer on the endpoint or gateway is malformed even though its non-streaming responses are valid. A working non-streaming response does not imply that streaming works. Some inference gateways proxy plain requests correctly but corrupt the SSE stream, for example by emitting `message_start` twice for one request. -Agent runs always use the streaming path, so without this check the defect would first surface inside the sandbox as a runtime failure, such as Hermes reporting that no final response was produced. +OpenClaw uses the streaming path, so without this check the defect would first surface inside the sandbox as a runtime failure. + +Hermes and OpenAI-compatible-only agents use the endpoint's `/v1/chat/completions` surface for custom Anthropic selections instead. +Current onboarding validates that surface and does not reject those agents because of a malformed native `/v1/messages` stream they will not use. +An older Hermes sandbox that still uses native Anthropic Messages can report that no final response was produced; re-run onboarding to select and validate the managed Chat Completions route. Fix the streaming layer on the endpoint or gateway, or onboard with a different Anthropic-compatible endpoint. The official Anthropic provider does not run this check and is not affected. -If a sandbox created by an older release fails at runtime with an empty final response on an Anthropic-compatible endpoint, re-run `$$nemoclaw onboard` so the streaming check can diagnose the endpoint. +If an OpenClaw sandbox created by an older release fails at runtime with an empty final response on an Anthropic-compatible endpoint, re-run `$$nemoclaw onboard` so the streaming check can diagnose the endpoint. If the endpoint serves a reasoning-only model, set `NEMOCLAW_REASONING=true` to skip the streaming check. Streaming defects then surface at runtime instead of during onboarding. diff --git a/src/lib/adapters/http/curl-args.ts b/src/lib/adapters/http/curl-args.ts index 72bf2b66c97..9ab69d40d25 100644 --- a/src/lib/adapters/http/curl-args.ts +++ b/src/lib/adapters/http/curl-args.ts @@ -244,7 +244,7 @@ export function buildValidatedCurlCommandArgs( return [...args, url]; } -export type CurlProbeMode = "json" | "chat-stream" | "event-stream"; +export type CurlProbeMode = "json" | "chat-stream" | "event-stream" | "event-stream-with-status"; export function buildCurlProbeSpawnArgs( args: string[], @@ -254,7 +254,8 @@ export function buildCurlProbeSpawnArgs( ): string[] { const outputArgs = mode === "json" ? ["-o", bodyFile, "-w", "%{http_code}"] : ["-N", "-o", bodyFile]; - const statusArgs = mode === "chat-stream" ? ["-w", "%{http_code}"] : []; + const statusArgs = + mode === "chat-stream" || mode === "event-stream-with-status" ? ["-w", "%{http_code}"] : []; // lgtm[js/file-access-to-http] URL/argv are validated; file-backed config paths must be explicitly trusted. return [...args, ...outputArgs, ...statusArgs, url]; } diff --git a/src/lib/adapters/http/probe.test.ts b/src/lib/adapters/http/probe.test.ts index 3871c7c66d4..5fa66f75e98 100644 --- a/src/lib/adapters/http/probe.test.ts +++ b/src/lib/adapters/http/probe.test.ts @@ -804,13 +804,13 @@ describe("runStreamingEventProbe", () => { describe("runAnthropicStreamingEventProbe", () => { /** Helper to build a spawnSyncImpl that writes SSE content to the -o file. */ - function mockStreaming(sseBody: string, exitCode = 0) { + function mockStreaming(sseBody: string, exitCode = 0, httpStatus = "200") { return (_command: string, args: readonly string[]) => { writeCurlOutputBody(args, sseBody); return { pid: 1, output: [], - stdout: "", + stdout: httpStatus, stderr: "", status: exitCode, signal: null, @@ -852,6 +852,17 @@ describe("runAnthropicStreamingEventProbe", () => { expect(result.sequenceErrors).toEqual([]); }); + it("rejects a non-2xx response even when its body looks like valid SSE", () => { + const result = runAnthropicStreamingEventProbe( + ["-sS", "--max-time", "15", "https://example.test/v1/messages"], + { spawnSyncImpl: mockStreaming(healthyStream, 0, "503") }, + ); + + expect(result.ok).toBe(false); + expect(result.httpStatus).toBe(503); + expect(result.message).toContain("HTTP 503"); + }); + it("fails when message_stop is emitted twice for one request", () => { const duplicatedStop = [ "event: message_start", @@ -1119,7 +1130,7 @@ describe("runAnthropicStreamingEventProbe", () => { return { pid: 1, output: [], - stdout: "", + stdout: "200", stderr: "", status: 0, signal: null, diff --git a/src/lib/adapters/http/probe.ts b/src/lib/adapters/http/probe.ts index c3e1045b139..9e197a3b321 100644 --- a/src/lib/adapters/http/probe.ts +++ b/src/lib/adapters/http/probe.ts @@ -447,6 +447,7 @@ export function runStreamingEventProbe( interface SseEventCaptureResult { ok: boolean; + httpStatus: number; curlStatus: number; /** Transport/execution error detail when `ok` is false. */ detail: string; @@ -465,13 +466,19 @@ function captureSseEventCounts( argv: string[], opts: CurlProbeOptions, tempPrefix: string, + captureHttpStatus = false, ): SseEventCaptureResult { const bodyFile = secureTempFile(tempPrefix, ".sse"); try { const { args, url } = validateCurlProbeArgs(argv, opts); const spawnSyncImpl = opts.spawnSyncImpl ?? spawnSync; const timeout = resolveCurlProcessTimeoutMs(argv, opts); - const curlArgs = buildCurlProbeSpawnArgs(args, url, bodyFile, "event-stream"); + const curlArgs = buildCurlProbeSpawnArgs( + args, + url, + bodyFile, + captureHttpStatus ? "event-stream-with-status" : "event-stream", + ); const result = spawnSyncImpl( "curl", // lgtm[js/file-access-to-http] curlArgs were validated and rebuilt from safe probe fields. @@ -495,7 +502,32 @@ function captureSseEventCounts( const detail = result.error ? String(result.error.message || result.error) : String(result.stderr || ""); - return { ok: false, curlStatus, detail, eventCounts: new Map(), eventSequence: [] }; + return { + ok: false, + httpStatus: 0, + curlStatus, + detail, + eventCounts: new Map(), + eventSequence: [], + }; + } + + const status = captureHttpStatus ? Number(String(result.stdout || "").trim()) : 0; + const httpStatus = captureHttpStatus && Number.isFinite(status) ? status : 0; + if (captureHttpStatus && (httpStatus < 200 || httpStatus >= 300)) { + return { + ok: false, + httpStatus, + curlStatus: result.status ?? 0, + detail: summarizeProbeFailure( + body, + httpStatus, + result.status ?? 0, + String(result.stderr || ""), + ), + eventCounts: new Map(), + eventSequence: [], + }; } // Parse SSE event types from the raw output. @@ -510,7 +542,14 @@ function captureSseEventCounts( eventSequence.push(eventType); } } - return { ok: true, curlStatus: result.status ?? 0, detail: "", eventCounts, eventSequence }; + return { + ok: true, + httpStatus, + curlStatus: result.status ?? 0, + detail: "", + eventCounts, + eventSequence, + }; } finally { cleanupTempDir(bodyFile, tempPrefix); } @@ -601,6 +640,8 @@ const SINGLETON_ANTHROPIC_STREAMING_EVENTS = ["message_start", "message_stop"]; export interface AnthropicStreamingProbeResult { ok: boolean; + /** HTTP response status, or 0 when no HTTP response was received. */ + httpStatus: number; /** curl exit status, including 28 when a bounded stream timed out. */ curlStatus: number; missingEvents: string[]; @@ -672,10 +713,11 @@ function runAnthropicStreamingEventProbeImpl( opts: CurlProbeOptions = {}, ): AnthropicStreamingProbeResult { try { - const capture = captureSseEventCounts(argv, opts, "nemoclaw-anthropic-streaming-probe"); + const capture = captureSseEventCounts(argv, opts, "nemoclaw-anthropic-streaming-probe", true); if (!capture.ok) { emitCurlResultTraceEvent({ ok: false, + http_status: capture.httpStatus, missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, duplicate_events_count: 0, sequence_errors_count: 0, @@ -683,6 +725,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + httpStatus: capture.httpStatus, curlStatus: capture.curlStatus, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], @@ -715,6 +758,7 @@ function runAnthropicStreamingEventProbeImpl( } emitCurlResultTraceEvent({ ok: false, + http_status: capture.httpStatus, missing_events_count: missing.length, duplicate_events_count: duplicates.length, sequence_errors_count: sequenceErrors.length, @@ -722,6 +766,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + httpStatus: capture.httpStatus, curlStatus: capture.curlStatus, missingEvents: missing, duplicateEvents: duplicates, @@ -734,6 +779,7 @@ function runAnthropicStreamingEventProbeImpl( emitCurlResultTraceEvent({ ok: true, + http_status: capture.httpStatus, missing_events_count: 0, duplicate_events_count: 0, sequence_errors_count: 0, @@ -741,6 +787,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: true, + httpStatus: capture.httpStatus, curlStatus: capture.curlStatus, missingEvents: [], duplicateEvents: [], @@ -753,6 +800,7 @@ function runAnthropicStreamingEventProbeImpl( typeof error === "object" && error && "status" in error ? Number(error.status) || 1 : 1; emitCurlResultTraceEvent({ ok: false, + http_status: 0, missing_events_count: REQUIRED_ANTHROPIC_STREAMING_EVENTS.length, duplicate_events_count: 0, sequence_errors_count: 0, @@ -760,6 +808,7 @@ function runAnthropicStreamingEventProbeImpl( }); return { ok: false, + httpStatus: 0, curlStatus, missingEvents: REQUIRED_ANTHROPIC_STREAMING_EVENTS, duplicateEvents: [], diff --git a/src/lib/inference/config.test.ts b/src/lib/inference/config.test.ts index c982d59424d..8078a8a4d64 100644 --- a/src/lib/inference/config.test.ts +++ b/src/lib/inference/config.test.ts @@ -12,6 +12,7 @@ import { DEFAULT_OLLAMA_MODEL, DEFAULT_ROUTE_CREDENTIAL_ENV, DEFAULT_ROUTE_PROFILE, + getCompatibleAnthropicOpenAiSurfaceBaseUrl, getOpenClawPrimaryModel, getProviderSelectionConfig, getSandboxInferenceConfig, @@ -22,6 +23,7 @@ import { parseGatewayInference, planInferenceRouteReconcile, resolveAgentInferenceApi, + resolveAgentProviderInferenceApi, sanitizeRouteValueForDisplay, VLLM_LOCAL_CREDENTIAL_ENV, } from "./config"; @@ -46,6 +48,35 @@ describe("resolveAgentInferenceApi", () => { }); }); +describe("resolveAgentProviderInferenceApi", () => { + it("uses Chat Completions for an OpenAI-only DCode agent on a custom Anthropic provider (#6294)", () => { + const dcodeAgent = { + name: "langchain-deepagents-code", + inference: { provider_type: "openai_compatible" }, + }; + + expect( + resolveAgentProviderInferenceApi( + dcodeAgent.name, + dcodeAgent, + "compatible-anthropic-endpoint", + "anthropic-messages", + ), + ).toBe("openai-completions"); + }); +}); + +describe("getCompatibleAnthropicOpenAiSurfaceBaseUrl", () => { + it.each([ + ["https://proxy.example.com", "https://proxy.example.com/v1"], + ["https://proxy.example.com/tenant", "https://proxy.example.com/tenant/v1"], + ["https://proxy.example.com/v1", "https://proxy.example.com/v1"], + ["https://proxy.example.com/v1/", "https://proxy.example.com/v1"], + ])("maps %s to the runtime Chat Completions base", (endpointUrl, expected) => { + expect(getCompatibleAnthropicOpenAiSurfaceBaseUrl(endpointUrl)).toBe(expected); + }); +}); + describe("inference selection config", () => { it("exposes the curated cloud model picker options", () => { expect(CLOUD_MODEL_OPTIONS).toEqual([ diff --git a/src/lib/inference/config.ts b/src/lib/inference/config.ts index 70e5410cd08..0f9ae6d8347 100644 --- a/src/lib/inference/config.ts +++ b/src/lib/inference/config.ts @@ -109,6 +109,19 @@ export function resolveAgentInferenceApi( : preferredInferenceApi; } +/** + * Return the OpenAI-compatible base used when a custom Anthropic endpoint is + * routed through the managed Chat Completions frontend. Anthropic endpoint + * normalization intentionally strips a trailing `/v1`; OpenShell's OpenAI + * provider appends `/chat/completions`, so restore `/v1` exactly once here. + */ +export function getCompatibleAnthropicOpenAiSurfaceBaseUrl( + endpointUrl: string | null | undefined, +): string { + const trimmed = String(endpointUrl ?? "").replace(/\/+$/, ""); + return trimmed.endsWith("/v1") ? trimmed : `${trimmed}/v1`; +} + export function getProviderSelectionConfig( provider: string, model?: string, @@ -300,6 +313,20 @@ export function coerceAgentInferenceApi( return preferredInferenceApi; } +/** Resolve the runtime API after applying both agent capability and provider overrides. */ +export function resolveAgentProviderInferenceApi( + agentName: string | null | undefined, + agent: unknown, + provider: string | null | undefined, + preferredInferenceApi: string | null, +): string | null { + return resolveAgentInferenceApi( + agentName, + provider, + coerceAgentInferenceApi(agent, preferredInferenceApi), + ); +} + export function parseGatewayInference(output: string | null | undefined): GatewayInference | null { if (!output) return null; const stripped = output.replace(/\u001b\[[0-9;]*m/g, ""); diff --git a/src/lib/inference/probe-anthropic.test.ts b/src/lib/inference/probe-anthropic.test.ts index 46902c84acd..960113359fe 100644 --- a/src/lib/inference/probe-anthropic.test.ts +++ b/src/lib/inference/probe-anthropic.test.ts @@ -89,6 +89,7 @@ describe("probeAnthropicEndpoint", () => { }); const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: true, + httpStatus: 200, curlStatus: 0, missingEvents: [], duplicateEvents: [], @@ -124,6 +125,7 @@ describe("probeAnthropicEndpoint", () => { streamingOpts = opts; return { ok: true, + httpStatus: 200, curlStatus: 0, missingEvents: [], duplicateEvents: [], @@ -165,6 +167,7 @@ describe("probeAnthropicEndpoint", () => { }); vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: false, + httpStatus: 200, curlStatus: 0, missingEvents: [], duplicateEvents: ["message_start"], @@ -185,7 +188,7 @@ describe("probeAnthropicEndpoint", () => { expect(result.ok).toBe(false); expect(result.failures?.[0]).toMatchObject({ name: "Anthropic Messages API (streaming)", - httpStatus: 0, + httpStatus: 200, curlStatus: 0, diagnosticCodes: ["anthropic-streaming-duplicate-message-start"], }); @@ -203,6 +206,7 @@ describe("probeAnthropicEndpoint", () => { }); vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: false, + httpStatus: 200, curlStatus: 28, missingEvents: ["message_stop"], duplicateEvents: [], @@ -219,7 +223,7 @@ describe("probeAnthropicEndpoint", () => { expect(result.failures?.[0]).toMatchObject({ name: "Anthropic Messages API (streaming)", - httpStatus: 0, + httpStatus: 200, curlStatus: 28, }); expect(getProbeRecovery(result)).toMatchObject({ @@ -240,6 +244,7 @@ describe("probeAnthropicEndpoint", () => { }); const streamSpy = vi.spyOn(probe, "runAnthropicStreamingEventProbe").mockReturnValue({ ok: true, + httpStatus: 200, curlStatus: 0, missingEvents: [], duplicateEvents: [], diff --git a/src/lib/inference/probe-anthropic.ts b/src/lib/inference/probe-anthropic.ts index 9b455277295..921143e616b 100644 --- a/src/lib/inference/probe-anthropic.ts +++ b/src/lib/inference/probe-anthropic.ts @@ -176,7 +176,7 @@ export function probeAnthropicEndpoint( failures: [ { name: "Anthropic Messages API (streaming)", - httpStatus: 0, + httpStatus: streamResult.httpStatus, curlStatus: streamResult.curlStatus, message: streamResult.message, diagnosticCodes: anthropicStreamingDiagnosticCodes(streamResult), diff --git a/src/lib/onboard.ts b/src/lib/onboard.ts index a42ee5dc3a9..7ab0d83ae09 100644 --- a/src/lib/onboard.ts +++ b/src/lib/onboard.ts @@ -3202,7 +3202,7 @@ type SetupNimSelectionState = type SetupNimSelectionResult = "selected" | "retry-selection"; // biome-ignore format: keep src/lib/onboard.ts net-neutral for growth guardrail. -type RemoteProviderSelectionArgs = { selected: ProviderChoice; requestedModel: string | null; recoveredFromSandbox: boolean; recoveredModel: string | null; sandboxName: string | null }; +type RemoteProviderSelectionArgs = { selected: ProviderChoice; requestedModel: string | null; recoveredFromSandbox: boolean; recoveredModel: string | null; sandboxName: string | null; intendedInferenceApi: string | null }; async function handleVllmSelection( state: SetupNimSelectionState, @@ -3465,7 +3465,7 @@ async function handleNimLocalSelection( // biome-ignore format: keep src/lib/onboard.ts net-neutral for growth guardrail. async function handleRemoteProviderSelection(args: RemoteProviderSelectionArgs, state: SetupNimSelectionState, recoveredRegistryRoute: RebuildRouteHandoff["route"] | null): Promise { - const { selected, requestedModel, recoveredFromSandbox, recoveredModel, sandboxName } = args; + const { selected, requestedModel, recoveredFromSandbox, recoveredModel, sandboxName, intendedInferenceApi } = args; const remoteConfig = REMOTE_PROVIDER_CONFIG[selected.key]; state.provider = remoteConfig.providerName; state.credentialEnv = remoteConfig.credentialEnv; @@ -3704,11 +3704,12 @@ async function handleRemoteProviderSelection(args: RemoteProviderSelectionArgs, const validationResult = state.reuseGatewayCredentialWithoutLocalKey ? "selected" - : await validateSelectedRemoteModel({ + : await validateSelectedRemoteModel({ selected, remoteConfig, state, selectedCredentialEnv, + intendedInferenceApi, }); if (validationResult === "selected") break; if (validationResult === "retry-selection") return "retry-selection"; diff --git a/src/lib/onboard/inference-providers/remote.ts b/src/lib/onboard/inference-providers/remote.ts index 3f1dfdd3f65..466818404f5 100644 --- a/src/lib/onboard/inference-providers/remote.ts +++ b/src/lib/onboard/inference-providers/remote.ts @@ -6,6 +6,7 @@ // onboard.setupInference (#767). Bedrock Runtime is delegated to // `onboard/bedrock-runtime.ts` exactly as the inline branch did. +import { getCompatibleAnthropicOpenAiSurfaceBaseUrl } from "../../inference/config"; import { readGatewayProviderMetadata } from "../gateway-provider-metadata"; import { deleteProviderWithRecovery, parseAttachedSandboxes } from "../sandbox-provider-cleanup"; import type { RemoteProviderDeps, SetupInferenceResult } from "./types"; @@ -238,10 +239,8 @@ export async function setupRemoteProviderInference( // to /v1/chat/completions, deduping only bases that // already end in /v1. Re-add the suffix so the probe and the runtime // route exercise the identical URL. - const trimmedSurfaceBase = String(resolvedEndpointUrl ?? "").replace(/\/+$/, ""); - const openAiSurfaceBaseUrl = trimmedSurfaceBase.endsWith("/v1") - ? trimmedSurfaceBase - : `${trimmedSurfaceBase}/v1`; + const openAiSurfaceBaseUrl = + getCompatibleAnthropicOpenAiSurfaceBaseUrl(resolvedEndpointUrl); const surfaceProbe = probeOpenAiSurface(openAiSurfaceBaseUrl, model, credentialValue, { skipResponsesProbe: true, }); diff --git a/src/lib/onboard/inference-selection-validation.test.ts b/src/lib/onboard/inference-selection-validation.test.ts index 40ef0c7dba6..bfcd0e2233e 100644 --- a/src/lib/onboard/inference-selection-validation.test.ts +++ b/src/lib/onboard/inference-selection-validation.test.ts @@ -87,7 +87,7 @@ describe("inference selection validation", () => { } }); - it("requests streaming validation for custom Anthropic-compatible endpoints (#6289)", async () => { + it("requests streaming validation for OpenClaw custom Anthropic endpoints (#6289)", async () => { const probeAnthropicEndpoint = vi.fn(() => ({ ok: true, api: "anthropic-messages", @@ -96,7 +96,7 @@ describe("inference selection validation", () => { const log = vi.spyOn(console, "log").mockImplementation(() => {}); const helpers = createInferenceSelectionValidationHelpers({ isNonInteractive: () => false, - agentProductName: () => "Hermes", + agentProductName: () => "OpenClaw", getCredential: () => "test-key", probeAnthropicEndpoint, promptValidationRecovery: vi.fn(async () => "selection" as const), @@ -122,6 +122,57 @@ describe("inference selection validation", () => { } }); + it("validates Hermes custom Anthropic routes on their intended Chat Completions surface (#6289)", async () => { + const probeAnthropicEndpoint = vi.fn(() => ({ + ok: false, + message: "duplicate message_start", + failures: [ + { + name: "Anthropic Messages API (streaming)", + httpStatus: 200, + curlStatus: 0, + message: "duplicate message_start", + }, + ], + })); + const probeOpenAiLikeEndpoint = vi.fn(() => ({ + ok: true, + api: "openai-completions", + label: "Chat Completions API", + })); + const log = vi.spyOn(console, "log").mockImplementation(() => {}); + const helpers = createInferenceSelectionValidationHelpers({ + isNonInteractive: () => false, + agentProductName: () => "Hermes", + getCredential: () => "test-key", + probeAnthropicEndpoint, + probeOpenAiLikeEndpoint, + promptValidationRecovery: vi.fn(async () => "selection" as const), + }); + + try { + await expect( + helpers.validateCustomAnthropicSelection( + "Custom Anthropic endpoint", + "https://compatible.example", + "nvidia/nemotron-3-super-v3", + "COMPATIBLE_ANTHROPIC_API_KEY", + null, + { intendedApi: "openai-completions" }, + ), + ).resolves.toEqual({ ok: true, api: "openai-completions" }); + expect(probeOpenAiLikeEndpoint).toHaveBeenCalledWith( + "https://compatible.example/v1", + "nvidia/nemotron-3-super-v3", + "test-key", + { skipResponsesProbe: true }, + ); + expect(probeAnthropicEndpoint).not.toHaveBeenCalled(); + } finally { + log.mockRestore(); + } + }); + it("skips Anthropic streaming validation in reasoning mode", async () => { vi.stubEnv("NEMOCLAW_REASONING", "yes"); const probeAnthropicEndpoint = vi.fn(() => ({ @@ -132,7 +183,7 @@ describe("inference selection validation", () => { const log = vi.spyOn(console, "log").mockImplementation(() => {}); const helpers = createInferenceSelectionValidationHelpers({ isNonInteractive: () => false, - agentProductName: () => "Hermes", + agentProductName: () => "OpenClaw", getCredential: () => "test-key", probeAnthropicEndpoint, promptValidationRecovery: vi.fn(async () => "selection" as const), @@ -157,7 +208,7 @@ describe("inference selection validation", () => { } }); - it("routes a malformed-streaming probe failure through validation recovery (#6289)", async () => { + it("keeps rejecting malformed native Anthropic streams for OpenClaw (#6289)", async () => { const probeAnthropicEndpoint = vi.fn(() => ({ ok: false, message: @@ -166,7 +217,7 @@ describe("inference selection validation", () => { failures: [ { name: "Anthropic Messages API (streaming)", - httpStatus: 0, + httpStatus: 200, curlStatus: 0, message: "duplicate message_start", diagnosticCodes: ["anthropic-streaming-duplicate-message-start"], @@ -177,7 +228,7 @@ describe("inference selection validation", () => { const error = vi.spyOn(console, "error").mockImplementation(() => {}); const helpers = createInferenceSelectionValidationHelpers({ isNonInteractive: () => false, - agentProductName: () => "Hermes", + agentProductName: () => "OpenClaw", getCredential: () => "test-key", probeAnthropicEndpoint, promptValidationRecovery, diff --git a/src/lib/onboard/inference-selection-validation.ts b/src/lib/onboard/inference-selection-validation.ts index 9a226e85de4..b9e032a428d 100644 --- a/src/lib/onboard/inference-selection-validation.ts +++ b/src/lib/onboard/inference-selection-validation.ts @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 import { getCredential } from "../credentials/store"; +import { getCompatibleAnthropicOpenAiSurfaceBaseUrl } from "../inference/config"; const { probeAnthropicEndpoint, probeOpenAiLikeEndpoint } = require("../inference/onboard-probes") as { @@ -80,6 +81,9 @@ export interface InferenceSelectionValidationHelpers { model: string, credentialEnv: string, helpUrl?: string | null, + options?: { + intendedApi?: "anthropic-messages" | "openai-completions"; + }, ): Promise; } @@ -228,21 +232,39 @@ export function createInferenceSelectionValidationHelpers( model: string, credentialEnv: string, helpUrl: string | null = null, + options: { + intendedApi?: "anthropic-messages" | "openai-completions"; + } = {}, ): Promise { const apiKey = resolveCredential(credentialEnv); const reasoningEnabled = normalizeReasoningFlag(process.env.NEMOCLAW_REASONING) === "true"; - // Streaming validation catches Anthropic-compatible gateways whose - // non-streaming responses are valid but whose SSE streams are malformed - // (duplicate message_start, missing content deltas) — the agent runtime - // only uses the streaming path, so onboarding must exercise it (#6289). - // Reasoning-only compatible endpoints often reject streaming probes, so - // mirror the custom OpenAI-compatible path and skip streaming for them. - const probe = runAnthropicProbe(endpointUrl, model, apiKey, { - probeStreaming: !reasoningEnabled, - }); + const intendedApi = options.intendedApi ?? "anthropic-messages"; + // Validate the protocol surface that the selected agent will actually use. + // Hermes routes custom Anthropic providers through the managed OpenAI + // frontend, while native Anthropic consumers require strict SSE validation + // for duplicate/missing/out-of-order events (#6289). + const probe = + intendedApi === "openai-completions" + ? runOpenAiLikeProbe( + getCompatibleAnthropicOpenAiSurfaceBaseUrl(endpointUrl), + model, + apiKey, + { skipResponsesProbe: true }, + ) + : runAnthropicProbe(endpointUrl, model, apiKey, { + // Reasoning-only compatible endpoints often reject streaming probes, + // so mirror the custom OpenAI-compatible path and skip streaming. + probeStreaming: !reasoningEnabled, + }); if (probe.ok) { - console.log(` ${probe.label} available — ${deps.agentProductName()} will use ${probe.api}.`); - return { ok: true, api: probe.api }; + if (probe.note) { + console.log(` ℹ ${probe.note}`); + } else { + console.log( + ` ${probe.label} available — ${deps.agentProductName()} will use ${intendedApi}.`, + ); + } + return { ok: true, api: intendedApi }; } printValidationFailure(label, probe); if (deps.isNonInteractive()) { diff --git a/src/lib/onboard/machine/handlers/provider-inference.ts b/src/lib/onboard/machine/handlers/provider-inference.ts index 63fe90ac960..23d64b363af 100644 --- a/src/lib/onboard/machine/handlers/provider-inference.ts +++ b/src/lib/onboard/machine/handlers/provider-inference.ts @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -import { coerceAgentInferenceApi, resolveAgentInferenceApi } from "../../../inference/config"; +import { resolveAgentProviderInferenceApi } from "../../../inference/config"; import type { WebSearchConfig } from "../../../inference/web-search"; import type { HermesAuthMethod, Session, SessionUpdates } from "../../../state/onboard-session"; import { withInferenceTrace, withProviderSelectionTrace } from "../../tracing"; @@ -259,10 +259,11 @@ export async function handleProviderInferenceState({ // selected agent cannot safely use. Normalize the seed before the resume // shortcut so the gateway provider is revalidated and, when necessary, // re-registered on the matching protocol surface before sandbox creation. - let preferredInferenceApi = resolveAgentInferenceApi( + let preferredInferenceApi = resolveAgentProviderInferenceApi( agentName(agent), + agent, provider, - coerceAgentInferenceApi(agent, initial.preferredInferenceApi), + initial.preferredInferenceApi, ); let compatibleEndpointReasoning = initial.compatibleEndpointReasoning; let nimContainer = initial.nimContainer; @@ -408,8 +409,9 @@ export async function handleProviderInferenceState({ const selectedModel = selected.model; provider = selectedProvider; model = selectedModel; - preferredInferenceApi = resolveAgentInferenceApi( + preferredInferenceApi = resolveAgentProviderInferenceApi( agentName(agent), + agent, provider, preferredInferenceApi, ); diff --git a/src/lib/onboard/probe-diagnostics.test.ts b/src/lib/onboard/probe-diagnostics.test.ts index c864c9cae39..b5783dc8d75 100644 --- a/src/lib/onboard/probe-diagnostics.test.ts +++ b/src/lib/onboard/probe-diagnostics.test.ts @@ -49,7 +49,7 @@ describe("summarizeProbeForDisplay", () => { failures: [ { name: "Anthropic Messages API (streaming)", - httpStatus: 0, + httpStatus: 200, curlStatus: 0, message: "raw provider response with secret-key", diagnosticCodes: [ @@ -65,6 +65,25 @@ describe("summarizeProbeForDisplay", () => { expect(summary).not.toContain("provider-controlled-diagnostic"); }); + it("preserves streaming timeout recovery when a partial HTTP 200 stream times out", () => { + const summary = summarizeProbeForDisplay({ + message: "partial stream with secret-key", + failures: [ + { + name: "Anthropic Messages API (streaming)", + httpStatus: 200, + curlStatus: 28, + message: "partial stream with secret-key", + diagnosticCodes: ["anthropic-streaming-missing-message-stop"], + }, + ], + }); + + expect(summary).toBe("Anthropic Messages API (streaming): curl exit 28"); + expect(summary).not.toContain("secret-key"); + expect(summary).not.toContain("partial stream"); + }); + it("falls back to coarse message classification", () => { expect(summarizeProbeForDisplay({ message: "HTTP 404: not found for secret-key" })).toBe( "HTTP 404", diff --git a/src/lib/onboard/probe-diagnostics.ts b/src/lib/onboard/probe-diagnostics.ts index a5202ab36e6..cb96a233549 100644 --- a/src/lib/onboard/probe-diagnostics.ts +++ b/src/lib/onboard/probe-diagnostics.ts @@ -26,10 +26,13 @@ function summarizeProbeFailureForDisplay(failure: Record): stri const name = typeof failure.name === "string" ? failure.name : "probe"; const httpStatus = typeof failure.httpStatus === "number" ? failure.httpStatus : 0; const curlStatus = typeof failure.curlStatus === "number" ? failure.curlStatus : 0; - if (httpStatus > 0) return `${name}: HTTP ${httpStatus}`; + if (httpStatus > 0 && (httpStatus < 200 || httpStatus >= 300)) { + return `${name}: HTTP ${httpStatus}`; + } if (curlStatus !== 0) return `${name}: curl exit ${curlStatus}`; const diagnostics = summarizeSafeProbeDiagnostics(failure.diagnosticCodes); if (diagnostics.length > 0) return `${name}: ${diagnostics.join("; ")}`; + if (httpStatus > 0) return `${name}: HTTP ${httpStatus}`; return `${name}: no HTTP response`; } diff --git a/src/lib/onboard/setup-nim-flow.test.ts b/src/lib/onboard/setup-nim-flow.test.ts index 31693ad5a61..c9d5ee31702 100644 --- a/src/lib/onboard/setup-nim-flow.test.ts +++ b/src/lib/onboard/setup-nim-flow.test.ts @@ -295,6 +295,7 @@ describe("createSetupNim", () => { recoveredFromSandbox: true, recoveredModel: "handoff-model", sandboxName: "target-sandbox", + intendedInferenceApi: null, }); expect(recoveredRoute).toBe(recoveredRegistryRoute); state.model = args.recoveredModel; @@ -383,4 +384,35 @@ describe("createSetupNim", () => { preferredInferenceApi: "openai-completions", }); }); + + it("validates DCode custom Anthropic selections on the OpenAI surface (#6294)", async () => { + const agent = { + name: "langchain-deepagents-code", + inference: { provider_type: "openai_compatible" }, + } as AgentDefinition; + const handleRemoteProviderSelection = vi.fn( + async (args, state) => { + expect(args.intendedInferenceApi).toBe("openai-completions"); + state.model = "custom-model"; + state.provider = "compatible-anthropic-endpoint"; + state.endpointUrl = "https://compatible.example"; + state.credentialEnv = "COMPATIBLE_ANTHROPIC_API_KEY"; + state.preferredInferenceApi = args.intendedInferenceApi; + return "selected"; + }, + ); + const setupNim = createSetupNim( + makeDeps({ + isNonInteractive: () => true, + getNonInteractiveProvider: () => "anthropicCompatible", + getNonInteractiveModel: () => "custom-model", + handleRemoteProviderSelection, + }), + ); + + const result = await setupNim(null, null, agent); + + expect(handleRemoteProviderSelection).toHaveBeenCalledOnce(); + expect(result.preferredInferenceApi).toBe("openai-completions"); + }); }); diff --git a/src/lib/onboard/setup-nim-flow.ts b/src/lib/onboard/setup-nim-flow.ts index 490ef8a9922..cc122127a17 100644 --- a/src/lib/onboard/setup-nim-flow.ts +++ b/src/lib/onboard/setup-nim-flow.ts @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 import type { AgentDefinition } from "../agent/defs"; +import { resolveAgentProviderInferenceApi } from "../inference/config"; import type { VllmProfile } from "../inference/vllm"; import { isBackToSelection } from "../navigation"; import type { HermesAuthMethod } from "./hermes-auth"; @@ -32,6 +33,7 @@ export interface SetupNimRemoteSelectionArgs { recoveredFromSandbox: boolean; recoveredModel: string | null; sandboxName: string | null; + intendedInferenceApi: string | null; } export type SetupNim = ( @@ -142,6 +144,20 @@ function requireSelectedProvider( return selected; } +function resolveValidationInferenceApi( + selectedKey: string, + provider: string, + agent: AgentDefinition | null, +): string | null { + if (selectedKey !== "anthropicCompatible") return null; + return resolveAgentProviderInferenceApi( + agent?.name ?? "openclaw", + agent, + provider, + "anthropic-messages", + ); +} + function clearReasoningUnlessCompatible( provider: string, current: string | null, @@ -319,7 +335,18 @@ export function createSetupNim( nvidiaFeaturedModels, }; const result = await deps.handleRemoteProviderSelection( - { selected, requestedModel, recoveredFromSandbox, recoveredModel, sandboxName }, + { + selected, + requestedModel, + recoveredFromSandbox, + recoveredModel, + sandboxName, + intendedInferenceApi: resolveValidationInferenceApi( + selected.key, + deps.remoteProviderConfig[selected.key].providerName, + agent, + ), + }, state, recoveredRegistryRoute, ); diff --git a/src/lib/onboard/setup-nim-selection.test.ts b/src/lib/onboard/setup-nim-selection.test.ts index 89207158036..d546463da02 100644 --- a/src/lib/onboard/setup-nim-selection.test.ts +++ b/src/lib/onboard/setup-nim-selection.test.ts @@ -66,6 +66,62 @@ describe("setupNim selection state helpers", () => { }); describe("createRemoteModelValidator", () => { + it.each([ + "openai-completions", + "anthropic-messages", + ] as const)("uses the intended %s runtime API when validating custom Anthropic selections (#6289)", async (expectedApi) => { + const state = makeState(); + state.provider = "compatible-anthropic-endpoint"; + state.endpointUrl = "https://compatible.example"; + state.model = "custom-model"; + let validatedApi: string | undefined; + const { validateSelectedRemoteModel } = createRemoteModelValidator({ + OPENAI_ENDPOINT_URL: "https://default-openai.example/v1", + ANTHROPIC_ENDPOINT_URL: "https://default-anthropic.example/v1", + requireValue: (value, message) => { + if (value === null || value === undefined) throw new Error(message); + return value; + }, + isBackToSelection: (_value): _value is never => false, + validateCustomOpenAiLikeSelection: async () => ({ ok: false, retry: "selection" }), + validateCustomAnthropicSelection: async ( + _label, + _endpointUrl, + _model, + _credentialEnv, + _helpUrl, + options, + ) => { + validatedApi = options?.intendedApi; + return { ok: true, api: validatedApi ?? null }; + }, + validateAnthropicSelectionWithRetryMessage: async () => ({ + ok: false, + retry: "selection", + }), + validateOpenAiLikeSelection: async () => ({ ok: false, retry: "selection" }), + shouldRequireResponsesToolCalling: () => false, + shouldSkipResponsesProbe: () => false, + getProbeAuthMode: () => undefined, + }); + + const result = await validateSelectedRemoteModel({ + selected: { key: "anthropicCompatible" }, + remoteConfig: { + label: "Other Anthropic-compatible endpoint", + endpointUrl: "https://compatible.example", + helpUrl: null, + }, + state, + selectedCredentialEnv: "COMPATIBLE_ANTHROPIC_API_KEY", + intendedInferenceApi: expectedApi, + }); + + assert.equal(result, "selected"); + assert.equal(validatedApi, expectedApi); + assert.equal(state.preferredInferenceApi, expectedApi); + }); + it("forces custom compatible endpoints to chat completions unless the API is explicit", async () => { const state = makeState(); state.provider = "openai-compatible"; diff --git a/src/lib/onboard/setup-nim-selection.ts b/src/lib/onboard/setup-nim-selection.ts index 174ccfb0ce3..547bfc43367 100644 --- a/src/lib/onboard/setup-nim-selection.ts +++ b/src/lib/onboard/setup-nim-selection.ts @@ -126,6 +126,9 @@ type RemoteModelValidatorDeps = { model: string, credentialEnv: string, helpUrl: string | null, + options?: { + intendedApi?: "anthropic-messages" | "openai-completions"; + }, ) => Promise; validateAnthropicSelectionWithRetryMessage: ( label: string, @@ -156,6 +159,7 @@ type ValidateSelectedRemoteModelArgs = { remoteConfig: RemoteProviderConfig; state: SetupNimSelectionState; selectedCredentialEnv: string; + intendedInferenceApi?: string | null; }; function shouldRetryModel(validation: ValidationResult): boolean { @@ -167,6 +171,13 @@ function shouldRetryModel(validation: ValidationResult): boolean { ); } +function requireCustomAnthropicRuntimeApi( + value: string | null, +): "anthropic-messages" | "openai-completions" { + if (value === "anthropic-messages" || value === "openai-completions") return value; + throw new Error(`Unsupported custom Anthropic runtime API: ${String(value)}`); +} + export function createRemoteModelValidator(deps: RemoteModelValidatorDeps): { validateSelectedRemoteModel: ( args: ValidateSelectedRemoteModelArgs, @@ -178,6 +189,7 @@ export function createRemoteModelValidator(deps: RemoteModelValidatorDeps): { remoteConfig, state, selectedCredentialEnv, + intendedInferenceApi = "anthropic-messages", }) => { const selectedModel = deps.requireValue( deps.isBackToSelection(state.model) ? null : state.model, @@ -226,12 +238,14 @@ export function createRemoteModelValidator(deps: RemoteModelValidatorDeps): { } if (selected.key === "anthropicCompatible") { + const intendedApi = requireCustomAnthropicRuntimeApi(intendedInferenceApi); const validation = await deps.validateCustomAnthropicSelection( remoteConfig.label, state.endpointUrl || deps.ANTHROPIC_ENDPOINT_URL, selectedModel, selectedCredentialEnv, remoteConfig.helpUrl, + { intendedApi }, ); if (validation.ok) { state.preferredInferenceApi = validation.api; From 675f42f23f60cf25b21dd2f9ea6fc27a9d5f9298 Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 12:56:38 -0700 Subject: [PATCH 10/12] refactor(onboard): keep validation adapter compact Signed-off-by: Carlos Villela --- src/lib/onboard.ts | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/src/lib/onboard.ts b/src/lib/onboard.ts index 7ab0d83ae09..891dbe6ea40 100644 --- a/src/lib/onboard.ts +++ b/src/lib/onboard.ts @@ -3704,13 +3704,10 @@ async function handleRemoteProviderSelection(args: RemoteProviderSelectionArgs, const validationResult = state.reuseGatewayCredentialWithoutLocalKey ? "selected" - : await validateSelectedRemoteModel({ - selected, - remoteConfig, - state, - selectedCredentialEnv, - intendedInferenceApi, - }); + : await validateSelectedRemoteModel( + // biome-ignore format: keep src/lib/onboard.ts net-neutral for growth guardrail. + { selected, remoteConfig, state, selectedCredentialEnv, intendedInferenceApi }, + ); if (validationResult === "selected") break; if (validationResult === "retry-selection") return "retry-selection"; } From 5c4ca135fd0de92447f23b64931aaf97eee4f8a2 Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 12:59:26 -0700 Subject: [PATCH 11/12] test(onboard): reuse value helper in validator coverage Signed-off-by: Carlos Villela --- src/lib/onboard/setup-nim-selection.test.ts | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/lib/onboard/setup-nim-selection.test.ts b/src/lib/onboard/setup-nim-selection.test.ts index d546463da02..960ecc6cf60 100644 --- a/src/lib/onboard/setup-nim-selection.test.ts +++ b/src/lib/onboard/setup-nim-selection.test.ts @@ -5,6 +5,7 @@ import assert from "node:assert/strict"; import { describe, it } from "vitest"; +import { requireValue } from "../core/require-value"; import { applyCloudFallbackSelection, clearNimContainerBeforeRetry, @@ -78,10 +79,7 @@ describe("createRemoteModelValidator", () => { const { validateSelectedRemoteModel } = createRemoteModelValidator({ OPENAI_ENDPOINT_URL: "https://default-openai.example/v1", ANTHROPIC_ENDPOINT_URL: "https://default-anthropic.example/v1", - requireValue: (value, message) => { - if (value === null || value === undefined) throw new Error(message); - return value; - }, + requireValue, isBackToSelection: (_value): _value is never => false, validateCustomOpenAiLikeSelection: async () => ({ ok: false, retry: "selection" }), validateCustomAnthropicSelection: async ( From 60ef9768f4772b14b7b1b2da83e310f5caa29edb Mon Sep 17 00:00:00 2001 From: Carlos Villela Date: Mon, 6 Jul 2026 13:21:15 -0700 Subject: [PATCH 12/12] test(onboard): cover stale provider replacement Signed-off-by: Carlos Villela --- .../remote-openai-surface.test.ts | 195 ++++++++++++++++++ 1 file changed, 195 insertions(+) create mode 100644 src/lib/onboard/inference-providers/remote-openai-surface.test.ts diff --git a/src/lib/onboard/inference-providers/remote-openai-surface.test.ts b/src/lib/onboard/inference-providers/remote-openai-surface.test.ts new file mode 100644 index 00000000000..34256a20388 --- /dev/null +++ b/src/lib/onboard/inference-providers/remote-openai-surface.test.ts @@ -0,0 +1,195 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +import { describe, expect, it, vi } from "vitest"; + +import { setupRemoteProviderInference } from "./remote"; +import type { RemoteProviderDeps } from "./types"; + +const PROVIDER = "compatible-anthropic-endpoint"; +const MODEL = "custom-model"; +const ENDPOINT = "https://inference.example"; +const OPENAI_SURFACE = `${ENDPOINT}/v1`; +const CREDENTIAL_ENV = "COMPATIBLE_ANTHROPIC_API_KEY"; +const SANDBOX = "target-box"; +const SUCCESS = { status: 0, stdout: "", stderr: "" }; + +function makeArgs(sandboxName: string | null) { + return { + sandboxName, + model: MODEL, + provider: PROVIDER, + endpointUrl: ENDPOINT, + credentialEnv: CREDENTIAL_ENV, + preferredInferenceApi: "openai-completions", + }; +} + +function createHarness() { + const runOpenshell = vi.fn(() => SUCCESS); + const upsertProvider = vi.fn(() => ({ ok: true })); + const probeOpenAiLikeEndpoint = vi.fn(() => ({ ok: true })); + const readGatewayProviderMetadata = vi.fn(() => ({ + name: PROVIDER, + type: "anthropic", + credentialKeys: [CREDENTIAL_ENV], + configKeys: ["ANTHROPIC_BASE_URL"], + })); + const deleteGatewayProvider = vi.fn(() => ({ ok: true })); + const exitProcess = vi.fn((code: number): never => { + throw new Error(`EXIT_CALLED:${code}`); + }); + const error = vi.fn(); + const deps = { + runOpenshell, + upsertProvider, + verifyInferenceRoute: vi.fn(), + verifyOnboardInferenceSmoke: vi.fn(), + isNonInteractive: vi.fn(() => true), + registry: { updateSandbox: vi.fn() }, + exitProcess, + error, + log: vi.fn(), + REMOTE_PROVIDER_CONFIG: { + anthropicCompatible: { + label: "Other Anthropic-compatible endpoint", + providerName: PROVIDER, + providerType: "anthropic", + credentialEnv: CREDENTIAL_ENV, + endpointUrl: ENDPOINT, + helpUrl: null, + modelMode: "input", + defaultModel: MODEL, + }, + }, + hydrateCredentialEnv: vi.fn(() => "test-secret"), + promptValidationRecovery: vi.fn(async () => "selection" as const), + classifyApplyFailure: vi.fn(() => "unknown"), + LOCAL_INFERENCE_TIMEOUT_SECS: 60, + bedrockRuntimeOnboard: { + setupBedrockRuntimeInference: vi.fn(async () => ({ handled: false as const })), + }, + redact: vi.fn((value: string) => value), + compactText: vi.fn((value: string) => value.trim()), + probeOpenAiLikeEndpoint, + readGatewayProviderMetadata, + deleteGatewayProvider, + } satisfies RemoteProviderDeps; + + return { + deps, + runOpenshell, + upsertProvider, + probeOpenAiLikeEndpoint, + readGatewayProviderMetadata, + deleteGatewayProvider, + exitProcess, + error, + }; +} + +describe("custom Anthropic provider replacement on the OpenAI surface", () => { + it("probes chat completions before replacing a stale Anthropic provider as OpenAI (#6294)", async () => { + const harness = createHarness(); + + await expect(setupRemoteProviderInference(makeArgs(SANDBOX), harness.deps)).resolves.toEqual({ + done: false, + }); + + expect(harness.probeOpenAiLikeEndpoint).toHaveBeenCalledWith( + OPENAI_SURFACE, + MODEL, + "test-secret", + { skipResponsesProbe: true }, + ); + expect(harness.readGatewayProviderMetadata).toHaveBeenCalledWith( + PROVIDER, + harness.runOpenshell, + ); + expect(harness.runOpenshell).toHaveBeenNthCalledWith(1, ["provider", "delete", PROVIDER], { + ignoreError: true, + suppressOutput: true, + }); + expect(harness.probeOpenAiLikeEndpoint.mock.invocationCallOrder[0]).toBeLessThan( + harness.runOpenshell.mock.invocationCallOrder[0], + ); + expect(harness.upsertProvider).toHaveBeenCalledWith( + PROVIDER, + "openai", + CREDENTIAL_ENV, + OPENAI_SURFACE, + { [CREDENTIAL_ENV]: "test-secret" }, + ); + expect(harness.probeOpenAiLikeEndpoint.mock.invocationCallOrder[0]).toBeLessThan( + harness.upsertProvider.mock.invocationCallOrder[0], + ); + }); + + it("authorizes detach recovery only for the current sandbox (#6294)", async () => { + const harness = createHarness(); + harness.runOpenshell.mockReturnValueOnce({ + status: 1, + stdout: "", + stderr: `provider '${PROVIDER}' is attached to sandbox(es): ${SANDBOX}`, + }); + + await expect(setupRemoteProviderInference(makeArgs(SANDBOX), harness.deps)).resolves.toEqual({ + done: false, + }); + + expect(harness.deleteGatewayProvider).toHaveBeenCalledWith(PROVIDER, { + runOpenshell: harness.runOpenshell, + allowedSandboxes: [SANDBOX], + }); + expect(harness.upsertProvider).toHaveBeenCalledWith( + PROVIDER, + "openai", + CREDENTIAL_ENV, + OPENAI_SURFACE, + { [CREDENTIAL_ENV]: "test-secret" }, + ); + expect(harness.deleteGatewayProvider.mock.invocationCallOrder[0]).toBeLessThan( + harness.upsertProvider.mock.invocationCallOrder[0], + ); + }); + + it("fails closed when a foreign sandbox is attached (#6294)", async () => { + const harness = createHarness(); + harness.runOpenshell.mockReturnValueOnce({ + status: 1, + stdout: "", + stderr: `provider '${PROVIDER}' is attached to sandbox(es): ${SANDBOX}, foreign-box`, + }); + + await expect(setupRemoteProviderInference(makeArgs(SANDBOX), harness.deps)).rejects.toThrow( + "EXIT_CALLED:1", + ); + + expect(harness.exitProcess).toHaveBeenCalledWith(1); + expect(harness.error).toHaveBeenCalledWith( + expect.stringContaining("attached to other sandbox(es) (foreign-box)"), + ); + expect(harness.deleteGatewayProvider).not.toHaveBeenCalled(); + expect(harness.upsertProvider).not.toHaveBeenCalled(); + }); + + it("refuses detach recovery without a confirmed sandbox (#6294)", async () => { + const harness = createHarness(); + harness.runOpenshell.mockReturnValueOnce({ + status: 1, + stdout: "", + stderr: `provider '${PROVIDER}' is attached to sandbox(es): ${SANDBOX}`, + }); + + await expect(setupRemoteProviderInference(makeArgs(null), harness.deps)).rejects.toThrow( + "EXIT_CALLED:1", + ); + + expect(harness.exitProcess).toHaveBeenCalledWith(1); + expect(harness.error).toHaveBeenCalledWith( + expect.stringContaining("no target sandbox was confirmed"), + ); + expect(harness.deleteGatewayProvider).not.toHaveBeenCalled(); + expect(harness.upsertProvider).not.toHaveBeenCalled(); + }); +});