diff --git a/Cargo.lock b/Cargo.lock
index 937ead564a..7a2d3536c1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -807,7 +807,9 @@ dependencies = [
  "buzz-sdk",
  "chrono",
  "clap",
+ "dirs",
  "evalexpr",
+ "fs2",
  "futures-util",
  "hex",
  "httparse",
@@ -818,6 +820,7 @@ dependencies = [
  "serde",
  "serde_json",
  "sha2 0.11.0",
+ "tempfile",
  "thiserror 2.0.18",
  "tokio",
  "tokio-tungstenite 0.29.0",
@@ -2986,6 +2989,16 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "fs2"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9564fc758e15025b46aa6643b1b77d047d1a56a1aea6e01002ac0c7026876213"
+dependencies = [
+ "libc",
+ "winapi",
+]
+
 [[package]]
 name = "fs_extra"
 version = "1.3.0"
diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml
index d047849806..d573f3e522 100644
--- a/crates/buzz-acp/Cargo.toml
+++ b/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"
diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs
index 93109fa94d..46e683139f 100644
--- a/crates/buzz-acp/src/acp.rs
+++ b/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.
diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs
index 0c4e5f158c..b4d3bd2aa1 100644
--- a/crates/buzz-acp/src/lib.rs
+++ b/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,6 +2203,7 @@ 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 {
@@ -2194,6 +2211,7 @@ async fn tokio_main() -> Result<()> {
                                             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,
         }
     }
 
diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs
index 8430307d9c..0975fe84a1 100644
--- a/crates/buzz-acp/src/pool.rs
+++ b/crates/buzz-acp/src/pool.rs
@@ -80,6 +80,54 @@ pub struct AgentModelCapabilities {
     pub available_models_raw: Option<serde_json::Value>,
 }
 
+fn model_capabilities_from_response(
+    response: &serde_json::Value,
+) -> Option<AgentModelCapabilities> {
+    let config_options_raw = extract_model_config_options(response);
+    let available_models_raw = response
+        .get("models")
+        .filter(|models| models.is_object())
+        .cloned();
+    if config_options_raw.is_empty() && available_models_raw.is_none() {
+        None
+    } else {
+        Some(AgentModelCapabilities {
+            config_options_raw,
+            available_models_raw,
+        })
+    }
+}
+
+#[derive(Debug, PartialEq)]
+enum LoadModelResolution {
+    Method(ModelSwitchMethod),
+    Unsupported,
+    Unverifiable,
+}
+
+fn resolve_load_model_switch(
+    load_response: &serde_json::Value,
+    cached: Option<&AgentModelCapabilities>,
+    desired_model: &str,
+) -> LoadModelResolution {
+    if model_capabilities_from_response(load_response).is_some() {
+        return resolve_model_switch_method(load_response, desired_model)
+            .map(LoadModelResolution::Method)
+            .unwrap_or(LoadModelResolution::Unsupported);
+    }
+
+    let Some(cached) = cached else {
+        return LoadModelResolution::Unverifiable;
+    };
+    let cached_response = serde_json::json!({
+        "configOptions": cached.config_options_raw,
+        "models": cached.available_models_raw,
+    });
+    resolve_model_switch_method(&cached_response, desired_model)
+        .map(LoadModelResolution::Method)
+        .unwrap_or(LoadModelResolution::Unsupported)
+}
+
 /// Per-channel session IDs and turn counters.
 ///
 /// Separated from `OwnedAgent` so the state machine is testable without
@@ -170,6 +218,8 @@ pub struct OwnedAgent {
     pub goose_system_prompt_supported: Option<bool>,
     /// Protocol version reported by the agent in its initialize response.
     pub protocol_version: u32,
+    /// Whether the agent advertised `agentCapabilities.loadSession` at init.
+    pub supports_load_session: bool,
 }
 
 /// Package name reported by `claude-agent-acp` in its `initialize` response.
@@ -269,6 +319,25 @@ fn apply_completed_before_control_signal(
     }
 }
 
+fn control_signal_discards_session(control_signal: &ControlSignal) -> bool {
+    matches!(
+        control_signal,
+        ControlSignal::Rotate | ControlSignal::SwitchModel(_)
+    )
+}
+
+pub(crate) fn clear_durable_channel_binding(ctx: &PromptContext, channel_id: &Uuid) -> bool {
+    ctx.session_store
+        .remove(&ctx.agent_command, &ctx.agent_args, channel_id)
+}
+
+fn clear_durable_source_binding(ctx: &PromptContext, source: &PromptSource) -> bool {
+    match source {
+        PromptSource::Channel(channel_id) => clear_durable_channel_binding(ctx, channel_id),
+        PromptSource::Heartbeat => false,
+    }
+}
+
 /// Control signal for an in-flight channel turn.
 ///
 /// Not `Copy`: `SwitchModel` carries an owned `String`. Callers must clone when
@@ -564,6 +633,12 @@ pub struct PromptContext {
     /// the desktop keys per (agent, relay) pair, e.g. `session_config_captured`,
     /// mirroring the `managed_agent_runtime_lifecycle` frames.
     pub relay_url: String,
+    /// Agent binary as configured (for durable session binding identity).
+    pub agent_command: String,
+    /// Agent args as configured (for durable session binding identity).
+    pub agent_args: Vec<String>,
+    /// Durable channel→session bindings surviving harness restarts.
+    pub session_store: std::sync::Arc<crate::session_store::SessionStore>,
 }
 
 impl AgentPool {
@@ -876,6 +951,129 @@ async fn resolve_new_session_channel_context(
     (is_dm, title_channel, Some(info.channel_type))
 }
 
+/// Try to restore a durable channel session via `session/load`.
+///
+/// Returns `Some(session_id)` on success. On miss, capability absence, or load
+/// failure, clears the stale binding (when present) and returns `None` so the
+/// caller can fall through to `session/new`.
+async fn try_load_persisted_session(
+    agent: &mut OwnedAgent,
+    ctx: &PromptContext,
+    channel_id: &Uuid,
+    _agent_core: Option<&str>,
+    _agent_canvas: Option<&str>,
+) -> Option<String> {
+    if !agent.supports_load_session {
+        return None;
+    }
+    let stored = ctx
+        .session_store
+        .get(&ctx.agent_command, &ctx.agent_args, channel_id)?;
+    match agent
+        .acp
+        .session_load_full(&ctx.cwd, &stored, ctx.mcp_servers.clone())
+        .await
+    {
+        Ok(resp) => {
+            if agent.model_capabilities.is_none() {
+                agent.model_capabilities = model_capabilities_from_response(&resp.raw);
+            }
+            // Re-apply desired model after load when present.
+            if let Some(ref desired) = agent.desired_model {
+                match resolve_load_model_switch(
+                    &resp.raw,
+                    agent.model_capabilities.as_ref(),
+                    desired,
+                ) {
+                    LoadModelResolution::Method(method) => {
+                        if let Err(e) =
+                            apply_model_switch(&mut agent.acp, &resp.session_id, desired, &method)
+                                .await
+                        {
+                            tracing::warn!(
+                                target: "pool::session",
+                                error = %e,
+                                "model re-apply after session/load failed — continuing with loaded session"
+                            );
+                        }
+                    }
+                    LoadModelResolution::Unsupported => {
+                        tracing::warn!(
+                            target: "pool::model",
+                            "desired model {desired} is absent from the advertised model catalog after session/load"
+                        );
+                    }
+                    LoadModelResolution::Unverifiable => {
+                        tracing::debug!(
+                            target: "pool::model",
+                            "session/load returned no model catalog and none is cached — leaving desired model unverifiable"
+                        );
+                    }
+                }
+            }
+            if !ctx.permission_mode.is_default()
+                && agent_supports_mode(&resp.raw, ctx.permission_mode.as_wire_str())
+            {
+                if let Err(e) =
+                    apply_permission_mode(&mut agent.acp, &resp.session_id, &ctx.permission_mode)
+                        .await
+                {
+                    tracing::warn!(
+                        target: "pool::session",
+                        error = %e,
+                        "permission mode after session/load failed — continuing"
+                    );
+                }
+            }
+            Some(resp.session_id)
+        }
+        Err(e) if load_failure_is_definitive(&e) => {
+            tracing::warn!(
+                target: "pool::session",
+                session_id = %stored,
+                channel_id = %channel_id,
+                error = %e,
+                "session/load rejected by agent — clearing stale binding (if unchanged) and creating a new session"
+            );
+            // Only drop the binding we failed to load. A concurrent process may
+            // already have written a newer session for this channel.
+            let _ = ctx.session_store.remove_if_equals(
+                &ctx.agent_command,
+                &ctx.agent_args,
+                channel_id,
+                &stored,
+            );
+            None
+        }
+        Err(e) => {
+            tracing::warn!(
+                target: "pool::session",
+                session_id = %stored,
+                channel_id = %channel_id,
+                error = %e,
+                "session/load outcome indeterminate — keeping binding and creating a new session; \
+                 the stored session may still be live on the provider"
+            );
+            None
+        }
+    }
+}
+
+/// Whether a failed `session/load` proves the stored binding is dead.
+///
+/// Only a JSON-RPC error response is definitive: the provider answered and
+/// refused, so the session is genuinely gone and the binding is safe to drop.
+///
+/// Everything else is indeterminate. A timeout, transport failure or malformed
+/// response does NOT prove the provider failed to load — it may hold the session
+/// open. Dropping the binding on those and falling through to `session/new`
+/// would fork hidden provider state: two live sessions, one unreachable. Keeping
+/// the mapping is self-healing, because a provider that has genuinely lost the
+/// session answers `AgentError` on a later attempt and that clears it then.
+fn load_failure_is_definitive(error: &AcpError) -> bool {
+    matches!(error, AcpError::AgentError { .. })
+}
+
 /// Create a new ACP session via `session_new_full()`, populate model capabilities
 /// on the agent (first session only), and apply `desired_model` if set.
 ///
@@ -1597,6 +1795,24 @@ pub async fn run_prompt_task(
         PromptSource::Channel(cid) => {
             if let Some(sid) = agent.state.sessions.get(cid) {
                 (sid.clone(), false)
+            } else if let Some(sid) = try_load_persisted_session(
+                &mut agent,
+                &ctx,
+                cid,
+                agent_core.as_deref(),
+                agent_canvas.as_deref(),
+            )
+            .await
+            {
+                tracing::info!(
+                    target: "pool::session",
+                    "loaded session {sid} for channel {cid}"
+                );
+                agent.state.sessions.insert(*cid, sid.clone());
+                if let Some((pending_cid, section)) = pending_canvas.take() {
+                    agent.state.canvas_sections.insert(pending_cid, section);
+                }
+                (sid, false)
             } else {
                 // The title is channel-qualified (`Agent · #channel`) so one
                 // agent in several channels doesn't produce identical session
@@ -1619,6 +1835,8 @@ pub async fn run_prompt_task(
                             "created session {sid} for channel {cid}"
                         );
                         agent.state.sessions.insert(*cid, sid.clone());
+                        ctx.session_store
+                            .put(&ctx.agent_command, &ctx.agent_args, cid, &sid);
                         // Commit canvas only after session creation succeeds (I3).
                         if let Some((pending_cid, section)) = pending_canvas.take() {
                             agent.state.canvas_sections.insert(pending_cid, section);
@@ -1996,6 +2214,8 @@ pub async fn run_prompt_task(
                 ) => result,
                 mode = rx => {
                     let control_signal = mode.unwrap_or(ControlSignal::Cancel);
+                    let discard_durable_session =
+                        control_signal_discards_session(&control_signal);
                     // Land the model switch before any cancel/requeue work: setting
                     // `desired_model` here means the fresh session created by the
                     // requeued turn (busy) or the next turn (already-completed)
@@ -2016,6 +2236,9 @@ pub async fn run_prompt_task(
                             Ok(stop_reason) => {
                                 log_stop_reason(&source, &stop_reason);
                                 agent.state.invalidate(&source);
+                                if discard_durable_session {
+                                    clear_durable_source_binding(&ctx, &source);
+                                }
                                 let retry_batch =
                                     requeue_cancelled_batch(&ctx, control_signal, batch);
 
@@ -2054,6 +2277,9 @@ pub async fn run_prompt_task(
                                 } else {
                                     agent.state.invalidate(&source);
                                 }
+                                if discard_durable_session {
+                                    clear_durable_source_binding(&ctx, &source);
+                                }
 
                                 let usage = agent.acp.take_turn_usage();
                                 publish_agent_turn_metric(
@@ -2110,6 +2336,9 @@ pub async fn run_prompt_task(
                             &source,
                             &control_signal,
                         );
+                        if discard_durable_session {
+                            clear_durable_source_binding(&ctx, &source);
+                        }
                         let usage = agent.acp.take_turn_usage();
                         publish_agent_turn_metric(
                             &ctx,
@@ -2169,6 +2398,7 @@ pub async fn run_prompt_task(
                     "rotating session for {source:?} after {stop_reason:?}",
                 );
                 agent.state.invalidate(&source);
+                clear_durable_source_binding(&ctx, &source);
             }
 
             let core_stop = acp_stop_to_core(&stop_reason);
@@ -4026,6 +4256,38 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec<String>)
 
 #[cfg(test)]
 mod tests {
+
+    /// A `session/load` failure only clears the durable binding when the
+    /// provider actually answered and refused. Timeouts, transport failures and
+    /// malformed responses are indeterminate: the provider may hold the session
+    /// open, and clearing the binding there would fork hidden state into two
+    /// live sessions with one unreachable.
+    #[test]
+    fn only_an_agent_error_is_a_definitive_session_load_failure() {
+        use std::time::Duration;
+
+        assert!(super::load_failure_is_definitive(&AcpError::AgentError {
+            code: -32602,
+            message: "no such session".into(),
+        }));
+
+        for indeterminate in [
+            AcpError::Timeout(Duration::from_secs(1)),
+            AcpError::IdleTimeout(Duration::from_secs(1)),
+            AcpError::WriteTimeout(Duration::from_secs(1)),
+            AcpError::CancelDrainTimeout(Duration::from_secs(1)),
+            AcpError::HardTimeout {
+                silence: Duration::from_secs(1),
+            },
+            AcpError::AgentExited,
+            AcpError::Protocol("truncated frame".into()),
+        ] {
+            assert!(
+                !super::load_failure_is_definitive(&indeterminate),
+                "{indeterminate:?} must not clear the binding"
+            );
+        }
+    }
     use super::*;
     use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
     use serde_json::json;
@@ -5279,6 +5541,62 @@ mod tests {
         (s, ch_a, ch_b)
     }
 
+    #[test]
+    fn load_without_catalog_does_not_poison_model_capabilities() {
+        let response = json!({"sessionId": "loaded"});
+        assert!(model_capabilities_from_response(&response).is_none());
+    }
+
+    #[test]
+    fn load_uses_cached_catalog_when_response_omits_it() {
+        let cached_response = json!({
+            "configOptions": [{
+                "configId": "model",
+                "category": "model",
+                "options": [{"value": "model-a"}]
+            }]
+        });
+        let cached = model_capabilities_from_response(&cached_response)
+            .expect("model catalog should be captured");
+
+        assert_eq!(
+            resolve_load_model_switch(&json!({}), Some(&cached), "model-a"),
+            LoadModelResolution::Method(ModelSwitchMethod::ConfigOption {
+                config_id: "model".into(),
+                option_value: "model-a".into(),
+            })
+        );
+    }
+
+    #[test]
+    fn advertised_load_catalog_is_authoritative_even_when_empty() {
+        let cached_response = json!({
+            "models": {
+                "availableModels": [{"modelId": "model-a"}]
+            }
+        });
+        let cached = model_capabilities_from_response(&cached_response)
+            .expect("model catalog should be captured");
+        let load_response = json!({
+            "models": {
+                "availableModels": []
+            }
+        });
+
+        assert_eq!(
+            resolve_load_model_switch(&load_response, Some(&cached), "model-a"),
+            LoadModelResolution::Unsupported
+        );
+    }
+
+    #[test]
+    fn load_without_any_catalog_is_unverifiable() {
+        assert_eq!(
+            resolve_load_model_switch(&json!({}), None, "model-a"),
+            LoadModelResolution::Unverifiable
+        );
+    }
+
     #[test]
     fn test_rotate_after_natural_completion_invalidates_channel_state() {
         let (mut s, ch_a, ch_b) = make_state();
@@ -6018,6 +6336,7 @@ mod tests {
             agent_name: "unknown".into(),
             goose_system_prompt_supported: None,
             protocol_version: 2,
+            supports_load_session: false,
         };
 
         // Simulate dispatch: install a steer receiver (normally done by
@@ -6076,6 +6395,7 @@ mod tests {
             agent_name: "unknown".into(),
             goose_system_prompt_supported: None,
             protocol_version: 2,
+            supports_load_session: false,
         };
 
         // Simulate a completed turn: `steer_rx` was consumed by the read loop
@@ -6542,9 +6862,38 @@ mod tests {
             memory_enabled: false,
             harness_name: "goose".to_string(),
             relay_url: "ws://127.0.0.1:3000".to_string(),
+            agent_command: "goose".to_string(),
+            agent_args: vec!["acp".to_string()],
+            session_store: std::sync::Arc::new(crate::session_store::SessionStore::open(
+                std::env::temp_dir().join(format!(
+                    "buzz-acp-test-sessions-{}.json",
+                    uuid::Uuid::new_v4()
+                )),
+            )),
         }
     }
 
+    #[test]
+    fn explicit_rotation_clears_the_durable_channel_binding() {
+        let ctx = make_prompt_context_no_owner();
+        let channel_id = Uuid::new_v4();
+        ctx.session_store.put(
+            &ctx.agent_command,
+            &ctx.agent_args,
+            &channel_id,
+            "session-before-rotate",
+        );
+
+        assert!(clear_durable_source_binding(
+            &ctx,
+            &PromptSource::Channel(channel_id)
+        ));
+        assert!(ctx
+            .session_store
+            .get(&ctx.agent_command, &ctx.agent_args, &channel_id)
+            .is_none());
+    }
+
     // ── render_canvas_section ────────────────────────────────────────────────
 
     #[test]
diff --git a/crates/buzz-acp/src/session_store.rs b/crates/buzz-acp/src/session_store.rs
new file mode 100644
index 0000000000..d663524fdc
--- /dev/null
+++ b/crates/buzz-acp/src/session_store.rs
@@ -0,0 +1,487 @@
+//! Durable channel → ACP session bindings for harness restarts.
+//!
+//! `SessionState` is in-memory only. Agents that advertise `loadSession` (e.g.
+//! Hermes) can restore a prior ACP conversation after the harness respawns if
+//! the channel→session mapping survives. This module persists that mapping as
+//! a small JSON sidecar under the process data directory.
+//!
+//! Keyed by `(agent_command_identity, agent_args, channel_id)` so different
+//! agent binaries / profiles do not share bindings. Heartbeats are never
+//! stored — they stay ephemeral.
+//!
+//! Cross-process safety: the store is a shared file. Every read and mutation
+//! takes a sibling lockfile, reloads the on-disk map under that lock, then
+//! writes atomically. A process-local cache alone is unsafe when two
+//! `buzz-acp` processes share the same agent command/args identity.
+
+use std::collections::HashMap;
+use std::ffi::OsString;
+use std::fs::{self, File, OpenOptions};
+use std::path::{Path, PathBuf};
+
+use fs2::FileExt;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::config::normalize_agent_command_identity;
+
+/// Environment override for the session store path (tests / operators).
+pub const SESSION_STORE_ENV: &str = "BUZZ_ACP_SESSION_STORE";
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)]
+struct StoreFile {
+    /// version for future migrations
+    version: u32,
+    /// map key → ACP session id
+    sessions: HashMap<String, String>,
+}
+
+/// Durable session binding store shared across buzz-acp processes.
+pub struct SessionStore {
+    path: PathBuf,
+    lock_path: PathBuf,
+}
+
+/// RAII wrapper that unlocks the OS file lock on drop.
+struct StoreLock {
+    file: File,
+}
+
+impl Drop for StoreLock {
+    fn drop(&mut self) {
+        let _ = FileExt::unlock(&self.file);
+    }
+}
+
+impl SessionStore {
+    /// Open or create the store at the resolved path.
+    ///
+    /// Does not cache file contents; each operation reloads under lock.
+    pub fn open(path: PathBuf) -> Self {
+        let lock_path = sibling_lock_path(&path);
+        Self { path, lock_path }
+    }
+
+    /// Resolve the default store path for this agent identity.
+    pub fn default_path(agent_command: &str, agent_args: &[String]) -> PathBuf {
+        if let Ok(override_path) = std::env::var(SESSION_STORE_ENV) {
+            if !override_path.trim().is_empty() {
+                return PathBuf::from(override_path);
+            }
+        }
+        let identity = store_identity(agent_command, agent_args);
+        let base = dirs::data_local_dir()
+            .or_else(dirs::data_dir)
+            .unwrap_or_else(|| PathBuf::from("."));
+        base.join("buzz-acp")
+            .join("sessions")
+            .join(format!("{identity}.json"))
+    }
+
+    /// Look up a stored ACP session id for a channel.
+    pub fn get(
+        &self,
+        agent_command: &str,
+        agent_args: &[String],
+        channel_id: &Uuid,
+    ) -> Option<String> {
+        let key = binding_key(agent_command, agent_args, channel_id);
+        let _lock = self.acquire_lock(false)?;
+        match load_store(&self.path) {
+            Ok(data) => data.sessions.get(&key).cloned(),
+            Err(e) => {
+                self.warn_io("failed to read ACP session bindings", &e);
+                None
+            }
+        }
+    }
+
+    /// Persist a channel → session binding.
+    pub fn put(
+        &self,
+        agent_command: &str,
+        agent_args: &[String],
+        channel_id: &Uuid,
+        session_id: &str,
+    ) {
+        let key = binding_key(agent_command, agent_args, channel_id);
+        let Some(_lock) = self.acquire_lock(true) else {
+            return;
+        };
+        let mut data = match load_store(&self.path) {
+            Ok(data) => data,
+            Err(e) if e.kind() == std::io::ErrorKind::InvalidData => {
+                // Corrupt sidecar: log and recover empty rather than wedging puts forever.
+                self.warn_io(
+                    "corrupt ACP session store on update — rewriting from empty map",
+                    &e,
+                );
+                StoreFile::default()
+            }
+            Err(e) => {
+                self.warn_io("failed to read ACP session bindings before update", &e);
+                return;
+            }
+        };
+        data.version = 1;
+        data.sessions.insert(key, session_id.to_owned());
+        if let Err(e) = save_store(&self.path, &data) {
+            self.warn_io("failed to persist ACP session binding", &e);
+        }
+    }
+
+    /// Remove the current binding for a channel, regardless of session id.
+    ///
+    /// This is reserved for explicit discard semantics such as owner-requested
+    /// rotation. Failed loads must use [`Self::remove_if_equals`] so they cannot
+    /// delete a newer binding written by another process.
+    ///
+    /// Returns `true` when a binding was removed.
+    pub fn remove(&self, agent_command: &str, agent_args: &[String], channel_id: &Uuid) -> bool {
+        let key = binding_key(agent_command, agent_args, channel_id);
+        let Some(_lock) = self.acquire_lock(true) else {
+            return false;
+        };
+        match load_store(&self.path) {
+            Ok(mut data) => {
+                if data.sessions.remove(&key).is_none() {
+                    return false;
+                }
+                if let Err(e) = save_store(&self.path, &data) {
+                    self.warn_io("failed to persist ACP session binding removal", &e);
+                    return false;
+                }
+                true
+            }
+            Err(e) => {
+                self.warn_io("failed to read ACP session bindings before removal", &e);
+                false
+            }
+        }
+    }
+
+    /// Remove a binding only if it still points at `expected_session_id`.
+    ///
+    /// Used after a failed `session/load`: another process may have already
+    /// written a newer session for the same channel, and a key-only remove
+    /// would delete that fresher binding.
+    ///
+    /// Returns `true` when a matching binding was removed.
+    pub fn remove_if_equals(
+        &self,
+        agent_command: &str,
+        agent_args: &[String],
+        channel_id: &Uuid,
+        expected_session_id: &str,
+    ) -> bool {
+        let key = binding_key(agent_command, agent_args, channel_id);
+        let Some(_lock) = self.acquire_lock(true) else {
+            return false;
+        };
+        match load_store(&self.path) {
+            Ok(mut data) => {
+                let matches = data
+                    .sessions
+                    .get(&key)
+                    .is_some_and(|current| current == expected_session_id);
+                if !matches {
+                    return false;
+                }
+                data.sessions.remove(&key);
+                if let Err(e) = save_store(&self.path, &data) {
+                    self.warn_io(
+                        "failed to persist conditional ACP session binding removal",
+                        &e,
+                    );
+                    return false;
+                }
+                true
+            }
+            Err(e) => {
+                self.warn_io(
+                    "failed to read ACP session bindings before conditional removal",
+                    &e,
+                );
+                false
+            }
+        }
+    }
+
+    fn acquire_lock(&self, exclusive: bool) -> Option<StoreLock> {
+        if let Some(parent) = self.lock_path.parent() {
+            if let Err(e) = fs::create_dir_all(parent) {
+                self.warn_io("failed to create ACP session store directory", &e);
+                return None;
+            }
+        }
+        let file = match OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(&self.lock_path)
+        {
+            Ok(file) => file,
+            Err(e) => {
+                self.warn_io("failed to open ACP session store lock", &e);
+                return None;
+            }
+        };
+        let result = if exclusive {
+            FileExt::lock_exclusive(&file)
+        } else {
+            FileExt::lock_shared(&file)
+        };
+        if let Err(e) = result {
+            self.warn_io("failed to lock ACP session store", &e);
+            return None;
+        }
+        Some(StoreLock { file })
+    }
+
+    fn warn_io(&self, message: &'static str, error: &std::io::Error) {
+        tracing::warn!(
+            target: "session_store",
+            path = %self.path.display(),
+            lock_path = %self.lock_path.display(),
+            error = %error,
+            "{message}"
+        );
+    }
+}
+
+fn sibling_lock_path(path: &Path) -> PathBuf {
+    let mut name = path.as_os_str().to_owned();
+    name.push(OsString::from(".lock"));
+    PathBuf::from(name)
+}
+
+fn store_identity(agent_command: &str, agent_args: &[String]) -> String {
+    let cmd = normalize_agent_command_identity(agent_command);
+    let args = agent_args.join(" ");
+    let raw = if args.is_empty() {
+        cmd
+    } else {
+        format!("{cmd} {args}")
+    };
+    // Keep the filename filesystem-safe and short.
+    let mut out = String::with_capacity(raw.len());
+    for ch in raw.chars() {
+        if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
+            out.push(ch);
+        } else {
+            out.push('_');
+        }
+    }
+    if out.is_empty() {
+        "agent".into()
+    } else {
+        out
+    }
+}
+
+fn binding_key(agent_command: &str, agent_args: &[String], channel_id: &Uuid) -> String {
+    format!(
+        "{}|{}|{}",
+        normalize_agent_command_identity(agent_command),
+        agent_args.join("\u{1f}"),
+        channel_id
+    )
+}
+
+fn load_store(path: &Path) -> std::io::Result<StoreFile> {
+    match fs::read_to_string(path) {
+        Ok(text) => serde_json::from_str(&text)
+            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(StoreFile::default()),
+        Err(e) => Err(e),
+    }
+}
+
+fn save_store(path: &Path, data: &StoreFile) -> std::io::Result<()> {
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent)?;
+    }
+    let tmp = path.with_extension("json.tmp");
+    let json = serde_json::to_string_pretty(data)
+        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
+    fs::write(&tmp, json)?;
+    fs::rename(&tmp, path)?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::tempdir;
+
+    #[test]
+    fn round_trip_binding() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        let store = SessionStore::open(path);
+        let channel = Uuid::new_v4();
+        assert!(store.get("hermes", &["acp".into()], &channel).is_none());
+        store.put("hermes", &["acp".into()], &channel, "sess-1");
+        assert_eq!(
+            store.get("hermes", &["acp".into()], &channel).as_deref(),
+            Some("sess-1")
+        );
+        // Re-open from disk.
+        let store2 = SessionStore::open(store.path.clone());
+        assert_eq!(
+            store2.get("hermes", &["acp".into()], &channel).as_deref(),
+            Some("sess-1")
+        );
+        assert!(store2.remove_if_equals("hermes", &["acp".into()], &channel, "sess-1"));
+        assert!(store2.get("hermes", &["acp".into()], &channel).is_none());
+    }
+
+    #[test]
+    fn different_args_are_isolated() {
+        let dir = tempdir().unwrap();
+        let store = SessionStore::open(dir.path().join("s.json"));
+        let channel = Uuid::new_v4();
+        store.put("hermes", &["acp".into()], &channel, "a");
+        store.put(
+            "hermes",
+            &["-p".into(), "chad".into(), "acp".into()],
+            &channel,
+            "b",
+        );
+        assert_eq!(
+            store.get("hermes", &["acp".into()], &channel).as_deref(),
+            Some("a")
+        );
+        assert_eq!(
+            store
+                .get(
+                    "hermes",
+                    &["-p".into(), "chad".into(), "acp".into()],
+                    &channel
+                )
+                .as_deref(),
+            Some("b")
+        );
+    }
+
+    #[test]
+    fn independently_opened_stores_do_not_lose_updates() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        let store_a = SessionStore::open(path.clone());
+        let store_b = SessionStore::open(path.clone());
+        let channel_a = Uuid::new_v4();
+        let channel_b = Uuid::new_v4();
+        let channel_c = Uuid::new_v4();
+        let args = ["acp".into()];
+
+        store_a.put("hermes", &args, &channel_a, "session-a");
+        store_b.put("hermes", &args, &channel_b, "session-b");
+
+        let reopened = SessionStore::open(path.clone());
+        assert_eq!(
+            reopened.get("hermes", &args, &channel_a).as_deref(),
+            Some("session-a")
+        );
+        assert_eq!(
+            reopened.get("hermes", &args, &channel_b).as_deref(),
+            Some("session-b")
+        );
+
+        // Open both before either mutation. A stale process-local snapshot would
+        // resurrect channel A when the second store writes channel C.
+        let remover = SessionStore::open(path.clone());
+        let writer = SessionStore::open(path.clone());
+        assert!(remover.remove_if_equals("hermes", &args, &channel_a, "session-a"));
+        writer.put("hermes", &args, &channel_c, "session-c");
+
+        let final_store = SessionStore::open(path);
+        assert!(final_store.get("hermes", &args, &channel_a).is_none());
+        assert_eq!(
+            final_store.get("hermes", &args, &channel_b).as_deref(),
+            Some("session-b")
+        );
+        assert_eq!(
+            final_store.get("hermes", &args, &channel_c).as_deref(),
+            Some("session-c")
+        );
+    }
+
+    #[test]
+    fn put_recovers_from_corrupt_store() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        fs::write(&path, "{not-json").unwrap();
+        let store = SessionStore::open(path.clone());
+        let channel = Uuid::new_v4();
+        store.put("hermes", &["acp".into()], &channel, "recovered");
+        assert_eq!(
+            store.get("hermes", &["acp".into()], &channel).as_deref(),
+            Some("recovered")
+        );
+    }
+
+    #[test]
+    fn remove_if_equals_does_not_delete_newer_binding() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        let args = ["acp".into()];
+        let channel = Uuid::new_v4();
+
+        // Process A reads X.
+        let process_a = SessionStore::open(path.clone());
+        process_a.put("hermes", &args, &channel, "session-x");
+        let read_x = process_a
+            .get("hermes", &args, &channel)
+            .expect("process A read X");
+        assert_eq!(read_x, "session-x");
+
+        // Process B writes Y for the same channel.
+        let process_b = SessionStore::open(path.clone());
+        process_b.put("hermes", &args, &channel, "session-y");
+        assert_eq!(
+            process_b.get("hermes", &args, &channel).as_deref(),
+            Some("session-y")
+        );
+
+        // Process A's failed load of X must not delete Y.
+        let removed = process_a.remove_if_equals("hermes", &args, &channel, &read_x);
+        assert!(!removed);
+
+        let final_store = SessionStore::open(path);
+        assert_eq!(
+            final_store.get("hermes", &args, &channel).as_deref(),
+            Some("session-y")
+        );
+    }
+
+    #[test]
+    fn remove_if_equals_clears_matching_stale_binding() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        let args = ["acp".into()];
+        let channel = Uuid::new_v4();
+        let store = SessionStore::open(path.clone());
+        store.put("hermes", &args, &channel, "session-x");
+        assert!(store.remove_if_equals("hermes", &args, &channel, "session-x"));
+        assert!(store.get("hermes", &args, &channel).is_none());
+        // No-op when already gone.
+        assert!(!store.remove_if_equals("hermes", &args, &channel, "session-x"));
+    }
+
+    #[test]
+    fn remove_clears_whichever_binding_is_current() {
+        let dir = tempdir().unwrap();
+        let path = dir.path().join("sessions.json");
+        let args = ["acp".into()];
+        let channel = Uuid::new_v4();
+        let store = SessionStore::open(path);
+        store.put("hermes", &args, &channel, "session-x");
+        store.put("hermes", &args, &channel, "session-y");
+
+        assert!(store.remove("hermes", &args, &channel));
+        assert!(store.get("hermes", &args, &channel).is_none());
+        assert!(!store.remove("hermes", &args, &channel));
+    }
+}