diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs
index 9917880b4b5..14882a3dca1 100644
--- a/crates/buzz-acp/src/acp.rs
+++ b/crates/buzz-acp/src/acp.rs
@@ -646,10 +646,13 @@ impl AcpClient {
     /// - `Some(SystemPromptTransport::ClaudeMeta(text))` — `_meta.systemPrompt`
     ///   as `{"append": text}`, keeping claude-agent-acp's native preset intact.
     ///
-    /// `session_title` rides in `_meta.sessionTitle` when `Some`; `_meta` is
-    /// omitted entirely otherwise, since adapters may distinguish an absent
-    /// member from a null one. When both `ClaudeMeta` and `session_title` are
-    /// present the two `_meta` members are merged into a single object.
+    /// `session_title` rides in `_meta.sessionTitle` when `Some`.
+    /// `session_key` rides in `_meta.sessionKey` when `Some` — OpenClaw's ACP
+    /// mapper uses that as the Gateway store key (else the process `--session`
+    /// default). `_meta` is omitted entirely when neither is set, since adapters
+    /// may distinguish an absent member from a null one. When `ClaudeMeta`,
+    /// `session_title`, and/or `session_key` are present the `_meta` members
+    /// are merged into a single object.
     ///
     /// Callers use [`extract_model_config_options`] and [`extract_model_state`]
     /// to pull model info from the raw result.
@@ -659,6 +662,7 @@ impl AcpClient {
         mcp_servers: Vec<McpServer>,
         system_prompt: Option<SystemPromptTransport<'_>>,
         session_title: Option<&str>,
+        session_key: Option<&str>,
     ) -> Result<SessionNewResponse, AcpError> {
         let mut params = serde_json::json!({
             "cwd": cwd,
@@ -669,7 +673,7 @@ impl AcpClient {
                 params["systemPrompt"] = serde_json::Value::String(sp.to_owned());
             }
             Some(SystemPromptTransport::ClaudeMeta(sp)) => {
-                // Merge into _meta so sessionTitle (set below) is not clobbered.
+                // Merge into _meta so sessionTitle / sessionKey (set below) are not clobbered.
                 params["_meta"]["systemPrompt"] = serde_json::json!({ "append": sp });
             }
             None => {}
@@ -678,6 +682,9 @@ impl AcpClient {
             // Merge — _meta may already carry systemPrompt from ClaudeMeta above.
             params["_meta"]["sessionTitle"] = serde_json::Value::String(title.to_owned());
         }
+        if let Some(key) = session_key {
+            params["_meta"]["sessionKey"] = serde_json::Value::String(key.to_owned());
+        }
         let result = self.send_request("session/new", params).await?;
         let session_id = result["sessionId"]
             .as_str()
@@ -700,9 +707,10 @@ impl AcpClient {
         mcp_servers: Vec<McpServer>,
         system_prompt: Option<SystemPromptTransport<'_>>,
         session_title: Option<&str>,
+        session_key: Option<&str>,
     ) -> Result<String, AcpError> {
         Ok(self
-            .session_new_full(cwd, mcp_servers, system_prompt, session_title)
+            .session_new_full(cwd, mcp_servers, system_prompt, session_title, session_key)
             .await?
             .session_id)
     }
@@ -3492,6 +3500,7 @@ mod tests {
                 vec![],
                 Some(SystemPromptTransport::Field("Custom system prompt")),
                 None,
+                None,
             )
             .await
             .expect("session_new_full should succeed");
@@ -3577,7 +3586,7 @@ mod tests {
             .expect("initialize should succeed");
 
         let resp = client
-            .session_new_full("/tmp", vec![], None, None)
+            .session_new_full("/tmp", vec![], None, None, None)
             .await
             .expect("session_new_full should succeed");
 
@@ -3605,7 +3614,7 @@ mod tests {
             .expect("initialize should succeed");
 
         let resp = client
-            .session_new_full("/tmp", vec![], None, Some("Fizz · #buzz-dev"))
+            .session_new_full("/tmp", vec![], None, Some("Fizz · #buzz-dev"), None)
             .await
             .expect("session_new_full should succeed");
 
@@ -3633,7 +3642,7 @@ mod tests {
             .expect("initialize should succeed");
 
         let resp = client
-            .session_new_full("/tmp", vec![], None, None)
+            .session_new_full("/tmp", vec![], None, None, None)
             .await
             .expect("session_new_full should succeed");
 
@@ -3669,6 +3678,7 @@ mod tests {
                 vec![],
                 Some(SystemPromptTransport::ClaudeMeta("Be concise")),
                 None,
+                None,
             )
             .await
             .expect("session_new_full should succeed");
@@ -3708,6 +3718,7 @@ mod tests {
                 vec![],
                 Some(SystemPromptTransport::ClaudeMeta("Be concise")),
                 Some("Fizz · #buzz-dev"),
+                None,
             )
             .await
             .expect("session_new_full should succeed");
@@ -3725,6 +3736,44 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn session_new_full_sends_session_key_and_title_in_meta() {
+        let script = r#"
+            read -t 2 _init
+            echo '{"jsonrpc":"2.0","id":0,"result":{"protocolVersion":1,"agentCapabilities":{}}}'
+            read -t 2 REQ
+            echo '{"jsonrpc":"2.0","id":1,"result":{"sessionId":"ses_key","_receivedRequest":'"$REQ"'}}'
+            sleep 1
+        "#;
+        let mut client = spawn_script(script).await;
+        client
+            .initialize()
+            .await
+            .expect("initialize should succeed");
+
+        let resp = client
+            .session_new_full(
+                "/tmp",
+                vec![],
+                None,
+                Some("Captain · #general"),
+                Some("agent:captain:buzz:channel:11111111-1111-1111-1111-111111111111"),
+            )
+            .await
+            .expect("session_new_full should succeed");
+
+        let received = &resp.raw["_receivedRequest"];
+        assert_eq!(
+            received["params"]["_meta"]["sessionTitle"].as_str(),
+            Some("Captain · #general"),
+        );
+        assert_eq!(
+            received["params"]["_meta"]["sessionKey"].as_str(),
+            Some("agent:captain:buzz:channel:11111111-1111-1111-1111-111111111111"),
+            "OpenClaw Gateway key must ride in _meta.sessionKey"
+        );
+    }
+
     // ── Goose-native steer scaffold (PR follow-up to #1160) ──────────────
 
     /// Helper: spawn an inert `cat` subprocess so we have a real AcpClient
diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs
index 02ea491cd79..1d6c1561a2a 100644
--- a/crates/buzz-acp/src/lib.rs
+++ b/crates/buzz-acp/src/lib.rs
@@ -6,6 +6,7 @@ mod engram_fetch;
 mod filter;
 mod last_mile;
 mod observer;
+mod openclaw_session;
 mod pool;
 mod pool_lifecycle;
 mod queue;
@@ -2238,6 +2239,7 @@ 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(),
+        openclaw_agent_id: crate::openclaw_session::parse_openclaw_agent_id(&config.agent_args),
     });
 
     if !config.memory_enabled {
@@ -4047,7 +4049,12 @@ fn handle_prompt_result(
         // The task may have invalidated this session before returning. Never
         // resurrect delivery state for a dead session; its replacement must
         // receive fresh standing context and history.
-        if let Some(live_session_id) = result.agent.state.sessions.get(channel_id).cloned() {
+        let conversation = result
+            .batch
+            .as_ref()
+            .map(crate::openclaw_session::ConversationKey::from_batch)
+            .unwrap_or_else(|| crate::openclaw_session::ConversationKey::channel(*channel_id));
+        if let Some(live_session_id) = result.agent.state.sessions.get(&conversation).cloned() {
             let event_ids = successful_steer_deliveries
                 .into_iter()
                 .filter(|delivery| delivery.session_id == live_session_id)
@@ -4055,7 +4062,7 @@ fn handle_prompt_result(
             result
                 .agent
                 .state
-                .mark_channel_delivery_success(*channel_id, false, event_ids);
+                .mark_channel_delivery_success(conversation, false, event_ids);
         }
     }
 
@@ -5088,7 +5095,9 @@ async fn run_models(args: ModelsArgs) -> Result<()> {
     // so shutdown() runs on all paths (success, error, timeout).
     let protocol_result = tokio::time::timeout(MODELS_TIMEOUT, async {
         let init = client.initialize().await?;
-        let session = client.session_new_full(&cwd, vec![], None, None).await?;
+        let session = client
+            .session_new_full(&cwd, vec![], None, None, None)
+            .await?;
         Ok::<_, acp::AcpError>((init, session))
     })
     .await;
@@ -7524,14 +7533,14 @@ mod error_outcome_emission_tests {
         let channel_id = Uuid::new_v4();
         let steer_event_id = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
         let mut agent = dummy_agent(0).await;
-        agent
-            .state
-            .sessions
-            .insert(channel_id, "live-session".into());
-        agent
-            .state
-            .deliveries
-            .insert(channel_id, Default::default());
+        agent.state.sessions.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            "live-session".into(),
+        );
+        agent.state.deliveries.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            Default::default(),
+        );
 
         let mut pool = AgentPool::from_slots(vec![None]);
         let task_id = pool.join_set.spawn(async {}).id();
@@ -7587,7 +7596,8 @@ mod error_outcome_emission_tests {
         );
 
         let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
-        assert!(returned.state.deliveries[&channel_id]
+        assert!(returned.state.deliveries
+            [&crate::openclaw_session::ConversationKey::channel(channel_id)]
             .delivered_event_ids
             .contains(steer_event_id));
     }
@@ -7596,14 +7606,14 @@ mod error_outcome_emission_tests {
     async fn in_flight_stale_native_steer_ack_cannot_update_replacement_session() {
         let channel_id = Uuid::new_v4();
         let mut agent = dummy_agent(0).await;
-        agent
-            .state
-            .sessions
-            .insert(channel_id, "replacement-session".into());
-        agent
-            .state
-            .deliveries
-            .insert(channel_id, Default::default());
+        agent.state.sessions.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            "replacement-session".into(),
+        );
+        agent.state.deliveries.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            Default::default(),
+        );
 
         let mut pool = AgentPool::from_slots(vec![None]);
         let task_id = pool.join_set.spawn(async {}).id();
@@ -7659,7 +7669,8 @@ mod error_outcome_emission_tests {
         );
 
         let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
-        assert!(returned.state.deliveries[&channel_id]
+        assert!(returned.state.deliveries
+            [&crate::openclaw_session::ConversationKey::channel(channel_id)]
             .delivered_event_ids
             .is_empty());
     }
@@ -7669,14 +7680,14 @@ mod error_outcome_emission_tests {
         let channel_id = Uuid::new_v4();
         let steer_event_id = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
         let mut agent = dummy_agent(0).await;
-        agent
-            .state
-            .sessions
-            .insert(channel_id, "live-session".into());
-        agent
-            .state
-            .deliveries
-            .insert(channel_id, Default::default());
+        agent.state.sessions.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            "live-session".into(),
+        );
+        agent.state.deliveries.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            Default::default(),
+        );
         let mut pool = AgentPool::from_slots(vec![Some(agent)]);
 
         assert!(pool.record_successful_steer(
@@ -7685,7 +7696,8 @@ mod error_outcome_emission_tests {
             "live-session".into(),
         ));
         let returned = pool.agents_mut()[0].as_ref().expect("idle returned agent");
-        assert!(returned.state.deliveries[&channel_id]
+        assert!(returned.state.deliveries
+            [&crate::openclaw_session::ConversationKey::channel(channel_id)]
             .delivered_event_ids
             .contains(steer_event_id));
     }
@@ -7694,14 +7706,14 @@ mod error_outcome_emission_tests {
     async fn late_native_steer_ack_cannot_update_replacement_session() {
         let channel_id = Uuid::new_v4();
         let mut agent = dummy_agent(0).await;
-        agent
-            .state
-            .sessions
-            .insert(channel_id, "replacement-session".into());
-        agent
-            .state
-            .deliveries
-            .insert(channel_id, Default::default());
+        agent.state.sessions.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            "replacement-session".into(),
+        );
+        agent.state.deliveries.insert(
+            crate::openclaw_session::ConversationKey::channel(channel_id),
+            Default::default(),
+        );
         let mut pool = AgentPool::from_slots(vec![Some(agent)]);
 
         assert!(!pool.record_successful_steer(
@@ -7710,7 +7722,8 @@ mod error_outcome_emission_tests {
             "old-session".into(),
         ));
         let returned = pool.agents_mut()[0].as_ref().expect("replacement agent");
-        assert!(returned.state.deliveries[&channel_id]
+        assert!(returned.state.deliveries
+            [&crate::openclaw_session::ConversationKey::channel(channel_id)]
             .delivered_event_ids
             .is_empty());
     }
@@ -7773,7 +7786,9 @@ mod error_outcome_emission_tests {
         );
 
         let returned = pool.agents_mut()[0].as_ref().expect("returned agent");
-        assert!(!returned.state.deliveries.contains_key(&channel_id));
+        assert!(!returned.state.deliveries.contains_key(
+            &crate::openclaw_session::ConversationKey::channel(channel_id)
+        ));
     }
 
     /// Drive one error outcome through `handle_prompt_result` and return how
diff --git a/crates/buzz-acp/src/openclaw_session.rs b/crates/buzz-acp/src/openclaw_session.rs
new file mode 100644
index 00000000000..9a90239bf74
--- /dev/null
+++ b/crates/buzz-acp/src/openclaw_session.rs
@@ -0,0 +1,308 @@
+//! OpenClaw Gateway session keys for last-mile ACP.
+//!
+//! One `openclaw acp` child stays up for the life of the unit. `--session
+//! agent:<id>:buzz` is that process's default Gateway key (heartbeats keep
+//! falling through to it). Conversation `session/new` must send
+//! `_meta.sessionKey` or OpenClaw's mapper reuses that leftover and every
+//! Buzz room piles into one Control UI row.
+//!
+//! Key shape mirrors Slack (`agent:<id>:<channel>:<peer>[:thread:<id>]`) with
+//! a `buzz` namespace. The store key is ids only — never a channel name.
+//! [`crate::config::compose_session_title`] remains the human label
+//! (`_meta.sessionTitle`).
+
+use uuid::Uuid;
+
+use crate::queue::{parse_thread_tags, FlushBatch};
+
+/// Buzz-side conversation that maps to one OpenClaw Gateway session.
+///
+/// Distinct from the ACP `sessionId` OpenClaw returns. The harness still
+/// caches that id in [`crate::pool::SessionState`]; this key is what the
+/// Gateway stores and what the Control UI lists.
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+pub(crate) struct ConversationKey {
+    pub channel_id: Uuid,
+    /// NIP-10 `root` (or lone `reply`) event id from [`parse_thread_tags`].
+    /// `None` for a top-level channel or DM conversation.
+    pub thread_root: Option<String>,
+}
+
+impl ConversationKey {
+    pub(crate) fn channel(channel_id: Uuid) -> Self {
+        Self {
+            channel_id,
+            thread_root: None,
+        }
+    }
+
+    /// Conversation for a flush: channel of the batch plus the last event's
+    /// thread root (`ThreadTags.root_event_id`), when any.
+    pub(crate) fn from_batch(batch: &FlushBatch) -> Self {
+        let thread_root = batch
+            .events
+            .last()
+            .and_then(|event| parse_thread_tags(&event.event).root_event_id)
+            .and_then(normalize_thread_root);
+        Self {
+            channel_id: batch.channel_id,
+            thread_root,
+        }
+    }
+}
+
+fn normalize_thread_root(raw: String) -> Option<String> {
+    let normalized = raw.trim().to_ascii_lowercase();
+    if normalized.is_empty() {
+        None
+    } else {
+        Some(normalized)
+    }
+}
+
+/// Parse `agent:<id>` out of `openclaw acp --session agent:<id>:buzz`.
+///
+/// The leftover `--session` flag stays on the process (do not drop it). New
+/// conversation work must not keep appending to that key.
+pub(crate) fn parse_openclaw_agent_id(agent_args: &[String]) -> Option<String> {
+    let mut args = agent_args.iter();
+    while let Some(arg) = args.next() {
+        let value = if let Some(value) = arg.strip_prefix("--session=") {
+            value
+        } else if arg == "--session" {
+            args.next().map(String::as_str)?
+        } else {
+            continue;
+        };
+        return agent_id_from_session_flag(value);
+    }
+    None
+}
+
+fn agent_id_from_session_flag(flag: &str) -> Option<String> {
+    let mut parts = flag.split(':');
+    if parts.next()? != "agent" {
+        return None;
+    }
+    let id = parts.next()?.trim();
+    if id.is_empty() {
+        return None;
+    }
+    let normalized: String = id
+        .chars()
+        .map(|c| c.to_ascii_lowercase())
+        .filter(|c| c.is_ascii_alphanumeric() || *c == '-' || *c == '_')
+        .collect();
+    if normalized.is_empty() {
+        None
+    } else {
+        Some(normalized)
+    }
+}
+
+/// Stable OpenClaw Gateway key for one Buzz conversation.
+///
+/// - Channel (stream / private / group, no thread):
+///   `agent:<agentId>:buzz:channel:<channelUuid>`
+/// - Thread: `agent:<agentId>:buzz:channel:<channelUuid>:thread:<rootEventId>`
+/// - DM: `agent:<agentId>:buzz:direct:<channelUuid>`
+/// - DM thread: `agent:<agentId>:buzz:direct:<channelUuid>:thread:<rootEventId>`
+pub(crate) fn compose_openclaw_session_key(
+    agent_id: &str,
+    channel_id: Uuid,
+    thread_root: Option<&str>,
+    is_dm: bool,
+) -> String {
+    let peer = if is_dm { "direct" } else { "channel" };
+    let mut key = format!("agent:{agent_id}:buzz:{peer}:{channel_id}");
+    if let Some(root) = thread_root.and_then(|root| normalize_thread_root(root.to_string())) {
+        key.push_str(":thread:");
+        key.push_str(&root);
+    }
+    key
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::queue::BatchEvent;
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+    use std::time::Instant;
+
+    fn channel(n: u8) -> Uuid {
+        Uuid::from_bytes([n; 16])
+    }
+
+    fn root_id(byte: u8) -> String {
+        hex::encode([byte; 32])
+    }
+
+    fn make_event(content: &str, tags: Vec<Tag>) -> nostr::Event {
+        let keys = Keys::generate();
+        EventBuilder::new(Kind::Custom(9), content)
+            .tags(tags)
+            .sign_with_keys(&keys)
+            .expect("sign")
+    }
+
+    fn batch_with(channel_id: Uuid, event: nostr::Event) -> FlushBatch {
+        FlushBatch {
+            channel_id,
+            events: vec![BatchEvent {
+                event,
+                prompt_tag: "@mention".into(),
+                received_at: Instant::now(),
+            }],
+            cancelled_events: vec![],
+            cancel_reason: None,
+        }
+    }
+
+    #[test]
+    fn parse_session_flag_reads_agent_id_from_leftover_buzz_key() {
+        assert_eq!(
+            parse_openclaw_agent_id(&[
+                "acp".into(),
+                "--session".into(),
+                "agent:captain:buzz".into()
+            ])
+            .as_deref(),
+            Some("captain")
+        );
+        assert_eq!(
+            parse_openclaw_agent_id(&["acp".into(), "--session=agent:Mo:buzz".into()]).as_deref(),
+            Some("mo")
+        );
+        assert_eq!(
+            parse_openclaw_agent_id(&[
+                "acp".into(),
+                "--session".into(),
+                "agent:korg:buzz:channel:x".into()
+            ])
+            .as_deref(),
+            Some("korg")
+        );
+    }
+
+    #[test]
+    fn parse_session_flag_ignores_non_openclaw_args() {
+        assert_eq!(parse_openclaw_agent_id(&["acp".into()]), None);
+        assert_eq!(
+            parse_openclaw_agent_id(&["acp".into(), "--session".into(), "main".into()]),
+            None
+        );
+        assert_eq!(parse_openclaw_agent_id(&[]), None);
+    }
+
+    #[test]
+    fn key_is_stable_per_agent_channel_thread_or_dm() {
+        let general = channel(1);
+        let dm = channel(2);
+        let root = root_id(0xab);
+        let channel_key = compose_openclaw_session_key("captain", general, None, false);
+        let thread_key = compose_openclaw_session_key("captain", general, Some(&root), false);
+        let dm_key = compose_openclaw_session_key("captain", dm, None, true);
+
+        assert_eq!(channel_key, format!("agent:captain:buzz:channel:{general}"));
+        assert_eq!(
+            thread_key,
+            format!("agent:captain:buzz:channel:{general}:thread:{root}")
+        );
+        assert_eq!(dm_key, format!("agent:captain:buzz:direct:{dm}"));
+
+        assert_eq!(
+            compose_openclaw_session_key("captain", general, None, false),
+            channel_key,
+            "same inbound channel must reuse the key"
+        );
+        assert_eq!(
+            compose_openclaw_session_key("captain", general, Some(&root), false),
+            thread_key,
+            "same inbound thread must reuse the key"
+        );
+        assert_eq!(
+            compose_openclaw_session_key("captain", dm, None, true),
+            dm_key,
+            "same inbound DM must reuse the key"
+        );
+    }
+
+    #[test]
+    fn different_channels_threads_and_dms_differ() {
+        let a = channel(1);
+        let b = channel(2);
+        let root_a = root_id(0x11);
+        let root_b = root_id(0x22);
+        let keys = [
+            compose_openclaw_session_key("captain", a, None, false),
+            compose_openclaw_session_key("captain", b, None, false),
+            compose_openclaw_session_key("captain", a, Some(&root_a), false),
+            compose_openclaw_session_key("captain", a, Some(&root_b), false),
+            compose_openclaw_session_key("captain", a, None, true),
+            compose_openclaw_session_key("mo", a, None, false),
+        ];
+        for (i, left) in keys.iter().enumerate() {
+            for (j, right) in keys.iter().enumerate() {
+                if i != j {
+                    assert_ne!(
+                        left, right,
+                        "keys {i} and {j} must differ: {left} vs {right}"
+                    );
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn key_uses_uuid_never_channel_name() {
+        let id = channel(9);
+        let key = compose_openclaw_session_key("quasar", id, None, false);
+        assert!(key.contains(&id.to_string()));
+        assert!(!key.contains("general"));
+        assert!(!key.contains("watercooler"));
+    }
+
+    #[test]
+    fn from_batch_uses_root_event_id_not_parent_or_trigger() {
+        let channel_id = channel(3);
+        let root = root_id(0xcd);
+        let parent = root_id(0xef);
+        let event = make_event(
+            "@captain follow up",
+            vec![
+                Tag::parse(["e", &root, "", "root"]).expect("root tag"),
+                Tag::parse(["e", &parent, "", "reply"]).expect("reply tag"),
+            ],
+        );
+        let trigger = event.id.to_hex();
+        let key = ConversationKey::from_batch(&batch_with(channel_id, event));
+        assert_eq!(key.channel_id, channel_id);
+        assert_eq!(key.thread_root.as_deref(), Some(root.as_str()));
+        assert_ne!(key.thread_root.as_deref(), Some(parent.as_str()));
+        assert_ne!(key.thread_root.as_deref(), Some(trigger.as_str()));
+    }
+
+    #[test]
+    fn from_batch_top_level_mention_has_no_thread() {
+        let channel_id = channel(4);
+        let event = make_event("@captain hello", vec![]);
+        let key = ConversationKey::from_batch(&batch_with(channel_id, event));
+        assert_eq!(key, ConversationKey::channel(channel_id));
+        assert_eq!(
+            compose_openclaw_session_key(
+                "captain",
+                key.channel_id,
+                key.thread_root.as_deref(),
+                false
+            ),
+            format!("agent:captain:buzz:channel:{channel_id}")
+        );
+    }
+
+    #[test]
+    fn leftover_agent_buzz_key_is_not_a_conversation_key() {
+        let composed = compose_openclaw_session_key("captain", channel(1), None, false);
+        assert_ne!(composed, "agent:captain:buzz");
+        assert!(composed.starts_with("agent:captain:buzz:"));
+    }
+}
diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs
index b457b0c1249..8eedff9fcf0 100644
--- a/crates/buzz-acp/src/pool.rs
+++ b/crates/buzz-acp/src/pool.rs
@@ -36,6 +36,7 @@ use crate::acp::{
 };
 use crate::config::{compose_session_title, DedupMode, PermissionMode};
 use crate::observer;
+use crate::openclaw_session::{compose_openclaw_session_key, ConversationKey};
 use crate::queue::{
     CancelReason, ContextMessage, ConversationContext, FlushBatch, PromptChannelInfo,
     PromptProfile, PromptProfileLookup, ThreadTags,
@@ -106,18 +107,21 @@ pub struct ChannelDeliveryState {
     pub delivered_event_ids: HashSet<String>,
 }
 
-/// Per-channel session IDs, turn counters, and delivery state.
+/// Per-conversation session IDs, turn counters, and delivery state.
 ///
 /// Separated from `OwnedAgent` so the state machine is testable without
-/// spawning a real agent subprocess.
+/// spawning a real agent subprocess. ACP `sessionId` is cached per
+/// [`ConversationKey`] (channel + optional thread root) so a thread gets its
+/// own `session/new` — and therefore its own OpenClaw `_meta.sessionKey` —
+/// instead of reusing the parent channel session.
 #[derive(Default)]
 pub struct SessionState {
-    /// channel_id → session_id
-    pub sessions: HashMap<Uuid, String>,
+    /// conversation → ACP session_id
+    pub sessions: HashMap<ConversationKey, String>,
     pub heartbeat_session: Option<String>,
-    /// Per-channel turn counters for proactive session rotation.
+    /// Per-conversation turn counters for proactive session rotation.
     /// Incremented on each successful prompt; reset when the session is rotated.
-    pub turn_counts: HashMap<Uuid, u32>,
+    pub turn_counts: HashMap<ConversationKey, u32>,
     /// Turn counter for the heartbeat session.
     pub heartbeat_turn_count: u32,
     /// Whether the live heartbeat session has successfully received `[Base]`.
@@ -132,9 +136,9 @@ pub struct SessionState {
     /// fetch fails — all fail open. Cleared on session invalidation alongside
     /// `core_sections` so the next session picks up any canvas change.
     pub canvas_sections: HashMap<Uuid, String>,
-    /// Per-channel successful-delivery state. Created with the ACP session and
-    /// cleared atomically with every invalidation path.
-    pub deliveries: HashMap<Uuid, ChannelDeliveryState>,
+    /// Per-conversation successful-delivery state. Created with the ACP session
+    /// and cleared atomically with every invalidation path.
+    pub deliveries: HashMap<ConversationKey, ChannelDeliveryState>,
 }
 
 impl SessionState {
@@ -152,14 +156,33 @@ impl SessionState {
         }
     }
 
-    /// Invalidate a single channel's session and turn counter.
-    /// Returns `true` if the channel had an active session.
+    /// Invalidate every conversation session for `channel_id` (membership
+    /// removal, idle model switch). Returns `true` if any session existed.
     pub fn invalidate_channel(&mut self, channel_id: &Uuid) -> bool {
-        self.turn_counts.remove(channel_id);
+        self.turn_counts
+            .retain(|key, _| key.channel_id != *channel_id);
         self.core_sections.remove(channel_id);
         self.canvas_sections.remove(channel_id);
-        self.deliveries.remove(channel_id);
-        self.sessions.remove(channel_id).is_some()
+        self.deliveries
+            .retain(|key, _| key.channel_id != *channel_id);
+        let had_session = self
+            .sessions
+            .keys()
+            .any(|key| key.channel_id == *channel_id);
+        self.sessions.retain(|key, _| key.channel_id != *channel_id);
+        had_session
+    }
+
+    /// Drop one conversation's ACP session without touching sibling threads
+    /// or the parent channel session.
+    pub fn invalidate_conversation(&mut self, key: &ConversationKey) -> bool {
+        self.turn_counts.remove(key);
+        self.deliveries.remove(key);
+        if key.thread_root.is_none() {
+            self.core_sections.remove(&key.channel_id);
+            self.canvas_sections.remove(&key.channel_id);
+        }
+        self.sessions.remove(key).is_some()
     }
 
     /// Invalidate all sessions and turn counters (e.g. after agent exit).
@@ -176,22 +199,32 @@ impl SessionState {
 
     pub(crate) fn mark_channel_delivery_success(
         &mut self,
-        channel_id: Uuid,
+        conversation: ConversationKey,
         standing_context_sent: bool,
         event_ids: impl IntoIterator<Item = String>,
     ) {
-        let delivery = self.deliveries.entry(channel_id).or_default();
+        let delivery = self.deliveries.entry(conversation).or_default();
         delivery.standing_context_sent |= standing_context_sent;
         delivery.delivered_event_ids.extend(event_ids);
     }
 
+    pub(crate) fn has_session_for_channel(&self, channel_id: Uuid) -> bool {
+        self.sessions.keys().any(|key| key.channel_id == channel_id)
+    }
+
     #[cfg(test)]
     fn has_channel_state(&self, channel_id: &Uuid) -> bool {
-        self.sessions.contains_key(channel_id)
-            || self.turn_counts.contains_key(channel_id)
+        self.has_session_for_channel(*channel_id)
+            || self
+                .turn_counts
+                .keys()
+                .any(|key| key.channel_id == *channel_id)
             || self.core_sections.contains_key(channel_id)
             || self.canvas_sections.contains_key(channel_id)
-            || self.deliveries.contains_key(channel_id)
+            || self
+                .deliveries
+                .keys()
+                .any(|key| key.channel_id == *channel_id)
     }
 }
 
@@ -641,6 +674,10 @@ 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,
+    /// OpenClaw agent id parsed from `--session agent:<id>:buzz`. When set,
+    /// conversation `session/new` sends `_meta.sessionKey`. Heartbeats omit it
+    /// so they keep the process-lifetime leftover key.
+    pub openclaw_agent_id: Option<String>,
 }
 
 impl AgentPool {
@@ -672,7 +709,7 @@ impl AgentPool {
         if let Some(cid) = channel_id {
             let idx = self.agents.iter().position(|slot| {
                 slot.as_ref()
-                    .map(|a| a.state.sessions.contains_key(&cid))
+                    .map(|a| a.state.has_session_for_channel(cid))
                     .unwrap_or(false)
             });
             if let Some(i) = idx {
@@ -711,7 +748,7 @@ impl AgentPool {
     pub fn has_session_for(&self, channel_id: Uuid) -> bool {
         self.agents.iter().any(|slot| {
             slot.as_ref()
-                .map(|a| a.state.sessions.contains_key(&channel_id))
+                .map(|a| a.state.has_session_for_channel(channel_id))
                 .unwrap_or(false)
         })
     }
@@ -796,14 +833,25 @@ impl AgentPool {
             return true;
         }
 
-        let Some(agent) = self.agents.iter_mut().flatten().find(|agent| {
-            agent.state.sessions.get(&channel_id).map(String::as_str) == Some(session_id.as_str())
-        }) else {
+        let Some(agent) = self
+            .agents
+            .iter_mut()
+            .flatten()
+            .find(|agent| agent.state.sessions.values().any(|sid| sid == &session_id))
+        else {
+            return false;
+        };
+        let Some(conversation) = agent
+            .state
+            .sessions
+            .iter()
+            .find_map(|(key, sid)| (sid == &session_id).then(|| key.clone()))
+        else {
             return false;
         };
         agent
             .state
-            .mark_channel_delivery_success(channel_id, false, [event_id]);
+            .mark_channel_delivery_success(conversation, false, [event_id]);
         true
     }
 
@@ -885,7 +933,7 @@ impl AgentPool {
             .agents
             .iter_mut()
             .flatten()
-            .find(|a| a.state.sessions.contains_key(&channel_id))
+            .find(|a| a.state.has_session_for_channel(channel_id))
         else {
             return IdleSwitchResult::NoIdleAgent;
         };
@@ -1003,6 +1051,9 @@ struct NewSessionChannelContext<'a> {
     name: Option<&'a str>,
     id: Option<Uuid>,
     channel_type: Option<&'a str>,
+    /// OpenClaw Gateway key (`_meta.sessionKey`). `None` for heartbeats and
+    /// non-OpenClaw agents so the process `--session` default remains leftover.
+    session_key: Option<&'a str>,
 }
 
 async fn create_session_and_apply_model(
@@ -1055,6 +1106,7 @@ async fn create_session_and_apply_model(
                 combined_system_prompt.as_deref(),
             ),
             session_title.as_deref(),
+            channel.session_key,
         )
         .await?;
 
@@ -1776,6 +1828,7 @@ pub async fn run_prompt_task(
     turn_id: String,
 ) {
     // Is this a channel prompt or a heartbeat?
+    let conversation = batch.as_ref().map(ConversationKey::from_batch);
     let source = match &batch {
         Some(b) => PromptSource::Channel(b.channel_id),
         None => PromptSource::Heartbeat,
@@ -1883,8 +1936,10 @@ pub async fn run_prompt_task(
         if let (PromptSource::Channel(cid), Some(owner_pk)) =
             (&source, ctx.agent_owner_pubkey.as_ref())
         {
-            let is_new_channel_session = !agent.state.sessions.contains_key(cid);
-            if is_new_channel_session && !agent.state.core_sections.contains_key(cid) {
+            let is_new_conversation = conversation
+                .as_ref()
+                .is_none_or(|key| !agent.state.sessions.contains_key(key));
+            if is_new_conversation && !agent.state.core_sections.contains_key(cid) {
                 // Bounded — we'd rather start the session with no core hint
                 // than block session creation on a stalled relay.
                 const CORE_FETCH_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3);
@@ -1937,9 +1992,11 @@ pub async fn run_prompt_task(
     let mut title_channel: Option<String> = None;
     let mut origin_channel_type: Option<String> = None;
     if let PromptSource::Channel(cid) = &source {
-        let is_new_channel_session = !agent.state.sessions.contains_key(cid);
-        let needs_canvas = is_new_channel_session && !agent.state.canvas_sections.contains_key(cid);
-        if is_new_channel_session {
+        let is_new_conversation = conversation
+            .as_ref()
+            .is_none_or(|key| !agent.state.sessions.contains_key(key));
+        let needs_canvas = is_new_conversation && !agent.state.canvas_sections.contains_key(cid);
+        if is_new_conversation {
             let (is_dm, resolved_channel, resolved_channel_type) =
                 resolve_new_session_channel_context(&ctx.channel_info, *cid).await;
             title_channel = resolved_channel;
@@ -1979,13 +2036,28 @@ pub async fn run_prompt_task(
 
     let (session_id, is_new_session) = match &source {
         PromptSource::Channel(cid) => {
-            if let Some(sid) = agent.state.sessions.get(cid) {
+            let conversation_key = conversation
+                .clone()
+                .unwrap_or_else(|| ConversationKey::channel(*cid));
+            if let Some(sid) = agent.state.sessions.get(&conversation_key) {
                 (sid.clone(), false)
             } else {
                 // The title is channel-qualified (`Agent · #channel`) so one
                 // agent in several channels doesn't produce identical session
                 // rows; `title_channel` comes from the single resolve above and
                 // is `None` for DM, unresolved, and unnamed channels.
+                //
+                // OpenClaw's mapper uses `_meta.sessionKey` when present, else
+                // the process `--session` leftover. Heartbeats omit the key.
+                // Confirmed DMs use `direct:`; unresolved types stay `channel:`.
+                let session_key = ctx.openclaw_agent_id.as_deref().map(|agent_id| {
+                    compose_openclaw_session_key(
+                        agent_id,
+                        conversation_key.channel_id,
+                        conversation_key.thread_root.as_deref(),
+                        origin_channel_type.as_deref() == Some("dm"),
+                    )
+                });
                 match create_session_and_apply_model(
                     &mut agent,
                     &ctx,
@@ -1996,6 +2068,7 @@ pub async fn run_prompt_task(
                         name: title_channel.as_deref(),
                         id: Some(*cid),
                         channel_type: origin_channel_type.as_deref(),
+                        session_key: session_key.as_deref(),
                     },
                 )
                 .await
@@ -2003,13 +2076,17 @@ pub async fn run_prompt_task(
                     Ok(sid) => {
                         tracing::info!(
                             target: "pool::session",
+                            conversation = ?conversation_key,
                             "created session {sid} for channel {cid}"
                         );
-                        agent.state.sessions.insert(*cid, sid.clone());
+                        agent
+                            .state
+                            .sessions
+                            .insert(conversation_key.clone(), sid.clone());
                         agent
                             .state
                             .deliveries
-                            .insert(*cid, ChannelDeliveryState::default());
+                            .insert(conversation_key, ChannelDeliveryState::default());
                         // Seed a zero usage baseline: buzz-acp spawned this session
                         // so prior usage is zero by definition — first turn is reliable.
                         agent.acp.notify_session_spawned(&sid);
@@ -2061,6 +2138,7 @@ pub async fn run_prompt_task(
                         name: None,
                         id: None,
                         channel_type: None,
+                        session_key: None,
                     },
                 )
                 .await
@@ -2139,11 +2217,16 @@ pub async fn run_prompt_task(
     // sessions created before this field existed fail safe by behaving as
     // undelivered once, rather than silently omitting standing context.
     let mut standing_context_sent = match &source {
-        PromptSource::Channel(cid) => agent
-            .state
-            .deliveries
-            .get(cid)
-            .is_some_and(|delivery| delivery.standing_context_sent),
+        PromptSource::Channel(cid) => {
+            let key = conversation
+                .clone()
+                .unwrap_or_else(|| ConversationKey::channel(*cid));
+            agent
+                .state
+                .deliveries
+                .get(&key)
+                .is_some_and(|delivery| delivery.standing_context_sent)
+        }
         PromptSource::Heartbeat => agent.state.heartbeat_standing_context_sent,
     };
 
@@ -2183,7 +2266,10 @@ pub async fn run_prompt_task(
                     // prompt below must not repeat it. Every other arm returns.
                     standing_context_sent = true;
                     if !agent.has_system_prompt_support() {
-                        agent.state.mark_channel_delivery_success(*cid, true, []);
+                        let key = conversation
+                            .clone()
+                            .unwrap_or_else(|| ConversationKey::channel(*cid));
+                        agent.state.mark_channel_delivery_success(key, true, []);
                     }
                     let usage = agent.acp.take_turn_usage();
                     publish_agent_turn_metric(
@@ -2351,7 +2437,7 @@ pub async fn run_prompt_task(
         let delivered_ids = agent
             .state
             .deliveries
-            .get(&b.channel_id)
+            .get(&ConversationKey::from_batch(b))
             .map(|delivery| &delivery.delivered_event_ids)
             .cloned()
             .unwrap_or_default();
@@ -2621,8 +2707,11 @@ pub async fn run_prompt_task(
                         }
                         if let PromptSource::Channel(cid) = &source {
                             let standing_sent = !agent.has_system_prompt_support();
+                            let key = conversation
+                                .clone()
+                                .unwrap_or_else(|| ConversationKey::channel(*cid));
                             agent.state.mark_channel_delivery_success(
-                                *cid,
+                                key,
                                 standing_sent,
                                 pending_delivered_event_ids.iter().cloned(),
                             );
@@ -2669,8 +2758,11 @@ pub async fn run_prompt_task(
 
             if let PromptSource::Channel(cid) = &source {
                 let standing_sent = !agent.has_system_prompt_support();
+                let key = conversation
+                    .clone()
+                    .unwrap_or_else(|| ConversationKey::channel(*cid));
                 agent.state.mark_channel_delivery_success(
-                    *cid,
+                    key,
                     standing_sent,
                     pending_delivered_event_ids.iter().cloned(),
                 );
@@ -2688,7 +2780,10 @@ pub async fn run_prompt_task(
                 if limit > 0 {
                     match &source {
                         PromptSource::Channel(cid) => {
-                            let count = agent.state.turn_counts.entry(*cid).or_insert(0);
+                            let key = conversation
+                                .clone()
+                                .unwrap_or_else(|| ConversationKey::channel(*cid));
+                            let count = agent.state.turn_counts.entry(key).or_insert(0);
                             *count += 1;
                             *count >= limit
                         }
@@ -2707,7 +2802,11 @@ pub async fn run_prompt_task(
                     target: "pool::session",
                     "rotating session for {source:?} after {stop_reason:?}",
                 );
-                agent.state.invalidate(&source);
+                if let Some(key) = conversation.as_ref() {
+                    agent.state.invalidate_conversation(key);
+                } else {
+                    agent.state.invalidate(&source);
+                }
             }
 
             let core_stop = acp_stop_to_core(&stop_reason);
@@ -4733,6 +4832,10 @@ mod tests {
     use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
     use serde_json::json;
 
+    fn ck(id: Uuid) -> ConversationKey {
+        ConversationKey::channel(id)
+    }
+
     fn test_mcp_server() -> McpServer {
         McpServer {
             name: "dev".into(),
@@ -6123,11 +6226,11 @@ done"#
         agent
             .state
             .sessions
-            .insert(channel_id, "live-session".into());
+            .insert(ck(channel_id), "live-session".into());
         agent
             .state
             .deliveries
-            .insert(channel_id, ChannelDeliveryState::default());
+            .insert(ck(channel_id), ChannelDeliveryState::default());
 
         let mut ctx = make_prompt_context_no_owner();
         ctx.base_prompt = Some("standing-once");
@@ -6167,7 +6270,7 @@ done"#
                     PromptOutcome::Ok(StopReason::EndTurn)
                 )),
             }
-            let delivery = &result.agent.state.deliveries[&channel_id];
+            let delivery = &result.agent.state.deliveries[&ck(channel_id)];
             assert_eq!(
                 delivery.standing_context_sent,
                 turn >= 2,
@@ -6298,11 +6401,11 @@ done"#
         agent
             .state
             .sessions
-            .insert(channel_id, "live-session".into());
+            .insert(ck(channel_id), "live-session".into());
         agent
             .state
             .deliveries
-            .insert(channel_id, ChannelDeliveryState::default());
+            .insert(ck(channel_id), ChannelDeliveryState::default());
 
         let mut ctx = make_prompt_context_no_owner();
         ctx.context_message_limit = 10;
@@ -6344,7 +6447,7 @@ done"#
             ));
             agent = result.agent;
         }
-        let delivery = &agent.state.deliveries[&channel_id];
+        let delivery = &agent.state.deliveries[&ck(channel_id)];
         assert!(delivery.delivered_event_ids.contains(&carry_over_id));
         assert!(delivery.delivered_event_ids.contains(&new_event_id));
         agent.acp.shutdown().await;
@@ -6451,11 +6554,11 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         agent
             .state
             .sessions
-            .insert(channel_id, "live-session".into());
+            .insert(ck(channel_id), "live-session".into());
         agent
             .state
             .deliveries
-            .insert(channel_id, ChannelDeliveryState::default());
+            .insert(ck(channel_id), ChannelDeliveryState::default());
 
         // Model the adversarial ordering: the task result has already retired
         // its TaskMeta and returned the agent before the successful ack arrives.
@@ -6529,19 +6632,19 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let mut state = SessionState::default();
         state
             .deliveries
-            .insert(channel, ChannelDeliveryState::default());
+            .insert(ck(channel), ChannelDeliveryState::default());
 
         // Building or attempting a prompt does not mutate delivery state.
-        let delivery = state.deliveries.get(&channel).unwrap();
+        let delivery = state.deliveries.get(&ck(channel)).unwrap();
         assert!(!delivery.standing_context_sent);
         assert!(delivery.delivered_event_ids.is_empty());
 
         state.mark_channel_delivery_success(
-            channel,
+            ck(channel),
             true,
             ["trigger".to_string(), "context".to_string()],
         );
-        let delivery = state.deliveries.get(&channel).unwrap();
+        let delivery = state.deliveries.get(&ck(channel)).unwrap();
         assert!(delivery.standing_context_sent);
         assert_eq!(delivery.delivered_event_ids.len(), 2);
     }
@@ -6550,17 +6653,17 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
     fn delivery_state_is_cleared_on_rotation_and_restarts_empty() {
         let channel = Uuid::new_v4();
         let mut state = SessionState::default();
-        state.sessions.insert(channel, "old-session".into());
-        state.mark_channel_delivery_success(channel, true, ["old-event".to_string()]);
+        state.sessions.insert(ck(channel), "old-session".into());
+        state.mark_channel_delivery_success(ck(channel), true, ["old-event".to_string()]);
 
         assert!(state.invalidate_channel(&channel));
-        assert!(!state.deliveries.contains_key(&channel));
+        assert!(!state.deliveries.contains_key(&ck(channel)));
 
-        state.sessions.insert(channel, "new-session".into());
+        state.sessions.insert(ck(channel), "new-session".into());
         state
             .deliveries
-            .insert(channel, ChannelDeliveryState::default());
-        let delivery = state.deliveries.get(&channel).unwrap();
+            .insert(ck(channel), ChannelDeliveryState::default());
+        let delivery = state.deliveries.get(&ck(channel)).unwrap();
         assert!(!delivery.standing_context_sent);
         assert!(delivery.delivered_event_ids.is_empty());
     }
@@ -6667,21 +6770,21 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let ch_a = Uuid::new_v4();
         let ch_b = Uuid::new_v4();
         let mut s = SessionState::default();
-        s.sessions.insert(ch_a, "sess-a".into());
-        s.sessions.insert(ch_b, "sess-b".into());
-        s.turn_counts.insert(ch_a, 5);
-        s.turn_counts.insert(ch_b, 3);
+        s.sessions.insert(ck(ch_a), "sess-a".into());
+        s.sessions.insert(ck(ch_b), "sess-b".into());
+        s.turn_counts.insert(ck(ch_a), 5);
+        s.turn_counts.insert(ck(ch_b), 3);
         s.core_sections.insert(ch_a, "core-a".into());
         s.core_sections.insert(ch_b, "core-b".into());
         s.deliveries.insert(
-            ch_a,
+            ck(ch_a),
             ChannelDeliveryState {
                 standing_context_sent: true,
                 delivered_event_ids: HashSet::from(["event-a".into()]),
             },
         );
         s.deliveries.insert(
-            ch_b,
+            ck(ch_b),
             ChannelDeliveryState {
                 standing_context_sent: true,
                 delivered_event_ids: HashSet::from(["event-b".into()]),
@@ -6703,12 +6806,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             &ControlSignal::Rotate,
         );
 
-        assert!(!s.sessions.contains_key(&ch_a));
-        assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.sessions.contains_key(&ck(ch_a)));
+        assert!(!s.turn_counts.contains_key(&ck(ch_a)));
         assert!(!s.core_sections.contains_key(&ch_a));
         assert!(!s.has_channel_state(&ch_a));
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
         assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
         assert_eq!(s.heartbeat_turn_count, 7);
@@ -6724,10 +6827,10 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             &ControlSignal::Cancel,
         );
 
-        assert_eq!(s.sessions.get(&ch_a).unwrap(), "sess-a");
-        assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
+        assert_eq!(s.sessions.get(&ck(ch_a)).unwrap(), "sess-a");
+        assert_eq!(*s.turn_counts.get(&ck(ch_a)).unwrap(), 5);
         assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
     }
 
     #[test]
@@ -6735,13 +6838,13 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let (mut s, ch_a, ch_b) = make_state();
         s.invalidate(&PromptSource::Channel(ch_a));
 
-        assert!(!s.sessions.contains_key(&ch_a));
-        assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.sessions.contains_key(&ck(ch_a)));
+        assert!(!s.turn_counts.contains_key(&ck(ch_a)));
         assert!(!s.core_sections.contains_key(&ch_a));
         assert!(!s.has_channel_state(&ch_a));
         // ch_b untouched
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
         // heartbeat untouched
         assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
@@ -6758,8 +6861,8 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         assert!(!s.heartbeat_standing_context_sent);
         // channels untouched
         assert_eq!(s.sessions.len(), 2);
-        assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(*s.turn_counts.get(&ck(ch_a)).unwrap(), 5);
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
@@ -6786,8 +6889,8 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         // Everything still intact.
         assert_eq!(s.sessions.len(), 2);
         assert_eq!(s.turn_counts.len(), 2);
-        assert_eq!(*s.turn_counts.get(&ch_a).unwrap(), 5);
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(*s.turn_counts.get(&ck(ch_a)).unwrap(), 5);
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
@@ -6805,19 +6908,42 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
     fn test_invalidate_channel_returns_true_when_session_existed() {
         let (mut s, ch_a, ch_b) = make_state();
         assert!(s.invalidate_channel(&ch_a));
-        assert!(!s.sessions.contains_key(&ch_a));
-        assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.sessions.contains_key(&ck(ch_a)));
+        assert!(!s.turn_counts.contains_key(&ck(ch_a)));
         assert!(!s.core_sections.contains_key(&ch_a));
         assert!(!s.has_channel_state(&ch_a));
         // ch_b untouched
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
         // heartbeat untouched
         assert_eq!(s.heartbeat_session.as_deref(), Some("sess-hb"));
         assert_eq!(s.heartbeat_turn_count, 7);
     }
 
+    #[test]
+    fn test_thread_conversation_does_not_reuse_parent_channel_session() {
+        let (mut s, ch_a, ch_b) = make_state();
+        let thread = ConversationKey {
+            channel_id: ch_a,
+            thread_root: Some("ab".repeat(32)),
+        };
+        s.sessions.insert(thread.clone(), "sess-thread".into());
+
+        assert_eq!(s.sessions.get(&ck(ch_a)).unwrap(), "sess-a");
+        assert_eq!(s.sessions.get(&thread).unwrap(), "sess-thread");
+        assert_ne!(
+            s.sessions.get(&ck(ch_a)).unwrap(),
+            s.sessions.get(&thread).unwrap()
+        );
+
+        s.invalidate_conversation(&thread);
+        assert!(s.sessions.get(&thread).is_none());
+        assert_eq!(s.sessions.get(&ck(ch_a)).unwrap(), "sess-a");
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(s.core_sections.get(&ch_a).unwrap(), "core-a");
+    }
+
     #[test]
     fn test_invalidate_channel_returns_false_when_no_session() {
         let (mut s, _ch_a, _ch_b) = make_state();
@@ -6837,12 +6963,12 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         for ch in &removed {
             s.invalidate_channel(ch);
         }
-        assert!(!s.sessions.contains_key(&ch_a));
-        assert!(!s.turn_counts.contains_key(&ch_a));
+        assert!(!s.sessions.contains_key(&ck(ch_a)));
+        assert!(!s.turn_counts.contains_key(&ck(ch_a)));
         assert!(!s.core_sections.contains_key(&ch_a));
         assert!(!s.has_channel_state(&ch_a));
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
         assert_eq!(s.core_sections.get(&ch_b).unwrap(), "core-b");
     }
 
@@ -6865,8 +6991,8 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
 
         assert!(!s.has_channel_state(&ch_a));
         // ch_b untouched — the switch is channel-scoped.
-        assert_eq!(s.sessions.get(&ch_b).unwrap(), "sess-b");
-        assert_eq!(*s.turn_counts.get(&ch_b).unwrap(), 3);
+        assert_eq!(s.sessions.get(&ck(ch_b)).unwrap(), "sess-b");
+        assert_eq!(*s.turn_counts.get(&ck(ch_b)).unwrap(), 3);
     }
 
     // ── requeue_cancelled_batch ────────────────────────────────────────────
@@ -7992,6 +8118,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             memory_enabled: false,
             harness_name: "goose".to_string(),
             relay_url: "ws://127.0.0.1:3000".to_string(),
+            openclaw_agent_id: None,
         }
     }
 
@@ -8097,14 +8224,14 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
     fn test_invalidate_channel_clears_canvas_section() {
         let ch = Uuid::new_v4();
         let mut s = SessionState::default();
-        s.sessions.insert(ch, "sess".into());
+        s.sessions.insert(ck(ch), "sess".into());
         s.canvas_sections
             .insert(ch, "[Channel Canvas]\nrev abc".into());
 
         s.invalidate_channel(&ch);
 
         assert!(!s.canvas_sections.contains_key(&ch));
-        assert!(!s.sessions.contains_key(&ch));
+        assert!(!s.sessions.contains_key(&ck(ch)));
     }
 
     #[test]
@@ -8114,7 +8241,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let mut s = SessionState::default();
         s.canvas_sections.insert(ch_a, "canvas-a".into());
         s.canvas_sections.insert(ch_b, "canvas-b".into());
-        s.sessions.insert(ch_a, "sess-a".into());
+        s.sessions.insert(ck(ch_a), "sess-a".into());
 
         s.invalidate_all();
 
@@ -8127,8 +8254,8 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         let ch_a = Uuid::new_v4();
         let ch_b = Uuid::new_v4();
         let mut s = SessionState::default();
-        s.sessions.insert(ch_a, "sess-a".into());
-        s.sessions.insert(ch_b, "sess-b".into());
+        s.sessions.insert(ck(ch_a), "sess-a".into());
+        s.sessions.insert(ck(ch_b), "sess-b".into());
         s.canvas_sections.insert(ch_a, "canvas-a".into());
         s.canvas_sections.insert(ch_b, "canvas-b".into());
 
@@ -8620,6 +8747,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8657,6 +8785,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8691,6 +8820,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8724,6 +8854,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8764,6 +8895,7 @@ exit 0"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8891,6 +9023,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -8962,6 +9095,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9017,6 +9151,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9059,6 +9194,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9100,6 +9236,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9166,6 +9303,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9203,6 +9341,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9276,6 +9415,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await
@@ -9317,6 +9457,7 @@ done"#
                 name: None,
                 id: None,
                 channel_type: None,
+                session_key: None,
             },
         )
         .await