diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md
index e2ca8ae54fe..877b58f2b21 100644
--- a/docs/PROVIDERS.md
+++ b/docs/PROVIDERS.md
@@ -887,9 +887,13 @@ reset-credit lookup. If ACP also fails, the original login error is shown, not
 an ACP error. It never runs for an isolated token account, a saved account row,
 or an OAuth principal, so it can never read a different account's usage:
 
-- The installed `grok agent stdio` is started with no console window and its
-  stderr discarded. Requests are newline-delimited JSON-RPC 2.0 on stdin and
-  stdout.
+- The installed `grok agent stdio` is found the same way as the login flow: the
+  native `~/.grok/bin/grok.exe` first, then `PATH`. It is started with no console
+  window and its stderr discarded. Requests are newline-delimited JSON-RPC 2.0 on
+  stdin and stdout.
+- While waiting for a reply, the client skips non-JSON lines, notifications, and
+  any message with a `method`, even when its `id` matches. Those are requests
+  from the server, not replies.
 - `initialize` is sent first with `protocolVersion` `"1"` and filesystem and
   terminal capabilities off, with a 4-second budget. Then `_x.ai/billing` is
   sent with empty params, with a 3-second budget. The method name is written
diff --git a/rust/examples/fake_grok_acp.rs b/rust/examples/fake_grok_acp.rs
new file mode 100644
index 00000000000..33be6ad4e9c
--- /dev/null
+++ b/rust/examples/fake_grok_acp.rs
@@ -0,0 +1,236 @@
+//! Scripted stand-in for `grok agent stdio`, used by the ACP billing tests.
+//!
+//! Usage: `fake_grok_acp agent stdio <scenario> [pid-file]`
+//!
+//! It speaks newline-delimited JSON-RPC 2.0 on stdin and stdout, as the real CLI
+//! does. When a pid file is given, it first writes `"<pid> <console>"`, where
+//! `<console>` is `1` when the process has a console window and `0` otherwise.
+//! The tests use that to check that CodexBar starts the child hidden.
+//!
+//! This is a cargo example, not a shipped binary. `cargo test` builds it next to
+//! the test binary, so no dependency or PATH setup is needed.
+
+use std::fs;
+use std::io::{self, BufRead, Write};
+use std::thread;
+use std::time::Duration;
+
+use serde_json::{Value, json};
+
+const BILLING_METHOD: &str = "_x.ai/billing";
+const METHOD_NOT_FOUND: i64 = -32601;
+const INVALID_PARAMS: i64 = -32602;
+/// Larger than the client's 256 KiB line limit.
+const OVERSIZED_LINE_BYTES: usize = 300 * 1024;
+
+#[cfg(windows)]
+unsafe extern "system" {
+    fn GetConsoleWindow() -> *mut std::ffi::c_void;
+}
+
+fn has_console() -> bool {
+    #[cfg(windows)]
+    {
+        // SAFETY: GetConsoleWindow takes no arguments and only reads state.
+        !unsafe { GetConsoleWindow() }.is_null()
+    }
+    #[cfg(not(windows))]
+    {
+        false
+    }
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() < 4 || args[1] != "agent" || args[2] != "stdio" {
+        eprintln!("usage: fake_grok_acp agent stdio <scenario> [pid-file]");
+        std::process::exit(2);
+    }
+    let scenario = args[3].as_str();
+    if let Some(path) = args.get(4) {
+        let record = format!("{} {}", std::process::id(), u8::from(has_console()));
+        if fs::write(path, record).is_err() {
+            std::process::exit(3);
+        }
+    }
+    serve(scenario, io::stdin().lock().lines());
+}
+
+/// Writes one JSON-RPC message per line. A failed write means the client has
+/// gone away, so the server exits.
+fn send(value: &Value) {
+    send_raw(&format!("{value}\n"));
+}
+
+fn send_raw(text: &str) {
+    let mut out = io::stdout().lock();
+    if out
+        .write_all(text.as_bytes())
+        .and_then(|()| out.flush())
+        .is_err()
+    {
+        std::process::exit(0);
+    }
+}
+
+/// Blocks until the client kills this process.
+fn hang() -> ! {
+    loop {
+        thread::sleep(Duration::from_secs(60));
+    }
+}
+
+fn next_request(lines: &mut impl Iterator<Item = io::Result<String>>) -> Option<Value> {
+    loop {
+        let line = lines.next()?.ok()?;
+        if line.trim().is_empty() {
+            continue;
+        }
+        return serde_json::from_str(&line).ok();
+    }
+}
+
+fn error_reply(id: &Value, code: i64, message: &str) -> Value {
+    json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
+}
+
+fn serve(scenario: &str, mut lines: impl Iterator<Item = io::Result<String>>) {
+    let Some(init) = next_request(&mut lines) else {
+        return;
+    };
+    match scenario {
+        // The CLI reads the handshake and then exits without answering.
+        "exit_early" => return,
+        "hang_init" => hang(),
+        _ => {}
+    }
+    let init_id = init.get("id").cloned().unwrap_or(Value::Null);
+    if init.get("method").and_then(Value::as_str) != Some("initialize")
+        || init
+            .pointer("/params/protocolVersion")
+            .and_then(Value::as_str)
+            != Some("1")
+    {
+        send(&error_reply(&init_id, INVALID_PARAMS, "bad initialize"));
+        return;
+    }
+    if scenario == "auth_init" {
+        send(&error_reply(&init_id, -32000, "Authentication required"));
+        return;
+    }
+    noise(scenario, &init_id);
+    send(&json!({
+        "jsonrpc": "2.0",
+        "id": init_id,
+        "result": {
+            "protocolVersion": "1",
+            "agentCapabilities": { "loadSession": false },
+        },
+    }));
+
+    let Some(billing) = next_request(&mut lines) else {
+        return;
+    };
+    let billing_id = billing.get("id").cloned().unwrap_or(Value::Null);
+    match scenario {
+        "hang" => hang(),
+        "partial_eof" => {
+            // Half a frame, then the process ends.
+            send_raw("{\"jsonrpc\":\"2.0\",\"id\":2,\"res");
+            return;
+        }
+        "oversized" => {
+            send_raw(&"a".repeat(OVERSIZED_LINE_BYTES));
+            send_raw("\n");
+            hang();
+        }
+        "method_not_found" => {
+            send(&error_reply(
+                &billing_id,
+                METHOD_NOT_FOUND,
+                "Method not found",
+            ));
+            return;
+        }
+        "auth_billing" => {
+            send(&error_reply(
+                &billing_id,
+                -32000,
+                "Authentication required. Run `grok login`.",
+            ));
+            return;
+        }
+        _ => {}
+    }
+    if billing.get("method").and_then(Value::as_str) != Some(BILLING_METHOD) {
+        send(&error_reply(
+            &billing_id,
+            METHOD_NOT_FOUND,
+            "unexpected method",
+        ));
+        return;
+    }
+    noise(scenario, &billing_id);
+    let result = if scenario == "legacy" {
+        legacy_billing_result()
+    } else {
+        credits_billing_result()
+    };
+    send(&json!({ "jsonrpc": "2.0", "id": billing_id, "result": result }));
+}
+
+/// Traffic that is not the reply to the pending request: a truncated frame,
+/// non-JSON text, a notification, a reply to another id, and a server request
+/// that reuses the pending id. Only the first two are sent by `garbled`.
+fn noise(scenario: &str, pending_id: &Value) {
+    match scenario {
+        "garbled" => {
+            send_raw("{\"jsonrpc\":\"2.0\",\"id\":\n");
+            send_raw("not json at all\n");
+        }
+        "noise" => {
+            send_raw("not json at all\n");
+            send(&json!({
+                "jsonrpc": "2.0",
+                "method": "session/update",
+                "params": { "sessionId": "s1", "update": { "sessionUpdate": "agent_message_chunk" } },
+            }));
+            send(&json!({ "jsonrpc": "2.0", "id": 999, "result": {} }));
+            send(&json!({
+                "jsonrpc": "2.0",
+                "id": pending_id,
+                "method": "fs/read_text_file",
+                "params": { "path": "/nonexistent" },
+            }));
+        }
+        _ => {}
+    }
+}
+
+/// Current `_x.ai/billing` result: the nested `config` credits payload with the
+/// plan at the top level, as upstream reads it.
+fn credits_billing_result() -> Value {
+    json!({
+        "config": {
+            "creditUsagePercent": 42,
+            "currentPeriod": {
+                "start": "2026-01-01T00:00:00Z",
+                "end": "2099-02-01T00:00:00Z",
+            },
+        },
+        "subscription_tier": "SUPERGROK_HEAVY",
+    })
+}
+
+/// Older top-level result: monetary values are `{ "val": <cents> }`.
+fn legacy_billing_result() -> Value {
+    json!({
+        "billingCycle": {
+            "billingPeriodStart": "2026-01-01T00:00:00Z",
+            "billingPeriodEnd": "2099-02-01T00:00:00Z",
+        },
+        "monthlyLimit": { "val": 10000 },
+        "usage": { "totalUsed": { "val": 2500 } },
+        "subscription_tier": "SUPERGROK",
+    })
+}
diff --git a/rust/src/providers/grok/accounts.rs b/rust/src/providers/grok/accounts.rs
index 1523064ffc3..ccdebeed52a 100644
--- a/rust/src/providers/grok/accounts.rs
+++ b/rust/src/providers/grok/accounts.rs
@@ -1,7 +1,7 @@
 //! Saved Grok CLI logins. Only auth.json identity and tokens move; sessions,
 //! skills, and the rest of ~/.grok stay in the ambient home.
 
-mod login;
+pub(super) mod login;
 mod orca;
 mod orca_runtime;
 mod refresh;
diff --git a/rust/src/providers/grok/acp.rs b/rust/src/providers/grok/acp.rs
index 56202e6d750..fe85a4f0815 100644
--- a/rust/src/providers/grok/acp.rs
+++ b/rust/src/providers/grok/acp.rs
@@ -10,6 +10,7 @@
 //! rows never reach it.
 
 use std::io;
+use std::path::Path;
 use std::process::Stdio;
 use std::time::Duration;
 
@@ -37,6 +38,8 @@ const MAX_LINE_BYTES: usize = 256 * 1024;
 /// How long a killed child gets to be reaped before the caller moves on.
 const EXIT_GRACE: Duration = Duration::from_secs(1);
 const USAGE_UNAVAILABLE: &str = "Grok CLI billing returned no usage, reset or prepaid balance";
+/// Shown when the CLI cannot be started because it is not installed.
+pub(super) const CLI_NOT_FOUND: &str = "Grok CLI not found. Install it, then run `grok login`.";
 
 #[cfg(windows)]
 const CREATE_NO_WINDOW: u32 = 0x0800_0000;
@@ -63,7 +66,7 @@ impl Default for AcpTimeouts {
 /// The child has no console window, its stderr is discarded, and it is killed
 /// before this returns on every path: success, error or timeout.
 pub(super) async fn fetch_billing(
-    program: &str,
+    program: &Path,
     args: &[&str],
     timeouts: AcpTimeouts,
 ) -> Result<BearerBilling, ProviderError> {
@@ -73,7 +76,7 @@ pub(super) async fn fetch_billing(
     outcome
 }
 
-fn spawn_child(program: &str, args: &[&str]) -> Result<Child, ProviderError> {
+fn spawn_child(program: &Path, args: &[&str]) -> Result<Child, ProviderError> {
     let mut command = Command::new(program);
     command
         .args(args)
@@ -86,9 +89,7 @@ fn spawn_child(program: &str, args: &[&str]) -> Result<Child, ProviderError> {
 
     command.spawn().map_err(|error| {
         if error.kind() == io::ErrorKind::NotFound {
-            ProviderError::NotInstalled(
-                "Grok CLI not found. Install it, then run `grok login`.".to_string(),
-            )
+            ProviderError::NotInstalled(CLI_NOT_FOUND.to_string())
         } else {
             ProviderError::Other("Grok CLI failed to start".to_string())
         }
@@ -246,9 +247,13 @@ async fn read_bounded_line<R: AsyncBufRead + Unpin>(reader: &mut R) -> io::Resul
     }
 }
 
-/// The reply to request `id`, or `None` for notifications and replies to other
-/// requests.
+/// The reply to request `id`, or `None` for notifications, requests from the
+/// server, and replies to other requests. A message with a `method` is never a
+/// reply, even when its `id` happens to match ours.
 fn reply_for(message: &Value, id: u64) -> Option<Result<Value, ProviderError>> {
+    if message.get("method").is_some() {
+        return None;
+    }
     let message_id = message.get("id")?.as_u64()?;
     if message_id != id {
         return None;
@@ -360,3 +365,7 @@ fn legacy_billing(result: &Value, now: DateTime<Utc>) -> Option<GrokBillingSnaps
 #[cfg(test)]
 #[path = "acp_tests.rs"]
 mod tests;
+
+#[cfg(test)]
+#[path = "acp_process_tests.rs"]
+mod process_tests;
diff --git a/rust/src/providers/grok/acp_process_tests.rs b/rust/src/providers/grok/acp_process_tests.rs
new file mode 100644
index 00000000000..866fe6ebd51
--- /dev/null
+++ b/rust/src/providers/grok/acp_process_tests.rs
@@ -0,0 +1,459 @@
+//! End-to-end ACP tests against a real child process. `examples/fake_grok_acp`
+//! stands in for `grok agent stdio`, and every call goes through the production
+//! `spawn_child` path: hidden console, piped stdio, stderr discarded.
+//!
+//! `cargo test` builds the example next to the test binary. The fake writes
+//! `"<pid> <console>"` to a pid file before its handshake, so each test can see
+//! whether the child was hidden and whether it is gone afterwards.
+
+use std::path::{Path, PathBuf};
+use std::process::{Command as StdCommand, Stdio};
+use std::time::{Duration, Instant};
+
+use super::{AcpTimeouts, fetch_billing};
+use crate::core::{FetchContext, ProviderError};
+use crate::providers::grok::accounts::GrokAuthKind;
+use crate::providers::grok::credits_proxy::BearerBilling;
+use crate::providers::grok::{GrokCredentials, GrokProvider};
+
+/// Budgets for tests that expect a real answer. Process start-up is slow on a
+/// loaded machine, so these are far above the production budgets.
+fn generous() -> AcpTimeouts {
+    AcpTimeouts {
+        initialize: Duration::from_secs(20),
+        billing: Duration::from_secs(20),
+    }
+}
+
+/// Budgets for tests that expect a timeout.
+fn short() -> AcpTimeouts {
+    AcpTimeouts {
+        initialize: Duration::from_millis(300),
+        billing: Duration::from_millis(300),
+    }
+}
+
+fn fake_grok() -> PathBuf {
+    let exe = std::env::current_exe().expect("test binary path");
+    // target/<profile>/deps/<test>.exe -> target/<profile>/examples/fake_grok_acp.exe
+    let profile_dir = exe
+        .parent()
+        .and_then(Path::parent)
+        .expect("cargo target layout");
+    let program = profile_dir
+        .join("examples")
+        .join(format!("fake_grok_acp{}", std::env::consts::EXE_SUFFIX));
+    assert!(
+        program.is_file(),
+        "{} is missing; cargo test builds the fake_grok_acp example",
+        program.display()
+    );
+    program
+}
+
+/// What the fake child recorded about itself before the handshake.
+struct ChildRecord {
+    pid: u32,
+    console: bool,
+}
+
+/// One fake child's scratch directory and pid file.
+struct FakeRun {
+    _dir: tempfile::TempDir,
+    pid_file: PathBuf,
+}
+
+impl FakeRun {
+    fn new() -> Self {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let pid_file = dir.path().join("child.txt");
+        Self {
+            _dir: dir,
+            pid_file,
+        }
+    }
+
+    /// Arguments for `agent stdio <scenario> <pid-file>`.
+    fn args(&self, scenario: &str) -> Vec<String> {
+        vec![
+            "agent".to_string(),
+            "stdio".to_string(),
+            scenario.to_string(),
+            self.pid_file.to_string_lossy().into_owned(),
+        ]
+    }
+
+    /// `None` when the child never started far enough to write its record.
+    fn record(&self) -> Option<ChildRecord> {
+        let text = std::fs::read_to_string(&self.pid_file).ok()?;
+        let mut parts = text.split_whitespace();
+        Some(ChildRecord {
+            pid: parts.next()?.parse().ok()?,
+            console: parts.next()? == "1",
+        })
+    }
+}
+
+async fn run_scenario(
+    scenario: &str,
+    timeouts: AcpTimeouts,
+) -> (Result<BearerBilling, ProviderError>, FakeRun) {
+    let run = FakeRun::new();
+    let args = run.args(scenario);
+    let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+    let outcome = fetch_billing(&fake_grok(), &refs, timeouts).await;
+    (outcome, run)
+}
+
+fn expect_error<T>(outcome: Result<T, ProviderError>) -> ProviderError {
+    match outcome {
+        Ok(_) => panic!("expected the ACP call to fail"),
+        Err(error) => error,
+    }
+}
+
+/// The child was killed and reaped: its process object is gone or has exited.
+async fn assert_child_stopped(record: &ChildRecord) {
+    let deadline = Instant::now() + Duration::from_secs(5);
+    while process_is_running(record.pid) && Instant::now() < deadline {
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        !process_is_running(record.pid),
+        "child {} is still running after the call returned",
+        record.pid
+    );
+}
+
+#[cfg(windows)]
+fn process_is_running(pid: u32) -> bool {
+    use windows::Win32::Foundation::{CloseHandle, STILL_ACTIVE};
+    use windows::Win32::System::Threading::{
+        GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
+    };
+
+    // SAFETY: the handle is opened, queried and closed here, and never escapes.
+    unsafe {
+        let Ok(handle) = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, false, pid) else {
+            return false;
+        };
+        let mut code = 0u32;
+        let queried = GetExitCodeProcess(handle, &mut code).is_ok();
+        CloseHandle(handle).ok();
+        queried && code == STILL_ACTIVE.0 as u32
+    }
+}
+
+/// Linux only: a reaped child has no `/proc` entry.
+#[cfg(not(windows))]
+fn process_is_running(pid: u32) -> bool {
+    Path::new(&format!("/proc/{pid}")).exists()
+}
+
+fn missing_login_error() -> ProviderError {
+    ProviderError::NotInstalled("Grok auth.json not found. Run `grok login`.".to_string())
+}
+
+/// Guards the no-zombie assertions: the probe must see a live child as running
+/// and a killed, reaped one as stopped. Otherwise `assert_child_stopped` would
+/// pass for the wrong reason.
+#[test]
+fn the_liveness_probe_tells_a_running_child_from_a_stopped_one() {
+    let run = FakeRun::new();
+    // Stdin stays open, so the child blocks on its first read and keeps running.
+    let mut command = StdCommand::new(fake_grok());
+    command
+        .args(run.args("hang_init"))
+        .stdin(Stdio::piped())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null());
+    // A console-less test runner would otherwise give the child a visible console.
+    #[cfg(windows)]
+    {
+        use std::os::windows::process::CommandExt;
+        command.creation_flags(0x0800_0000); // CREATE_NO_WINDOW
+    }
+    let mut child = command.spawn().expect("probe child starts");
+    let deadline = Instant::now() + Duration::from_secs(10);
+    let record = loop {
+        if let Some(record) = run.record() {
+            break record;
+        }
+        assert!(Instant::now() < deadline, "probe child never started");
+        std::thread::sleep(Duration::from_millis(10));
+    };
+
+    assert!(process_is_running(record.pid), "a live child must be seen");
+    child.kill().expect("probe child is killed");
+    child.wait().expect("probe child is reaped");
+    assert!(
+        !process_is_running(record.pid),
+        "a reaped child must not be seen as running"
+    );
+}
+
+// ---------------------------------------------------------------- success
+
+#[tokio::test]
+async fn credits_answer_comes_back_from_a_real_child_which_is_hidden_and_stopped() {
+    let (outcome, run) = run_scenario("ok", generous()).await;
+
+    let bearer = outcome.expect("billing answer");
+    assert_eq!(bearer.billing.used_percent, Some(42.0));
+    assert_eq!(bearer.subscription_tier.as_deref(), Some("SuperGrok Heavy"));
+    assert!(bearer.billing.resets_at.is_some());
+
+    let record = run.record().expect("child recorded itself");
+    assert!(
+        !record.console,
+        "the production spawn must hide the console window"
+    );
+    assert_child_stopped(&record).await;
+}
+
+/// Control for the hidden-spawn check: a child started without
+/// `CREATE_NO_WINDOW` inherits this test's console. If this test has no console,
+/// the control proves nothing and is skipped.
+#[cfg(windows)]
+#[test]
+fn the_console_probe_reports_a_console_when_one_is_inherited() {
+    use windows::Win32::System::Console::GetConsoleWindow;
+
+    // SAFETY: GetConsoleWindow takes no arguments and only reads state.
+    let parent_has_console = !unsafe { GetConsoleWindow() }.0.is_null();
+    if !parent_has_console {
+        eprintln!("skipped: the test process has no console to inherit");
+        return;
+    }
+    let run = FakeRun::new();
+    let status = std::process::Command::new(fake_grok())
+        .args(run.args("exit_early"))
+        .stdin(Stdio::null())
+        .stdout(Stdio::null())
+        .stderr(Stdio::null())
+        .status()
+        .expect("control child runs");
+    assert!(status.success());
+    let record = run.record().expect("control child recorded itself");
+    assert!(
+        record.console,
+        "the probe must see an inherited console, or the hidden check proves nothing"
+    );
+}
+
+#[tokio::test]
+async fn noise_and_a_colliding_server_request_are_skipped_before_the_reply() {
+    // Non-JSON text, a notification, a reply to another id, and a server
+    // request that reuses the pending id arrive before each reply.
+    let (outcome, run) = run_scenario("noise", generous()).await;
+
+    assert_eq!(
+        outcome.expect("answer after noise").billing.used_percent,
+        Some(42.0)
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn truncated_and_garbled_frames_are_skipped_until_the_reply() {
+    let (outcome, run) = run_scenario("garbled", generous()).await;
+
+    assert_eq!(
+        outcome
+            .expect("answer after garbled frames")
+            .billing
+            .used_percent,
+        Some(42.0)
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn the_older_top_level_billing_shape_is_read_from_a_real_child() {
+    let (outcome, run) = run_scenario("legacy", generous()).await;
+
+    let bearer = outcome.expect("legacy answer");
+    // 2500 of 10000 cents.
+    assert_eq!(bearer.billing.used_percent, Some(25.0));
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+// ---------------------------------------------------------------- errors
+
+#[tokio::test]
+async fn method_not_found_means_this_cli_has_no_acp_billing() {
+    let (outcome, run) = run_scenario("method_not_found", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(
+        matches!(&error, ProviderError::Other(message) if message.contains("does not provide billing")),
+        "{error}"
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn an_authentication_error_on_billing_is_auth_required() {
+    let (outcome, run) = run_scenario("auth_billing", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(matches!(error, ProviderError::AuthRequired), "{error}");
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn an_authentication_error_on_initialize_is_auth_required() {
+    let (outcome, run) = run_scenario("auth_init", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(matches!(error, ProviderError::AuthRequired), "{error}");
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn a_child_that_never_answers_initialize_times_out_and_is_killed() {
+    let run = FakeRun::new();
+    let args = run.args("hang_init");
+    let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+    let started = Instant::now();
+
+    let outcome = fetch_billing(&fake_grok(), &refs, short()).await;
+
+    assert!(matches!(expect_error(outcome), ProviderError::Timeout));
+    assert!(started.elapsed() < Duration::from_secs(10));
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn a_child_that_never_answers_billing_times_out_and_is_killed() {
+    // The handshake completes, then billing hangs. The short budget is only
+    // for billing, so initialize has room to finish.
+    let run = FakeRun::new();
+    let args = run.args("hang");
+    let refs: Vec<&str> = args.iter().map(String::as_str).collect();
+    let timeouts = AcpTimeouts {
+        initialize: Duration::from_secs(20),
+        billing: Duration::from_millis(300),
+    };
+
+    let outcome = fetch_billing(&fake_grok(), &refs, timeouts).await;
+
+    assert!(matches!(expect_error(outcome), ProviderError::Timeout));
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn a_child_that_exits_before_replying_is_an_error_not_a_hang() {
+    let (outcome, run) = run_scenario("exit_early", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(
+        matches!(&error, ProviderError::Other(message) if message.contains("closed")),
+        "{error}"
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn a_half_written_frame_followed_by_exit_is_an_error() {
+    let (outcome, run) = run_scenario("partial_eof", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(
+        matches!(&error, ProviderError::Other(message) if message.contains("closed")),
+        "{error}"
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn an_oversized_line_is_rejected_and_the_child_is_killed() {
+    let (outcome, run) = run_scenario("oversized", generous()).await;
+
+    let error = expect_error(outcome);
+    assert!(
+        matches!(&error, ProviderError::Other(message) if message.contains("oversized")),
+        "{error}"
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+// ---------------------------------------------------------------- fallback
+
+#[tokio::test]
+async fn a_missing_cli_login_is_answered_by_the_real_child() {
+    let run = FakeRun::new();
+    let provider =
+        GrokProvider::new().with_acp_process_for_tests(fake_grok(), run.args("ok"), generous());
+
+    let result = provider
+        .fetch_cli_acp_without_login(missing_login_error(), &FetchContext::default())
+        .await
+        .expect("ACP answers without a login");
+
+    assert_eq!(result.usage.primary.used_percent, 42.0);
+    assert_eq!(result.source_label, "grok-cli");
+    assert_eq!(result.usage.account_email, None);
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn an_expired_cli_login_is_answered_by_the_real_child() {
+    // The error comes from the real parser, not a constructed value.
+    let text = r#"{"https://accounts.x.ai/sign-in": {"key": "cli-token", "auth_mode": "session", "expires_at": "2000-01-01T00:00:00Z"}}"#;
+    let expired = GrokCredentials::parse_for_kind(text, GrokAuthKind::Cli).unwrap_err();
+    let run = FakeRun::new();
+    let provider =
+        GrokProvider::new().with_acp_process_for_tests(fake_grok(), run.args("ok"), generous());
+
+    let result = provider
+        .fetch_cli_acp_without_login(expired, &FetchContext::default())
+        .await
+        .expect("ACP answers for an expired login");
+
+    assert_eq!(result.usage.primary.used_percent, 42.0);
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn a_failing_child_keeps_the_original_login_error() {
+    let run = FakeRun::new();
+    let provider = GrokProvider::new().with_acp_process_for_tests(
+        fake_grok(),
+        run.args("method_not_found"),
+        generous(),
+    );
+
+    let error = provider
+        .fetch_cli_acp_without_login(missing_login_error(), &FetchContext::default())
+        .await
+        .unwrap_err();
+
+    assert!(
+        matches!(&error, ProviderError::NotInstalled(message) if message.contains("grok login")),
+        "{error}"
+    );
+    assert_child_stopped(&run.record().expect("child recorded itself")).await;
+}
+
+#[tokio::test]
+async fn an_isolated_account_never_starts_the_child() {
+    let run = FakeRun::new();
+    let provider =
+        GrokProvider::new().with_acp_process_for_tests(fake_grok(), run.args("ok"), generous());
+    let isolated = FetchContext {
+        token_account_isolated: true,
+        ..FetchContext::default()
+    };
+
+    let error = provider
+        .fetch_cli_acp_without_login(missing_login_error(), &isolated)
+        .await
+        .unwrap_err();
+
+    assert!(matches!(error, ProviderError::NotInstalled(_)), "{error}");
+    assert!(
+        run.record().is_none(),
+        "an isolated account must not start the ambient CLI"
+    );
+}
diff --git a/rust/src/providers/grok/acp_tests.rs b/rust/src/providers/grok/acp_tests.rs
index 9846e28b585..3e5b7aa44ab 100644
--- a/rust/src/providers/grok/acp_tests.rs
+++ b/rust/src/providers/grok/acp_tests.rs
@@ -489,7 +489,7 @@ const SILENT_CHILD: (&str, &[&str]) = ("sleep", &["120"]);
 #[tokio::test]
 async fn timed_out_child_is_stopped_and_reaped() {
     let (program, args) = SILENT_CHILD;
-    let mut child = spawn_child(program, args).expect("silent child starts");
+    let mut child = spawn_child(std::path::Path::new(program), args).expect("silent child starts");
 
     let outcome = run_child(&mut child, short_timeouts(), now()).await;
     // `fetch_billing` runs exactly this sequence.
@@ -511,7 +511,7 @@ async fn timed_out_child_is_stopped_and_reaped() {
 #[tokio::test]
 async fn a_missing_grok_binary_is_reported_as_not_installed() {
     let error = fetch_billing(
-        "codexbar-test-no-such-grok-binary",
+        std::path::Path::new("codexbar-test-no-such-grok-binary"),
         &["agent", "stdio"],
         short_timeouts(),
     )
diff --git a/rust/src/providers/grok/mod.rs b/rust/src/providers/grok/mod.rs
index 26f8e7afa62..bfe2ac36a03 100644
--- a/rust/src/providers/grok/mod.rs
+++ b/rust/src/providers/grok/mod.rs
@@ -37,7 +37,6 @@ const REMAINING_RESETS_ENDPOINT: &str =
 const CLI_SETTINGS_ENDPOINT: &str = "https://cli-chat-proxy.grok.com/v1/settings";
 const RESET_CREDITS_TIMEOUT: Duration = Duration::from_secs(2);
 const RESET_CREDITS_JOIN_GRACE: Duration = Duration::from_millis(250);
-const GROK_CLI_PROGRAM: &str = "grok";
 const GROK_CLI_ACP_ARGS: [&str; 2] = ["agent", "stdio"];
 
 /// Reset-window pace (upstream `GrokProviderDescriptor.pace`, v0.70.0): weekly
@@ -55,11 +54,18 @@ pub struct GrokProvider {
 }
 
 /// Where the ACP billing fallback gets its answer. `Installed` runs the Grok
-/// CLI; tests substitute a fake to pin the fallback ordering.
+/// CLI; tests substitute a scripted child process or a fake to pin the
+/// fallback ordering.
 #[derive(Clone)]
 enum CliBillingRunner {
     Installed,
     #[cfg(test)]
+    Process {
+        program: PathBuf,
+        args: Vec<String>,
+        timeouts: acp::AcpTimeouts,
+    },
+    #[cfg(test)]
     Fake(std::sync::Arc<dyn Fn() -> Result<BearerBilling, ProviderError> + Send + Sync>),
 }
 
@@ -67,12 +73,21 @@ impl CliBillingRunner {
     async fn fetch(&self) -> Result<BearerBilling, ProviderError> {
         match self {
             Self::Installed => {
-                acp::fetch_billing(
-                    GROK_CLI_PROGRAM,
-                    &GROK_CLI_ACP_ARGS,
-                    acp::AcpTimeouts::default(),
-                )
-                .await
+                // The native installer puts grok under ~/.grok/bin, which is
+                // often not on PATH. The login flow's resolver checks that
+                // location first, so the ACP fallback uses the same answer.
+                let program = accounts::login::executable()
+                    .map_err(|_| ProviderError::NotInstalled(acp::CLI_NOT_FOUND.to_string()))?;
+                acp::fetch_billing(&program, &GROK_CLI_ACP_ARGS, acp::AcpTimeouts::default()).await
+            }
+            #[cfg(test)]
+            Self::Process {
+                program,
+                args,
+                timeouts,
+            } => {
+                let args: Vec<&str> = args.iter().map(String::as_str).collect();
+                acp::fetch_billing(program, &args, *timeouts).await
             }
             #[cfg(test)]
             Self::Fake(run) => run(),
@@ -119,6 +134,21 @@ impl GrokProvider {
         self
     }
 
+    #[cfg(test)]
+    fn with_acp_process_for_tests(
+        mut self,
+        program: PathBuf,
+        args: Vec<String>,
+        timeouts: acp::AcpTimeouts,
+    ) -> Self {
+        self.cli_billing = CliBillingRunner::Process {
+            program,
+            args,
+            timeouts,
+        };
+        self
+    }
+
     #[cfg(test)]
     fn with_cli_billing_for_tests(
         mut self,
@@ -629,7 +659,7 @@ impl GrokProvider {
     }
 
     fn detect_cli_version() -> Option<String> {
-        let mut command = std::process::Command::new("grok");
+        let mut command = std::process::Command::new(accounts::login::executable().ok()?);
         command.arg("--version");
         hide_windows_console(&mut command);
         let output = command.output().ok()?;