feat(acp): Hermes runtime discovery and durable session load

crates/buzz-acp/Cargo.toml

@@ -68,6 +68,11 @@ clap = { version = "4", features = ["derive", "env"] }
 # Config file
 toml = "1.0"
 
+# Durable ACP session binding store location
+dirs = "6"
+# Cross-process flock for shared session bindings
+fs2 = "0.4"
+
 # Filter expressions
 evalexpr = { workspace = true }
 
@@ -78,4 +83,5 @@ nix = { version = "0.31", default-features = false, features = ["signal"] }
 
 [dev-dependencies]
 tokio = { workspace = true, features = ["test-util"] }
+tempfile = "3"
 httparse = "1"

crates/buzz-acp/src/acp.rs

@@ -689,6 +689,50 @@ impl AcpClient {
             .session_id)
     }
 
+    /// Send `session/load` for an existing ACP session id.
+    ///
+    /// Used after harness restart when a durable channel→session binding is
+    /// known and the agent advertised `agentCapabilities.loadSession`.
+    /// History-replay `session/update` notifications are consumed by the
+    /// request loop without entering the observer feed, so relay observers do
+    /// not republish the loaded transcript.
+    pub async fn session_load_full(
+        &mut self,
+        cwd: &str,
+        session_id: &str,
+        mcp_servers: Vec<McpServer>,
+    ) -> Result<SessionNewResponse, AcpError> {
+        let params = serde_json::json!({
+            "cwd": cwd,
+            "sessionId": session_id,
+            "mcpServers": mcp_servers,
+        });
+        let result = self
+            .send_request_with_session_update_observer("session/load", params, false)
+            .await?;
+        // Spec-compliant agents may omit sessionId on load (it is implied).
+        // Prefer the request id so callers always have a concrete binding.
+        let resolved_id = result
+            .get("sessionId")
+            .and_then(|v| v.as_str())
+            .unwrap_or(session_id)
+            .to_owned();
+        tracing::info!(target: "acp::session", "session loaded: {resolved_id}");
+        Ok(SessionNewResponse {
+            session_id: resolved_id,
+            raw: result,
+        })
+    }
+
+    /// Returns true when an initialize result advertises `loadSession`.
+    pub fn agent_supports_load_session(init_result: &serde_json::Value) -> bool {
+        init_result
+            .get("agentCapabilities")
+            .and_then(|caps| caps.get("loadSession"))
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false)
+    }
+
     /// Send Goose's custom system-prompt request after `session/new`.
     pub async fn session_set_goose_system_prompt(
         &mut self,
@@ -1067,7 +1111,7 @@ impl AcpClient {
     /// Send a JSON-RPC request and wait for the matching response.
     ///
     /// Assigns the next available id, writes the NDJSON line to stdin,
-    /// then calls [`read_until_response`](Self::read_until_response).
+    /// then reads until the matching response arrives.
     ///
     /// The write phase is bounded by `WRITE_TIMEOUT` (30s) and the read phase
     /// by `REQUEST_TIMEOUT` (60s), so worst-case wall clock is ~90s. Non-prompt
@@ -1077,6 +1121,16 @@ impl AcpClient {
         &mut self,
         method: &str,
         params: serde_json::Value,
+    ) -> Result<serde_json::Value, AcpError> {
+        self.send_request_with_session_update_observer(method, params, true)
+            .await
+    }
+
+    async fn send_request_with_session_update_observer(
+        &mut self,
+        method: &str,
+        params: serde_json::Value,
+        observe_session_updates: bool,
     ) -> Result<serde_json::Value, AcpError> {
         let id = self.next_id;
         self.next_id += 1;
@@ -1099,7 +1153,12 @@ impl AcpClient {
             Err(_) => return Err(AcpError::Timeout(timeout)),
         }
 
-        match tokio::time::timeout(timeout, self.read_until_response(id)).await {
+        match tokio::time::timeout(
+            timeout,
+            self.read_until_response_with_session_update_observer(id, observe_session_updates),
+        )
+        .await
+        {
             Ok(result) => result,
             Err(_) => Err(AcpError::Timeout(timeout)),
         }
@@ -1109,7 +1168,7 @@ impl AcpClient {
     ///
     /// After a [`AcpError::Timeout`] from [`send_request`], the agent may
     /// eventually send the late response. That stale message will sit in the
-    /// `BufReader` buffer and be silently skipped by the next `read_until_response`
+    /// `BufReader` buffer and be silently skipped by the next response-read
     /// call (ID mismatch). However, if the caller wants a clean slate — e.g.
     /// before retrying the same method — they can call this to consume any
     /// buffered data with a short deadline.
@@ -1169,9 +1228,10 @@ impl AcpClient {
     ///
     /// Compares the incoming `id` field as a `serde_json::Value` against
     /// `json!(expected_id)` so that both numeric and string IDs work correctly.
-    async fn read_until_response(
+    async fn read_until_response_with_session_update_observer(
         &mut self,
         expected_id: u64,
+        observe_session_updates: bool,
     ) -> Result<serde_json::Value, AcpError> {
         loop {
             // LinesCodec::new_with_max_length enforces MAX_LINE_SIZE at the
@@ -1215,7 +1275,11 @@ impl AcpClient {
                     continue;
                 }
             };
-            self.observe("acp_read", msg.clone());
+            let is_session_update =
+                msg.get("method").and_then(|v| v.as_str()) == Some("session/update");
+            if observe_session_updates || !is_session_update {
+                self.observe("acp_read", msg.clone());
+            }
 
             // Check if this is a response to our expected request (has matching id
             // AND no `method` field — a `method` field means it's an agent-initiated
@@ -1262,7 +1326,7 @@ impl AcpClient {
         }
     }
 
-    /// Idle-aware message loop: like [`read_until_response`] but resets an idle
+    /// Idle-aware message loop: like the regular response-read path but resets an idle
     /// deadline on every stdout line. Fires [`AcpError::IdleTimeout`] on silence
     /// or [`AcpError::HardTimeout`] on absolute wall-clock cap.
     ///
@@ -3253,6 +3317,57 @@ mod tests {
         assert_eq!(result.unwrap()["worked"], serde_json::json!(true));
     }
 
+    #[tokio::test]
+    async fn session_load_suppresses_replayed_updates_from_observer_only() {
+        let script = r#"
+            read -t 2 _load
+            echo '{"jsonrpc":"2.0","method":"session/update","params":{"marker":"replayed"}}'
+            echo '{"jsonrpc":"2.0","id":0,"result":{}}'
+            read -t 2 _next
+            echo '{"jsonrpc":"2.0","method":"session/update","params":{"marker":"live"}}'
+            echo '{"jsonrpc":"2.0","id":1,"result":{"worked":true}}'
+            sleep 1
+        "#;
+        let mut client = spawn_script(script).await;
+        let observer = crate::observer::ObserverHandle::in_process();
+        client.set_observer(Some(observer.clone()), 0);
+
+        let loaded = client
+            .session_load_full("/", "sess-existing", Vec::new())
+            .await
+            .expect("session/load should succeed");
+        assert_eq!(loaded.session_id, "sess-existing");
+
+        let next = client
+            .send_request("test/echo", serde_json::json!({}))
+            .await
+            .expect("follow-up request should succeed");
+        assert_eq!(next["worked"], serde_json::json!(true));
+
+        let observed_reads: Vec<_> = observer
+            .snapshot()
+            .into_iter()
+            .filter(|event| event.kind == "acp_read")
+            .map(|event| event.payload)
+            .collect();
+        assert!(
+            !observed_reads
+                .iter()
+                .any(|payload| payload["params"]["marker"] == "replayed"),
+            "session/load replay updates must not enter the observer feed"
+        );
+        assert!(
+            observed_reads
+                .iter()
+                .any(|payload| payload["params"]["marker"] == "live"),
+            "normal session updates must remain observable after load"
+        );
+        assert!(
+            observed_reads.iter().any(|payload| payload["id"] == 0),
+            "the session/load response itself must remain observable"
+        );
+    }
+
     #[tokio::test]
     async fn keepalive_resets_idle_past_deadline() {
         // Keepalive session/update lines every 50ms against a 100ms idle deadline.

crates/buzz-acp/src/lib.rs

@@ -9,6 +9,7 @@ mod pool;
 mod pool_lifecycle;
 mod queue;
 mod relay;
+mod session_store;
 mod setup_mode;
 mod usage;
 
@@ -1140,8 +1141,9 @@ fn any_respawn_in_flight(crash_history: &[SlotCircuit]) -> bool {
 /// Result of a background respawn task.
 struct RespawnResult {
     index: usize,
-    /// Tuple: (initialized client, protocol version, agent name).
-    result: Result<(AcpClient, u32, String)>,
+    /// Tuple: (initialized client, protocol version, agent name,
+    /// supports session/load).
+    result: Result<(AcpClient, u32, String, bool)>,
 }
 
 /// Outcome of a non-cancelling steer attempt, forwarded from a per-attempt
@@ -1185,7 +1187,7 @@ impl RespawnGuard {
     /// Send the result and disarm the guard. Uses `try_send` (sync) so there
     /// is no await boundary between marking `sent` and actually enqueueing —
     /// cancellation cannot slip between the two.
-    fn send(mut self, result: Result<(AcpClient, u32, String)>) {
+    fn send(mut self, result: Result<(AcpClient, u32, String, bool)>) {
         // Invariant: try_send succeeds because the channel capacity equals the
         // slot count, and respawn_in_flight guarantees at most one outstanding
         // result per slot. If this ever fails, the channel sizing or the
@@ -1612,6 +1614,14 @@ async fn tokio_main() -> Result<()> {
         memory_enabled: config.memory_enabled,
         harness_name: crate::config::normalize_agent_command_identity(&config.agent_command),
         relay_url: config.relay_url.clone(),
+        agent_command: config.agent_command.clone(),
+        agent_args: config.agent_args.clone(),
+        session_store: std::sync::Arc::new(crate::session_store::SessionStore::open(
+            crate::session_store::SessionStore::default_path(
+                &config.agent_command,
+                &config.agent_args,
+            ),
+        )),
     });
 
     if !config.memory_enabled {
@@ -1854,7 +1864,7 @@ async fn tokio_main() -> Result<()> {
         while let Ok(rr) = respawn_rx.try_recv() {
             crash_history[rr.index].respawn_in_flight = false;
             match rr.result {
-                Ok((acp, protocol_version, agent_name)) => {
+                Ok((acp, protocol_version, agent_name, supports_load_session)) => {
                     let agent = OwnedAgent {
                         index: rr.index,
                         acp,
@@ -1865,6 +1875,7 @@ async fn tokio_main() -> Result<()> {
                         agent_name,
                         goose_system_prompt_supported: None,
                         protocol_version,
+                        supports_load_session,
                     };
                     pool.return_agent(agent);
                     tracing::info!(agent = rr.index, "respawn complete");
@@ -2179,6 +2190,11 @@ async fn tokio_main() -> Result<()> {
                             if is_rotate {
                                 if let Some(owner) = owner_cache.get() {
                                     if buzz_event.event.pubkey.to_hex() == *owner {
+                                        let durable_cleared =
+                                            pool::clear_durable_channel_binding(
+                                                &ctx,
+                                                &buzz_event.channel_id,
+                                            );
                                         let fired = signal_in_flight_task(
                                             &mut pool,
                                             buzz_event.channel_id,
@@ -2187,13 +2203,15 @@ async fn tokio_main() -> Result<()> {
                                         if fired {
                                             tracing::info!(
                                                 channel_id = %buzz_event.channel_id,
+                                                durable_cleared,
                                                 "!rotate received — cancelling in-flight turn and rotating session"
                                             );
                                         } else {
                                             let invalidated = pool.invalidate_channel_sessions(buzz_event.channel_id);
                                             tracing::info!(
                                                 channel_id = %buzz_event.channel_id,
                                                 invalidated,
+                                                durable_cleared,
                                                 "!rotate received — invalidated idle channel session(s)"
                                             );
                                         }
@@ -2785,7 +2803,7 @@ async fn tokio_main() -> Result<()> {
     // Drain any respawn results that completed before the abort. Explicitly
     // shut down returned agents instead of relying on AcpClient::Drop.
     while let Ok(rr) = respawn_rx.try_recv() {
-        if let Ok((mut acp, _, _)) = rr.result {
+        if let Ok((mut acp, _, _, _)) = rr.result {
             acp.shutdown().await;
             tracing::debug!(agent = rr.index, "reaped respawned agent on shutdown");
         }
@@ -3931,6 +3949,8 @@ async fn initialize_agent_pool(
                             }),
                         );
                         let agent_name = normalized_agent_name(&init_result);
+                        let supports_load_session =
+                            AcpClient::agent_supports_load_session(&init_result);
                         agent_slots.push(Some(OwnedAgent {
                             index: i,
                             acp,
@@ -3941,6 +3961,7 @@ async fn initialize_agent_pool(
                             agent_name,
                             goose_system_prompt_supported: None,
                             protocol_version,
+                            supports_load_session,
                         }));
                     }
                     Ok(Err(e)) => {
@@ -3991,7 +4012,7 @@ async fn spawn_and_init(
     has_generated_codex_config: bool,
     agent_index: usize,
     observer: Option<observer::ObserverHandle>,
-) -> Result<(AcpClient, u32, String)> {
+) -> Result<(AcpClient, u32, String, bool)> {
     let mut acp = AcpClient::spawn(command, args, extra_env, has_generated_codex_config)
         .await
         .map_err(|e| anyhow::anyhow!("failed to spawn agent: {e}"))?;
@@ -4001,6 +4022,7 @@ async fn spawn_and_init(
         Ok(init_result) => {
             tracing::info!("agent initialized: {init_result}");
             let protocol_version = init_result["protocolVersion"].as_u64().unwrap_or(1) as u32;
+            let supports_load_session = AcpClient::agent_supports_load_session(&init_result);
             acp.observe(
                 "agent_initialized",
                 serde_json::json!({
@@ -4009,7 +4031,7 @@ async fn spawn_and_init(
                 }),
             );
             let agent_name = normalized_agent_name(&init_result);
-            Ok((acp, protocol_version, agent_name))
+            Ok((acp, protocol_version, agent_name, supports_load_session))
         }
         Err(e) => {
             // Explicitly shut down the spawned child to prevent zombie/leak.
@@ -5396,6 +5418,7 @@ mod error_outcome_emission_tests {
             // Error branches under test never read this; 1 is the legacy
             // non-systemPrompt path, the simplest valid value.
             protocol_version: 1,
+            supports_load_session: false,
         }
     }