diff --git a/crates/tui/src/client.rs b/crates/tui/src/client.rs
index cf53fdb9de..2fda2e43db 100644
--- a/crates/tui/src/client.rs
+++ b/crates/tui/src/client.rs
@@ -279,6 +279,19 @@ pub struct DeepSeekClient {
     /// this list closes the gap for bare provider tokens with no recognizable
     /// prefix (for example token-plan and provider-specific keys).
     model_bound_secret_values: Arc<Vec<String>>,
+    /// Exact values a catalog endpoint could echo back at us: the active API
+    /// key and every user-configured HTTP header value, plus everything in
+    /// `model_bound_secret_values`.
+    ///
+    /// Deliberately a second, wider list rather than a widening of that one:
+    /// they answer different questions at different trust boundaries. That
+    /// list is "what must never reach a *model*", which is why it covers only
+    /// auth-shaped headers and values of at least
+    /// `MIN_EXACT_SECRET_CHARS`. This one is "what must never come back out
+    /// of an endpoint the user typed during setup" (#6173), where a
+    /// three-character key is still a key and a custom header the user
+    /// configured is still theirs.
+    catalog_error_secret_values: Arc<Vec<String>>,
     /// Whether credential-shaped tool output is masked before it is sent to an
     /// upstream model. The safe default is `true`; it is `false` only after the
     /// user disabled `[redaction] model_bound` and confirmed the opt-out on the
@@ -577,6 +590,7 @@ impl Clone for DeepSeekClient {
             http1_client: self.http1_client.clone(),
             api_key: self.api_key.clone(),
             model_bound_secret_values: Arc::clone(&self.model_bound_secret_values),
+            catalog_error_secret_values: Arc::clone(&self.catalog_error_secret_values),
             model_bound_masking: self.model_bound_masking,
             base_url: self.base_url.clone(),
             api_provider: self.api_provider,
@@ -754,6 +768,58 @@ fn configured_model_bound_secret_values(config: &Config, active_api_key: &str) -
     values
 }
 
+/// Everything a catalog probe could have sent that must not come back.
+///
+/// Longest first, so a value that contains another is masked whole.
+fn catalog_error_secret_values(
+    active_api_key: &str,
+    http_headers: &HashMap<String, String>,
+    model_bound: &[String],
+) -> Vec<String> {
+    let mut values: Vec<String> = Vec::new();
+    let mut push = |value: &str| {
+        let value = value.trim();
+        if !value.is_empty() && !values.iter().any(|existing| existing == value) {
+            values.push(value.to_string());
+        }
+    };
+    // No length floor here, unlike `push_model_bound_secret`: a short key
+    // echoed back by an untrusted endpoint is still a leaked key. The cost of
+    // being wrong is a suppressed message, which is what this path did before.
+    push(active_api_key);
+    for value in http_headers.values() {
+        push(value);
+    }
+    for value in model_bound {
+        push(value);
+    }
+    values.sort_by_key(|value| std::cmp::Reverse(value.len()));
+    values
+}
+
+/// The opaque values in a request's URL query — the pagination cursor.
+///
+/// Collected in both forms: decoded, as a provider that parsed the cursor
+/// would echo it, and raw, as one that quoted the URL back would.
+fn request_query_secret_values(url: &reqwest::Url) -> Vec<String> {
+    let mut values: Vec<String> = Vec::new();
+    let mut push = |value: String| {
+        if !value.trim().is_empty() && !values.contains(&value) {
+            values.push(value);
+        }
+    };
+    for (_, value) in url.query_pairs() {
+        push(value.into_owned());
+    }
+    for pair in url.query().unwrap_or_default().split('&') {
+        if let Some((_, value)) = pair.split_once('=') {
+            push(value.to_string());
+        }
+    }
+    values.sort_by_key(|value| std::cmp::Reverse(value.len()));
+    values
+}
+
 fn redact_model_bound_text(text: &str, exact_secret_values: &[String]) -> String {
     let mut redacted = text.to_string();
     for secret in exact_secret_values {
@@ -770,6 +836,24 @@ fn redact_model_bound_text(text: &str, exact_secret_values: &[String]) -> String
 /// Maximum bytes to read from an error response body (64 KB).
 pub(super) const ERROR_BODY_MAX_BYTES: usize = 64 * 1024;
 
+/// How much of a provider's HTTP error body may be shown to the user.
+///
+/// The catalog probe is the one request that contacts a `base_url` the user
+/// typed during setup, and its URL carries an opaque pagination cursor, so a
+/// provider — or anything answering at that URL — can echo a key, a custom
+/// header value or the cursor back inside an error body. #3385 answered that
+/// by discarding the body entirely, and the cost of the blunt version was
+/// #6173: Gemini's real reason ("User location is not supported for the API
+/// use") reached the user as `Invalid request (400): ` with nothing after the
+/// colon, so a geo-block, a bad key and a wrong endpoint were indistinguishable.
+pub(super) enum ErrorBodyDisclosure {
+    /// The endpoint is already established. Surface the provider's message.
+    Full,
+    /// Untrusted endpoint. Surface only what survives redaction — and nothing
+    /// at all if a known secret is still in the result.
+    Guarded { request_secrets: Vec<String> },
+}
+
 /// Read/overall timeout for the shared client's non-streaming requests
 /// (`/models` listing, catalog refresh, health probes). Streaming requests
 /// keep their own idle-timeout envelope; without this, a provider that accepts
@@ -1568,12 +1652,18 @@ impl DeepSeekClient {
         )?
         .build()?;
 
+        let catalog_error_secret_values = Arc::new(catalog_error_secret_values(
+            &api_key,
+            &http_headers,
+            &model_bound_secret_values,
+        ));
         Ok(Self {
             http_client,
             models_http_client,
             http1_client,
             api_key,
             model_bound_secret_values,
+            catalog_error_secret_values,
             model_bound_masking,
             base_url,
             api_provider,
@@ -2910,10 +3000,18 @@ impl DeepSeekClient {
                         .timeout(NON_STREAMING_HTTP_TIMEOUT)
                 };
                 let response = match mode {
-                    ModelsRequestMode::Interactive => self
-                        .send_with_retry_error_body(build, false)
-                        .await
-                        .map_err(ModelsFetchError::Interactive)?,
+                    // #6173: the provider's own words, minus this client's
+                    // secrets and this request's cursor. The endpoint is not
+                    // established here — it is whatever the user typed during
+                    // setup — so the body is guarded rather than trusted.
+                    ModelsRequestMode::Interactive => {
+                        let disclosure = ErrorBodyDisclosure::Guarded {
+                            request_secrets: request_query_secret_values(&url),
+                        };
+                        self.send_with_retry_error_body(build, &disclosure)
+                            .await
+                            .map_err(ModelsFetchError::Interactive)?
+                    }
                     ModelsRequestMode::Refresh => build()
                         .send()
                         .await
@@ -3399,28 +3497,67 @@ impl DeepSeekClient {
         }
     }
 
+    /// Apply `disclosure` to one provider error body.
+    ///
+    /// Redaction runs on the raw bytes, *before* `sanitize_http_error_body`
+    /// truncates them, so a secret can never be split across the truncation
+    /// boundary and survive as a fragment. The result is then checked against
+    /// every value this client knows is secret; if one is still there the
+    /// whole body is dropped, which is exactly the behaviour this path had
+    /// before #6173. The fallback is the old contract, not a weaker one.
+    ///
+    /// Known limitation: this removes what the *client* knows is secret. A
+    /// credential the user configured outside Codewhale — in a proxy, say —
+    /// is not in that set and would pass through, the same limit
+    /// `redact_model_bound_text` has.
+    fn disclosed_http_error_body(
+        &self,
+        disclosure: &ErrorBodyDisclosure,
+        status: u16,
+        raw: &str,
+    ) -> String {
+        let provider = Some(self.api_provider.display_name());
+        let ErrorBodyDisclosure::Guarded { request_secrets } = disclosure else {
+            return sanitize_http_error_body(provider, status, raw);
+        };
+        let mut redacted = raw.to_string();
+        for secret in self
+            .catalog_error_secret_values
+            .iter()
+            .chain(request_secrets.iter())
+        {
+            redacted = redacted.replace(secret.as_str(), codewhale_config::persistence::REDACTED);
+        }
+        let message = sanitize_http_error_body(provider, status, &redacted);
+        let leaked = self
+            .catalog_error_secret_values
+            .iter()
+            .chain(request_secrets.iter())
+            .any(|secret| message.contains(secret.as_str()));
+        if leaked { String::new() } else { message }
+    }
+
     pub(super) async fn send_with_retry<F>(&self, build: F) -> Result<reqwest::Response>
     where
         F: FnMut() -> reqwest::RequestBuilder,
     {
-        self.send_with_retry_error_body(build, true).await
+        self.send_with_retry_error_body(build, &ErrorBodyDisclosure::Full)
+            .await
     }
 
-    // Model-list errors can echo opaque cursors or credentials. Keep status and
-    // Retry-After classification, but suppress their bodies before retry logs
-    // and state updates. Other requests retain their existing error details.
+    /// Model-list errors can echo opaque cursors or credentials. Keep status
+    /// and Retry-After classification either way; `disclosure` decides how
+    /// much of the body reaches retry logs, state updates and the user.
     async fn send_with_retry_error_body<F>(
         &self,
         mut build: F,
-        include_error_body: bool,
+        disclosure: &ErrorBodyDisclosure,
     ) -> Result<reqwest::Response>
     where
         F: FnMut() -> reqwest::RequestBuilder,
     {
         if self.isolated_request_state {
-            return self
-                .send_with_isolated_retry(build, include_error_body)
-                .await;
+            return self.send_with_isolated_retry(build, disclosure).await;
         }
         let retry_cfg: LlmRetryConfig = self.retry.clone().into();
         let request_result = with_retry(
@@ -3446,16 +3583,8 @@ impl DeepSeekClient {
                         return Ok(response);
                     }
                     let retry_after = extract_retry_after(response.headers());
-                    let body = if include_error_body {
-                        let body = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
-                        sanitize_http_error_body(
-                            Some(self.api_provider.display_name()),
-                            status.as_u16(),
-                            &body,
-                        )
-                    } else {
-                        String::new()
-                    };
+                    let raw = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
+                    let body = self.disclosed_http_error_body(disclosure, status.as_u16(), &raw);
                     Err(LlmError::from_http_response_with_retry_after(
                         status.as_u16(),
                         &body,
@@ -3514,7 +3643,7 @@ impl DeepSeekClient {
     async fn send_with_isolated_retry<F>(
         &self,
         mut build: F,
-        include_error_body: bool,
+        disclosure: &ErrorBodyDisclosure,
     ) -> Result<reqwest::Response>
     where
         F: FnMut() -> reqwest::RequestBuilder,
@@ -3535,16 +3664,8 @@ impl DeepSeekClient {
                         return Ok(response);
                     }
                     let retry_after = extract_retry_after(response.headers());
-                    let body = if include_error_body {
-                        let body = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
-                        sanitize_http_error_body(
-                            Some(self.api_provider.display_name()),
-                            status.as_u16(),
-                            &body,
-                        )
-                    } else {
-                        String::new()
-                    };
+                    let raw = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
+                    let body = self.disclosed_http_error_body(disclosure, status.as_u16(), &raw);
                     Err(LlmError::from_http_response_with_retry_after(
                         status.as_u16(),
                         &body,
diff --git a/crates/tui/src/client/catalog_tests.rs b/crates/tui/src/client/catalog_tests.rs
index 3e86650e0b..11346963d8 100644
--- a/crates/tui/src/client/catalog_tests.rs
+++ b/crates/tui/src/client/catalog_tests.rs
@@ -859,6 +859,52 @@ fn assert_no_canaries(error: &anyhow::Error) {
     }
 }
 
+/// #6173: a geo-blocked key produced `Invalid request (400): ` — the colon
+/// that introduces the provider's reason, with nothing after it, because the
+/// catalog path discarded the body wholesale. A geo-block, a bad key and a
+/// wrong endpoint were then indistinguishable, and the reporter had to change
+/// VPN exits to find out which one it was. The reason is the provider's own
+/// words; only this client's secrets have to go.
+#[tokio::test]
+async fn catalog_errors_surface_the_provider_reason_without_client_secrets() {
+    const REASON: &str = "User location is not supported for the API use.";
+
+    let server = MockServer::start().await;
+    mount_page(
+        &server,
+        None,
+        ResponseTemplate::new(400).set_body_json(json!({
+            "error": {"code": 400, "message": REASON, "status": "FAILED_PRECONDITION"}
+        })),
+    )
+    .await;
+    let client = anthropic_client(&server.uri());
+    let error = client.list_models().await.unwrap_err();
+    assert!(
+        format!("{error:#}").contains(REASON),
+        "the provider's reason must reach the user: {error:#}"
+    );
+    assert_no_canaries(&error);
+
+    // The same reason, from an endpoint that also echoes back things only
+    // this client could have sent it. The reason survives; they do not.
+    let echoing = MockServer::start().await;
+    mount_page(
+        &echoing,
+        None,
+        ResponseTemplate::new(400).set_body_json(json!({
+            "error": {"message": format!("{REASON} key={KEY} header=custom-header-canary")}
+        })),
+    )
+    .await;
+    let client = anthropic_client(&echoing.uri());
+    crate::retry_status::clear();
+    let error = client.list_models().await.unwrap_err();
+    assert!(format!("{error:#}").contains(REASON), "{error:#}");
+    assert_no_canaries(&error);
+    crate::retry_status::clear();
+}
+
 #[tokio::test]
 async fn later_page_http_and_transport_errors_do_not_expose_cursor_or_key() {
     for isolated in [false, true] {
diff --git a/crates/tui/src/tui/app.rs b/crates/tui/src/tui/app.rs
index a878e5d7aa..04af62da03 100644
--- a/crates/tui/src/tui/app.rs
+++ b/crates/tui/src/tui/app.rs
@@ -70,9 +70,9 @@ pub(crate) enum RedactionGateNotice {
 }
 pub use types::{
     AppAction, AppModeUi, AutomationAction, ComposerDensity, ComposerSubmitAction,
-    ComposerSubmitChord, InitialInput, McpUiAction, QueuedMessage, ScreenMode, SettingSelection,
-    ShellJobAction, SubmitDisposition, TaskPanelEntry, TaskPanelEntryKind, ToolCollapseMode,
-    ToolDetailRecord, TranscriptSpacing, TuiOptions, VimMode,
+    ComposerSubmitChord, InflightSteer, InitialInput, McpUiAction, QueuedMessage, ScreenMode,
+    SettingSelection, ShellJobAction, SubmitDisposition, TaskPanelEntry, TaskPanelEntryKind,
+    ToolCollapseMode, ToolDetailRecord, TranscriptSpacing, TuiOptions, VimMode,
 };
 pub(crate) use types::{
     CacheReplayTarget, GoalControlIntent, PendingGoalControl, WORKFLOW_DRAFT_INSTRUCTION_PREFIX,
@@ -2275,6 +2275,12 @@ pub struct App {
     /// channel and the bucket renders with a rejected-steer label when
     /// populated.
     pub rejected_steers: VecDeque<String>,
+    /// Steers accepted by the steer channel but not yet seen in the engine's
+    /// record. Rendered through the same "sending into turn" preview bucket as
+    /// `pending_steers`; promoted to a transcript cell by
+    /// `apply_engine_session_projection`, or moved to `rejected_steers` by
+    /// `TurnComplete` when the turn ended without them (#6190).
+    pub inflight_steers: VecDeque<InflightSteer>,
     /// Legacy resend flag for pending steer recovery.
     pub submit_pending_steers_after_interrupt: bool,
     /// Start time for current turn
diff --git a/crates/tui/src/tui/app/init.rs b/crates/tui/src/tui/app/init.rs
index b05154141a..7ad6107cc9 100644
--- a/crates/tui/src/tui/app/init.rs
+++ b/crates/tui/src/tui/app/init.rs
@@ -1086,6 +1086,7 @@ impl App {
             queued_draft: None,
             pending_steers: VecDeque::new(),
             rejected_steers: VecDeque::new(),
+            inflight_steers: VecDeque::new(),
             submit_pending_steers_after_interrupt: false,
             turn_started_at: None,
             turn_last_activity_at: None,
diff --git a/crates/tui/src/tui/app/types.rs b/crates/tui/src/tui/app/types.rs
index 5a9cf84ed3..4e1cc715a6 100644
--- a/crates/tui/src/tui/app/types.rs
+++ b/crates/tui/src/tui/app/types.rs
@@ -284,6 +284,32 @@ pub struct QueuedMessage {
     pub history_echoed: bool,
 }
 
+/// A steer handed to the engine that the engine has not yet recorded.
+///
+/// Live-only, and deliberately not in `api_messages`: `EngineHandle::steer`
+/// succeeding means the channel took the text, not that a turn accepted it.
+/// The engine commits a steer at a step boundary and drops one whose turn has
+/// already moved on, so painting a settled transcript cell at send time
+/// produced a cell that could sit above the work it followed, or survive
+/// forever for a steer the model never saw (#6190). It becomes a real cell
+/// when the engine's own record shows it, and a "could not send" receipt when
+/// the turn ends without it.
+#[derive(Debug, Clone)]
+pub struct InflightSteer {
+    /// The composed message, carried so acceptance can paint the same cell
+    /// (including the queue-time echo it may already own).
+    pub message: QueuedMessage,
+    /// Exactly what was handed to `EngineHandle::steer`. The engine records
+    /// this as the first text block of the accepted user message, which is
+    /// what acceptance matches on.
+    pub content: String,
+    /// `api_messages.len()` when the steer was sent — the lower bound for the
+    /// acceptance search, so an identical earlier message cannot claim it.
+    pub sent_after_index: usize,
+    /// Held until acceptance knows the message index to anchor them to.
+    pub references: Vec<codewhale_core::ContextReference>,
+}
+
 /// Prefix for the bounded, tool-less model turn produced by `/workflow`.
 ///
 /// The marker travels with the queued message so a draft that waits behind an
diff --git a/crates/tui/src/tui/external_editor.rs b/crates/tui/src/tui/external_editor.rs
index 072a54115f..063e79a560 100644
--- a/crates/tui/src/tui/external_editor.rs
+++ b/crates/tui/src/tui/external_editor.rs
@@ -204,6 +204,17 @@ fn with_suspended_tui(
     use_bracketed_paste: bool,
     body: impl FnOnce() -> io::Result<EditorOutcome>,
 ) -> io::Result<EditorOutcome> {
+    // 0. Stop reading the tty.
+    // #6165: suspending crossterm state is not enough. The input pump runs on
+    // its own thread and keeps calling `event::read()` whatever mode the
+    // terminal is in, so a child launched without this pause competes with
+    // Codewhale for every keystroke — the editor cannot be quit and the
+    // fragments land in the composer. Pausing here, rather than at each call
+    // site, is what makes `/hooks edit` and the composer editor correct by
+    // the same construction. Fail closed: a pump that will not stop means the
+    // handoff would reproduce the defect, so the editor does not run.
+    let input_pause = crate::tui::ui::pause_terminal_input_for_child()?;
+
     // 1. Suspend.
     // Focus reporting is about to be disabled. Fail closed to the quiet state
     // so a stale FocusLost cannot authorize a surprise notification while an
@@ -244,6 +255,9 @@ fn with_suspended_tui(
     // viewport stale.
     let _ = terminal.clear();
 
+    // 4. Take the tty back, after the modes it reads under are restored.
+    drop(input_pause);
+
     result
 }
 
diff --git a/crates/tui/src/tui/ui.rs b/crates/tui/src/tui/ui.rs
index d71b3341ce..fe86256bbb 100644
--- a/crates/tui/src/tui/ui.rs
+++ b/crates/tui/src/tui/ui.rs
@@ -911,6 +911,9 @@ mod terminal;
 mod terminal_input;
 use remote_control_bridge::*;
 use terminal_input::*;
+// #6165: `external_editor` is a sibling of `ui`, and the pump pause now lives
+// inside its `with_suspended_tui` so no editor entry point can forget it.
+pub(crate) use terminal_input::pause_terminal_input_for_child;
 
 pub(crate) use dispatch::*;
 pub(crate) use motion::*;
diff --git a/crates/tui/src/tui/ui/dispatch.rs b/crates/tui/src/tui/ui/dispatch.rs
index ea7a761084..ece0b9a3e7 100644
--- a/crates/tui/src/tui/ui/dispatch.rs
+++ b/crates/tui/src/tui/ui/dispatch.rs
@@ -1214,28 +1214,95 @@ pub(crate) async fn steer_user_message(
     }
     app.last_submitted_prompt = Some(message.display.clone());
 
-    // Flush any streaming thinking/tool content into history before
-    // inserting the steer message, so the steer appears after (below)
-    // the content that chronologically preceded it.
-    app.flush_active_cell();
-
-    // Mirror steer input in local transcript/session state. A message echoed
-    // at queue time already owns a transcript cell; rewrite that cell into the
-    // steer form instead of painting a second bubble.
-    let history_cell = paint_user_turn_cell(app, &message, format!("+ {}", message.display));
-    app.record_context_references(history_cell, message_index, references);
-    app.push_api_message(Message {
-        role: Role::User,
-        content: vec![ContentBlock::Text {
-            text: content.clone(),
-            cache_control: None,
-        }],
-    });
+    // #6190: the steer channel accepting the text is not the turn accepting
+    // it. The engine commits a steer at the next step boundary and discards
+    // one whose turn has already moved on, so painting a settled cell and
+    // pushing `api_messages` here produced two defects at once: the cell sat
+    // above the assistant content the record places before it, and a dropped
+    // steer left a transcript entry the model never saw. Hold it as in-flight
+    // instead — it renders in the "sending into turn" preview until the
+    // engine's own `SessionUpdated` shows it, which is also where it learns
+    // its real message index.
+    app.inflight_steers
+        .push_back(crate::tui::app::InflightSteer {
+            message,
+            content,
+            sent_after_index: message_index,
+            references,
+        });
+    app.needs_redraw = true;
 
     app.status_message = Some("Steering current turn...".to_string());
     Ok(true)
 }
 
+/// Promote every in-flight steer the engine's record now contains.
+///
+/// Called from `apply_engine_session_projection` after the projection lands,
+/// so the transcript cell is appended in the position the record gives it:
+/// below the assistant work that preceded the steer, as the newest entry.
+/// Matching is on the exact text handed to `EngineHandle::steer`, which the
+/// engine stores as the accepted user message's first text block, searched
+/// from the index the steer was sent after so an identical earlier message
+/// cannot claim it.
+pub(crate) fn settle_accepted_steers(app: &mut App) {
+    if app.inflight_steers.is_empty() {
+        return;
+    }
+    let mut claimed: Vec<usize> = Vec::new();
+    let mut unsettled = VecDeque::new();
+    for steer in std::mem::take(&mut app.inflight_steers) {
+        let Some(index) = accepted_steer_index(app, &steer, &claimed) else {
+            unsettled.push_back(steer);
+            continue;
+        };
+        claimed.push(index);
+        // Settle the streaming thinking/tool content that chronologically
+        // preceded the steer before the steer's own cell is appended.
+        app.flush_active_cell();
+        let display = format!("+ {}", steer.message.display);
+        let history_cell = paint_user_turn_cell(app, &steer.message, display);
+        app.record_context_references(history_cell, index, steer.references);
+        app.needs_redraw = true;
+    }
+    app.inflight_steers = unsettled;
+}
+
+fn accepted_steer_index(
+    app: &App,
+    steer: &crate::tui::app::InflightSteer,
+    claimed: &[usize],
+) -> Option<usize> {
+    let start = steer.sent_after_index.min(app.api_messages.len());
+    app.api_messages
+        .iter()
+        .enumerate()
+        .skip(start)
+        .find(|(index, message)| {
+            !claimed.contains(index)
+                && message.role == Role::User
+                && matches!(
+                    message.content.first(),
+                    Some(ContentBlock::Text { text, .. }) if text == &steer.content
+                )
+        })
+        .map(|(index, _)| index)
+}
+
+/// A turn that ended without accepting a steer dropped it. Say so where the
+/// steer was already being shown, instead of leaving it "sending" forever or
+/// — as before #6190 — leaving a transcript cell for input the model never
+/// received.
+pub(crate) fn settle_unaccepted_steers_at_turn_end(app: &mut App) {
+    if app.inflight_steers.is_empty() {
+        return;
+    }
+    for steer in std::mem::take(&mut app.inflight_steers) {
+        app.rejected_steers.push_back(steer.message.display);
+    }
+    app.needs_redraw = true;
+}
+
 pub(crate) fn snapshot_steer_paused_state(app: &App) -> SteerPausedSnapshot {
     SteerPausedSnapshot {
         paused: app.paused,
diff --git a/crates/tui/src/tui/ui/event_loop.rs b/crates/tui/src/tui/ui/event_loop.rs
index 6bb7a74b70..41c018c0e5 100644
--- a/crates/tui/src/tui/ui/event_loop.rs
+++ b/crates/tui/src/tui/ui/event_loop.rs
@@ -61,6 +61,11 @@ pub(super) fn apply_engine_session_projection(
     }
     app.context_token_cache.borrow_mut().clear();
     app.set_api_messages(messages);
+    // #6190: the projection is the engine's own record, so it is where a
+    // steer's acceptance becomes observable — and the only place the steer's
+    // real message index is known. Promote before anything else reads the
+    // transcript, so live order equals record order by construction.
+    crate::tui::ui::dispatch::settle_accepted_steers(app);
     app.system_prompt = system_prompt;
     if app.auto_model {
         app.last_effective_model = Some(model);
@@ -2434,6 +2439,10 @@ pub(crate) async fn run_event_loop(
                         if flush_gate_receipts_for(app, None) {
                             transcript_batch_updated = true;
                         }
+                        // A steer the turn never accepted was dropped by the
+                        // engine. Report it instead of leaving it "sending"
+                        // (#6190).
+                        crate::tui::ui::dispatch::settle_unaccepted_steers_at_turn_end(app);
                         let completed_turn = app.active_turn.take();
                         // The in-flight provisional estimate hands off to the
                         // authoritative cumulative price accrued below; the
diff --git a/crates/tui/src/tui/ui/frame.rs b/crates/tui/src/tui/ui/frame.rs
index f016457d4b..749f0958b5 100644
--- a/crates/tui/src/tui/ui/frame.rs
+++ b/crates/tui/src/tui/ui/frame.rs
@@ -1399,9 +1399,13 @@ pub(crate) fn build_pending_input_preview(app: &App) -> PendingInputPreview {
             }
         })
         .collect();
+    // #6190: a steer the engine has not recorded yet is exactly what this
+    // bucket's "sending into turn" label describes, so it shares it rather
+    // than growing a fourth bucket and a fifteenth locale string.
     preview.pending_steers = app
         .pending_steers
         .iter()
+        .chain(app.inflight_steers.iter().map(|steer| &steer.message))
         .map(|m| m.display.clone())
         .collect();
     preview.rejected_steers = app.rejected_steers.iter().cloned().collect();
diff --git a/crates/tui/src/tui/ui/terminal_input.rs b/crates/tui/src/tui/ui/terminal_input.rs
index e08e124baa..5afb227cfe 100644
--- a/crates/tui/src/tui/ui/terminal_input.rs
+++ b/crates/tui/src/tui/ui/terminal_input.rs
@@ -3,7 +3,7 @@
 use std::cell::Cell;
 use std::io;
 use std::sync::{
-    Arc,
+    Arc, Mutex,
     atomic::{AtomicBool, Ordering},
 };
 use std::thread::{self, JoinHandle};
@@ -44,6 +44,105 @@ impl ObservedTerminalEvent {
     }
 }
 
+/// Process-wide handle on the one terminal input pump's pause flags.
+///
+/// There is one stdin per process and exactly one [`TerminalInputPump`]
+/// reading it, so this is a singleton by construction rather than by
+/// convention. It exists so that *handing the terminal to a child* can be one
+/// operation instead of a rule every call site has to remember (#6165):
+/// suspending raw mode and the alternate screen does not stop the pump
+/// thread, which keeps calling `event::read()` on the same tty and splits the
+/// user's keystrokes between the child and the composer.
+///
+/// Known limitation: the gate only stops the pump reading. It cannot drain
+/// input the pump already buffered, and it cannot refuse the handoff when a
+/// cancellation key is pending — both need the receiver and the event loop's
+/// pending queue, so they stay with [`super::prepare_terminal_input_handoff`]
+/// at the call sites that have them.
+static CHILD_TERMINAL_GATE: Mutex<Option<ChildTerminalGate>> = Mutex::new(None);
+
+#[derive(Clone)]
+struct ChildTerminalGate {
+    paused: Arc<AtomicBool>,
+    paused_ack: Arc<AtomicBool>,
+}
+
+/// Publish this pump as the process's terminal input owner.
+pub(super) fn publish_child_terminal_gate(paused: &Arc<AtomicBool>, paused_ack: &Arc<AtomicBool>) {
+    if let Ok(mut gate) = CHILD_TERMINAL_GATE.lock() {
+        *gate = Some(ChildTerminalGate {
+            paused: Arc::clone(paused),
+            paused_ack: Arc::clone(paused_ack),
+        });
+    }
+}
+
+/// Retract `paused`'s pump, but only if it is still the published one — a
+/// detached wedged thread must not unpublish the replacement that took over.
+fn retract_child_terminal_gate(paused: &Arc<AtomicBool>) {
+    if let Ok(mut gate) = CHILD_TERMINAL_GATE.lock()
+        && gate
+            .as_ref()
+            .is_some_and(|current| Arc::ptr_eq(&current.paused, paused))
+    {
+        *gate = None;
+    }
+}
+
+/// The terminal input pump, paused for as long as this guard is alive.
+///
+/// Held by [`crate::tui::external_editor::with_suspended_tui`] across the
+/// whole child handoff, so the pump resumes on every path out — including a
+/// child that failed to spawn or a panic unwinding through it.
+pub(crate) struct ChildTerminalInputPause {
+    gate: Option<ChildTerminalGate>,
+}
+
+/// Stop the process's terminal input pump before a child takes the tty.
+///
+/// Fails closed: if the pump does not acknowledge the pause, the caller must
+/// not run the child, because that is exactly the keystroke-splitting state
+/// this guards against. A process with no pump published (tests, non-TUI
+/// callers) has nothing to pause and succeeds with an inert guard, matching
+/// [`TerminalInputPump::pause_for_child_terminal`]'s `handle.is_none()` case.
+pub(crate) fn pause_terminal_input_for_child() -> io::Result<ChildTerminalInputPause> {
+    let Some(gate) = CHILD_TERMINAL_GATE
+        .lock()
+        .ok()
+        .and_then(|gate| gate.clone())
+    else {
+        return Ok(ChildTerminalInputPause { gate: None });
+    };
+    gate.paused.store(true, Ordering::Release);
+    let deadline = Instant::now() + TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT;
+    while !gate.paused_ack.load(Ordering::Acquire) {
+        if Instant::now() >= deadline {
+            gate.paused_ack.store(false, Ordering::Release);
+            gate.paused.store(false, Ordering::Release);
+            return Err(io::Error::new(
+                io::ErrorKind::TimedOut,
+                "terminal input pump did not pause before child terminal handoff",
+            ));
+        }
+        // Blocking-call convention (#6149): a bounded retry, capped by
+        // `TERMINAL_INPUT_CHILD_PAUSE_TIMEOUT`, in a synchronous API whose
+        // caller is about to block this very thread on a foreground editor
+        // for as long as the user keeps it open. `tokio::time` is not
+        // reachable from here and would not change what the thread does.
+        thread::sleep(TERMINAL_INPUT_CHILD_PAUSE_POLL_INTERVAL);
+    }
+    Ok(ChildTerminalInputPause { gate: Some(gate) })
+}
+
+impl Drop for ChildTerminalInputPause {
+    fn drop(&mut self) {
+        if let Some(gate) = self.gate.take() {
+            gate.paused_ack.store(false, Ordering::Release);
+            gate.paused.store(false, Ordering::Release);
+        }
+    }
+}
+
 pub(crate) struct TerminalInputPump {
     pub(super) rx: std::sync::mpsc::Receiver<TerminalInputMessage>,
     pub(super) stop: Arc<AtomicBool>,
@@ -64,6 +163,7 @@ pub(super) struct TerminalInputPumpParts {
 impl TerminalInputPump {
     pub(super) fn spawn() -> io::Result<Self> {
         let parts = Self::spawn_parts()?;
+        publish_child_terminal_gate(&parts.paused, &parts.paused_ack);
         Ok(Self {
             rx: parts.rx,
             stop: parts.stop,
@@ -249,10 +349,12 @@ impl TerminalInputPump {
     pub(super) fn detach_current_thread(&mut self) {
         self.stop.store(true, Ordering::Release);
         let _ = self.handle.take();
+        retract_child_terminal_gate(&self.paused);
     }
 
     /// Adopt freshly spawned pump parts and reset the liveness clock.
     pub(super) fn install_parts(&mut self, parts: TerminalInputPumpParts) {
+        publish_child_terminal_gate(&parts.paused, &parts.paused_ack);
         self.rx = parts.rx;
         self.stop = parts.stop;
         self.paused = parts.paused;
@@ -271,6 +373,7 @@ impl Drop for TerminalInputPump {
         // (or its send fails because `rx` was dropped) and exits on its own.
         self.stop.store(true, Ordering::Release);
         let _ = self.handle.take();
+        retract_child_terminal_gate(&self.paused);
     }
 }
 
diff --git a/crates/tui/src/tui/ui/tests.rs b/crates/tui/src/tui/ui/tests.rs
index 1346961ca2..8a6cd434a4 100644
--- a/crates/tui/src/tui/ui/tests.rs
+++ b/crates/tui/src/tui/ui/tests.rs
@@ -11639,10 +11639,146 @@ async fn live_steer_crosses_message_submit_transform_exactly_once() {
         Some("transformed steer")
     );
     assert_eq!(std::fs::read_to_string(count).expect("hook count"), "x");
-    assert!(app.api_messages.iter().any(|message| matches!(
-        &message.content[0],
-        ContentBlock::Text { text, .. } if text == "transformed steer"
-    )));
+    // #6190: the transform's output is what the engine was handed, and it is
+    // held as in-flight until the engine's own record shows it. Pushing it
+    // into `api_messages` at send time was the reorder this fix removes.
+    assert_eq!(
+        app.inflight_steers
+            .iter()
+            .map(|steer| steer.content.as_str())
+            .collect::<Vec<_>>(),
+        vec!["transformed steer"]
+    );
+    assert!(
+        !app.api_messages.iter().any(|message| matches!(
+            &message.content[0],
+            ContentBlock::Text { text, .. } if text == "transformed steer"
+        )),
+        "a steer the engine has not recorded yet must not be in the local transcript"
+    );
+}
+
+/// #6190: steering did not place the steer as the newest transcript entry —
+/// it was painted at send time, so it sat above assistant work the engine's
+/// record places before it, and the live transcript disagreed with the
+/// replayed one. The steer now becomes a cell when, and where, the engine
+/// records it.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn steer_becomes_the_newest_transcript_entry_only_when_the_engine_records_it() {
+    let _environment = crate::test_support::lock_test_env();
+    let mut app = create_test_app();
+    let session = super::event_loop::ensure_runtime_session_id(&mut app);
+    app.api_messages = vec![text_message("user", "original request")];
+    app.add_message(HistoryCell::Assistant {
+        content: "work produced before the steer arrived".to_string(),
+        streaming: false,
+    });
+    app.is_loading = true;
+    let mut engine = crate::core::engine::mock_engine_handle();
+
+    attempt_steer_with_queue_fallback(
+        &mut app,
+        &Config::default(),
+        &engine.handle,
+        QueuedMessage::new("actually use the other file".to_string(), None),
+        DispatchRecovery::Immediate,
+    )
+    .await;
+    assert_eq!(
+        engine.rx_steer.recv().await.as_deref(),
+        Some("actually use the other file")
+    );
+
+    // The channel took it; the turn has not. Nothing is settled yet.
+    assert!(
+        !app.history
+            .iter()
+            .any(|cell| matches!(cell, HistoryCell::User { .. })),
+        "a steer must not own a transcript cell before the engine records it"
+    );
+    assert_eq!(app.api_messages.len(), 1);
+    assert_eq!(
+        build_pending_input_preview(&app).pending_steers,
+        vec!["actually use the other file".to_string()],
+        "an unaccepted steer is shown as still sending, not as transcript"
+    );
+
+    // The engine commits the steer at its step boundary, after the assistant
+    // message it followed. `is_loading` is cleared first only because the
+    // mid-turn checkpoint branch of the projection is not what this pins.
+    app.is_loading = false;
+    let model = app.model.clone();
+    let workspace = app.workspace.clone();
+    assert!(super::event_loop::apply_engine_session_projection(
+        &mut app,
+        &Config::default(),
+        EngineEvent::SessionUpdated {
+            session_id: session,
+            messages: vec![
+                text_message("user", "original request"),
+                text_message("assistant", "work produced before the steer arrived"),
+                text_message("user", "actually use the other file"),
+            ],
+            system_prompt: None,
+            model,
+            workspace,
+        }
+    ));
+
+    assert!(app.inflight_steers.is_empty(), "the steer was accepted");
+    assert!(build_pending_input_preview(&app).pending_steers.is_empty());
+    assert!(
+        matches!(
+            app.history.last(),
+            Some(HistoryCell::User { content }) if content == "+ actually use the other file"
+        ),
+        "the steer must be the newest transcript entry: {:?}",
+        app.history.last()
+    );
+}
+
+/// #6190 case D: `next_turn_steer` drains and *discards* a steer stamped with
+/// a turn that has already moved on. The toast said "sent into turn" and the
+/// cell stayed in history forever, for input the model never received. The
+/// steer now settles as a "could not send" receipt instead.
+#[cfg(not(windows))]
+#[tokio::test]
+async fn steer_the_turn_never_accepted_is_reported_not_left_in_the_transcript() {
+    let mut app = create_test_app();
+    app.is_loading = true;
+    let mut engine = crate::core::engine::mock_engine_handle();
+
+    attempt_steer_with_queue_fallback(
+        &mut app,
+        &Config::default(),
+        &engine.handle,
+        QueuedMessage::new("too late to matter".to_string(), None),
+        DispatchRecovery::Immediate,
+    )
+    .await;
+    assert_eq!(
+        engine.rx_steer.recv().await.as_deref(),
+        Some("too late to matter")
+    );
+
+    // The turn ends without the engine ever recording it.
+    super::dispatch::settle_unaccepted_steers_at_turn_end(&mut app);
+
+    assert!(app.inflight_steers.is_empty());
+    assert!(
+        !app.history
+            .iter()
+            .any(|cell| matches!(cell, HistoryCell::User { .. })),
+        "a dropped steer must not leave a transcript cell the record never had"
+    );
+    assert!(app.api_messages.is_empty());
+    let preview = build_pending_input_preview(&app);
+    assert!(preview.pending_steers.is_empty());
+    assert_eq!(
+        preview.rejected_steers,
+        vec!["too late to matter".to_string()]
+    );
 }
 
 #[cfg(not(windows))]
@@ -26061,8 +26197,104 @@ async fn terminal_input_handoff_preserves_pending_cancellation_keys() {
     input.resume_after_child_terminal();
 }
 
+/// `CHILD_TERMINAL_GATE` is process-global — one stdin, one pump. Any test
+/// that publishes or reads it must hold this, including the restart test,
+/// which publishes through `install_parts`.
+static CHILD_TERMINAL_GATE_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+/// #6165: `/hooks edit` handed the terminal to `$EDITOR` with the pump still
+/// reading stdin, so `Esc` and `Enter` were eaten by the composer while `:`
+/// and `!` reached `vi`. The pause now lives inside `with_suspended_tui`, so
+/// this pins what that guard must do to the pump: stop it reading for the
+/// whole handoff, and start it again on the way out.
+#[test]
+fn child_terminal_pause_stops_the_pump_reading_and_resumes_it_on_drop() {
+    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+
+    let paused = std::sync::Arc::new(AtomicBool::new(false));
+    let paused_ack = std::sync::Arc::new(AtomicBool::new(false));
+    let stop = std::sync::Arc::new(AtomicBool::new(false));
+    let reads = std::sync::Arc::new(AtomicUsize::new(0));
+    // Mirrors the real pump loop's pause handling: acknowledge and stop
+    // reading while paused, read otherwise. Spawning the crossterm pump here
+    // would need an interactive terminal.
+    let worker = {
+        let (paused, paused_ack, stop, reads) = (
+            std::sync::Arc::clone(&paused),
+            std::sync::Arc::clone(&paused_ack),
+            std::sync::Arc::clone(&stop),
+            std::sync::Arc::clone(&reads),
+        );
+        std::thread::spawn(move || {
+            while !stop.load(Ordering::Acquire) {
+                if paused.load(Ordering::Acquire) {
+                    paused_ack.store(true, Ordering::Release);
+                } else {
+                    paused_ack.store(false, Ordering::Release);
+                    reads.fetch_add(1, Ordering::Release);
+                }
+                std::thread::sleep(Duration::from_millis(1));
+            }
+        })
+    };
+    super::terminal_input::publish_child_terminal_gate(&paused, &paused_ack);
+
+    let guard = pause_terminal_input_for_child().expect("the pump acknowledges the pause");
+    assert!(paused.load(Ordering::Acquire));
+    let while_child_owns_it = reads.load(Ordering::Acquire);
+    std::thread::sleep(Duration::from_millis(30));
+    assert_eq!(
+        reads.load(Ordering::Acquire),
+        while_child_owns_it,
+        "the pump must not read the tty while a child owns the terminal"
+    );
+
+    drop(guard);
+    assert!(!paused.load(Ordering::Acquire));
+    let deadline = Instant::now() + Duration::from_secs(2);
+    while reads.load(Ordering::Acquire) == while_child_owns_it {
+        assert!(
+            Instant::now() < deadline,
+            "the pump must read again once the child releases the terminal"
+        );
+        std::thread::sleep(Duration::from_millis(1));
+    }
+
+    stop.store(true, Ordering::Release);
+    let _ = worker.join();
+}
+
+/// A pump that will not stop means the handoff would reproduce #6165, so the
+/// editor must not run — and the refusal must leave the pump reading.
+#[test]
+fn child_terminal_pause_refuses_the_handoff_when_the_pump_never_acknowledges() {
+    use std::sync::atomic::{AtomicBool, Ordering};
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
+
+    let paused = std::sync::Arc::new(AtomicBool::new(false));
+    let paused_ack = std::sync::Arc::new(AtomicBool::new(false));
+    super::terminal_input::publish_child_terminal_gate(&paused, &paused_ack);
+
+    let error = pause_terminal_input_for_child()
+        .err()
+        .expect("an unacknowledged pause must fail the handoff");
+    assert_eq!(error.kind(), std::io::ErrorKind::TimedOut);
+    assert!(
+        !paused.load(Ordering::Acquire),
+        "a refused handoff must leave the pump reading, not wedged paused"
+    );
+}
+
 #[test]
 fn input_pump_restart_detaches_wedged_thread_and_installs_fresh_parts() {
+    let _serialized = CHILD_TERMINAL_GATE_TEST_LOCK
+        .lock()
+        .unwrap_or_else(|poisoned| poisoned.into_inner());
     // A "wedged" pump thread blocked forever on a channel recv stands in for
     // a crossterm `event::read` that never returns (stalled Windows console
     // poll, or a Unix tty that stopped delivering bytes). Joining it would