fix(runtime): report the steer the engine actually took, not the one it was sent (#6276)

Steering was the only way to redirect a running turn, and the runtime API
reported every steer as delivered before the engine decided anything. It
persisted the item as already-`Completed` and emitted `turn.steered` +
`item.completed` the moment the text entered the mailbox. The engine then
discards any steer whose turn has moved on, and an interrupted or failed turn
drops whatever it had queued — so a GUI showed "Guidance sent", cleared the
composer, and the words were gone. The Ratatui path was made honest in #6190;
the layer serving GPUI, ACP and SDK clients never was.

The engine now answers for each steer. `SteerInput` carries an optional
verdict channel, and `SteerInput::into_pending` is the one way to claim a
steer off it. The resulting `PendingSteer` reports `Accepted` from `commit()`
and `Dropped` from `Drop`, so every exit path — wrong turn, empty text,
interrupt, stream failure, early return, the engine itself going away —
settles exactly once without anyone having to hunt for it.

`steer_turn` persists the item `queued` and hands the verdict channel to a
detached settler, never to the caller: a cancelled request or a timed-out one
therefore cannot orphan the verdict and strand the item. On `Accepted` the
item becomes `completed`, `steer_count` rises, and the existing events fire.
On `Dropped` it becomes `canceled` and `turn.steer_dropped` carries the text,
a reason, and the settled item, so a client can resend instead of clearing a
composer over guidance nobody read. The API observes that settlement for two
seconds — a steer into a streaming turn settles in milliseconds — and behind a
long tool call falls back to the honest `queued` receipt rather than hanging
for the length of the tool call. A dropped steer maps to 409.

History rebuild honours the same record: a `queued` or `canceled` user message
is skipped, because replaying it would feed the model text the receipt says it
never saw.

Tests: `cargo nextest run -p codewhale-tui --lib -E 'test(steer) or
test(runtime_threads) or test(runtime_api) or test(restart_rebuild)'` —
492 run, 492 passed, 0 failed. The new drop test fails without the fix
(returns Ok with steer_count 1 instead of the 409). `cargo fmt --all --
--check` clean. Clippy on this crate reports the same 17 pre-existing
`too_many_arguments` errors with and without this change, none in these files.

Not from this change and still red on a clean tree:
`core::engine::tests::automatic_compaction_continues_one_task_and_suppresses_failed_passes`.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Signed-off-by: CodeWhale Bot <[email protected]>

CHANGELOG.md

@@ -122,6 +122,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- A steer the engine never delivered is no longer reported as sent. The runtime
+  API persisted the steer item as already-`Completed` and emitted
+  `turn.steered` + `item.completed` the moment the text entered the engine's
+  mailbox — before the engine decided anything. The engine discards a steer
+  whose turn has moved on, and an interrupted or failed turn drops whatever it
+  had queued, so a GUI could show "Guidance sent", clear the composer, and lose
+  the user's words. The engine now returns a verdict for every steer on every
+  exit path, the item settles `completed` or `canceled` to match, a dropped
+  steer emits `turn.steer_dropped` and answers `409` so a client can resend,
+  and `steer_count` counts steers the model actually received (#6276).
 - `<recommended_plugins>` suggestions stop nagging: a plugin id is now
   injected at most once per engine lifetime, and a plugin whose name a
   loaded skill already covers is never suggested — the local skill owns

crates/tui/src/core/engine.rs

@@ -1288,7 +1288,15 @@ impl Engine {
         }
     }
 
-    fn next_turn_steer(&mut self) -> Option<String> {
+    /// Take the next steer belonging to the active turn.
+    ///
+    /// Steers addressed to a turn that has already moved on are discarded
+    /// here; dropping their [`handle::SteerInput`] reports
+    /// [`handle::SteerOutcome::Dropped`] to the sender, so a discard is never
+    /// silent (#6276). The returned [`handle::PendingSteer`] is unsettled:
+    /// the caller must `commit()` it once the text is in the turn's record,
+    /// and dropping it otherwise reports `Dropped` too.
+    fn next_turn_steer(&mut self) -> Option<handle::PendingSteer> {
         let active_id = self
             .turn_controls
             .lock()
@@ -1298,7 +1306,7 @@ impl Engine {
             .map(|control| control.id);
         while let Ok(steer) = self.rx_steer.try_recv() {
             if steer.turn_id == active_id {
-                return Some(steer.content);
+                return Some(steer.into_pending());
             }
         }
         None
@@ -7985,7 +7993,7 @@ impl SubAgentWiring {
 mod approval;
 mod compaction;
 mod context;
-mod handle;
+pub(crate) mod handle;
 pub mod preview;
 use crate::compaction::estimate_input_tokens_conservative;
 #[cfg(test)]

crates/tui/src/core/engine/handle.rs

@@ -11,7 +11,7 @@
 use anyhow::Result;
 use std::collections::VecDeque;
 use std::sync::{Arc, Mutex as StdMutex};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, oneshot};
 use tokio_util::sync::CancellationToken;
 
 use codewhale_config::AppMode;
@@ -80,6 +80,77 @@ impl Drop for TurnControlGuard {
 pub(crate) struct SteerInput {
     pub(super) turn_id: Option<u64>,
     pub(crate) content: String,
+    pub(super) outcome: Option<oneshot::Sender<SteerOutcome>>,
+}
+
+/// The engine's verdict on one steer. A steer whose turn had already moved
+/// on is discarded by `next_turn_steer`; the verdict tells the sender which
+/// happened, because "the channel accepted the text" is not "the model saw
+/// it" (#6276).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum SteerOutcome {
+    /// The steer's text was committed into the session record inside the
+    /// turn it was sent to; the model received it.
+    Accepted,
+    /// The turn had already moved on (or ended) before the steer reached a
+    /// commit boundary. The model never saw the text.
+    Dropped,
+}
+
+/// A steer the engine has taken ownership of. Committing it reports
+/// [`SteerOutcome::Accepted`]; any other exit — interrupt, failure, early
+/// return, silent drop of the pending queue — reports `Dropped` from `Drop`,
+/// so no path can lose a verdict.
+pub(crate) struct PendingSteer {
+    pub(crate) content: String,
+    outcome: Option<oneshot::Sender<SteerOutcome>>,
+}
+
+impl PendingSteer {
+    pub(crate) fn new(content: String, outcome: Option<oneshot::Sender<SteerOutcome>>) -> Self {
+        Self { content, outcome }
+    }
+
+    /// Commit the steer into the turn's record: report `Accepted`, then hand
+    /// back the text. Consuming `self` without calling this reports
+    /// `Dropped` via `Drop`.
+    pub(crate) fn commit(mut self) -> String {
+        if let Some(outcome) = self.outcome.take() {
+            let _ = outcome.send(SteerOutcome::Accepted);
+        }
+        // `Drop` runs after this returns and finds `outcome` already taken,
+        // so the verdict stays exactly one `Accepted`.
+        std::mem::take(&mut self.content)
+    }
+}
+
+impl Drop for PendingSteer {
+    fn drop(&mut self) {
+        if let Some(outcome) = self.outcome.take() {
+            let _ = outcome.send(SteerOutcome::Dropped);
+        }
+    }
+}
+
+impl SteerInput {
+    /// Take ownership of this steer as an unsettled [`PendingSteer`].
+    ///
+    /// This is the only way to claim a steer off the channel. Whatever the
+    /// claimant then does — `commit()` or drop — settles it exactly once, so
+    /// there is one settlement mechanism rather than two (#6276).
+    pub(crate) fn into_pending(mut self) -> PendingSteer {
+        // Both fields are taken, so the `Drop` below finds nothing left to
+        // settle and the verdict travels with the `PendingSteer`.
+        PendingSteer::new(std::mem::take(&mut self.content), self.outcome.take())
+    }
+}
+
+impl Drop for SteerInput {
+    fn drop(&mut self) {
+        if let Some(outcome) = self.outcome.take() {
+            let _ = outcome.send(SteerOutcome::Dropped);
+        }
+    }
 }
 
 impl std::ops::Deref for SteerInput {
@@ -105,8 +176,24 @@ impl SteerPermit {
         self.permit.send(SteerInput {
             turn_id: self.turn_id,
             content,
+            outcome: None,
         });
     }
+
+    /// Send a steer and receive the engine's verdict on it. The receiver
+    /// resolves to [`SteerOutcome::Accepted`] when the turn commits the text
+    /// into its record, [`SteerOutcome::Dropped`] when the turn moved on
+    /// first, and closes without a verdict only if the engine itself is gone
+    /// (#6276).
+    pub(crate) fn send_with_outcome(self, content: String) -> oneshot::Receiver<SteerOutcome> {
+        let (outcome_tx, outcome_rx) = oneshot::channel();
+        self.permit.send(SteerInput {
+            turn_id: self.turn_id,
+            content,
+            outcome: Some(outcome_tx),
+        });
+        outcome_rx
+    }
 }
 
 impl EngineHandle {

crates/tui/src/core/engine/tests.rs

@@ -5217,6 +5217,7 @@ async fn user_steer_during_parent_answer_still_gets_a_reply_with_healthy_childre
             .try_send(handle::SteerInput {
                 turn_id,
                 content: "Also read state.txt and include its evidence.".to_string(),
+                outcome: None,
             })
             .expect("steer channel open");
         canned::simple_text_turn("The workflow is still running.")

crates/tui/src/core/engine/turn_loop.rs

@@ -45,7 +45,11 @@ struct StreamOutcome {
     pending_message_complete: bool,
     last_text_index: Option<usize>,
     stream_errors: u32,
-    pending_steers: Vec<String>,
+    /// Unsettled steers queued mid-stream. Each is committed into the turn's
+    /// record at a step boundary, or dropped — and dropping one reports
+    /// `SteerOutcome::Dropped` to its sender, so an interrupted or failed
+    /// turn cannot silently swallow user guidance (#6276).
+    pending_steers: Vec<handle::PendingSteer>,
     /// Typed, engine-internal drop-recovery state. `Option` + consume-once
     /// means one drop schedules exactly one resume; see [`StreamResume`].
     pending_resume: Option<StreamResume>,
@@ -806,11 +810,12 @@ impl Engine {
             }
 
             let mut accepted_steer = false;
-            while let Some(steer) = self.next_turn_steer() {
-                let steer = steer.trim().to_string();
-                if steer.is_empty() {
+            while let Some(pending) = self.next_turn_steer() {
+                if pending.content.trim().is_empty() {
+                    // Nothing to deliver; dropping `pending` settles it.
                     continue;
                 }
+                let steer = pending.commit().trim().to_string();
                 accepted_steer = true;
                 self.session
                     .working_set
@@ -2077,7 +2082,8 @@ impl Engine {
                         turn.stop_diagnostics
                             .permission_denial_rounds_without_progress = 0;
                     }
-                    for steer in pending_steers.drain(..) {
+                    for pending in pending_steers.drain(..) {
+                        let steer = pending.commit().trim().to_string();
                         self.session
                             .working_set
                             .observe_user_message(&steer, &self.session.workspace);
@@ -2681,7 +2687,8 @@ impl Engine {
 
             let accepted_steer_after_tools = !pending_steers.is_empty();
             if !pending_steers.is_empty() {
-                for steer in pending_steers.drain(..) {
+                for pending in pending_steers.drain(..) {
+                    let steer = pending.commit().trim().to_string();
                     self.session
                         .working_set
                         .observe_user_message(&steer, &self.session.workspace);
@@ -4672,7 +4679,7 @@ impl Engine {
         // content-block delta delivered to the consumer).
         let mut any_content_received = false;
         let mut transparent_stream_retries = 0u32;
-        let mut pending_steers: Vec<String> = Vec::new();
+        let mut pending_steers: Vec<handle::PendingSteer> = Vec::new();
         // `stream_start` is reset on a transparent retry so the wall-clock
         // budget restarts with the fresh stream.
         let mut stream_start = Instant::now();
@@ -4729,18 +4736,16 @@ impl Engine {
             let Some(event_result) = poll_outcome else {
                 break;
             };
-            while let Some(steer) = self.next_turn_steer() {
-                let steer = steer.trim().to_string();
-                if steer.is_empty() {
+            while let Some(pending) = self.next_turn_steer() {
+                if pending.content.trim().is_empty() {
+                    // Nothing to deliver; dropping `pending` settles it.
                     continue;
                 }
-                pending_steers.push(steer.clone());
+                let preview = summarize_text(pending.content.trim(), 120);
+                pending_steers.push(pending);
                 let _ = self
                     .tx_event
-                    .send(Event::status(format!(
-                        "Steer input queued: {}",
-                        summarize_text(&steer, 120)
-                    )))
+                    .send(Event::status(format!("Steer input queued: {preview}")))
                     .await;
             }
 

crates/tui/src/runtime_api.rs

@@ -8311,6 +8311,10 @@ fn map_thread_err(err: anyhow::Error) -> ApiError {
     } else if message.contains("already has an active turn")
         || message.contains("No active turn")
         || message.contains("is not active")
+        // A steer the engine dropped: the turn moved on before the model saw
+        // it. 409 lets a client keep the text and resend rather than trust a
+        // delivery that never happened (#6276).
+        || message.contains("moved on before the steer")
         || lower.contains("operation_key is already bound")
         || lower.contains("operation_key binding is incomplete")
         || lower.contains("operation_key binding does not match")

crates/tui/src/runtime_api/tests.rs

@@ -4415,7 +4415,10 @@ async fn steer_and_interrupt_endpoints_work_on_active_turn() -> Result<()> {
         let _ = tx_event
             .send(EngineEvent::MessageStarted { index: 0 })
             .await;
-        if let Some(steer_text) = rx_steer.recv().await {
+        if let Some(steer) = rx_steer.recv().await {
+            // An engine that uses the steer has committed it; `commit()` is
+            // what reports that acceptance back to `steer_turn`.
+            let steer_text = steer.into_pending().commit();
             let _ = tx_event
                 .send(EngineEvent::MessageDelta {
                     index: 0,

crates/tui/src/runtime_threads.rs

@@ -31,6 +31,7 @@ use crate::compaction::CompactionConfig;
 #[cfg(test)]
 use crate::config::DEFAULT_TEXT_MODEL;
 use crate::config::{ApiProvider, Config, MAX_SUBAGENTS, ProviderIdentity};
+use crate::core::engine::handle::SteerOutcome;
 use crate::core::engine::{
     EngineConfig, EngineHandle, spawn_engine_with_authoritative_route_config,
 };
@@ -79,6 +80,14 @@ const MAX_PENDING_DYNAMIC_TOOL_CALLS: usize = 128;
 const SUMMARY_LIMIT: usize = 280;
 const STREAM_DELTA_BATCH_MAX_LATENCY: Duration = Duration::from_millis(32);
 const STREAM_DELTA_BATCH_MAX_BYTES: usize = 16 * 1024;
+/// How long `steer_turn` observes the engine's verdict before answering with
+/// the honest `Queued` receipt instead. A steer sent into a streaming turn
+/// settles in milliseconds; one sent behind a long tool call cannot, and an
+/// API request must not hang for the length of a tool call (#6276).
+const STEER_SETTLE_WAIT: Duration = Duration::from_secs(2);
+/// Why a steer never reached the model, in the words a client can show.
+const STEER_DROPPED_REASON: &str =
+    "the turn moved on before the engine committed it, so the model never saw it — resend it";
 const EVENT_TRANSACTION_LOCK_TIMEOUT: Duration = Duration::from_secs(5);
 const EVENT_TRANSACTION_LOCK_POLL: Duration = Duration::from_millis(5);
 const EVENT_TRANSACTION_LOCK_FILE: &str = "events.lock";
@@ -8887,58 +8896,133 @@ impl RuntimeThreadManager {
         acceptance_rx
     }
 
-    fn spawn_steer_receipts(
+    /// Settle one steer against the engine's real verdict.
+    ///
+    /// The detached worker — not the API caller — owns `outcome_rx`, so a
+    /// cancelled request or a timed-out caller can never orphan the verdict
+    /// and strand the item in `Queued` (#6276). Exactly one of two
+    /// settlements is written, record first and events after:
+    ///
+    /// - `Accepted`: the item flips to `Completed`, `steer_count` rises, and
+    ///   `turn.steered` + `item.completed` are emitted — the happy path
+    ///   clients already know.
+    /// - `Dropped` (including a closed channel, which means the engine is
+    ///   gone): the item flips to `Canceled` and `turn.steer_dropped` carries
+    ///   the settled item, so a client can requeue the text it was told had
+    ///   been delivered.
+    fn spawn_steer_settlement(
         &self,
         turn: TurnRecord,
         item: TurnItemRecord,
         prompt: String,
-    ) -> oneshot::Receiver<TurnRecord> {
-        let (receipt_tx, receipt_rx) = oneshot::channel();
+        outcome_rx: oneshot::Receiver<SteerOutcome>,
+    ) -> oneshot::Receiver<(SteerOutcome, TurnRecord)> {
+        let (settle_tx, settle_rx) = oneshot::channel();
         let manager = Arc::new(self.clone());
         let worker = tokio::spawn(async move {
             use futures_util::FutureExt;
-            let receipts = std::panic::AssertUnwindSafe(async {
-                if let Err(err) = manager
-                    .emit_event(
-                        &turn.thread_id,
-                        Some(&turn.id),
-                        Some(&item.id),
-                        "turn.steered",
-                        json!({
-                            "thread_id": turn.thread_id.clone(),
-                            "turn_id": turn.id.clone(),
-                            "input": prompt,
-                        }),
-                    )
-                    .await
-                {
-                    tracing::warn!("Failed to persist turn.steered after engine acceptance: {err}");
-                }
-                if let Err(err) = manager
-                    .emit_event(
-                        &turn.thread_id,
-                        Some(&turn.id),
-                        Some(&item.id),
-                        "item.completed",
-                        json!({ "item": item }),
-                    )
-                    .await
-                {
-                    tracing::warn!("Failed to persist steer item.completed: {err}");
+            // The engine sends exactly one verdict on every exit path. A
+            // closed channel means the engine itself went away without one,
+            // which is a drop by any honest reading.
+            let outcome = outcome_rx.await.unwrap_or(SteerOutcome::Dropped);
+            let mut item = item;
+            item.status = match outcome {
+                SteerOutcome::Accepted => TurnItemLifecycleStatus::Completed,
+                SteerOutcome::Dropped => TurnItemLifecycleStatus::Canceled,
+            };
+            item.ended_at = Some(Utc::now());
+            let settled_turn = std::panic::AssertUnwindSafe(async {
+                let persisted = {
+                    let _turn_mutation = manager.store.turn_mutation.lock();
+                    (|| -> Result<TurnRecord> {
+                        manager.store.save_item(&item)?;
+                        let mut turn = manager.store.load_turn(&item.turn_id)?;
+                        if outcome == SteerOutcome::Accepted {
+                            turn.steer_count = turn.steer_count.saturating_add(1);
+                        }
+                        manager.store.save_turn(&turn)?;
+                        Ok(turn)
+                    })()
+                };
+                let settled_turn = match persisted {
+                    Ok(turn) => turn,
+                    Err(err) => {
+                        tracing::error!("Failed to settle steer item {}: {err}", item.id);
+                        turn.clone()
+                    }
+                };
+                match outcome {
+                    SteerOutcome::Accepted => {
+                        if let Err(err) = manager
+                            .emit_event(
+                                &turn.thread_id,
+                                Some(&turn.id),
+                                Some(&item.id),
+                                "turn.steered",
+                                json!({
+                                    "thread_id": turn.thread_id.clone(),
+                                    "turn_id": turn.id.clone(),
+                                    "input": prompt,
+                                }),
+                            )
+                            .await
+                        {
+                            tracing::warn!(
+                                "Failed to persist turn.steered after engine acceptance: {err}"
+                            );
+                        }
+                        if let Err(err) = manager
+                            .emit_event(
+                                &turn.thread_id,
+                                Some(&turn.id),
+                                Some(&item.id),
+                                "item.completed",
+                                json!({ "item": item }),
+                            )
+                            .await
+                        {
+                            tracing::warn!("Failed to persist steer item.completed: {err}");
+                        }
+                    }
+                    SteerOutcome::Dropped => {
+                        if let Err(err) = manager
+                            .emit_event(
+                                &turn.thread_id,
+                                Some(&turn.id),
+                                Some(&item.id),
+                                "turn.steer_dropped",
+                                json!({
+                                    "thread_id": turn.thread_id.clone(),
+                                    "turn_id": turn.id.clone(),
+                                    "input": prompt,
+                                    "reason": STEER_DROPPED_REASON,
+                                    "item": item,
+                                }),
+                            )
+                            .await
+                        {
+                            tracing::warn!("Failed to persist turn.steer_dropped: {err}");
+                        }
+                    }
                 }
+                settled_turn
             })
             .catch_unwind()
             .await;
-            if let Err(payload) = receipts {
-                tracing::error!(
-                    "Steer receipt task panicked after engine acceptance: {}",
-                    panic_payload_message(&*payload)
-                );
-            }
-            let _ = receipt_tx.send(turn);
+            let settled_turn = match settled_turn {
+                Ok(turn) => turn,
+                Err(payload) => {
+                    tracing::error!(
+                        "Steer settlement task panicked: {}",
+                        panic_payload_message(&*payload)
+                    );
+                    turn
+                }
+            };
+            let _ = settle_tx.send((outcome, settled_turn));
         });
         self.track_receipt_worker(worker);
-        receipt_rx
+        settle_rx
     }
 
     pub async fn start_turn(&self, thread_id: &str, req: StartTurnRequest) -> Result<TurnRecord> {
@@ -9612,20 +9696,24 @@ impl RuntimeThreadManager {
             .map_err(|error| anyhow!("Failed to steer turn: {error}"))?;
 
         let now = Utc::now();
+        let queued_turn;
         let item = TurnItemRecord {
             schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
             id: format!("item_{}", &Uuid::new_v4().to_string()[..8]),
             turn_id: turn_id.to_string(),
             kind: TurnItemKind::UserMessage,
-            status: TurnItemLifecycleStatus::Completed,
+            // Queued, not Completed: the text is in the engine's mailbox, not
+            // yet in the turn's record. `spawn_steer_settlement` flips it once
+            // the engine reports what actually happened (#6276).
+            status: TurnItemLifecycleStatus::Queued,
             summary: summarize_text(&prompt, SUMMARY_LIMIT),
             detail: Some(prompt.clone()),
             metadata: None,
             artifact_refs: Vec::new(),
             started_at: Some(now),
-            ended_at: Some(now),
+            ended_at: None,
         };
-        let receipt_rx = {
+        let settle_rx = {
             let mut active = self.active.lock().await;
             let Some(active_thread) = active.engines.get(thread_id) else {
                 bail!("Thread is not loaded");
@@ -9649,7 +9737,8 @@ impl RuntimeThreadManager {
                     bail!("Turn {turn_id} is no longer in progress and cannot be steered");
                 }
                 self.store.save_item(&item)?;
-                turn.steer_count = turn.steer_count.saturating_add(1);
+                // `steer_count` counts steers the model actually received, so
+                // it rises in the settlement path, not here.
                 if !turn.item_ids.iter().any(|id| id == &item.id) {
                     turn.item_ids.push(item.id.clone());
                 }
@@ -9670,13 +9759,26 @@ impl RuntimeThreadManager {
             };
             // The reserved send has no await/failure point. From here the
             // engine and durable record agree even if the API caller drops.
-            permit.send(prompt.clone());
+            let outcome_rx = permit.send_with_outcome(prompt.clone());
             touch_lru(&mut active.lru, thread_id);
-            self.spawn_steer_receipts(turn, item, prompt)
+            queued_turn = turn.clone();
+            self.spawn_steer_settlement(turn, item, prompt, outcome_rx)
         };
-        receipt_rx
-            .await
-            .map_err(|_| anyhow!("Steer receipt task ended before acknowledgement"))
+
+        // The settler owns the verdict; this is only an observation of it.
+        // Steering a streaming turn settles in milliseconds, so the caller
+        // almost always gets the truth in-band. Behind a long tool call the
+        // engine will not look at its mailbox for minutes, and an API request
+        // must not hang that long — so the wait is bounded and the caller
+        // falls back to the honest `Queued` receipt it already has.
+        match tokio::time::timeout(STEER_SETTLE_WAIT, settle_rx).await {
+            Ok(Ok((SteerOutcome::Accepted, turn))) => Ok(turn),
+            Ok(Ok((SteerOutcome::Dropped, _))) => bail!(
+                "Turn {turn_id} moved on before the steer reached the model; {STEER_DROPPED_REASON}"
+            ),
+            Ok(Err(_)) => bail!("Steer settlement task ended before acknowledgement"),
+            Err(_) => Ok(queued_turn),
+        }
     }
 
     pub async fn compact_thread(
@@ -10508,6 +10610,17 @@ impl RuntimeThreadManager {
             for item in items {
                 match item.kind {
                     TurnItemKind::UserMessage => {
+                        // A steer the engine never committed is recorded
+                        // `queued` or `canceled` precisely because the model
+                        // never saw it. Replaying it here would put words into
+                        // the context that the record says were not delivered
+                        // — the record is right, so it stays out (#6276).
+                        if matches!(
+                            item.status,
+                            TurnItemLifecycleStatus::Queued | TurnItemLifecycleStatus::Canceled
+                        ) {
+                            continue;
+                        }
                         flush_assistant(&mut assistant_blocks, &mut messages);
                         user_blocks.extend(item.user_content()?);
                     }

crates/tui/src/runtime_threads/tests.rs

@@ -13539,7 +13539,9 @@ async fn steer_turn_on_active_turn_records_item_and_event() -> Result<()> {
                 })
                 .await;
             if let Some(steer) = rx_steer.recv().await {
-                let _ = steer_seen_tx.send(steer.content);
+                // Model an engine that commits the steer into its record;
+                // `commit()` is what reports acceptance back to `steer_turn`.
+                let _ = steer_seen_tx.send(steer.into_pending().commit());
             }
             let _ = tx_event
                 .send(EngineEvent::MessageStarted { index: 0 })
@@ -13670,7 +13672,7 @@ async fn steer_receipts_outlive_caller_cancellation_after_engine_acceptance() ->
     assert_eq!(
         tokio::time::timeout(Duration::from_secs(2), rx_steer.recv())
             .await?
-            .map(|steer| steer.content),
+            .map(|steer| steer.into_pending().commit()),
         Some("keep the accepted steer".to_string())
     );
     steer_task.abort();
@@ -13741,6 +13743,121 @@ async fn steer_receipts_outlive_caller_cancellation_after_engine_acceptance() ->
     Ok(())
 }
 
+#[tokio::test]
+async fn steer_dropped_by_the_engine_is_reported_as_undelivered() -> Result<()> {
+    // #6276: the channel accepting the text is not the model seeing it. An
+    // engine that takes a steer and never commits it — the turn moved on, was
+    // interrupted, or failed — must not produce a delivery receipt.
+    let manager = test_manager(test_runtime_dir())?;
+    let thread = manager
+        .create_thread(CreateThreadRequest::default())
+        .await?;
+    let harness = install_mock_engine(&manager, &thread.id).await;
+    let mut rx_op = harness.rx_op;
+    let mut rx_steer = harness.rx_steer;
+    let tx_event = harness.tx_event;
+
+    let turn = manager
+        .start_turn(
+            &thread.id,
+            StartTurnRequest {
+                prompt: "initial".to_string(),
+                ..Default::default()
+            },
+        )
+        .await?;
+    assert!(matches!(
+        rx_op.recv().await,
+        Some(Op::SendMessage(TurnSpec { .. }))
+    ));
+
+    // The engine receives the steer and drops it without committing, exactly
+    // as `next_turn_steer` does for a turn that has already moved on.
+    let drop_driver = tokio::spawn(async move {
+        let steer = rx_steer.recv().await;
+        drop(steer);
+        rx_steer
+    });
+
+    let error = manager
+        .steer_turn(
+            &thread.id,
+            &turn.id,
+            SteerTurnRequest {
+                prompt: "never reaches the model".to_string(),
+            },
+        )
+        .await
+        .expect_err("a dropped steer must not report delivery");
+    assert!(
+        error.to_string().contains("moved on before the steer"),
+        "drop error must name the cause, got: {error}"
+    );
+    let _rx_steer = drop_driver.await?;
+
+    // The durable record agrees: canceled item, no delivery count.
+    let items = manager.store.list_items_for_turn(&turn.id)?;
+    let steer_item = items
+        .iter()
+        .find(|item| item.detail.as_deref() == Some("never reaches the model"))
+        .context("the attempted steer must stay on the record")?;
+    assert_eq!(steer_item.status, TurnItemLifecycleStatus::Canceled);
+    assert_eq!(manager.store.load_turn(&turn.id)?.steer_count, 0);
+
+    let events = manager.events_since(&thread.id, None)?;
+    let dropped = events
+        .iter()
+        .find(|ev| ev.event == "turn.steer_dropped")
+        .context("a dropped steer must be announced so a client can requeue")?;
+    assert_eq!(
+        dropped.payload.get("input").and_then(Value::as_str),
+        Some("never reaches the model")
+    );
+    assert!(
+        !events.iter().any(|ev| ev.event == "turn.steered"),
+        "a dropped steer must not emit turn.steered"
+    );
+    assert!(
+        !events.iter().any(|ev| {
+            ev.event == "item.completed"
+                && ev
+                    .payload
+                    .get("item")
+                    .and_then(|item| item.get("detail"))
+                    .and_then(Value::as_str)
+                    == Some("never reaches the model")
+        }),
+        "a dropped steer must not emit item.completed"
+    );
+
+    tx_event
+        .send(EngineEvent::MessageStarted { index: 0 })
+        .await?;
+    tx_event
+        .send(EngineEvent::MessageDelta {
+            index: 0,
+            content: "unsteered response".to_string(),
+        })
+        .await?;
+    tx_event
+        .send(EngineEvent::MessageComplete { index: 0 })
+        .await?;
+    tx_event
+        .send(EngineEvent::TurnComplete {
+            usage: Usage::default(),
+            parent_route_usage: Usage::default(),
+            routed_usage_dropped_records: 0,
+            status: TurnOutcomeStatus::Completed,
+            error: None,
+            tool_catalog: None,
+            base_url: None,
+        })
+        .await?;
+    let terminal = wait_for_terminal_turn(&manager, &turn.id).await?;
+    assert_eq!(terminal.status, RuntimeTurnStatus::Completed);
+    Ok(())
+}
+
 #[tokio::test]
 async fn steer_rejects_a_terminal_durable_turn_without_dispatch_or_item() -> Result<()> {
     let manager = test_manager(test_runtime_dir())?;
@@ -15456,6 +15573,101 @@ fn restart_rebuild_keeps_in_flight_tool_call_identity() -> Result<()> {
     Ok(())
 }
 
+/// A steer the engine never committed is recorded `canceled`/`queued` exactly
+/// because the model never saw it. Rebuilding history must honour that: #6276
+/// made the receipt honest, and replaying the text here would put it back into
+/// the context the receipt says it never reached.
+#[test]
+fn restart_rebuild_skips_steers_the_engine_never_delivered() -> Result<()> {
+    let dir = test_runtime_dir();
+    let manager = test_manager(dir.clone())?;
+    let thread = sample_thread("thr_rebuild_6276");
+    manager.store.save_thread(&thread)?;
+
+    let now = Utc::now();
+    let turn_id = "turn_6276_rebuild".to_string();
+    let item = |id: &str, text: &str, status: TurnItemLifecycleStatus| TurnItemRecord {
+        schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+        id: id.to_string(),
+        turn_id: turn_id.clone(),
+        kind: TurnItemKind::UserMessage,
+        status,
+        summary: text.to_string(),
+        detail: Some(text.to_string()),
+        metadata: None,
+        artifact_refs: Vec::new(),
+        started_at: Some(now),
+        ended_at: Some(now),
+    };
+    let delivered = item("item_6276_ok", "hello", TurnItemLifecycleStatus::Completed);
+    let dropped = item(
+        "item_6276_drop",
+        "never seen by the model",
+        TurnItemLifecycleStatus::Canceled,
+    );
+    let pending = item(
+        "item_6276_queue",
+        "not settled yet",
+        TurnItemLifecycleStatus::Queued,
+    );
+    manager.store.save_item(&delivered)?;
+    manager.store.save_item(&dropped)?;
+    manager.store.save_item(&pending)?;
+    manager.store.save_turn(&TurnRecord {
+        max_output_tokens: None,
+        schema_version: CURRENT_RUNTIME_SCHEMA_VERSION,
+        id: turn_id.clone(),
+        thread_id: thread.id.clone(),
+        status: RuntimeTurnStatus::Completed,
+        input_summary: "hello".to_string(),
+        created_at: now,
+        started_at: Some(now),
+        ended_at: Some(now),
+        duration_ms: None,
+        usage: None,
+        routing_settlement: false,
+        effective_route_usage: None,
+        permission_posture: None,
+        effective_provider: None,
+        effective_provider_id: None,
+        effective_openrouter_vendor: None,
+        effective_billing_surface: None,
+        effective_endpoint_fingerprint: None,
+        effective_provider_live_pricing: None,
+        effective_billing_mode: None,
+        effective_dispatched_at: None,
+        effective_model: None,
+        routed_usage: Vec::new(),
+        routed_usage_drop_records: Vec::new(),
+        routed_usage_source_ids: Vec::new(),
+        routed_usage_dropped_records: 0,
+        model_request_diagnostics: None,
+        error: None,
+        item_ids: vec![delivered.id.clone(), dropped.id.clone(), pending.id.clone()],
+        steer_count: 0,
+        agent_mail_message_id: None,
+    })?;
+
+    let turns = manager.store.list_turns_for_thread(&thread.id)?;
+    let messages = manager.reconstruct_messages_from_turns(&turns)?;
+    let replayed: Vec<String> = messages
+        .iter()
+        .flat_map(|message| message.content.iter())
+        .filter_map(|block| match block {
+            ContentBlock::Text { text, .. } => Some(text.clone()),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(
+        replayed,
+        vec!["hello".to_string()],
+        "only the delivered user message may be rebuilt"
+    );
+
+    let _ = std::fs::remove_dir_all(dir);
+    Ok(())
+}
+
 /// Snapshots persisted before tool identity was durable (metadata absent)
 /// must contribute nothing to rebuilt history: an empty id/name tool_call
 /// with `arguments: "null"` is rejected by strict OpenAI-compatible

docs/RUNTIME_API.md

@@ -762,7 +762,7 @@ accept an empty string to clear a previously-set value. Added in v0.8.10 (#562):
 
 **Turns** (within a thread)
 - `POST /v1/threads/{id}/turns`
-- `POST /v1/threads/{id}/turns/{turn_id}/steer`
+- `POST /v1/threads/{id}/turns/{turn_id}/steer` - inject guidance into the running turn. The response is a receipt for what actually happened, not for what was attempted; see [Steer delivery](#steer-delivery).
 - `POST /v1/threads/{id}/turns/{turn_id}/interrupt`
 - `POST /v1/threads/{id}/compact` (manual compaction)
 - `POST /v1/threads/{id}/undo` - fork the thread with the last N turns removed (`{"depth": N}`, default 0 = last turn only); returns the forked thread plus `original_user_text` so a GUI can pre-populate the input box
@@ -1697,8 +1697,32 @@ Compatibility notes:
   is an equivalent alias for clients that use `created_at` naming elsewhere; do
   not require both fields to be present.
 
+### Steer delivery
+
+Putting a steer into the engine's mailbox is not the same as the model reading
+it. The engine discards a steer whose turn has already moved on, and an
+interrupted or failed turn drops whatever it had queued. The API reports the
+engine's real verdict rather than the attempt:
+
+- The item is persisted `queued` when the steer is accepted into the mailbox.
+- **Delivered.** The engine committed the text into the turn's record: the item
+  becomes `completed`, `steer_count` rises, and `turn.steered` + `item.completed`
+  are emitted. `POST .../steer` returns `200` with that turn.
+- **Not delivered.** The turn moved on, was interrupted, or failed first: the
+  item becomes `canceled`, `steer_count` does not rise, and `turn.steer_dropped`
+  is emitted carrying `input`, `reason`, and the settled `item`. `POST .../steer`
+  returns `409`, so a client can keep the user's text and resend it rather than
+  clearing a composer over guidance that was never seen.
+- **Still pending.** A steer sent while the engine is inside a long tool call
+  cannot settle until that call returns, and the request does not hang for it.
+  After a short wait `POST .../steer` returns `200` with the item still `queued`;
+  the eventual `turn.steered` or `turn.steer_dropped` event carries the verdict.
+
+A client that treats `200` as "the model saw it" is therefore wrong in the third
+case: read the item's status, or wait for the event.
+
 Common event names: `thread.started`, `thread.forked`, `turn.started`,
-`turn.lifecycle`, `turn.steered`, `turn.interrupt_requested`,
+`turn.lifecycle`, `turn.steered`, `turn.steer_dropped`, `turn.interrupt_requested`,
 `turn.completed`, `item.started`, `item.delta`, `item.completed`,
 `item.failed`, `item.interrupted`, `approval.required`, `approval.decided`,
 `approval.timeout`, `user_input.required`, `user_input.answered`,