diff --git a/crates/buzz-acp/src/conversation.rs b/crates/buzz-acp/src/conversation.rs
new file mode 100644
index 00000000000..f2d1dc0b3c5
--- /dev/null
+++ b/crates/buzz-acp/src/conversation.rs
@@ -0,0 +1,126 @@
+use nostr::Event;
+use sha2::{Digest, Sha256};
+use uuid::Uuid;
+
+/// Return the scheduler/session identity for an inbound event.
+///
+/// Channel UUIDs remain the identity for DMs. Channel messages use their
+/// NIP-10 root event, or their own event ID when starting a new thread. This
+/// lets multiple threads in one channel occupy independent pool slots without
+/// changing the queue and session-state APIs that already accept UUIDs.
+pub fn id_for_event(channel_id: Uuid, event: &Event, is_dm: bool) -> Uuid {
+    if is_dm {
+        return channel_id;
+    }
+
+    let root = crate::queue::parse_thread_tags(event)
+        .root_event_id
+        .unwrap_or_else(|| event.id.to_hex());
+    deterministic_id(channel_id, &root)
+}
+
+/// Recover the real channel UUID carried by a NIP-29 event.
+///
+/// Tests and legacy callers may construct events without an `h` tag, so users
+/// should fall back to the queue key when this returns `None`.
+pub fn routing_channel_id(event: &Event) -> Option<Uuid> {
+    event.tags.iter().find_map(|tag| {
+        let parts = tag.as_slice();
+        (parts.first().map(String::as_str) == Some("h"))
+            .then(|| parts.get(1))
+            .flatten()
+            .and_then(|value| Uuid::parse_str(value).ok())
+    })
+}
+
+fn deterministic_id(channel_id: Uuid, root_event_id: &str) -> Uuid {
+    let mut digest = Sha256::new();
+    digest.update(b"buzz-acp-conversation-v1");
+    digest.update(channel_id.as_bytes());
+    digest.update(root_event_id.as_bytes());
+    let hash = digest.finalize();
+    let mut bytes = [0_u8; 16];
+    bytes.copy_from_slice(&hash[..16]);
+    Uuid::from_bytes(bytes)
+}
+
+#[cfg(test)]
+mod tests {
+    use std::time::Instant;
+
+    use nostr::{EventBuilder, Keys, Kind, Tag};
+
+    use super::*;
+    use crate::config::DedupMode;
+    use crate::queue::{EventQueue, QueuedEvent};
+
+    fn event(tags: Vec<Tag>) -> Event {
+        EventBuilder::new(Kind::TextNote, "task")
+            .tags(tags)
+            .sign_with_keys(&Keys::generate())
+            .expect("test event signs")
+    }
+
+    #[test]
+    fn top_level_channel_events_get_distinct_conversation_ids() {
+        let channel = Uuid::new_v4();
+        let first = event(vec![]);
+        let second = event(vec![]);
+
+        assert_ne!(
+            id_for_event(channel, &first, false),
+            id_for_event(channel, &second, false)
+        );
+    }
+
+    #[test]
+    fn replies_to_same_root_share_a_conversation_id() {
+        let channel = Uuid::new_v4();
+        let root = event(vec![]);
+        let root_id = root.id.to_hex();
+        let reply_tag = Tag::parse(["e", root_id.as_str(), "", "reply"]).expect("valid reply tag");
+        let first = event(vec![reply_tag.clone()]);
+        let second = event(vec![reply_tag]);
+
+        assert_eq!(
+            id_for_event(channel, &first, false),
+            id_for_event(channel, &second, false)
+        );
+    }
+
+    #[test]
+    fn dm_events_keep_channel_identity() {
+        let channel = Uuid::new_v4();
+        assert_eq!(id_for_event(channel, &event(vec![]), true), channel);
+    }
+
+    #[test]
+    fn two_threads_in_one_channel_can_be_in_flight_together() {
+        let channel = Uuid::new_v4();
+        let channel_tag =
+            Tag::parse(["h", channel.to_string().as_str()]).expect("valid channel tag");
+        let first = event(vec![channel_tag.clone()]);
+        let second = event(vec![channel_tag]);
+        let first_id = id_for_event(channel, &first, false);
+        let second_id = id_for_event(channel, &second, false);
+        let mut queue = EventQueue::new(DedupMode::Queue);
+
+        for (conversation, event) in [(first_id, first), (second_id, second)] {
+            assert!(queue.push(QueuedEvent {
+                channel_id: conversation,
+                event,
+                received_at: Instant::now(),
+                prompt_tag: "mention".to_string(),
+            }));
+        }
+
+        let first_batch = queue.flush_next().expect("first thread flushes");
+        let second_batch = queue
+            .flush_next()
+            .expect("second thread flushes while first is in flight");
+
+        assert_ne!(first_batch.channel_id, second_batch.channel_id);
+        assert_eq!(first_batch.routing_channel_id(), channel);
+        assert_eq!(second_batch.routing_channel_id(), channel);
+    }
+}
diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs
index 30a47e3ee17..d6d9520e89c 100644
--- a/crates/buzz-acp/src/lib.rs
+++ b/crates/buzz-acp/src/lib.rs
@@ -2,6 +2,7 @@
 
 mod acp;
 mod config;
+mod conversation;
 mod engram_fetch;
 mod filter;
 mod observer;
@@ -10,6 +11,9 @@ mod pool_lifecycle;
 mod queue;
 mod relay;
 mod setup_mode;
+mod thread_workspace;
+#[cfg(test)]
+mod thread_workspace_tests;
 mod usage;
 
 pub use usage::TurnUsage;
@@ -47,6 +51,12 @@ use tokio::sync::{mpsc, watch};
 use tracing_subscriber::EnvFilter;
 use uuid::Uuid;
 
+#[derive(Debug, Clone)]
+struct TypingState {
+    routing_channel_id: Uuid,
+    thread_tags: ThreadTags,
+}
+
 /// Check if argv[1] matches a subcommand name, before any clap parsing.
 ///
 /// This avoids clap rejecting harness flags (like `--private-key`) that aren't
@@ -905,22 +915,42 @@ fn handle_cancel_turn_control(
         tracing::warn!("observer cancel_turn control frame missing valid channelId");
         return;
     };
+    let conversation_id = payload
+        .get("conversationId")
+        .and_then(|value| value.as_str())
+        .and_then(|value| value.parse::<Uuid>().ok());
+    let turn_id = payload
+        .get("turnId")
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.is_empty());
 
-    let fired = signal_in_flight_task(pool, channel_id, ControlSignal::Cancel);
+    let fired = if let Some(turn_id) = turn_id {
+        signal_in_flight_turn(pool, turn_id, ControlSignal::Cancel)
+    } else {
+        signal_in_flight_task(
+            pool,
+            conversation_id.unwrap_or(channel_id),
+            channel_id,
+            ControlSignal::Cancel,
+        )
+    };
     let status = if fired { "sent" } else { "no_active_turn" };
     if let Some(observer) = observer {
+        let context = observer::context_for_conversation(
+            Some(channel_id),
+            conversation_id,
+            None,
+            turn_id.map(ToOwned::to_owned),
+        );
         observer.emit(
             "control_result",
             None,
-            &observer::ObserverContext {
-                channel_id: Some(channel_id.to_string()),
-                session_id: None,
-                turn_id: None,
-                started_at: None,
-            },
+            &context,
             serde_json::json!({
                 "type": "cancel_turn",
                 "status": status,
+                "conversationId": conversation_id.map(|id| id.to_string()),
+                "turnId": turn_id,
             }),
         );
     }
@@ -954,31 +984,51 @@ fn handle_switch_model_control(
         tracing::warn!("observer switch_model control frame missing modelId");
         return;
     };
-
-    // A turn is in flight for this channel iff a task_map entry exists. The
-    // agent is moved out of the pool during a turn, so the control oneshot is
-    // the only reachable lever; an idle channel has no such entry.
-    let turn_in_flight = pool
-        .task_map()
-        .values()
-        .any(|m| m.channel_id == Some(channel_id));
+    let conversation_id = payload
+        .get("conversationId")
+        .and_then(|value| value.as_str())
+        .and_then(|value| value.parse::<Uuid>().ok());
+    let turn_id = payload
+        .get("turnId")
+        .and_then(|value| value.as_str())
+        .filter(|value| !value.is_empty());
+    let target_conversation_id = conversation_id.unwrap_or(channel_id);
+
+    // Prefer the exact turn emitted by the observer. The conversation ID
+    // remains the stable fallback when the turn completed between the
+    // desktop snapshot and this control frame.
+    let turn_in_flight = pool.task_map().values().any(|meta| {
+        turn_id
+            .map(|turn_id| meta.turn_id == turn_id)
+            .unwrap_or(meta.channel_id == Some(target_conversation_id))
+    });
 
     let status = if turn_in_flight {
         // Busy path: deliver over the oneshot. `false` means the oneshot was
         // already consumed this turn (a prior cancel/interrupt) — the turn is
         // already ending, so the switch cannot land on it.
-        if signal_in_flight_task(
-            pool,
-            channel_id,
-            ControlSignal::SwitchModel(model_id.to_string()),
-        ) {
+        let fired = if let Some(turn_id) = turn_id {
+            signal_in_flight_turn(
+                pool,
+                turn_id,
+                ControlSignal::SwitchModel(model_id.to_string()),
+            )
+        } else {
+            signal_in_flight_task(
+                pool,
+                target_conversation_id,
+                channel_id,
+                ControlSignal::SwitchModel(model_id.to_string()),
+            )
+        };
+        if fired {
             "sent"
         } else {
             "turn_ending"
         }
     } else {
         // Idle path: validate against the cached catalog before invalidating.
-        match pool.switch_idle_agent_model(channel_id, model_id) {
+        match pool.switch_idle_agent_model(target_conversation_id, model_id) {
             IdleSwitchResult::Switched => "switched",
             IdleSwitchResult::UnsupportedModel => "unsupported_model",
             IdleSwitchResult::NoIdleAgent => "no_active_turn",
@@ -986,19 +1036,22 @@ fn handle_switch_model_control(
     };
 
     if let Some(observer) = observer {
+        let context = observer::context_for_conversation(
+            Some(channel_id),
+            conversation_id,
+            None,
+            turn_id.map(ToOwned::to_owned),
+        );
         observer.emit(
             "control_result",
             None,
-            &observer::ObserverContext {
-                channel_id: Some(channel_id.to_string()),
-                session_id: None,
-                turn_id: None,
-                started_at: None,
-            },
+            &context,
             serde_json::json!({
                 "type": "switch_model",
                 "status": status,
                 "modelId": model_id,
+                "conversationId": conversation_id.map(|id| id.to_string()),
+                "turnId": turn_id,
             }),
         );
     }
@@ -1598,7 +1651,7 @@ async fn tokio_main() -> Result<()> {
     } else {
         None
     };
-    let mut typing_channels: HashMap<Uuid, ThreadTags> = HashMap::new();
+    let mut typing_channels: HashMap<Uuid, TypingState> = HashMap::new();
     let mut presence_task: Option<tokio::task::JoinHandle<()>> = None;
 
     // Runs at the TOP of every loop iteration via Instant check — cannot be
@@ -1986,7 +2039,7 @@ async fn tokio_main() -> Result<()> {
                                     // removed channel. Events already in-flight will
                                     // complete normally (the relay may reject actions if
                                     // the agent lost access).
-                                    let drained_ids = queue.drain_channel(ch);
+                                    let drained_ids = queue.drain_routing_channel(ch);
                                     let invalidated = if pool_ready {
                                         pool.invalidate_channel_sessions(ch)
                                     } else {
@@ -1995,7 +2048,9 @@ async fn tokio_main() -> Result<()> {
                                     // Track removed channels so checked-out agents get
                                     // their sessions stripped when they return to the pool.
                                     removed_channels.insert(ch);
-                                    typing_channels.remove(&ch);
+                                    typing_channels.retain(|_, state| {
+                                        state.routing_channel_id != ch
+                                    });
                                     // Best-effort: clean up 👀 on drained events.
                                     // Note: the relay revokes membership before
                                     // emitting the notification, so this DELETE may
@@ -2053,6 +2108,14 @@ async fn tokio_main() -> Result<()> {
                                 // contain "!shutdown" from a non-owner.
                             }
 
+                            let inbound_is_dm =
+                                is_dm_channel(buzz_event.channel_id, &ctx.channel_info).await;
+                            let inbound_conversation_id = conversation::id_for_event(
+                                buzz_event.channel_id,
+                                &buzz_event.event,
+                                inbound_is_dm,
+                            );
+
                             // Mirrors !shutdown: kind:9, content "!cancel", from
                             // owner, mentions THIS agent. Must be BEFORE
                             // queue.push() — the event content is moved by push.
@@ -2071,6 +2134,7 @@ async fn tokio_main() -> Result<()> {
                                     if buzz_event.event.pubkey.to_hex() == *owner {
                                         let fired = signal_in_flight_task(
                                             &mut pool,
+                                            inbound_conversation_id,
                                             buzz_event.channel_id,
                                             ControlSignal::Cancel,
                                         );
@@ -2109,6 +2173,7 @@ async fn tokio_main() -> Result<()> {
                                     if buzz_event.event.pubkey.to_hex() == *owner {
                                         let fired = signal_in_flight_task(
                                             &mut pool,
+                                            inbound_conversation_id,
                                             buzz_event.channel_id,
                                             ControlSignal::Rotate,
                                         );
@@ -2148,7 +2213,7 @@ async fn tokio_main() -> Result<()> {
                                 // to DM) so allowlist/anyone modes cannot be
                                 // exercised by non-owner authors inside DMs.
                                 let is_dm =
-                                    is_dm_channel(buzz_event.channel_id, &ctx.channel_info).await;
+                                    inbound_is_dm;
                                 let allowed = author_allowed(
                                     &config.respond_to,
                                     &config.respond_to_allowlist,
@@ -2195,7 +2260,7 @@ async fn tokio_main() -> Result<()> {
                             let event_for_steer = buzz_event.event.clone();
                             let prompt_tag_for_steer = prompt_tag.clone();
                             let accepted = queue.push(QueuedEvent {
-                                channel_id: buzz_event.channel_id,
+                                channel_id: inbound_conversation_id,
                                 event: buzz_event.event,
                                 received_at: std::time::Instant::now(),
                                 prompt_tag,
@@ -2215,7 +2280,7 @@ async fn tokio_main() -> Result<()> {
                             // Event is already queued. If mode requires it AND
                             // the channel has an in-flight task, fire cancel —
                             // OR take the non-cancelling (ACP steer) fork for Steer signals.
-                            if accepted && queue.is_channel_in_flight(buzz_event.channel_id) {
+                            if accepted && queue.is_channel_in_flight(inbound_conversation_id) {
                                 // Author eligibility (owner ∪ allowlist ∪ siblings)
                                 // is already enforced by the inbound author gate
                                 // above, so the mid-turn signal fires for every
@@ -2242,6 +2307,7 @@ async fn tokio_main() -> Result<()> {
                                         && try_native_steer(
                                             &mut pool,
                                             &mut queue,
+                                            inbound_conversation_id,
                                             buzz_event.channel_id,
                                             event_for_steer,
                                             prompt_tag_for_steer,
@@ -2250,6 +2316,7 @@ async fn tokio_main() -> Result<()> {
                                     if !native_attempted {
                                         signal_in_flight_task(
                                             &mut pool,
+                                            inbound_conversation_id,
                                             buzz_event.channel_id,
                                             signal,
                                         );
@@ -2328,14 +2395,17 @@ async fn tokio_main() -> Result<()> {
                     // Use try_publish (non-blocking) for typing indicators —
                     // they're ephemeral and must not block the main loop during
                     // relay reconnection (#35).
-                    for (&ch, thread_tags) in &typing_channels {
+                    for state in typing_channels.values() {
                         if let Ok(event) = relay.build_typing_event(
-                            ch,
-                            thread_tags.root_event_id.as_deref(),
-                            thread_tags.parent_event_id.as_deref(),
+                            state.routing_channel_id,
+                            state.thread_tags.root_event_id.as_deref(),
+                            state.thread_tags.parent_event_id.as_deref(),
                         ) {
                             if let Err(e) = relay.try_publish_event(event) {
-                                tracing::debug!("typing indicator dropped for {ch}: {e}");
+                                tracing::debug!(
+                                    channel_id = %state.routing_channel_id,
+                                    "typing indicator dropped: {e}"
+                                );
                             }
                         }
                     }
@@ -2517,8 +2587,15 @@ async fn tokio_main() -> Result<()> {
                     Ok(pool::SteerAck::PromptCompletedNeutral) => (true, false, false),
                     Err(_recv_err) => (true, false, false),
                 };
+                let routing_channel_id = pool
+                    .task_map()
+                    .values()
+                    .find(|meta| meta.channel_id == Some(channel_id))
+                    .and_then(|meta| meta.routing_channel_id)
+                    .unwrap_or(channel_id);
                 tracing::info!(
-                    channel = %channel_id,
+                    channel = %routing_channel_id,
+                    conversation = %channel_id,
                     event_id = %event_id,
                     ?ack,
                     release_withheld,
@@ -2541,7 +2618,12 @@ async fn tokio_main() -> Result<()> {
                     // front of `queues[channel_id]`, so the cancel
                     // will pick it up as part of the merged batch and
                     // re-prompt the agent.
-                    signal_in_flight_task(&mut pool, channel_id, ControlSignal::Steer);
+                    signal_in_flight_task(
+                        &mut pool,
+                        channel_id,
+                        routing_channel_id,
+                        ControlSignal::Steer,
+                    );
                 }
                 // After releasing a withheld event, give dispatch a chance
                 // to re-flush. If the prompt is still in flight, the
@@ -2779,17 +2861,54 @@ fn mode_gate_signal(
 /// Returns `true` if a signal was sent, `false` if no in-flight task was found.
 fn signal_in_flight_task(
     pool: &mut AgentPool,
-    channel_id: uuid::Uuid,
+    conversation_id: uuid::Uuid,
+    routing_channel_id: uuid::Uuid,
     mode: ControlSignal,
 ) -> bool {
-    let entry = pool
-        .task_map_mut()
-        .values_mut()
-        .find(|m| m.channel_id == Some(channel_id));
+    let exact_task = pool
+        .task_map()
+        .iter()
+        .find_map(|(task_id, meta)| (meta.channel_id == Some(conversation_id)).then_some(*task_id));
+    let task_id = exact_task.or_else(|| {
+        let mut routing_matches = pool.task_map().iter().filter_map(|(task_id, meta)| {
+            (meta.routing_channel_id == Some(routing_channel_id)).then_some(*task_id)
+        });
+        let first = routing_matches.next()?;
+        if routing_matches.next().is_some() {
+            tracing::warn!(
+                channel = %routing_channel_id,
+                conversation = %conversation_id,
+                "control signal is ambiguous across multiple in-flight threads"
+            );
+            return None;
+        }
+        Some(first)
+    });
 
-    if let Some(meta) = entry {
+    if let Some(meta) = task_id.and_then(|task_id| pool.task_map_mut().get_mut(&task_id)) {
         if let Some(tx) = meta.control_tx.take() {
-            tracing::info!(channel = %channel_id, ?mode, "control signal sent to in-flight task");
+            tracing::info!(
+                channel = %routing_channel_id,
+                conversation = %conversation_id,
+                ?mode,
+                "control signal sent to in-flight task"
+            );
+            let _ = tx.send(mode);
+            return true;
+        }
+    }
+    false
+}
+
+/// Send a control signal to one exact observer turn.
+fn signal_in_flight_turn(pool: &mut AgentPool, turn_id: &str, mode: ControlSignal) -> bool {
+    let task_id = pool
+        .task_map()
+        .iter()
+        .find_map(|(task_id, meta)| (meta.turn_id == turn_id).then_some(*task_id));
+    if let Some(meta) = task_id.and_then(|task_id| pool.task_map_mut().get_mut(&task_id)) {
+        if let Some(tx) = meta.control_tx.take() {
+            tracing::info!(turn = %turn_id, ?mode, "control signal sent to exact in-flight turn");
             let _ = tx.send(mode);
             return true;
         }
@@ -2824,7 +2943,8 @@ fn signal_in_flight_task(
 fn try_native_steer(
     pool: &mut AgentPool,
     queue: &mut EventQueue,
-    channel_id: uuid::Uuid,
+    conversation_id: uuid::Uuid,
+    routing_channel_id: uuid::Uuid,
     event: nostr::Event,
     prompt_tag: String,
     steer_ack_tx: &mpsc::UnboundedSender<SteerAckEvent>,
@@ -2849,7 +2969,7 @@ fn try_native_steer(
         prompt_tag: prompt_tag.clone(),
         received_at: std::time::Instant::now(),
     };
-    let event_block = queue::format_event_block(channel_id, None, &be, None);
+    let event_block = queue::format_event_block(routing_channel_id, None, &be, None);
     let body = format!("{header}\n\n[Buzz event: {prompt_tag}]\n{event_block}\n\n{closing}");
 
     let (ack_tx, ack_rx) = tokio::sync::oneshot::channel::<pool::SteerAck>();
@@ -2858,14 +2978,14 @@ fn try_native_steer(
         ack_tx,
     };
 
-    match pool.send_steer(channel_id, request) {
+    match pool.send_steer(conversation_id, request) {
         Ok(()) => {
             // Withhold the queued event synchronously BEFORE spawning
             // the watcher: this closes the race where `mark_complete`
             // clears `in_flight_channels` and a stray `flush_next` could
             // re-deliver the event via normal dispatch. See
             // `EventQueue::mark_native_steer_pending` docs at queue.rs:606.
-            let withheld = queue.mark_native_steer_pending(channel_id, &event_id_hex);
+            let withheld = queue.mark_native_steer_pending(conversation_id, &event_id_hex);
             if !withheld {
                 // Race: the event was already drained out of the queue
                 // before we got here (e.g. a concurrent flush picked it
@@ -2875,7 +2995,8 @@ fn try_native_steer(
                 // the same message twice). Log so this is visible if it
                 // ever happens in production.
                 tracing::warn!(
-                    channel = %channel_id,
+                    channel = %routing_channel_id,
+                    conversation = %conversation_id,
                     event_id = %event_id_hex,
                     "native steer accepted by read loop but event was not in queue to withhold \
                      — possible duplicate delivery if steer succeeds"
@@ -2886,7 +3007,7 @@ fn try_native_steer(
             tokio::spawn(async move {
                 let ack = ack_rx.await;
                 let _ = ack_tx_clone.send(SteerAckEvent {
-                    channel_id,
+                    channel_id: conversation_id,
                     event_id: event_id_for_watcher,
                     ack,
                 });
@@ -2895,7 +3016,8 @@ fn try_native_steer(
         }
         Err(e) => {
             tracing::info!(
-                channel = %channel_id,
+                channel = %routing_channel_id,
+                conversation = %conversation_id,
                 error = ?e,
                 "non-cancelling steer not accepted — falling back to cancel+merge"
             );
@@ -2911,7 +3033,7 @@ fn dispatch_pending(
     pool: &mut AgentPool,
     queue: &mut EventQueue,
     ctx: &Arc<PromptContext>,
-) -> Vec<(Uuid, ThreadTags)> {
+) -> Vec<(Uuid, TypingState)> {
     let mut dispatched_channels = Vec::new();
     loop {
         let batch = match queue.flush_next() {
@@ -2919,6 +3041,7 @@ fn dispatch_pending(
             None => break,
         };
         let channel_id = batch.channel_id;
+        let routing_channel_id = batch.routing_channel_id();
         let typing_scope = batch
             .events
             .last()
@@ -2983,13 +3106,20 @@ fn dispatch_pending(
             pool::TaskMeta {
                 agent_index,
                 channel_id: Some(channel_id),
+                routing_channel_id: Some(routing_channel_id),
                 turn_id,
                 recoverable_batch,
                 control_tx: Some(control_tx),
                 steer_tx,
             },
         );
-        dispatched_channels.push((channel_id, typing_scope));
+        dispatched_channels.push((
+            channel_id,
+            TypingState {
+                routing_channel_id,
+                thread_tags: typing_scope,
+            },
+        ));
     }
     tracing::debug!(
         dispatched = dispatched_channels.len(),
@@ -3044,7 +3174,7 @@ fn spawn_failure_notice(
             .map(|be| queue::parse_thread_tags(&be.event))
             .unwrap_or_default();
         let rest = rest.clone();
-        let channel_id = batch.channel_id;
+        let channel_id = batch.routing_channel_id();
         tokio::spawn(async move {
             pool::post_failure_notice(&rest, channel_id, &thread_tags, &content).await;
         });
@@ -3067,6 +3197,11 @@ fn handle_prompt_result(
 ) -> LoopAction {
     let before = pool.task_map().len();
     let agent_index = result.agent.index;
+    let result_routing_channel_id = result.batch.as_ref().map(FlushBatch::routing_channel_id);
+    let result_conversation_id = match &result.source {
+        PromptSource::Channel(id) => Some(*id),
+        PromptSource::Heartbeat => None,
+    };
     pool.task_map_mut()
         .retain(|_, meta| meta.agent_index != agent_index);
     debug_assert_eq!(before, pool.task_map().len() + 1);
@@ -3086,9 +3221,10 @@ fn handle_prompt_result(
     // every retry starts at attempt 1 — defeating exponential backoff and
     // dead-letter protection.
     if let Some(batch) = result.batch.take() {
+        let routing_channel_id = batch.routing_channel_id();
         // Don't requeue batches for channels the agent was removed from —
         // those events are stale and should be silently dropped.
-        if !removed_channels.contains(&batch.channel_id) {
+        if !removed_channels.contains(&routing_channel_id) {
             if matches!(
                 result.outcome,
                 PromptOutcome::Cancelled | PromptOutcome::CancelDrainTimeout(_)
@@ -3116,7 +3252,7 @@ fn handle_prompt_result(
                 })
             ) {
                 tracing::error!(
-                    channel_id = %batch.channel_id,
+                    channel_id = %routing_channel_id,
                     events = batch.events.len(),
                     "dead-lettering batch after hard-cap timeout (no recent activity) — discarding {} events",
                     batch.events.len(),
@@ -3134,7 +3270,7 @@ fn handle_prompt_result(
                 })
             ) {
                 tracing::warn!(
-                    channel_id = %batch.channel_id,
+                    channel_id = %routing_channel_id,
                     events = batch.events.len(),
                     "hard-cap timeout with recent activity — requeueing for retry"
                 );
@@ -3154,7 +3290,7 @@ fn handle_prompt_result(
                 // delays the visible failure. Dead-letter immediately and tell
                 // the user to re-authenticate the CLI.
                 tracing::warn!(
-                    channel_id = %batch.channel_id,
+                    channel_id = %routing_channel_id,
                     events = batch.events.len(),
                     "dead-lettering batch immediately — non-retryable auth error"
                 );
@@ -3180,7 +3316,7 @@ fn handle_prompt_result(
             }
         } else {
             tracing::debug!(
-                channel_id = %batch.channel_id,
+                channel_id = %routing_channel_id,
                 events = batch.events.len(),
                 "dropping failed batch for removed channel"
             );
@@ -3198,6 +3334,7 @@ fn handle_prompt_result(
     // only touches idle agents.
     for ch in removed_channels {
         result.agent.state.invalidate_channel(ch);
+        result.agent.state.invalidate_routing_channel(*ch);
     }
 
     let outcome_label = match &result.outcome {
@@ -3224,10 +3361,7 @@ fn handle_prompt_result(
         .to_string();
     let harness_pid = std::process::id();
 
-    let channel_id = match &result.source {
-        PromptSource::Channel(ch) => Some(*ch),
-        PromptSource::Heartbeat => None,
-    };
+    let channel_id = result_routing_channel_id;
     let turn_id = result.turn_id.clone();
     let emit_turn_error = |error_msg: &str, error_code: Option<i64>| {
         if let Some(ref observer) = observer {
@@ -3241,7 +3375,12 @@ fn handle_prompt_result(
             observer.emit(
                 "turn_error",
                 Some(agent_index),
-                &observer::context_for(channel_id, None, Some(turn_id.clone())),
+                &observer::context_for_conversation(
+                    channel_id,
+                    result_conversation_id,
+                    None,
+                    Some(turn_id.clone()),
+                ),
                 payload,
             );
         }
@@ -3427,7 +3566,7 @@ fn recover_panicked_agent(
     join_error: tokio::task::JoinError,
     heartbeat_in_flight: &mut bool,
     removed_channels: &HashSet<Uuid>,
-    typing_channels: &mut HashMap<Uuid, ThreadTags>,
+    typing_channels: &mut HashMap<Uuid, TypingState>,
     crash_history: &mut [SlotCircuit],
     respawn_tx: &mpsc::Sender<RespawnResult>,
     respawn_tasks: &mut tokio::task::JoinSet<()>,
@@ -3443,14 +3582,17 @@ fn recover_panicked_agent(
     // Requeue BEFORE mark_complete (same rationale as handle_prompt_result).
     if let Some(batch) = meta.recoverable_batch {
         if let Some(ch) = meta.channel_id {
-            if !removed_channels.contains(&ch) {
+            if !meta
+                .routing_channel_id
+                .is_some_and(|routing| removed_channels.contains(&routing))
+            {
                 // Dead-letter on exhaustion is logged inside requeue(); a
                 // panic path has no outcome to report, so no notice here.
                 let _ = queue.requeue(batch);
                 tracing::warn!("requeued batch for panicked agent {i}");
             } else {
                 tracing::debug!(
-                    channel_id = %ch,
+                    channel_id = %meta.routing_channel_id.unwrap_or(ch),
                     "dropping panicked batch for removed channel"
                 );
             }
@@ -3470,7 +3612,12 @@ fn recover_panicked_agent(
         observer.emit(
             "agent_panic",
             Some(i),
-            &observer::context_for(meta.channel_id, None, Some(meta.turn_id)),
+            &observer::context_for_conversation(
+                meta.routing_channel_id,
+                meta.channel_id,
+                None,
+                Some(meta.turn_id),
+            ),
             serde_json::json!({
                 "outcome": "panic",
                 "error": format!("Agent task panicked: {join_error}"),
@@ -3525,7 +3672,7 @@ fn drain_ready_join_results(
     config: &Config,
     heartbeat_in_flight: &mut bool,
     removed_channels: &HashSet<Uuid>,
-    typing_channels: &mut HashMap<Uuid, ThreadTags>,
+    typing_channels: &mut HashMap<Uuid, TypingState>,
     crash_history: &mut [SlotCircuit],
     respawn_tx: &mpsc::Sender<RespawnResult>,
     respawn_tasks: &mut tokio::task::JoinSet<()>,
@@ -3596,6 +3743,7 @@ fn dispatch_heartbeat(
         pool::TaskMeta {
             agent_index,
             channel_id: None,
+            routing_channel_id: None,
             turn_id,
             recoverable_batch: None,
             control_tx: None,
@@ -4385,6 +4533,7 @@ mod owner_control_command_tests {
             pool::TaskMeta {
                 agent_index: 0,
                 channel_id: Some(channel_id),
+                routing_channel_id: Some(channel_id),
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: Some(control_tx),
@@ -4395,20 +4544,113 @@ mod owner_control_command_tests {
         assert!(!signal_in_flight_task(
             &mut pool,
             other_channel_id,
+            other_channel_id,
             ControlSignal::Rotate
         ));
         assert!(signal_in_flight_task(
             &mut pool,
             channel_id,
+            channel_id,
             ControlSignal::Rotate
         ));
         assert_eq!(control_rx.await.unwrap(), ControlSignal::Rotate);
         assert!(!signal_in_flight_task(
             &mut pool,
             channel_id,
+            channel_id,
             ControlSignal::Rotate
         ));
     }
+
+    #[tokio::test]
+    async fn signal_in_flight_task_rejects_ambiguous_channel_control() {
+        let mut pool = AgentPool::from_slots(vec![]);
+        let routing_channel_id = Uuid::new_v4();
+        let first_conversation_id = Uuid::new_v4();
+        let second_conversation_id = Uuid::new_v4();
+        let (first_tx, mut first_rx) = tokio::sync::oneshot::channel();
+        let (second_tx, mut second_rx) = tokio::sync::oneshot::channel();
+
+        for (conversation_id, control_tx) in [
+            (first_conversation_id, first_tx),
+            (second_conversation_id, second_tx),
+        ] {
+            let abort_handle = pool.join_set.spawn(async {});
+            pool.task_map_mut().insert(
+                abort_handle.id(),
+                pool::TaskMeta {
+                    agent_index: 0,
+                    channel_id: Some(conversation_id),
+                    routing_channel_id: Some(routing_channel_id),
+                    turn_id: format!("turn-{conversation_id}"),
+                    recoverable_batch: None,
+                    control_tx: Some(control_tx),
+                    steer_tx: None,
+                },
+            );
+        }
+
+        assert!(!signal_in_flight_task(
+            &mut pool,
+            routing_channel_id,
+            routing_channel_id,
+            ControlSignal::Cancel
+        ));
+        assert!(matches!(
+            first_rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+        assert!(matches!(
+            second_rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+
+        assert!(signal_in_flight_task(
+            &mut pool,
+            first_conversation_id,
+            routing_channel_id,
+            ControlSignal::Cancel
+        ));
+        assert_eq!(first_rx.await.unwrap(), ControlSignal::Cancel);
+    }
+
+    #[tokio::test]
+    async fn signal_in_flight_turn_controls_only_the_selected_thread() {
+        let mut pool = AgentPool::from_slots(vec![]);
+        let routing_channel_id = Uuid::new_v4();
+        let (first_tx, mut first_rx) = tokio::sync::oneshot::channel();
+        let (second_tx, second_rx) = tokio::sync::oneshot::channel();
+
+        for (turn_id, control_tx) in [("turn-a", first_tx), ("turn-b", second_tx)] {
+            let abort_handle = pool.join_set.spawn(async {});
+            pool.task_map_mut().insert(
+                abort_handle.id(),
+                pool::TaskMeta {
+                    agent_index: 0,
+                    channel_id: Some(Uuid::new_v4()),
+                    routing_channel_id: Some(routing_channel_id),
+                    turn_id: turn_id.to_string(),
+                    recoverable_batch: None,
+                    control_tx: Some(control_tx),
+                    steer_tx: None,
+                },
+            );
+        }
+
+        assert!(signal_in_flight_turn(
+            &mut pool,
+            "turn-b",
+            ControlSignal::SwitchModel("opus".into())
+        ));
+        assert!(matches!(
+            first_rx.try_recv(),
+            Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+        ));
+        assert_eq!(
+            second_rx.await.unwrap(),
+            ControlSignal::SwitchModel("opus".into())
+        );
+    }
 }
 
 #[cfg(test)]
@@ -4921,6 +5163,7 @@ mod observer_chunk_coalescer_tests {
             kind: "acp_read".to_string(),
             agent_index: Some(0),
             channel_id: Some("channel-1".to_string()),
+            conversation_id: None,
             session_id: Some("session-1".to_string()),
             turn_id: Some("turn-1".to_string()),
             started_at: None,
@@ -4949,6 +5192,7 @@ mod observer_chunk_coalescer_tests {
             kind: "turn_started".to_string(),
             agent_index: Some(0),
             channel_id: Some("channel-1".to_string()),
+            conversation_id: None,
             session_id: Some("session-1".to_string()),
             turn_id: Some("turn-1".to_string()),
             started_at: None,
@@ -5330,6 +5574,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -5406,6 +5651,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: Some(channel_id),
+                routing_channel_id: Some(channel_id),
                 turn_id: "panic-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -5498,6 +5744,7 @@ mod error_outcome_emission_tests {
                 crate::pool::TaskMeta {
                     agent_index: 0,
                     channel_id: None,
+                    routing_channel_id: None,
                     turn_id: "test-turn-id".to_string(),
                     recoverable_batch: None,
                     control_tx: None,
@@ -5589,6 +5836,7 @@ mod error_outcome_emission_tests {
                 crate::pool::TaskMeta {
                     agent_index: 0,
                     channel_id: None,
+                    routing_channel_id: None,
                     turn_id: "test-turn-id".to_string(),
                     recoverable_batch: None,
                     control_tx: None,
@@ -5694,6 +5942,7 @@ mod error_outcome_emission_tests {
                 crate::pool::TaskMeta {
                     agent_index: 0,
                     channel_id: None,
+                    routing_channel_id: None,
                     turn_id: "test-turn-id".to_string(),
                     recoverable_batch: None,
                     control_tx: None,
@@ -5770,6 +6019,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -5864,6 +6114,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -5980,6 +6231,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -6119,6 +6371,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -6307,6 +6560,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -6392,6 +6646,7 @@ mod error_outcome_emission_tests {
             crate::pool::TaskMeta {
                 agent_index: 0,
                 channel_id: None,
+                routing_channel_id: None,
                 turn_id: "test-turn-id".to_string(),
                 recoverable_batch: None,
                 control_tx: None,
@@ -6455,6 +6710,7 @@ mod observer_payload_trim_tests {
             kind: kind.to_string(),
             agent_index: Some(0),
             channel_id: Some("11111111-1111-1111-1111-111111111111".to_string()),
+            conversation_id: None,
             session_id: Some("sess-1".to_string()),
             turn_id: Some("turn-1".to_string()),
             started_at: None,
diff --git a/crates/buzz-acp/src/observer.rs b/crates/buzz-acp/src/observer.rs
index 7029e5af6d5..d78ca64e1b8 100644
--- a/crates/buzz-acp/src/observer.rs
+++ b/crates/buzz-acp/src/observer.rs
@@ -22,6 +22,9 @@ const OBSERVER_BUFFER_CAP: usize = 1_000;
 pub struct ObserverContext {
     /// Buzz channel UUID for the current turn, when channel-scoped.
     pub channel_id: Option<String>,
+    /// Scheduler/session identity. For channel work this identifies one
+    /// top-level conversation or thread independently of the real channel.
+    pub conversation_id: Option<String>,
     /// ACP session ID associated with the current turn, once known.
     pub session_id: Option<String>,
     /// Local UUID for one prompt turn.
@@ -67,6 +70,8 @@ pub struct ObserverEvent {
     pub agent_index: Option<usize>,
     /// Buzz channel UUID for channel-scoped events.
     pub channel_id: Option<String>,
+    /// Scheduler/session identity for the exact conversation or thread.
+    pub conversation_id: Option<String>,
     /// ACP session ID when known.
     pub session_id: Option<String>,
     /// Local UUID for one prompt turn.
@@ -114,6 +119,7 @@ impl ObserverHandle {
             kind: kind.into(),
             agent_index,
             channel_id: context.channel_id.clone(),
+            conversation_id: context.conversation_id.clone(),
             session_id: context.session_id.clone(),
             turn_id: context.turn_id.clone(),
             started_at: context.started_at.clone(),
@@ -137,6 +143,7 @@ impl ObserverHandle {
 }
 
 /// Build observer context values from optional channel/session/turn IDs.
+#[cfg(test)]
 pub fn context_for(
     channel_id: Option<uuid::Uuid>,
     session_id: Option<String>,
@@ -144,6 +151,7 @@ pub fn context_for(
 ) -> ObserverContext {
     ObserverContext {
         channel_id: channel_id.map(|id| id.to_string()),
+        conversation_id: None,
         session_id,
         turn_id,
         started_at: None,
@@ -151,6 +159,7 @@ pub fn context_for(
 }
 
 /// Attach the authoritative start timestamp to every observer frame for a turn.
+#[cfg(test)]
 pub fn context_for_turn(
     channel_id: Option<uuid::Uuid>,
     session_id: Option<String>,
@@ -159,8 +168,42 @@ pub fn context_for_turn(
 ) -> ObserverContext {
     ObserverContext {
         channel_id: channel_id.map(|id| id.to_string()),
+        conversation_id: None,
         session_id,
         turn_id: Some(turn_id),
         started_at: Some(started_at),
     }
 }
+
+/// Build observer context for one exact channel conversation and turn.
+pub fn context_for_conversation_turn(
+    channel_id: Option<uuid::Uuid>,
+    conversation_id: Option<uuid::Uuid>,
+    session_id: Option<String>,
+    turn_id: String,
+    started_at: String,
+) -> ObserverContext {
+    ObserverContext {
+        channel_id: channel_id.map(|id| id.to_string()),
+        conversation_id: conversation_id.map(|id| id.to_string()),
+        session_id,
+        turn_id: Some(turn_id),
+        started_at: Some(started_at),
+    }
+}
+
+/// Build observer context for one exact channel conversation.
+pub fn context_for_conversation(
+    channel_id: Option<uuid::Uuid>,
+    conversation_id: Option<uuid::Uuid>,
+    session_id: Option<String>,
+    turn_id: Option<String>,
+) -> ObserverContext {
+    ObserverContext {
+        channel_id: channel_id.map(|id| id.to_string()),
+        conversation_id: conversation_id.map(|id| id.to_string()),
+        session_id,
+        turn_id,
+        started_at: None,
+    }
+}
diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs
index d1e005cbcce..495aa9bd44a 100644
--- a/crates/buzz-acp/src/pool.rs
+++ b/crates/buzz-acp/src/pool.rs
@@ -50,7 +50,10 @@ const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60);
 /// Metadata stored per in-flight task for panic recovery.
 pub struct TaskMeta {
     pub agent_index: usize,
+    /// Scheduler/session identity. For channel work this identifies a thread.
     pub channel_id: Option<Uuid>,
+    /// Real NIP-29 channel used for relay operations and observer context.
+    pub routing_channel_id: Option<Uuid>,
     /// Identifies terminal events when the task panics before returning a result.
     pub turn_id: String,
     /// Clone of batch for Queue mode panic recovery.
@@ -102,6 +105,8 @@ 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>,
+    /// Conversation identity → real NIP-29 channel.
+    pub routing_channels: HashMap<Uuid, Uuid>,
 }
 
 impl SessionState {
@@ -124,9 +129,25 @@ impl SessionState {
         self.turn_counts.remove(channel_id);
         self.core_sections.remove(channel_id);
         self.canvas_sections.remove(channel_id);
+        self.routing_channels.remove(channel_id);
         self.sessions.remove(channel_id).is_some()
     }
 
+    /// Invalidate every thread session belonging to a real channel.
+    pub fn invalidate_routing_channel(&mut self, routing_channel_id: Uuid) -> usize {
+        let conversations: Vec<Uuid> = self
+            .routing_channels
+            .iter()
+            .filter_map(|(conversation, routing)| {
+                (*routing == routing_channel_id).then_some(*conversation)
+            })
+            .collect();
+        conversations
+            .iter()
+            .filter(|conversation| self.invalidate_channel(conversation))
+            .count()
+    }
+
     /// Invalidate all sessions and turn counters (e.g. after agent exit).
     pub fn invalidate_all(&mut self) {
         self.sessions.clear();
@@ -135,6 +156,7 @@ impl SessionState {
         self.heartbeat_turn_count = 0;
         self.core_sections.clear();
         self.canvas_sections.clear();
+        self.routing_channels.clear();
     }
 
     #[cfg(test)]
@@ -143,6 +165,7 @@ impl SessionState {
             || self.turn_counts.contains_key(channel_id)
             || self.core_sections.contains_key(channel_id)
             || self.canvas_sections.contains_key(channel_id)
+            || self.routing_channels.contains_key(channel_id)
     }
 }
 
@@ -729,9 +752,8 @@ impl AgentPool {
         let mut count = 0;
         for slot in &mut self.agents {
             if let Some(agent) = slot.as_mut() {
-                if agent.state.invalidate_channel(&channel_id) {
-                    count += 1;
-                }
+                count += usize::from(agent.state.invalidate_channel(&channel_id));
+                count += agent.state.invalidate_routing_channel(channel_id);
             }
         }
         count
@@ -755,12 +777,14 @@ impl AgentPool {
         channel_id: Uuid,
         model_id: &str,
     ) -> IdleSwitchResult {
-        let Some(agent) = self
-            .agents
-            .iter_mut()
-            .flatten()
-            .find(|a| a.state.sessions.contains_key(&channel_id))
-        else {
+        let Some(agent) = self.agents.iter_mut().flatten().find(|agent| {
+            agent.state.sessions.contains_key(&channel_id)
+                || agent
+                    .state
+                    .routing_channels
+                    .values()
+                    .any(|routing| *routing == channel_id)
+        }) else {
             return IdleSwitchResult::NoIdleAgent;
         };
 
@@ -779,6 +803,7 @@ impl AgentPool {
         agent.desired_model = Some(model_id.to_string());
         agent.model_overridden = true;
         agent.state.invalidate_channel(&channel_id);
+        agent.state.invalidate_routing_channel(channel_id);
         IdleSwitchResult::Switched
     }
 }
@@ -868,6 +893,7 @@ async fn resolve_new_session_channel_context(
 async fn create_session_and_apply_model(
     agent: &mut OwnedAgent,
     ctx: &PromptContext,
+    session_cwd: &str,
     agent_core: Option<&str>,
     agent_canvas: Option<&str>,
     channel_name: Option<&str>,
@@ -882,7 +908,7 @@ async fn create_session_and_apply_model(
     let combined_system_prompt = with_canvas(
         with_core(
             with_team(
-                framed_system_prompt(&ctx.cwd, ctx.base_prompt, ctx.system_prompt.as_deref()),
+                framed_system_prompt(session_cwd, ctx.base_prompt, ctx.system_prompt.as_deref()),
                 ctx.team_instructions.as_deref(),
             ),
             agent_core,
@@ -898,7 +924,7 @@ async fn create_session_and_apply_model(
     let resp = agent
         .acp
         .session_new_full(
-            &ctx.cwd,
+            session_cwd,
             ctx.mcp_servers.clone(),
             session_new_system_prompt(
                 is_goose,
@@ -1347,17 +1373,18 @@ pub async fn run_prompt_task(
         Some(b) => PromptSource::Channel(b.channel_id),
         None => PromptSource::Heartbeat,
     };
-    let observer_channel_id = match &source {
-        PromptSource::Channel(channel_id) => Some(*channel_id),
-        PromptSource::Heartbeat => None,
-    };
+    let observer_channel_id = batch.as_ref().map(FlushBatch::routing_channel_id);
+    let observer_conversation_id = batch.as_ref().map(|batch| batch.channel_id);
     let turn_started_at = chrono::Utc::now().to_rfc3339();
-    agent.acp.set_observer_context(observer::context_for_turn(
-        observer_channel_id,
-        None,
-        turn_id.clone(),
-        turn_started_at.clone(),
-    ));
+    agent
+        .acp
+        .set_observer_context(observer::context_for_conversation_turn(
+            observer_channel_id,
+            observer_conversation_id,
+            None,
+            turn_id.clone(),
+            turn_started_at.clone(),
+        ));
     let triggering_event_ids: Vec<String> = batch
         .as_ref()
         .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect())
@@ -1381,6 +1408,7 @@ pub async fn run_prompt_task(
         agent.acp.observer_handle(),
         agent.acp.observer_agent_index(),
         observer_channel_id,
+        observer_conversation_id,
         turn_id.clone(),
     );
 
@@ -1400,8 +1428,9 @@ pub async fn run_prompt_task(
     let liveness = run_turn_liveness(
         agent.acp.observer_handle(),
         agent.acp.observer_agent_index(),
-        observer::context_for_turn(
+        observer::context_for_conversation_turn(
             observer_channel_id,
+            observer_conversation_id,
             None,
             turn_id.clone(),
             turn_started_at.clone(),
@@ -1421,6 +1450,26 @@ pub async fn run_prompt_task(
         .unwrap_or_default();
     let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone());
 
+    let session_cwd = match (&source, &batch) {
+        (PromptSource::Channel(cid), Some(batch)) if !agent.state.sessions.contains_key(cid) => {
+            match resolve_thread_session_cwd(batch, &ctx).await {
+                Ok(cwd) => cwd,
+                Err(error) => {
+                    send_prompt_result(
+                        &result_tx,
+                        &turn_id,
+                        agent,
+                        source,
+                        PromptOutcome::Error(AcpError::Protocol(error.to_string())),
+                        requeue_batch_if_queue(&ctx, Some(batch.clone())),
+                    );
+                    return;
+                }
+            }
+        }
+        _ => ctx.cwd.clone(),
+    };
+
     //
     // Core memory is delivered inside the system prompt the harness already
     // builds (system role for protocol >= 2, the `[System]` user-message
@@ -1502,17 +1551,20 @@ pub async fn run_prompt_task(
     // canvas DM check uses — see `resolve_new_session_channel_context`.
     let mut title_channel: Option<String> = None;
     if let PromptSource::Channel(cid) = &source {
+        let routing_channel_id = observer_channel_id.unwrap_or(*cid);
         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);
         let needs_title = is_new_channel_session && ctx.session_title.is_some();
         if needs_canvas || needs_title {
             let (is_dm, resolved_channel) =
-                resolve_new_session_channel_context(&ctx.channel_info, *cid).await;
+                resolve_new_session_channel_context(&ctx.channel_info, routing_channel_id).await;
             title_channel = resolved_channel;
             // A confirmed DM never receives a canvas section; an undeterminable
             // channel type fails closed as a DM for the same reason.
             if needs_canvas && !is_dm {
-                if let Some(section) = fetch_canvas_section(*cid, &ctx.rest_client).await {
+                if let Some(section) =
+                    fetch_canvas_section(routing_channel_id, &ctx.rest_client).await
+                {
                     pending_canvas = Some((*cid, section));
                 }
             }
@@ -1550,6 +1602,7 @@ pub async fn run_prompt_task(
                 match create_session_and_apply_model(
                     &mut agent,
                     &ctx,
+                    &session_cwd,
                     agent_core.as_deref(),
                     agent_canvas.as_deref(),
                     title_channel.as_deref(),
@@ -1562,6 +1615,10 @@ pub async fn run_prompt_task(
                             "created session {sid} for channel {cid}"
                         );
                         agent.state.sessions.insert(*cid, sid.clone());
+                        agent
+                            .state
+                            .routing_channels
+                            .insert(*cid, observer_channel_id.unwrap_or(*cid));
                         // 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);
@@ -1600,7 +1657,9 @@ pub async fn run_prompt_task(
             if let Some(sid) = &agent.state.heartbeat_session {
                 (sid.clone(), false)
             } else {
-                match create_session_and_apply_model(&mut agent, &ctx, None, None, None).await {
+                match create_session_and_apply_model(&mut agent, &ctx, &ctx.cwd, None, None, None)
+                    .await
+                {
                     Ok(sid) => {
                         tracing::info!(
                             target: "pool::session",
@@ -1637,12 +1696,15 @@ pub async fn run_prompt_task(
             }
         }
     };
-    agent.acp.set_observer_context(observer::context_for_turn(
-        observer_channel_id,
-        Some(session_id.clone()),
-        turn_id.clone(),
-        turn_started_at,
-    ));
+    agent
+        .acp
+        .set_observer_context(observer::context_for_conversation_turn(
+            observer_channel_id,
+            observer_conversation_id,
+            Some(session_id.clone()),
+            turn_id.clone(),
+            turn_started_at,
+        ));
     // Backfill liveness's shared session ID so ticks after this point carry
     // it too, matching every other observer frame for this turn.
     liveness_guard.set_session_id(session_id.clone());
@@ -1818,7 +1880,7 @@ pub async fn run_prompt_task(
     } else if let Some(ref b) = batch {
         // Build prompt from batch with context enrichment.
         // Try startup cache first; lazy-fetch via REST for dynamic channels.
-        let channel_info = ctx.channel_info.resolve(b.channel_id).await;
+        let channel_info = ctx.channel_info.resolve(b.routing_channel_id()).await;
 
         let conversation_context = if ctx.context_message_limit > 0 {
             fetch_conversation_context(b, &channel_info, &ctx).await
@@ -1839,7 +1901,7 @@ pub async fn run_prompt_task(
         if let Some(ref cmd) = slash_command {
             tracing::info!(
                 target: "pool::prompt",
-                channel = %b.channel_id,
+                channel = %b.routing_channel_id(),
                 command = %cmd,
                 "slash-command pass-through"
             );
@@ -2600,17 +2662,156 @@ async fn fetch_conversation_context(
     let last_event = batch.events.last()?;
     let tags = crate::queue::parse_thread_tags(&last_event.event);
     if let Some(root_id) = tags.root_event_id {
-        return fetch_thread_context(batch.channel_id, &root_id, limit, &ctx.rest_client).await;
+        return fetch_thread_context(
+            batch.routing_channel_id(),
+            &root_id,
+            limit,
+            &ctx.rest_client,
+        )
+        .await;
     }
 
     // DM non-reply: fetch recent conversation history.
     if is_dm {
-        return fetch_dm_context(batch.channel_id, limit, &ctx.rest_client).await;
+        return fetch_dm_context(batch.routing_channel_id(), limit, &ctx.rest_client).await;
     }
 
     None
 }
 
+async fn resolve_thread_session_cwd(
+    batch: &FlushBatch,
+    ctx: &PromptContext,
+) -> anyhow::Result<String> {
+    let Some(last_event) = batch.events.last() else {
+        return Ok(ctx.cwd.clone());
+    };
+    let owner = ctx
+        .agent_owner_pubkey
+        .as_ref()
+        .map(nostr::PublicKey::to_hex);
+    let thread_root_id = crate::queue::parse_thread_tags(&last_event.event).root_event_id;
+    let root_event_id = thread_root_id
+        .clone()
+        .unwrap_or_else(|| last_event.event.id.to_hex());
+
+    let fetched_root = if owner.is_some() && thread_root_id.is_some() {
+        // Once a thread exists, its root freezes the repository authority.
+        // Never trust a later reply's freshly-resolved Project path: the
+        // Project may have been relinked between agent handoffs.
+        fetch_thread_context(
+            batch.routing_channel_id(),
+            &root_event_id,
+            1,
+            &ctx.rest_client,
+        )
+        .await
+    } else {
+        None
+    };
+    let trusted_content = if let Some(owner) = owner.as_deref() {
+        select_trusted_workspace_content(
+            owner,
+            &last_event.event.pubkey.to_hex(),
+            &last_event.event.content,
+            thread_root_id.is_some(),
+            fetched_root.as_ref(),
+        )?
+    } else {
+        None
+    };
+
+    let Some(content) = trusted_content else {
+        return Ok(ctx.cwd.clone());
+    };
+    let Some(workspace) = crate::thread_workspace::parse_project_workspace(content)? else {
+        return Ok(ctx.cwd.clone());
+    };
+    let worktree =
+        crate::thread_workspace::ensure_thread_worktree(&workspace, &root_event_id).await?;
+    tracing::info!(
+        channel = %batch.routing_channel_id(),
+        root = %root_event_id,
+        cwd = %worktree.display(),
+        "resolved isolated thread worktree"
+    );
+    Ok(worktree.to_string_lossy().into_owned())
+}
+
+fn trusted_fetched_root_content<'a>(
+    owner: &str,
+    context: Option<&'a ConversationContext>,
+) -> anyhow::Result<Option<&'a str>> {
+    let Some(context) = context else {
+        anyhow::bail!("could not verify thread root before creating an agent session");
+    };
+    let ConversationContext::Thread { messages, .. } = context else {
+        return Ok(None);
+    };
+    let Some(root) = messages.first() else {
+        anyhow::bail!("thread root lookup returned no messages");
+    };
+    Ok(root
+        .pubkey
+        .eq_ignore_ascii_case(owner)
+        .then_some(root.content.as_str()))
+}
+
+fn select_trusted_workspace_content<'a>(
+    owner: &str,
+    direct_author: &str,
+    direct_content: &'a str,
+    is_thread_reply: bool,
+    fetched_root: Option<&'a ConversationContext>,
+) -> anyhow::Result<Option<&'a str>> {
+    if is_thread_reply {
+        return trusted_fetched_root_content(owner, fetched_root);
+    }
+    Ok(direct_author
+        .eq_ignore_ascii_case(owner)
+        .then_some(direct_content))
+}
+
+#[cfg(test)]
+mod thread_session_cwd_tests {
+    use super::{select_trusted_workspace_content, trusted_fetched_root_content};
+    use crate::queue::{ContextMessage, ConversationContext};
+
+    #[test]
+    fn missing_root_context_fails_closed() {
+        let error = trusted_fetched_root_content("owner", None).unwrap_err();
+        assert!(error.to_string().contains("could not verify thread root"));
+    }
+
+    #[test]
+    fn handoff_reply_cannot_replace_root_workspace_authority() {
+        let root = ConversationContext::Thread {
+            messages: vec![ContextMessage {
+                pubkey: "owner".into(),
+                timestamp: "2026-07-31T00:00:00Z".into(),
+                content: "buzz://project-workspace?path=%2Frepo-a".into(),
+            }],
+            total: 1,
+            truncated: false,
+        };
+
+        let selected = select_trusted_workspace_content(
+            "owner",
+            "owner",
+            "buzz://project-workspace?path=%2Frepo-b",
+            true,
+            Some(&root),
+        )
+        .unwrap();
+
+        assert_eq!(
+            selected,
+            Some("buzz://project-workspace?path=%2Frepo-a"),
+            "a later owner reply must not move the thread to a relinked Project"
+        );
+    }
+}
+
 /// Normalize AND validate a pubkey for the batch profile API request.
 /// Returns `None` for malformed input — only valid 64-char hex passes.
 /// See also: `normalize_lookup_key` in queue.rs (normalize-only, no validation).
@@ -3364,6 +3565,7 @@ struct TurnCompletionGuard {
     observer: Option<observer::ObserverHandle>,
     agent_index: Option<usize>,
     channel_id: Option<uuid::Uuid>,
+    conversation_id: Option<uuid::Uuid>,
     turn_id: String,
 }
 
@@ -3372,12 +3574,14 @@ impl TurnCompletionGuard {
         observer: Option<observer::ObserverHandle>,
         agent_index: Option<usize>,
         channel_id: Option<uuid::Uuid>,
+        conversation_id: Option<uuid::Uuid>,
         turn_id: String,
     ) -> Self {
         Self {
             observer,
             agent_index,
             channel_id,
+            conversation_id,
             turn_id,
         }
     }
@@ -3386,7 +3590,12 @@ impl TurnCompletionGuard {
 impl Drop for TurnCompletionGuard {
     fn drop(&mut self) {
         if let Some(observer) = self.observer.take() {
-            let context = observer::context_for(self.channel_id, None, Some(self.turn_id.clone()));
+            let context = observer::context_for_conversation(
+                self.channel_id,
+                self.conversation_id,
+                None,
+                Some(self.turn_id.clone()),
+            );
             observer.emit(
                 "turn_completed",
                 self.agent_index,
diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs
index 029bf86dbf4..12847b9967c 100644
--- a/crates/buzz-acp/src/queue.rs
+++ b/crates/buzz-acp/src/queue.rs
@@ -89,6 +89,21 @@ pub struct FlushBatch {
     pub cancel_reason: Option<CancelReason>,
 }
 
+impl FlushBatch {
+    /// Real NIP-29 channel for relay operations.
+    ///
+    /// `channel_id` is the scheduler identity and may represent a thread.
+    /// Production events carry the real channel in their `h` tag; the fallback
+    /// preserves compatibility with tests and older event producers.
+    pub fn routing_channel_id(&self) -> Uuid {
+        self.events
+            .last()
+            .or_else(|| self.cancelled_events.last())
+            .and_then(|event| crate::conversation::routing_channel_id(&event.event))
+            .unwrap_or(self.channel_id)
+    }
+}
+
 /// Per-channel event queue with per-channel in-flight enforcement.
 ///
 /// # State Machine
@@ -641,6 +656,41 @@ impl EventQueue {
         ids
     }
 
+    /// Drop queued conversations belonging to a real NIP-29 channel.
+    ///
+    /// Conversation-scoped queue keys are intentionally opaque UUIDs, so
+    /// membership cleanup resolves ownership from each event's `h` tag.
+    pub fn drain_routing_channel(&mut self, routing_channel_id: Uuid) -> Vec<String> {
+        let mut conversations = HashSet::new();
+        for (conversation, events) in &self.queues {
+            if events.iter().any(|queued| {
+                crate::conversation::routing_channel_id(&queued.event) == Some(routing_channel_id)
+            }) {
+                conversations.insert(*conversation);
+            }
+        }
+        for (conversation, events) in &self.cancelled_batches {
+            if events.iter().any(|queued| {
+                crate::conversation::routing_channel_id(&queued.event) == Some(routing_channel_id)
+            }) {
+                conversations.insert(*conversation);
+            }
+        }
+        for (conversation, events) in &self.withheld_native_steer {
+            if events.iter().any(|queued| {
+                crate::conversation::routing_channel_id(&queued.event) == Some(routing_channel_id)
+            }) {
+                conversations.insert(*conversation);
+            }
+        }
+
+        let mut ids = Vec::new();
+        for conversation in conversations {
+            ids.extend(self.drain_channel(conversation));
+        }
+        ids
+    }
+
     /// Whether a prompt is currently in-flight for the given channel.
     pub fn is_channel_in_flight(&self, channel_id: Uuid) -> bool {
         self.in_flight_channels.contains(&channel_id)
@@ -1404,6 +1454,7 @@ pub(crate) fn base_section(base_prompt: &str) -> String {
 /// For agents with `protocol_version >= 2`, base_prompt and system_prompt are
 /// delivered via the system role in `session/new` and omitted from this message.
 pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<String> {
+    let routing_channel_id = batch.routing_channel_id();
     // Scope is always derived from the LAST event in the batch — that's the
     // one the agent is responding to. Thread/DM context is supplementary info
     // included alongside, not a scope override. This prevents mixed batches
@@ -1479,7 +1530,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
         )
     };
     sections.push(format_context_hints(
-        batch.channel_id,
+        routing_channel_id,
         args.channel_info,
         &thread_tags,
         is_dm,
@@ -1509,7 +1560,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
                 "\n\n--- Event {} ({}) ---\n{}",
                 i + 1,
                 be.prompt_tag,
-                format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
+                format_event_block(
+                    routing_channel_id,
+                    args.channel_info,
+                    be,
+                    args.profile_lookup
+                )
             ));
         }
         sections.push(s);
@@ -1523,13 +1579,23 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
                 "{}\n\n--- Event 1 ({}) ---\n{}",
                 framing.new_header_single,
                 be.prompt_tag,
-                format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
+                format_event_block(
+                    routing_channel_id,
+                    args.channel_info,
+                    be,
+                    args.profile_lookup
+                )
             )
         } else {
             format!(
                 "[Buzz event: {}]\n{}",
                 be.prompt_tag,
-                format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
+                format_event_block(
+                    routing_channel_id,
+                    args.channel_info,
+                    be,
+                    args.profile_lookup
+                )
             )
         }
     } else {
@@ -1548,7 +1614,12 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
                 "\n\n--- Event {} ({}) ---\n{}",
                 i + 1,
                 be.prompt_tag,
-                format_event_block(batch.channel_id, args.channel_info, be, args.profile_lookup)
+                format_event_block(
+                    routing_channel_id,
+                    args.channel_info,
+                    be,
+                    args.profile_lookup
+                )
             ));
         }
         s
diff --git a/crates/buzz-acp/src/thread_workspace.rs b/crates/buzz-acp/src/thread_workspace.rs
new file mode 100644
index 00000000000..5e08cc3b84c
--- /dev/null
+++ b/crates/buzz-acp/src/thread_workspace.rs
@@ -0,0 +1,176 @@
+use std::{
+    fs,
+    path::{Path, PathBuf},
+};
+
+use anyhow::{bail, Context, Result};
+use tokio::process::Command;
+
+const CONTEXT_URL_PREFIX: &str = "buzz://project-workspace?";
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ProjectWorkspace {
+    pub repo_address: String,
+    pub local_path: PathBuf,
+}
+
+pub fn parse_project_workspace(content: &str) -> Result<Option<ProjectWorkspace>> {
+    let Some(start) = content.find(CONTEXT_URL_PREFIX) else {
+        return Ok(None);
+    };
+    let suffix = &content[start..];
+    let end = suffix
+        .find(['>', ' ', '\n', '\r', '\t'])
+        .unwrap_or(suffix.len());
+    let url = url::Url::parse(&suffix[..end]).context("invalid Project workspace URL")?;
+    let mut repo_address = None;
+    let mut local_path = None;
+    for (key, value) in url.query_pairs() {
+        match key.as_ref() {
+            "repo" => repo_address = Some(value.into_owned()),
+            "path" => local_path = Some(PathBuf::from(value.into_owned())),
+            _ => {}
+        }
+    }
+    let repo_address = repo_address.context("Project workspace URL is missing repo")?;
+    let local_path = local_path.context("Project workspace URL is missing path")?;
+    if repo_address.trim().is_empty() || !local_path.is_absolute() {
+        bail!("Project workspace metadata is invalid");
+    }
+    Ok(Some(ProjectWorkspace {
+        repo_address,
+        local_path,
+    }))
+}
+
+/// Ensure the deterministic worktree for a thread exists and return its cwd.
+pub async fn ensure_thread_worktree(
+    workspace: &ProjectWorkspace,
+    root_event_id: &str,
+) -> Result<PathBuf> {
+    validate_root_event_id(root_event_id)?;
+    let selected_path = fs::canonicalize(&workspace.local_path).with_context(|| {
+        format!(
+            "Project workspace does not exist: {}",
+            workspace.local_path.display()
+        )
+    })?;
+    let repo_root = git_output(&selected_path, ["rev-parse", "--show-toplevel"]).await?;
+    let repo_root =
+        fs::canonicalize(repo_root.trim()).context("could not canonicalize git repository root")?;
+    let common_git = git_output(&repo_root, ["rev-parse", "--git-common-dir"]).await?;
+    let common_git = canonical_git_path(&repo_root, common_git.trim())
+        .context("could not canonicalize git common directory")?;
+
+    let short_root = &root_event_id[..12];
+    let repo_name = repo_root
+        .file_name()
+        .and_then(|name| name.to_str())
+        .unwrap_or("project");
+    let parent = repo_root
+        .parent()
+        .context("git repository has no parent directory")?
+        .join(".buzz-worktrees");
+    let worktree_path = parent.join(format!("{repo_name}-{short_root}"));
+    let branch = format!("buzz/{short_root}");
+
+    if verify_worktree(&worktree_path, &common_git).await {
+        return Ok(worktree_path);
+    }
+    fs::create_dir_all(&parent).context("could not create Buzz worktree directory")?;
+
+    let create = Command::new("git")
+        .arg("-C")
+        .arg(&repo_root)
+        .args(["worktree", "add", "-b", &branch])
+        .arg(&worktree_path)
+        .arg("HEAD")
+        .kill_on_drop(true)
+        .output()
+        .await
+        .context("could not start git worktree add")?;
+
+    if !create.status.success() {
+        // Another harness may have won the same idempotent create race.
+        for _ in 0..10 {
+            if verify_worktree(&worktree_path, &common_git).await {
+                return Ok(worktree_path);
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
+        }
+        // The deterministic branch can outlive a manually removed worktree.
+        // Reattach it instead of treating that recoverable state as a task
+        // failure. Git still rejects a branch checked out somewhere else.
+        let attach = Command::new("git")
+            .arg("-C")
+            .arg(&repo_root)
+            .args(["worktree", "add"])
+            .arg(&worktree_path)
+            .arg(&branch)
+            .kill_on_drop(true)
+            .output()
+            .await
+            .context("could not start git worktree reattach")?;
+        if attach.status.success() && verify_worktree(&worktree_path, &common_git).await {
+            return Ok(worktree_path);
+        }
+        let stderr = String::from_utf8_lossy(&create.stderr);
+        bail!("git worktree add failed: {}", stderr.trim());
+    }
+
+    if !verify_worktree(&worktree_path, &common_git).await {
+        bail!("created worktree failed repository verification");
+    }
+    Ok(worktree_path)
+}
+
+async fn verify_worktree(path: &Path, expected_common_git: &Path) -> bool {
+    let Ok(root) = git_output(path, ["rev-parse", "--show-toplevel"]).await else {
+        return false;
+    };
+    let Ok(common) = git_output(path, ["rev-parse", "--git-common-dir"]).await else {
+        return false;
+    };
+    let Ok(root) = fs::canonicalize(root.trim()) else {
+        return false;
+    };
+    let Ok(common_path) = canonical_git_path(&root, common.trim()) else {
+        return false;
+    };
+    common_path == expected_common_git
+}
+
+fn canonical_git_path(repo_root: &Path, path: &str) -> std::io::Result<PathBuf> {
+    let path = Path::new(path);
+    fs::canonicalize(if path.is_absolute() {
+        path.to_path_buf()
+    } else {
+        repo_root.join(path)
+    })
+}
+
+async fn git_output<I, S>(cwd: &Path, args: I) -> Result<String>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<std::ffi::OsStr>,
+{
+    let output = Command::new("git")
+        .arg("-C")
+        .arg(cwd)
+        .args(args)
+        .kill_on_drop(true)
+        .output()
+        .await
+        .context("could not start git")?;
+    if !output.status.success() {
+        bail!("{}", String::from_utf8_lossy(&output.stderr).trim());
+    }
+    String::from_utf8(output.stdout).context("git returned non-UTF-8 output")
+}
+
+fn validate_root_event_id(root_event_id: &str) -> Result<()> {
+    if root_event_id.len() != 64 || !root_event_id.chars().all(|c| c.is_ascii_hexdigit()) {
+        bail!("thread root event ID must be 64 hex characters");
+    }
+    Ok(())
+}
diff --git a/crates/buzz-acp/src/thread_workspace_tests.rs b/crates/buzz-acp/src/thread_workspace_tests.rs
new file mode 100644
index 00000000000..4e4545f557e
--- /dev/null
+++ b/crates/buzz-acp/src/thread_workspace_tests.rs
@@ -0,0 +1,66 @@
+use std::{fs, path::PathBuf};
+
+use tokio::process::Command;
+use uuid::Uuid;
+
+use crate::thread_workspace::{ensure_thread_worktree, parse_project_workspace, ProjectWorkspace};
+
+#[test]
+fn parses_encoded_project_workspace_context() {
+    let content = "[ctx]: <buzz://project-workspace?repo=github.com%2Facme%2Fapp&path=%2Ftmp%2Fapp> \"Project\"\n\nFix it";
+    let workspace = parse_project_workspace(content)
+        .expect("valid context")
+        .expect("context present");
+    assert_eq!(workspace.repo_address, "github.com/acme/app");
+    assert_eq!(workspace.local_path, PathBuf::from("/tmp/app"));
+}
+
+#[test]
+fn rejects_relative_workspace_path() {
+    let content = "buzz://project-workspace?repo=acme%2Fapp&path=relative";
+    assert!(parse_project_workspace(content).is_err());
+}
+
+#[tokio::test]
+async fn concurrent_ensure_calls_converge_on_one_worktree() {
+    let fixture = std::env::temp_dir().join(format!("buzz-worktree-test-{}", Uuid::new_v4()));
+    let repo = fixture.join("project");
+    fs::create_dir_all(&repo).expect("fixture directory");
+    run_git(&repo, &["init", "-b", "main"]).await;
+    run_git(&repo, &["config", "user.email", "test@example.com"]).await;
+    run_git(&repo, &["config", "user.name", "Test"]).await;
+    fs::write(repo.join("README.md"), "fixture").expect("fixture file");
+    run_git(&repo, &["add", "README.md"]).await;
+    run_git(&repo, &["commit", "-m", "fixture"]).await;
+
+    let workspace = ProjectWorkspace {
+        repo_address: "fixture/project".to_string(),
+        local_path: repo,
+    };
+    let root = "a".repeat(64);
+    let (first, second) = tokio::join!(
+        ensure_thread_worktree(&workspace, &root),
+        ensure_thread_worktree(&workspace, &root)
+    );
+    let first = first.expect("first ensure succeeds");
+    let second = second.expect("second ensure converges");
+    assert_eq!(first, second);
+    assert!(first.join("README.md").is_file());
+
+    fs::remove_dir_all(&fixture).expect("fixture cleanup");
+}
+
+async fn run_git(cwd: &std::path::Path, args: &[&str]) {
+    let output = Command::new("git")
+        .arg("-C")
+        .arg(cwd)
+        .args(args)
+        .output()
+        .await
+        .expect("git starts");
+    assert!(
+        output.status.success(),
+        "git failed: {}",
+        String::from_utf8_lossy(&output.stderr)
+    );
+}
diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts
index 459fa757432..4b282540dda 100644
--- a/desktop/playwright.config.ts
+++ b/desktop/playwright.config.ts
@@ -1,5 +1,8 @@
 import { defineConfig, devices } from "@playwright/test";
 
+const e2ePort = process.env.BUZZ_E2E_PORT ?? "4173";
+const e2eBaseUrl = `http://127.0.0.1:${e2ePort}`;
+
 export default defineConfig({
   testDir: "./tests/e2e",
   timeout: 30_000,
@@ -10,7 +13,8 @@ export default defineConfig({
     ["html", { open: "never", outputFolder: "playwright-report" }],
   ],
   use: {
-    baseURL: "http://127.0.0.1:4173",
+    baseURL: e2eBaseUrl,
+    permissions: ["clipboard-read", "clipboard-write"],
     screenshot: "only-on-failure",
     trace: "on-first-retry",
     video: "retain-on-failure",
@@ -161,9 +165,9 @@ export default defineConfig({
     },
   ],
   webServer: {
-    command: "python3 -m http.server 4173 -d dist",
+    command: `python3 -m http.server ${e2ePort} -d dist`,
     cwd: ".",
     reuseExistingServer: !process.env.CI,
-    url: "http://127.0.0.1:4173",
+    url: e2eBaseUrl,
   },
 });
diff --git a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs
index d78d1d21afe..090b8d34978 100644
--- a/desktop/src/features/agents/activeAgentTurnsStore.test.mjs
+++ b/desktop/src/features/agents/activeAgentTurnsStore.test.mjs
@@ -5,6 +5,7 @@ import {
   syncAgentTurnsFromEvents,
   syncActiveAgentTurnsFromObserver,
   getActiveTurnsForAgent,
+  getActiveTurnControlTargetsForAgent,
   getActiveTurnsByChannel,
   resetActiveAgentTurnsStore,
   subscribeActiveAgentTurns,
@@ -86,6 +87,42 @@ describe("activeAgentTurnsStore", () => {
     });
   });
 
+  it("keeps exact control targets for concurrent threads in one channel", () => {
+    syncAgentTurnsFromEvents(AGENT, [
+      makeEvent({
+        seq: 1,
+        channelId: "shared-channel",
+        conversationId: "thread-a",
+        turnId: "turn-a",
+      }),
+      makeEvent({
+        seq: 2,
+        timestamp: "2024-01-01T00:00:01Z",
+        channelId: "shared-channel",
+        conversationId: "thread-b",
+        turnId: "turn-b",
+      }),
+    ]);
+
+    assert.deepEqual(getActiveTurnControlTargetsForAgent(AGENT), [
+      {
+        channelId: "shared-channel",
+        conversationId: "thread-a",
+        turnId: "turn-a",
+      },
+      {
+        channelId: "shared-channel",
+        conversationId: "thread-b",
+        turnId: "turn-b",
+      },
+    ]);
+    assert.equal(
+      getActiveTurnsForAgent(AGENT).length,
+      1,
+      "visual channel badge remains aggregated",
+    );
+  });
+
   describe("seq restart detection", () => {
     it("processes post-restart events whose timestamp climbs past the watermark", () => {
       // Process events up to seq 50.
diff --git a/desktop/src/features/agents/activeAgentTurnsStore.ts b/desktop/src/features/agents/activeAgentTurnsStore.ts
index 3af41ae0789..44b57965d5f 100644
--- a/desktop/src/features/agents/activeAgentTurnsStore.ts
+++ b/desktop/src/features/agents/activeAgentTurnsStore.ts
@@ -40,6 +40,7 @@ const PRUNE_INTERVAL_MS = 5_000;
 type ActiveTurn = {
   turnId: string;
   channelId: string;
+  conversationId: string;
   startedAt: number;
   lastActivityAt: number;
 };
@@ -50,6 +51,13 @@ export type ActiveTurnSummary = {
   anchorAt: number;
 };
 
+/** Exact target for observer controls; never collapsed by real channel. */
+export type ActiveTurnControlTarget = {
+  channelId: string;
+  conversationId: string;
+  turnId: string;
+};
+
 /** One channel with active agent work, aggregated across agents. */
 export type ActiveChannelTurnSummary = {
   channelId: string;
@@ -83,6 +91,7 @@ const clockOffsetByAgent = new Map<string, number>();
 // Cached snapshots for useSyncExternalStore reference stability.
 // Only regenerated when the underlying turn map for an agent actually changes.
 const cachedTurnSummaries = new Map<string, ActiveTurnSummary[]>();
+const cachedControlTargets = new Map<string, ActiveTurnControlTarget[]>();
 let cachedChannelTurnSummaries: ActiveChannelTurnSummary[] | null = null;
 
 // Composite watermark per agent: the newest observer event processed, by
@@ -103,6 +112,7 @@ let pruneInterval: ReturnType<typeof setInterval> | null = null;
 
 function invalidateCache(agentKey: string) {
   cachedTurnSummaries.delete(agentKey);
+  cachedControlTargets.delete(agentKey);
   cachedChannelTurnSummaries = null;
 }
 
@@ -137,6 +147,7 @@ function parseTimestamp(timestamp: string): number | null {
 function startTurn(
   agentPubkey: string,
   channelId: string,
+  conversationId: string,
   turnId: string,
   timestamp: string,
 ) {
@@ -166,6 +177,7 @@ function startTurn(
   agentTurns.set(turnId, {
     turnId,
     channelId,
+    conversationId,
     startedAt,
     lastActivityAt: Date.now(),
   });
@@ -214,7 +226,13 @@ function resurrectTurn(agentPubkey: string, event: ObserverEvent): boolean {
     frameAt !== null && startedAtMs !== null && startedAtMs <= frameAt
       ? startedAt
       : event.timestamp;
-  startTurn(agentPubkey, event.channelId, event.turnId, safeStartedAt);
+  startTurn(
+    agentPubkey,
+    event.channelId,
+    event.conversationId ?? event.channelId,
+    event.turnId,
+    safeStartedAt,
+  );
   return true;
 }
 
@@ -351,6 +369,7 @@ function processEvent(agentPubkey: string, event: ObserverEvent) {
         startTurn(
           agentPubkey,
           event.channelId,
+          event.conversationId ?? event.channelId,
           event.turnId ?? `seq-${event.seq}`,
           event.timestamp,
         );
@@ -456,7 +475,35 @@ export function getActiveTurnsForAgent(
   return result;
 }
 
+export function getActiveTurnControlTargetsForAgent(
+  agentPubkey: string | null | undefined,
+): ActiveTurnControlTarget[] {
+  if (!agentPubkey) return EMPTY_CONTROL_TARGETS;
+  const key = normalizePubkey(agentPubkey);
+  const agentTurns = activeTurnsByAgent.get(key);
+  if (!agentTurns || agentTurns.size === 0) return EMPTY_CONTROL_TARGETS;
+
+  const cached = cachedControlTargets.get(key);
+  if (cached) return cached;
+
+  const result = [...agentTurns.values()]
+    .map(({ channelId, conversationId, turnId }) => ({
+      channelId,
+      conversationId,
+      turnId,
+    }))
+    .sort(
+      (a, b) =>
+        a.channelId.localeCompare(b.channelId) ||
+        a.conversationId.localeCompare(b.conversationId) ||
+        a.turnId.localeCompare(b.turnId),
+    );
+  cachedControlTargets.set(key, result);
+  return result;
+}
+
 const EMPTY_TURNS: ActiveTurnSummary[] = [];
+const EMPTY_CONTROL_TARGETS: ActiveTurnControlTarget[] = [];
 const EMPTY_CHANNEL_TURNS: ActiveChannelTurnSummary[] = [];
 
 /**
@@ -535,6 +582,16 @@ export function useActiveAgentTurns(
   return React.useSyncExternalStore(subscribeActiveAgentTurns, getSnapshot);
 }
 
+export function useActiveAgentTurnControlTargets(
+  agentPubkey: string | null | undefined,
+): ActiveTurnControlTarget[] {
+  const getSnapshot = React.useCallback(
+    () => getActiveTurnControlTargetsForAgent(agentPubkey),
+    [agentPubkey],
+  );
+  return React.useSyncExternalStore(subscribeActiveAgentTurns, getSnapshot);
+}
+
 /**
  * Hook: returns channels with active agent work across all tracked agents.
  * Re-renders when the channel set changes — not when the clock ticks.
diff --git a/desktop/src/features/agents/lib/liveSwitchOutcome.test.mjs b/desktop/src/features/agents/lib/liveSwitchOutcome.test.mjs
index 737d84b8620..b768d91b003 100644
--- a/desktop/src/features/agents/lib/liveSwitchOutcome.test.mjs
+++ b/desktop/src/features/agents/lib/liveSwitchOutcome.test.mjs
@@ -6,7 +6,13 @@ import { awaitLiveSwitchOutcome } from "./liveSwitchOutcome.ts";
 const MODEL = "goose-claude-fable-5";
 
 function frame(status, overrides = {}) {
-  return { type: "switch_model", status, modelId: MODEL, ...overrides };
+  return {
+    type: "switch_model",
+    status,
+    modelId: MODEL,
+    turnId: "turn-1",
+    ...overrides,
+  };
 }
 
 /**
@@ -26,7 +32,10 @@ function harness(channelCount) {
   });
 
   const outcome = awaitLiveSwitchOutcome({
-    channelCount,
+    targetTurnIds: Array.from(
+      { length: channelCount },
+      (_, index) => `turn-${index + 1}`,
+    ),
     modelId: MODEL,
     subscribe: (fn) => {
       listener = fn;
@@ -65,8 +74,8 @@ test("awaitLiveSwitchOutcome fast sent on one channel does not mask a later unsu
   const h = harness(2);
   // Channel A acks fast as `sent`; a first-ack-resolves impl would settle "ok"
   // here. The fail-fast contract must keep waiting and then reject on B.
-  h.push(frame("sent"));
-  h.push(frame("unsupported_model"));
+  h.push(frame("sent", { turnId: "turn-1" }));
+  h.push(frame("unsupported_model", { turnId: "turn-2" }));
   assert.equal(await h.outcome, "unsupported");
 });
 
@@ -88,18 +97,25 @@ test("awaitLiveSwitchOutcome resolves ok only after the last channel acks", asyn
     }
   };
 
-  h.push(frame("sent"));
+  h.push(frame("sent", { turnId: "turn-1" }));
   await drainMicrotasks();
   assert.equal(settled, false, "must not resolve on the first ack");
 
-  h.push(frame("switched"));
+  h.push(frame("switched", { turnId: "turn-2" }));
   await drainMicrotasks();
   assert.equal(settled, false, "must not resolve before the last ack");
 
-  h.push(frame("turn_ending"));
+  h.push(frame("sent", { turnId: "turn-3" }));
   assert.equal(await h.outcome, "ok");
 });
 
+test("awaitLiveSwitchOutcome does not report success when a targeted turn is already ending", async () => {
+  const h = harness(2);
+  h.push(frame("sent", { turnId: "turn-1" }));
+  h.push(frame("turn_ending", { turnId: "turn-2" }));
+  assert.equal(await h.outcome, "not_applied");
+});
+
 test("awaitLiveSwitchOutcome rejects on unsupported immediately and unsubscribes exactly once", async () => {
   const h = harness(2);
   h.push(frame("unsupported_model"));
@@ -117,6 +133,7 @@ test("awaitLiveSwitchOutcome ignores frames for a different model or control typ
   const h = harness(1);
   h.push(frame("sent", { modelId: "some-other-model" }));
   h.push({ type: "cancel_turn", status: "sent", modelId: MODEL });
+  h.push(frame("sent", { turnId: "some-other-turn" }));
   let settled = false;
   void h.outcome.then(() => {
     settled = true;
@@ -128,10 +145,10 @@ test("awaitLiveSwitchOutcome ignores frames for a different model or control typ
   assert.equal(await h.outcome, "ok");
 });
 
-test("awaitLiveSwitchOutcome resolves ok via the timeout fallback when the harness never replies", async () => {
+test("awaitLiveSwitchOutcome reports unconfirmed when the harness never replies", async () => {
   const h = harness(2);
   h.fireTimeout();
-  assert.equal(await h.outcome, "ok");
+  assert.equal(await h.outcome, "unconfirmed");
   assert.equal(h.unsubscribeCalls, 1, "timeout fallback unsubscribes");
 });
 
@@ -144,11 +161,11 @@ test("awaitLiveSwitchOutcome fires the per-channel sends after subscribing", asy
   assert.equal(await h.outcome, "ok");
 });
 
-test("awaitLiveSwitchOutcome with zero channels resolves ok at the timeout (no acks expected)", async () => {
+test("awaitLiveSwitchOutcome with zero targets is unconfirmed at timeout", async () => {
   // No active turns means channelCount 0: remaining starts at 0 but the success
   // resolve only fires inside a frame callback, so with no frames the timeout
   // fallback is what settles it. This documents the degenerate path.
   const h = harness(0);
   h.fireTimeout();
-  assert.equal(await h.outcome, "ok");
+  assert.equal(await h.outcome, "unconfirmed");
 });
diff --git a/desktop/src/features/agents/lib/liveSwitchOutcome.ts b/desktop/src/features/agents/lib/liveSwitchOutcome.ts
index d12261e5968..ca86996a542 100644
--- a/desktop/src/features/agents/lib/liveSwitchOutcome.ts
+++ b/desktop/src/features/agents/lib/liveSwitchOutcome.ts
@@ -1,60 +1,70 @@
 import type { ControlResultFrame } from "@/shared/api/types";
 
 /**
- * Resolve the outcome of a live `switch_model` across one or more channels.
+ * Resolve the outcome of a live `switch_model` across exact observer turns.
  *
- * A live switch fires a `switch_model` frame per active channel and learns each
- * channel's result asynchronously over the observer relay. The fail-fast rule:
+ * A live switch fires a `switch_model` frame per active turn and learns each
+ * turn's result asynchronously over the observer relay. The fail-fast rule:
  * any single `unsupported_model` result rejects the whole pick immediately;
- * every other status must arrive from every channel before resolving success.
- * If the harness never replies, the fallback timeout resolves `"ok"` — the
- * override still rides the requeued/next session, we just can't confirm it
- * synchronously.
+ * a terminal/no-active result reports that the switch did not apply, and every
+ * successful acknowledgement must arrive exactly once before resolving.
+ * Timeout is explicitly unconfirmed rather than a false success.
  *
  * The counting lives here, isolated from React and the relay so it can be unit
  * tested with synthetic frames and a fake clock. The caller injects the
- * relay subscription, the per-channel sends, and the timeout scheduler.
+ * relay subscription, the per-turn sends, and the timeout scheduler.
  */
 export async function awaitLiveSwitchOutcome({
-  channelCount,
+  targetTurnIds,
   modelId,
   subscribe,
   sendSwitches,
   scheduleTimeout,
 }: {
-  /** Number of channels the switch was fired to — the success threshold. */
-  channelCount: number;
+  /** Exact observer turns targeted by the switch. */
+  targetTurnIds: readonly string[];
   /** Model being switched to; frames for any other model are ignored. */
   modelId: string;
   /** Register a control-result listener; returns an unsubscribe function. */
   subscribe: (listener: (frame: ControlResultFrame) => void) => () => void;
-  /** Fire the per-channel `switch_model` sends. Resolves when all are sent. */
+  /** Fire the per-turn `switch_model` sends. Resolves when all are sent. */
   sendSwitches: () => Promise<void>;
   /** Schedule the no-reply fallback; returns a cancel function. */
   scheduleTimeout: (onTimeout: () => void) => () => void;
-}): Promise<"ok" | "unsupported"> {
-  const settled = new Promise<"ok" | "unsupported">((resolve) => {
+}): Promise<"ok" | "unsupported" | "not_applied" | "unconfirmed"> {
+  const settled = new Promise<
+    "ok" | "unsupported" | "not_applied" | "unconfirmed"
+  >((resolve) => {
     let unsubscribe = () => {};
     let cancelTimeout = () => {};
-    let remaining = channelCount;
-    const finish = (outcome: "ok" | "unsupported") => {
+    const remaining = new Set(targetTurnIds);
+    const finish = (
+      outcome: "ok" | "unsupported" | "not_applied" | "unconfirmed",
+    ) => {
       cancelTimeout();
       unsubscribe();
       resolve(outcome);
     };
-    cancelTimeout = scheduleTimeout(() => finish("ok"));
+    cancelTimeout = scheduleTimeout(() => finish("unconfirmed"));
     unsubscribe = subscribe((frame) => {
       if (frame.type !== "switch_model" || frame.modelId !== modelId) {
         return;
       }
+      if (!frame.turnId || !remaining.has(frame.turnId)) {
+        return;
+      }
       if (frame.status === "unsupported_model") {
         // Any single failure rejects the whole pick immediately.
         finish("unsupported");
         return;
       }
-      // sent / switched / turn_ending — count as success for this channel.
-      remaining -= 1;
-      if (remaining <= 0) {
+      if (frame.status === "turn_ending" || frame.status === "no_active_turn") {
+        finish("not_applied");
+        return;
+      }
+      // sent / switched — count each exact turn at most once.
+      remaining.delete(frame.turnId);
+      if (remaining.size === 0) {
         finish("ok");
       }
     });
diff --git a/desktop/src/features/agents/ui/ModelPicker.tsx b/desktop/src/features/agents/ui/ModelPicker.tsx
index f7bafde99b5..824f5a8eb90 100644
--- a/desktop/src/features/agents/ui/ModelPicker.tsx
+++ b/desktop/src/features/agents/ui/ModelPicker.tsx
@@ -10,7 +10,7 @@ import { getAgentModels, updateManagedAgent } from "@/shared/api/tauri";
 import { switchManagedAgentModel } from "@/shared/api/agentControl";
 import { awaitLiveSwitchOutcome } from "@/features/agents/lib/liveSwitchOutcome";
 import { subscribeControlResults } from "@/features/agents/observerRelayStore";
-import { useActiveAgentTurns } from "@/features/agents/activeAgentTurnsStore";
+import { useActiveAgentTurnControlTargets } from "@/features/agents/activeAgentTurnsStore";
 import {
   useAgentConfigSurface,
   managedAgentsQueryKey,
@@ -43,7 +43,7 @@ export function ModelPicker({
   const queryClient = useQueryClient();
 
   const isRunning = agent.status === "running" || agent.status === "deployed";
-  const activeTurns = useActiveAgentTurns(agent.pubkey);
+  const activeTurns = useActiveAgentTurnControlTargets(agent.pubkey);
   // A live switch rides the agent's running session(s) instead of persisting a
   // new default. It applies only to a persona-linked running agent with at
   // least one active turn — those are the channels the desktop can name in the
@@ -107,22 +107,27 @@ export function ModelPicker({
     return labels[origin] ?? null;
   }, [configSurface]);
 
-  // Send a live `switch_model` frame to each channel the agent is working in
+  // Send a live `switch_model` frame to each exact turn the agent is working in
   // and wait for the harness to acknowledge. Any single `unsupported_model`
-  // result rejects the whole pick immediately; all other statuses must arrive
-  // from every channel before resolving success.
+  // result rejects the whole pick immediately; only exact per-turn success
+  // acknowledgements resolve the switch.
   const sendLiveSwitch = React.useCallback(
     (modelId: string) => {
-      const channelIds = activeTurns.map((turn) => turn.channelId);
       return awaitLiveSwitchOutcome({
-        channelCount: channelIds.length,
+        targetTurnIds: activeTurns.map((turn) => turn.turnId),
         modelId,
         subscribe: (listener) =>
           subscribeControlResults(agent.pubkey, listener),
         sendSwitches: async () => {
           await Promise.all(
-            channelIds.map((channelId) =>
-              switchManagedAgentModel(agent.pubkey, channelId, modelId),
+            activeTurns.map((turn) =>
+              switchManagedAgentModel(
+                agent.pubkey,
+                turn.channelId,
+                turn.conversationId,
+                turn.turnId,
+                modelId,
+              ),
             ),
           );
         },
@@ -147,6 +152,14 @@ export function ModelPicker({
           toast.error("That model isn't available for this agent.");
           return;
         }
+        if (outcome === "not_applied") {
+          toast.error("The active turn ended before the model switch landed.");
+          return;
+        }
+        if (outcome === "unconfirmed") {
+          toast.warning("Model switch sent, but the agent did not confirm it.");
+          return;
+        }
         toast.success("Model switched for this session.");
         onModelChanged?.();
         return;
diff --git a/desktop/src/features/agents/ui/agentSessionTypes.ts b/desktop/src/features/agents/ui/agentSessionTypes.ts
index 578f98076cd..6f0115e175a 100644
--- a/desktop/src/features/agents/ui/agentSessionTypes.ts
+++ b/desktop/src/features/agents/ui/agentSessionTypes.ts
@@ -6,6 +6,7 @@ export type ObserverEvent = {
   kind: string;
   agentIndex: number | null;
   channelId: string | null;
+  conversationId?: string | null;
   sessionId: string | null;
   turnId: string | null;
   startedAt?: string | null;
diff --git a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx
index a907f872458..c449d5d428a 100644
--- a/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx
+++ b/desktop/src/features/channels/ui/AgentSessionThreadPanel.tsx
@@ -9,6 +9,7 @@ import {
 import { toast } from "sonner";
 
 import { useAgentWorking } from "@/features/agents/agentWorkingSignal";
+import { useActiveAgentTurnControlTargets } from "@/features/agents/activeAgentTurnsStore";
 import { isManagedAgentActive } from "@/features/agents/lib/managedAgentControlActions";
 import {
   mergeObserverEventWindows,
@@ -103,7 +104,18 @@ export function AgentSessionThreadPanel({
     agent.pubkey,
     sessionChannelId,
   );
-  const canStopCurrentTurn = isWorking && canInterruptTurn;
+  const activeControlTargets = useActiveAgentTurnControlTargets(agent.pubkey);
+  const scopedControlTargets = React.useMemo(
+    () =>
+      sessionChannelId
+        ? activeControlTargets.filter(
+            (target) => target.channelId === sessionChannelId,
+          )
+        : activeControlTargets,
+    [activeControlTargets, sessionChannelId],
+  );
+  const canStopCurrentTurn =
+    isWorking && canInterruptTurn && scopedControlTargets.length > 0;
   useEscapeKey(onClose, isOverlay || isSinglePanelView);
 
   const scrollRef = React.useRef<HTMLDivElement>(null);
@@ -239,13 +251,20 @@ export function AgentSessionThreadPanel({
     : "All channels";
   const animateActivity = useTranscriptAnimationEnabled();
   const showTimestamps = useTranscriptTimestampsEnabled();
-  async function handleInterruptTurn() {
-    if (!channel) {
+  async function handleInterruptTurn(
+    target: (typeof scopedControlTargets)[number] | undefined,
+  ) {
+    if (!target) {
       return;
     }
 
     try {
-      await cancelManagedAgentTurn(agent.pubkey, channel.id);
+      await cancelManagedAgentTurn(
+        agent.pubkey,
+        target.channelId,
+        target.conversationId,
+        target.turnId,
+      );
       toast.success(
         `Stop signal sent to ${agent.name}. It may take a moment to respond.`,
       );
@@ -374,35 +393,59 @@ export function AgentSessionThreadPanel({
               />
             </DropdownMenuItem>
             <DropdownMenuSeparator />
-            <DropdownMenuItem
-              className="items-start gap-3"
-              data-testid="agent-session-stop-turn"
-              disabled={!canStopCurrentTurn}
-              onSelect={() => {
-                void handleInterruptTurn();
-              }}
-              title={
-                canStopCurrentTurn
-                  ? "Interrupt the current ACP turn without stopping the agent process."
-                  : isWorking
-                    ? "Only locally managed agents can be interrupted from this community."
-                    : "Available while the agent is working."
-              }
-            >
-              <Octagon className="mt-0.5 h-4 w-4 text-muted-foreground" />
-              <span className="min-w-0 flex-1">
-                <span className="block text-sm font-medium">
-                  Stop current turn
-                </span>
-                {!canStopCurrentTurn ? (
-                  <span className="mt-0.5 block text-xs text-muted-foreground">
-                    {isWorking
-                      ? "Only available for locally managed agents."
-                      : "Available while the agent is working."}
+            {scopedControlTargets.length > 1 && canInterruptTurn ? (
+              scopedControlTargets.map((target, index) => (
+                <DropdownMenuItem
+                  className="items-start gap-3"
+                  data-testid={`agent-session-stop-turn-${target.turnId}`}
+                  key={target.turnId}
+                  onSelect={() => {
+                    void handleInterruptTurn(target);
+                  }}
+                  title="Interrupt this exact ACP turn without stopping the agent process."
+                >
+                  <Octagon className="mt-0.5 h-4 w-4 text-muted-foreground" />
+                  <span className="min-w-0 flex-1">
+                    <span className="block text-sm font-medium">
+                      Stop active turn {index + 1}
+                    </span>
+                    <span className="mt-0.5 block text-xs text-muted-foreground">
+                      Thread {target.conversationId.slice(0, 8)}
+                    </span>
                   </span>
-                ) : null}
-              </span>
-            </DropdownMenuItem>
+                </DropdownMenuItem>
+              ))
+            ) : (
+              <DropdownMenuItem
+                className="items-start gap-3"
+                data-testid="agent-session-stop-turn"
+                disabled={!canStopCurrentTurn}
+                onSelect={() => {
+                  void handleInterruptTurn(scopedControlTargets[0]);
+                }}
+                title={
+                  canStopCurrentTurn
+                    ? "Interrupt the current ACP turn without stopping the agent process."
+                    : isWorking
+                      ? "Only locally managed agents can be interrupted from this community."
+                      : "Available while the agent is working."
+                }
+              >
+                <Octagon className="mt-0.5 h-4 w-4 text-muted-foreground" />
+                <span className="min-w-0 flex-1">
+                  <span className="block text-sm font-medium">
+                    Stop current turn
+                  </span>
+                  {!canStopCurrentTurn ? (
+                    <span className="mt-0.5 block text-xs text-muted-foreground">
+                      {isWorking
+                        ? "Only available for locally managed agents."
+                        : "Available while the agent is working."}
+                    </span>
+                  ) : null}
+                </span>
+              </DropdownMenuItem>
+            )}
           </DropdownMenuContent>
         </DropdownMenu>
       ) : null}
diff --git a/desktop/src/features/messages/lib/projectThreadAgentRouting.test.mjs b/desktop/src/features/messages/lib/projectThreadAgentRouting.test.mjs
new file mode 100644
index 00000000000..9857363278b
--- /dev/null
+++ b/desktop/src/features/messages/lib/projectThreadAgentRouting.test.mjs
@@ -0,0 +1,44 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { resolveProjectThreadAgentRouting } from "./projectThreadAgentRouting.ts";
+
+const context =
+  "[ctx]: <buzz://project-workspace?repo=acme%2Fapp&path=%2Ftmp%2Fapp>";
+
+test("new Project task wakes first agent and keeps later agents as references", () => {
+  assert.deepEqual(
+    resolveProjectThreadAgentRouting({
+      content: `${context}\n\n@planner plan, @builder build, @reviewer review`,
+      explicitAgentPubkeys: ["planner", "builder", "reviewer"],
+      isThreadReply: false,
+      mentionPubkeys: ["human", "planner", "builder", "reviewer"],
+    }),
+    {
+      mentionPubkeys: ["human", "planner"],
+      referencePubkeys: ["builder", "reviewer"],
+    },
+  );
+});
+
+test("ordinary channels and thread replies retain all notifying mentions", () => {
+  const mentions = ["planner", "builder"];
+  assert.deepEqual(
+    resolveProjectThreadAgentRouting({
+      content: "@planner and @builder",
+      explicitAgentPubkeys: mentions,
+      isThreadReply: false,
+      mentionPubkeys: mentions,
+    }),
+    { mentionPubkeys: mentions, referencePubkeys: [] },
+  );
+  assert.deepEqual(
+    resolveProjectThreadAgentRouting({
+      content: context,
+      explicitAgentPubkeys: mentions,
+      isThreadReply: true,
+      mentionPubkeys: mentions,
+    }),
+    { mentionPubkeys: mentions, referencePubkeys: [] },
+  );
+});
diff --git a/desktop/src/features/messages/lib/projectThreadAgentRouting.ts b/desktop/src/features/messages/lib/projectThreadAgentRouting.ts
new file mode 100644
index 00000000000..fc20094f7cc
--- /dev/null
+++ b/desktop/src/features/messages/lib/projectThreadAgentRouting.ts
@@ -0,0 +1,49 @@
+import { normalizePubkey } from "@/shared/lib/pubkey";
+
+const PROJECT_WORKSPACE_CONTEXT = "buzz://project-workspace?";
+
+type RoutingInput = {
+  content: string;
+  explicitAgentPubkeys: string[];
+  isThreadReply: boolean;
+  mentionPubkeys: string[];
+};
+
+export type ProjectThreadAgentRouting = {
+  mentionPubkeys: string[];
+  referencePubkeys: string[];
+};
+
+/**
+ * A new Project task wakes only the first explicitly ordered agent. Remaining
+ * agents stay renderable as non-notifying references and can be handed work in
+ * later thread replies. Ordinary chat, non-Project channels, and replies keep
+ * their existing mention behavior.
+ */
+export function resolveProjectThreadAgentRouting({
+  content,
+  explicitAgentPubkeys,
+  isThreadReply,
+  mentionPubkeys,
+}: RoutingInput): ProjectThreadAgentRouting {
+  const orderedAgents = uniquePubkeys(explicitAgentPubkeys);
+  if (
+    isThreadReply ||
+    !content.includes(PROJECT_WORKSPACE_CONTEXT) ||
+    orderedAgents.length < 2
+  ) {
+    return { mentionPubkeys, referencePubkeys: [] };
+  }
+
+  const deferred = new Set(orderedAgents.slice(1));
+  return {
+    mentionPubkeys: mentionPubkeys.filter(
+      (pubkey) => !deferred.has(normalizePubkey(pubkey)),
+    ),
+    referencePubkeys: [...deferred],
+  };
+}
+
+function uniquePubkeys(pubkeys: Iterable<string>) {
+  return [...new Set([...pubkeys].map(normalizePubkey))].filter(Boolean);
+}
diff --git a/desktop/src/features/messages/ui/useMentionSendFlow.ts b/desktop/src/features/messages/ui/useMentionSendFlow.ts
index 41a2b3e882e..79289047a5d 100644
--- a/desktop/src/features/messages/ui/useMentionSendFlow.ts
+++ b/desktop/src/features/messages/ui/useMentionSendFlow.ts
@@ -14,6 +14,7 @@ import { resolvePersonaRuntime } from "@/features/agents/lib/resolvePersonaRunti
 import { useAddChannelMembersMutation } from "@/features/channels/hooks";
 import { resolveCurrentProjectChannelAgentMessage } from "@/features/projects/lib/project-local-workspace-runtime";
 import { filterEffectiveExplicitAgentPubkeys } from "@/features/messages/lib/effectiveExplicitAgentPubkeys";
+import { resolveProjectThreadAgentRouting } from "@/features/messages/lib/projectThreadAgentRouting";
 import type { UseChannelLinksResult } from "@/features/messages/lib/useChannelLinks";
 import type { UseEmojiAutocompleteResult } from "@/features/messages/lib/useEmojiAutocomplete";
 import {
@@ -537,10 +538,20 @@ export function useMentionSendFlow({
         }
 
         try {
-          await onSendRef.current(
-            finalContent,
+          const taskRouting = resolveProjectThreadAgentRouting({
+            content: finalContent,
+            explicitAgentPubkeys: effectiveExplicitAgentPubkeys,
+            isThreadReply: draft.capturedThreadContext !== null,
             mentionPubkeys,
+          });
+          const routedOutgoingTags = mergeOutgoingTagsWithReferenceMentions(
             outgoingTags,
+            taskRouting.referencePubkeys,
+          );
+          await onSendRef.current(
+            finalContent,
+            taskRouting.mentionPubkeys,
+            routedOutgoingTags,
             sendChannelId,
             draft.capturedThreadContext,
           );
diff --git a/desktop/src/features/projects/lib/project-channel-agent-context.ts b/desktop/src/features/projects/lib/project-channel-agent-context.ts
index 7d457e3ff73..3c407a5f645 100644
--- a/desktop/src/features/projects/lib/project-channel-agent-context.ts
+++ b/desktop/src/features/projects/lib/project-channel-agent-context.ts
@@ -126,14 +126,19 @@ export function appendProjectChannelAgentContext(
   }
   const title = [
     `Project ${context.repoAddress}.`,
-    `Use workspace absolute path ${context.localPath}.`,
-    "session/new.cwd remains unchanged.",
+    `Source workspace ${context.localPath}.`,
+    "The harness provisions one isolated worktree per thread.",
   ]
     .join(" ")
     .replaceAll("\\", "\\\\")
     .replaceAll('"', '\\"');
   const label = `buzz-project-context-${globalThis.crypto.randomUUID()}`;
-  return `[${label}]: <buzz://project-workspace> "${title}"\n\n${content}`;
+  const workspaceUrl = [
+    "buzz://project-workspace",
+    `?repo=${encodeURIComponent(context.repoAddress)}`,
+    `&path=${encodeURIComponent(context.localPath)}`,
+  ].join("");
+  return `[${label}]: <${workspaceUrl}> "${title}"\n\n${content}`;
 }
 
 export async function resolveProjectChannelAgentMessage(
diff --git a/desktop/src/features/projects/project-channel-agent-context-contract.test.mjs b/desktop/src/features/projects/project-channel-agent-context-contract.test.mjs
index ca2aa380d61..df1cbad523c 100644
--- a/desktop/src/features/projects/project-channel-agent-context-contract.test.mjs
+++ b/desktop/src/features/projects/project-channel-agent-context-contract.test.mjs
@@ -90,7 +90,7 @@ test("each send resolves the current path rather than caching an older link", ()
   assert.equal(after.localPath, "/Users/oscar/Projects/Nuncio Crew v2");
 });
 
-test("agent context names the absolute workspace but keeps cwd unchanged", () => {
+test("agent context encodes the source workspace for per-thread provisioning", () => {
   const context = projectContextForChannel(CHANNEL_ID, [project()]);
   const outgoing = appendProjectChannelAgentContext(
     "Inspect the tests.",
@@ -100,8 +100,8 @@ test("agent context names the absolute workspace but keeps cwd unchanged", () =>
   assert.match(outgoing, /Inspect the tests\./);
   assert.match(outgoing, /30617:/);
   assert.match(outgoing, /\/Users\/oscar\/Projects\/Nuncio Crew/);
-  assert.match(outgoing, /absolute path/i);
-  assert.match(outgoing, /session\/new\.cwd.+unchanged/i);
+  assert.match(outgoing, /path=%2FUsers%2Foscar%2FProjects%2FNuncio%20Crew/);
+  assert.match(outgoing, /isolated worktree per thread/i);
 });
 
 test("machine context is invisible in rendered CommonMark", () => {
diff --git a/desktop/src/features/projects/project-local-workspace-live-relay.test.mjs b/desktop/src/features/projects/project-local-workspace-live-relay.test.mjs
index 4e96b170c40..ee1cb6fc074 100644
--- a/desktop/src/features/projects/project-local-workspace-live-relay.test.mjs
+++ b/desktop/src/features/projects/project-local-workspace-live-relay.test.mjs
@@ -173,7 +173,10 @@ test("links and relinks a Project through a real Buzz relay", {
     );
     assert.match(agentMessage, /30617:/);
     assert.match(agentMessage, /Nuncio Crew 二/);
-    assert.match(agentMessage, /session\/new\.cwd remains unchanged/);
+    assert.match(
+      agentMessage,
+      /harness provisions one isolated worktree per thread/,
+    );
     assert.doesNotMatch(agentMessage, /Đồ án/);
   } finally {
     relay.close();
diff --git a/desktop/src/shared/api/agentControl.ts b/desktop/src/shared/api/agentControl.ts
index 677f0ffad49..1e49a2cfcee 100644
--- a/desktop/src/shared/api/agentControl.ts
+++ b/desktop/src/shared/api/agentControl.ts
@@ -4,10 +4,14 @@ import type { CancelManagedAgentTurnResult } from "@/shared/api/types";
 export async function cancelManagedAgentTurn(
   pubkey: string,
   channelId: string,
+  conversationId: string,
+  turnId: string,
 ): Promise<CancelManagedAgentTurnResult> {
   await sendAgentObserverControl(pubkey, {
     type: "cancel_turn",
     channelId,
+    conversationId,
+    turnId,
   });
   return { status: "sent" };
 }
@@ -21,11 +25,15 @@ export async function cancelManagedAgentTurn(
 export async function switchManagedAgentModel(
   pubkey: string,
   channelId: string,
+  conversationId: string,
+  turnId: string,
   modelId: string,
 ): Promise<void> {
   await sendAgentObserverControl(pubkey, {
     type: "switch_model",
     channelId,
+    conversationId,
+    turnId,
     modelId,
   });
 }
diff --git a/desktop/src/shared/api/types.ts b/desktop/src/shared/api/types.ts
index 689c400b03c..09c78dace2c 100644
--- a/desktop/src/shared/api/types.ts
+++ b/desktop/src/shared/api/types.ts
@@ -480,10 +480,8 @@ export type CancelManagedAgentTurnResult = {
 };
 
 /**
- * Outcome of a live `switch_model` control frame, surfaced asynchronously via
- * the agent's `control_result` observer frame. Busy path: `sent` (cancel +
- * requeue on the new model) or `turn_ending` (oneshot already consumed this
- * turn). Idle path: `switched`, `unsupported_model`, or `no_active_turn`.
+ * Outcome of a live `switch_model` control frame. Busy: `sent`/`turn_ending`.
+ * Idle: `switched`, `unsupported_model`, or `no_active_turn`.
  */
 export type SwitchManagedAgentModelStatus =
   | "sent"
@@ -496,6 +494,8 @@ export type ControlResultFrame = {
   type: "cancel_turn" | "switch_model";
   status: string;
   modelId?: string;
+  conversationId?: string | null;
+  turnId?: string | null;
 };
 
 export type GitBashPrerequisite = {
diff --git a/docs/crew/DECISIONS.md b/docs/crew/DECISIONS.md
index be417df8d4d..247f92c9189 100644
--- a/docs/crew/DECISIONS.md
+++ b/docs/crew/DECISIONS.md
@@ -250,3 +250,25 @@ workspaces.
 Buzz workflow source files remain unchanged for upstream synchronization.
 Inherited automatic workflows are disabled in GitHub repository state only
 after the additive Crew gate passes, and can be re-enabled as rollback.
+
+## D-018 — Scope managed agent execution by Project thread
+
+- **Status:** Accepted
+- **Date:** 2026-07-31
+
+Each non-DM channel thread owns an independent ACP queue/session identity.
+Top-level event IDs establish that identity; NIP-10 replies reuse the root ID.
+The real NIP-29 channel remains separate for relay queries, reactions,
+observer frames, membership cleanup, and thread-scoped typing indicators.
+
+For an owner-authored Project task, Crew encodes the linked source workspace
+as hidden composer metadata. Before a new ACP session, the harness validates
+the source Git repository and creates one deterministic worktree and branch
+from the immutable thread-root event ID. All agents handed work inside that
+thread converge on the same path. Invalid metadata or worktree failure stops
+the task instead of falling back to the source checkout.
+
+A new multi-agent Project task notifies only the first explicitly ordered
+agent. Later agents remain visible through non-notifying reference tags and
+are woken by explicit mentions in subsequent thread replies. Ordinary chat,
+single-agent prompts, DMs, and non-Project channels keep existing routing.
diff --git a/docs/crew/LOCAL-BUILD.md b/docs/crew/LOCAL-BUILD.md
index 360c9b80385..f7966637bf7 100644
--- a/docs/crew/LOCAL-BUILD.md
+++ b/docs/crew/LOCAL-BUILD.md
@@ -72,17 +72,21 @@ The relay is authoritative, but no separate CLI registration is required:
 7. Quit and reopen NuncioCrew.
 8. Return to **Projects** and confirm the same Project is reconstructed from
    the relay.
-9. Mention an agent in the Project's bound channel and ask it to inspect a
-   harmless file by absolute path.
+9. Start two top-level task messages in the Project channel and mention an
+   agent in each. Confirm both thread activity indicators can run together.
+10. In one task, mention a second agent in a reply and confirm it sees the
+    first agent's changes in the same worktree.
 
 Expected result:
 
 - Project identity remains `(pubkey, identifier)`;
 - the local path is location metadata on kind `30617`;
-- no Git remote is inspected and no `clone` tag is fabricated in this slice;
-- the selected folder is not cloned, initialized, or modified;
-- `session/new.cwd` remains unchanged;
-- the agent receives the path through Project-channel context.
+- no `clone` tag is fabricated and the selected source checkout is not used
+  as an agent write target;
+- Crew creates a sibling `.buzz-worktrees` directory with one deterministic
+  worktree per thread root;
+- `session/new.cwd` is the thread worktree for Project tasks;
+- ordinary chat and non-Project sessions retain the process cwd.
 
 ## Rebuild checks
 
diff --git a/docs/crew/STATE.md b/docs/crew/STATE.md
index b38426f944f..cf55b2b2aa7 100644
--- a/docs/crew/STATE.md
+++ b/docs/crew/STATE.md
@@ -24,21 +24,21 @@ In scope:
 - Project create and update through the existing kind `30617` relay lifecycle;
 - folder-first `+ → Repository` creation in the Projects page;
 - canonical `buzz-channel` binding and relay acknowledgement;
-- Project-channel context containing the absolute path;
+- Project-channel context containing the absolute source path;
+- per-thread ACP scheduling and isolated Git worktree cwd;
+- ordered multi-agent Project task routing through normal composer mentions;
 - one-machine, one-manager use;
 - provider compatibility through existing ACP paths.
 
 Out of scope for this slice:
 
-- changing `session/new.cwd`;
-- a per-Project Rust dispatcher;
-- Git repository or worktree management;
+- a per-Project Rust dispatcher outside the ACP harness;
 - clone, init, fetch, pull, push, branch, or remote validation;
 - commit-diff loading for an exact linked workspace;
 - board implementation;
 - mobile;
 - multi-user local-path sharing;
-- final mention model syntax.
+- automatic semantic branch renaming after an agent proposes a human title;
 - Windows drive and UNC workspace paths.
 
 ## Local desktop build
@@ -93,7 +93,8 @@ Out of scope for this slice:
 
 ## Verified evidence
 
-- `buzz-acp` currently captures one process cwd for its prompt context.
+- `buzz-acp` uses the process cwd for ordinary sessions and one validated,
+  deterministic worktree cwd for each owner-authored Project task thread.
 - Project announcements already support `buzz-channel` binding.
 - `buzz-dev-mcp` accepts absolute paths and shell `workdir`.
 - Codex, Claude Code, Cursor, and Devin all completed an isolated absolute
@@ -154,8 +155,18 @@ publishing a new real relay event, which was intentionally not done.
   checkout collision isolation, empty-state create access, Markdown isolation,
   live relay reconstruction, exact local path resolution, mismatch rejection,
   no fallback, and truthful Local source state.
-- Latest full desktop suite: `3863` passed, `1` gated live-relay test skipped,
+- Latest full desktop suite: `3873` passed, `1` gated live-relay test skipped,
   zero failed.
+- Full `just ci` passed on the thread-worktree orchestration branch, including
+  Rust workspace tests, `1905` native desktop tests (`14` ignored), `906`
+  mobile tests (`1` skipped), frontend builds, lint, typecheck, and formatting.
+- Focused browser verification passed `62` Project composer, mention,
+  messaging, thread-anchor, and boot-flow scenarios against the E2E bridge.
+- Exact observer controls preserve `conversationId` and `turnId`; concurrent
+  same-channel turns can be stopped independently, and unconfirmed model
+  switches no longer surface as successful.
+- The local macOS arm64 bundle was built and ad-hoc signed at
+  `desktop/src-tauri/target/aarch64-apple-darwin/release/bundle/macos/NuncioCrew Local.app`.
 - Earlier focused live relay test: `1/1` passed with an isolated Buzz relay.
 - Typecheck, file-size gate, Biome checks, production build, and
   `git diff --check` passed.
diff --git a/plans/20260731-0200-thread-worktree-orchestration/plan.md b/plans/20260731-0200-thread-worktree-orchestration/plan.md
new file mode 100644
index 00000000000..7dce473f5d7
--- /dev/null
+++ b/plans/20260731-0200-thread-worktree-orchestration/plan.md
@@ -0,0 +1,28 @@
+# Thread worktree orchestration
+
+- **Status:** Complete
+- **Date:** 2026-07-31
+
+## Goal
+
+Let one Project channel run several independent task threads concurrently,
+with one deterministic Git worktree shared by every agent handoff in a thread.
+
+## Phases
+
+1. [x] Key ACP queue, affinity, session, control, and typing state by thread.
+2. [x] Preserve the real NIP-29 channel for relay operations and UI activity.
+3. [x] Encode trusted Project workspace context in the normal composer.
+4. [x] Provision an idempotent, fail-closed worktree before `session/new`.
+5. [x] Wake the first ordered agent while retaining later agent references.
+6. [x] Complete desktop, Rust, concurrency, package, and CI verification.
+7. [x] Record release-ready evidence and complete handoff.
+
+## Invariants
+
+- Normal channel messages and single-agent mentions keep existing behavior.
+- No Project prompt button or modal; the composer remains the only entry point.
+- A thread root is the idempotency key; prompt text never names a worktree.
+- Project workspace metadata is trusted only when authored by the agent owner.
+- Provisioning failure never falls back to the shared source checkout.
+- Later agent handoffs resolve the same thread root and therefore the same cwd.