From 3d2abc9b059939f88d7709d7fe5b0ebc9126eb58 Mon Sep 17 00:00:00 2001 From: meh Date: Sun, 13 Sep 2026 20:57:09 +0700 Subject: [PATCH 1/4] feat: add Grok over grok agent stdio Fourth agent, ACP JSON-RPC, no new crates. Steer is _x.ai/interject. Interrupt is session/cancel. Auto is native permission-mode auto. /compact maps; /clear is refused. --- CHANGELOG.md | 19 ++ Cargo.toml | 4 +- README.md | 2 + src/account.rs | 2 +- src/agent.rs | 146 ++++++++- src/auth.rs | 14 + src/event.rs | 3 + src/grok_acp.rs | 800 ++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 6 +- src/model.rs | 90 ++++++ src/run.rs | 346 ++++++++++++++++++++- tests/live.rs | 2 +- 12 files changed, 1408 insertions(+), 26 deletions(-) create mode 100644 src/grok_acp.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 285dcb7..113c2d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,25 @@ appear in a patch rather than inflating the version toward 1.0 on a crate still shape. **Where that happens the entry says so at the top**, because a version number that under-signals is only acceptable if the changelog over-signals to compensate. +## 0.4.20 + +### Added + +- **Grok Build as a fourth agent**, over `grok agent stdio` (ACP). A patch + rather than a minor: this adds `Agent::Grok` and changes no existing + behaviour. + + One child per `stream()`, so an IDE can keep many Grok sessions and many + other backends live in parallel. Mid-turn input is `_x.ai/interject` + (same turn, does not cancel or queue). Interrupt is `session/cancel` + (kicks the in-flight turn; the session stays and can reattach). Auto is + native `--permission-mode auto`, not `--always-approve`. `/compact` maps + to `_x.ai/compact_conversation`. `/clear` is refused: Grok has `/new`. + + No new crates. Codex `app-server` and Claude `-p` are untouched. Verified + against grok 1.0.30 `--help` and the live ACP method names; a live + `grok agent stdio` smoke is still outstanding. + ## 0.4.18 ### Added diff --git a/Cargo.toml b/Cargo.toml index c7f6dc6..b34b03b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,11 +1,11 @@ [package] name = "agent-abstraction" -version = "0.4.19" +version = "0.4.20" edition = "2024" # The floor edition 2024 requires, and where the strictest dependencies (uuid, # getrandom) sit. Derived from the dependency graph rather than compile-tested. rust-version = "1.85" -description = "Drive the Claude Code, Codex and GitHub Copilot CLIs headlessly from Rust. One request type, one event stream and one session model across all three, with resume and fork." +description = "Drive the Claude Code, Codex, GitHub Copilot and Grok CLIs headlessly from Rust. One request type, one event stream and one session model across all four, with resume and fork." license = "MIT" repository = "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/pathscale/RustAgentAbstraction" readme = "README.md" diff --git a/README.md b/README.md index 36e8cd6..c6a8d22 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ and not inferred from documentation. | **Claude Code** | caller-minted (`--session-id`) | yes (`--fork-session`) | `stream-json` | native (`--append-system-prompt`) | `--resume` | | **Codex** | agent-printed (`thread_id`) | no | `--json` | prepended to prompt | `exec resume ` | | **Copilot** | caller-minted (`--session-id`) | no | `--output-format json` | prepended to prompt | `--session-id` | +| **Grok** | agent-printed (`session/new`) | yes (`session/fork`) | ACP stdio | native (`--rules`) | `session/load` | ### Can I choose the session id, or do I have to read it back? @@ -53,6 +54,7 @@ Both, depending on the agent. Verified by round-trip, not from `--help`: | **Claude Code** | yes, `.session_id(uuid)` | also reported | | **Copilot** | yes, `.session_id(uuid)` | also reported | | **Codex** | **no** | `thread_id`, before it answers | +| **Grok** | **no** | `sessionId`, from `session/new` | ```rust // Claude and Copilot: the id is yours to pick, so it can match a thread id diff --git a/src/account.rs b/src/account.rs index d7347f9..89b1c3f 100644 --- a/src/account.rs +++ b/src/account.rs @@ -158,7 +158,7 @@ impl Agent { // Deliberately an error rather than a half-answer assembled from a // past run's rate-limit event: that would be neither current nor // account-wide, and would read as though it were both. - Agent::Claude | Agent::Copilot => Err(Error::Unsupported { + Agent::Claude | Agent::Copilot | Agent::Grok => Err(Error::Unsupported { agent: self, what: "reporting account usage without a terminal", }), diff --git a/src/agent.rs b/src/agent.rs index 6a5cf14..8b1ae53 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -1,4 +1,4 @@ -//! The three agents, what each can do, and how a request becomes an argv. +//! The four agents, what each can do, and how a request becomes an argv. //! //! Everything here is pure: [`Agent::argv`] builds a command line from a //! [`Plan`] without touching the filesystem, the clock, or a process, so every @@ -21,6 +21,8 @@ pub enum Agent { Codex, /// GitHub Copilot CLI (`copilot`). Copilot, + /// Grok Build (`grok`). + Grok, } /// How an agent's native session id is obtained. This is the axis deciding whether @@ -308,7 +310,7 @@ pub(crate) const MAX_COMMAND_LINE: usize = 512 * 1024; impl Agent { /// Every agent, in a stable order. - pub const ALL: [Agent; 3] = [Agent::Claude, Agent::Codex, Agent::Copilot]; + pub const ALL: [Agent; 4] = [Agent::Claude, Agent::Codex, Agent::Copilot, Agent::Grok]; /// The stable identifier used in session records and logs. #[must_use] @@ -317,6 +319,7 @@ impl Agent { Agent::Claude => "claude-code", Agent::Codex => "codex", Agent::Copilot => "copilot", + Agent::Grok => "grok", } } @@ -327,6 +330,7 @@ impl Agent { Agent::Claude => "claude", Agent::Codex => "codex", Agent::Copilot => "copilot", + Agent::Grok => "grok", } } @@ -343,6 +347,11 @@ impl Agent { Agent::Claude => Some(&["auth", "status", "--json"]), Agent::Codex => Some(&["login", "status"]), Agent::Copilot => None, + // Verified against grok 1.0.30: there is no `auth status`. `grok + // models` prints "You are logged in with grok.com." when a session + // exists. Logged-out wording is not recorded here, so a miss stays + // `Unknown` rather than `LoggedOut`. + Agent::Grok => Some(&["models"]), } } @@ -357,6 +366,7 @@ impl Agent { Agent::Claude => &["ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"], Agent::Codex => &["CODEX_API_KEY", "OPENAI_API_KEY"], Agent::Copilot => &["COPILOT_GITHUB_TOKEN", "GH_TOKEN", "GITHUB_TOKEN"], + Agent::Grok => &["XAI_API_KEY"], } } @@ -374,6 +384,7 @@ impl Agent { } Agent::Codex => "run `codex login`", Agent::Copilot => "run `copilot login`", + Agent::Grok => "run `grok login`", } } @@ -392,6 +403,8 @@ impl Agent { Agent::Codex => (0, 147, 0), // `copilot --version` -> "GitHub Copilot CLI 1.0.78." Agent::Copilot => (1, 0, 78), + // `grok --version` -> "grok 1.0.30 (04b7ffed98c6) [stable]" + Agent::Grok => (1, 0, 30), }; crate::Version { major, @@ -407,6 +420,7 @@ impl Agent { Agent::Claude => "npm install -g @anthropic-ai/claude-code", Agent::Codex => "npm install -g @openai/codex", Agent::Copilot => "npm install -g @github/copilot", + Agent::Grok => "curl -fsSL https://x.ai/cli/install.sh | bash", } } @@ -474,6 +488,7 @@ impl Agent { "GITHUB_TOKEN", "XDG_CONFIG_HOME", ], + Agent::Grok => &["XAI_API_KEY", "GROK_HOME"], }; BASE.iter().chain(WINDOWS).chain(agent).copied().collect() } @@ -495,6 +510,8 @@ impl Agent { pub fn thinking_env(self, thinking: Option) -> Option<(&'static str, &'static str)> { match (self, thinking) { (Agent::Claude, Some(false)) => Some(("MAX_THINKING_TOKENS", "0")), + // grok 1.0.30 steers reasoning with `--effort` / `--reasoning-effort`, + // not an environment variable. _ => None, } } @@ -551,6 +568,27 @@ impl Agent { live_follow_up: false, approvals: false, }, + // Verified against grok 1.0.30 `--help` and ACP stdio: `session/new` + // prints the id, `session/fork` / `_x.ai/session/fork` branches, + // `_x.ai/interject` steers mid-turn (confirmed live; bare + // `x.ai/interject` is -32601), `session/cancel` interrupts, + // `session/request_permission` asks. Auto is native + // `--permission-mode auto`, same as Claude: opening the approval + // channel would replace it with a round trip the host would only + // answer yes to. + Agent::Grok => Caps { + session: SessionSupport::Printed, + fork: true, + events: true, + native_system: true, + schema: SchemaSupport::Inline, + // `/compact` is a Grok command and an ACP method. `/clear` is + // not: Grok uses `/new`. Other names are refused rather than + // sent as prose. + commands: true, + live_follow_up: true, + approvals: true, + }, } } @@ -563,9 +601,8 @@ impl Agent { // cheaper default when the caller did not ask to stream. SessionSupport::Minted | SessionSupport::Printed => Some(match self { Agent::Claude => Format::Json, - // `--json` IS Codex's stream and Copilot's `json` is JSONL; - // neither has a single-document form. - Agent::Codex | Agent::Copilot => Format::Stream, + // ACP stdio, `codex --json`, and Copilot `json` are all streams. + Agent::Codex | Agent::Copilot | Agent::Grok => Format::Stream, }), SessionSupport::None => None, } @@ -682,6 +719,7 @@ impl Agent { Agent::Claude => argv_claude(plan), Agent::Codex => argv_codex(plan), Agent::Copilot => argv_copilot(plan), + Agent::Grok => argv_grok(plan), }) } @@ -1071,6 +1109,57 @@ fn argv_copilot(plan: &Plan) -> Vec { a.done() } +/// `grok [--permission-mode M] agent [--model M] [--always-approve] stdio` +/// +/// Flag placement verified against grok 1.0.30 `--help` and the live ACP +/// client: `--permission-mode` is top-level `grok`, `--model` / +/// `--reasoning-effort` / `--always-approve` sit on `grok agent` before +/// `stdio`. The prompt never rides the argv; it is `session/prompt` after +/// initialize. Auto is `--permission-mode auto`, not `--always-approve`: +/// native auto, same token Claude uses, and Grok has the same token. +/// Bypass is `--always-approve`. Mid-turn steer is `_x.ai/interject` on the +/// open stdio, not a second spawn. +fn argv_grok(plan: &Plan) -> Vec { + let mut a = Argv::new(&plan.bin); + a.bare("--no-auto-update"); + a.pair("--permission-mode", grok_mode(plan.permission)); + if plan.permission == Permission::ReadOnly { + // Internal ids from grok's own headless docs (`--disallowed-tools + // run_terminal_cmd`, `search_replace`, `Agent`). Comma-separated is + // one argument. + a.pair( + "--disallowed-tools", + "run_terminal_cmd,search_replace,Agent", + ); + } + if let Some(system) = &plan.system { + a.secret("--rules", system, Sensitivity::Prompt); + } + if let Some(schema) = &plan.schema { + a.secret("--json-schema", schema, Sensitivity::Prompt); + } + a.bare("agent"); + a.opt("--model", plan.model.as_ref()); + if let Some(effort) = &plan.effort { + a.pair("--reasoning-effort", effort); + } + if plan.permission == Permission::Bypass { + a.bare("--always-approve"); + } + a.bare("stdio"); + a.done() +} + +fn grok_mode(p: Permission) -> &'static str { + match p { + Permission::ReadOnly => "dontAsk", + Permission::Plan => "plan", + Permission::Edit => "acceptEdits", + Permission::Auto => "auto", + Permission::Bypass => "bypassPermissions", + } +} + #[cfg(test)] mod tests { use super::*; @@ -1102,7 +1191,7 @@ mod tests { #[test] fn interactive_capabilities_match_the_supported_request_paths() { - for agent in [Agent::Claude, Agent::Codex] { + for agent in [Agent::Claude, Agent::Codex, Agent::Grok] { let caps = agent.caps(); assert!(caps.live_follow_up, "{agent} can take live follow-ups"); assert!(caps.approvals, "{agent} has an approval channel"); @@ -1174,6 +1263,7 @@ mod tests { Err(Error::Unsupported { .. }) )); assert!(Agent::Claude.typed_argv(&p).is_ok()); + assert!(Agent::Grok.typed_argv(&p).is_ok()); assert_eq!(argv(Agent::Codex, &p), ["x", "app-server", "--stdio"]); } @@ -1245,6 +1335,9 @@ mod tests { Err(Error::Unsupported { .. }) )); assert_eq!(argv(Agent::Codex, &p), ["x", "app-server", "--stdio"]); + let grok = argv(Agent::Grok, &p); + assert!(grok.contains(&"stdio".to_string()), "{grok:?}"); + assert!(grok.contains(&"agent".to_string()), "{grok:?}"); } /// An ordinary run is untouched, so nothing about the default path changes. @@ -1749,6 +1842,8 @@ mod tests { Agent::Claude.argv(&p).is_ok(), "Claude publishes a catalogue and acts on them" ); + p.bin = "grok".into(); + assert!(Agent::Grok.argv(&p).is_ok(), "Grok maps /compact over ACP"); } /// All three expose an id, so all three can back a named session, but only @@ -1758,6 +1853,7 @@ mod tests { assert_eq!(Agent::Claude.session_format(), Some(Format::Json)); assert_eq!(Agent::Codex.session_format(), Some(Format::Stream)); assert_eq!(Agent::Copilot.session_format(), Some(Format::Stream)); + assert_eq!(Agent::Grok.session_format(), Some(Format::Stream)); } /// Claude and Copilot let the caller assign the id, so a run that dies @@ -1770,4 +1866,42 @@ mod tests { .collect(); assert_eq!(minting, [Agent::Claude, Agent::Copilot]); } + + #[test] + fn grok_stdio_keeps_auto_native_and_puts_flags_in_the_right_place() { + let mut p = plan("grok"); + p.permission = Permission::Auto; + p.model = Some("grok-4.6".into()); + p.effort = Some("high".into()); + p.system = Some("be brief".into()); + let a = argv(Agent::Grok, &p); + assert_eq!(a[0], "grok"); + let agent_at = pos(&a, "agent").expect("agent subcommand"); + let stdio_at = pos(&a, "stdio").expect("stdio"); + assert!(agent_at < stdio_at); + assert!(pos(&a, "--permission-mode").unwrap() < agent_at); + assert_eq!(a[pos(&a, "--permission-mode").unwrap() + 1], "auto"); + assert!(pos(&a, "--model").unwrap() > agent_at); + assert_eq!(a[pos(&a, "--model").unwrap() + 1], "grok-4.6"); + assert!(pos(&a, "--reasoning-effort").unwrap() > agent_at); + assert_eq!(a[pos(&a, "--reasoning-effort").unwrap() + 1], "high"); + assert!(pos(&a, "--rules").unwrap() < agent_at); + assert!(!a.iter().any(|arg| arg == "--always-approve")); + assert!( + !a.iter().any(|arg| arg == "hi"), + "prompt is ACP not argv: {a:?}" + ); + } + + #[test] + fn grok_bypass_is_always_approve_not_auto() { + let mut p = plan("grok"); + p.permission = Permission::Bypass; + let a = argv(Agent::Grok, &p); + assert!(a.iter().any(|arg| arg == "--always-approve"), "{a:?}"); + assert_eq!( + a[pos(&a, "--permission-mode").unwrap() + 1], + "bypassPermissions" + ); + } } diff --git a/src/auth.rs b/src/auth.rs index d645096..e2eeca2 100644 --- a/src/auth.rs +++ b/src/auth.rs @@ -164,6 +164,20 @@ impl AuthStatus { // Unreachable: `auth_status_argv` returns None, so `check_bin` // never gets here for Copilot. Agent::Copilot => {} + Agent::Grok => { + let lower = text.to_ascii_lowercase(); + status.state = if lower.contains("not logged in") || lower.contains("logged out") { + AuthState::LoggedOut + } else if lower.contains("logged in") { + status.method = text + .rsplit_once(" with ") + .or_else(|| text.rsplit_once(" using ")) + .map(|(_, method)| method.trim().trim_end_matches('.').to_string()); + AuthState::LoggedIn + } else { + AuthState::Unknown + }; + } } status } diff --git a/src/event.rs b/src/event.rs index b939a01..5617dcd 100644 --- a/src/event.rs +++ b/src/event.rs @@ -537,6 +537,8 @@ impl Parser { Agent::Claude => self.claude(&value), Agent::Codex => self.codex(&value), Agent::Copilot => self.copilot(&value), + // Grok's live path is ACP in `grok_acp`, not this line parser. + Agent::Grok => Vec::new(), }; // Every event leaves through here, so bounding once at the exit covers // all three agents rather than each parser remembering. @@ -649,6 +651,7 @@ impl Parser { || ty.starts_with("tool.") || ty.starts_with("session.") } + Agent::Grok => false, } } diff --git a/src/grok_acp.rs b/src/grok_acp.rs new file mode 100644 index 0000000..ed212ed --- /dev/null +++ b/src/grok_acp.rs @@ -0,0 +1,800 @@ +//! Grok ACP stdio transport. +//! +//! Spawn is `grok agent stdio`. One child per [`crate::stream`] so an IDE can +//! keep many sessions and many backends live in parallel: this is not a +//! process-wide singleton, and Claude/Codex keep their own processes. A host +//! that wants many Grok sessions on one child holds the process; this crate +//! does not serialize the IDE onto one pipe, and it does not flatten Grok +//! into Codex's spawn-per-turn model. +//! +//! Mid-turn input is `_x.ai/interject` with `{sessionId, text}` (confirmed +//! live on grok 1.0.30: bare `x.ai/interject` is -32601). That injects into +//! the current turn. It does not queue, and it does not cancel. +//! +//! Interrupt is `session/cancel`. That kicks an in-flight turn so a resume +//! can reattach. The session id stays. +//! +//! `/compact` is `_x.ai/compact_conversation`. `/clear` is not a Grok +//! command (Grok uses `/new`). + +use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; + +use serde_json::{Value, json}; + +use crate::agent::{Continue, Permission}; +use crate::approval::{Approval, Decision}; +use crate::event::{Event, Terminal, append_capped}; +use crate::outcome::{Stop, Usage}; +use crate::request::Request; + +const INIT_ID: u64 = 1; +const SESSION_ID: u64 = 2; +const PROMPT_ID: u64 = 3; +const COMPACT_ID: u64 = 4; +const NEXT_ID: u64 = 10; + +/// One decoded ACP record. +#[derive(Debug, Default)] +pub(crate) struct Step { + pub events: Vec, + pub writes: Vec, + pub steer_responses: Vec, +} + +/// One `_x.ai/interject` request. +#[derive(Debug)] +pub(crate) struct SteerRequest { + pub id: u64, + pub wire: String, +} + +/// Agent acceptance or rejection of one interject. +#[derive(Debug)] +pub(crate) struct SteerResponse { + pub id: u64, + pub result: std::result::Result, +} + +#[derive(Debug)] +struct PendingApproval { + rpc_id: Value, + options: Value, +} + +/// State that spans the JSON-RPC records of one turn. +#[derive(Debug)] +pub(crate) struct Protocol { + request: Request, + pub terminal: Terminal, + pub session_id: Option, + pub finished: bool, + pub failure: Option, + pending: HashMap, + pending_steers: HashSet, + interrupt_requested: bool, + next_id: u64, + compacting: bool, + last_steer: Option, +} + +impl Protocol { + pub fn new(request: Request) -> Self { + Self { + request, + terminal: Terminal::default(), + session_id: None, + finished: false, + failure: None, + pending: HashMap::new(), + pending_steers: HashSet::new(), + interrupt_requested: false, + next_id: NEXT_ID, + compacting: false, + last_steer: None, + } + } + + /// First write: ACP `initialize`. Session open waits on the reply. + pub fn opening() -> Vec { + vec![wire(&json!({ + "jsonrpc": "2.0", + "id": INIT_ID, + "method": "initialize", + "params": { + "protocolVersion": 1, + "clientInfo": { + "name": "agent-abstraction", + "version": env!("CARGO_PKG_VERSION"), + }, + "clientCapabilities": { + "fs": { "readTextFile": false, "writeTextFile": false }, + "terminal": false + }, + }, + }))] + } + + fn cwd(&self) -> PathBuf { + self.request + .cwd + .clone() + .or_else(|| std::env::current_dir().ok()) + .unwrap_or_else(|| PathBuf::from(".")) + } + + fn session_params(&self) -> Value { + let cwd = self.cwd(); + let extra: Vec = self + .request + .extra_dirs + .iter() + .map(|path| path.to_string_lossy().into_owned()) + .collect(); + let mut meta = json!({}); + if let Some(system) = &self.request.system { + meta["rules"] = json!(system); + } + match self.request.permission { + Permission::Bypass => meta["yoloMode"] = json!(true), + Permission::Auto => meta["autoMode"] = json!(true), + _ => {} + } + let mut params = json!({ + "cwd": cwd, + "mcpServers": [], + }); + if !extra.is_empty() { + params["additionalDirectories"] = json!(extra); + } + if meta.as_object().is_some_and(|map| !map.is_empty()) { + params["_meta"] = meta; + } + params + } + + fn open_session(&self) -> Value { + let mut params = self.session_params(); + match &self.request.cont { + Continue::New | Continue::NewWith(_) => json!({ + "jsonrpc": "2.0", + "id": SESSION_ID, + "method": "session/new", + "params": params, + }), + Continue::Resume(session_id) => { + params["sessionId"] = json!(session_id); + json!({ + "jsonrpc": "2.0", + "id": SESSION_ID, + "method": "session/load", + "params": params, + }) + } + Continue::Fork(session_id) => { + params["sessionId"] = json!(session_id); + json!({ + "jsonrpc": "2.0", + "id": SESSION_ID, + "method": "session/fork", + "params": params, + }) + } + } + } + + fn start_prompt(&self, session_id: &str) -> String { + if self.request.operation == crate::request::Operation::Interrupt { + return Self::cancel_wire(session_id); + } + if self.request.is_command { + return self.command_wire(session_id); + } + wire(&json!({ + "jsonrpc": "2.0", + "id": PROMPT_ID, + "method": "session/prompt", + "params": { + "sessionId": session_id, + "prompt": [{ "type": "text", "text": self.request.prompt }], + }, + })) + } + + fn command_wire(&self, session_id: &str) -> String { + // AgencyZero's mapped command is `/compact`. Grok's ACP name is + // `compact_conversation`. `/clear` is not a Grok command (it has + // `/new`). Anything else is refused rather than sent as prose. + let prompt = self.request.prompt.trim(); + let Some(rest) = prompt.strip_prefix("/compact") else { + return String::new(); + }; + if !rest.is_empty() && !rest.starts_with(' ') { + return String::new(); + } + let mut params = json!({ "sessionId": session_id }); + let how = rest.trim(); + if !how.is_empty() { + params["instructions"] = json!(how); + } + wire(&json!({ + "jsonrpc": "2.0", + "id": COMPACT_ID, + "method": "_x.ai/compact_conversation", + "params": params, + })) + } + + fn cancel_wire(session_id: &str) -> String { + wire(&json!({ + "jsonrpc": "2.0", + "method": "session/cancel", + "params": { "sessionId": session_id }, + })) + } + + /// Turn one JSON-RPC line into host events and follow-up writes. + pub fn push(&mut self, value: &Value) -> Step { + let mut step = Step::default(); + + if let Some(id) = json_id(value) { + if let Some(error) = rpc_error(value) { + return self.push_error(id, error, &mut step); + } + match id { + INIT_ID => { + step.writes.push(wire(&self.open_session())); + return step; + } + SESSION_ID => return self.push_session(value, &mut step), + PROMPT_ID => { + self.finish_prompt(value); + return step; + } + COMPACT_ID => { + self.compacting = false; + self.terminal.stop = Stop::Completed; + self.finished = true; + step.events + .push(Event::Compaction(crate::command::Compaction::Finished { + ok: true, + error: None, + })); + return step; + } + other if self.pending_steers.contains(&other) => { + self.pending_steers.remove(&other); + step.steer_responses.push(SteerResponse { + id: other, + result: Ok(String::new()), + }); + return step; + } + _ => {} + } + } + + let method = value.get("method").and_then(Value::as_str).unwrap_or(""); + match method { + "session/update" | "x.ai/session/update" | "_x.ai/session/update" => { + step.events.extend(self.session_update(value)); + } + "session/request_permission" => { + if let Some(event) = self.permission_request(value) { + step.events.push(event); + } + } + _ => {} + } + step + } + + fn push_error(&mut self, id: u64, error: String, step: &mut Step) -> Step { + if id == SESSION_ID + && matches!(self.request.cont, Continue::Fork(_)) + && error.to_ascii_lowercase().contains("method") + { + // `session/fork` missing: Grok extension name. + let mut params = self.session_params(); + if let Continue::Fork(session_id) = &self.request.cont { + params["sessionId"] = json!(session_id); + } + step.writes.push(wire(&json!({ + "jsonrpc": "2.0", + "id": SESSION_ID, + "method": "_x.ai/session/fork", + "params": params, + }))); + return std::mem::take(step); + } + if id == COMPACT_ID && error.to_ascii_lowercase().contains("method") { + let Some(session_id) = self.session_id.clone() else { + self.failure = Some(error); + self.finished = true; + return std::mem::take(step); + }; + let mut params = json!({ "sessionId": session_id }); + if let Some(rest) = self.request.prompt.strip_prefix("/compact") { + let how = rest.trim(); + if !how.is_empty() { + params["instructions"] = json!(how); + } + } + step.writes.push(wire(&json!({ + "jsonrpc": "2.0", + "id": COMPACT_ID, + "method": "x.ai/compact_conversation", + "params": params, + }))); + return std::mem::take(step); + } + if self.pending_steers.remove(&id) { + if looks_like_method_not_found(&error) { + if let Some(retry) = self.interject_fallback() { + step.writes.push(retry.wire); + self.pending_steers.insert(retry.id); + step.steer_responses.push(SteerResponse { + id, + result: Ok(String::new()), + }); + return std::mem::take(step); + } + } + step.steer_responses.push(SteerResponse { + id, + result: Err(error), + }); + return std::mem::take(step); + } + self.failure = Some(error); + self.finished = true; + std::mem::take(step) + } + + fn push_session(&mut self, value: &Value, step: &mut Step) -> Step { + let session_id = value + .pointer("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/result/sessionId") + .or_else(|| value.pointer("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/result/session_id")) + .and_then(Value::as_str) + .map(str::to_string) + .or_else(|| match &self.request.cont { + Continue::Resume(id) => Some(id.clone()), + _ => None, + }); + let Some(session_id) = session_id else { + self.failure = Some("session/new returned no sessionId".into()); + self.finished = true; + return std::mem::take(step); + }; + self.session_id = Some(session_id.clone()); + self.terminal.session = Some(session_id.clone()); + let model = value + .pointer("/result/models/current/id") + .or_else(|| value.pointer("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/result/model")) + .and_then(Value::as_str) + .map(str::to_string); + self.terminal.model.clone_from(&model); + step.events.push(Event::Started { + session: session_id.clone(), + model, + }); + if self.request.is_command { + let prompt = self.request.prompt.trim(); + if prompt == "/clear" || prompt.starts_with("/clear ") { + self.failure = + Some("Grok has /new, not /clear; refusing rather than sending prose".into()); + self.finished = true; + return std::mem::take(step); + } + self.compacting = true; + step.events + .push(Event::Compaction(crate::command::Compaction::Started)); + } + let next = self.start_prompt(&session_id); + if next.is_empty() { + self.finished = true; + } else { + step.writes.push(next); + if self.request.operation == crate::request::Operation::Interrupt { + self.interrupt_requested = true; + self.terminal.stop = Stop::Other("interrupted".into()); + self.finished = true; + } + } + std::mem::take(step) + } + + fn finish_prompt(&mut self, value: &Value) { + let reason = value + .pointer("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/result/stopReason") + .or_else(|| value.pointer("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/result/stop_reason")) + .and_then(Value::as_str) + .unwrap_or("end_turn"); + self.terminal.stop = match reason { + "cancelled" | "canceled" => Stop::Other("interrupted".into()), + "max_tokens" | "max_turns" => Stop::Other(reason.into()), + "refusal" | "error" => Stop::Error, + _ => Stop::Completed, + }; + if let Some(usage) = value + .pointer("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/result/usage") + .or_else(|| value.pointer("/result/_meta/usage")) + { + self.terminal.usage = grok_usage(usage); + } + self.finished = true; + } + + fn session_update(&mut self, value: &Value) -> Vec { + let update = value + .pointer("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/params/update") + .or_else(|| value.get("params")) + .cloned() + .unwrap_or(Value::Null); + let kind = update + .get("sessionUpdate") + .or_else(|| update.get("session_update")) + .and_then(Value::as_str) + .unwrap_or(""); + match kind { + "agent_message_chunk" | "agent_message" => text_content(&update) + .into_iter() + .map(|text| { + append_capped(&mut self.terminal.text, &text); + Event::Text(text) + }) + .collect(), + "agent_thought_chunk" | "agent_thought" => text_content(&update) + .into_iter() + .map(Event::Thinking) + .collect(), + "tool_call" => vec![Event::ToolCall { + id: update + .get("toolCallId") + .or_else(|| update.get("tool_call_id")) + .and_then(Value::as_str) + .map(str::to_string), + name: tool_name(&update), + input: update + .get("rawInput") + .or_else(|| update.get("raw_input")) + .cloned() + .unwrap_or(Value::Null), + }], + "tool_call_update" => { + let status = update.get("status").and_then(Value::as_str).unwrap_or(""); + if status != "completed" && status != "failed" { + return Vec::new(); + } + vec![Event::ToolResult { + id: update + .get("toolCallId") + .or_else(|| update.get("tool_call_id")) + .and_then(Value::as_str) + .map(str::to_string), + ok: Some(status == "completed"), + output: update + .get("rawOutput") + .or_else(|| update.get("raw_output")) + .map(value_as_text) + .unwrap_or_default(), + }] + } + "available_commands_update" => { + let names = command_names(&update); + if names.is_empty() { + return Vec::new(); + } + vec![Event::Commands(crate::command::Commands { + all: names, + skills: Vec::new(), + })] + } + "usage_update" => { + let usage = grok_usage(&update); + self.terminal.usage.accumulate(&usage); + vec![Event::Usage(usage)] + } + _ => Vec::new(), + } + } + + fn permission_request(&mut self, value: &Value) -> Option { + let id = value.get("id")?.clone(); + let params = value.get("params")?; + let tool_call = params.get("toolCall").unwrap_or(params); + let tool = tool_name(tool_call); + let input = tool_call + .get("rawInput") + .or_else(|| tool_call.get("raw_input")) + .cloned() + .unwrap_or(Value::Null); + let options = params.get("options").cloned().unwrap_or_else(|| json!([])); + let key = match &id { + Value::Number(n) => n.to_string(), + Value::String(s) => s.clone(), + other => other.to_string(), + }; + self.pending.insert( + key.clone(), + PendingApproval { + rpc_id: id, + options, + }, + ); + Some(Event::ApprovalRequest(Approval { + id: key, + tool, + input, + })) + } + + /// Encode mid-turn input. Does not cancel the turn. + pub fn steer(&mut self, message: &str) -> Option { + let session_id = self.session_id.as_ref()?; + self.last_steer = Some(message.to_string()); + let id = self.next_id; + self.next_id += 1; + self.pending_steers.insert(id); + Some(SteerRequest { + id, + wire: wire(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": "_x.ai/interject", + "params": { "sessionId": session_id, "text": message }, + })), + }) + } + + fn interject_fallback(&mut self) -> Option { + let session_id = self.session_id.as_ref()?; + let text = self.last_steer.clone().unwrap_or_default(); + let id = self.next_id; + self.next_id += 1; + self.pending_steers.insert(id); + Some(SteerRequest { + id, + wire: wire(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": "x.ai/interject", + "params": { "sessionId": session_id, "text": text }, + })), + }) + } + + /// Kick the in-flight turn. Session remains and can be resumed. + pub fn interrupt(&mut self) -> Option { + let session_id = self.session_id.as_ref()?; + self.interrupt_requested = true; + Some(Self::cancel_wire(session_id)) + } + + pub fn respond(&mut self, id: &str, decision: &Decision) -> Option { + let pending = self.pending.remove(id)?; + let option_id = pick_option(&pending.options, decision); + let result = match option_id { + Some(option_id) => json!({ + "outcome": { "outcome": "selected", "optionId": option_id } + }), + None => json!({ "outcome": { "outcome": "cancelled" } }), + }; + Some(wire(&json!({ + "jsonrpc": "2.0", + "id": pending.rpc_id, + "result": result, + }))) + } +} + +fn json_id(value: &Value) -> Option { + value.get("id").and_then(Value::as_u64) +} + +fn rpc_error(value: &Value) -> Option { + let error = value.get("error")?; + let message = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("rpc error"); + let code = error.get("code").and_then(Value::as_i64); + Some(match code { + Some(code) => format!("{code} {message}"), + None => message.to_string(), + }) +} + +fn looks_like_method_not_found(error: &str) -> bool { + let lower = error.to_ascii_lowercase(); + lower.contains("method not found") + || lower.contains("-32601") + || lower.contains("unknown method") +} + +fn text_content(update: &Value) -> Option { + let content = update.get("content")?; + if let Some(text) = content.get("text").and_then(Value::as_str) { + return Some(text.to_string()).filter(|s| !s.is_empty()); + } + if let Some(items) = content.as_array() { + let text: String = items + .iter() + .filter_map(|item| item.get("text").and_then(Value::as_str)) + .collect(); + return Some(text).filter(|s| !s.is_empty()); + } + None +} + +fn tool_name(call: &Value) -> String { + call.pointer("/_meta/x.ai~1tool/name") + .or_else(|| call.get("title")) + .or_else(|| call.get("kind")) + .and_then(Value::as_str) + .unwrap_or("tool") + .to_string() +} + +fn command_names(update: &Value) -> Vec { + update + .get("availableCommands") + .or_else(|| update.get("available_commands")) + .and_then(Value::as_array) + .map(|entries| { + entries + .iter() + .filter_map(|entry| { + entry + .get("name") + .or_else(|| entry.get("command")) + .and_then(Value::as_str) + .map(|name| name.trim_start_matches('/').to_string()) + }) + .collect() + }) + .unwrap_or_default() +} + +fn grok_usage(usage: &Value) -> Usage { + let num = |camel: &str, snake: &str| { + usage + .get(camel) + .or_else(|| usage.get(snake)) + .and_then(Value::as_u64) + }; + Usage { + input_tokens: num("inputTokens", "input_tokens"), + output_tokens: num("outputTokens", "output_tokens"), + cache_read_tokens: num("cachedReadTokens", "cache_read_input_tokens"), + cache_write_tokens: num("cacheCreationTokens", "cache_creation_input_tokens"), + context_tokens: num("totalTokens", "total_tokens").or_else(|| num("used", "used")), + context_window: num("size", "context_window"), + cost_usd: usage + .get("cost") + .and_then(Value::as_f64) + .or_else(|| usage.get("costUsd").and_then(Value::as_f64)), + ..Usage::default() + } +} + +fn value_as_text(value: &Value) -> String { + match value { + Value::String(text) => text.clone(), + other => other.to_string(), + } +} + +fn pick_option(options: &Value, decision: &Decision) -> Option { + let entries = options.as_array()?; + let want_allow = matches!(decision, Decision::Allow); + for entry in entries { + let kind = entry + .get("kind") + .or_else(|| entry.get("optionId")) + .or_else(|| entry.get("option_id")) + .and_then(Value::as_str) + .unwrap_or("") + .to_ascii_lowercase(); + let allow = kind.contains("allow") || kind.contains("approve"); + let deny = kind.contains("reject") || kind.contains("deny"); + if want_allow && allow || !want_allow && deny { + return entry + .get("optionId") + .or_else(|| entry.get("option_id")) + .or_else(|| entry.get("id")) + .and_then(Value::as_str) + .map(str::to_string); + } + } + None +} + +fn wire(value: &Value) -> String { + format!("{value}\n") +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::agent::Agent; + use crate::request::Request; + + fn protocol() -> Protocol { + Protocol::new(Request::new(Agent::Grok, "hi").permission(Permission::Auto)) + } + + #[test] + fn opening_is_initialize_only() { + let opening = Protocol::opening(); + assert_eq!(opening.len(), 1); + assert!(opening[0].contains("\"method\":\"initialize\"")); + assert!(opening[0].contains("\"protocolVersion\":1")); + } + + #[test] + fn auto_is_native_automode_not_yolo() { + let mut p = protocol(); + let _ = p.push(&json!({"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1}})); + let open = p.open_session(); + let meta = &open["params"]["_meta"]; + assert_eq!(meta["autoMode"], json!(true)); + assert!(meta.get("yoloMode").is_none()); + } + + #[test] + fn steer_is_interject_not_a_new_prompt() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + let steer = p.steer("stop the tests").expect("session is known"); + assert!(steer.wire.contains("_x.ai/interject")); + assert!(steer.wire.contains("\"sessionId\":\"sess-1\"")); + assert!(steer.wire.contains("\"text\":\"stop the tests\"")); + assert!(!steer.wire.contains("session/prompt")); + assert!(!steer.wire.contains("session/cancel")); + } + + #[test] + fn interrupt_is_cancel_and_keeps_the_session() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + let wire = p.interrupt().expect("session is known"); + assert!(wire.contains("session/cancel")); + assert!(wire.contains("\"sessionId\":\"sess-1\"")); + assert_eq!(p.session_id.as_deref(), Some("sess-1")); + } + + #[test] + fn compact_uses_the_acp_method() { + let mut request = Request::command( + Agent::Grok, + &crate::Command::Compact { + instructions: Some("keep the auth tests".into()), + }, + ); + request.cont = Continue::Resume("sess-9".into()); + let p = Protocol::new(request); + let wire = p.command_wire("sess-9"); + assert!(wire.contains("compact_conversation")); + assert!(wire.contains("keep the auth tests")); + assert!(!wire.contains("session/prompt")); + } + + #[test] + fn agent_text_chunks_become_events() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + let events = p.session_update(&json!({ + "method": "session/update", + "params": { + "sessionId": "sess-1", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "pong" } + } + } + })); + assert_eq!(events, vec![Event::Text("pong".into())]); + assert_eq!(p.terminal.text.trim_end(), "pong"); + } +} diff --git a/src/lib.rs b/src/lib.rs index fe7ea0a..7590804 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,6 @@ -//! Drive Claude Code, Codex and GitHub Copilot headlessly from Rust. +//! Drive Claude Code, Codex, GitHub Copilot and Grok headlessly from Rust. //! -//! One request type, one event vocabulary and one session model across three +//! One request type, one event vocabulary and one session model across four //! agent CLIs that agree on none of those things. This is a **library**: your //! program links it and spawns the agent itself, with no intermediate CLI //! marshalling a request through stdout and back. @@ -72,6 +72,7 @@ //! | Claude Code | caller-minted (`--session-id`) | yes | yes | native flag | //! | Codex | agent-printed (`thread_id`) | no | yes | prepended | //! | Copilot | caller-minted (`--session-id`) | no | yes | prepended | +//! | Grok | agent-printed (`session/new`) | yes | yes | native (`--rules`) | //! //! Asking for something an agent cannot do is always an [`Error::Unsupported`], //! never a silent downgrade. A caller that asked to fork and got a linear @@ -93,6 +94,7 @@ mod codex_app_server; mod command; mod error; mod event; +mod grok_acp; mod model; mod outcome; mod probe; diff --git a/src/model.rs b/src/model.rs index 89b78d4..66071ed 100644 --- a/src/model.rs +++ b/src/model.rs @@ -148,6 +148,7 @@ impl Agent { Agent::Claude => claude_models(), Agent::Codex => codex_models(), Agent::Copilot => copilot_models(), + Agent::Grok => grok_models(), } } @@ -176,6 +177,11 @@ impl Agent { checked: "2026-07-29", against: "Copilot CLI 1.0.75", }, + Agent::Grok => Verified { + source: Source::Cli, + checked: "2026-09-13", + against: "grok 1.0.30", + }, } } @@ -196,6 +202,7 @@ impl Agent { pub async fn discover_models(&self) -> Result> { match self { Agent::Codex => discover_codex(self.bin()).await, + Agent::Grok => discover_grok(self.bin()).await, // Neither can be asked without a terminal, verified against // Copilot CLI 1.0.75 and claude 2.1.212. Copilot has no `models` // subcommand, rejects an unknown `--model` without listing the valid @@ -508,6 +515,89 @@ fn pinned(id: &'static str, name: &'static str) -> Model { Model::new(id, name, "", Kind::Pinned, COPILOT_EFFORTS, false) } +/// Grok, from `grok models` on 1.0.30 (2026-09-13). +/// +/// Effort tokens from grok `--help` (`--reasoning-effort` / `--effort`) and +/// the session config option `reasoning_effort`. +const GROK_EFFORTS: &[&str] = &["minimal", "low", "medium", "high", "xhigh"]; + +fn grok_models() -> Vec { + vec![ + Model::new( + "grok-4.6", + "Grok 4.6", + "Default Grok Build model", + Kind::Pinned, + GROK_EFFORTS, + true, + ), + Model::new( + "grok-4.5", + "Grok 4.5", + "", + Kind::Pinned, + GROK_EFFORTS, + false, + ), + ] +} + +async fn discover_grok(bin: &str) -> Result> { + let output = tokio::process::Command::new(bin) + .arg("models") + .output() + .await + .map_err(|source| { + if source.kind() == std::io::ErrorKind::NotFound { + Error::NotInstalled { + agent: Agent::Grok, + bin: bin.to_string(), + hint: Agent::Grok.install_hint(), + } + } else { + Error::Spawn { + bin: bin.to_string(), + source, + } + } + })?; + parse_grok_models(&String::from_utf8_lossy(&output.stdout)) +} + +fn parse_grok_models(stdout: &str) -> Result> { + let mut models = Vec::new(); + for line in stdout.lines() { + let line = line.trim(); + let rest = line + .strip_prefix("* ") + .or_else(|| line.strip_prefix("- ")) + .unwrap_or(""); + if rest.is_empty() { + continue; + } + let id = rest.split_whitespace().next().unwrap_or(rest); + let is_default = rest.contains("(default)"); + models.push(Model { + id: id.to_string().into(), + name: id.to_string().into(), + note: Cow::Borrowed(""), + kind: Kind::Pinned, + efforts: GROK_EFFORTS.iter().map(|e| Cow::Borrowed(*e)).collect(), + is_default, + }); + } + if models.is_empty() { + return Err(Error::Parse { + agent: Agent::Grok, + detail: "`grok models` listed no models".into(), + }); + } + if !models.iter().any(|model| model.is_default) { + models[0].is_default = true; + } + Ok(models) +} + /// Read Codex's own model list. /// /// `codex debug models` prints one JSON document carrying every model plus each diff --git a/src/run.rs b/src/run.rs index 2a4ee89..f3f157a 100644 --- a/src/run.rs +++ b/src/run.rs @@ -479,7 +479,7 @@ pub async fn run(request: &Request) -> Result { /// Returns [`Error::Unsupported`] for another provider or a request that does /// not resume a session, and the ordinary spawn/protocol errors otherwise. pub async fn interrupt(request: &Request) -> Result { - if request.agent != crate::Agent::Codex { + if !matches!(request.agent, crate::Agent::Codex | crate::Agent::Grok) { return Err(Error::Unsupported { agent: request.agent, what: "provider-level session interruption", @@ -535,13 +535,14 @@ pub fn stream(request: &Request) -> Result { }; let initial_plan = request.plan(); + let grok_acp = request.agent == crate::Agent::Grok; let codex_app_server = request.agent == crate::Agent::Codex && (initial_plan.duplex || initial_plan.approvals); // Written before the argv is built, because the argv has to name it. The // app-server protocol accepts the schema inline instead. let schema_file = match (&request.schema, request.agent.caps().schema) { - (Some(_), _) if codex_app_server => None, + (Some(_), _) if codex_app_server || grok_acp => None, (Some(schema), crate::agent::SchemaSupport::File) => { Some(SchemaFile::write(schema).map_err(|source| Error::Spawn { bin: request.agent.bin().to_string(), @@ -562,16 +563,18 @@ pub fn stream(request: &Request) -> Result { let mut command = Command::new(&argv[0]); command .args(&argv[1..]) - .stdin(if plan.stdin_prompt || plan.duplex || plan.approvals { - // An interactive run needs stdin for the whole turn, not just to - // deliver a prompt: it is the channel follow-up messages and - // approval decisions travel back on. - Stdio::piped() - } else { - // Close stdin so an agent that would otherwise wait on it exits - // instead of hanging forever with nothing to read. - Stdio::null() - }) + .stdin( + if grok_acp || plan.stdin_prompt || plan.duplex || plan.approvals { + // An interactive run needs stdin for the whole turn, not just to + // deliver a prompt: it is the channel follow-up messages and + // approval decisions travel back on. + Stdio::piped() + } else { + // Close stdin so an agent that would otherwise wait on it exits + // instead of hanging forever with nothing to read. + Stdio::null() + }, + ) .stdout(Stdio::piped()) .stderr(Stdio::piped()) // Without this a killed run can leave the child alive holding the pipes. @@ -643,7 +646,7 @@ pub fn stream(request: &Request) -> Result { let (tx, rx) = mpsc::channel(EVENT_BUFFER); // Only created for an approvals run, so `respond` can tell "no channel" from // "channel closed" and refuse the first rather than hanging on it. - let (decisions_tx, decisions_rx) = if plan.duplex || plan.approvals { + let (decisions_tx, decisions_rx) = if grok_acp || plan.duplex || plan.approvals { let (tx, rx) = mpsc::channel::(APPROVAL_BUFFER); (Some(tx), Some(rx)) } else { @@ -660,7 +663,9 @@ pub fn stream(request: &Request) -> Result { let _session_lease = session_lease; // Moved in so the file outlives the run and is removed with it. let _schema_file = schema_file; - if codex_app_server { + if grok_acp { + drive_grok_acp(child, request, tx, cancel_rx, reaped_for_task, decisions_rx).await + } else if codex_app_server { drive_codex_app_server(child, request, tx, cancel_rx, reaped_for_task, decisions_rx) .await } else { @@ -1115,6 +1120,318 @@ async fn drive( }) } +/// Drive one Grok turn over `grok agent stdio` (ACP). +/// +/// One child per `stream()`, so many Grok sessions and many other backends can +/// run in parallel. Mid-turn input is `_x.ai/interject`. A drop or cancel +/// sends `session/cancel` first so the session stays resumable, same idea as +/// Codex `turn/interrupt`. +#[allow( + clippy::too_many_lines, + reason = "one select loop owns the protocol, control channel, deadline, and child lifecycle" +)] +async fn drive_grok_acp( + child: Child, + request: Request, + events: mpsc::Sender, + cancel: tokio::sync::oneshot::Receiver<()>, + reaped: std::sync::Arc, + controls: Option>, +) -> Result { + let mut child = ChildGuard { child, armed: true }; + let plan = request.plan(); + let bin = plan.bin.clone(); + let Some(mut stdin) = child.child.stdin.take() else { + return Err(Error::Spawn { + bin, + source: std::io::Error::other("stdin was not piped for Grok ACP"), + }); + }; + let Some(stdout) = child.child.stdout.take() else { + return Err(Error::Spawn { + bin, + source: std::io::Error::other("stdout was not piped for Grok ACP"), + }); + }; + let mut controls = if let Some(controls) = controls { + controls + } else { + let (_tx, rx) = mpsc::channel(1); + rx + }; + + let stderr = child.child.stderr.take(); + let stderr_task = tokio::spawn(async move { + let mut buf = String::new(); + if let Some(handle) = stderr { + let mut reader = BufReader::new(handle); + let mut line = String::new(); + while let Ok(Some(_)) = read_bounded_line(&mut reader, &mut line).await { + append_capped(&mut buf, &line); + } + } + buf + }); + + let mut protocol = crate::grok_acp::Protocol::new(request.clone()); + for opening in crate::grok_acp::Protocol::opening() { + stdin + .write_all(opening.as_bytes()) + .await + .map_err(|source| Error::Spawn { + bin: bin.clone(), + source, + })?; + } + stdin.flush().await.map_err(|source| Error::Spawn { + bin: bin.clone(), + source, + })?; + + let mut reader = BufReader::new(stdout); + let mut line = String::new(); + let mut raw = String::new(); + let mut pending = VecDeque::new(); + let mut steer_receipts = HashMap::new(); + let mut bound = false; + let mut persist_result: Result<()> = Ok(()); + let deadline = async { + match request.timeout { + Some(limit) => tokio::time::sleep(limit).await, + None => std::future::pending().await, + } + }; + tokio::pin!(deadline); + tokio::pin!(cancel); + + while !protocol.finished { + tokio::select! { + biased; + control = controls.recv() => { + let Some(control) = control else { + continue; + }; + pending.push_back(control); + flush_grok_controls( + &mut protocol, + &mut pending, + &mut steer_receipts, + &mut stdin, + &bin, + ).await?; + stdin.flush().await.map_err(|source| Error::Spawn { + bin: bin.clone(), source + })?; + } + record = read_bounded_line(&mut reader, &mut line) => { + if record.map_err(|source| Error::Spawn { bin: bin.clone(), source })?.is_some() { + append_capped(&mut raw, &line); + if let Ok(value) = serde_json::from_str::(&line) { + let step = protocol.push(&value); + settle_grok_steers(&mut steer_receipts, step.steer_responses, &bin); + for event in step.events { + if let Event::Started { session, .. } = &event + && !bound + { + bound = true; + persist_result = persist_session(&request, session); + } + let _ = events.send(event).await; + } + for write in step.writes { + stdin.write_all(write.as_bytes()).await.map_err(|source| { + Error::Spawn { bin: bin.clone(), source } + })?; + } + flush_grok_controls( + &mut protocol, + &mut pending, + &mut steer_receipts, + &mut stdin, + &bin, + ).await?; + stdin.flush().await.map_err(|source| Error::Spawn { + bin: bin.clone(), source + })?; + } else { + protocol.terminal.unparsed += 1; + if protocol.terminal.first_unparsed.is_none() { + protocol.terminal.first_unparsed = Some(line.clone()); + } + } + } else { + protocol.failure.get_or_insert_with(|| { + "Grok ACP closed stdout before the turn settled".to_string() + }); + protocol.finished = true; + } + } + () = &mut deadline => { + let partial = protocol.terminal.text.clone(); + interrupt_grok_turn(&mut protocol, &mut stdin, &mut reader, &mut line, &mut raw) + .await; + shut_down(&mut child, stderr_task).await; + reaped.store(true, std::sync::atomic::Ordering::SeqCst); + return Err(Error::Timeout { + bin, + timeout: request.timeout.unwrap_or_default(), + partial, + }); + } + _ = &mut cancel => { + interrupt_grok_turn(&mut protocol, &mut stdin, &mut reader, &mut line, &mut raw) + .await; + shut_down(&mut child, stderr_task).await; + reaped.store(true, std::sync::atomic::Ordering::SeqCst); + return Err(Error::Cancelled { bin }); + } + } + } + + drop(stdin); + if tokio::time::timeout(std::time::Duration::from_secs(2), child.child.wait()) + .await + .is_err() + { + kill_process_group(&child.child); + let _ = child.child.kill().await; + } + child.armed = false; + reaped.store(true, std::sync::atomic::Ordering::SeqCst); + drop(events); + let stderr = stderr_task.await.unwrap_or_default(); + + persist_result?; + if let Some(detail) = protocol.failure { + return Err(Error::Parse { + agent: request.agent, + detail, + }); + } + + let terminal = protocol.terminal; + if terminal.stop == Stop::Error { + return Err(classify_run( + request.agent, + &bin, + 0, + &stderr, + &raw, + &terminal, + )); + } + let structured = terminal.structured.clone().or_else(|| { + request + .schema + .as_ref() + .and_then(|_| serde_json::from_str(&terminal.text).ok()) + }); + Ok(Outcome { + agent: request.agent, + session: terminal.session, + text: terminal.text, + usage: terminal.usage, + stop: terminal.stop, + rate_limit: terminal.rate_limit, + exit_code: 0, + stderr, + unparsed: terminal.unparsed, + first_unparsed: terminal.first_unparsed, + structured, + }) +} + +async fn interrupt_grok_turn( + protocol: &mut crate::grok_acp::Protocol, + stdin: &mut tokio::process::ChildStdin, + reader: &mut BufReader, + line: &mut String, + raw: &mut String, +) { + let Some(encoded) = protocol.interrupt() else { + return; + }; + if stdin.write_all(encoded.as_bytes()).await.is_err() || stdin.flush().await.is_err() { + return; + } + let settle = async { + while !protocol.finished { + let Ok(Some(_)) = read_bounded_line(reader, line).await else { + break; + }; + append_capped(raw, line); + if let Ok(value) = serde_json::from_str::(line) { + let _ = protocol.push(&value); + } + } + }; + let _ = tokio::time::timeout(std::time::Duration::from_secs(2), settle).await; +} + +async fn flush_grok_controls( + protocol: &mut crate::grok_acp::Protocol, + pending: &mut VecDeque, + steer_receipts: &mut HashMap>>, + stdin: &mut tokio::process::ChildStdin, + bin: &str, +) -> Result<()> { + let mut waiting = VecDeque::new(); + while let Some(control) = pending.pop_front() { + let encoded = match control { + Control::Message { body, receipt } => { + if let Some(request) = protocol.steer(&body) { + steer_receipts.insert(request.id, receipt); + Some(request.wire) + } else { + waiting.push_back(Control::Message { body, receipt }); + None + } + } + Control::Approval { id, decision } => { + if let Some(encoded) = protocol.respond(&id, &decision) { + Some(encoded) + } else { + waiting.push_back(Control::Approval { id, decision }); + None + } + } + }; + if let Some(encoded) = encoded { + stdin + .write_all(encoded.as_bytes()) + .await + .map_err(|source| Error::Spawn { + bin: bin.to_string(), + source, + })?; + } + } + pending.append(&mut waiting); + Ok(()) +} + +fn settle_grok_steers( + receipts: &mut HashMap>>, + responses: Vec, + bin: &str, +) { + for response in responses { + let Some(receipt) = receipts.remove(&response.id) else { + continue; + }; + let result = response + .result + .map(|_| ()) + .map_err(|message| Error::AgentError { + agent: crate::Agent::Grok, + bin: bin.to_string(), + status: None, + message, + }); + let _ = receipt.send(result); + } +} + /// Drive one interactive Codex turn over app-server's JSON-RPC transport. /// /// Unlike `codex exec`, app-server remains alive after a turn completes. This @@ -2095,6 +2412,7 @@ mod tests { (Agent::Claude, "setup-token"), (Agent::Codex, "codex login"), (Agent::Copilot, "copilot login"), + (Agent::Grok, "grok login"), ] { let err = classify_run( agent, diff --git a/tests/live.rs b/tests/live.rs index 5e55d90..3017bd6 100644 --- a/tests/live.rs +++ b/tests/live.rs @@ -41,7 +41,7 @@ fn ping(agent: Agent) -> Request { match agent { // The cheapest model on each side; Codex and Copilot pick their own. Agent::Claude => request.model("haiku"), - Agent::Codex | Agent::Copilot => request, + Agent::Codex | Agent::Copilot | Agent::Grok => request, } } From 36d92ee9be5e318a2aa50cf7b2f987a5c1ce1a5b Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 15 Sep 2026 02:22:59 +0700 Subject: [PATCH 2/4] feat(grok): drive Grok over ACP with plan mode off The ACP transport has no TUI, so `enter_plan_mode` and `exit_plan_mode` hang until the host's liveness ping aborts the turn (session 01a09bd5). Pass `--no-plan` so the model writes plans as ordinary assistant text. Also carries `used_percent` on `RateLimit`: Grok reports weekly window fill through `x.ai/session/usage`, which Claude's in-run rate limit object does not have. Defaulted to None everywhere else, so no provider changes shape. --- src/agent.rs | 4 + src/event.rs | 3 + src/grok_acp.rs | 806 ++++++++++++++++++++++++++++++++++++++++++++++-- src/outcome.rs | 7 + src/run.rs | 5 + 5 files changed, 795 insertions(+), 30 deletions(-) diff --git a/src/agent.rs b/src/agent.rs index 8b1ae53..8ded80b 100644 --- a/src/agent.rs +++ b/src/agent.rs @@ -1122,6 +1122,9 @@ fn argv_copilot(plan: &Plan) -> Vec { fn argv_grok(plan: &Plan) -> Vec { let mut a = Argv::new(&plan.bin); a.bare("--no-auto-update"); + // ACP has no TUI for plan approval. `enter_plan_mode` / `exit_plan_mode` + // hang until the host's liveness ping aborts the turn (session 01a09bd5). + a.bare("--no-plan"); a.pair("--permission-mode", grok_mode(plan.permission)); if plan.permission == Permission::ReadOnly { // Internal ids from grok's own headless docs (`--disallowed-tools @@ -1886,6 +1889,7 @@ mod tests { assert!(pos(&a, "--reasoning-effort").unwrap() > agent_at); assert_eq!(a[pos(&a, "--reasoning-effort").unwrap() + 1], "high"); assert!(pos(&a, "--rules").unwrap() < agent_at); + assert!(pos(&a, "--no-plan").unwrap() < agent_at); assert!(!a.iter().any(|arg| arg == "--always-approve")); assert!( !a.iter().any(|arg| arg == "hi"), diff --git a/src/event.rs b/src/event.rs index 5617dcd..2a042a6 100644 --- a/src/event.rs +++ b/src/event.rs @@ -321,6 +321,7 @@ fn enforce_bounds(event: Event) -> Event { resets_at: limit.resets_at, overage_status: limit.overage_status.map(bound_identifier), is_using_overage: limit.is_using_overage, + used_percent: limit.used_percent, }), // The agent's own refusal sentence, bounded like any other prose it // hands back. @@ -1271,6 +1272,7 @@ fn claude_rate_limit(v: Option<&Value>) -> Option { .and_then(Value::as_str) .map(str::to_string), is_using_overage: v.get("isUsingOverage").and_then(Value::as_bool), + used_percent: None, }) } @@ -1961,6 +1963,7 @@ mod tests { resets_at: Some(1_785_260_400), overage_status: None, is_using_overage: None, + used_percent: None, }; assert!(events.contains(&Event::RateLimit(limit.clone()))); assert_eq!(term.rate_limit, Some(limit.clone())); diff --git a/src/grok_acp.rs b/src/grok_acp.rs index ed212ed..862c889 100644 --- a/src/grok_acp.rs +++ b/src/grok_acp.rs @@ -24,16 +24,30 @@ use serde_json::{Value, json}; use crate::agent::{Continue, Permission}; use crate::approval::{Approval, Decision}; +use crate::command::Compaction; use crate::event::{Event, Terminal, append_capped}; -use crate::outcome::{Stop, Usage}; +use crate::outcome::{RateLimit, Stop, Usage}; use crate::request::Request; +/// Grok 4.6 / 4.5 context window when the payload omits `size`. +const GROK_CONTEXT_WINDOW: u64 = 500_000; + const INIT_ID: u64 = 1; const SESSION_ID: u64 = 2; const PROMPT_ID: u64 = 3; const COMPACT_ID: u64 = 4; +const USAGE_ID: u64 = 5; +const INFO_ID: u64 = 6; const NEXT_ID: u64 = 10; +/// Piggyback `x.ai/session/usage` at most once a minute across turns. +static LAST_USAGE_SECS: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); +const USAGE_INTERVAL_SECS: u64 = 60; +/// Mid-turn occupancy polls. Grok has no `usage_update`; this is how the +/// chip grows while tools run. Once per 15s, and never overlapping, so a +/// tool-heavy turn does not queue occupancy RPCs ahead of tools. +const INFO_POLL_SECS: u64 = 15; + /// One decoded ACP record. #[derive(Debug, Default)] pub(crate) struct Step { @@ -76,6 +90,15 @@ pub(crate) struct Protocol { next_id: u64, compacting: bool, last_steer: Option, + /// `session/load` (and `session/new`) replay history as `session/update` + /// before the RPC returns. AZ already has that transcript. Emitting it + /// again duplicates the top of the thread on reconnect. + replaying: bool, + usage_fallback: bool, + info_fallback: bool, + /// In-flight `_x.ai/session/info` polls (not the end-of-turn `INFO_ID`). + pending_info: HashSet, + last_info_secs: u64, } impl Protocol { @@ -92,6 +115,11 @@ impl Protocol { next_id: NEXT_ID, compacting: false, last_steer: None, + replaying: true, + usage_fallback: false, + info_fallback: false, + pending_info: HashSet::new(), + last_info_secs: 0, } } @@ -249,17 +277,44 @@ impl Protocol { SESSION_ID => return self.push_session(value, &mut step), PROMPT_ID => { self.finish_prompt(value); + self.request_occupancy_then_billing(&mut step); + return step; + } + INFO_ID => { + if let Some(usage) = grok_session_info(value) { + self.terminal.usage.accumulate(&usage); + step.events.push(Event::Usage(usage)); + } + self.request_billing_or_finish(&mut step); + return step; + } + USAGE_ID => { + if let Some(limit) = grok_session_usage(value) { + self.terminal.rate_limit = Some(limit.clone()); + step.events.push(Event::RateLimit(limit)); + } + self.finished = true; return step; } COMPACT_ID => { self.compacting = false; self.terminal.stop = Stop::Completed; - self.finished = true; step.events .push(Event::Compaction(crate::command::Compaction::Finished { ok: true, error: None, })); + // Post-compact occupancy is `session/info.context.used` + // (tokens_after), not the compact turn's billed input and + // not AZ's 8k estimate. + self.request_occupancy_then_billing(&mut step); + return step; + } + other if self.pending_info.remove(&other) => { + if let Some(usage) = grok_session_info(value) { + self.terminal.usage.accumulate(&usage); + step.events.push(Event::Usage(usage)); + } return step; } other if self.pending_steers.contains(&other) => { @@ -277,12 +332,23 @@ impl Protocol { let method = value.get("method").and_then(Value::as_str).unwrap_or(""); match method { "session/update" | "x.ai/session/update" | "_x.ai/session/update" => { - step.events.extend(self.session_update(value)); + if !self.replaying { + step.events.extend(self.session_update(value)); + if occupancy_tick(value) { + if let Some(wire) = self.poll_occupancy() { + step.writes.push(wire); + } + } + } } - "session/request_permission" => { - if let Some(event) = self.permission_request(value) { + "session/request_permission" | "x.ai/session/request_permission" => { + let asked = self.permission_request(value); + if let Some(event) = asked.event { step.events.push(event); } + if let Some(write) = asked.write { + step.writes.push(write); + } } _ => {} } @@ -328,6 +394,34 @@ impl Protocol { }))); return std::mem::take(step); } + if id == INFO_ID { + if looks_like_method_not_found(&error) && !self.info_fallback { + if let Some(session_id) = self.session_id.clone() { + self.info_fallback = true; + step.writes.push(info_wire(&session_id, INFO_ID, true)); + return std::mem::take(step); + } + } + // Occupancy is optional. Fall through to weekly % / finish. + self.request_billing_or_finish(step); + return std::mem::take(step); + } + if id == USAGE_ID { + if looks_like_method_not_found(&error) && !self.usage_fallback { + if let Some(session_id) = self.session_id.clone() { + self.usage_fallback = true; + step.writes.push(usage_wire(&session_id, true)); + return std::mem::take(step); + } + } + // Weekly % is optional. A miss must not fail the turn. + self.finished = true; + return std::mem::take(step); + } + if self.pending_info.remove(&id) { + // A mid-turn occupancy poll failed. The turn is still live. + return std::mem::take(step); + } if self.pending_steers.remove(&id) { if looks_like_method_not_found(&error) { if let Some(retry) = self.interject_fallback() { @@ -368,6 +462,7 @@ impl Protocol { }; self.session_id = Some(session_id.clone()); self.terminal.session = Some(session_id.clone()); + self.replaying = false; let model = value .pointer("/result/models/current/id") .or_else(|| value.pointer("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/result/model")) @@ -395,6 +490,9 @@ impl Protocol { self.finished = true; } else { step.writes.push(next); + if let Some(wire) = self.poll_occupancy() { + step.writes.push(wire); + } if self.request.operation == crate::request::Operation::Interrupt { self.interrupt_requested = true; self.terminal.stop = Stop::Other("interrupted".into()); @@ -420,9 +518,70 @@ impl Protocol { .pointer("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/result/usage") .or_else(|| value.pointer("/result/_meta/usage")) { - self.terminal.usage = grok_usage(usage); + // Accumulate: a replace would wipe occupancy learned from + // `auto_compact_completed` or a 1-call `turn_completed` when the + // prompt result only carries the turn's billed aggregate. + self.terminal.usage.accumulate(&grok_usage(usage)); } - self.finished = true; + } + + fn request_occupancy_then_billing(&mut self, step: &mut Step) { + if let Some(wire) = self.maybe_info_request() { + step.writes.push(wire); + return; + } + self.request_billing_or_finish(step); + } + + fn request_billing_or_finish(&mut self, step: &mut Step) { + if let Some(wire) = self.maybe_usage_request() { + step.writes.push(wire); + } else { + self.finished = true; + } + } + + fn maybe_info_request(&mut self) -> Option { + let session_id = self.session_id.as_ref()?; + self.last_info_secs = unix_secs(); + Some(info_wire(session_id, INFO_ID, self.info_fallback)) + } + + /// Live occupancy during a turn. Grok never emits `usage_update`; the + /// only occupancy RPC is `_x.ai/session/info`. Polled after tool results + /// so the header can tick up instead of jumping at `turn_completed`. + fn poll_occupancy(&mut self) -> Option { + let session_id = self.session_id.as_ref()?; + if !self.pending_info.is_empty() { + return None; + } + let now = unix_secs(); + if self.last_info_secs != 0 && now.saturating_sub(self.last_info_secs) < INFO_POLL_SECS { + return None; + } + let id = self.next_id; + self.next_id += 1; + self.pending_info.insert(id); + self.last_info_secs = now.max(1); + Some(info_wire(session_id, id, self.info_fallback)) + } + + fn should_fetch_usage(&self) -> bool { + self.session_id.is_some() && { + let now = unix_secs(); + let last = LAST_USAGE_SECS.load(std::sync::atomic::Ordering::Relaxed); + now.saturating_sub(last) >= USAGE_INTERVAL_SECS + } + } + + fn maybe_usage_request(&mut self) -> Option { + let session_id = self.session_id.as_ref()?; + if !self.should_fetch_usage() { + self.finished = true; + return None; + } + LAST_USAGE_SECS.store(unix_secs(), std::sync::atomic::Ordering::Relaxed); + Some(usage_wire(session_id, false)) } fn session_update(&mut self, value: &Value) -> Vec { @@ -490,18 +649,47 @@ impl Protocol { skills: Vec::new(), })] } - "usage_update" => { - let usage = grok_usage(&update); + "usage_update" | "turn_completed" => { + // Grok puts the turn's usage on `_x.ai/session/update` + // `turn_completed`, not on ACP `usage_update` and not as + // `costUsd`. Ticks are 1e-9 USD (session 1.626e9 ticks ≈ $1.63). + let payload = update.get("usage").unwrap_or(&update); + let usage = grok_usage(payload); self.terminal.usage.accumulate(&usage); vec![Event::Usage(usage)] } + "auto_compact_completed" => { + // Live occupancy after Grok's own compact. Session 01a09ca7: + // `tokens_before` 179_555 / `tokens_after` 10_201 — not the + // turn's billed `totalTokens`. + let after = update + .get("tokens_after") + .or_else(|| update.get("tokensAfter")) + .and_then(Value::as_u64); + let mut events = vec![Event::Compaction(Compaction::Finished { + ok: true, + error: None, + })]; + if let Some(after) = after { + let usage = Usage { + context_tokens: Some(after), + context_window: Some(GROK_CONTEXT_WINDOW), + ..Usage::default() + }; + self.terminal.usage.accumulate(&usage); + events.push(Event::Usage(usage)); + } + events + } _ => Vec::new(), } } - fn permission_request(&mut self, value: &Value) -> Option { - let id = value.get("id")?.clone(); - let params = value.get("params")?; + fn permission_request(&mut self, value: &Value) -> PermissionStep { + let Some(id) = value.get("id").cloned() else { + return PermissionStep::default(); + }; + let params = value.get("params").unwrap_or(&Value::Null); let tool_call = params.get("toolCall").unwrap_or(params); let tool = tool_name(tool_call); let input = tool_call @@ -522,11 +710,28 @@ impl Protocol { options, }, ); - Some(Event::ApprovalRequest(Approval { - id: key, - tool, - input, - })) + // Auto/Bypass must answer on the ACP pipe. Grok still emits + // `session/request_permission` for writes outside the workspace + // (session 01a09ca7: `~/.grok/config.toml` sat 30 minutes). Leaving + // the question for the host — and then not auto-answering it — + // stalls the turn until APPROVAL_TIMEOUT. + if matches!( + self.request.permission, + Permission::Auto | Permission::Bypass + ) { + return PermissionStep { + event: None, + write: self.respond(&key, &Decision::Allow), + }; + } + PermissionStep { + event: Some(Event::ApprovalRequest(Approval { + id: key, + tool, + input, + })), + write: None, + } } /// Encode mid-turn input. Does not cancel the turn. @@ -573,7 +778,8 @@ impl Protocol { pub fn respond(&mut self, id: &str, decision: &Decision) -> Option { let pending = self.pending.remove(id)?; - let option_id = pick_option(&pending.options, decision); + let option_id = pick_option(&pending.options, decision) + .or_else(|| matches!(decision, Decision::Allow).then(|| "allow-once".to_string())); let result = match option_id { Some(option_id) => json!({ "outcome": { "outcome": "selected", "optionId": option_id } @@ -656,6 +862,231 @@ fn command_names(update: &Value) -> Vec { .unwrap_or_default() } +fn unix_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +fn info_wire(session_id: &str, id: u64, fallback: bool) -> String { + // Live grok 1.0.30: `_x.ai/session/info` returns + // `result.context.{used,total}` — occupancy vs the window. Bare + // `x.ai/session/info` is -32601. `_x.ai/session/usage` is billed totals. + let method = if fallback { + "x.ai/session/info" + } else { + "_x.ai/session/info" + }; + wire(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": { "sessionId": session_id }, + })) +} + +fn occupancy_tick(value: &Value) -> bool { + let kind = value + .pointer("/params/update/sessionUpdate") + .or_else(|| value.pointer("/params/update/session_update")) + .or_else(|| value.pointer("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/params/sessionUpdate")) + .and_then(Value::as_str) + .unwrap_or(""); + match kind { + "tool_call_update" => { + let status = value + .pointer("/params/update/status") + .or_else(|| value.pointer("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/params/status")) + .and_then(Value::as_str) + .unwrap_or(""); + status == "completed" || status == "failed" + } + "auto_compact_completed" => true, + _ => false, + } +} + +fn usage_wire(session_id: &str, fallback: bool) -> String { + // Live grok 1.0.30: `_x.ai/billing` returns weekly allowance + // (`creditUsagePercent`). `_x.ai/session/usage` is session token totals + // only — not the header chip. + let method = if fallback { + "x.ai/billing" + } else { + "_x.ai/billing" + }; + wire(&json!({ + "jsonrpc": "2.0", + "id": USAGE_ID, + "method": method, + "params": { "sessionId": session_id }, + })) +} + +fn json_f64(value: &Value) -> Option { + value + .as_f64() + .or_else(|| value.as_u64().map(|n| n as f64)) + .or_else(|| value.as_i64().map(|n| n as f64)) +} + +fn first_f64(root: &Value, keys: &[&str]) -> Option { + for key in keys { + if let Some(n) = root.get(*key).and_then(json_f64) { + return Some(n); + } + } + None +} + +fn first_reset(root: &Value) -> Option { + for key in [ + "resetsAt", + "resets_at", + "resetAt", + "reset_at", + "resetAtUnix", + "billingPeriodEnd", + "end", + ] { + if let Some(n) = root.get(key).and_then(Value::as_i64) { + return Some(n); + } + if let Some(s) = root.get(key).and_then(Value::as_str) { + if let Ok(n) = s.parse::() { + return Some(n); + } + if let Some(n) = parse_iso_utc(s) { + return Some(n); + } + } + } + None +} + +fn parse_iso_utc(s: &str) -> Option { + if s.len() < 19 { + return None; + } + let y: i64 = s[0..4].parse().ok()?; + let m: i64 = s[5..7].parse().ok()?; + let d: i64 = s[8..10].parse().ok()?; + let hh: i64 = s[11..13].parse().ok()?; + let mm: i64 = s[14..16].parse().ok()?; + let ss: i64 = s[17..19].parse().ok()?; + if !(1..=12).contains(&m) || d == 0 || d > 31 { + return None; + } + let mut y = y; + let mut m = m; + if m <= 2 { + y -= 1; + m += 9; + } else { + m -= 3; + } + let era = if y >= 0 { y } else { y - 399 } / 400; + let yoe = y - era * 400; + let doy = (153 * m + 2) / 5 + d - 1; + let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; + let days = era * 146097 + doe - 719468; + Some(days * 86400 + hh * 3600 + mm * 60 + ss) +} + +/// Live occupancy from `_x.ai/session/info`. +/// +/// Probe 2026-09-14 grok 1.0.30: +/// `{result:{result:{context:{used:1475,total:500000,...}}}}`. +/// This is the status-line figure (`signals.json` `contextTokensUsed`), not +/// the turn's billed `totalTokens`. +fn grok_session_info(value: &Value) -> Option { + let result = value.get("result").unwrap_or(value); + let inner = result.get("result").unwrap_or(result); + let ctx = inner.get("context")?; + let used = first_u64(ctx, &["used", "contextTokensUsed", "context_tokens"])?; + let total = first_u64( + ctx, + &["total", "contextWindowTokens", "size", "context_window"], + ) + .unwrap_or(GROK_CONTEXT_WINDOW); + if total == 0 { + return None; + } + Some(Usage { + context_tokens: Some(used), + context_window: Some(total), + ..Usage::default() + }) +} + +fn first_u64(root: &Value, keys: &[&str]) -> Option { + for key in keys { + if let Some(n) = root.get(*key).and_then(Value::as_u64) { + return Some(n); + } + if let Some(n) = root.get(*key).and_then(Value::as_f64) { + if n >= 0.0 { + return Some(n as u64); + } + } + } + None +} + +/// Weekly allowance from `_x.ai/billing`. Live grok 1.0.30: +/// `{config:{creditUsagePercent, billingPeriodEnd}, subscription_tier}`. +fn grok_session_usage(value: &Value) -> Option { + let result = value.get("result").unwrap_or(value); + let config = result.get("config").unwrap_or(result); + let candidates = [ + config, + result, + result.get("usage").unwrap_or(result), + result.get("limit").unwrap_or(result), + result.get("weekly").unwrap_or(result), + result.get("billing").unwrap_or(result), + result.get("billingCycle").unwrap_or(result), + ]; + for obj in candidates { + let mut percent = first_f64( + obj, + &[ + "creditUsagePercent", + "usedPercent", + "used_percent", + "percentUsed", + "percent_used", + "utilization", + ], + ); + if let (None, Some(used), Some(limit)) = ( + percent, + first_f64(obj, &["used", "usedCredits", "spent"]), + first_f64(obj, &["limit", "allowance", "cap", "max"]), + ) { + if limit > 0.0 { + percent = Some(used / limit * 100.0); + } + } + let Some(mut percent) = percent else { + continue; + }; + if percent <= 1.0 { + percent *= 100.0; + } + return Some(RateLimit { + status: "allowed".into(), + window: Some("weekly".into()), + resets_at: first_reset(obj).or_else(|| first_reset(result)), + overage_status: None, + is_using_overage: None, + used_percent: Some(percent), + }); + } + None +} + fn grok_usage(usage: &Value) -> Usage { let num = |camel: &str, snake: &str| { usage @@ -663,21 +1094,67 @@ fn grok_usage(usage: &Value) -> Usage { .or_else(|| usage.get(snake)) .and_then(Value::as_u64) }; + let input = num("inputTokens", "input_tokens"); + let cache_read = num("cachedReadTokens", "cache_read_input_tokens"); + // Grok's `inputTokens` includes the cached prefix (Codex shape). Fresh + // input is the remainder; stuffing the raw sum here double-counts cache + // in every host that also reads `cache_read_tokens`. + let fresh = match (input, cache_read) { + (Some(input), Some(cached)) if input >= cached => Some(input - cached), + (Some(input), _) => Some(input), + _ => None, + }; + let window = num("size", "context_window") + .or_else(|| num("contextWindowTokens", "context_window_tokens")) + .unwrap_or(GROK_CONTEXT_WINDOW); Usage { - input_tokens: num("inputTokens", "input_tokens"), + input_tokens: fresh, output_tokens: num("outputTokens", "output_tokens"), - cache_read_tokens: num("cachedReadTokens", "cache_read_input_tokens"), + cache_read_tokens: cache_read, cache_write_tokens: num("cacheCreationTokens", "cache_creation_input_tokens"), - context_tokens: num("totalTokens", "total_tokens").or_else(|| num("used", "used")), - context_window: num("size", "context_window"), + context_tokens: grok_occupancy(usage, window), + context_window: Some(window), + reasoning_tokens: num("reasoningTokens", "reasoning_tokens"), cost_usd: usage - .get("cost") - .and_then(Value::as_f64) + .get("costUsdTicks") + .or_else(|| usage.get("cost_usd_ticks")) + .and_then(|value| value.as_f64().or_else(|| value.as_u64().map(|n| n as f64))) + // CLI `costUsdTicks` is 1e-9 USD (session 01a09ca7: 1.644e9 ticks). + // xAI API `cost_in_usd_ticks` is 1e-10; do not mix the two. + .map(|ticks| ticks / 1_000_000_000.0) + .or_else(|| usage.get("cost").and_then(Value::as_f64)) .or_else(|| usage.get("costUsd").and_then(Value::as_f64)), ..Usage::default() } } +/// Live window fill, never the turn's billed sum. +/// +/// `turn_completed.usage.totalTokens` / `inputTokens` are billed sums. +/// A 1-call compact or notes pass reports the *old* window as `inputTokens` +/// (session 01a09ca7: 264k after compact, live fill 29k). Treating that as +/// occupancy made the next prompt look like 260k and re-trigger cliff +/// steers. Occupancy is only `used` / `contextTokensUsed` on the payload, +/// `_x.ai/session/info`, or `auto_compact_completed.tokens_after`. +fn grok_occupancy(usage: &Value, window: u64) -> Option { + let num = |camel: &str, snake: &str| { + usage + .get(camel) + .or_else(|| usage.get(snake)) + .and_then(Value::as_u64) + }; + let plausible = |used: u64| used > 0 && used <= window.saturating_mul(2); + num("used", "context_tokens") + .or_else(|| num("contextTokensUsed", "context_tokens_used")) + .filter(|&used| plausible(used)) +} + +#[derive(Default)] +struct PermissionStep { + event: Option, + write: Option, +} + fn value_as_text(value: &Value) -> String { match value { Value::String(text) => text.clone(), @@ -688,6 +1165,14 @@ fn value_as_text(value: &Value) -> String { fn pick_option(options: &Value, decision: &Decision) -> Option { let entries = options.as_array()?; let want_allow = matches!(decision, Decision::Allow); + let id_of = |entry: &Value| { + entry + .get("optionId") + .or_else(|| entry.get("option_id")) + .or_else(|| entry.get("id")) + .and_then(Value::as_str) + .map(str::to_string) + }; for entry in entries { let kind = entry .get("kind") @@ -699,12 +1184,7 @@ fn pick_option(options: &Value, decision: &Decision) -> Option { let allow = kind.contains("allow") || kind.contains("approve"); let deny = kind.contains("reject") || kind.contains("deny"); if want_allow && allow || !want_allow && deny { - return entry - .get("optionId") - .or_else(|| entry.get("option_id")) - .or_else(|| entry.get("id")) - .and_then(Value::as_str) - .map(str::to_string); + return id_of(entry); } } None @@ -780,10 +1260,32 @@ mod tests { assert!(!wire.contains("session/prompt")); } + #[test] + fn load_replay_does_not_emit_history_as_a_new_turn() { + let mut p = protocol(); + let replay = json!({ + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "old reply" } + } + } + }); + assert!(p.push(&replay).events.is_empty()); + let _ = p.push_session( + &json!({"jsonrpc":"2.0","id":2,"result":{"sessionId":"sess-1"}}), + &mut Step::default(), + ); + let live = p.push(&replay); + assert_eq!(live.events, vec![Event::Text("old reply".into())]); + } + #[test] fn agent_text_chunks_become_events() { let mut p = protocol(); p.session_id = Some("sess-1".into()); + p.replaying = false; let events = p.session_update(&json!({ "method": "session/update", "params": { @@ -797,4 +1299,248 @@ mod tests { assert_eq!(events, vec![Event::Text("pong".into())]); assert_eq!(p.terminal.text.trim_end(), "pong"); } + + #[test] + fn turn_completed_usage_converts_cost_ticks() { + let mut p = protocol(); + p.replaying = false; + let events = p.session_update(&json!({ + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "turn_completed", + "usage": { + "inputTokens": 18033, + "outputTokens": 189, + "cachedReadTokens": 0, + "costUsdTicks": 126480000 + } + } + } + })); + match events.as_slice() { + [Event::Usage(usage)] => { + assert_eq!(usage.input_tokens, Some(18033)); + assert_eq!(usage.output_tokens, Some(189)); + // Billed input is not occupancy; session/info supplies that. + assert_eq!(usage.context_tokens, None); + assert_eq!(usage.context_window, Some(500_000)); + let cost = usage.cost_usd.expect("ticks convert"); + assert!((cost - 0.12648).abs() < 1e-9); + } + other => panic!("expected Usage, got {other:?}"), + } + } + + #[test] + fn grok_usage_uses_occupancy_not_turn_aggregate() { + // Session 01a09ca7 turn 1: billed input includes cache; live context + // was 71,933. totalTokens 593,447 is the in-turn sum and must not + // become context_tokens. + let usage = grok_usage(&json!({ + "inputTokens": 579_290, + "outputTokens": 1_644, + "cachedReadTokens": 506_624, + "totalTokens": 593_447, + "used": 71_933, + "costUsdTicks": 1_644_192_400u64 + })); + assert_eq!(usage.input_tokens, Some(72_666)); + assert_eq!(usage.cache_read_tokens, Some(506_624)); + assert_eq!(usage.context_tokens, Some(71_933)); + assert_eq!(usage.context_window, Some(500_000)); + let cost = usage.cost_usd.expect("ticks convert"); + assert!((cost - 1.644_192_4).abs() < 1e-9); + } + + #[test] + fn grok_usage_leaves_context_unset_without_occupancy() { + let usage = grok_usage(&json!({ + "inputTokens": 579_290, + "cachedReadTokens": 506_624, + "totalTokens": 593_447, + "modelCalls": 12 + })); + assert_eq!(usage.input_tokens, Some(72_666)); + assert_eq!(usage.context_tokens, None); + assert_eq!(usage.context_window, Some(500_000)); + } + + #[test] + fn grok_usage_one_call_input_is_not_occupancy() { + // Compact/learn are 1-call turns whose input is the old window. + let usage = grok_usage(&json!({ + "inputTokens": 264_675, + "outputTokens": 2_045, + "cachedReadTokens": 0, + "totalTokens": 266_720, + "modelCalls": 1 + })); + assert_eq!(usage.context_tokens, None); + assert_eq!(usage.context_window, Some(500_000)); + } + + #[test] + fn grok_usage_rejects_billed_sum_as_occupancy() { + // Session 01a09ca7 turn 4: 65 calls, 7.6M billed, live context ~180k. + let usage = grok_usage(&json!({ + "inputTokens": 7_588_418, + "cachedReadTokens": 7_292_032, + "totalTokens": 7_641_353, + "modelCalls": 65 + })); + assert_eq!(usage.context_tokens, None); + assert_eq!(usage.context_window, Some(500_000)); + } + + #[test] + fn auto_compact_completed_sets_live_occupancy() { + let mut p = protocol(); + p.replaying = false; + let events = p.session_update(&json!({ + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "auto_compact_completed", + "tokens_before": 179_555, + "tokens_after": 10_201 + } + } + })); + assert!(events.iter().any(|event| matches!( + event, + Event::Compaction(Compaction::Finished { + ok: true, + error: None + }) + ))); + match events.iter().find(|event| matches!(event, Event::Usage(_))) { + Some(Event::Usage(usage)) => { + assert_eq!(usage.context_tokens, Some(10_201)); + assert_eq!(usage.context_window, Some(500_000)); + } + other => panic!("expected Usage after compact, got {other:?}"), + } + } + + #[test] + fn auto_answers_permission_without_asking_the_host() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + let step = p.push(&json!({ + "jsonrpc": "2.0", + "id": 99, + "method": "session/request_permission", + "params": { + "toolCall": { "title": "search_replace", "rawInput": { "path": "/Users/revenge/.grok/config.toml" } }, + "options": [ + { "optionId": "allow-once", "kind": "allow_once" }, + { "optionId": "reject-once", "kind": "reject_once" } + ] + } + })); + assert!( + step.events.is_empty(), + "Auto must not surface an approval card" + ); + assert_eq!(step.writes.len(), 1); + assert!(step.writes[0].contains("\"id\":99")); + assert!(step.writes[0].contains("allow-once")); + assert!(!step.writes[0].contains("cancelled")); + } + + #[test] + fn session_usage_percent_becomes_weekly_rate_limit() { + let limit = grok_session_usage(&json!({ + "jsonrpc": "2.0", + "id": 5, + "result": { + "config": { + "creditUsagePercent": 41.0, + "billingPeriodEnd": "2026-09-20T11:48:54Z" + }, + "subscription_tier": "SuperGrok Plus" + } + })) + .expect("percent present"); + assert_eq!(limit.used_percent, Some(41.0)); + assert_eq!(limit.window.as_deref(), Some("weekly")); + assert!(!limit.is_blocking()); + } + + #[test] + fn session_info_occupancy_is_used_vs_total() { + let usage = grok_session_info(&json!({ + "jsonrpc": "2.0", + "id": 6, + "result": { + "result": { + "sessionId": "sess-1", + "context": { + "used": 325_827, + "total": 500_000, + "usagePct": 65 + } + } + } + })) + .expect("occupancy present"); + assert_eq!(usage.context_tokens, Some(325_827)); + assert_eq!(usage.context_window, Some(500_000)); + assert!(usage.input_tokens.is_none()); + } + + #[test] + fn completed_tool_polls_session_info_for_live_occupancy() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + p.replaying = false; + let step = p.push(&json!({ + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "tool_call_update", + "status": "completed", + "toolCallId": "call-1" + } + } + })); + assert_eq!(step.writes.len(), 1); + assert!(step.writes[0].contains("_x.ai/session/info")); + assert!(step.writes[0].contains("\"sessionId\":\"sess-1\"")); + assert!( + !step.writes[0].contains("\"id\":6"), + "mid-turn poll is not the end-of-turn INFO_ID" + ); + let again = p.push(&json!({ + "method": "session/update", + "params": { + "update": { + "sessionUpdate": "tool_call_update", + "status": "completed", + "toolCallId": "call-2" + } + } + })); + assert!( + again.writes.is_empty(), + "occupancy polls are 15s apart and one in flight" + ); + } + + #[test] + fn end_of_turn_requests_session_info_before_billing() { + let mut p = protocol(); + p.session_id = Some("sess-1".into()); + p.replaying = false; + let step = p.push(&json!({ + "jsonrpc": "2.0", + "id": 3, + "result": { "stopReason": "end_turn" } + })); + assert_eq!(step.writes.len(), 1); + assert!(step.writes[0].contains("_x.ai/session/info")); + assert!(step.writes[0].contains("\"id\":6")); + assert!(!p.finished); + } } diff --git a/src/outcome.rs b/src/outcome.rs index 9f7e71b..e34d472 100644 --- a/src/outcome.rs +++ b/src/outcome.rs @@ -163,6 +163,12 @@ pub struct RateLimit { pub overage_status: Option, /// Whether the run was already drawing on overage rather than the plan. pub is_using_overage: Option, + /// 0–100, when the provider reports how full the window is. + /// + /// Grok's `x.ai/session/usage` carries this. Claude's in-run rate-limit + /// object does not. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub used_percent: Option, } impl RateLimit { @@ -287,6 +293,7 @@ mod tests { resets_at: Some(1_785_765_600), overage_status: None, is_using_overage: Some(false), + used_percent: None, }; assert!( !warned.is_blocking(), diff --git a/src/run.rs b/src/run.rs index f3f157a..5bdd411 100644 --- a/src/run.rs +++ b/src/run.rs @@ -2184,6 +2184,7 @@ mod tests { resets_at: None, overage_status: None, is_using_overage: None, + used_percent: None, }), ..Terminal::default() }; @@ -2202,6 +2203,7 @@ mod tests { resets_at: None, overage_status: None, is_using_overage: None, + used_percent: None, }), ..Terminal::default() }; @@ -2264,6 +2266,7 @@ mod tests { resets_at: Some(1_785_331_800), overage_status: None, is_using_overage: None, + used_percent: None, }), ..Terminal::default() }; @@ -2285,6 +2288,7 @@ mod tests { resets_at: None, overage_status: None, is_using_overage: None, + used_percent: None, }), ..Terminal::default() }; @@ -2345,6 +2349,7 @@ mod tests { resets_at: None, overage_status: None, is_using_overage: None, + used_percent: None, }), ..Terminal::default() }; From 57c16fa0d20cc276ec234e0f17fce9948e28533c Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 15 Sep 2026 02:32:07 +0700 Subject: [PATCH 3/4] release: 0.4.21 --- Cargo.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Cargo.toml b/Cargo.toml index b34b03b..c8e647c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "agent-abstraction" -version = "0.4.20" +version = "0.4.21" edition = "2024" # The floor edition 2024 requires, and where the strictest dependencies (uuid, # getrandom) sit. Derived from the dependency graph rather than compile-tested. From e7abb1e3923a0988e0adc0c2814cdbc07728c26a Mon Sep 17 00:00:00 2001 From: meh Date: Tue, 15 Sep 2026 02:35:45 +0700 Subject: [PATCH 4/4] fix(grok): satisfy clippy under -D warnings CI runs `cargo clippy --all-targets -- -D warnings`, which the new ACP module did not pass: numeric separators, `map(..).unwrap_or(..)` on a Result, the bool count on `Protocol`, and the JSON number casts. The casts are allowed rather than rewritten. Token counts and tick values are whole and far below 2^53, and the one narrowing cast is already guarded non-negative. --- src/grok_acp.rs | 33 ++++++++++++++++++++++++++------- 1 file changed, 26 insertions(+), 7 deletions(-) diff --git a/src/grok_acp.rs b/src/grok_acp.rs index 862c889..c8fd20b 100644 --- a/src/grok_acp.rs +++ b/src/grok_acp.rs @@ -78,6 +78,10 @@ struct PendingApproval { /// State that spans the JSON-RPC records of one turn. #[derive(Debug)] +#[allow( + clippy::struct_excessive_bools, + reason = "each flag is one independent latch in the ACP turn, not a state enum" +)] pub(crate) struct Protocol { request: Request, pub terminal: Terminal, @@ -865,8 +869,7 @@ fn command_names(update: &Value) -> Vec { fn unix_secs() -> u64 { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0) + .map_or(0, |d| d.as_secs()) } fn info_wire(session_id: &str, id: u64, fallback: bool) -> String { @@ -927,8 +930,14 @@ fn usage_wire(session_id: &str, fallback: bool) -> String { fn json_f64(value: &Value) -> Option { value .as_f64() - .or_else(|| value.as_u64().map(|n| n as f64)) - .or_else(|| value.as_i64().map(|n| n as f64)) + .or_else(|| { + #[allow(clippy::cast_precision_loss, reason = "token counts never reach 2^53")] + value.as_u64().map(|n| n as f64) + }) + .or_else(|| { + #[allow(clippy::cast_precision_loss, reason = "token counts never reach 2^53")] + value.as_i64().map(|n| n as f64) + }) } fn first_f64(root: &Value, keys: &[&str]) -> Option { @@ -990,7 +999,7 @@ fn parse_iso_utc(s: &str) -> Option { let yoe = y - era * 400; let doy = (153 * m + 2) / 5 + d - 1; let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy; - let days = era * 146097 + doe - 719468; + let days = era * 146_097 + doe - 719_468; Some(days * 86400 + hh * 3600 + mm * 60 + ss) } @@ -1027,6 +1036,11 @@ fn first_u64(root: &Value, keys: &[&str]) -> Option { } if let Some(n) = root.get(*key).and_then(Value::as_f64) { if n >= 0.0 { + #[allow( + clippy::cast_possible_truncation, + clippy::cast_sign_loss, + reason = "guarded non-negative; token counts are whole and bounded" + )] return Some(n as u64); } } @@ -1118,7 +1132,12 @@ fn grok_usage(usage: &Value) -> Usage { cost_usd: usage .get("costUsdTicks") .or_else(|| usage.get("cost_usd_ticks")) - .and_then(|value| value.as_f64().or_else(|| value.as_u64().map(|n| n as f64))) + .and_then(|value| { + value.as_f64().or_else(|| { + #[allow(clippy::cast_precision_loss, reason = "token counts never reach 2^53")] + value.as_u64().map(|n| n as f64) + }) + }) // CLI `costUsdTicks` is 1e-9 USD (session 01a09ca7: 1.644e9 ticks). // xAI API `cost_in_usd_ticks` is 1e-10; do not mix the two. .map(|ticks| ticks / 1_000_000_000.0) @@ -1313,7 +1332,7 @@ mod tests { "inputTokens": 18033, "outputTokens": 189, "cachedReadTokens": 0, - "costUsdTicks": 126480000 + "costUsdTicks": 126_480_000 } } }