From 04c44ab87c7940109079f8b361720dabf9402415 Mon Sep 17 00:00:00 2001 From: storyicon Date: Sun, 2 Aug 2026 14:09:20 +0800 Subject: [PATCH] feat(harness): add Grok Build CLI provider Introduce a first-class harness provider for the local Grok Build CLI so AgentField can drive Grok headless sessions the same way it drives Codex, OpenCode, Gemini, and Claude Code. The provider wraps the Grok CLI with a PTY (via script) because plain pipes currently fail with "Device not configured", feeds prompts through --prompt-file to avoid argv limits, and normalizes missing token usage to zero so metrics aggregation does not crash on incomplete usage objects. --- .../agentfield/harness/_availability.py | 7 + sdk/python/agentfield/harness/_runner.py | 11 +- .../agentfield/harness/providers/_factory.py | 6 +- .../agentfield/harness/providers/grok.py | 302 ++++++++++++++++++ sdk/python/agentfield/types.py | 5 +- sdk/python/tests/test_harness_factory.py | 8 + .../tests/test_harness_grok_provider.py | 44 +++ 7 files changed, 377 insertions(+), 6 deletions(-) create mode 100644 sdk/python/agentfield/harness/providers/grok.py create mode 100644 sdk/python/tests/test_harness_grok_provider.py diff --git a/sdk/python/agentfield/harness/_availability.py b/sdk/python/agentfield/harness/_availability.py index 43319b00c..069cbe768 100644 --- a/sdk/python/agentfield/harness/_availability.py +++ b/sdk/python/agentfield/harness/_availability.py @@ -33,6 +33,13 @@ class ProviderSpec: install_command="curl -fsSL https://opencode.ai/install | bash", auth_env_vars=(), ), + "grok": ProviderSpec( + binary="grok", + version_args=("--version",), + # Grok Build CLI is distributed by xAI; install path may vary by platform. + install_command="install the Grok Build CLI and run `grok login`", + auth_env_vars=("XAI_API_KEY",), + ), } # Providers without a PROVIDER_SPECS entry (claude-code today, or any future diff --git a/sdk/python/agentfield/harness/_runner.py b/sdk/python/agentfield/harness/_runner.py index 4ba73bed7..168fd03e0 100644 --- a/sdk/python/agentfield/harness/_runner.py +++ b/sdk/python/agentfield/harness/_runner.py @@ -174,6 +174,7 @@ def _resolve_options( "codex_bin", "gemini_bin", "opencode_bin", + "grok_bin", "schema_max_retries", "schema_mode", ]: @@ -209,10 +210,12 @@ def _accumulate_metrics( if raw.metrics.session_id: session_id = raw.metrics.session_id all_messages.extend(raw.messages) - tokens["input_tokens"] += raw.metrics.input_tokens - tokens["output_tokens"] += raw.metrics.output_tokens - tokens["cache_read_tokens"] += raw.metrics.cache_read_tokens - tokens["cache_creation_tokens"] += raw.metrics.cache_creation_tokens + # Coerce None → 0: some CLI providers omit token fields and previously + # leaked None into Metrics, crashing aggregation with int += None. + tokens["input_tokens"] += raw.metrics.input_tokens or 0 + tokens["output_tokens"] += raw.metrics.output_tokens or 0 + tokens["cache_read_tokens"] += raw.metrics.cache_read_tokens or 0 + tokens["cache_creation_tokens"] += raw.metrics.cache_creation_tokens or 0 if raw.metrics.model and not tokens["model"]: tokens["model"] = raw.metrics.model diff --git a/sdk/python/agentfield/harness/providers/_factory.py b/sdk/python/agentfield/harness/providers/_factory.py index a7d364605..5b0ad05dd 100644 --- a/sdk/python/agentfield/harness/providers/_factory.py +++ b/sdk/python/agentfield/harness/providers/_factory.py @@ -6,7 +6,7 @@ from agentfield.harness.providers._base import HarnessProvider from agentfield.types import HarnessConfig -SUPPORTED_PROVIDERS = {"claude-code", "codex", "gemini", "opencode"} +SUPPORTED_PROVIDERS = {"claude-code", "codex", "gemini", "opencode", "grok"} def build_provider(config: "HarnessConfig") -> "HarnessProvider": @@ -34,4 +34,8 @@ def build_provider(config: "HarnessConfig") -> "HarnessProvider": return OpenCodeProvider( bin_path=getattr(config, "opencode_bin", "opencode"), ) + if provider_name == "grok": + from agentfield.harness.providers.grok import GrokProvider + + return GrokProvider(bin_path=getattr(config, "grok_bin", "grok")) raise NotImplementedError(f"Provider {provider_name!r} is not yet implemented.") diff --git a/sdk/python/agentfield/harness/providers/grok.py b/sdk/python/agentfield/harness/providers/grok.py new file mode 100644 index 000000000..14db40228 --- /dev/null +++ b/sdk/python/agentfield/harness/providers/grok.py @@ -0,0 +1,302 @@ +"""Grok Build CLI provider (local `grok` binary). + +Invokes headless multi-turn Grok agent sessions via: + + script -q /dev/null grok --output-format json --prompt-file … … + +``script`` allocates a PTY: the Grok CLI currently errors with +``Device not configured (os error 6)`` when started with plain pipes. +""" + +from __future__ import annotations + +import json +import os +import re +import shutil +import tempfile +import time +from typing import Any, Dict, List, Optional + +from agentfield.harness._availability import ensure_cli_available, provider_unavailable +from agentfield.harness._cli import resolve_model_and_variant, run_cli, strip_ansi +from agentfield.harness._result import FailureType, Metrics, RawResult + +_JSON_OBJECT_RE = re.compile(r"\{[\s\S]*\}\s*$") + + +def _permission_mode(options: dict[str, object]) -> str: + mode = options.get("permission_mode") + if mode == "plan": + return "plan" + if mode in {"auto", "acceptEdits", "bypassPermissions", "dontAsk", "default"}: + # Harness runs are unattended; prefer non-interactive approvals. + if mode == "plan": + return "plan" + if mode in {"auto", "acceptEdits", "bypassPermissions"}: + return "bypassPermissions" if mode in {"auto", "bypassPermissions"} else "acceptEdits" + return str(mode) + return "bypassPermissions" + + +def _extract_json_payload(stdout: str) -> Optional[dict[str, Any]]: + """Parse Grok ``--output-format json`` payload, tolerating script PTY noise.""" + cleaned = strip_ansi(stdout or "") + # script(1) may prefix with control characters (e.g. ^D). + cleaned = cleaned.replace("\x04", "").strip() + if not cleaned: + return None + try: + data = json.loads(cleaned) + return data if isinstance(data, dict) else None + except json.JSONDecodeError: + pass + match = _JSON_OBJECT_RE.search(cleaned) + if not match: + return None + try: + data = json.loads(match.group(0)) + return data if isinstance(data, dict) else None + except json.JSONDecodeError: + return None + + +def _result_text(payload: dict[str, Any]) -> Optional[str]: + for key in ("text", "result", "message", "content", "output"): + value = payload.get(key) + if isinstance(value, str) and value.strip(): + return value + return None + + +def _usage_metrics(payload: dict[str, Any], model: Optional[str]) -> Metrics: + usage = payload.get("usage") if isinstance(payload.get("usage"), dict) else {} + model_usage = payload.get("modelUsage") if isinstance(payload.get("modelUsage"), dict) else {} + # Prefer first modelUsage entry for model id / token totals when present. + resolved_model = model + first_model_usage: dict[str, Any] = {} + if model_usage: + first_key = next(iter(model_usage.keys()), None) + if not resolved_model and first_key is not None: + resolved_model = first_key + if first_key is not None and isinstance(model_usage.get(first_key), dict): + first_model_usage = model_usage[first_key] + + def _int(value: object, default: int = 0) -> int: + # Token fields on Metrics are non-optional ints. Missing usage must be + # 0 — returning None breaks _accumulate_metrics (int += None). + if isinstance(value, bool): + return default + if isinstance(value, (int, float)): + return int(value) + return default + + def _token(*keys: str) -> int: + for key in keys: + if key in usage: + return _int(usage.get(key)) + if key in first_model_usage: + return _int(first_model_usage.get(key)) + return 0 + + cost = payload.get("total_cost_usd") + if not isinstance(cost, (int, float)) or isinstance(cost, bool): + cost = None + else: + cost = float(cost) + + turns = payload.get("num_turns") + if not isinstance(turns, int): + turns = 0 + + session_id = payload.get("sessionId") or payload.get("session_id") or "" + return Metrics( + num_turns=turns, + total_cost_usd=cost, + session_id=str(session_id) if session_id else "", + input_tokens=_token("input_tokens", "prompt_tokens"), + output_tokens=_token("output_tokens", "completion_tokens"), + cache_read_tokens=_token( + "cache_read_input_tokens", "cache_read_tokens", "cached_input_tokens" + ), + cache_creation_tokens=_token( + "cache_creation_input_tokens", "cache_creation_tokens" + ), + model=str(resolved_model) if resolved_model else model, + ) + + +class GrokProvider: + """Grok Build CLI provider. Invokes local ``grok`` headless sessions.""" + + def __init__(self, bin_path: str = "grok"): + self._bin = bin_path + + async def execute(self, prompt: str, options: dict[str, object]) -> RawResult: + ensure_cli_available("grok", self._bin) + + root = options.get("project_dir") or options.get("cwd") + if not isinstance(root, str) or not root.strip(): + root = os.getcwd() + + model_value, variant_value = resolve_model_and_variant(options) + max_turns = options.get("max_turns") + if not isinstance(max_turns, int) or max_turns <= 0: + max_turns = 30 + + permission = _permission_mode(options) + + # Long prompts exceed argv limits; always feed via --prompt-file. + prompt_path: Optional[str] = None + try: + with tempfile.NamedTemporaryFile( + "w", + encoding="utf-8", + prefix="agentfield-grok-prompt-", + suffix=".txt", + delete=False, + ) as handle: + handle.write(prompt) + prompt_path = handle.name + + # Grok requires a PTY; wrap with script(1) when available. + grok_cmd = [ + self._bin, + "--cwd", + root, + "--permission-mode", + permission, + "--always-approve", + "--output-format", + "json", + "--max-turns", + str(max_turns), + "--no-alt-screen", + "--prompt-file", + prompt_path, + ] + if model_value: + grok_cmd.extend(["-m", model_value]) + if variant_value: + grok_cmd.extend(["--reasoning-effort", variant_value]) + + system_prompt = options.get("system_prompt") + if isinstance(system_prompt, str) and system_prompt.strip(): + grok_cmd.extend(["--system-prompt-override", system_prompt]) + + if shutil.which("script") is not None: + # -q quiet, /dev/null typescript file; stdout still carries agent JSON. + cmd = ["script", "-q", "/dev/null", *grok_cmd] + else: + cmd = grok_cmd + + env: Dict[str, str] = {} + env_value = options.get("env") + if isinstance(env_value, dict): + env = { + str(key): str(value) + for key, value in env_value.items() + if isinstance(key, str) and isinstance(value, str) + } + # Ensure localhost control-plane traffic is not proxy-hijacked. + env.setdefault("NO_PROXY", "127.0.0.1,localhost,::1") + env.setdefault("no_proxy", "127.0.0.1,localhost,::1") + + start_api = time.monotonic() + try: + stdout, stderr, returncode = await run_cli(cmd, env=env, cwd=root) + except FileNotFoundError as exc: + raise provider_unavailable("grok", self._bin) from exc + except TimeoutError as exc: + return RawResult( + is_error=True, + error_message=str(exc), + failure_type=FailureType.TIMEOUT, + metrics=Metrics(model=model_value), + ) + + api_ms = int((time.monotonic() - start_api) * 1000) + payload = _extract_json_payload(stdout) + # Grok sometimes emits {"type":"error","message":"..."} on stdout with rc!=0. + if payload and payload.get("type") == "error": + err_msg = payload.get("message") or payload.get("error") or str(payload) + return RawResult( + result=None, + messages=[payload], + metrics=Metrics(model=model_value), + is_error=True, + error_message=str(err_msg)[:1000], + failure_type=FailureType.CRASH, + returncode=returncode, + ) + + result_text = _result_text(payload) if payload else None + metrics = ( + _usage_metrics(payload, model_value) + if payload + else Metrics(model=model_value) + ) + metrics.duration_api_ms = api_ms + + messages: List[Dict[str, Any]] = [payload] if payload else [] + clean_stderr = strip_ansi(stderr.strip()) if stderr else "" + + if returncode < 0: + return RawResult( + result=result_text, + messages=messages, + metrics=metrics, + is_error=True, + error_message=( + f"Process killed by signal {-returncode}. stderr: {clean_stderr[:500]}" + if clean_stderr + else f"Process killed by signal {-returncode}." + ), + failure_type=FailureType.CRASH, + returncode=returncode, + ) + + if returncode != 0 and result_text is None: + return RawResult( + result=None, + messages=messages, + metrics=metrics, + is_error=True, + error_message=( + clean_stderr[:1000] + if clean_stderr + else f"Process exited with code {returncode} and produced no output." + ), + failure_type=FailureType.CRASH, + returncode=returncode, + ) + + if result_text is None and payload is None: + return RawResult( + result=stdout.strip() or None, + messages=messages, + metrics=metrics, + is_error=True, + error_message=( + "Grok produced no parseable JSON output. " + + (clean_stderr[:500] if clean_stderr else f"stdout[:500]={stdout[:500]!r}") + ), + failure_type=FailureType.CRASH, + returncode=returncode, + ) + + return RawResult( + result=result_text, + messages=messages, + metrics=metrics, + is_error=False, + error_message=None, + failure_type=FailureType.NONE, + returncode=returncode, + ) + finally: + if prompt_path: + try: + os.unlink(prompt_path) + except OSError: + pass diff --git a/sdk/python/agentfield/types.py b/sdk/python/agentfield/types.py index 301b859cb..0557abc31 100644 --- a/sdk/python/agentfield/types.py +++ b/sdk/python/agentfield/types.py @@ -276,7 +276,7 @@ class DiscoveryResult: class HarnessConfig(BaseModel): provider: str = Field( ..., - description='Coding agent provider: "claude-code" | "codex" | "gemini" | "opencode"', + description='Coding agent provider: "claude-code" | "codex" | "gemini" | "opencode" | "grok"', ) model: str = Field(default="sonnet", description="Default model identifier.") max_turns: int = Field(default=30, description="Maximum agent iterations.") @@ -321,6 +321,9 @@ class HarnessConfig(BaseModel): opencode_bin: str = Field( default="opencode", description="Path to opencode binary." ) + grok_bin: str = Field( + default="grok", description="Path to Grok Build CLI binary." + ) schema_mode: str = Field( default="single", description=( diff --git a/sdk/python/tests/test_harness_factory.py b/sdk/python/tests/test_harness_factory.py index 8c3faccf7..5f31252a5 100644 --- a/sdk/python/tests/test_harness_factory.py +++ b/sdk/python/tests/test_harness_factory.py @@ -35,6 +35,7 @@ def test_supported_providers_contains_expected_names(): assert "codex" in SUPPORTED_PROVIDERS assert "gemini" in SUPPORTED_PROVIDERS assert "opencode" in SUPPORTED_PROVIDERS + assert "grok" in SUPPORTED_PROVIDERS # --------------------------------------------------------------------------- @@ -208,6 +209,13 @@ def test_claude_code_provider_satisfies_harness_provider_protocol(): assert isinstance(provider, HarnessProvider) +def test_grok_provider_satisfies_harness_provider_protocol(): + from agentfield.harness.providers.grok import GrokProvider + + provider = GrokProvider() + assert isinstance(provider, HarnessProvider) + + def test_harness_provider_protocol_requires_execute_method(): """Any object with async execute(prompt, options) satisfies the Protocol.""" diff --git a/sdk/python/tests/test_harness_grok_provider.py b/sdk/python/tests/test_harness_grok_provider.py new file mode 100644 index 000000000..c76d3374d --- /dev/null +++ b/sdk/python/tests/test_harness_grok_provider.py @@ -0,0 +1,44 @@ +"""Unit tests for the Grok Build CLI harness provider.""" + +from __future__ import annotations + +from agentfield.harness.providers.grok import ( + GrokProvider, + _extract_json_payload, + _usage_metrics, +) +from agentfield.harness.providers._factory import SUPPORTED_PROVIDERS, build_provider +from agentfield.types import HarnessConfig + + +def test_supported_providers_includes_grok() -> None: + assert "grok" in SUPPORTED_PROVIDERS + + +def test_build_provider_grok_default_bin() -> None: + provider = build_provider(HarnessConfig(provider="grok")) + assert isinstance(provider, GrokProvider) + assert provider._bin == "grok" + + +def test_build_provider_grok_custom_bin() -> None: + provider = build_provider(HarnessConfig(provider="grok", grok_bin="/opt/grok")) + assert isinstance(provider, GrokProvider) + assert provider._bin == "/opt/grok" + + +def test_extract_json_payload_tolerates_script_noise() -> None: + raw = "\x04" + '{"text":"{\\"action\\":\\"approve\\"}","num_turns":1}\n' + payload = _extract_json_payload(raw) + assert payload is not None + assert payload["num_turns"] == 1 + + +def test_usage_metrics_missing_tokens_are_zero_not_none() -> None: + metrics = _usage_metrics({"usage": {}, "num_turns": 2}, "grok-4.5") + assert metrics.input_tokens == 0 + assert metrics.output_tokens == 0 + assert metrics.cache_read_tokens == 0 + assert metrics.cache_creation_tokens == 0 + assert metrics.num_turns == 2 + assert metrics.model == "grok-4.5"