From f03bf44fbd5a5e03c970ba710ff693cefdbc1aaa Mon Sep 17 00:00:00 2001
From: signet-bot-claude <261785442+signet-bot-claude@users.noreply.github.com>
Date: Fri, 16 Oct 2026 16:39:12 -0400
Subject: [PATCH 1/4] Grok: requeue a batch lost when the leader restarts
 behind the stdio bridge

Grok's leader restarts itself to apply an auto-update. hcom's stdio bridge
reconnects to the new leader without a word, so the batch hcom queued in the
gap never reached Grok and no response or queue event ever mentioned it. It
stayed in flight forever: nothing later was queued, roster polls stopped too
(one was lost the same way), and unread climbed until someone typed into the
agent's terminal.

A new leader PID in the socket's lock file, or a roster or session/load
request unanswered for 30 s, now counts as a lost connection. The existing
reconnect path detaches the in-flight batch, follows it through the new
connection's queue, and queues it again if it is gone. stop_leader reads the
lock through the same helper.

Tests: leader PID parsing from the lock beside the socket; a changed PID (and
only a changed, readable one) is a lost connection; a roster or load request
unanswered for REQUEST_TIMEOUT is a lost connection; and an end-to-end run
against a fake bridge whose leader restarts as the batch goes out, which
must reconnect once and deliver the batch.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/delivery/grok.rs | 296 ++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 278 insertions(+), 18 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 277be5a9..185e07b1 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -26,6 +26,12 @@
 //! is part of the model's turn, or when the prompt request returns a result.
 //! A batch that never started (removed from the queue, transport lost) stays
 //! unread and is queued again later: at-least-once, never silently dropped.
+//!
+//! The bridge reconnects on its own when the leader restarts (Grok's leader
+//! restarts itself to apply an auto-update) and tells hcom nothing. Requests
+//! written while the leader was down are lost and never answered. hcom
+//! treats a new leader PID in the socket's lock file, or a roster or load
+//! request left unanswered, as a lost connection.
 
 use std::collections::{HashMap, HashSet};
 use std::io::{BufRead, BufReader, Write};
@@ -58,9 +64,14 @@ const ROSTER_FAST_WINDOW: Duration = Duration::from_secs(5);
 const ROSTER_SLOW_POLL: Duration = Duration::from_secs(15);
 /// Wait before re-queueing a batch whose prompt was dropped before it ran
 /// (user removed it from the queue, prompt error, transport lost).
-const REQUEUE_DELAY: Duration = Duration::from_secs(10);
+const REQUEUE_DELAY: Duration = Duration::from_secs(if cfg!(test) { 1 } else { 10 });
 /// Wait before loading the bound session again after a failed load.
 const LOAD_RETRY_DELAY: Duration = Duration::from_secs(2);
+/// Grok answers a roster or `session/load` request in well under a second. One
+/// still unanswered after this long was lost in transit.
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+/// How often to check which leader process holds the socket.
+const LEADER_CHECK: Duration = Duration::from_secs(2);
 /// Prompts hcom queues carry this `promptId` prefix.
 const HCOM_PROMPT_ID_PREFIX: &str = "hcom-";
 /// Leader-mode Grok ignores these (it warns and continues). Since hcom must
@@ -149,11 +160,11 @@ impl Launch {
     /// PID to the socket's sibling `.lock` (`x.sock` -> `x.lock`).
     pub(crate) fn stop_leader(&self) {
         let socket = std::path::Path::new(&self.socket);
-        let lock = socket.with_extension("lock");
-        let pid = std::fs::read_to_string(&lock)
-            .ok()
-            .and_then(|content| content.trim().parse::<u32>().ok());
-        if let Some(pid) = pid.filter(|pid| crate::sys::process::is_alive(*pid)) {
+        let lock = self.lock_path();
+        if let Some(pid) = self
+            .leader_pid()
+            .filter(|pid| crate::sys::process::is_alive(*pid))
+        {
             crate::sys::process::terminate(pid);
             let deadline = Instant::now() + Duration::from_secs(3);
             while crate::sys::process::is_alive(pid) && Instant::now() < deadline {
@@ -168,6 +179,19 @@ impl Launch {
         let _ = std::fs::remove_file(lock);
     }
 
+    fn lock_path(&self) -> std::path::PathBuf {
+        std::path::Path::new(&self.socket).with_extension("lock")
+    }
+
+    /// PID of the leader that holds this launch's socket, from its `.lock`.
+    fn leader_pid(&self) -> Option<u32> {
+        std::fs::read_to_string(self.lock_path())
+            .ok()?
+            .trim()
+            .parse()
+            .ok()
+    }
+
     /// Env for the TUI and the ACP client. The leader is spawned by whichever
     /// connects first and inherits it; `--no-subagents` itself has no effect
     /// in leader mode.
@@ -830,6 +854,38 @@ enum Pending {
     Load(String),
 }
 
+/// Requests awaiting a response on the current connection, and when each was
+/// sent.
+type Outstanding = HashMap<u64, (Pending, Instant)>;
+
+fn awaits(pending: &Outstanding, wanted: impl Fn(&Pending) -> bool) -> bool {
+    pending.values().any(|(request, _)| wanted(request))
+}
+
+/// A request unanswered for `REQUEST_TIMEOUT`, as its method name.
+fn stalled_request(pending: &Outstanding, now: Instant) -> Option<&'static str> {
+    pending
+        .values()
+        .find(|(_, sent)| now.saturating_duration_since(*sent) >= REQUEST_TIMEOUT)
+        .map(|(request, _)| match request {
+            Pending::Roster => "x.ai/sessions/list",
+            Pending::Load(_) => "session/load",
+        })
+}
+
+/// Whether another leader took over the socket since hcom attached. The
+/// bridge follows a restarted leader without a word, so this is the only sign
+/// that requests written in between were lost. An unreadable lock proves
+/// nothing.
+fn leader_change(attached: Option<u32>, current: Option<u32>) -> Option<String> {
+    match (attached, current) {
+        (Some(old), Some(new)) if old != new => {
+            Some(format!("Grok leader restarted (pid {old} -> {new})"))
+        }
+        _ => None,
+    }
+}
+
 /// The roster result, which the stdio bridge may wrap once more.
 fn roster_sessions(result: &Value) -> &[Value] {
     result
@@ -937,7 +993,7 @@ pub(super) fn run(
     let mut client: Option<Client> = None;
     let mut tracker = Tracker::default();
     let mut loaded: HashSet<String> = HashSet::new();
-    let mut pending: HashMap<u64, Pending> = HashMap::new();
+    let mut pending: Outstanding = HashMap::new();
     let mut roster_due: Option<Instant> = None;
     let mut roster_fast_until = Instant::now();
     let mut in_flight: Option<InFlight> = None;
@@ -947,6 +1003,8 @@ pub(super) fn run(
     let mut connect_failures: u32 = 0;
     let mut retry_at = Instant::now();
     let mut requeue_at = Instant::now();
+    let mut attached_leader: Option<u32> = None;
+    let mut leader_checked = Instant::now();
     let launched_at = Instant::now();
     while running.load(Ordering::Acquire) {
         super::refresh_title_state(super::TitleRefresh {
@@ -1005,6 +1063,8 @@ pub(super) fn run(
                 Ok(connected) => {
                     client = Some(connected);
                     connect_failures = 0;
+                    attached_leader = launch.leader_pid();
+                    leader_checked = Instant::now();
                     tracker.reconnected();
                     loaded.clear();
                     pending.clear();
@@ -1064,7 +1124,10 @@ pub(super) fn run(
                 }
                 Event::Response(value) => {
                     let id = value["id"].as_u64();
-                    match id.and_then(|id| pending.remove(&id)) {
+                    match id
+                        .and_then(|id| pending.remove(&id))
+                        .map(|(request, _)| request)
+                    {
                         Some(Pending::Roster) => match value.get("error") {
                             Some(error) => {
                                 log_warn("native", "grok.acp.roster_failed", &error.to_string())
@@ -1164,9 +1227,7 @@ pub(super) fn run(
                 }
                 Effect::Load { session, cwd } => {
                     if loaded.contains(&session)
-                        || pending
-                            .values()
-                            .any(|p| matches!(p, Pending::Load(s) if *s == session))
+                        || awaits(&pending, |p| matches!(p, Pending::Load(s) if *s == session))
                     {
                         continue;
                     }
@@ -1181,7 +1242,7 @@ pub(super) fn run(
                                "_meta": {"noReplay": true}}),
                     ) {
                         Ok(id) => {
-                            pending.insert(id, Pending::Load(session));
+                            pending.insert(id, (Pending::Load(session), Instant::now()));
                         }
                         Err(error) => closed = Some(format!("write failed: {error:#}")),
                     }
@@ -1194,7 +1255,7 @@ pub(super) fn run(
                 }
             }
         }
-        let roster_pending = pending.values().any(|p| matches!(p, Pending::Roster));
+        let roster_pending = awaits(&pending, |p| matches!(p, Pending::Roster));
         if !roster_pending && roster_due.is_none() {
             let fast = tracker.bound.is_none() || Instant::now() < roster_fast_until;
             let interval = if fast {
@@ -1208,11 +1269,28 @@ pub(super) fn run(
             roster_due = None;
             match conn.send("_x.ai/sessions/list", json!({})) {
                 Ok(id) => {
-                    pending.insert(id, Pending::Roster);
+                    pending.insert(id, (Pending::Roster, Instant::now()));
                 }
                 Err(error) => closed = Some(format!("write failed: {error:#}")),
             }
         }
+        // The bridge is still up, but what was written to it may never be
+        // answered: drop it like a lost connection, so the in-flight batch is
+        // followed through the next connection's queue and requeued if gone.
+        if closed.is_none() && leader_checked.elapsed() >= LEADER_CHECK {
+            leader_checked = Instant::now();
+            let current = launch.leader_pid();
+            closed = leader_change(attached_leader, current);
+            attached_leader = attached_leader.or(current);
+        }
+        if closed.is_none()
+            && let Some(method) = stalled_request(&pending, Instant::now())
+        {
+            closed = Some(format!(
+                "{method} unanswered after {}s",
+                REQUEST_TIMEOUT.as_secs()
+            ));
+        }
         if let Some(error) = closed {
             client = None;
             retry_at = Instant::now() + Duration::from_secs(1);
@@ -1293,9 +1371,7 @@ pub(super) fn run(
         // is retried.
         if let Some(session) = tracker.bound.clone()
             && !loaded.contains(&session)
-            && !pending
-                .values()
-                .any(|p| matches!(p, Pending::Load(s) if *s == session))
+            && !awaits(&pending, |p| matches!(p, Pending::Load(s) if *s == session))
             && Instant::now() >= load_retry_at
         {
             let cwd = tracker
@@ -1312,7 +1388,7 @@ pub(super) fn run(
                        "_meta": {"noReplay": true}}),
             ) {
                 Ok(id) => {
-                    pending.insert(id, Pending::Load(session));
+                    pending.insert(id, (Pending::Load(session), Instant::now()));
                 }
                 Err(error) => {
                     log_warn("native", "grok.acp.write_failed", &format!("{error:#}"));
@@ -1872,4 +1948,188 @@ mod tests {
         let plain = Launch::new("grok", &[], &["--", "--no-subagents"]).unwrap();
         assert!(plain.child_env().is_empty());
     }
+
+    fn launch_at(command: &str, socket: &std::path::Path) -> Launch {
+        Launch {
+            command: command.into(),
+            prefix: Vec::new(),
+            socket: socket.to_string_lossy().into_owned(),
+            no_subagents: false,
+        }
+    }
+
+    #[test]
+    fn leader_pid_comes_from_the_lock_beside_the_socket() {
+        let dir = tempfile::tempdir().unwrap();
+        let launch = launch_at("grok", &dir.path().join("hcom-grok-x.sock"));
+        assert_eq!(launch.leader_pid(), None);
+        let lock = dir.path().join("hcom-grok-x.lock");
+        std::fs::write(&lock, "77001\n").unwrap();
+        assert_eq!(launch.leader_pid(), Some(77001));
+        std::fs::write(&lock, "").unwrap();
+        assert_eq!(launch.leader_pid(), None);
+    }
+
+    #[test]
+    fn a_new_leader_pid_means_the_connection_was_lost() {
+        assert!(
+            leader_change(Some(77001), Some(94509))
+                .is_some_and(|why| why.contains("77001 -> 94509"))
+        );
+        assert_eq!(leader_change(Some(77001), Some(77001)), None);
+        // An unreadable lock (mid-restart, never written) proves nothing.
+        assert_eq!(leader_change(Some(77001), None), None);
+        assert_eq!(leader_change(None, Some(94509)), None);
+    }
+
+    #[test]
+    fn a_roster_or_load_left_unanswered_was_lost() {
+        let sent = Instant::now();
+        let mut pending: Outstanding = HashMap::new();
+        pending.insert(1, (Pending::Roster, sent));
+        assert_eq!(
+            stalled_request(&pending, sent + Duration::from_secs(1)),
+            None
+        );
+        assert_eq!(
+            stalled_request(&pending, sent + REQUEST_TIMEOUT),
+            Some("x.ai/sessions/list")
+        );
+        pending.clear();
+        pending.insert(2, (Pending::Load("s".into()), sent));
+        assert_eq!(
+            stalled_request(&pending, sent + REQUEST_TIMEOUT),
+            Some("session/load")
+        );
+    }
+
+    /// Stands in for `grok agent --leader --leader-socket <sock> stdio`. The
+    /// first bridge sees the leader restart as hcom's batch goes out: the
+    /// lock gets the new leader's PID, and nothing written to that bridge is
+    /// ever answered again. The bridge itself stays up, as Grok's does.
+    #[cfg(unix)]
+    const FAKE_BRIDGE: &str = r#"#!/bin/sh
+dir=$(dirname "$0")
+lock="${4%.sock}.lock"
+run=$(( $(cat "$dir/runs" 2>/dev/null || echo 0) + 1 ))
+echo "$run" > "$dir/runs"
+lost=
+while IFS= read -r line; do
+  [ -z "$lost" ] || continue
+  id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p')
+  case "$line" in
+    *'"method":"initialize"'*|*'"method":"session/load"'*)
+      echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{}}"
+      # The reconnected leader's queue snapshot: the lost batch is not in it.
+      case "$line" in *session/load*)
+        echo '{"jsonrpc":"2.0","method":"_x.ai/queue/changed","params":{"sessionId":"s1","entries":[]}}';;
+      esac ;;
+    *'"method":"_x.ai/sessions/list"'*)
+      echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{\"sessions\":[{\"sessionId\":\"s1\",\"resident\":true,\"activity\":\"idle\",\"lastChangeUnixMs\":1}]}}" ;;
+    *'"method":"session/prompt"'*)
+      if [ "$run" = 1 ]; then
+        echo 2002 > "$lock"
+        lost=1
+      else
+        prompt=$(printf '%s' "$line" | sed -n 's/.*"promptId":"\([^"]*\)".*/\1/p')
+        echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$prompt\",\"runningKind\":\"prompt\",\"entries\":[]}}"
+      fi ;;
+  esac
+done
+"#;
+
+    /// Observed: the leader auto-updated and restarted while idle; the
+    /// bridge reconnected without a word and the batch queued during the gap
+    /// never reached Grok. With no response or queue event for it, the batch
+    /// stayed in flight, no later mail was queued, and unread climbed until
+    /// someone typed into the agent's terminal.
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn batch_lost_in_a_silent_leader_restart_is_queued_again() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let (dir, hcom_dir, _home, _guard) = crate::hooks::test_helpers::isolated_test_env();
+        let db_path = hcom_dir.join("test.db");
+        let mut db = HcomDb::open_at(&db_path).unwrap();
+        let work = dir.path().join("work");
+        std::fs::create_dir_all(&work).unwrap();
+        db.save_instance_named(
+            "nova",
+            &serde_json::Map::from_iter([
+                ("tool".into(), json!("grok")),
+                ("session_id".into(), json!("s1")),
+                ("directory".into(), json!(work.to_string_lossy())),
+                ("status".into(), json!(ST_LISTENING)),
+                ("created_at".into(), json!(1.0)),
+            ]),
+        )
+        .unwrap();
+        db.set_session_binding("s1", "nova").unwrap();
+        db.set_process_binding("proc-1", "s1", "nova").unwrap();
+        let message = json!({"from": "luna", "text": "are you there?",
+                             "scope": "mentions", "mentions": ["nova"]});
+        db.conn()
+            .execute(
+                "INSERT INTO events (type, timestamp, instance, data)
+                 VALUES ('message', '2026-01-01T00:00:00Z', 'luna', ?1)",
+                rusqlite::params![message.to_string()],
+            )
+            .unwrap();
+        let message_id = db.conn().last_insert_rowid();
+
+        let bridge = dir.path().join("grok");
+        std::fs::write(&bridge, FAKE_BRIDGE).unwrap();
+        std::fs::set_permissions(&bridge, std::fs::Permissions::from_mode(0o755)).unwrap();
+        let socket = dir.path().join("leader.sock");
+        std::fs::write(socket.with_extension("lock"), "1001\n").unwrap();
+        let launch = launch_at(&bridge.to_string_lossy(), &socket);
+
+        let running = Arc::new(AtomicBool::new(true));
+        let watcher = {
+            let running = running.clone();
+            let db_path = db_path.clone();
+            std::thread::spawn(move || {
+                let db = HcomDb::open_at(&db_path).unwrap();
+                let deadline = Instant::now() + Duration::from_secs(30);
+                while db.get_cursor("nova") < message_id && Instant::now() < deadline {
+                    std::thread::sleep(Duration::from_millis(50));
+                }
+                running.store(false, Ordering::Release);
+            })
+        };
+        let notify = NotifyServer::new().unwrap();
+        let state = DeliveryState {
+            screen: Arc::new(RwLock::new(Default::default())),
+            grok_acp: None,
+            launch_phase_active: Arc::new(AtomicBool::new(false)),
+            inject_port: 0,
+            user_activity_cooldown_ms: 0,
+        };
+        let mut name = "nova".to_string();
+        run(
+            &launch,
+            &running,
+            &mut db,
+            &notify,
+            &state,
+            "proc-1",
+            &mut name,
+            &ToolConfig::for_tool(crate::tool::Tool::Grok),
+            &None,
+            &None,
+            &None,
+            &mut super::super::host_label::HostLabel::resolve(),
+            &mut LaunchOutcome::Ready,
+        );
+        watcher.join().unwrap();
+
+        assert_eq!(
+            db.get_cursor("nova"),
+            message_id,
+            "the batch was never acked"
+        );
+        let runs = std::fs::read_to_string(dir.path().join("runs")).unwrap();
+        assert_eq!(runs.trim(), "2", "hcom should reconnect exactly once");
+    }
 }

From 415db7c8fea8a6cad379931957bda774deb79292 Mon Sep 17 00:00:00 2001
From: signet-bot-claude <261785442+signet-bot-claude@users.noreply.github.com>
Date: Sat, 17 Oct 2026 11:10:43 -0400
Subject: [PATCH 2/4] Grok: leave a batch the new leader still queues to wait
 for its turn

After a reconnect, a batch that Grok's new connection still lists in its
queue was dropped anyway once DETACHED_WAIT passed with no other evidence,
and hcom queued the same mail a second time behind the first copy. The new
leader-restart and unanswered-request triggers make that reconnect common:
a batch can reach the restarted leader before hcom notices the restart.

A queue snapshot that lists the detached batch now stops that timeout; the
batch then follows the queue until it runs or leaves it. A later drop re-arms
the timeout, since the next connection must show it again.

Tests: the fake bridge can keep the lock's PID (only the unanswered request
shows the loss, end to end) or list the batch behind a user turn and run it
after DETACHED_WAIT (no second session/prompt). REQUEST_TIMEOUT and
DETACHED_WAIT are short under cfg(test), like REQUEUE_DELAY.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/delivery/grok.rs | 127 ++++++++++++++++++++++++++++++++++---------
 1 file changed, 102 insertions(+), 25 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 185e07b1..1d18980c 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -69,7 +69,7 @@ const REQUEUE_DELAY: Duration = Duration::from_secs(if cfg!(test) { 1 } else { 1
 const LOAD_RETRY_DELAY: Duration = Duration::from_secs(2);
 /// Grok answers a roster or `session/load` request in well under a second. One
 /// still unanswered after this long was lost in transit.
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(if cfg!(test) { 3 } else { 30 });
 /// How often to check which leader process holds the socket.
 const LEADER_CHECK: Duration = Duration::from_secs(2);
 /// Prompts hcom queues carry this `promptId` prefix.
@@ -770,13 +770,17 @@ struct InFlight {
     /// prompt when its client disconnects, so the batch is followed through
     /// the reconnected client's queue snapshots, not treated as dropped.
     detached_at: Option<Instant>,
+    /// Whether a snapshot since the drop still lists it in Grok's queue. It
+    /// then waits for its turn there, however long, rather than for
+    /// `DETACHED_WAIT`: queuing it again would deliver it twice.
+    queued_after_detach: bool,
 }
 
 /// How long to wait after `removedFromQueue` for the snapshot showing a merge.
 const MERGE_EVIDENCE_WAIT: Duration = Duration::from_secs(2);
 /// How long a batch whose connection dropped waits for a queue snapshot
 /// before the transcript decides.
-const DETACHED_WAIT: Duration = Duration::from_secs(30);
+const DETACHED_WAIT: Duration = Duration::from_secs(if cfg!(test) { 2 } else { 30 });
 
 #[derive(Debug, PartialEq, Eq)]
 enum Outcome {
@@ -801,16 +805,21 @@ fn queue_outcome(params: &Value, flight: &InFlight) -> Option<Outcome> {
     // still queued; gone means it ran (the transcript will show it) or was
     // removed.
     if flight.detached_at.is_some() {
-        let queued = params["entries"]
-            .as_array()
-            .is_some_and(|entries| entries.iter().any(|e| e["id"] == flight.prompt_id));
-        return (!queued).then(|| Outcome::Dropped("gone from Grok's queue".into()));
+        return (!still_queued(params, flight))
+            .then(|| Outcome::Dropped("gone from Grok's queue".into()));
     }
     // Absence proves nothing, even after `removedFromQueue`: the merge
     // snapshot can come later. `MERGE_EVIDENCE_WAIT` decides that case.
     None
 }
 
+/// Whether a queue snapshot lists our prompt as waiting.
+fn still_queued(params: &Value, flight: &InFlight) -> bool {
+    params["entries"]
+        .as_array()
+        .is_some_and(|entries| entries.iter().any(|e| e["id"] == flight.prompt_id))
+}
+
 /// Whether the session transcript shows the prompt text as sent: durable
 /// proof it ran when the live evidence was lost (bridge disconnect, a merge
 /// snapshot that never arrived).
@@ -1179,10 +1188,13 @@ pub(super) fn run(
                 }
                 Event::Notification { method, params } => {
                     if method == "x.ai/queue/changed"
-                        && let Some(flight) = in_flight.as_ref()
+                        && let Some(flight) = in_flight.as_mut()
                         && params["sessionId"].as_str() == Some(flight.session.as_str())
                     {
                         outcome = queue_outcome(&params, flight);
+                        if flight.detached_at.is_some() && still_queued(&params, flight) {
+                            flight.queued_after_detach = true;
+                        }
                     }
                     effects.extend(tracker.on_notification(&method, &params));
                 }
@@ -1296,7 +1308,12 @@ pub(super) fn run(
             retry_at = Instant::now() + Duration::from_secs(1);
             if let Some(flight) = in_flight.as_mut() {
                 flight.request_id = None;
-                flight.detached_at.get_or_insert_with(Instant::now);
+                // A batch seen queued before this drop needs fresh evidence
+                // from the next connection.
+                if flight.queued_after_detach || flight.detached_at.is_none() {
+                    flight.detached_at = Some(Instant::now());
+                }
+                flight.queued_after_detach = false;
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
@@ -1311,6 +1328,7 @@ pub(super) fn run(
         if outcome.is_none()
             && in_flight
                 .as_ref()
+                .filter(|flight| !flight.queued_after_detach)
                 .and_then(|flight| flight.detached_at)
                 .is_some_and(|at| at.elapsed() >= DETACHED_WAIT)
         {
@@ -1442,6 +1460,7 @@ pub(super) fn run(
                         ack: prepared.ack,
                         removed_at: None,
                         detached_at: None,
+                        queued_after_detach: false,
                     });
                 }
                 Err(error) => {
@@ -1480,6 +1499,7 @@ mod tests {
             },
             removed_at: removed.then(Instant::now),
             detached_at: None,
+            queued_after_detach: false,
         }
     }
 
@@ -2007,6 +2027,11 @@ mod tests {
     /// first bridge sees the leader restart as hcom's batch goes out: the
     /// lock gets the new leader's PID, and nothing written to that bridge is
     /// ever answered again. The bridge itself stays up, as Grok's does.
+    /// Flag files beside it vary the restart: `same_pid` leaves the lock
+    /// alone, so only the unanswered requests show the loss; `queued` has
+    /// the batch reach the new leader, listed behind a user turn and run
+    /// only after `DETACHED_WAIT`. Every prompt the second bridge receives
+    /// is counted in `requeued`.
     #[cfg(unix)]
     const FAKE_BRIDGE: &str = r#"#!/bin/sh
 dir=$(dirname "$0")
@@ -2020,33 +2045,38 @@ while IFS= read -r line; do
   case "$line" in
     *'"method":"initialize"'*|*'"method":"session/load"'*)
       echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{}}"
-      # The reconnected leader's queue snapshot: the lost batch is not in it.
       case "$line" in *session/load*)
-        echo '{"jsonrpc":"2.0","method":"_x.ai/queue/changed","params":{"sessionId":"s1","entries":[]}}';;
+        if [ -f "$dir/queued" ]; then
+          # The batch reached the new leader and waits behind a user turn.
+          held=$(cat "$dir/prompt")
+          echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"user-1\",\"entries\":[{\"id\":\"$held\"}]}}"
+          (sleep 4; echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$held\",\"runningKind\":\"prompt\",\"entries\":[]}}") &
+        else
+          # The reconnected leader's queue snapshot: the lost batch is not in it.
+          echo '{"jsonrpc":"2.0","method":"_x.ai/queue/changed","params":{"sessionId":"s1","entries":[]}}'
+        fi;;
       esac ;;
     *'"method":"_x.ai/sessions/list"'*)
       echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{\"sessions\":[{\"sessionId\":\"s1\",\"resident\":true,\"activity\":\"idle\",\"lastChangeUnixMs\":1}]}}" ;;
     *'"method":"session/prompt"'*)
+      prompt=$(printf '%s' "$line" | sed -n 's/.*"promptId":"\([^"]*\)".*/\1/p')
       if [ "$run" = 1 ]; then
-        echo 2002 > "$lock"
+        echo "$prompt" > "$dir/prompt"
+        [ -f "$dir/same_pid" ] || echo 2002 > "$lock"
         lost=1
       else
-        prompt=$(printf '%s' "$line" | sed -n 's/.*"promptId":"\([^"]*\)".*/\1/p')
+        echo "$prompt" >> "$dir/requeued"
         echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$prompt\",\"runningKind\":\"prompt\",\"entries\":[]}}"
       fi ;;
   esac
 done
 "#;
 
-    /// Observed: the leader auto-updated and restarted while idle; the
-    /// bridge reconnected without a word and the batch queued during the gap
-    /// never reached Grok. With no response or queue event for it, the batch
-    /// stayed in flight, no later mail was queued, and unread climbed until
-    /// someone typed into the agent's terminal.
+    /// Runs one delivery through the fake bridge with the given flag files
+    /// until the batch is acked or 30 s pass. Returns whether it was acked,
+    /// how many bridges hcom started, and the prompts the second one got.
     #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn batch_lost_in_a_silent_leader_restart_is_queued_again() {
+    fn deliver_across_restart(flags: &[&str]) -> (bool, String, Vec<String>) {
         use std::os::unix::fs::PermissionsExt;
 
         let (dir, hcom_dir, _home, _guard) = crate::hooks::test_helpers::isolated_test_env();
@@ -2081,6 +2111,9 @@ done
         let bridge = dir.path().join("grok");
         std::fs::write(&bridge, FAKE_BRIDGE).unwrap();
         std::fs::set_permissions(&bridge, std::fs::Permissions::from_mode(0o755)).unwrap();
+        for flag in flags {
+            std::fs::write(dir.path().join(flag), "").unwrap();
+        }
         let socket = dir.path().join("leader.sock");
         std::fs::write(socket.with_extension("lock"), "1001\n").unwrap();
         let launch = launch_at(&bridge.to_string_lossy(), &socket);
@@ -2124,12 +2157,56 @@ done
         );
         watcher.join().unwrap();
 
+        let read = |name: &str| std::fs::read_to_string(dir.path().join(name)).unwrap_or_default();
+        (
+            db.get_cursor("nova") == message_id,
+            read("runs").trim().to_string(),
+            read("requeued").lines().map(str::to_string).collect(),
+        )
+    }
+
+    /// Observed: the leader auto-updated and restarted while idle; the
+    /// bridge reconnected without a word and the batch queued during the gap
+    /// never reached Grok. With no response or queue event for it, the batch
+    /// stayed in flight, no later mail was queued, and unread climbed until
+    /// someone typed into the agent's terminal.
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn batch_lost_in_a_silent_leader_restart_is_queued_again() {
+        let (acked, runs, requeued) = deliver_across_restart(&[]);
+        assert!(acked, "the batch was never acked");
+        assert_eq!(runs, "2", "hcom should reconnect exactly once");
+        assert_eq!(requeued.len(), 1, "the lost batch is queued again once");
+    }
+
+    /// The lock can keep the same PID (a restart between checks, or a lost
+    /// write with no restart at all). The unanswered roster or load request
+    /// alone must still drop the connection and requeue the batch.
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn batch_lost_behind_unanswered_requests_is_queued_again() {
+        let (acked, runs, requeued) = deliver_across_restart(&["same_pid"]);
+        assert!(acked, "the batch was never acked");
+        assert_eq!(runs, "2", "hcom should reconnect exactly once");
+        assert_eq!(requeued.len(), 1, "the lost batch is queued again once");
+    }
+
+    /// A batch that did reach the new leader, listed in its queue behind a
+    /// long user turn, must wait there: queuing it again would show Grok the
+    /// same mail twice.
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn batch_still_queued_after_a_restart_is_not_queued_twice() {
+        let (acked, runs, requeued) = deliver_across_restart(&["queued"]);
+        assert!(acked, "the batch was never acked");
+        assert_eq!(runs, "2", "hcom should reconnect exactly once");
         assert_eq!(
-            db.get_cursor("nova"),
-            message_id,
-            "the batch was never acked"
+            requeued,
+            Vec::<String>::new(),
+            "the queued batch was sent again"
         );
-        let runs = std::fs::read_to_string(dir.path().join("runs")).unwrap();
-        assert_eq!(runs.trim(), "2", "hcom should reconnect exactly once");
     }
 }

From a3cdd02f43d046148230c877e606a5185c500e7b Mon Sep 17 00:00:00 2001
From: aannoo <aannoo@users.noreply.github.com>
Date: Tue, 20 Oct 2026 03:53:58 +0100
Subject: [PATCH 3/4] Grok: treat the bridge's leader_reconnected as a lost
 connection

The stdio bridge announces a leader restart with x.ai/leader_reconnected,
so drop the lock-file PID polling and request timeouts and reuse the
existing reconnect path on that notification.
---
 src/delivery/grok.rs | 238 +++++++++----------------------------------
 1 file changed, 50 insertions(+), 188 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 1d18980c..7ad78f22 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -27,11 +27,10 @@
 //! A batch that never started (removed from the queue, transport lost) stays
 //! unread and is queued again later: at-least-once, never silently dropped.
 //!
-//! The bridge reconnects on its own when the leader restarts (Grok's leader
-//! restarts itself to apply an auto-update) and tells hcom nothing. Requests
-//! written while the leader was down are lost and never answered. hcom
-//! treats a new leader PID in the socket's lock file, or a roster or load
-//! request left unanswered, as a lost connection.
+//! When the leader restarts (it does so to apply an auto-update), the bridge
+//! reconnects to the new one, replays the session loads and announces it with
+//! `x.ai/leader_reconnected`. Requests the old leader had not answered never
+//! will be, so hcom handles it as a lost connection.
 
 use std::collections::{HashMap, HashSet};
 use std::io::{BufRead, BufReader, Write};
@@ -67,11 +66,6 @@ const ROSTER_SLOW_POLL: Duration = Duration::from_secs(15);
 const REQUEUE_DELAY: Duration = Duration::from_secs(if cfg!(test) { 1 } else { 10 });
 /// Wait before loading the bound session again after a failed load.
 const LOAD_RETRY_DELAY: Duration = Duration::from_secs(2);
-/// Grok answers a roster or `session/load` request in well under a second. One
-/// still unanswered after this long was lost in transit.
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(if cfg!(test) { 3 } else { 30 });
-/// How often to check which leader process holds the socket.
-const LEADER_CHECK: Duration = Duration::from_secs(2);
 /// Prompts hcom queues carry this `promptId` prefix.
 const HCOM_PROMPT_ID_PREFIX: &str = "hcom-";
 /// Leader-mode Grok ignores these (it warns and continues). Since hcom must
@@ -160,11 +154,11 @@ impl Launch {
     /// PID to the socket's sibling `.lock` (`x.sock` -> `x.lock`).
     pub(crate) fn stop_leader(&self) {
         let socket = std::path::Path::new(&self.socket);
-        let lock = self.lock_path();
-        if let Some(pid) = self
-            .leader_pid()
-            .filter(|pid| crate::sys::process::is_alive(*pid))
-        {
+        let lock = socket.with_extension("lock");
+        let pid = std::fs::read_to_string(&lock)
+            .ok()
+            .and_then(|content| content.trim().parse::<u32>().ok());
+        if let Some(pid) = pid.filter(|pid| crate::sys::process::is_alive(*pid)) {
             crate::sys::process::terminate(pid);
             let deadline = Instant::now() + Duration::from_secs(3);
             while crate::sys::process::is_alive(pid) && Instant::now() < deadline {
@@ -179,19 +173,6 @@ impl Launch {
         let _ = std::fs::remove_file(lock);
     }
 
-    fn lock_path(&self) -> std::path::PathBuf {
-        std::path::Path::new(&self.socket).with_extension("lock")
-    }
-
-    /// PID of the leader that holds this launch's socket, from its `.lock`.
-    fn leader_pid(&self) -> Option<u32> {
-        std::fs::read_to_string(self.lock_path())
-            .ok()?
-            .trim()
-            .parse()
-            .ok()
-    }
-
     /// Env for the TUI and the ACP client. The leader is spawned by whichever
     /// connects first and inherits it; `--no-subagents` itself has no effect
     /// in leader mode.
@@ -863,38 +844,6 @@ enum Pending {
     Load(String),
 }
 
-/// Requests awaiting a response on the current connection, and when each was
-/// sent.
-type Outstanding = HashMap<u64, (Pending, Instant)>;
-
-fn awaits(pending: &Outstanding, wanted: impl Fn(&Pending) -> bool) -> bool {
-    pending.values().any(|(request, _)| wanted(request))
-}
-
-/// A request unanswered for `REQUEST_TIMEOUT`, as its method name.
-fn stalled_request(pending: &Outstanding, now: Instant) -> Option<&'static str> {
-    pending
-        .values()
-        .find(|(_, sent)| now.saturating_duration_since(*sent) >= REQUEST_TIMEOUT)
-        .map(|(request, _)| match request {
-            Pending::Roster => "x.ai/sessions/list",
-            Pending::Load(_) => "session/load",
-        })
-}
-
-/// Whether another leader took over the socket since hcom attached. The
-/// bridge follows a restarted leader without a word, so this is the only sign
-/// that requests written in between were lost. An unreadable lock proves
-/// nothing.
-fn leader_change(attached: Option<u32>, current: Option<u32>) -> Option<String> {
-    match (attached, current) {
-        (Some(old), Some(new)) if old != new => {
-            Some(format!("Grok leader restarted (pid {old} -> {new})"))
-        }
-        _ => None,
-    }
-}
-
 /// The roster result, which the stdio bridge may wrap once more.
 fn roster_sessions(result: &Value) -> &[Value] {
     result
@@ -1002,7 +951,7 @@ pub(super) fn run(
     let mut client: Option<Client> = None;
     let mut tracker = Tracker::default();
     let mut loaded: HashSet<String> = HashSet::new();
-    let mut pending: Outstanding = HashMap::new();
+    let mut pending: HashMap<u64, Pending> = HashMap::new();
     let mut roster_due: Option<Instant> = None;
     let mut roster_fast_until = Instant::now();
     let mut in_flight: Option<InFlight> = None;
@@ -1012,8 +961,6 @@ pub(super) fn run(
     let mut connect_failures: u32 = 0;
     let mut retry_at = Instant::now();
     let mut requeue_at = Instant::now();
-    let mut attached_leader: Option<u32> = None;
-    let mut leader_checked = Instant::now();
     let launched_at = Instant::now();
     while running.load(Ordering::Acquire) {
         super::refresh_title_state(super::TitleRefresh {
@@ -1072,8 +1019,6 @@ pub(super) fn run(
                 Ok(connected) => {
                     client = Some(connected);
                     connect_failures = 0;
-                    attached_leader = launch.leader_pid();
-                    leader_checked = Instant::now();
                     tracker.reconnected();
                     loaded.clear();
                     pending.clear();
@@ -1133,10 +1078,7 @@ pub(super) fn run(
                 }
                 Event::Response(value) => {
                     let id = value["id"].as_u64();
-                    match id
-                        .and_then(|id| pending.remove(&id))
-                        .map(|(request, _)| request)
-                    {
+                    match id.and_then(|id| pending.remove(&id)) {
                         Some(Pending::Roster) => match value.get("error") {
                             Some(error) => {
                                 log_warn("native", "grok.acp.roster_failed", &error.to_string())
@@ -1187,6 +1129,10 @@ pub(super) fn run(
                     }
                 }
                 Event::Notification { method, params } => {
+                    if method == "x.ai/leader_reconnected" {
+                        closed = Some("Grok leader restarted".into());
+                        break;
+                    }
                     if method == "x.ai/queue/changed"
                         && let Some(flight) = in_flight.as_mut()
                         && params["sessionId"].as_str() == Some(flight.session.as_str())
@@ -1239,7 +1185,9 @@ pub(super) fn run(
                 }
                 Effect::Load { session, cwd } => {
                     if loaded.contains(&session)
-                        || awaits(&pending, |p| matches!(p, Pending::Load(s) if *s == session))
+                        || pending
+                            .values()
+                            .any(|p| matches!(p, Pending::Load(s) if *s == session))
                     {
                         continue;
                     }
@@ -1254,7 +1202,7 @@ pub(super) fn run(
                                "_meta": {"noReplay": true}}),
                     ) {
                         Ok(id) => {
-                            pending.insert(id, (Pending::Load(session), Instant::now()));
+                            pending.insert(id, Pending::Load(session));
                         }
                         Err(error) => closed = Some(format!("write failed: {error:#}")),
                     }
@@ -1267,7 +1215,7 @@ pub(super) fn run(
                 }
             }
         }
-        let roster_pending = awaits(&pending, |p| matches!(p, Pending::Roster));
+        let roster_pending = pending.values().any(|p| matches!(p, Pending::Roster));
         if !roster_pending && roster_due.is_none() {
             let fast = tracker.bound.is_none() || Instant::now() < roster_fast_until;
             let interval = if fast {
@@ -1281,28 +1229,11 @@ pub(super) fn run(
             roster_due = None;
             match conn.send("_x.ai/sessions/list", json!({})) {
                 Ok(id) => {
-                    pending.insert(id, (Pending::Roster, Instant::now()));
+                    pending.insert(id, Pending::Roster);
                 }
                 Err(error) => closed = Some(format!("write failed: {error:#}")),
             }
         }
-        // The bridge is still up, but what was written to it may never be
-        // answered: drop it like a lost connection, so the in-flight batch is
-        // followed through the next connection's queue and requeued if gone.
-        if closed.is_none() && leader_checked.elapsed() >= LEADER_CHECK {
-            leader_checked = Instant::now();
-            let current = launch.leader_pid();
-            closed = leader_change(attached_leader, current);
-            attached_leader = attached_leader.or(current);
-        }
-        if closed.is_none()
-            && let Some(method) = stalled_request(&pending, Instant::now())
-        {
-            closed = Some(format!(
-                "{method} unanswered after {}s",
-                REQUEST_TIMEOUT.as_secs()
-            ));
-        }
         if let Some(error) = closed {
             client = None;
             retry_at = Instant::now() + Duration::from_secs(1);
@@ -1389,7 +1320,9 @@ pub(super) fn run(
         // is retried.
         if let Some(session) = tracker.bound.clone()
             && !loaded.contains(&session)
-            && !awaits(&pending, |p| matches!(p, Pending::Load(s) if *s == session))
+            && !pending
+                .values()
+                .any(|p| matches!(p, Pending::Load(s) if *s == session))
             && Instant::now() >= load_retry_at
         {
             let cwd = tracker
@@ -1406,7 +1339,7 @@ pub(super) fn run(
                        "_meta": {"noReplay": true}}),
             ) {
                 Ok(id) => {
-                    pending.insert(id, (Pending::Load(session), Instant::now()));
+                    pending.insert(id, Pending::Load(session));
                 }
                 Err(error) => {
                     log_warn("native", "grok.acp.write_failed", &format!("{error:#}"));
@@ -1969,78 +1902,19 @@ mod tests {
         assert!(plain.child_env().is_empty());
     }
 
-    fn launch_at(command: &str, socket: &std::path::Path) -> Launch {
-        Launch {
-            command: command.into(),
-            prefix: Vec::new(),
-            socket: socket.to_string_lossy().into_owned(),
-            no_subagents: false,
-        }
-    }
-
-    #[test]
-    fn leader_pid_comes_from_the_lock_beside_the_socket() {
-        let dir = tempfile::tempdir().unwrap();
-        let launch = launch_at("grok", &dir.path().join("hcom-grok-x.sock"));
-        assert_eq!(launch.leader_pid(), None);
-        let lock = dir.path().join("hcom-grok-x.lock");
-        std::fs::write(&lock, "77001\n").unwrap();
-        assert_eq!(launch.leader_pid(), Some(77001));
-        std::fs::write(&lock, "").unwrap();
-        assert_eq!(launch.leader_pid(), None);
-    }
-
-    #[test]
-    fn a_new_leader_pid_means_the_connection_was_lost() {
-        assert!(
-            leader_change(Some(77001), Some(94509))
-                .is_some_and(|why| why.contains("77001 -> 94509"))
-        );
-        assert_eq!(leader_change(Some(77001), Some(77001)), None);
-        // An unreadable lock (mid-restart, never written) proves nothing.
-        assert_eq!(leader_change(Some(77001), None), None);
-        assert_eq!(leader_change(None, Some(94509)), None);
-    }
-
-    #[test]
-    fn a_roster_or_load_left_unanswered_was_lost() {
-        let sent = Instant::now();
-        let mut pending: Outstanding = HashMap::new();
-        pending.insert(1, (Pending::Roster, sent));
-        assert_eq!(
-            stalled_request(&pending, sent + Duration::from_secs(1)),
-            None
-        );
-        assert_eq!(
-            stalled_request(&pending, sent + REQUEST_TIMEOUT),
-            Some("x.ai/sessions/list")
-        );
-        pending.clear();
-        pending.insert(2, (Pending::Load("s".into()), sent));
-        assert_eq!(
-            stalled_request(&pending, sent + REQUEST_TIMEOUT),
-            Some("session/load")
-        );
-    }
-
-    /// Stands in for `grok agent --leader --leader-socket <sock> stdio`. The
-    /// first bridge sees the leader restart as hcom's batch goes out: the
-    /// lock gets the new leader's PID, and nothing written to that bridge is
-    /// ever answered again. The bridge itself stays up, as Grok's does.
-    /// Flag files beside it vary the restart: `same_pid` leaves the lock
-    /// alone, so only the unanswered requests show the loss; `queued` has
-    /// the batch reach the new leader, listed behind a user turn and run
-    /// only after `DETACHED_WAIT`. Every prompt the second bridge receives
-    /// is counted in `requeued`.
+    /// Stands in for `grok agent --leader --leader-socket <sock> stdio`. On
+    /// the first bridge the leader restarts as hcom's batch goes out: the
+    /// bridge announces `x.ai/leader_reconnected`, and the old leader never
+    /// answers the prompt. With a `queued` flag file beside it, the batch
+    /// reached the new leader, listed behind a user turn and run only after
+    /// `DETACHED_WAIT`. Every prompt the second bridge receives is counted in
+    /// `requeued`.
     #[cfg(unix)]
     const FAKE_BRIDGE: &str = r#"#!/bin/sh
 dir=$(dirname "$0")
-lock="${4%.sock}.lock"
 run=$(( $(cat "$dir/runs" 2>/dev/null || echo 0) + 1 ))
 echo "$run" > "$dir/runs"
-lost=
 while IFS= read -r line; do
-  [ -z "$lost" ] || continue
   id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9][0-9]*\).*/\1/p')
   case "$line" in
     *'"method":"initialize"'*|*'"method":"session/load"'*)
@@ -2052,7 +1926,7 @@ while IFS= read -r line; do
           echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"user-1\",\"entries\":[{\"id\":\"$held\"}]}}"
           (sleep 4; echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$held\",\"runningKind\":\"prompt\",\"entries\":[]}}") &
         else
-          # The reconnected leader's queue snapshot: the lost batch is not in it.
+          # The new leader's queue snapshot: the lost batch is not in it.
           echo '{"jsonrpc":"2.0","method":"_x.ai/queue/changed","params":{"sessionId":"s1","entries":[]}}'
         fi;;
       esac ;;
@@ -2062,8 +1936,7 @@ while IFS= read -r line; do
       prompt=$(printf '%s' "$line" | sed -n 's/.*"promptId":"\([^"]*\)".*/\1/p')
       if [ "$run" = 1 ]; then
         echo "$prompt" > "$dir/prompt"
-        [ -f "$dir/same_pid" ] || echo 2002 > "$lock"
-        lost=1
+        echo '{"jsonrpc":"2.0","method":"x.ai/leader_reconnected","params":{}}'
       else
         echo "$prompt" >> "$dir/requeued"
         echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$prompt\",\"runningKind\":\"prompt\",\"entries\":[]}}"
@@ -2114,9 +1987,16 @@ done
         for flag in flags {
             std::fs::write(dir.path().join(flag), "").unwrap();
         }
-        let socket = dir.path().join("leader.sock");
-        std::fs::write(socket.with_extension("lock"), "1001\n").unwrap();
-        let launch = launch_at(&bridge.to_string_lossy(), &socket);
+        let launch = Launch {
+            command: bridge.to_string_lossy().into_owned(),
+            prefix: Vec::new(),
+            socket: dir
+                .path()
+                .join("leader.sock")
+                .to_string_lossy()
+                .into_owned(),
+            no_subagents: false,
+        };
 
         let running = Arc::new(AtomicBool::new(true));
         let watcher = {
@@ -2165,34 +2045,20 @@ done
         )
     }
 
-    /// Observed: the leader auto-updated and restarted while idle; the
-    /// bridge reconnected without a word and the batch queued during the gap
-    /// never reached Grok. With no response or queue event for it, the batch
-    /// stayed in flight, no later mail was queued, and unread climbed until
-    /// someone typed into the agent's terminal.
+    /// Observed: the leader auto-updated and restarted while idle, and the
+    /// batch queued across the restart was never answered. It stayed in
+    /// flight, no later mail was queued, and unread climbed until someone
+    /// typed into the agent's terminal.
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
-    fn batch_lost_in_a_silent_leader_restart_is_queued_again() {
+    fn batch_lost_in_a_leader_restart_is_queued_again() {
         let (acked, runs, requeued) = deliver_across_restart(&[]);
         assert!(acked, "the batch was never acked");
         assert_eq!(runs, "2", "hcom should reconnect exactly once");
         assert_eq!(requeued.len(), 1, "the lost batch is queued again once");
     }
 
-    /// The lock can keep the same PID (a restart between checks, or a lost
-    /// write with no restart at all). The unanswered roster or load request
-    /// alone must still drop the connection and requeue the batch.
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn batch_lost_behind_unanswered_requests_is_queued_again() {
-        let (acked, runs, requeued) = deliver_across_restart(&["same_pid"]);
-        assert!(acked, "the batch was never acked");
-        assert_eq!(runs, "2", "hcom should reconnect exactly once");
-        assert_eq!(requeued.len(), 1, "the lost batch is queued again once");
-    }
-
     /// A batch that did reach the new leader, listed in its queue behind a
     /// long user turn, must wait there: queuing it again would show Grok the
     /// same mail twice.
@@ -2203,10 +2069,6 @@ done
         let (acked, runs, requeued) = deliver_across_restart(&["queued"]);
         assert!(acked, "the batch was never acked");
         assert_eq!(runs, "2", "hcom should reconnect exactly once");
-        assert_eq!(
-            requeued,
-            Vec::<String>::new(),
-            "the queued batch was sent again"
-        );
+        assert!(requeued.is_empty(), "the queued batch was sent again");
     }
 }

From 1235fbb7c677c3b8d49fd1152c97240fbc17b0bf Mon Sep 17 00:00:00 2001
From: aannoo <aannoo@users.noreply.github.com>
Date: Tue, 20 Oct 2026 04:31:50 +0100
Subject: [PATCH 4/4] Grok: requeue a detached batch only once its session is
 idle

Reconnecting brings no queue snapshot, so a batch still waiting behind a
long turn looked gone after DETACHED_WAIT and was queued twice. Grok runs
a queued prompt as soon as its session is idle, so wait for a roster
since the drop that shows the session neither working nor waiting on an
approval (needs_input) before the transcript decides. This replaces
queued_after_detach, which relied on a snapshot that never comes.
---
 src/delivery/grok.rs | 113 ++++++++++++++++++++++++++-----------------
 1 file changed, 68 insertions(+), 45 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 7ad78f22..0441fe3f 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -751,16 +751,19 @@ struct InFlight {
     /// prompt when its client disconnects, so the batch is followed through
     /// the reconnected client's queue snapshots, not treated as dropped.
     detached_at: Option<Instant>,
-    /// Whether a snapshot since the drop still lists it in Grok's queue. It
-    /// then waits for its turn there, however long, rather than for
-    /// `DETACHED_WAIT`: queuing it again would deliver it twice.
-    queued_after_detach: bool,
+    /// Whether the latest roster since the drop shows its session idle.
+    /// Reconnecting brings no queue snapshot, and Grok runs a queued prompt
+    /// as soon as its session is idle, so only an idle session proves the
+    /// batch is no longer waiting behind a turn. (A row the user holds for
+    /// editing at the front can still park it, for up to Grok's edit-hold
+    /// TTL; that stays a possible duplicate.)
+    idle_after_detach: bool,
 }
 
 /// How long to wait after `removedFromQueue` for the snapshot showing a merge.
 const MERGE_EVIDENCE_WAIT: Duration = Duration::from_secs(2);
 /// How long a batch whose connection dropped waits for a queue snapshot
-/// before the transcript decides.
+/// before the transcript decides, once its session is idle.
 const DETACHED_WAIT: Duration = Duration::from_secs(if cfg!(test) { 2 } else { 30 });
 
 #[derive(Debug, PartialEq, Eq)]
@@ -786,21 +789,16 @@ fn queue_outcome(params: &Value, flight: &InFlight) -> Option<Outcome> {
     // still queued; gone means it ran (the transcript will show it) or was
     // removed.
     if flight.detached_at.is_some() {
-        return (!still_queued(params, flight))
-            .then(|| Outcome::Dropped("gone from Grok's queue".into()));
+        let queued = params["entries"]
+            .as_array()
+            .is_some_and(|entries| entries.iter().any(|e| e["id"] == flight.prompt_id));
+        return (!queued).then(|| Outcome::Dropped("gone from Grok's queue".into()));
     }
     // Absence proves nothing, even after `removedFromQueue`: the merge
     // snapshot can come later. `MERGE_EVIDENCE_WAIT` decides that case.
     None
 }
 
-/// Whether a queue snapshot lists our prompt as waiting.
-fn still_queued(params: &Value, flight: &InFlight) -> bool {
-    params["entries"]
-        .as_array()
-        .is_some_and(|entries| entries.iter().any(|e| e["id"] == flight.prompt_id))
-}
-
 /// Whether the session transcript shows the prompt text as sent: durable
 /// proof it ran when the live evidence was lost (bridge disconnect, a merge
 /// snapshot that never arrived).
@@ -854,6 +852,15 @@ fn roster_sessions(result: &Value) -> &[Value] {
         .unwrap_or_default()
 }
 
+/// Whether the roster shows `session` in a turn: running, or waiting on an
+/// approval or question (`needs_input` wins over `working` mid-turn).
+fn roster_busy(sessions: &[Value], session: &str) -> bool {
+    sessions.iter().any(|entry| {
+        entry["sessionId"] == session
+            && matches!(entry["activity"].as_str(), Some("working" | "needs_input"))
+    })
+}
+
 /// Point the instance at `session`, as a hook-based SessionStart would.
 /// Returns the instance name, which a switch to a session bound to another
 /// identity changes.
@@ -1084,7 +1091,14 @@ pub(super) fn run(
                                 log_warn("native", "grok.acp.roster_failed", &error.to_string())
                             }
                             None => {
-                                effects.extend(tracker.on_roster(roster_sessions(&value["result"])))
+                                let sessions = roster_sessions(&value["result"]);
+                                if let Some(flight) = in_flight.as_mut()
+                                    && flight.detached_at.is_some()
+                                {
+                                    flight.idle_after_detach =
+                                        !roster_busy(sessions, &flight.session);
+                                }
+                                effects.extend(tracker.on_roster(sessions))
                             }
                         },
                         Some(Pending::Load(session)) => match value.get("error") {
@@ -1134,13 +1148,10 @@ pub(super) fn run(
                         break;
                     }
                     if method == "x.ai/queue/changed"
-                        && let Some(flight) = in_flight.as_mut()
+                        && let Some(flight) = in_flight.as_ref()
                         && params["sessionId"].as_str() == Some(flight.session.as_str())
                     {
                         outcome = queue_outcome(&params, flight);
-                        if flight.detached_at.is_some() && still_queued(&params, flight) {
-                            flight.queued_after_detach = true;
-                        }
                     }
                     effects.extend(tracker.on_notification(&method, &params));
                 }
@@ -1239,12 +1250,8 @@ pub(super) fn run(
             retry_at = Instant::now() + Duration::from_secs(1);
             if let Some(flight) = in_flight.as_mut() {
                 flight.request_id = None;
-                // A batch seen queued before this drop needs fresh evidence
-                // from the next connection.
-                if flight.queued_after_detach || flight.detached_at.is_none() {
-                    flight.detached_at = Some(Instant::now());
-                }
-                flight.queued_after_detach = false;
+                flight.detached_at.get_or_insert_with(Instant::now);
+                flight.idle_after_detach = false;
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
@@ -1259,7 +1266,7 @@ pub(super) fn run(
         if outcome.is_none()
             && in_flight
                 .as_ref()
-                .filter(|flight| !flight.queued_after_detach)
+                .filter(|flight| flight.idle_after_detach)
                 .and_then(|flight| flight.detached_at)
                 .is_some_and(|at| at.elapsed() >= DETACHED_WAIT)
         {
@@ -1393,7 +1400,7 @@ pub(super) fn run(
                         ack: prepared.ack,
                         removed_at: None,
                         detached_at: None,
-                        queued_after_detach: false,
+                        idle_after_detach: false,
                     });
                 }
                 Err(error) => {
@@ -1432,7 +1439,7 @@ mod tests {
             },
             removed_at: removed.then(Instant::now),
             detached_at: None,
-            queued_after_detach: false,
+            idle_after_detach: false,
         }
     }
 
@@ -1493,6 +1500,18 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn roster_busy_includes_pending_approvals() {
+        let roster = |activity: &str| {
+            vec![json!({"sessionId": "s", "resident": true, "activity": activity})]
+        };
+        assert!(roster_busy(&roster("working"), "s"));
+        // An approval mid-turn reports needs_input, not working.
+        assert!(roster_busy(&roster("needs_input"), "s"));
+        assert!(!roster_busy(&roster("idle"), "s"));
+        assert!(!roster_busy(&roster("working"), "other"));
+    }
+
     #[test]
     fn transcript_proves_a_batch_ran() {
         let dir = tempfile::tempdir().unwrap();
@@ -1904,11 +1923,13 @@ mod tests {
 
     /// Stands in for `grok agent --leader --leader-socket <sock> stdio`. On
     /// the first bridge the leader restarts as hcom's batch goes out: the
-    /// bridge announces `x.ai/leader_reconnected`, and the old leader never
-    /// answers the prompt. With a `queued` flag file beside it, the batch
-    /// reached the new leader, listed behind a user turn and run only after
-    /// `DETACHED_WAIT`. Every prompt the second bridge receives is counted in
-    /// `requeued`.
+    /// bridge announces `x.ai/leader_reconnected`, the old leader never
+    /// answers the prompt, and the new one has an empty queue. With a
+    /// `queued` flag file beside it, only the bridge drops instead: the
+    /// batch stays queued behind a turn that ends after `DETACHED_WAIT`. As
+    /// with Grok, loading a session sends no queue snapshot; only the
+    /// roster shows the turn. Every prompt the second bridge receives is
+    /// counted in `requeued`.
     #[cfg(unix)]
     const FAKE_BRIDGE: &str = r#"#!/bin/sh
 dir=$(dirname "$0")
@@ -1920,22 +1941,24 @@ while IFS= read -r line; do
     *'"method":"initialize"'*|*'"method":"session/load"'*)
       echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{}}"
       case "$line" in *session/load*)
-        if [ -f "$dir/queued" ]; then
-          # The batch reached the new leader and waits behind a user turn.
+        if [ -f "$dir/turn" ]; then
+          # The turn ends and the held batch runs.
           held=$(cat "$dir/prompt")
-          echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"user-1\",\"entries\":[{\"id\":\"$held\"}]}}"
-          (sleep 4; echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$held\",\"runningKind\":\"prompt\",\"entries\":[]}}") &
-        else
-          # The new leader's queue snapshot: the lost batch is not in it.
-          echo '{"jsonrpc":"2.0","method":"_x.ai/queue/changed","params":{"sessionId":"s1","entries":[]}}'
+          (sleep 4; rm -f "$dir/turn"; echo "{\"jsonrpc\":\"2.0\",\"method\":\"_x.ai/queue/changed\",\"params\":{\"sessionId\":\"s1\",\"runningPromptId\":\"$held\",\"runningKind\":\"prompt\",\"entries\":[]}}") &
         fi;;
       esac ;;
     *'"method":"_x.ai/sessions/list"'*)
-      echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{\"sessions\":[{\"sessionId\":\"s1\",\"resident\":true,\"activity\":\"idle\",\"lastChangeUnixMs\":1}]}}" ;;
+      activity=idle
+      [ -f "$dir/turn" ] && activity=working
+      echo "{\"jsonrpc\":\"2.0\",\"id\":$id,\"result\":{\"sessions\":[{\"sessionId\":\"s1\",\"resident\":true,\"activity\":\"$activity\",\"lastChangeUnixMs\":1}]}}" ;;
     *'"method":"session/prompt"'*)
       prompt=$(printf '%s' "$line" | sed -n 's/.*"promptId":"\([^"]*\)".*/\1/p')
       if [ "$run" = 1 ]; then
         echo "$prompt" > "$dir/prompt"
+        if [ -f "$dir/queued" ]; then
+          touch "$dir/turn"
+          exit 0
+        fi
         echo '{"jsonrpc":"2.0","method":"x.ai/leader_reconnected","params":{}}'
       else
         echo "$prompt" >> "$dir/requeued"
@@ -2059,13 +2082,13 @@ done
         assert_eq!(requeued.len(), 1, "the lost batch is queued again once");
     }
 
-    /// A batch that did reach the new leader, listed in its queue behind a
-    /// long user turn, must wait there: queuing it again would show Grok the
-    /// same mail twice.
+    /// Observed: only the bridge dropped while the batch waited behind a long
+    /// tool call. Reconnecting brought no queue snapshot, hcom queued the
+    /// batch again, and Grok ran both copies once the tool finished.
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
-    fn batch_still_queued_after_a_restart_is_not_queued_twice() {
+    fn batch_queued_behind_a_turn_across_a_bridge_drop_is_not_queued_twice() {
         let (acked, runs, requeued) = deliver_across_restart(&["queued"]);
         assert!(acked, "the batch was never acked");
         assert_eq!(runs, "2", "hcom should reconnect exactly once");