diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs
index c493560c93..d2161a45e5 100644
--- a/crates/buzz-acp/src/lib.rs
+++ b/crates/buzz-acp/src/lib.rs
@@ -635,6 +635,9 @@ struct QueuedNormalListenerEvent {
     /// The admitted event as a native steer would render it, including an
     /// edit's resolved original-message routing.
     steer_event: queue::BatchEvent,
+    /// The listener's DM classification of the channel (unresolved counts as
+    /// a DM); see [`EventQueue::in_flight_accepts_steer`].
+    channel_is_dm: bool,
 }
 
 impl QueuedNormalListenerEvent {
@@ -664,13 +667,18 @@ impl QueuedNormalListenerEvent {
             return;
         };
         // A native steer keeps the running turn's `<context>`, so it may only
-        // carry a message that replies in the same thread. Under the channel
-        // policy one session spans threads; a message for another thread takes
-        // the cancel+merge path, whose re-prompt carries its own `<context>`.
-        let same_reply_thread = queue.in_flight_reply_thread(&self.scope)
-            == Some(self.steer_event.reply_thread().as_str());
+        // carry a message that replies in the same place. A channel-policy or
+        // DM session spans several destinations; a message for another one
+        // takes the cancel+merge path, whose re-prompt carries its own
+        // `<context>`. The DM rule applies only when the running turn's
+        // prompt itself was rendered as a DM.
+        let same_reply_route = queue.in_flight_accepts_steer(
+            &self.scope,
+            &self.steer_event.reply_route(),
+            self.channel_is_dm,
+        );
         let native_attempted = matches!(signal, ControlSignal::Steer)
-            && same_reply_thread
+            && same_reply_route
             && try_native_steer(
                 pool,
                 queue,
@@ -714,6 +722,7 @@ impl NormalListenerIngress {
         self,
         queue: &mut EventQueue,
         session_scope: scope::SessionScope,
+        channel_is_dm: bool,
     ) -> QueuedNormalListenerEvent {
         let Self {
             buzz_event,
@@ -744,6 +753,7 @@ impl NormalListenerIngress {
             effective_author,
             reaction_target_id,
             steer_event,
+            channel_is_dm,
         }
     }
 }
@@ -3521,14 +3531,13 @@ async fn run_harness(
                             // channel-keyed routing. Telemetry only for now —
                             // queue/pool partitioning by scope lands in a
                             // follow-up (see ticket outline steps 2–4).
-                            let session_scope = ingress.session_scope(
-                                config.session_policy,
-                                is_dm_channel(
-                                    ingress.buzz_event.channel_id,
-                                    &ctx.channel_info,
-                                )
-                                .await,
-                            );
+                            let channel_is_dm = is_dm_channel(
+                                ingress.buzz_event.channel_id,
+                                &ctx.channel_info,
+                            )
+                            .await;
+                            let session_scope =
+                                ingress.session_scope(config.session_policy, channel_is_dm);
                             tracing::debug!(
                                 channel_id = %session_scope.channel_id(),
                                 scope = %session_scope.telemetry_label(),
@@ -3537,7 +3546,7 @@ async fn run_harness(
                                 policy = %config.session_policy,
                                 "admitted event — resolved session scope"
                             );
-                            let queued = ingress.push(&mut queue, session_scope);
+                            let queued = ingress.push(&mut queue, session_scope, channel_is_dm);
                             // 👀 — immediate "seen" reaction, only if the event
                             // was actually queued (not dropped by DedupMode::Drop).
                             // Fire-and-forget: on rare fast-failure paths the
@@ -4350,8 +4359,9 @@ fn try_native_steer(
     // channel context and the actor's profile in the original prompt,
     // duplicating it here would defeat the point of non-cancelling
     // steering (which is to inject only what's new).
-    // The caller steers natively only a message in the running turn's reply
-    // thread, so the turn's own `<context>` still routes the reply. An edit's
+    // The caller steers natively only a message that replies where the
+    // running turn replies (`ReplyRoute::accepts_steer`), so the turn's own
+    // `<context>` still routes the reply. An edit's
     // block names its original (`Edit of:`) and the original's thread root.
     let event_id_hex = be.event.id.to_hex();
     let body = native_steer_body(channel_id, &be);
@@ -4873,6 +4883,9 @@ fn dispatch_pending(
             .state
             .set_scope_owner_generation(scope.clone(), owner_generation);
 
+        // The prompt task records how it classified the channel, for the
+        // native-steer guard (`EventQueue::in_flight_accepts_steer`).
+        let prompt_dm = queue.in_flight_prompt_dm(&scope).unwrap_or_default();
         let abort_handle = pool.join_set.spawn(async move {
             pool::run_prompt_task(
                 agent,
@@ -4882,6 +4895,7 @@ fn dispatch_pending(
                 result_tx,
                 Some(control_rx),
                 task_turn_id,
+                prompt_dm,
             )
             .await;
         });
@@ -5603,6 +5617,7 @@ fn dispatch_heartbeat(
             result_tx,
             None,
             task_turn_id,
+            Default::default(),
         )
         .await;
     });
@@ -10016,7 +10031,7 @@ mod edit_native_steer_tests {
         };
         let scope = ingress.session_scope(scope::SessionPolicy::Channel, false);
         let mut queue = EventQueue::new(config::DedupMode::Queue);
-        let queued = ingress.push(&mut queue, scope);
+        let queued = ingress.push(&mut queue, scope, false);
         assert_eq!(queued.steer_event.edit, Some(resolved));
         assert_eq!(queued.reaction_target_id, original.id.to_hex());
     }
@@ -10028,6 +10043,26 @@ mod edit_native_steer_tests {
     fn steer_edit_into_running_turn(
         running_event: nostr::Event,
         original: &nostr::Event,
+    ) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
+        let edit = edit_event(&original.id.to_hex(), &[]);
+        let resolved = queue::ResolvedEdit {
+            target_event_id: original.id.to_hex(),
+            target_thread_tags: queue::parse_thread_tags(original),
+        };
+        steer_into_running_turn(false, Some(false), running_event, edit, Some(resolved))
+    }
+
+    /// Drive `incoming` through the listener's steer decision while a turn
+    /// for `running_event` is in flight in the same conversation session
+    /// (the channel policy, or any DM). `prompt_is_dm` is how the running
+    /// turn's prompt classified the channel; `None` means the prompt has not
+    /// been formatted yet.
+    fn steer_into_running_turn(
+        is_dm: bool,
+        prompt_is_dm: Option<bool>,
+        running_event: nostr::Event,
+        incoming: nostr::Event,
+        incoming_edit: Option<queue::ResolvedEdit>,
     ) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
         let channel_id = Uuid::new_v4();
         let ingress =
@@ -10044,10 +10079,16 @@ mod edit_native_steer_tests {
 
         let mut queue = EventQueue::new(config::DedupMode::Queue);
         let running = ingress(running_event, None);
-        let scope = running.session_scope(scope::SessionPolicy::Channel, false);
-        running.push(&mut queue, scope.clone());
+        let scope = running.session_scope(scope::SessionPolicy::Channel, is_dm);
+        running.push(&mut queue, scope.clone(), is_dm);
         queue.flush_next().expect("running turn");
         assert!(queue.is_scope_in_flight(&scope));
+        if let Some(prompt_is_dm) = prompt_is_dm {
+            queue
+                .in_flight_prompt_dm(&scope)
+                .expect("in-flight turn records its prompt classification")
+                .record(prompt_is_dm);
+        }
 
         let mut pool = AgentPool::from_slots(vec![]);
         let (control_tx, mut control_rx) = tokio::sync::oneshot::channel();
@@ -10067,20 +10108,15 @@ mod edit_native_steer_tests {
             },
         );
 
-        let edit = edit_event(&original.id.to_hex(), &[]);
-        let resolved = queue::ResolvedEdit {
-            target_event_id: original.id.to_hex(),
-            target_thread_tags: queue::parse_thread_tags(original),
-        };
-        let edit_ingress = ingress(edit, Some(resolved));
+        let incoming_ingress = ingress(incoming, incoming_edit);
         assert_eq!(
-            edit_ingress.session_scope(scope::SessionPolicy::Channel, false),
+            incoming_ingress.session_scope(scope::SessionPolicy::Channel, is_dm),
             scope,
-            "channel policy: one session spans every thread"
+            "one conversation session spans every thread"
         );
         let (ack_tx, _ack_rx) = mpsc::unbounded_channel();
-        edit_ingress
-            .push(&mut queue, scope.clone())
+        incoming_ingress
+            .push(&mut queue, scope.clone(), is_dm)
             .steer_or_interrupt(
                 MultipleEventHandling::Steer,
                 None,
@@ -10151,6 +10187,309 @@ mod edit_native_steer_tests {
         assert!(steer.is_none(), "no native steer across top-level threads");
         assert_eq!(control, Some(ControlSignal::Steer));
     }
+
+    /// A top-level DM message's `<context>` names no reply target, so a
+    /// second top-level DM message replies in the same place and is steered
+    /// natively into the running turn rather than cancelling it.
+    #[tokio::test]
+    async fn dm_top_level_follow_up_steers_natively() {
+        let (steer, control) =
+            steer_into_running_turn(true, Some(true), message(None), message(None), None);
+
+        assert!(steer.is_some(), "DM follow-up is sent as a native steer");
+        assert_eq!(
+            control, None,
+            "native steer must not cancel the running turn"
+        );
+    }
+
+    /// Where a dispatched turn is held when the follow-up arrives: the agent
+    /// never answers the first request with this method.
+    enum HoldAt {
+        /// Session setup, before the turn's own prompt is formatted.
+        SessionNew,
+        /// The turn's own prompt is running.
+        Prompt,
+    }
+
+    /// Dispatch a turn for `running_event` through the production
+    /// `dispatch_pending` → `run_prompt_task` path to an agent that holds at
+    /// `hold_at`, then drive `incoming` through the real listener admission.
+    /// `channel_type` is the channel's metadata (`"dm"` or `"stream"`), known
+    /// to the listener and the prompt alike. Returns the native steer request,
+    /// if any, and the control signal sent to the running turn, if any.
+    async fn steer_into_dispatched_turn(
+        channel_type: &str,
+        initial_message: Option<&str>,
+        hold_at: HoldAt,
+        running_event: nostr::Event,
+        incoming: nostr::Event,
+    ) -> (Option<pool::SteerRequest>, Option<ControlSignal>) {
+        let channel_id = Uuid::new_v4();
+        let is_dm = channel_type == "dm";
+        let capture = std::env::temp_dir().join(format!(
+            "buzz-acp-dispatched-steer-{}.ndjson",
+            Uuid::new_v4()
+        ));
+        let quoted_capture = capture.to_string_lossy().replace('\'', "'\\''");
+        let held_method = match hold_at {
+            HoldAt::SessionNew => "session/new",
+            HoldAt::Prompt => "session/prompt",
+        };
+        // Record each request and never answer: the turn stays in flight at
+        // its first request, which is `held_method`.
+        let script = format!(
+            r#"while IFS= read -r line; do printf '%s\n' "$line" >> '{quoted_capture}'; done"#
+        );
+        let acp = acp::AcpClient::spawn("bash", &["-c".into(), script], &[], false)
+            .await
+            .expect("spawn holding ACP");
+        let ingress = |event: nostr::Event| NormalListenerIngress {
+            buzz_event: relay::BuzzEvent {
+                connection_generation: 0,
+                channel_id,
+                event,
+            },
+            effective_author: "author".into(),
+            prompt_tag: "@mention".into(),
+            edit: None,
+        };
+        let running = ingress(running_event);
+        let scope = running.session_scope(scope::SessionPolicy::Channel, is_dm);
+        let mut agent = pool::OwnedAgent {
+            index: 0,
+            acp,
+            state: pool::SessionState::default(),
+            model_capabilities: None,
+            desired_model: None,
+            model_overridden: false,
+            desired_model_request_id: None,
+            desired_model_pending_ack: false,
+            startup_effort: None,
+            agent_name: "dispatched-steer-test-agent".into(),
+            goose_system_prompt_supported: None,
+            protocol_version: 1,
+        };
+        if matches!(hold_at, HoldAt::Prompt) {
+            agent
+                .state
+                .sessions
+                .insert(scope.clone(), "live-session".into());
+        }
+        // The relay has no newer metadata, project, or history: every query
+        // returns no events, so the turn keeps the startup channel metadata.
+        let relay = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind relay fixture");
+        let base_url = format!("http://{}", relay.local_addr().unwrap());
+        let relay = tokio::spawn(async move {
+            use tokio::io::{AsyncReadExt, AsyncWriteExt};
+            loop {
+                let (mut socket, _) = relay.accept().await.expect("accept relay query");
+                tokio::spawn(async move {
+                    let mut request = [0; 16384];
+                    let _ = socket.read(&mut request).await;
+                    let _ = socket
+                        .write_all(
+                            b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 2\r\nConnection: close\r\n\r\n[]",
+                        )
+                        .await;
+                });
+            }
+        });
+        let mut ctx = pool::tests::make_prompt_context_no_owner();
+        ctx.rest_client.base_url = base_url;
+        ctx.initial_message = initial_message.map(str::to_string);
+        ctx.channel_info = pool::ChannelInfoResolver::new(
+            HashMap::from([(
+                channel_id,
+                relay::ChannelInfo {
+                    name: "test-channel".into(),
+                    channel_type: channel_type.into(),
+                    description: None,
+                },
+            )]),
+            ctx.rest_client.clone(),
+        );
+        let ctx = Arc::new(ctx);
+        let mut pool = AgentPool::from_slots(vec![Some(agent)]);
+        let mut queue = EventQueue::new(config::DedupMode::Queue);
+        running.push(&mut queue, scope.clone(), is_dm);
+        let mut last_activity = tokio::time::Instant::now();
+        let dispatched = dispatch_pending(&mut pool, &mut queue, &ctx, &mut last_activity, None);
+        assert_eq!(dispatched.len(), 1, "the running turn is dispatched");
+
+        // The turn has passed every step before `held_method` once that
+        // request reaches the agent.
+        let held_request = format!("\"{held_method}\"");
+        tokio::time::timeout(Duration::from_secs(10), async {
+            while !std::fs::read_to_string(&capture)
+                .unwrap_or_default()
+                .contains(&held_request)
+            {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("dispatched turn reaches its held request");
+
+        // Observe the listener's decision through the running task's own
+        // steer and control channels.
+        let (control_tx, mut control_rx) = tokio::sync::oneshot::channel();
+        let (steer_tx, mut steer_rx) = tokio::sync::mpsc::channel(1);
+        let meta = pool
+            .task_map_mut()
+            .values_mut()
+            .find(|meta| meta.scope.as_ref() == Some(&scope))
+            .expect("running task");
+        meta.control_tx = Some(control_tx);
+        meta.steer_tx = Some(steer_tx);
+
+        let (ack_tx, _ack_rx) = mpsc::unbounded_channel();
+        ingress(incoming)
+            .push(&mut queue, scope.clone(), is_dm)
+            .steer_or_interrupt(
+                MultipleEventHandling::Steer,
+                None,
+                &mut pool,
+                &mut queue,
+                &ack_tx,
+            );
+        let steer = steer_rx.try_recv().ok();
+        let control = control_rx.try_recv().ok();
+
+        pool.join_set.abort_all();
+        while pool.join_set.join_next().await.is_some() {}
+        relay.abort();
+        let _ = std::fs::remove_file(&capture);
+        (steer, control)
+    }
+
+    /// The DM steer rule depends on the running prompt's own classification,
+    /// which travels from the queue turn through `dispatch_pending` into
+    /// `run_prompt_task`, which records it. Drive that production handoff
+    /// with DM channel metadata, then the real listener admission: a
+    /// top-level follow-up must be steered natively.
+    #[tokio::test]
+    async fn dispatched_dm_turn_records_classification_for_native_steer() {
+        let (steer, control) =
+            steer_into_dispatched_turn("dm", None, HoldAt::Prompt, message(None), message(None))
+                .await;
+
+        assert!(steer.is_some(), "DM follow-up is sent as a native steer");
+        assert_eq!(
+            control, None,
+            "native steer must not cancel the running turn"
+        );
+    }
+
+    /// The prompt task records its classification as soon as channel
+    /// metadata resolves, so a top-level DM follow-up during a slow
+    /// `session/new` is steered natively too.
+    #[tokio::test]
+    async fn dm_follow_up_during_session_setup_steers_natively() {
+        let (steer, control) = steer_into_dispatched_turn(
+            "dm",
+            None,
+            HoldAt::SessionNew,
+            message(None),
+            message(None),
+        )
+        .await;
+
+        assert!(steer.is_some(), "DM follow-up is sent as a native steer");
+        assert_eq!(
+            control, None,
+            "native steer must not cancel the running turn"
+        );
+    }
+
+    /// An `initial_message` setup prompt reads the same steer mailbox, so a
+    /// DM follow-up must not be admitted for native steering before that
+    /// setup turn finishes; it takes the cancel+merge path.
+    #[tokio::test]
+    async fn dm_follow_up_before_initial_message_cancels_and_merges() {
+        let (steer, control) = steer_into_dispatched_turn(
+            "dm",
+            Some("set up the session"),
+            HoldAt::SessionNew,
+            message(None),
+            message(None),
+        )
+        .await;
+
+        assert!(
+            steer.is_none(),
+            "no native steer into the initial_message setup turn"
+        );
+        assert_eq!(control, Some(ControlSignal::Steer));
+    }
+
+    /// A known channel's prompt is never rendered as a DM, so a same-thread
+    /// follow-up is steered natively before the turn's classification is
+    /// recorded, as before DM steering existed: here the prompt task is
+    /// still waiting on `session/new` ahead of an `initial_message`.
+    #[tokio::test]
+    async fn channel_same_thread_follow_up_during_setup_steers_natively() {
+        let root = "ab".repeat(32);
+        let (steer, control) = steer_into_dispatched_turn(
+            "stream",
+            Some("set up the session"),
+            HoldAt::SessionNew,
+            message(Some(&root)),
+            message(Some(&root)),
+        )
+        .await;
+
+        assert!(steer.is_some(), "same-thread follow-up is a native steer");
+        assert_eq!(
+            control, None,
+            "native steer must not cancel the running turn"
+        );
+    }
+
+    /// A DM thread reply needs a `--reply-to` the running top-level turn's
+    /// `<context>` does not carry, so it takes the cancel+merge path.
+    #[tokio::test]
+    async fn dm_thread_reply_during_top_level_turn_cancels_and_merges() {
+        let root = "ab".repeat(32);
+        let (steer, control) =
+            steer_into_running_turn(true, Some(true), message(None), message(Some(&root)), None);
+
+        assert!(
+            steer.is_none(),
+            "no native steer into a different DM thread"
+        );
+        assert_eq!(control, Some(ControlSignal::Steer));
+    }
+
+    /// A DM turn whose prompt was rendered while channel metadata was
+    /// unavailable got channel-style `<context>`: its replies go to a thread
+    /// rooted at its trigger. The DM rule must not apply once metadata
+    /// resolves, so a top-level follow-up takes the cancel+merge path.
+    #[tokio::test]
+    async fn dm_follow_up_into_channel_formatted_turn_cancels_and_merges() {
+        let (steer, control) =
+            steer_into_running_turn(true, Some(false), message(None), message(None), None);
+
+        assert!(
+            steer.is_none(),
+            "steer guard follows the running prompt's classification"
+        );
+        assert_eq!(control, Some(ControlSignal::Steer));
+    }
+
+    /// Before the prompt task records its classification, nothing proves
+    /// which `<context>` a turn in a DM (or a channel of unresolved type)
+    /// will carry, so the follow-up takes the cancel+merge path.
+    #[tokio::test]
+    async fn dm_follow_up_before_prompt_classification_cancels_and_merges() {
+        let (steer, control) =
+            steer_into_running_turn(true, None, message(None), message(None), None);
+
+        assert!(steer.is_none(), "no native steer without a recorded prompt");
+        assert_eq!(control, Some(ControlSignal::Steer));
+    }
 }
 
 #[cfg(test)]
diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs
index 1278c17d75..992d894d59 100644
--- a/crates/buzz-acp/src/pool.rs
+++ b/crates/buzz-acp/src/pool.rs
@@ -2411,6 +2411,7 @@ fn send_prompt_result(
 ///
 /// The agent is ALWAYS returned — even on panic the `JoinSet` detects the
 /// abort and the caller uses `task_map` to recover the agent index.
+#[allow(clippy::too_many_arguments)]
 pub async fn run_prompt_task(
     mut agent: OwnedAgent,
     batch: Option<FlushBatch>,
@@ -2419,6 +2420,7 @@ pub async fn run_prompt_task(
     result_tx: mpsc::UnboundedSender<PromptResult>,
     control_rx: Option<tokio::sync::oneshot::Receiver<ControlSignal>>,
     turn_id: String,
+    prompt_dm: crate::queue::PromptDmClassification,
 ) {
     // Is this a channel prompt or a heartbeat?
     let source = match &batch {
@@ -2528,6 +2530,21 @@ pub async fn run_prompt_task(
         },
         PromptSource::Heartbeat => None,
     };
+    // Whether this turn's prompt renders the channel as a DM. `format_prompt`
+    // derives the same value from `resolved_channel_info`, which is fixed
+    // from here on, so the main loop's native-steer guard can rely on it now
+    // rather than after the session setup below. A follow-up steered before
+    // this turn's own prompt starts waits in the steer mailbox, so it must
+    // not be admitted while an `initial_message` setup prompt (which reads the
+    // same mailbox) is still to come; that case records after the setup turn.
+    let prompt_is_dm = resolved_channel_info
+        .as_ref()
+        .is_some_and(|info| info.channel_type == "dm");
+    let initial_message_pending = ctx.initial_message.is_some()
+        && matches!(&source, PromptSource::Channel(scope) if !agent.state.sessions.contains_key(scope));
+    if !initial_message_pending {
+        prompt_dm.record(prompt_is_dm);
+    }
 
     //
     // Core memory is delivered inside the system prompt the harness already
@@ -2985,6 +3002,8 @@ pub async fn run_prompt_task(
             }
         }
     }
+    // Any `initial_message` setup turn is done; see `prompt_is_dm`.
+    prompt_dm.record(prompt_is_dm);
 
     // When the batch is a single slash-command message (e.g. "@Eva /goal …"),
     // `slash_command` holds the bare command. It is sent as the FIRST prompt
@@ -3027,10 +3046,7 @@ pub async fn run_prompt_task(
         // reuse that exact typed result for prompt formatting.
         let channel_info = resolved_channel_info.clone();
 
-        let is_dm = channel_info
-            .as_ref()
-            .map(|info| info.channel_type == "dm")
-            .unwrap_or(false);
+        let is_dm = prompt_is_dm;
         let context_target = resolve_context_target(b, is_dm);
         let hydrated_thread_root = match &context_target {
             ContextTarget::Thread(root) => Some(root),
@@ -5660,7 +5676,7 @@ async fn clear_reactions(rest: crate::relay::RestClient, event_ids: Vec<String>)
 }
 
 #[cfg(test)]
-mod tests {
+pub(crate) mod tests {
     use super::*;
     use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp};
     use serde_json::json;
@@ -7134,6 +7150,7 @@ mod tests {
                 result_tx,
                 None,
                 "observer-test-turn".into(),
+                Default::default(),
             ));
             let liveness = tokio::time::timeout(Duration::from_secs(5), async {
                 request_rx.await.expect("channel context lookup started");
@@ -7227,6 +7244,7 @@ done"#
                 result_tx.clone(),
                 None,
                 format!("turn-{turn}"),
+                Default::default(),
             )
             .await;
             let result = result_rx.recv().await.expect("prompt result");
@@ -7350,6 +7368,7 @@ done"#
                 result_tx.clone(),
                 None,
                 format!("turn-{turn}"),
+                Default::default(),
             )
             .await;
             let result = result_rx.recv().await.expect("prompt result");
@@ -7563,6 +7582,7 @@ done"#
             result_tx.clone(),
             None,
             "first-turn".into(),
+            Default::default(),
         )
         .await;
         let first_result = result_rx.recv().await.expect("first prompt result");
@@ -7582,6 +7602,7 @@ done"#
             result_tx,
             None,
             "follow-up-turn".into(),
+            Default::default(),
         )
         .await;
         let mut result = result_rx.recv().await.expect("prompt result");
@@ -7752,6 +7773,7 @@ done"#
                 result_tx.clone(),
                 None,
                 turn_id.into(),
+                Default::default(),
             )
             .await;
             let result = result_rx.recv().await.expect("prompt result");
@@ -7916,6 +7938,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             result_tx,
             None,
             "next-turn".into(),
+            Default::default(),
         )
         .await;
         let mut result = result_rx.recv().await.expect("next prompt result");
@@ -10402,7 +10425,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
         );
     }
 
-    pub(super) fn make_prompt_context_no_owner() -> PromptContext {
+    pub(crate) fn make_prompt_context_no_owner() -> PromptContext {
         let agent_keys = nostr::Keys::generate();
         make_prompt_context_impl(&agent_keys, None)
     }
@@ -11140,6 +11163,7 @@ done"#
             result_tx,
             None,
             "indeterminate-project-turn".into(),
+            Default::default(),
         )
         .await;
 
diff --git a/crates/buzz-acp/src/queue.rs b/crates/buzz-acp/src/queue.rs
index a5b4b9d3f3..3f2ea34dbb 100644
--- a/crates/buzz-acp/src/queue.rs
+++ b/crates/buzz-acp/src/queue.rs
@@ -15,6 +15,7 @@
 
 use nostr::{Event, ToBech32};
 use std::collections::{HashMap, HashSet, VecDeque};
+use std::sync::{Arc, OnceLock};
 use std::time::{Duration, Instant};
 use uuid::Uuid;
 
@@ -154,6 +155,81 @@ impl BatchEvent {
     pub fn reply_thread(&self) -> String {
         reply_thread(&self.event, self.edit.as_ref())
     }
+
+    /// See [`ReplyRoute`].
+    pub fn reply_route(&self) -> ReplyRoute {
+        ReplyRoute {
+            root_event_id: self
+                .routing_thread_tags()
+                .root_event_id
+                .map(|root| root.to_ascii_lowercase()),
+            thread: self.reply_thread(),
+        }
+    }
+}
+
+/// Where a turn's `<context>` sends ordinary replies, as far as the
+/// native-steer guard needs to know.
+///
+/// A native steer adds a message to a running turn without a new `<context>`,
+/// so the turn keeps replying where its own `<context>` points. A message may
+/// be steered natively only when that destination is also right for it.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ReplyRoute {
+    /// Routed thread root, lowercase; `None` for a top-level message.
+    root_event_id: Option<String>,
+    /// See [`reply_thread`].
+    thread: String,
+}
+
+impl ReplyRoute {
+    /// Whether a turn whose `<context>` was rendered for `self` also replies
+    /// correctly to `incoming`.
+    ///
+    /// In a channel, a top-level message is answered in a new thread rooted at
+    /// that message, so two messages share a destination only when they share
+    /// a [`reply_thread`]. In a DM, a top-level message's `<context>` names no
+    /// reply target (see `format_prompt`): every top-level DM message replies
+    /// at the top of the same conversation. DM messages therefore share a
+    /// destination when they share a thread root, including none.
+    pub fn accepts_steer(&self, incoming: &ReplyRoute, is_dm: bool) -> bool {
+        if is_dm {
+            self.root_event_id == incoming.root_event_id
+        } else {
+            self.thread == incoming.thread
+        }
+    }
+}
+
+/// Whether the running turn's prompt rendered its channel as a DM.
+///
+/// The prompt task records this when it formats the turn's `<context>`
+/// (`format_prompt`'s `is_dm`). The steer guard reads the recorded value
+/// instead of classifying the incoming event again: channel metadata can
+/// become resolvable after the turn starts, and the guard must apply the rule
+/// that matches the `<context>` the agent actually received.
+#[derive(Debug, Clone, Default)]
+pub struct PromptDmClassification(Arc<OnceLock<bool>>);
+
+impl PromptDmClassification {
+    /// Record the prompt's classification. Later calls are ignored: a turn has
+    /// one `<context>`.
+    pub fn record(&self, is_dm: bool) {
+        let _ = self.0.set(is_dm);
+    }
+
+    /// The recorded classification; `None` before the prompt is formatted.
+    pub fn get(&self) -> Option<bool> {
+        self.0.get().copied()
+    }
+}
+
+/// Reply destination of an in-flight turn, and how its prompt classified the
+/// channel.
+#[derive(Debug, Clone)]
+struct InFlightReplyRoute {
+    route: ReplyRoute,
+    prompt_is_dm: PromptDmClassification,
 }
 
 /// Why a batch's prior turn was cancelled — controls how `format_prompt`
@@ -241,9 +317,9 @@ pub struct EventQueue {
     in_flight_deadlines: HashMap<SessionScope, Instant>,
     /// Number of events in each in-flight batch (for expiry logging).
     in_flight_batch_sizes: HashMap<SessionScope, usize>,
-    /// Reply thread of each in-flight turn: the [`reply_thread`] of the batch
+    /// Reply route of each in-flight turn: the [`ReplyRoute`] of the batch
     /// event whose `<context>` routes the turn's replies (its last event).
-    in_flight_reply_threads: HashMap<SessionScope, String>,
+    in_flight_reply_routes: HashMap<SessionScope, InFlightReplyRoute>,
     retry_after: HashMap<SessionScope, Instant>,
     /// Per-scope retry attempt counter for exponential backoff / dead-lettering.
     retry_counts: HashMap<SessionScope, u32>,
@@ -285,7 +361,7 @@ impl EventQueue {
             in_flight_scopes: HashSet::new(),
             in_flight_deadlines: HashMap::new(),
             in_flight_batch_sizes: HashMap::new(),
-            in_flight_reply_threads: HashMap::new(),
+            in_flight_reply_routes: HashMap::new(),
             retry_after: HashMap::new(),
             retry_counts: HashMap::new(),
             dedup_mode,
@@ -433,7 +509,7 @@ impl EventQueue {
             );
             self.in_flight_scopes.remove(&scope);
             self.in_flight_deadlines.remove(&scope);
-            self.in_flight_reply_threads.remove(&scope);
+            self.in_flight_reply_routes.remove(&scope);
             // Recover any withheld goose-native steer events for the expired
             // scope back to the queue front so normal dispatch delivers
             // them. Unlike the in-flight batch above (already delivered to a
@@ -477,7 +553,7 @@ impl EventQueue {
                             .insert(scope.clone(), now + self.in_flight_deadline);
                         self.in_flight_batch_sizes
                             .insert(scope.clone(), cancelled.len());
-                        self.record_in_flight_reply_thread(&scope, &cancelled);
+                        self.record_in_flight_reply_route(&scope, &cancelled);
                         return Some(FlushBatch {
                             channel_id: scope.channel_id(),
                             scope,
@@ -515,7 +591,7 @@ impl EventQueue {
             .insert(scope.clone(), now + self.in_flight_deadline);
         self.in_flight_batch_sizes
             .insert(scope.clone(), events.len());
-        self.record_in_flight_reply_thread(&scope, &events);
+        self.record_in_flight_reply_route(&scope, &events);
 
         // Merge any cancelled events stored by requeue_as_cancelled().
         let cancelled_events = self.cancelled_batches.remove(&scope).unwrap_or_default();
@@ -568,7 +644,7 @@ impl EventQueue {
         self.in_flight_scopes.remove(&scope);
         self.in_flight_deadlines.remove(&scope);
         self.in_flight_batch_sizes.remove(&scope);
-        self.in_flight_reply_threads.remove(&scope);
+        self.in_flight_reply_routes.remove(&scope);
         let now = Instant::now();
         match self.retry_after.get(&scope) {
             // Active throttle → scope was requeued; keep retry_counts intact.
@@ -787,7 +863,7 @@ impl EventQueue {
             );
             self.in_flight_scopes.remove(&scope);
             self.in_flight_deadlines.remove(&scope);
-            self.in_flight_reply_threads.remove(&scope);
+            self.in_flight_reply_routes.remove(&scope);
             // Symmetric with the flush_next expiry block: recover withheld
             // goose-native steer events for the expired scope so they are
             // not permanently orphaned in the side table.
@@ -943,25 +1019,58 @@ impl EventQueue {
         self.in_flight_scopes.contains(&scope.into_scope())
     }
 
-    /// The reply thread of the turn in flight for `scope`, if any.
+    /// Whether a message with reply route `incoming` may be steered natively
+    /// into the turn in flight for `scope`.
     ///
-    /// A native steer adds a message to a running turn without a new
-    /// `<context>`, so the turn keeps replying where its own `<context>`
-    /// points. A message whose [`reply_thread`] differs (possible under the
-    /// channel session policy) must not be steered natively; the cancel+merge
-    /// path re-dispatches it with its own full `<context>`.
-    pub fn in_flight_reply_thread(&self, scope: &SessionScope) -> Option<&str> {
-        self.in_flight_reply_threads.get(scope).map(String::as_str)
+    /// A message whose route the running turn does not accept (see
+    /// [`ReplyRoute::accepts_steer`]; possible whenever one session spans
+    /// several reply destinations: the channel session policy, or a DM) must
+    /// not be steered natively; the cancel+merge path re-dispatches it with
+    /// its own full `<context>`.
+    ///
+    /// The rule follows how the running turn's prompt classified the channel
+    /// (see [`PromptDmClassification`]). Until the prompt task records that,
+    /// `channel_is_dm` is the listener's own classification, where unresolved
+    /// metadata counts as a DM. A channel the listener knows is not a DM also
+    /// renders as one in the prompt, so the channel rule applies at once.
+    /// Otherwise the prompt may render either way, and the answer is `false`.
+    pub fn in_flight_accepts_steer(
+        &self,
+        scope: &SessionScope,
+        incoming: &ReplyRoute,
+        channel_is_dm: bool,
+    ) -> bool {
+        let Some(running) = self.in_flight_reply_routes.get(scope) else {
+            return false;
+        };
+        match running.prompt_is_dm.get() {
+            Some(is_dm) => running.route.accepts_steer(incoming, is_dm),
+            None if !channel_is_dm => running.route.accepts_steer(incoming, false),
+            None => false,
+        }
     }
 
-    fn record_in_flight_reply_thread(&mut self, scope: &SessionScope, events: &[BatchEvent]) {
+    /// The cell in which the prompt task for `scope`'s in-flight turn records
+    /// its DM classification (see [`PromptDmClassification`]).
+    pub fn in_flight_prompt_dm(&self, scope: &SessionScope) -> Option<PromptDmClassification> {
+        self.in_flight_reply_routes
+            .get(scope)
+            .map(|running| running.prompt_is_dm.clone())
+    }
+
+    fn record_in_flight_reply_route(&mut self, scope: &SessionScope, events: &[BatchEvent]) {
         match events.last() {
             Some(last) => {
-                self.in_flight_reply_threads
-                    .insert(scope.clone(), last.reply_thread());
+                self.in_flight_reply_routes.insert(
+                    scope.clone(),
+                    InFlightReplyRoute {
+                        route: last.reply_route(),
+                        prompt_is_dm: PromptDmClassification::default(),
+                    },
+                );
             }
             None => {
-                self.in_flight_reply_threads.remove(scope);
+                self.in_flight_reply_routes.remove(scope);
             }
         }
     }
@@ -1216,8 +1325,8 @@ pub(crate) fn reaction_target_id(event: &Event) -> String {
 /// the routed event itself when it is top-level (a reply opens a thread
 /// rooted there). Lowercase, so equivalent hex spellings compare equal.
 ///
-/// This is the thread-session key, and it decides whether a message may be
-/// steered natively into a running turn (see [`EventQueue::in_flight_reply_thread`]).
+/// This is the thread-session key. Outside DMs it also decides whether a
+/// message may be steered natively into a running turn (see [`ReplyRoute`]).
 pub(crate) fn reply_thread(event: &Event, edit: Option<&ResolvedEdit>) -> String {
     routing_thread_tags(event, edit)
         .root_event_id
@@ -2220,6 +2329,7 @@ pub fn format_prompt(batch: &FlushBatch, args: &FormatPromptArgs<'_>) -> Vec<Str
     //   - top-level     → anchor to the triggering event (it becomes the root)
     // Agent↔agent turns get no forced anchor — deep nesting is intentional
     // there. DMs are always 1:1 with a human, so they always anchor.
+    // `ReplyRoute::accepts_steer` mirrors this DM rule; keep them in sync.
     let sender_pubkey = last_event.event.pubkey.to_hex();
     let reply_anchor = if is_dm {
         thread_tags
@@ -2396,6 +2506,50 @@ pub(crate) fn native_steer_framing() -> (&'static str, &'static str) {
     (framing.new_tag, framing.closing_note)
 }
 
+#[cfg(test)]
+mod reply_route_tests {
+    use super::ReplyRoute;
+
+    fn route(root: Option<&str>, thread: &str) -> ReplyRoute {
+        ReplyRoute {
+            root_event_id: root.map(str::to_string),
+            thread: thread.to_string(),
+        }
+    }
+
+    /// A top-level message (no root) whose replies open thread `id`.
+    fn top_level(id: &str) -> ReplyRoute {
+        route(None, id)
+    }
+
+    /// A reply in the thread rooted at `root`.
+    fn in_thread(root: &str) -> ReplyRoute {
+        route(Some(root), root)
+    }
+
+    #[test]
+    fn channel_steers_only_within_one_reply_thread() {
+        assert!(in_thread("a").accepts_steer(&in_thread("a"), false));
+        assert!(!in_thread("a").accepts_steer(&in_thread("b"), false));
+        // Replies to a top-level message open a thread rooted at it.
+        assert!(top_level("a").accepts_steer(&in_thread("a"), false));
+        assert!(in_thread("a").accepts_steer(&top_level("a"), false));
+        assert!(!top_level("a").accepts_steer(&top_level("b"), false));
+        assert!(!top_level("a").accepts_steer(&in_thread("b"), false));
+    }
+
+    #[test]
+    fn dm_steers_within_one_thread_or_the_top_level() {
+        // Every top-level DM message replies at the top of the conversation.
+        assert!(top_level("a").accepts_steer(&top_level("b"), true));
+        assert!(in_thread("a").accepts_steer(&in_thread("a"), true));
+        assert!(!in_thread("a").accepts_steer(&in_thread("b"), true));
+        // A top-level DM turn has no `--reply-to`; a threaded one does.
+        assert!(!top_level("a").accepts_steer(&in_thread("a"), true));
+        assert!(!in_thread("a").accepts_steer(&top_level("b"), true));
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;