From 76d89c0e78d4f06af5ff86d0ced37e50cac2d874 Mon Sep 17 00:00:00 2001 From: Chris Nighswonger Date: Mon, 1 Jun 2026 22:00:39 +0000 Subject: [PATCH 1/2] feat(orch): api_delegate MCP tool for HTTP-only providers (Grok) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cli_delegate covers providers with a local CLI binary (claude, codex, gemini). Some useful providers — xAI Grok being the immediate case — only expose an HTTP API; wrapping them as fake CLIs is more friction than value. This change adds a sibling MCP tool, api_delegate, that treats those providers natively. New module orch/api_executor.py: - stdlib-only HTTP delegation via urllib + ssl.create_default_context - PROVIDERS dict keyed by short name; each entry declares endpoint, default model, key resolver, response-extraction function - DelegationResult shape identical to executor.execute_cli so callers (MCP tool layer, DB logger, history capture) treat both paths uniformly - xAI/Grok key resolution: XAI_API_KEY_PATH (file) → ~/.llm-relay/grok.key → ~/grok.key (legacy) → XAI_API_KEY (env var) - Comprehensive error handling: unknown provider, no key, HTTP error, transport error, non-JSON response, empty choices — each returns a populated DelegationResult with explanatory error field New MCP tool api_delegate: - Signature: provider, prompt, model="", system="", timeout=120, max_tokens=4000 - Logs to delegation DB with strategy="api-direct" - Respects LLM_RELAY_HISTORY=1 for session-history capture - Returns same JSON envelope as cli_delegate cli_status and cli_probe extended to surface API providers alongside CLI providers. Each row now carries a "kind" field ("cli-binary" vs "http-api") so callers can distinguish. Tests: 16 new in tests/test_orch/test_api_executor.py covering provider listing, status, all error paths, success path with mocked urlopen, header construction, model/system pass-through, and key resolution precedence. Existing TestCliStatus tests updated to mock the API- provider surface so they stay focused on CLI behavior; added a new test_includes_api_providers asserting the cross-surface integration. Full suite: 610 pass + 1 pre-existing failure (test_codex_basic, env- sensitive, unrelated to this change). --- src/llm_relay/mcp/server.py | 149 ++++++++++++++- src/llm_relay/orch/api_executor.py | 263 ++++++++++++++++++++++++++ tests/test_mcp/test_server.py | 26 ++- tests/test_orch/test_api_executor.py | 266 +++++++++++++++++++++++++++ 4 files changed, 693 insertions(+), 11 deletions(-) create mode 100644 src/llm_relay/orch/api_executor.py create mode 100644 tests/test_orch/test_api_executor.py diff --git a/src/llm_relay/mcp/server.py b/src/llm_relay/mcp/server.py index a841bc2..a5b9940 100644 --- a/src/llm_relay/mcp/server.py +++ b/src/llm_relay/mcp/server.py @@ -17,8 +17,9 @@ "llm-relay", instructions=( "CLI orchestration tools for delegating tasks to Claude Code, " - "OpenAI Codex, and Gemini CLI. Provides smart routing, " - "usage tracking, and multi-CLI session diagnostics." + "OpenAI Codex, and Gemini CLI, plus HTTP-API delegation for " + "providers without a local CLI (currently xAI Grok). Provides " + "smart routing, usage tracking, and multi-CLI session diagnostics." ), ) @@ -129,22 +130,127 @@ def cli_delegate( }) +# ── Tool 1b: api_delegate ── + + +@mcp.tool() +def api_delegate( + provider: str, + prompt: str, + model: str = "", + system: str = "", + timeout: int = 120, + max_tokens: int = 4000, +) -> str: + """Delegate a task to an HTTP-only LLM provider (no local CLI binary). + + Mirrors cli_delegate but targets providers that expose only an HTTP API, + not a CLI tool. Currently supported: "grok" (xAI Grok via chat-completions). + + API key resolution: reads from a file path first, then env var. + For grok: ~/.llm-relay/grok.key (or XAI_API_KEY_PATH); falls back to + ~/grok.key for backward compatibility; finally to XAI_API_KEY env var. + + Args: + provider: Which provider to use ("grok") + prompt: The user-role prompt content + model: Optional model override (default: grok-4.3 for grok) + system: Optional system-role prompt to prepend + timeout: Request timeout in seconds (default 120) + max_tokens: Max completion tokens (default 4000) + """ + from llm_relay.orch.api_executor import execute_api, list_api_providers + from llm_relay.orch.executor import prompt_hash, prompt_preview + + if provider not in list_api_providers(): + return _json({ + "success": False, + "error": "Unknown API provider {!r}. Available: {}".format( + provider, list_api_providers() + ), + }) + + result = execute_api( + provider, + prompt, + model=model or None, + system=system or None, + timeout=timeout, + max_tokens=max_tokens, + ) + + # Log to delegation DB using the same surface as cli_delegate. + try: + from llm_relay.orch.db import get_orch_conn, log_delegation + conn = get_orch_conn() + log_delegation( + conn, + cli_id=result.cli_id, + auth_method=result.auth_method.value, + prompt_hash=prompt_hash(prompt), + prompt_preview=prompt_preview(prompt), + model=model or None, + working_dir=None, + success=result.success, + exit_code=result.exit_code, + duration_ms=result.duration_ms, + output_chars=len(result.output), + error=result.error, + strategy="api-direct", + ) + conn.close() + except Exception: + logger.debug("Failed to log api_delegate", exc_info=True) + + if os.getenv("LLM_RELAY_HISTORY", "0") == "1": + try: + from llm_relay.proxy.db import get_conn as get_proxy_conn + from llm_relay.proxy.history import capture_delegation_turn + hconn = get_proxy_conn() + capture_delegation_turn( + hconn, + session_id="api-delegation-{}".format(int(time.time() * 1000)), + cli_id=result.cli_id, + prompt=prompt, + output=result.output, + model=model or None, + duration_ms=result.duration_ms, + ) + except Exception: + logger.debug("Failed to capture api_delegate history", exc_info=True) + + return _json({ + "success": result.success, + "cli_id": result.cli_id, + "output": result.output, + "error": result.error, + "duration_ms": round(result.duration_ms, 1), + "exit_code": result.exit_code, + "model_used": result.model_used, + }) + + # ── Tool 2: cli_status ── @mcp.tool() def cli_status() -> str: - """Check which CLI tools are installed and authenticated. + """Check which CLI tools and API providers are installed/authenticated. Returns the status of all registered CLI tools (Claude Code, Codex, Gemini) - including installation path, authentication status, and preferred auth method. + plus HTTP-only providers wired into api_delegate (currently xAI Grok). + Each entry includes installation/authentication status and the preferred + auth method. CLI tools and API providers are distinguished by the "kind" + field ("cli-binary" vs "http-api"). """ + from llm_relay.orch.api_executor import api_provider_status, list_api_providers from llm_relay.orch.discovery import discover_all statuses = discover_all() - return _json([ + out = [ { "cli_id": s.cli_id, + "kind": "cli-binary", "binary_name": s.binary_name, "installed": s.installed, "authenticated": s.cli_authenticated, @@ -154,7 +260,23 @@ def cli_status() -> str: "usable": s.is_usable(), } for s in statuses - ]) + ] + for short_name in list_api_providers(): + st = api_provider_status(short_name) + if "error" in st: + continue + out.append({ + "cli_id": st["provider_id"], + "kind": st["kind"], + "binary_name": short_name, + "installed": True, # HTTP providers don't need a local binary + "authenticated": st["api_key_available"], + "api_key_available": st["api_key_available"], + "preferred_auth": st["auth_method"], + "version": None, + "usable": st["usable"], + }) + return _json(out) # ── Tool 3: cli_probe ── @@ -162,11 +284,14 @@ def cli_status() -> str: @mcp.tool() def cli_probe(cli: str) -> str: - """Deep probe of a specific CLI: version, auth status, default model, binary path. + """Deep probe of a specific CLI or API provider. + + Returns version, auth status, default model, and binary/endpoint path. Args: - cli: Which CLI to probe ("claude", "codex", or "gemini") + cli: Which provider to probe ("claude", "codex", "gemini", or "grok") """ + from llm_relay.orch.api_executor import api_provider_status, list_api_providers from llm_relay.orch.discovery import discover_all cli_map = {"claude": "claude-code", "codex": "openai-codex", "gemini": "gemini-cli"} @@ -176,6 +301,7 @@ def cli_probe(cli: str) -> str: if s.cli_id == cli_id or s.binary_name == cli: return _json({ "cli_id": s.cli_id, + "kind": "cli-binary", "binary_name": s.binary_name, "binary_path": s.binary_path, "installed": s.installed, @@ -187,7 +313,12 @@ def cli_probe(cli: str) -> str: "usable": s.is_usable(), }) - return _json({"error": "CLI '{}' not found in registry".format(cli)}) + if cli in list_api_providers(): + st = api_provider_status(cli) + if "error" not in st: + return _json(st) + + return _json({"error": "Provider '{}' not found in CLI or API registry".format(cli)}) # ── Tool 4: orch_delegate ── diff --git a/src/llm_relay/orch/api_executor.py b/src/llm_relay/orch/api_executor.py new file mode 100644 index 0000000..c86277b --- /dev/null +++ b/src/llm_relay/orch/api_executor.py @@ -0,0 +1,263 @@ +"""HTTP API delegation for providers without a CLI binary -- stdlib only. + +Mirrors the shape of executor.py (subprocess CLI execution) but targets +HTTP-only providers. Initial provider: xAI Grok (chat-completions API). + +Auth key is read from a file path (default ~/.llm-relay/.key) or +an environment variable, in that order. Key file content is the bearer +token literal. + +Result shape matches DelegationResult so callers (MCP tool layer, DB +logger, history capture) work identically against CLI and API providers. +""" + +from __future__ import annotations + +import json +import logging +import os +import ssl +import time +import urllib.error +import urllib.request +from typing import Optional + +from llm_relay.orch.models import AuthMethod, DelegationResult + +logger = logging.getLogger(__name__) + + +# ── Provider config ────────────────────────────────────────────────────────── +# Each provider knows its endpoint, default model, key resolution, and how to +# extract the assistant text from its response. Adding a new HTTP provider +# means adding an entry here plus optional response-extraction logic. + + +def _xai_key() -> Optional[str]: + """Resolve the xAI/Grok API key. File first (XAI_API_KEY_PATH), then env.""" + path = os.environ.get( + "XAI_API_KEY_PATH", + os.path.expanduser("~/.llm-relay/grok.key"), + ) + # Backward-compat: pre-existing ~/grok.key also accepted. + if not os.path.isfile(path): + legacy = os.path.expanduser("~/grok.key") + if os.path.isfile(legacy): + path = legacy + if os.path.isfile(path): + try: + with open(path, encoding="utf-8") as f: + return f.read().strip() or None + except OSError: + logger.debug("xAI key file %s exists but is not readable", path) + env_key = os.environ.get("XAI_API_KEY", "").strip() + return env_key or None + + +def _xai_extract(payload: dict) -> str: + """Extract assistant text from an xAI chat-completions response.""" + try: + return payload["choices"][0]["message"]["content"] or "" + except (KeyError, IndexError, TypeError): + return "" + + +PROVIDERS = { + "grok": { + "provider_id": "xai-grok", + "endpoint": "https://api.x.ai/v1/chat/completions", + "default_model": "grok-4.3", + "key_resolver": _xai_key, + "extract": _xai_extract, + "auth_method": AuthMethod.API_KEY, + "api_key_name": "XAI_API_KEY (or ~/.llm-relay/grok.key)", + }, +} + + +# ── Public surface ─────────────────────────────────────────────────────────── + + +def list_api_providers() -> list[str]: + """Short names of API-only providers wired up here.""" + return list(PROVIDERS.keys()) + + +def api_provider_status(short_name: str) -> dict: + """Probe-style status for an API provider. Mirrors cli_probe output shape.""" + cfg = PROVIDERS.get(short_name) + if cfg is None: + return { + "error": "Unknown API provider: {!r}. Available: {}".format( + short_name, list_api_providers() + ) + } + key = cfg["key_resolver"]() + return { + "provider_id": cfg["provider_id"], + "kind": "http-api", + "endpoint": cfg["endpoint"], + "default_model": cfg["default_model"], + "auth_method": cfg["auth_method"].value, + "api_key_name": cfg["api_key_name"], + "api_key_available": bool(key), + "usable": bool(key), + } + + +def execute_api( + provider: str, + prompt: str, + *, + model: Optional[str] = None, + system: Optional[str] = None, + timeout: int = 120, + max_tokens: int = 4000, + temperature: float = 0.3, +) -> DelegationResult: + """Delegate a prompt to an HTTP-only provider. + + Returns a DelegationResult shaped identically to CLI execution so the + surrounding MCP/DB/history machinery can treat both paths uniformly. + """ + started = time.monotonic() + cfg = PROVIDERS.get(provider) + if cfg is None: + return DelegationResult( + cli_id="unknown-api", + auth_method=AuthMethod.NONE, + success=False, + output="", + error="Unknown API provider: {!r}".format(provider), + duration_ms=0.0, + exit_code=2, + ) + + provider_id = cfg["provider_id"] + key = cfg["key_resolver"]() + if not key: + return DelegationResult( + cli_id=provider_id, + auth_method=AuthMethod.NONE, + success=False, + output="", + error="No API key available for {}; checked {}".format( + provider, cfg["api_key_name"] + ), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=1, + ) + + messages = [] + if system: + messages.append({"role": "system", "content": system}) + messages.append({"role": "user", "content": prompt}) + + body = { + "model": model or cfg["default_model"], + "messages": messages, + "temperature": temperature, + "max_tokens": max_tokens, + } + data = json.dumps(body).encode("utf-8") + + req = urllib.request.Request( + cfg["endpoint"], + data=data, + method="POST", + headers={ + "Authorization": "Bearer {}".format(key), + "Content-Type": "application/json", + "User-Agent": "llm-relay/api-delegate (stdlib)", + }, + ) + + # Explicit TLS context so urllib uses the system trust store; corporate- + # proxy users running through an outbound MITM will need NODE_EXTRA_CA- + # equivalent setup via SSL_CERT_FILE, the same as CLI binaries do. + ctx = ssl.create_default_context() + + try: + with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: + status = resp.status + raw = resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as e: + body_excerpt = "" + try: + body_excerpt = e.read().decode("utf-8", errors="replace")[:500] + except Exception: + pass + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="HTTP {} from {}: {}".format(e.code, cfg["endpoint"], body_excerpt or e.reason), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=e.code, + model_used=body["model"], + ) + except urllib.error.URLError as e: + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="Transport error to {}: {}".format(cfg["endpoint"], e.reason), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=1, + model_used=body["model"], + ) + except Exception as e: + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="Unhandled exception: {}".format(e), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=1, + model_used=body["model"], + ) + + try: + payload = json.loads(raw) + except json.JSONDecodeError as e: + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="Provider returned non-JSON (HTTP {}): {}: {}".format(status, e, raw[:200]), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=status, + model_used=body["model"], + ) + + output = cfg["extract"](payload) + duration_ms = round((time.monotonic() - started) * 1000.0, 1) + + if not output: + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="Provider returned no assistant text. Raw payload keys: {}".format( + sorted(payload.keys()) if isinstance(payload, dict) else type(payload).__name__ + ), + duration_ms=duration_ms, + exit_code=status, + model_used=body["model"], + ) + + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=True, + output=output, + error=None, + duration_ms=duration_ms, + exit_code=status, + model_used=body["model"], + ) diff --git a/tests/test_mcp/test_server.py b/tests/test_mcp/test_server.py index 0a52131..8fed193 100644 --- a/tests/test_mcp/test_server.py +++ b/tests/test_mcp/test_server.py @@ -32,19 +32,41 @@ def _make_statuses(): class TestCliStatus: @patch("llm_relay.orch.discovery.discover_all", return_value=_make_statuses()) def test_returns_all_clis(self, mock_discover): + # cli_status surfaces both CLI providers (from discover_all) and + # API providers (from api_executor.list_api_providers). Patch the + # API surface to "no providers" so this test stays focused on CLI. from llm_relay.mcp.server import cli_status - result = json.loads(cli_status()) + with patch("llm_relay.orch.api_executor.list_api_providers", return_value=[]): + result = json.loads(cli_status()) assert len(result) == 3 assert result[0]["cli_id"] == "claude-code" + assert result[0]["kind"] == "cli-binary" assert result[0]["usable"] is True assert result[0]["version"] == "2.1.91" @patch("llm_relay.orch.discovery.discover_all", return_value=[]) def test_empty_when_no_clis(self, mock_discover): + # As above: scope to "no CLI providers AND no API providers" so + # the empty case stays empty. from llm_relay.mcp.server import cli_status - result = json.loads(cli_status()) + with patch("llm_relay.orch.api_executor.list_api_providers", return_value=[]): + result = json.loads(cli_status()) assert result == [] + @patch("llm_relay.orch.discovery.discover_all", return_value=_make_statuses()) + def test_includes_api_providers(self, mock_discover, tmp_path, monkeypatch): + # When an API provider has a usable key, cli_status appends it. + from llm_relay.mcp.server import cli_status + key_path = tmp_path / "grok.key" + key_path.write_text("xai-test", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + result = json.loads(cli_status()) + kinds = [r["kind"] for r in result] + assert "cli-binary" in kinds + assert "http-api" in kinds + api_rows = [r for r in result if r["kind"] == "http-api"] + assert any(r["cli_id"] == "xai-grok" for r in api_rows) + class TestCliProbe: @patch("llm_relay.orch.discovery.discover_all", return_value=_make_statuses()) diff --git a/tests/test_orch/test_api_executor.py b/tests/test_orch/test_api_executor.py new file mode 100644 index 0000000..f5f132b --- /dev/null +++ b/tests/test_orch/test_api_executor.py @@ -0,0 +1,266 @@ +"""Tests for orch/api_executor.py — HTTP API delegation for providers without a CLI.""" + +from __future__ import annotations + +import json +import os +import tempfile +import urllib.error +from unittest.mock import patch + +import pytest + +from llm_relay.orch.api_executor import ( + PROVIDERS, + api_provider_status, + execute_api, + list_api_providers, +) +from llm_relay.orch.models import AuthMethod + + +# ── list / status ──────────────────────────────────────────────────────────── + + +def test_list_api_providers_includes_grok(): + assert "grok" in list_api_providers() + + +def test_api_provider_status_unknown_returns_error(): + st = api_provider_status("does-not-exist") + assert "error" in st + + +def test_api_provider_status_grok_shape(monkeypatch): + # Force no key resolved so usable=False, but the shape should still be complete. + monkeypatch.setenv("XAI_API_KEY_PATH", "/nonexistent/path") + monkeypatch.delenv("XAI_API_KEY", raising=False) + # also point legacy fallback at nothing + with patch("os.path.expanduser", side_effect=lambda p: "/nonexistent/path/legacy" if "~/grok.key" in p else p): + st = api_provider_status("grok") + assert st["provider_id"] == "xai-grok" + assert st["kind"] == "http-api" + assert st["endpoint"].startswith("https://") + assert st["default_model"] == "grok-4.3" + assert st["auth_method"] == "api_key" + assert "usable" in st + + +def test_api_provider_status_grok_with_keyfile(tmp_path, monkeypatch): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-test-token-1234567890\n", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + st = api_provider_status("grok") + assert st["api_key_available"] is True + assert st["usable"] is True + + +# ── execute_api: error paths ───────────────────────────────────────────────── + + +def test_execute_api_unknown_provider(): + result = execute_api("nope", "hello") + assert result.success is False + assert result.cli_id == "unknown-api" + assert result.auth_method == AuthMethod.NONE + assert "Unknown API provider" in result.error + assert result.exit_code == 2 + + +def test_execute_api_no_key(monkeypatch, tmp_path): + monkeypatch.setenv("XAI_API_KEY_PATH", str(tmp_path / "missing-key")) + monkeypatch.delenv("XAI_API_KEY", raising=False) + # also redirect legacy ~/grok.key lookup + with patch("os.path.expanduser", side_effect=lambda p: str(tmp_path / "no-legacy") if "~/grok.key" in p else p): + result = execute_api("grok", "hello") + assert result.success is False + assert "No API key available" in result.error + assert result.exit_code == 1 + + +# ── execute_api: success path (urllib mocked) ──────────────────────────────── + + +def _mock_response(payload: dict, status: int = 200): + """Build a context-manager mock for urlopen()'s return value.""" + class _Resp: + def __init__(self, body, code): + self._body = body + self.status = code + + def read(self): + return self._body + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + return _Resp(json.dumps(payload).encode("utf-8"), status) + + +def test_execute_api_success(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-token\n", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + response_payload = { + "id": "x123", + "object": "chat.completion", + "choices": [ + {"index": 0, "message": {"role": "assistant", "content": "Hi from Grok"}} + ], + "usage": {"prompt_tokens": 5, "completion_tokens": 4, "total_tokens": 9}, + } + + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + return_value=_mock_response(response_payload), + ): + result = execute_api("grok", "hello there") + + assert result.success is True + assert result.cli_id == "xai-grok" + assert result.auth_method == AuthMethod.API_KEY + assert result.output == "Hi from Grok" + assert result.exit_code == 200 + assert result.model_used == "grok-4.3" + + +def test_execute_api_success_with_model_and_system(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-token", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + captured = {} + + def fake_urlopen(req, timeout=None, context=None): + captured["url"] = req.full_url + captured["body"] = json.loads(req.data.decode("utf-8")) + captured["auth"] = req.get_header("Authorization") + return _mock_response( + {"choices": [{"message": {"content": "Reviewed."}}]}, 200 + ) + + with patch("llm_relay.orch.api_executor.urllib.request.urlopen", side_effect=fake_urlopen): + result = execute_api("grok", "review this", model="grok-4.20-0309-reasoning", system="You are a reviewer.") + + assert result.success is True + assert captured["body"]["model"] == "grok-4.20-0309-reasoning" + assert captured["body"]["messages"][0] == {"role": "system", "content": "You are a reviewer."} + assert captured["body"]["messages"][1] == {"role": "user", "content": "review this"} + assert captured["auth"] == "Bearer xai-token" + + +def test_execute_api_http_error(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-bad", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + # urllib.error.HTTPError expects (url, code, msg, hdrs, fp); we synthesize one. + import io + err = urllib.error.HTTPError( + url="https://api.x.ai/v1/chat/completions", + code=401, + msg="Unauthorized", + hdrs=None, + fp=io.BytesIO(b'{"error":"invalid api key"}'), + ) + with patch("llm_relay.orch.api_executor.urllib.request.urlopen", side_effect=err): + result = execute_api("grok", "hello") + + assert result.success is False + assert result.exit_code == 401 + assert "HTTP 401" in result.error + assert "invalid api key" in result.error + + +def test_execute_api_transport_error(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-token", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + side_effect=urllib.error.URLError("network down"), + ): + result = execute_api("grok", "hello") + + assert result.success is False + assert "Transport error" in result.error + assert result.exit_code == 1 + + +def test_execute_api_non_json_response(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-token", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + class _PlainResp: + status = 200 + def read(self): + return b"upstream broken" + def __enter__(self): + return self + def __exit__(self, *a): + return False + + with patch("llm_relay.orch.api_executor.urllib.request.urlopen", return_value=_PlainResp()): + result = execute_api("grok", "hello") + + assert result.success is False + assert "non-JSON" in result.error + + +def test_execute_api_empty_choices(monkeypatch, tmp_path): + key_path = tmp_path / "grok.key" + key_path.write_text("xai-token", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + return_value=_mock_response({"choices": []}), + ): + result = execute_api("grok", "hello") + + assert result.success is False + assert "no assistant text" in result.error + + +# ── key resolution: file vs env vs legacy ──────────────────────────────────── + + +def test_xai_key_prefers_explicit_file_path(monkeypatch, tmp_path): + key_path = tmp_path / "explicit.key" + key_path.write_text("from-explicit-file", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + monkeypatch.setenv("XAI_API_KEY", "from-env-should-not-win") + + from llm_relay.orch.api_executor import _xai_key + assert _xai_key() == "from-explicit-file" + + +def test_xai_key_falls_back_to_env(monkeypatch, tmp_path): + monkeypatch.setenv("XAI_API_KEY_PATH", str(tmp_path / "no-file")) + monkeypatch.setenv("XAI_API_KEY", "from-env-fallback") + with patch("os.path.expanduser", side_effect=lambda p: str(tmp_path / "no-legacy") if "~/grok.key" in p else p): + from llm_relay.orch.api_executor import _xai_key + assert _xai_key() == "from-env-fallback" + + +def test_xai_key_returns_none_when_neither_set(monkeypatch, tmp_path): + monkeypatch.setenv("XAI_API_KEY_PATH", str(tmp_path / "no-file")) + monkeypatch.delenv("XAI_API_KEY", raising=False) + with patch("os.path.expanduser", side_effect=lambda p: str(tmp_path / "no-legacy") if "~/grok.key" in p else p): + from llm_relay.orch.api_executor import _xai_key + assert _xai_key() is None + + +def test_xai_key_strips_whitespace(monkeypatch, tmp_path): + key_path = tmp_path / "g.key" + key_path.write_text(" xai-with-padding \n\n", encoding="utf-8") + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + from llm_relay.orch.api_executor import _xai_key + assert _xai_key() == "xai-with-padding" From 6e4f436f87ebbe86e6c88d88955da89753d70b3d Mon Sep 17 00:00:00 2001 From: Chris Nighswonger Date: Mon, 1 Jun 2026 22:37:13 +0000 Subject: [PATCH 2/2] fix(orch): tighten api_delegate per Grok + Codex review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Round-trip review found three real bugs and three more on the second pass. All addressed: Grok review (3 fixes): - HTTPS scheme guard: refuse non-HTTPS endpoints before sending the bearer token. The PROVIDERS dict is in-tree so this is defense against future misedits, not external input. Guard fires before urlopen() is reached. - Key file permission gating: _read_key_file_if_safe() requires POSIX mode 0600 or stricter (no group / no other access). Loose-permission files are silently skipped with a debug log; resolver falls through to env var. Windows is exempt (st_mode bits don't carry equivalent meaning). - Payload-failure exit codes: JSON-parse failures and empty-extract failures previously returned exit_code=status (200) on a 2xx HTTP response, looking like success to int-checking callers. Now both use PAYLOAD_FAILURE_EXIT (502 "bad gateway") sentinel. Codex review (3 fixes): - exit_code=0 on success: Grok's payload-failure fix tightened error paths but left the success path returning exit_code=status (200). proxy/composition.py:814 explicitly checks `exit_code == 0`; the CLI sister module executor.py returns proc.returncode (0 on clean exit). api_executor now matches the convention. - Wrap key_resolver() in try/except: provider-supplied callable that raises previously bubbled. Now returns a clean DelegationResult. - Wrap extract() in try/except: same fix for the response-extraction hook. - Remove unused imports: os, tempfile, pytest, top-level PROVIDERS (each test that needs PROVIDERS imports it locally). Plus: capture model_used into a local variable before the request so exception handlers don't depend on body[] still being well-formed. Tests: 16 → 24 (+ HTTPS-guard, perm gating ×3, payload-failure exit ×2, key_resolver-raises, extract-raises). Full suite 616 → 618 pass. Live ~/grok.key chmod'd to 0600 (was 0664). --- src/llm_relay/orch/api_executor.py | 153 ++++++++++++++++---- tests/test_orch/test_api_executor.py | 199 +++++++++++++++++++++++++-- 2 files changed, 315 insertions(+), 37 deletions(-) diff --git a/src/llm_relay/orch/api_executor.py b/src/llm_relay/orch/api_executor.py index c86277b..f49789e 100644 --- a/src/llm_relay/orch/api_executor.py +++ b/src/llm_relay/orch/api_executor.py @@ -33,23 +33,60 @@ # means adding an entry here plus optional response-extraction logic. +def _read_key_file_if_safe(path: str) -> Optional[str]: + """Read a key file iff its mode is private (owner-only access). + + On POSIX systems, key files MUST NOT be readable by group or other + (mode bits 0o077 must be zero). A key file with looser permissions + is silently skipped with a debug log — the caller then falls through + to the env-var path. On Windows, where st_mode bits don't carry the + same meaning, mode checking is skipped. + + Returns the stripped key contents, or None if the file is missing, + has insecure permissions, or is unreadable. + """ + if not os.path.isfile(path): + return None + if os.name == "posix": + try: + mode = os.stat(path).st_mode & 0o777 + if mode & 0o077: + logger.debug( + "xAI key file %s has insecure mode %o (must be 0600 or stricter); skipping", + path, mode, + ) + return None + except OSError: + logger.debug("xAI key file %s could not be stat'd; skipping", path) + return None + try: + with open(path, encoding="utf-8") as f: + return f.read().strip() or None + except OSError: + logger.debug("xAI key file %s exists but is not readable", path) + return None + + def _xai_key() -> Optional[str]: - """Resolve the xAI/Grok API key. File first (XAI_API_KEY_PATH), then env.""" + """Resolve the xAI/Grok API key. File first (XAI_API_KEY_PATH), then env. + + Key files are accepted only when their POSIX mode is 0600 or stricter + (see _read_key_file_if_safe). Loose-permission files are silently + skipped so a misconfigured key never leaks into outbound requests. + """ path = os.environ.get( "XAI_API_KEY_PATH", os.path.expanduser("~/.llm-relay/grok.key"), ) - # Backward-compat: pre-existing ~/grok.key also accepted. - if not os.path.isfile(path): - legacy = os.path.expanduser("~/grok.key") - if os.path.isfile(legacy): - path = legacy - if os.path.isfile(path): - try: - with open(path, encoding="utf-8") as f: - return f.read().strip() or None - except OSError: - logger.debug("xAI key file %s exists but is not readable", path) + key = _read_key_file_if_safe(path) + if key: + return key + # Backward-compat: pre-existing ~/grok.key also accepted (same perm check). + legacy = os.path.expanduser("~/grok.key") + if legacy != path: + key = _read_key_file_if_safe(legacy) + if key: + return key env_key = os.environ.get("XAI_API_KEY", "").strip() return env_key or None @@ -134,7 +171,38 @@ def execute_api( ) provider_id = cfg["provider_id"] - key = cfg["key_resolver"]() + + # Endpoint scheme guard: bearer-token-bearing requests MUST go over HTTPS. + # The PROVIDERS dict is in-tree so a misconfigured http:// endpoint here + # would be a code bug, not external input — but a one-line guard prevents + # the worst-case credential leak if someone ever adds or edits an entry + # without noticing. + endpoint = cfg["endpoint"] + if not endpoint.startswith("https://"): + return DelegationResult( + cli_id=provider_id, + auth_method=AuthMethod.NONE, + success=False, + output="", + error="Refusing non-HTTPS endpoint for bearer-token request: {}".format(endpoint), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=2, + ) + + # Provider-supplied hook; defensively wrap in case a future provider's + # key_resolver raises (e.g., environment lookup that misuses the API). + try: + key = cfg["key_resolver"]() + except Exception as e: + return DelegationResult( + cli_id=provider_id, + auth_method=AuthMethod.NONE, + success=False, + output="", + error="API key resolution failed for {}: {}".format(provider, e), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=1, + ) if not key: return DelegationResult( cli_id=provider_id, @@ -153,8 +221,11 @@ def execute_api( messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) + # Capture model_used early so all exception branches can reference it + # without depending on body[] still being well-formed at exception time. + model_used = model or cfg["default_model"] body = { - "model": model or cfg["default_model"], + "model": model_used, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, @@ -162,7 +233,7 @@ def execute_api( data = json.dumps(body).encode("utf-8") req = urllib.request.Request( - cfg["endpoint"], + endpoint, data=data, method="POST", headers={ @@ -181,6 +252,7 @@ def execute_api( with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp: status = resp.status raw = resp.read().decode("utf-8", errors="replace") + content_type = resp.headers.get("Content-Type", "") except urllib.error.HTTPError as e: body_excerpt = "" try: @@ -192,10 +264,10 @@ def execute_api( auth_method=cfg["auth_method"], success=False, output="", - error="HTTP {} from {}: {}".format(e.code, cfg["endpoint"], body_excerpt or e.reason), + error="HTTP {} from {}: {}".format(e.code, endpoint, body_excerpt or e.reason), duration_ms=round((time.monotonic() - started) * 1000.0, 1), exit_code=e.code, - model_used=body["model"], + model_used=model_used, ) except urllib.error.URLError as e: return DelegationResult( @@ -203,10 +275,10 @@ def execute_api( auth_method=cfg["auth_method"], success=False, output="", - error="Transport error to {}: {}".format(cfg["endpoint"], e.reason), + error="Transport error to {}: {}".format(endpoint, e.reason), duration_ms=round((time.monotonic() - started) * 1000.0, 1), exit_code=1, - model_used=body["model"], + model_used=model_used, ) except Exception as e: return DelegationResult( @@ -217,9 +289,14 @@ def execute_api( error="Unhandled exception: {}".format(e), duration_ms=round((time.monotonic() - started) * 1000.0, 1), exit_code=1, - model_used=body["model"], + model_used=model_used, ) + # Sentinel for parse / extract failures that arrive over a 2xx HTTP status — + # we can't reuse `status` (200) as exit_code because callers treating + # non-zero-as-failure would then see a payload error as success. + PAYLOAD_FAILURE_EXIT = 502 # "bad gateway" — upstream returned an unusable body + try: payload = json.loads(raw) except json.JSONDecodeError as e: @@ -228,13 +305,29 @@ def execute_api( auth_method=cfg["auth_method"], success=False, output="", - error="Provider returned non-JSON (HTTP {}): {}: {}".format(status, e, raw[:200]), + error="Provider returned non-JSON (HTTP {}, Content-Type {!r}): {}: {}".format( + status, content_type, e, raw[:200] + ), duration_ms=round((time.monotonic() - started) * 1000.0, 1), - exit_code=status, - model_used=body["model"], + exit_code=PAYLOAD_FAILURE_EXIT, + model_used=model_used, ) - output = cfg["extract"](payload) + # Provider-supplied hook; defensively wrap so a misbehaving extract() + # returns a clean DelegationResult instead of bubbling an exception. + try: + output = cfg["extract"](payload) + except Exception as e: + return DelegationResult( + cli_id=provider_id, + auth_method=cfg["auth_method"], + success=False, + output="", + error="Provider response extraction failed: {}".format(e), + duration_ms=round((time.monotonic() - started) * 1000.0, 1), + exit_code=PAYLOAD_FAILURE_EXIT, + model_used=model_used, + ) duration_ms = round((time.monotonic() - started) * 1000.0, 1) if not output: @@ -247,10 +340,14 @@ def execute_api( sorted(payload.keys()) if isinstance(payload, dict) else type(payload).__name__ ), duration_ms=duration_ms, - exit_code=status, - model_used=body["model"], + exit_code=PAYLOAD_FAILURE_EXIT, + model_used=model_used, ) + # Success: exit_code follows the CLI-executor convention (0 == success); + # the HTTP status code is implicit (any 2xx that reached this branch is + # a successful round-trip). proxy/composition.py and other consumers + # rely on exit_code == 0 to flag success. return DelegationResult( cli_id=provider_id, auth_method=cfg["auth_method"], @@ -258,6 +355,6 @@ def execute_api( output=output, error=None, duration_ms=duration_ms, - exit_code=status, - model_used=body["model"], + exit_code=0, + model_used=model_used, ) diff --git a/tests/test_orch/test_api_executor.py b/tests/test_orch/test_api_executor.py index f5f132b..9cfa43b 100644 --- a/tests/test_orch/test_api_executor.py +++ b/tests/test_orch/test_api_executor.py @@ -3,15 +3,10 @@ from __future__ import annotations import json -import os -import tempfile import urllib.error from unittest.mock import patch -import pytest - from llm_relay.orch.api_executor import ( - PROVIDERS, api_provider_status, execute_api, list_api_providers, @@ -49,6 +44,7 @@ def test_api_provider_status_grok_shape(monkeypatch): def test_api_provider_status_grok_with_keyfile(tmp_path, monkeypatch): key_path = tmp_path / "grok.key" key_path.write_text("xai-test-token-1234567890\n", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) st = api_provider_status("grok") assert st["api_key_available"] is True @@ -81,12 +77,13 @@ def test_execute_api_no_key(monkeypatch, tmp_path): # ── execute_api: success path (urllib mocked) ──────────────────────────────── -def _mock_response(payload: dict, status: int = 200): +def _mock_response(payload: dict, status: int = 200, content_type: str = "application/json"): """Build a context-manager mock for urlopen()'s return value.""" class _Resp: - def __init__(self, body, code): + def __init__(self, body, code, ct): self._body = body self.status = code + self.headers = {"Content-Type": ct} def read(self): return self._body @@ -97,12 +94,13 @@ def __enter__(self): def __exit__(self, *args): return False - return _Resp(json.dumps(payload).encode("utf-8"), status) + return _Resp(json.dumps(payload).encode("utf-8"), status, content_type) def test_execute_api_success(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-token\n", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) response_payload = { @@ -124,13 +122,14 @@ def test_execute_api_success(monkeypatch, tmp_path): assert result.cli_id == "xai-grok" assert result.auth_method == AuthMethod.API_KEY assert result.output == "Hi from Grok" - assert result.exit_code == 200 + assert result.exit_code == 0 # CLI-executor convention: 0 == success assert result.model_used == "grok-4.3" def test_execute_api_success_with_model_and_system(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) captured = {} @@ -156,6 +155,7 @@ def fake_urlopen(req, timeout=None, context=None): def test_execute_api_http_error(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-bad", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) # urllib.error.HTTPError expects (url, code, msg, hdrs, fp); we synthesize one. @@ -179,6 +179,7 @@ def test_execute_api_http_error(monkeypatch, tmp_path): def test_execute_api_transport_error(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) with patch( @@ -195,10 +196,12 @@ def test_execute_api_transport_error(monkeypatch, tmp_path): def test_execute_api_non_json_response(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) class _PlainResp: status = 200 + headers = {"Content-Type": "text/html"} def read(self): return b"upstream broken" def __enter__(self): @@ -216,6 +219,7 @@ def __exit__(self, *a): def test_execute_api_empty_choices(monkeypatch, tmp_path): key_path = tmp_path / "grok.key" key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) with patch( @@ -234,6 +238,7 @@ def test_execute_api_empty_choices(monkeypatch, tmp_path): def test_xai_key_prefers_explicit_file_path(monkeypatch, tmp_path): key_path = tmp_path / "explicit.key" key_path.write_text("from-explicit-file", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) monkeypatch.setenv("XAI_API_KEY", "from-env-should-not-win") @@ -260,7 +265,183 @@ def test_xai_key_returns_none_when_neither_set(monkeypatch, tmp_path): def test_xai_key_strips_whitespace(monkeypatch, tmp_path): key_path = tmp_path / "g.key" key_path.write_text(" xai-with-padding \n\n", encoding="utf-8") + key_path.chmod(0o600) monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) from llm_relay.orch.api_executor import _xai_key assert _xai_key() == "xai-with-padding" + + +# ── Permission gating (new in fix round) ───────────────────────────────────── + + +def test_xai_key_rejects_world_readable_file(monkeypatch, tmp_path): + """A 0644 (or worse) key file is silently skipped — must fall through to env.""" + key_path = tmp_path / "loose.key" + key_path.write_text("secret-key-here", encoding="utf-8") + key_path.chmod(0o644) # group + other readable + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + monkeypatch.setenv("XAI_API_KEY", "fallback-from-env") + with patch("os.path.expanduser", side_effect=lambda p: str(tmp_path / "no-legacy") if "~/grok.key" in p else p): + from llm_relay.orch.api_executor import _xai_key + # Loose-perm file is skipped; env fallback wins. + assert _xai_key() == "fallback-from-env" + + +def test_xai_key_rejects_group_readable_file(monkeypatch, tmp_path): + """0640 also fails — any group access disqualifies.""" + key_path = tmp_path / "group.key" + key_path.write_text("secret", encoding="utf-8") + key_path.chmod(0o640) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + monkeypatch.delenv("XAI_API_KEY", raising=False) + with patch("os.path.expanduser", side_effect=lambda p: str(tmp_path / "no-legacy") if "~/grok.key" in p else p): + from llm_relay.orch.api_executor import _xai_key + # No env fallback, no usable file → None. + assert _xai_key() is None + + +def test_xai_key_accepts_0600(monkeypatch, tmp_path): + """0600 is the canonical safe mode and must be accepted.""" + key_path = tmp_path / "private.key" + key_path.write_text("good-key", encoding="utf-8") + key_path.chmod(0o600) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + from llm_relay.orch.api_executor import _xai_key + assert _xai_key() == "good-key" + + +# ── HTTPS enforcement (new in fix round) ───────────────────────────────────── + + +def test_execute_api_refuses_non_https_endpoint(monkeypatch, tmp_path): + """If a provider's endpoint is http://, the request must be refused before + the bearer token can leak.""" + key_path = tmp_path / "g.key" + key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + # Inject a malformed provider config to simulate a misedited PROVIDERS entry. + from llm_relay.orch.api_executor import PROVIDERS + PROVIDERS["bad-scheme-test"] = { + **PROVIDERS["grok"], + "endpoint": "http://api.x.ai/v1/chat/completions", + } + try: + from llm_relay.orch.api_executor import execute_api + # Pre-flight: ensure urlopen is NEVER reached (the guard must fire first). + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + side_effect=AssertionError("urlopen should not be called for non-HTTPS endpoint"), + ): + result = execute_api("bad-scheme-test", "hello") + assert result.success is False + assert "non-HTTPS" in result.error + assert result.exit_code == 2 + finally: + del PROVIDERS["bad-scheme-test"] + + +# ── Exit-code-on-payload-failure (new in fix round) ────────────────────────── + + +def test_execute_api_non_json_uses_payload_failure_exit_code(monkeypatch, tmp_path): + """A 200 response with non-JSON body must NOT return exit_code=200 + (which would look like success to int-based callers). Must use the + PAYLOAD_FAILURE_EXIT sentinel.""" + key_path = tmp_path / "g.key" + key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + class _PlainResp: + status = 200 + headers = {"Content-Type": "text/html"} + def read(self): + return b"broken" + def __enter__(self): + return self + def __exit__(self, *a): + return False + + with patch("llm_relay.orch.api_executor.urllib.request.urlopen", return_value=_PlainResp()): + result = execute_api("grok", "hello") + + assert result.success is False + assert result.exit_code != 200, "exit_code=200 on payload failure looks like success to int callers" + assert result.exit_code == 502 + # Content-Type should be surfaced in the error for diagnostics. + assert "text/html" in result.error + + +def test_execute_api_extract_callable_raises(monkeypatch, tmp_path): + """If a provider's extract() callable raises, execute_api returns a clean + DelegationResult instead of bubbling the exception.""" + key_path = tmp_path / "g.key" + key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + from llm_relay.orch.api_executor import PROVIDERS + + def _broken_extract(payload): + raise ValueError("simulated extract failure") + + PROVIDERS["broken-extract-test"] = { + **PROVIDERS["grok"], + "provider_id": "broken-extract-test-id", + "extract": _broken_extract, + } + try: + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + return_value=_mock_response({"choices": [{"message": {"content": "x"}}]}), + ): + result = execute_api("broken-extract-test", "hello") + assert result.success is False + assert "extraction failed" in result.error + assert "simulated extract failure" in result.error + assert result.exit_code == 502 + finally: + del PROVIDERS["broken-extract-test"] + + +def test_execute_api_key_resolver_raises(monkeypatch, tmp_path): + """If a provider's key_resolver() callable raises, execute_api returns a + clean DelegationResult.""" + from llm_relay.orch.api_executor import PROVIDERS + + def _broken_resolver(): + raise RuntimeError("simulated resolver failure") + + PROVIDERS["broken-resolver-test"] = { + **PROVIDERS["grok"], + "provider_id": "broken-resolver-test-id", + "key_resolver": _broken_resolver, + } + try: + result = execute_api("broken-resolver-test", "hello") + assert result.success is False + assert "key resolution failed" in result.error + assert "simulated resolver failure" in result.error + assert result.exit_code == 1 + finally: + del PROVIDERS["broken-resolver-test"] + + +def test_execute_api_empty_choices_uses_payload_failure_exit_code(monkeypatch, tmp_path): + key_path = tmp_path / "g.key" + key_path.write_text("xai-token", encoding="utf-8") + key_path.chmod(0o600) + monkeypatch.setenv("XAI_API_KEY_PATH", str(key_path)) + + with patch( + "llm_relay.orch.api_executor.urllib.request.urlopen", + return_value=_mock_response({"choices": []}, status=200), + ): + result = execute_api("grok", "hello") + + assert result.success is False + assert result.exit_code == 502 + assert result.exit_code != 200