diff --git a/adapters/python/claude/README.md b/adapters/python/claude/README.md index a6b8ac271..bfa13847c 100644 --- a/adapters/python/claude/README.md +++ b/adapters/python/claude/README.md @@ -169,7 +169,7 @@ config.enable_relay( ``` For each Relay-enabled Claude runtime, NeMo Fabric starts one `nemo-relay` gateway, -waits for its health endpoint, and stops it with the runtime. NeMo Fabric passes the +waits up to 30 seconds for its health endpoint, and stops it with the runtime. NeMo Fabric passes the gateway URL to the connected Claude Code process through `ANTHROPIC_BASE_URL` and `NEMO_RELAY_GATEWAY_URL`, and passes the selected explicit model endpoint to the gateway as its Anthropic upstream. It also stages a runtime-scoped Claude diff --git a/adapters/python/codex/README.md b/adapters/python/codex/README.md index 3cf1de779..ba7eab1b7 100644 --- a/adapters/python/codex/README.md +++ b/adapters/python/codex/README.md @@ -154,7 +154,9 @@ Codex adapter does not provide a separate `relay` extra; its `harness` and Enable Relay with `FabricConfig.enable_relay(...)`. The adapter starts the installed `nemo-relay` CLI as a supervised sidecar; do not start the gateway -separately. +separately. Gateway readiness has a 30-second budget covering configuration +validation and required worker initialization. Startup fails on early process exit +or an expired budget, and the gateway log is retained for diagnosis. NeMo Fabric routes the selected Responses-compatible provider through the gateway and passes its explicit `base_url` to Relay as the upstream endpoint. diff --git a/adapters/python/common/README.md b/adapters/python/common/README.md index 17e537cd5..df9e8aa93 100644 --- a/adapters/python/common/README.md +++ b/adapters/python/common/README.md @@ -122,6 +122,22 @@ state during `start`. Adapter stdout is reserved for the protocol; diagnostics are redirected to stderr. A host crash or protocol timeout terminates that runtime. +## Relay Gateway Readiness + +The Claude and Codex adapters wait up to 30 seconds for the supervised Relay +gateway's HTTP health endpoint. This budget includes configuration validation +and required worker initialization. Process exit is checked before each health +probe and again before accepting a successful response. An exit during an +unsuccessful probe is detected at the next poll. A gateway that never becomes +healthy fails when the budget expires and is stopped. +The adapters retain its log for diagnosis. Local Python hosts separately enforce +a 90-second adapter startup limit. + +Relay 0.10.3 includes an +[upstream startup-memory fix](https://github.com/NVIDIA/NeMo-Relay/pull/1375). +Use that release to obtain the fix on 0.10 installations. NeMo Fabric continues to +accept the supported `>=0.9,<0.11` CLI range. + ## Relay Request Correlation In-process Relay SDK adapters that own their Agent scope can use diff --git a/adapters/python/common/src/nemo_fabric_adapters/common/relay_gateway.py b/adapters/python/common/src/nemo_fabric_adapters/common/relay_gateway.py index 45cf04faa..f0e6d5cd3 100644 --- a/adapters/python/common/src/nemo_fabric_adapters/common/relay_gateway.py +++ b/adapters/python/common/src/nemo_fabric_adapters/common/relay_gateway.py @@ -20,7 +20,10 @@ from nemo_fabric_adapters.common.relay_version import is_supported_relay_version -RELAY_HEALTH_TIMEOUT_SECONDS = 10.0 +# Readiness includes plugin preflight and worker startup before health is served. +# This operational budget is separate from Relay's inner request timeouts and +# Fabric's outer adapter-start limit. +RELAY_STARTUP_TIMEOUT_SECONDS = 30.0 RELAY_STOP_TIMEOUT_SECONDS = 5.0 RELAY_VERSION_TIMEOUT_SECONDS = 5.0 @@ -110,24 +113,39 @@ def wait_for_relay_gateway( process: subprocess.Popen[Any], health_url: str, *, - timeout: float = RELAY_HEALTH_TIMEOUT_SECONDS, + timeout: float = RELAY_STARTUP_TIMEOUT_SECONDS, ) -> None: """Wait until the Relay health endpoint succeeds or startup fails.""" deadline = time.monotonic() + timeout - while time.monotonic() < deadline: + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + break returncode = process.poll() if returncode is not None: raise RelayGatewayError( f"NeMo Relay gateway exited with status {returncode} before becoming ready" ) try: - with urllib.request.urlopen(health_url, timeout=1) as response: + with urllib.request.urlopen( + health_url, timeout=min(1.0, remaining) + ) as response: if 200 <= response.status < 300: + returncode = process.poll() + if returncode is not None: + raise RelayGatewayError( + f"NeMo Relay gateway exited with status {returncode} before becoming ready" + ) + if time.monotonic() >= deadline: + break return except (OSError, urllib.error.URLError): pass - time.sleep(0.1) + remaining = deadline - time.monotonic() + if remaining <= 0: + break + time.sleep(min(0.1, remaining)) raise RelayGatewayError(f"NeMo Relay gateway did not become ready at {health_url}") diff --git a/adapters/typescript/pi/src/relay-gateway.ts b/adapters/typescript/pi/src/relay-gateway.ts index 163bcac09..99446bf14 100644 --- a/adapters/typescript/pi/src/relay-gateway.ts +++ b/adapters/typescript/pi/src/relay-gateway.ts @@ -147,26 +147,43 @@ export async function waitForRelayGateway( const fetchRequest = options.fetch ?? fetch; const pollIntervalMs = options.pollIntervalMs ?? 100; const deadline = performance.now() + timeoutMs; - while (performance.now() < deadline) { + while (true) { + const remaining = deadline - performance.now(); + if (remaining <= 0) { + break; + } if (processExited(child)) { throw new RelayGatewayError( `NeMo Relay gateway exited with status ${child.exitCode ?? child.signalCode} before becoming ready`, ); } + let ready = false; try { const response = await fetchRequest(healthUrl, { method: "HEAD", - signal: AbortSignal.timeout(1_000), + signal: AbortSignal.timeout(Math.min(1_000, Math.ceil(remaining))), }); - const ready = response.ok; await response.body?.cancel(); - if (ready) { - return; - } + ready = response.ok; } catch { // Retry until the gateway exits or the readiness deadline expires. } - await delay(Math.min(pollIntervalMs, Math.max(0, deadline - performance.now()))); + if (ready) { + if (processExited(child)) { + throw new RelayGatewayError( + `NeMo Relay gateway exited with status ${child.exitCode ?? child.signalCode} before becoming ready`, + ); + } + if (performance.now() >= deadline) { + break; + } + return; + } + const remainingAfterProbe = deadline - performance.now(); + if (remainingAfterProbe <= 0) { + break; + } + await delay(Math.min(pollIntervalMs, remainingAfterProbe)); } throw new RelayGatewayError(`NeMo Relay gateway did not become ready at ${healthUrl}`); } diff --git a/adapters/typescript/pi/test/relay.test.mjs b/adapters/typescript/pi/test/relay.test.mjs index ee0d541d5..b77e3c459 100644 --- a/adapters/typescript/pi/test/relay.test.mjs +++ b/adapters/typescript/pi/test/relay.test.mjs @@ -844,6 +844,77 @@ test("drains unsuccessful and successful Relay health responses", async () => { assert.deepEqual(cancelled, [0, 1]); }); +test("rejects a gateway exit during a successful health probe", async (t) => { + for (const boundary of ["fetch", "body cancellation"]) { + await t.test(boundary, async () => { + const child = new MockChild(); + await assert.rejects( + waitForRelayGateway(child, "http://127.0.0.1:41000/healthz", { + async fetch() { + if (boundary === "fetch") child.exitCode = 17; + return { + ok: true, + body: { + async cancel() { + if (boundary === "body cancellation") child.exitCode = 17; + }, + }, + }; + }, + }), + /exited with status 17 before becoming ready/u, + ); + }); + } +}); + +test("rejects a successful health probe at the readiness deadline", async (t) => { + for (const boundary of ["fetch", "body cancellation"]) { + await t.test(boundary, async (subtest) => { + let now = 0; + subtest.mock.method(performance, "now", () => now); + await assert.rejects( + waitForRelayGateway(new MockChild(), "http://127.0.0.1:41000/healthz", { + timeoutMs: 25, + async fetch() { + if (boundary === "fetch") now = 25; + return { + ok: true, + body: { + async cancel() { + if (boundary === "body cancellation") now = 25; + }, + }, + }; + }, + }), + /did not become ready/u, + ); + }); + } +}); + +test("limits a health probe to the remaining readiness budget", async (t) => { + let now = 0; + t.mock.method(performance, "now", () => now); + const timeouts = []; + t.mock.method(AbortSignal, "timeout", (milliseconds) => { + timeouts.push(milliseconds); + return new AbortController().signal; + }); + await assert.rejects( + waitForRelayGateway(new MockChild(), "http://127.0.0.1:41000/healthz", { + timeoutMs: 25, + async fetch() { + now = 25; + throw new Error("still initializing"); + }, + }), + /did not become ready/u, + ); + assert.deepEqual(timeouts, [25]); +}); + test("escalates gateway shutdown from terminate to kill and is safe after exit", async () => { const mockChild = new MockChild(); mockChild.kill = function (signal) { diff --git a/examples/openshell/run-demo.sh b/examples/openshell/run-demo.sh index 56d2b534f..4fe36156e 100644 --- a/examples/openshell/run-demo.sh +++ b/examples/openshell/run-demo.sh @@ -272,9 +272,11 @@ run_consumer() { if [[ "${EXAMPLE_MODE}" == caller_owned || "${EXAMPLE_MODE}" == both ]]; then CALLER_SANDBOX_ATTEMPTED=1 + # Guard empty arrays for Bash 3.2's set -u behavior without adding empty args. "${OPENSHELL_CLI}" --gateway-endpoint "http://127.0.0.1:${GATEWAY_PORT}" \ --workspace default sandbox create --name "${SANDBOX_NAME}" --from "${RUNTIME_IMAGE}" \ - --policy "${POLICY_FILE}" "${MODEL_ENV_ARGS[@]}" "${SERVER_ENV_ARGS[@]}" \ + --policy "${POLICY_FILE}" ${MODEL_ENV_ARGS[@]+"${MODEL_ENV_ARGS[@]}"} \ + ${SERVER_ENV_ARGS[@]+"${SERVER_ENV_ARGS[@]}"} \ --detach --no-tty -- fabric-agent-ctl serve SANDBOX_JSON="$("${OPENSHELL_CLI}" --gateway-endpoint "http://127.0.0.1:${GATEWAY_PORT}" \ --workspace default sandbox get "${SANDBOX_NAME}" --output json)" diff --git a/tests/_utils/utils.py b/tests/_utils/utils.py index 9b7ca63be..aa5521ae8 100644 --- a/tests/_utils/utils.py +++ b/tests/_utils/utils.py @@ -22,11 +22,17 @@ def assert_atof_skill_selection( output: Mapping[str, Any], expected_skill: str | None ) -> None: records = atof_records(output) - tool_records = [record for record in records if record["category"] == "tool"] + tool_records = [ + record + for record in records + if record["kind"] == "scope" and record["category"] == "tool" + ] llm_end_records = [ record for record in records - if record["category"] == "llm" and record["scope_category"] == "end" + if record["kind"] == "scope" + and record["category"] == "llm" + and record["scope_category"] == "end" ] serialized_tool_records = json.dumps(tool_records).replace("\\\\", "/") serialized_skill_calls = json.dumps(tool_records + llm_end_records).replace( @@ -55,7 +61,9 @@ def assert_atof_model(output: Mapping[str, Any], expected_model: str) -> None: llm_starts = [ record for record in events - if record["category"] == "llm" and record["scope_category"] == "start" + if record["kind"] == "scope" + and record["category"] == "llm" + and record["scope_category"] == "start" ] assert llm_starts, events assert {record["data"]["content"]["model"] for record in llm_starts} == { @@ -81,7 +89,9 @@ def assert_semantic_relay_artifacts( llm_starts = [ event for event in events - if event.get("category") == "llm" and event.get("scope_category") == "start" + if event["kind"] == "scope" + and event["category"] == "llm" + and event["scope_category"] == "start" ] assert llm_starts, events assert all( @@ -92,7 +102,9 @@ def assert_semantic_relay_artifacts( llm_ends = [ event for event in events - if event.get("category") == "llm" and event.get("scope_category") == "end" + if event["kind"] == "scope" + and event["category"] == "llm" + and event["scope_category"] == "end" ] assert llm_ends, events assert any(_relay_event_total_tokens(event) > 0 for event in llm_ends) diff --git a/tests/adapters/test_adapaters_common_utils.py b/tests/adapters/test_adapaters_common_utils.py index d1805f12b..0be298052 100644 --- a/tests/adapters/test_adapaters_common_utils.py +++ b/tests/adapters/test_adapaters_common_utils.py @@ -1332,8 +1332,10 @@ async def test_relay_initializes_v4_atof_atif_config( } common_utils.validate_relay_observability(plugin_config) assert plugin.validate_exact(plugin_config)["config"]["diagnostics"] == [] + # Activation includes discovered host plugins, so compare the same layers. + expected_diagnostics = plugin.validate(plugin_config)["config"]["diagnostics"] async with plugin.activate(plugin_config) as activation: - assert activation.report["config"]["diagnostics"] == [] + assert activation.report["config"]["diagnostics"] == expected_diagnostics def test_relay_validates_all_v4_otlp_fields(): diff --git a/tests/adapters/test_adapters_common_relay_gateway.py b/tests/adapters/test_adapters_common_relay_gateway.py index c307254ae..86c23ede2 100644 --- a/tests/adapters/test_adapters_common_relay_gateway.py +++ b/tests/adapters/test_adapters_common_relay_gateway.py @@ -239,6 +239,39 @@ def test_wait_for_relay_gateway_reports_early_exit(): ) +def test_wait_for_relay_gateway_rejects_exit_during_successful_probe(monkeypatch): + process = MagicMock() + process.poll.side_effect = [None, 17] + response = MagicMock() + response.__enter__.return_value.status = 200 + monkeypatch.setattr( + relay_gateway.urllib.request, "urlopen", MagicMock(return_value=response) + ) + + with pytest.raises(relay_gateway.RelayGatewayError, match="status 17"): + relay_gateway.wait_for_relay_gateway(process, "http://127.0.0.1:43210/healthz") + + +def test_wait_for_relay_gateway_rejects_success_after_deadline( + monkeypatch, gateway_clock +): + process = MagicMock() + process.poll.return_value = None + response = MagicMock() + response.__enter__.return_value.status = 200 + + def probe_health(url, *, timeout): + gateway_clock.monotonic.return_value += 0.26 + return response + + monkeypatch.setattr(relay_gateway.urllib.request, "urlopen", probe_health) + + with pytest.raises(relay_gateway.RelayGatewayError, match="did not become ready"): + relay_gateway.wait_for_relay_gateway( + process, "http://127.0.0.1:43210/healthz", timeout=0.25 + ) + + def test_wait_for_relay_gateway_times_out(): process = MagicMock() process.poll.return_value = None @@ -271,3 +304,66 @@ def test_stop_relay_gateway_is_idempotent_for_exited_process(): process.terminate.assert_not_called() process.kill.assert_not_called() + + +@pytest.fixture(name="gateway_clock") +def gateway_clock_fixture(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + mock_clock = MagicMock(spec=relay_gateway.time) + mock_clock.monotonic.return_value = 0.0 + + def advance(seconds: float) -> None: + mock_clock.monotonic.return_value += seconds + + mock_clock.sleep.side_effect = advance + monkeypatch.setattr(relay_gateway, "time", mock_clock) + return mock_clock + + +def test_wait_for_relay_gateway_allows_preflight_before_worker_readiness( + monkeypatch, gateway_clock +): + process = MagicMock() + process.poll.return_value = None + response = MagicMock() + response.__enter__.return_value.status = 200 + + def probe_health(url, *, timeout): + gateway_clock.monotonic.return_value += timeout + if gateway_clock.monotonic() < 14.0: + raise relay_gateway.urllib.error.URLError("still initializing") + return response + + monkeypatch.setattr(relay_gateway.urllib.request, "urlopen", probe_health) + + relay_gateway.wait_for_relay_gateway(process, "http://127.0.0.1:43210/healthz") + + assert 14.0 <= gateway_clock.monotonic() < 30.0 + + +@pytest.mark.parametrize( + ("probe_elapsed", "expected_sleep"), [(0.25, None), (0.20, 0.05)] +) +def test_wait_for_relay_gateway_limits_each_probe_to_remaining_startup_budget( + monkeypatch, gateway_clock, probe_elapsed, expected_sleep +): + process = MagicMock() + process.poll.return_value = None + + def probe_health(url, *, timeout): + gateway_clock.monotonic.return_value += probe_elapsed + raise relay_gateway.urllib.error.URLError("not ready") + + mock_probe = MagicMock(side_effect=probe_health) + monkeypatch.setattr(relay_gateway.urllib.request, "urlopen", mock_probe) + + with pytest.raises(relay_gateway.RelayGatewayError, match="did not become ready"): + relay_gateway.wait_for_relay_gateway( + process, "http://127.0.0.1:43210/healthz", timeout=0.25 + ) + + mock_probe.assert_called_once_with("http://127.0.0.1:43210/healthz", timeout=0.25) + if expected_sleep is None: + gateway_clock.sleep.assert_not_called() + else: + gateway_clock.sleep.assert_called_once_with(pytest.approx(expected_sleep)) + assert gateway_clock.monotonic() == pytest.approx(0.25) diff --git a/tests/scripts/test_openshell_launcher.py b/tests/scripts/test_openshell_launcher.py index 317b8052d..f54157496 100644 --- a/tests/scripts/test_openshell_launcher.py +++ b/tests/scripts/test_openshell_launcher.py @@ -12,6 +12,7 @@ import json import os from pathlib import Path +import shlex import shutil import socket import subprocess @@ -164,7 +165,7 @@ def _operations(root: Path) -> list[dict]: def _run(launcher, workload, env): script, _, base_env = launcher return subprocess.run( - ["bash", str(script), str(workload), "--input", "words with spaces"], + ["/bin/bash", str(script), str(workload), "--input", "words with spaces"], env={**base_env, **env}, text=True, capture_output=True, @@ -172,7 +173,22 @@ def _run(launcher, workload, env): ) -def test_generic_workload_gets_ownership_and_consumer_args(launcher): +@pytest.mark.parametrize( + ("model_env_args", "server_env_args"), + [ + pytest.param([], None, id="default-server-env"), + pytest.param([], [], id="no-env"), + pytest.param(["--env", "MODEL_HINT=words with spaces"], [], id="no-server-env"), + pytest.param( + ["--env", "MODEL_HINT=words with spaces"], + ["--env", "SERVER_HINT=words with spaces"], + id="quoted-env", + ), + ], +) +def test_generic_workload_gets_ownership_and_consumer_args( + launcher, model_env_args, server_env_args +): _, root, _ = launcher workload = root / "third-workload" workload.mkdir() @@ -185,6 +201,12 @@ def test_generic_workload_gets_ownership_and_consumer_args(launcher): stage_workload_image() { :; } run_workload_consumer() { local mode="$1"; shift; uv run --no-sync consumer "$mode" "$@"; } """ + + f"MODEL_ENV_ARGS=({shlex.join(model_env_args)})\n" + + ( + f"SERVER_ENV_ARGS=({shlex.join(server_env_args)})\n" + if server_env_args is not None + else "" + ) ) completed = _run(launcher, hook, {"OPENSHELL_EXAMPLE_MODE": "both"}) @@ -195,6 +217,26 @@ def test_generic_workload_gets_ownership_and_consumer_args(launcher): assert all(call["args"][-2:] == ["--input", "words with spaces"] for call in calls) assert all(call["provider"].endswith("/fabric-environment-openshell") for call in calls) assert not (root / "sandbox-state.json").exists() + create = next( + operation["args"] + for operation in _operations(root) + if operation["command"] == "openshell" + and "sandbox" in operation["args"] + and "create" in operation["args"] + ) + expected_server_env = ( + server_env_args + if server_env_args is not None + else [ + "--env", + "ADAPTER_PYTHON=/opt/fabric/bin/python", + "--env", + "TMPDIR=/sandbox/tmp", + ] + ) + assert create[create.index("--policy") + 2 : create.index("--detach")] == ( + model_env_args + expected_server_env + ) @pytest.mark.parametrize("mode", ["caller_owned", "fabric_owned"]) @@ -343,7 +385,7 @@ def test_invalid_platform_or_mode_stops_before_building(launcher, environment, m "stage_workload_image() { :; }\nrun_workload_consumer() { :; }\n" ) completed = subprocess.run( - ["bash", str(script), str(workload)], + ["/bin/bash", str(script), str(workload)], env={**base_env, **environment}, text=True, capture_output=True,