From fc62a9b3ff5df75685a1473034792eb9d2b010e8 Mon Sep 17 00:00:00 2001
From: zane668 <152239216+zane668@users.noreply.github.com>
Date: Sun, 18 Oct 2026 22:23:16 -0400
Subject: [PATCH 1/4] fix(grok): acknowledge steer-injected mail before
 requeueing

Recognize exact same-session interjection evidence for a removed queued batch and retain its identity during the existing retry pause. Recheck the durable live branch after rewind so surviving injections are acknowledged without accepting stale or removed-branch evidence.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/delivery/grok.rs | 576 ++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 567 insertions(+), 9 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 277be5a9..2e08be51 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -739,9 +739,15 @@ struct InFlight {
     /// The session's `updates.jsonl`, checked before re-queueing.
     transcript: std::path::PathBuf,
     /// When Grok answered `removedFromQueue`: either removed, or merged into
-    /// the prompt ahead of it (which then runs its text). Only a combined-turn
-    /// snapshot tells them apart, and it may arrive after the response.
+    /// the prompt ahead of it (which then runs its text), or injected by
+    /// steer. The confirming queue/update evidence may arrive later.
     removed_at: Option<Instant>,
+    /// Live steer evidence, pending its removal response. A rewind invalidates
+    /// the cache until its marker is durable and the live branch is rechecked.
+    interjection: SteerEvidence,
+    /// Keep tracking this batch during the existing re-queue pause: an
+    /// injection or its transcript can arrive after the merge wait.
+    requeue_pending: bool,
     /// When the ACP connection that queued it dropped. Grok keeps a queued
     /// prompt when its client disconnects, so the batch is followed through
     /// the reconnected client's queue snapshots, not treated as dropped.
@@ -762,6 +768,72 @@ enum Outcome {
     Dropped(String),
 }
 
+enum SteerEvidence {
+    Unseen,
+    Injected,
+    Rewound(Value),
+}
+
+/// Steer runs a queued prompt inside the current turn. Its display text is
+/// the original prompt, while content.text wraps it in Grok's instructions.
+fn is_interjection(params: &Value, flight: &InFlight) -> bool {
+    let update = &params["update"];
+    params["sessionId"].as_str() == Some(flight.session.as_str())
+        && update["sessionUpdate"] == "user_message_chunk"
+        && update["_meta"]["interjection"] == true
+        && update["content"]["type"] == "text"
+        && update["content"]["_meta"]["displayText"].as_str() == Some(flight.text.as_str())
+        && update["content"]["text"]
+            .as_str()
+            .is_some_and(|text| text.contains(&flight.text))
+}
+
+fn observe_interjection(method: &str, params: &Value, flight: &mut InFlight) {
+    if method == "session/update" && is_interjection(params, flight) {
+        flight.interjection = SteerEvidence::Injected;
+    } else if method == "x.ai/session/update"
+        && params["sessionId"].as_str() == Some(flight.session.as_str())
+        && params["update"]["sessionUpdate"] == "rewind_marker"
+    {
+        // A rewind can remove the injected turn before its file is flushed.
+        // Keep the exact marker, including its event metadata, so an older
+        // disk branch cannot restore the invalidated cache.
+        flight.interjection = SteerEvidence::Rewound(params.clone());
+    }
+}
+
+fn interjected_in_transcript(flight: &InFlight) -> bool {
+    let Ok(content) = std::fs::read_to_string(&flight.transcript) else {
+        return false;
+    };
+    if let SteerEvidence::Rewound(marker) = &flight.interjection
+        && !content.lines().any(|line| {
+            serde_json::from_str::<Value>(line).is_ok_and(|value| {
+                value["method"] == "_x.ai/session/update" && value["params"] == *marker
+            })
+        })
+    {
+        return false;
+    }
+    crate::transcript::grok::live_lines(&content)
+        .into_iter()
+        .rev()
+        .any(|line| {
+            serde_json::from_str::<Value>(line).is_ok_and(|value| {
+                value["method"] == "session/update" && is_interjection(&value["params"], flight)
+            })
+        })
+}
+
+fn interjection_outcome(flight: &InFlight) -> Option<Outcome> {
+    flight.removed_at?;
+    let injected = match flight.interjection {
+        SteerEvidence::Unseen | SteerEvidence::Rewound(_) => interjected_in_transcript(flight),
+        SteerEvidence::Injected => true,
+    };
+    injected.then_some(Outcome::Delivered)
+}
+
 /// What a queue snapshot proves about our prompt. Only identity counts: our
 /// prompt id running, or our exact text among the running combined texts
 /// (Grok merges queued plain prompts into the front one).
@@ -803,6 +875,9 @@ fn ran_in_transcript(path: &std::path::Path, text: &str) -> bool {
                 && serde_json::from_str::<Value>(line).is_ok_and(|value| {
                     let update = &value["params"]["update"];
                     update["sessionUpdate"] == "user_message_chunk"
+                        // A steer wrapper needs the stricter display-text
+                        // identity check above, not just a body substring.
+                        && update["_meta"]["interjection"] != true
                         && update["content"]["text"]
                             .as_str()
                             .is_some_and(|sent| sent.contains(text))
@@ -1115,6 +1190,9 @@ pub(super) fn run(
                     }
                 }
                 Event::Notification { method, params } => {
+                    if let Some(flight) = in_flight.as_mut() {
+                        observe_interjection(&method, &params, flight);
+                    }
                     if method == "x.ai/queue/changed"
                         && let Some(flight) = in_flight.as_ref()
                         && params["sessionId"].as_str() == Some(flight.session.as_str())
@@ -1222,9 +1300,15 @@ pub(super) fn run(
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
+        if outcome.is_none()
+            && let Some(flight) = in_flight.as_ref()
+        {
+            outcome = interjection_outcome(flight);
+        }
         if outcome.is_none()
             && in_flight
                 .as_ref()
+                .filter(|flight| !flight.requeue_pending)
                 .and_then(|flight| flight.removed_at)
                 .is_some_and(|at| at.elapsed() >= MERGE_EVIDENCE_WAIT)
         {
@@ -1233,6 +1317,7 @@ pub(super) fn run(
         if outcome.is_none()
             && in_flight
                 .as_ref()
+                .filter(|flight| !flight.requeue_pending)
                 .and_then(|flight| flight.detached_at)
                 .is_some_and(|at| at.elapsed() >= DETACHED_WAIT)
         {
@@ -1241,10 +1326,13 @@ pub(super) fn run(
             ));
         }
         if let Some(outcome) = outcome
-            && let Some(flight) = in_flight.take()
+            && let Some(mut flight) = in_flight.take()
         {
             let outcome = match outcome {
-                Outcome::Dropped(_) if ran_in_transcript(&flight.transcript, &flight.text) => {
+                Outcome::Dropped(_)
+                    if interjection_outcome(&flight).is_some()
+                        || ran_in_transcript(&flight.transcript, &flight.text) =>
+                {
                     Outcome::Delivered
                 }
                 other => other,
@@ -1277,14 +1365,33 @@ pub(super) fn run(
                 Outcome::Dropped(reason) => {
                     // Still unread; queue it again after a pause.
                     requeue_at = Instant::now() + REQUEUE_DELAY;
-                    log_warn(
-                        "native",
-                        "grok.acp.requeue",
-                        &format!("prompt={} {reason}; kept unread", flight.prompt_id),
-                    );
+                    if flight.removed_at.is_some() {
+                        flight.requeue_pending = true;
+                        in_flight = Some(flight);
+                    } else {
+                        log_warn(
+                            "native",
+                            "grok.acp.requeue",
+                            &format!("prompt={} {reason}; kept unread", flight.prompt_id),
+                        );
+                    }
                 }
             }
         }
+        if let Some(flight) = in_flight.as_ref()
+            && flight.requeue_pending
+            && Instant::now() >= requeue_at
+        {
+            log_warn(
+                "native",
+                "grok.acp.requeue",
+                &format!(
+                    "prompt={} removed from Grok's queue; kept unread",
+                    flight.prompt_id
+                ),
+            );
+            in_flight = None;
+        }
         let Some(conn) = client.as_mut() else {
             continue;
         };
@@ -1365,6 +1472,8 @@ pub(super) fn run(
                         text: prepared.formatted,
                         ack: prepared.ack,
                         removed_at: None,
+                        interjection: SteerEvidence::Unseen,
+                        requeue_pending: false,
                         detached_at: None,
                     });
                 }
@@ -1387,6 +1496,283 @@ mod tests {
 
     const BATCH: &str = "<hcom>[request #7] luna → nova: hi</hcom>";
 
+    // Synthetic packet shaped like an observed Grok steer injection. The
+    // original session/prompt response was not captured; removal responses
+    // below are inferred fixtures, not original-response replays.
+    const STEER_INJECTION_FIXTURE: &str = r#"{"timestamp":1000,"method":"session/update","params":{"sessionId":"fixture-session","update":{"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"The user sent a message while you were working:\n<user_query>\n<hcom>[request #7] luna → nova: Please continue the current task and acknowledge this message.</hcom>\n</user_query>\nIf the user is asking for a response, address the user first. After replying, complete any unfinished tasks from previous turns.","_meta":{"displayText":"<hcom>[request #7] luna → nova: Please continue the current task and acknowledge this message.</hcom>"}},"_meta":{"modelId":"grok-4.7-build-fast","interjection":true}},"_meta":{"eventId":"fixture-session-injection","agentTimestampMs":1000000}}}"#;
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn steer_removed_batch_is_not_requeued_when_injection_arrives_late() {
+        exercise_removed_steer_fixture(true, None);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn removed_without_injection_still_requeues_and_stays_unread() {
+        exercise_removed_steer_fixture(false, None);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn rewind_preserving_earlier_steer_does_not_requeue() {
+        exercise_removed_steer_fixture(true, Some((1, 0)));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn rewind_before_disk_rechecks_preserved_steer() {
+        exercise_removed_steer_fixture(true, Some((1, 3)));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn rewind_before_disk_removing_steer_stays_unread() {
+        exercise_removed_steer_fixture(true, Some((0, 3)));
+    }
+
+    #[cfg(unix)]
+    fn exercise_removed_steer_fixture(inject: bool, rewind: Option<(u64, u64)>) {
+        // Exercise the production run/Client/ACK path against fixture ACP,
+        // including an injection after the original two-second decision.
+        let (_home_dir, _hcom, home, _env) = crate::hooks::test_helpers::isolated_test_env();
+        // EnvGuard does not manage GROK_HOME. Restore this override before
+        // releasing its environment lock, including during an assertion unwind.
+        struct GrokHome(Option<std::ffi::OsString>);
+        impl Drop for GrokHome {
+            fn drop(&mut self) {
+                unsafe {
+                    match &self.0 {
+                        Some(value) => std::env::set_var("GROK_HOME", value),
+                        None => std::env::remove_var("GROK_HOME"),
+                    }
+                }
+            }
+        }
+        let _grok_home = GrokHome(std::env::var_os("GROK_HOME"));
+        unsafe { std::env::set_var("GROK_HOME", home.join(".grok")) };
+        let dir = tempfile::tempdir().unwrap();
+        let capture = dir.path().join("prompts.jsonl");
+        let script = dir.path().join("acp.sh");
+        std::fs::write(&script, r#"capture=$1
+roster=$2
+queue=$3
+injection=$4
+rewind=$5
+later=$6
+transcript=$7
+disk_delay=$8
+while IFS= read -r line; do
+    id=${line#*'"id":'}
+    id=${id%%,*}
+    case "$line" in
+        *'"method":"initialize"'*)
+            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
+        *'"method":"_x.ai/sessions/list"'*)
+            printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$id" "$roster" ;;
+        *'"method":"session/load"'*)
+            printf '%s\n' "$queue"
+            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
+        *'"method":"session/prompt"'*)
+            printf '%s\n' "$line" >> "$capture"
+            if [ "$rewind" != '{}' ] && [ "$sent" != yes ]; then
+                # Inject prompt 0, then rewind only the later prompt 1 (or
+                # delete prompt 0 for the negative). The response is inferred.
+                printf '%s\n%s\n' "$injection" "$later" > "$transcript"
+                printf '%s\n%s\n%s\n' "$injection" "$later" "$rewind"
+                if [ "$disk_delay" = 0 ]; then printf '%s\n' "$rewind" >> "$transcript"; fi
+                printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
+                if [ "$disk_delay" != 0 ]; then
+                    sleep "$disk_delay"
+                    printf '%s\n' "$rewind" >> "$transcript"
+                fi
+                sent=yes
+            else
+                # INFERRED response, not a injection production packet.
+                printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
+            fi
+            if [ "$rewind" = '{}' ] && [ "$sent" != yes ]; then
+                sleep 3
+                printf '%s\n' "$injection"
+                sent=yes
+            fi ;;
+    esac
+done
+"#).unwrap();
+        let injected: Value = serde_json::from_str(STEER_INJECTION_FIXTURE).unwrap();
+        let session = injected["params"]["sessionId"].as_str().unwrap();
+        let exact = injected["params"]["update"]["content"]["_meta"]["displayText"]
+            .as_str()
+            .unwrap();
+        let text = exact
+            .strip_prefix("<hcom>[request #7] luna → nova: ")
+            .unwrap()
+            .strip_suffix("</hcom>")
+            .unwrap();
+        let cwd = dir.path().to_str().unwrap();
+        let transcript = crate::transcript::grok::session_transcript_path(cwd, session);
+        std::fs::create_dir_all(transcript.parent().unwrap()).unwrap();
+        // Synthetic later-turn/rewind and response fixtures. The injection
+        // has the shape observed in a real steer delivery.
+        let marker = rewind.map(|(target, _)| {
+            json!({
+            "method": "_x.ai/session/update", "params": {
+                "sessionId": session, "update": {"sessionUpdate": "rewind_marker",
+                    "target_prompt_index": target},
+                "_meta": {"eventId": "fixture-rewind-500"}}})
+        });
+        let later = json!({"method": "session/update", "params": {
+            "sessionId": session, "update": {"sessionUpdate": "user_message_chunk",
+                "content": {"type": "text", "text": "later prompt"},
+                "_meta": {"promptIndex": 1}}}});
+        let roster = json!({"sessions": [{"sessionId": session, "cwd": cwd,
+            "activity": "idle", "resident": true, "lastActivityAt": 1}]});
+        let queue = json!({"method": "_x.ai/queue/changed", "params": {
+            "sessionId": session, "runningPromptId": "user-turn", "entries": []}});
+        let launch = Launch::new(
+            "/bin/sh",
+            &[
+                script.to_string_lossy().into_owned(),
+                capture.to_string_lossy().into_owned(),
+                roster.to_string(),
+                queue.to_string(),
+                if inject {
+                    STEER_INJECTION_FIXTURE
+                } else {
+                    "{}"
+                }
+                .to_string(),
+                marker
+                    .as_ref()
+                    .map_or_else(|| "{}".into(), Value::to_string),
+                later.to_string(),
+                transcript.to_string_lossy().into_owned(),
+                rewind.map_or(0, |(_, delay)| delay).to_string(),
+            ],
+            &[],
+        )
+        .unwrap();
+        let mut db = HcomDb::open_raw(&dir.path().join("test.db")).unwrap();
+        db.init_db().unwrap();
+        db.conn()
+            .execute(
+                "INSERT INTO instances (name, tool, session_id, directory, status, created_at)
+             VALUES ('nova', 'grok', ?1, ?2, 'listening', 0)",
+                rusqlite::params![session, cwd],
+            )
+            .unwrap();
+        db.set_session_binding(session, "nova").unwrap();
+        db.set_process_binding("fixture-process", session, "nova")
+            .unwrap();
+        let event = db
+            .insert_event_row(
+                "message",
+                "luna",
+                &json!({
+            "from": "luna", "text": text, "intent": "request", "scope": "mentions",
+            "mentions": ["nova"], "delivered_to": ["nova"]}),
+                None,
+            )
+            .unwrap();
+        db.conn()
+            .execute("UPDATE events SET id=7 WHERE id=?1", [event])
+            .unwrap();
+        // Consume the first-receipt tip, so the next production batch is
+        // byte-for-byte identical to the synthetic displayText.
+        common::prepare_pending_messages(&db, "nova").unwrap();
+        assert_eq!(
+            common::prepare_pending_messages(&db, "nova")
+                .unwrap()
+                .formatted,
+            exact
+        );
+        let running = Arc::new(AtomicBool::new(true));
+        let stop = Arc::clone(&running);
+        let check_stale_file = rewind.is_some_and(|(_, delay)| delay > 0);
+        let test_db = dir.path().join("test.db");
+        let observer = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_secs(2));
+            let early_cursor = if check_stale_file {
+                let conn = rusqlite::Connection::open_with_flags(
+                    test_db,
+                    rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+                )
+                .unwrap();
+                Some(
+                    conn.query_row(
+                        "SELECT last_event_id FROM instances WHERE name='nova'",
+                        [],
+                        |row| row.get::<_, i64>(0),
+                    )
+                    .unwrap(),
+                )
+            } else {
+                None
+            };
+            std::thread::sleep(Duration::from_secs(12));
+            stop.store(false, Ordering::Release);
+            early_cursor
+        });
+        let state = DeliveryState {
+            screen: Arc::new(RwLock::new(super::super::ScreenState {
+                ready: true,
+                input_text: Some("PRIVATE_DRAFT".into()),
+                prompt_empty: false,
+                ..Default::default()
+            })),
+            grok_acp: None,
+            launch_phase_active: Arc::new(AtomicBool::new(false)),
+            inject_port: 0,
+            user_activity_cooldown_ms: 0,
+        };
+        let notify = NotifyServer::new().unwrap();
+        let mut name = "nova".to_string();
+        run(
+            &launch,
+            &running,
+            &mut db,
+            &notify,
+            &state,
+            "fixture-process",
+            &mut name,
+            &ToolConfig::for_tool(crate::tool::Tool::Grok),
+            &None,
+            &None,
+            &None,
+            &mut super::super::host_label::HostLabel::resolve(),
+            &mut LaunchOutcome::Ready,
+        );
+        if let Some(early_cursor) = observer.join().unwrap() {
+            assert_eq!(
+                early_cursor, 0,
+                "a live rewind must veto the not-yet-updated disk branch"
+            );
+        }
+        let packets: Vec<Value> = std::fs::read_to_string(&capture)
+            .unwrap()
+            .lines()
+            .map(|line| serde_json::from_str(line).unwrap())
+            .collect();
+        eprintln!(
+            "late steer: prompts={} cursor={}",
+            packets.len(),
+            db.get_cursor("nova")
+        );
+        let delivered = inject && rewind.is_none_or(|(target, _)| target == 1);
+        assert_eq!(packets.len(), if delivered { 1 } else { 2 });
+        assert_eq!(packets[0]["params"]["_meta"]["sendNow"], false);
+        assert_eq!(db.get_cursor("nova"), if delivered { 7 } else { 0 });
+        assert_eq!(
+            state.screen.read().unwrap().input_text.as_deref(),
+            Some("PRIVATE_DRAFT")
+        );
+    }
+
     fn flight(removed: bool) -> InFlight {
         InFlight {
             request_id: Some(4),
@@ -1403,10 +1789,182 @@ mod tests {
                 mark_announced: false,
             },
             removed_at: removed.then(Instant::now),
+            interjection: SteerEvidence::Unseen,
+            requeue_pending: false,
             detached_at: None,
         }
     }
 
+    fn steer_flight(removed: bool) -> (InFlight, Value) {
+        let injection: Value = serde_json::from_str(STEER_INJECTION_FIXTURE).unwrap();
+        let mut flight = flight(removed);
+        flight.session = injection["params"]["sessionId"].as_str().unwrap().into();
+        flight.text = injection["params"]["update"]["content"]["_meta"]["displayText"]
+            .as_str()
+            .unwrap()
+            .into();
+        (flight, injection)
+    }
+
+    #[test]
+    fn steer_injection_and_inferred_removal_can_arrive_in_either_order() {
+        // Removal is inferred from the observed requeue branch; its original
+        // response was not captured. The notification shape is synthetic.
+        let inferred = json!({"id": 4, "result": {
+            "_meta": {"completionKind": "removedFromQueue"}}});
+        for inject_first in [false, true] {
+            let (mut flight, injection) = steer_flight(false);
+            if inject_first {
+                observe_interjection("session/update", &injection["params"], &mut flight);
+                assert_eq!(
+                    interjection_outcome(&flight),
+                    None,
+                    "injection alone cannot claim a batch"
+                );
+            }
+            assert_eq!(response_outcome(&inferred), None);
+            flight.removed_at = Some(Instant::now());
+            if !inject_first {
+                assert_eq!(
+                    interjection_outcome(&flight),
+                    None,
+                    "manual queue removal stays unread"
+                );
+                observe_interjection("session/update", &injection["params"], &mut flight);
+            }
+            assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+        }
+    }
+
+    #[test]
+    fn steer_evidence_requires_the_exact_batch_session_and_user_injection() {
+        let (flight, injection) = steer_flight(true);
+        assert!(is_interjection(&injection["params"], &flight));
+        let variants = [
+            ("/params/sessionId", json!("another-session")),
+            ("/params/update/sessionUpdate", json!("agent_message_chunk")),
+            ("/params/update/_meta/interjection", json!(false)),
+            ("/params/update/content/type", json!("image")),
+            ("/params/update/content/text", json!("different body")),
+            (
+                "/params/update/content/_meta/displayText",
+                json!(format!("{} ", flight.text)),
+            ),
+            (
+                "/params/update/content/_meta/displayText",
+                json!("partial batch text"),
+            ),
+        ];
+        for (pointer, replacement) in variants {
+            let mut wrong = injection.clone();
+            *wrong.pointer_mut(pointer).unwrap() = replacement;
+            let mut pending = steer_flight(true).0;
+            observe_interjection("session/update", &wrong["params"], &mut pending);
+            assert_eq!(interjection_outcome(&pending), None, "{pointer}");
+        }
+        let mut wrong_method = steer_flight(true).0;
+        observe_interjection(
+            "x.ai/session_notification",
+            &injection["params"],
+            &mut wrong_method,
+        );
+        assert_eq!(interjection_outcome(&wrong_method), None);
+    }
+
+    #[test]
+    fn late_steer_transcript_evidence_is_consumed_only_on_its_live_branch() {
+        let dir = tempfile::tempdir().unwrap();
+        let (mut flight, injection) = steer_flight(true);
+        flight.transcript = dir.path().join("updates.jsonl");
+        flight.removed_at = Some(Instant::now() - MERGE_EVIDENCE_WAIT);
+        flight.requeue_pending = true;
+        assert_eq!(interjection_outcome(&flight), None);
+        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
+        assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+        // Generic substring proof must not bypass the strict steer check.
+        assert!(!ran_in_transcript(&flight.transcript, &flight.text));
+        let rewind = json!({"method": "_x.ai/session/update", "params": {
+            "sessionId": flight.session, "update": {
+                "sessionUpdate": "rewind_marker", "target_prompt_index": 0}}});
+        observe_interjection("session/update", &injection["params"], &mut flight);
+        observe_interjection("x.ai/session/update", &rewind["params"], &mut flight);
+        assert_eq!(
+            interjection_outcome(&flight),
+            None,
+            "a live rewind vetoes a stale file"
+        );
+        std::fs::write(&flight.transcript, format!("{injection}\n{rewind}\n")).unwrap();
+        flight.interjection = SteerEvidence::Unseen;
+        assert_eq!(
+            interjection_outcome(&flight),
+            None,
+            "rewound injection is not delivery evidence"
+        );
+        for (pointer, replacement) in [
+            ("/params/sessionId", json!("another-session")),
+            ("/params/update/content/text", json!("different body")),
+            (
+                "/params/update/content/_meta/displayText",
+                json!(format!("{} ", flight.text)),
+            ),
+            ("/params/update/sessionUpdate", json!("agent_message_chunk")),
+        ] {
+            let mut wrong = injection.clone();
+            *wrong.pointer_mut(pointer).unwrap() = replacement;
+            std::fs::write(&flight.transcript, format!("{wrong}\n")).unwrap();
+            assert_eq!(interjection_outcome(&flight), None, "{pointer}");
+            assert!(!ran_in_transcript(&flight.transcript, &flight.text));
+        }
+    }
+
+    #[test]
+    fn matching_rewind_must_be_durable_before_rechecking_the_surviving_branch() {
+        let dir = tempfile::tempdir().unwrap();
+        let (mut flight, injection) = steer_flight(true);
+        flight.transcript = dir.path().join("updates.jsonl");
+        let later = json!({"method": "session/update", "params": {
+            "sessionId": flight.session, "update": {"sessionUpdate": "user_message_chunk",
+                "content": {"type": "text", "text": "later"}, "_meta": {"promptIndex": 1}}}});
+        let marker = json!({"method": "_x.ai/session/update", "params": {
+            "sessionId": flight.session, "update": {"sessionUpdate": "rewind_marker",
+                "target_prompt_index": 1}, "_meta": {"eventId": "rewind-current"}}});
+        let mut older = marker.clone();
+        older["params"]["_meta"]["eventId"] = json!("rewind-older");
+        observe_interjection("session/update", &injection["params"], &mut flight);
+        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
+        std::fs::write(
+            &flight.transcript,
+            format!("{injection}\n{later}\n{older}\n"),
+        )
+        .unwrap();
+        assert_eq!(
+            interjection_outcome(&flight),
+            None,
+            "an older rewind cannot unlock the stale file"
+        );
+        std::fs::write(
+            &flight.transcript,
+            format!("{injection}\n{later}\n{marker}\n"),
+        )
+        .unwrap();
+        assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+        let mut deleting = marker.clone();
+        deleting["params"]["update"]["target_prompt_index"] = json!(0);
+        deleting["params"]["_meta"]["eventId"] = json!("rewind-delete");
+        observe_interjection("x.ai/session/update", &deleting["params"], &mut flight);
+        assert_eq!(interjection_outcome(&flight), None);
+        std::fs::write(
+            &flight.transcript,
+            format!("{injection}\n{later}\n{marker}\n{deleting}\n"),
+        )
+        .unwrap();
+        assert_eq!(
+            interjection_outcome(&flight),
+            None,
+            "a removed injection stays unread"
+        );
+    }
+
     #[test]
     fn running_prompt_is_delivered() {
         let params = json!({"sessionId": "s", "runningPromptId": "hcom-1", "entries": []});

From bdd062a412f39b88c73cb8fd51de430364a38fc9 Mon Sep 17 00:00:00 2001
From: zane668 <152239216+zane668@users.noreply.github.com>
Date: Mon, 19 Oct 2026 00:39:05 -0400
Subject: [PATCH 2/4] fix(grok): preserve steer evidence across reconnects and
 retries

Recover exact durable interjections when a detached prompt loses its response, and keep the first retry deadline stable across repeated missing queue snapshots. Throttle pending transcript scans, invalidate them on rewind, and force a final read before retrying. Require the complete post-submission retry observation window in stdio regressions.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/delivery/grok.rs | 289 +++++++++++++++++++++++++++++++++++--------
 1 file changed, 235 insertions(+), 54 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 2e08be51..11cc2aa7 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -745,6 +745,8 @@ struct InFlight {
     /// Live steer evidence, pending its removal response. A rewind invalidates
     /// the cache until its marker is durable and the live branch is rechecked.
     interjection: SteerEvidence,
+    /// Limit full transcript scans while the live evidence is pending.
+    transcript_check_at: Option<Instant>,
     /// Keep tracking this batch during the existing re-queue pause: an
     /// injection or its transcript can arrive after the merge wait.
     requeue_pending: bool,
@@ -756,6 +758,7 @@ struct InFlight {
 
 /// How long to wait after `removedFromQueue` for the snapshot showing a merge.
 const MERGE_EVIDENCE_WAIT: Duration = Duration::from_secs(2);
+const INTERJECTION_TRANSCRIPT_POLL: Duration = Duration::from_secs(1);
 /// How long a batch whose connection dropped waits for a queue snapshot
 /// before the transcript decides.
 const DETACHED_WAIT: Duration = Duration::from_secs(30);
@@ -799,6 +802,7 @@ fn observe_interjection(method: &str, params: &Value, flight: &mut InFlight) {
         // Keep the exact marker, including its event metadata, so an older
         // disk branch cannot restore the invalidated cache.
         flight.interjection = SteerEvidence::Rewound(params.clone());
+        flight.transcript_check_at = None;
     }
 }
 
@@ -825,12 +829,24 @@ fn interjected_in_transcript(flight: &InFlight) -> bool {
         })
 }
 
-fn interjection_outcome(flight: &InFlight) -> Option<Outcome> {
-    flight.removed_at?;
-    let injected = match flight.interjection {
-        SteerEvidence::Unseen | SteerEvidence::Rewound(_) => interjected_in_transcript(flight),
-        SteerEvidence::Injected => true,
-    };
+fn interjection_outcome(flight: &mut InFlight, force_transcript: bool) -> Option<Outcome> {
+    // A detached request may have lost its removal response. Keep the
+    // original transcript recovery, but require the exact steer identity.
+    if flight.removed_at.is_none() && (flight.detached_at.is_none() || flight.request_id.is_some())
+    {
+        return None;
+    }
+    let injected =
+        if flight.removed_at.is_some() && matches!(flight.interjection, SteerEvidence::Injected) {
+            true
+        } else {
+            let now = Instant::now();
+            if !force_transcript && flight.transcript_check_at.is_some_and(|at| now < at) {
+                return None;
+            }
+            flight.transcript_check_at = Some(now + INTERJECTION_TRANSCRIPT_POLL);
+            interjected_in_transcript(flight)
+        };
     injected.then_some(Outcome::Delivered)
 }
 
@@ -1300,10 +1316,15 @@ pub(super) fn run(
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
+        let requeue_due = in_flight
+            .as_ref()
+            .is_some_and(|flight| flight.requeue_pending && Instant::now() >= requeue_at);
         if outcome.is_none()
-            && let Some(flight) = in_flight.as_ref()
+            && let Some(flight) = in_flight.as_mut()
         {
-            outcome = interjection_outcome(flight);
+            // The final check must not reuse a negative poll result before
+            // discarding the batch and submitting it again.
+            outcome = interjection_outcome(flight, requeue_due);
         }
         if outcome.is_none()
             && in_flight
@@ -1330,7 +1351,7 @@ pub(super) fn run(
         {
             let outcome = match outcome {
                 Outcome::Dropped(_)
-                    if interjection_outcome(&flight).is_some()
+                    if interjection_outcome(&mut flight, requeue_due).is_some()
                         || ran_in_transcript(&flight.transcript, &flight.text) =>
                 {
                     Outcome::Delivered
@@ -1364,7 +1385,9 @@ pub(super) fn run(
                 }
                 Outcome::Dropped(reason) => {
                     // Still unread; queue it again after a pause.
-                    requeue_at = Instant::now() + REQUEUE_DELAY;
+                    if !flight.requeue_pending {
+                        requeue_at = Instant::now() + REQUEUE_DELAY;
+                    }
                     if flight.removed_at.is_some() {
                         flight.requeue_pending = true;
                         in_flight = Some(flight);
@@ -1473,6 +1496,7 @@ pub(super) fn run(
                         ack: prepared.ack,
                         removed_at: None,
                         interjection: SteerEvidence::Unseen,
+                        transcript_check_at: None,
                         requeue_pending: false,
                         detached_at: None,
                     });
@@ -1505,39 +1529,53 @@ mod tests {
     #[test]
     #[serial_test::serial]
     fn steer_removed_batch_is_not_requeued_when_injection_arrives_late() {
-        exercise_removed_steer_fixture(true, None);
+        exercise_removed_steer_fixture(true, None, 0);
     }
 
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
     fn removed_without_injection_still_requeues_and_stays_unread() {
-        exercise_removed_steer_fixture(false, None);
+        exercise_removed_steer_fixture(false, None, 0);
     }
 
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
     fn rewind_preserving_earlier_steer_does_not_requeue() {
-        exercise_removed_steer_fixture(true, Some((1, 0)));
+        exercise_removed_steer_fixture(true, Some((1, 0)), 0);
     }
 
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
     fn rewind_before_disk_rechecks_preserved_steer() {
-        exercise_removed_steer_fixture(true, Some((1, 3)));
+        exercise_removed_steer_fixture(true, Some((1, 3)), 0);
     }
 
     #[cfg(unix)]
     #[test]
     #[serial_test::serial]
     fn rewind_before_disk_removing_steer_stays_unread() {
-        exercise_removed_steer_fixture(true, Some((0, 3)));
+        exercise_removed_steer_fixture(true, Some((0, 3)), 0);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn reconnect_with_lost_response_acknowledges_durable_steer() {
+        exercise_removed_steer_fixture(true, None, 1);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn reconnect_repeated_missing_snapshots_do_not_extend_retry_pause() {
+        exercise_removed_steer_fixture(false, None, 2);
     }
 
     #[cfg(unix)]
-    fn exercise_removed_steer_fixture(inject: bool, rewind: Option<(u64, u64)>) {
+    fn exercise_removed_steer_fixture(inject: bool, rewind: Option<(u64, u64)>, reconnect: u8) {
         // Exercise the production run/Client/ACK path against fixture ACP,
         // including an injection after the original two-second decision.
         let (_home_dir, _hcom, home, _env) = crate::hooks::test_helpers::isolated_test_env();
@@ -1567,6 +1605,7 @@ rewind=$5
 later=$6
 transcript=$7
 disk_delay=$8
+reconnect=$9
 while IFS= read -r line; do
     id=${line#*'"id":'}
     id=${id%%,*}
@@ -1574,12 +1613,28 @@ while IFS= read -r line; do
         *'"method":"initialize"'*)
             printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
         *'"method":"_x.ai/sessions/list"'*)
-            printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$id" "$roster" ;;
+            printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$id" "$roster"
+            if [ "$reconnect" != 0 ] && [ -f "$capture" ]; then
+                # Keep reporting the absent entry; each snapshot must not
+                # restart the retry pause. The hint keeps roster polls fast.
+                printf '%s\n' "$queue"
+                printf '{"method":"_x.ai/models/update","params":{}}\n'
+            fi ;;
         *'"method":"session/load"'*)
             printf '%s\n' "$queue"
             printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
         *'"method":"session/prompt"'*)
+            first=no
+            if [ ! -f "$capture" ]; then first=yes; fi
             printf '%s\n' "$line" >> "$capture"
+            if [ "$reconnect" != 0 ] && [ "$first" = yes ]; then
+                if [ "$injection" != '{}' ]; then printf '%s\n' "$injection" > "$transcript"; fi
+                if [ "$reconnect" = 2 ]; then
+                    # INFERRED response, not a captured production packet.
+                    printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
+                fi
+                exit 0
+            fi
             if [ "$rewind" != '{}' ] && [ "$sent" != yes ]; then
                 # Inject prompt 0, then rewind only the later prompt 1 (or
                 # delete prompt 0 for the negative). The response is inferred.
@@ -1593,7 +1648,7 @@ while IFS= read -r line; do
                 fi
                 sent=yes
             else
-                # INFERRED response, not a injection production packet.
+                # INFERRED response, not a captured production packet.
                 printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
             fi
             if [ "$rewind" = '{}' ] && [ "$sent" != yes ]; then
@@ -1653,6 +1708,7 @@ done
                 later.to_string(),
                 transcript.to_string_lossy().into_owned(),
                 rewind.map_or(0, |(_, delay)| delay).to_string(),
+                reconnect.to_string(),
             ],
             &[],
         )
@@ -1693,30 +1749,57 @@ done
         );
         let running = Arc::new(AtomicBool::new(true));
         let stop = Arc::clone(&running);
+        let delivered = inject && rewind.is_none_or(|(target, _)| target == 1);
         let check_stale_file = rewind.is_some_and(|(_, delay)| delay > 0);
         let test_db = dir.path().join("test.db");
+        let observed_capture = capture.clone();
         let observer = std::thread::spawn(move || {
-            std::thread::sleep(Duration::from_secs(2));
-            let early_cursor = if check_stale_file {
-                let conn = rusqlite::Connection::open_with_flags(
-                    test_db,
-                    rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
-                )
-                .unwrap();
-                Some(
-                    conn.query_row(
-                        "SELECT last_event_id FROM instances WHERE name='nova'",
-                        [],
-                        |row| row.get::<_, i64>(0),
-                    )
-                    .unwrap(),
-                )
-            } else {
-                None
-            };
-            std::thread::sleep(Duration::from_secs(12));
+            let deadline = Instant::now() + Duration::from_secs(22);
+            let mut first_prompt = None;
+            let mut early_cursor = None;
+            let mut covered_retry_window = false;
+            loop {
+                let prompts = std::fs::read_to_string(&observed_capture)
+                    .map_or(0, |content| content.lines().count());
+                if prompts > 0 {
+                    let first = *first_prompt.get_or_insert_with(Instant::now);
+                    if check_stale_file
+                        && early_cursor.is_none()
+                        && first.elapsed() >= Duration::from_secs(2)
+                    {
+                        let conn = rusqlite::Connection::open_with_flags(
+                            &test_db,
+                            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
+                        )
+                        .unwrap();
+                        early_cursor = Some(
+                            conn.query_row(
+                                "SELECT last_event_id FROM instances WHERE name='nova'",
+                                [],
+                                |row| row.get::<_, i64>(0),
+                            )
+                            .unwrap(),
+                        );
+                    }
+                    // Delivered cases must survive a complete retry period
+                    // measured from submission, not from thread startup.
+                    if !delivered && prompts >= 2 {
+                        break;
+                    }
+                    if first.elapsed()
+                        >= MERGE_EVIDENCE_WAIT + REQUEUE_DELAY + Duration::from_secs(2)
+                    {
+                        covered_retry_window = true;
+                        break;
+                    }
+                }
+                if Instant::now() >= deadline {
+                    break;
+                }
+                std::thread::sleep(Duration::from_millis(50));
+            }
             stop.store(false, Ordering::Release);
-            early_cursor
+            (early_cursor, covered_retry_window)
         });
         let state = DeliveryState {
             screen: Arc::new(RwLock::new(super::super::ScreenState {
@@ -1747,7 +1830,12 @@ done
             &mut super::super::host_label::HostLabel::resolve(),
             &mut LaunchOutcome::Ready,
         );
-        if let Some(early_cursor) = observer.join().unwrap() {
+        let (early_cursor, covered_retry_window) = observer.join().unwrap();
+        assert!(
+            !delivered || covered_retry_window,
+            "the deadline expired before a complete post-submission retry window"
+        );
+        if let Some(early_cursor) = early_cursor {
             assert_eq!(
                 early_cursor, 0,
                 "a live rewind must veto the not-yet-updated disk branch"
@@ -1763,7 +1851,6 @@ done
             packets.len(),
             db.get_cursor("nova")
         );
-        let delivered = inject && rewind.is_none_or(|(target, _)| target == 1);
         assert_eq!(packets.len(), if delivered { 1 } else { 2 });
         assert_eq!(packets[0]["params"]["_meta"]["sendNow"], false);
         assert_eq!(db.get_cursor("nova"), if delivered { 7 } else { 0 });
@@ -1790,6 +1877,7 @@ done
             },
             removed_at: removed.then(Instant::now),
             interjection: SteerEvidence::Unseen,
+            transcript_check_at: None,
             requeue_pending: false,
             detached_at: None,
         }
@@ -1817,7 +1905,7 @@ done
             if inject_first {
                 observe_interjection("session/update", &injection["params"], &mut flight);
                 assert_eq!(
-                    interjection_outcome(&flight),
+                    interjection_outcome(&mut flight, true),
                     None,
                     "injection alone cannot claim a batch"
                 );
@@ -1826,13 +1914,16 @@ done
             flight.removed_at = Some(Instant::now());
             if !inject_first {
                 assert_eq!(
-                    interjection_outcome(&flight),
+                    interjection_outcome(&mut flight, true),
                     None,
                     "manual queue removal stays unread"
                 );
                 observe_interjection("session/update", &injection["params"], &mut flight);
             }
-            assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+            assert_eq!(
+                interjection_outcome(&mut flight, true),
+                Some(Outcome::Delivered)
+            );
         }
     }
 
@@ -1860,7 +1951,7 @@ done
             *wrong.pointer_mut(pointer).unwrap() = replacement;
             let mut pending = steer_flight(true).0;
             observe_interjection("session/update", &wrong["params"], &mut pending);
-            assert_eq!(interjection_outcome(&pending), None, "{pointer}");
+            assert_eq!(interjection_outcome(&mut pending, true), None, "{pointer}");
         }
         let mut wrong_method = steer_flight(true).0;
         observe_interjection(
@@ -1868,7 +1959,91 @@ done
             &injection["params"],
             &mut wrong_method,
         );
-        assert_eq!(interjection_outcome(&wrong_method), None);
+        assert_eq!(interjection_outcome(&mut wrong_method, true), None);
+    }
+
+    #[test]
+    fn detached_lost_response_requires_strict_durable_injection() {
+        let dir = tempfile::tempdir().unwrap();
+        for invalid in [
+            None,
+            Some("/params/sessionId"),
+            Some("/params/update/content/_meta/displayText"),
+        ] {
+            let (mut flight, mut injection) = steer_flight(false);
+            flight.transcript = dir.path().join("updates.jsonl");
+            observe_interjection("session/update", &injection["params"], &mut flight);
+            assert_eq!(
+                interjection_outcome(&mut flight, true),
+                None,
+                "a connected request still needs its removal response"
+            );
+            flight.detached_at = Some(Instant::now());
+            assert_eq!(
+                interjection_outcome(&mut flight, true),
+                None,
+                "the old request id has not been detached yet"
+            );
+            flight.request_id = None;
+            assert_eq!(
+                interjection_outcome(&mut flight, true),
+                None,
+                "a cached live injection alone cannot replace the lost response"
+            );
+            if let Some(pointer) = invalid {
+                *injection.pointer_mut(pointer).unwrap() = json!("wrong identity");
+            }
+            std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
+            assert_eq!(
+                interjection_outcome(&mut flight, true),
+                invalid.is_none().then_some(Outcome::Delivered)
+            );
+            std::fs::remove_file(&flight.transcript).unwrap();
+        }
+    }
+
+    #[test]
+    fn rewind_invalidates_the_transcript_poll_deadline() {
+        let dir = tempfile::tempdir().unwrap();
+        let (mut flight, injection) = steer_flight(true);
+        flight.transcript = dir.path().join("updates.jsonl");
+        assert_eq!(interjection_outcome(&mut flight, false), None);
+        let later = json!({"method":"session/update", "params": {
+            "sessionId": flight.session, "update": {"sessionUpdate":"user_message_chunk",
+                "content":{"type":"text","text":"later"}, "_meta":{"promptIndex":1}}}});
+        let marker = json!({"method":"_x.ai/session/update", "params": {
+            "sessionId":flight.session, "update":{"sessionUpdate":"rewind_marker",
+                "target_prompt_index":1}, "_meta":{"eventId":"new-rewind"}}});
+        std::fs::write(
+            &flight.transcript,
+            format!("{injection}\n{later}\n{marker}\n"),
+        )
+        .unwrap();
+        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
+        assert_eq!(
+            interjection_outcome(&mut flight, false),
+            Some(Outcome::Delivered),
+            "a new rewind must invalidate a cached miss immediately"
+        );
+    }
+
+    #[test]
+    fn interjection_transcript_checks_are_throttled_between_polls() {
+        let dir = tempfile::tempdir().unwrap();
+        let (mut flight, injection) = steer_flight(true);
+        flight.transcript = dir.path().join("updates.jsonl");
+        assert_eq!(interjection_outcome(&mut flight, false), None);
+        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
+        assert_eq!(
+            interjection_outcome(&mut flight, false),
+            None,
+            "a second poll must not rescan the transcript within one second"
+        );
+        assert_eq!(
+            interjection_outcome(&mut flight, true),
+            Some(Outcome::Delivered),
+            "the last check before retry must bypass a cached miss"
+        );
     }
 
     #[test]
@@ -1878,9 +2053,12 @@ done
         flight.transcript = dir.path().join("updates.jsonl");
         flight.removed_at = Some(Instant::now() - MERGE_EVIDENCE_WAIT);
         flight.requeue_pending = true;
-        assert_eq!(interjection_outcome(&flight), None);
+        assert_eq!(interjection_outcome(&mut flight, true), None);
         std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
-        assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+        assert_eq!(
+            interjection_outcome(&mut flight, true),
+            Some(Outcome::Delivered)
+        );
         // Generic substring proof must not bypass the strict steer check.
         assert!(!ran_in_transcript(&flight.transcript, &flight.text));
         let rewind = json!({"method": "_x.ai/session/update", "params": {
@@ -1889,14 +2067,14 @@ done
         observe_interjection("session/update", &injection["params"], &mut flight);
         observe_interjection("x.ai/session/update", &rewind["params"], &mut flight);
         assert_eq!(
-            interjection_outcome(&flight),
+            interjection_outcome(&mut flight, true),
             None,
             "a live rewind vetoes a stale file"
         );
         std::fs::write(&flight.transcript, format!("{injection}\n{rewind}\n")).unwrap();
         flight.interjection = SteerEvidence::Unseen;
         assert_eq!(
-            interjection_outcome(&flight),
+            interjection_outcome(&mut flight, true),
             None,
             "rewound injection is not delivery evidence"
         );
@@ -1912,7 +2090,7 @@ done
             let mut wrong = injection.clone();
             *wrong.pointer_mut(pointer).unwrap() = replacement;
             std::fs::write(&flight.transcript, format!("{wrong}\n")).unwrap();
-            assert_eq!(interjection_outcome(&flight), None, "{pointer}");
+            assert_eq!(interjection_outcome(&mut flight, true), None, "{pointer}");
             assert!(!ran_in_transcript(&flight.transcript, &flight.text));
         }
     }
@@ -1938,7 +2116,7 @@ done
         )
         .unwrap();
         assert_eq!(
-            interjection_outcome(&flight),
+            interjection_outcome(&mut flight, true),
             None,
             "an older rewind cannot unlock the stale file"
         );
@@ -1947,19 +2125,22 @@ done
             format!("{injection}\n{later}\n{marker}\n"),
         )
         .unwrap();
-        assert_eq!(interjection_outcome(&flight), Some(Outcome::Delivered));
+        assert_eq!(
+            interjection_outcome(&mut flight, true),
+            Some(Outcome::Delivered)
+        );
         let mut deleting = marker.clone();
         deleting["params"]["update"]["target_prompt_index"] = json!(0);
         deleting["params"]["_meta"]["eventId"] = json!("rewind-delete");
         observe_interjection("x.ai/session/update", &deleting["params"], &mut flight);
-        assert_eq!(interjection_outcome(&flight), None);
+        assert_eq!(interjection_outcome(&mut flight, true), None);
         std::fs::write(
             &flight.transcript,
             format!("{injection}\n{later}\n{marker}\n{deleting}\n"),
         )
         .unwrap();
         assert_eq!(
-            interjection_outcome(&flight),
+            interjection_outcome(&mut flight, true),
             None,
             "a removed injection stays unread"
         );

From bb428c6bd270cfba582cb70fa41c3f9587439ef1 Mon Sep 17 00:00:00 2001
From: zane668 <152239216+zane668@users.noreply.github.com>
Date: Mon, 19 Oct 2026 00:39:05 -0400
Subject: [PATCH 3/4] fix(grok): resolve retry expiry with final delivery
 evidence

Keep detached batches with a lost prompt response through the same fixed retry pause so late durable injections can be acknowledged. Resolve expiry with a forced strict evidence check after slower transcript work and apply that outcome without a later independent clock decision.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/delivery/grok.rs | 125 ++++++++++++++++++++++++++++++++-----------
 1 file changed, 93 insertions(+), 32 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 11cc2aa7..8fe60118 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -850,6 +850,23 @@ fn interjection_outcome(flight: &mut InFlight, force_transcript: bool) -> Option
     injected.then_some(Outcome::Delivered)
 }
 
+/// The final transition out of the retry pause, evaluated at its deadline.
+fn expired_requeue_outcome(
+    flight: &mut InFlight,
+    deadline: Instant,
+    now: Instant,
+) -> Option<Outcome> {
+    if !flight.requeue_pending || now < deadline {
+        return None;
+    }
+    // Decide expiry and consume its last evidence in one transition. A
+    // deadline crossed after an earlier poll cannot reuse that cached miss.
+    Some(
+        interjection_outcome(flight, true)
+            .unwrap_or_else(|| Outcome::Dropped("removed from Grok's queue".into())),
+    )
+}
+
 /// What a queue snapshot proves about our prompt. Only identity counts: our
 /// prompt id running, or our exact text among the running combined texts
 /// (Grok merges queued plain prompts into the front one).
@@ -1316,15 +1333,10 @@ pub(super) fn run(
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
-        let requeue_due = in_flight
-            .as_ref()
-            .is_some_and(|flight| flight.requeue_pending && Instant::now() >= requeue_at);
         if outcome.is_none()
             && let Some(flight) = in_flight.as_mut()
         {
-            // The final check must not reuse a negative poll result before
-            // discarding the batch and submitting it again.
-            outcome = interjection_outcome(flight, requeue_due);
+            outcome = interjection_outcome(flight, false);
         }
         if outcome.is_none()
             && in_flight
@@ -1346,18 +1358,25 @@ pub(super) fn run(
                 "no queue evidence after reconnecting".into(),
             ));
         }
+        if matches!(outcome, Some(Outcome::Dropped(_)))
+            && let Some(flight) = in_flight.as_mut()
+            && (interjection_outcome(flight, false).is_some()
+                || ran_in_transcript(&flight.transcript, &flight.text))
+        {
+            outcome = Some(Outcome::Delivered);
+        }
+        // This is the only expiry decision. Do it after potentially slow
+        // transcript work, then apply its outcome without a second clock read.
+        let expired = in_flight
+            .as_mut()
+            .and_then(|flight| expired_requeue_outcome(flight, requeue_at, Instant::now()));
+        let requeue_expired = expired.is_some();
+        if outcome != Some(Outcome::Delivered) && expired.is_some() {
+            outcome = expired;
+        }
         if let Some(outcome) = outcome
             && let Some(mut flight) = in_flight.take()
         {
-            let outcome = match outcome {
-                Outcome::Dropped(_)
-                    if interjection_outcome(&mut flight, requeue_due).is_some()
-                        || ran_in_transcript(&flight.transcript, &flight.text) =>
-                {
-                    Outcome::Delivered
-                }
-                other => other,
-            };
             match outcome {
                 Outcome::Delivered if flight.ack.instance_name != *current_name => {
                     // Queued under an identity a session switch retired: its
@@ -1388,7 +1407,9 @@ pub(super) fn run(
                     if !flight.requeue_pending {
                         requeue_at = Instant::now() + REQUEUE_DELAY;
                     }
-                    if flight.removed_at.is_some() {
+                    if !requeue_expired
+                        && (flight.removed_at.is_some() || flight.detached_at.is_some())
+                    {
                         flight.requeue_pending = true;
                         in_flight = Some(flight);
                     } else {
@@ -1401,20 +1422,6 @@ pub(super) fn run(
                 }
             }
         }
-        if let Some(flight) = in_flight.as_ref()
-            && flight.requeue_pending
-            && Instant::now() >= requeue_at
-        {
-            log_warn(
-                "native",
-                "grok.acp.requeue",
-                &format!(
-                    "prompt={} removed from Grok's queue; kept unread",
-                    flight.prompt_id
-                ),
-            );
-            in_flight = None;
-        }
         let Some(conn) = client.as_mut() else {
             continue;
         };
@@ -1574,6 +1581,13 @@ mod tests {
         exercise_removed_steer_fixture(false, None, 2);
     }
 
+    #[cfg(unix)]
+    #[test]
+    #[serial_test::serial]
+    fn reconnect_lost_response_tracks_injection_written_during_retry_pause() {
+        exercise_removed_steer_fixture(true, None, 3);
+    }
+
     #[cfg(unix)]
     fn exercise_removed_steer_fixture(inject: bool, rewind: Option<(u64, u64)>, reconnect: u8) {
         // Exercise the production run/Client/ACK path against fixture ACP,
@@ -1622,13 +1636,20 @@ while IFS= read -r line; do
             fi ;;
         *'"method":"session/load"'*)
             printf '%s\n' "$queue"
-            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
+            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id"
+            if [ "$reconnect" = 3 ] && [ -f "$capture" ] && [ "$sent" != yes ]; then
+                # The missing snapshot reaches run before the injection is
+                # durable. There is no session/prompt response to rescue it.
+                sleep 3
+                printf '%s\n' "$injection" > "$transcript"
+                sent=yes
+            fi ;;
         *'"method":"session/prompt"'*)
             first=no
             if [ ! -f "$capture" ]; then first=yes; fi
             printf '%s\n' "$line" >> "$capture"
             if [ "$reconnect" != 0 ] && [ "$first" = yes ]; then
-                if [ "$injection" != '{}' ]; then printf '%s\n' "$injection" > "$transcript"; fi
+                if [ "$injection" != '{}' ] && [ "$reconnect" != 3 ]; then printf '%s\n' "$injection" > "$transcript"; fi
                 if [ "$reconnect" = 2 ]; then
                     # INFERRED response, not a captured production packet.
                     printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
@@ -1962,6 +1983,46 @@ done
         assert_eq!(interjection_outcome(&mut wrong_method, true), None);
     }
 
+    #[test]
+    fn retry_expiry_forces_strict_evidence_after_an_earlier_cached_miss() {
+        let dir = tempfile::tempdir().unwrap();
+        let (mut flight, injection) = steer_flight(true);
+        flight.transcript = dir.path().join("updates.jsonl");
+        flight.requeue_pending = true;
+        assert_eq!(interjection_outcome(&mut flight, false), None);
+        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
+        let deadline = Instant::now() + Duration::from_secs(60);
+        assert_eq!(
+            expired_requeue_outcome(&mut flight, deadline, deadline - Duration::from_nanos(1)),
+            None
+        );
+        // A scheduler pause crosses the deadline after the earlier poll.
+        // Explicit times make the boundary deterministic; no sleep race.
+        assert_eq!(
+            expired_requeue_outcome(&mut flight, deadline, deadline),
+            Some(Outcome::Delivered)
+        );
+        let marker = json!({"method":"_x.ai/session/update", "params": {
+            "sessionId":flight.session, "update":{"sessionUpdate":"rewind_marker",
+                "target_prompt_index":0}, "_meta":{"eventId":"expiry-rewind"}}});
+        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
+        assert!(
+            matches!(
+                expired_requeue_outcome(&mut flight, deadline, deadline),
+                Some(Outcome::Dropped(_))
+            ),
+            "expiry must not accept the stale branch before rewind is durable"
+        );
+        std::fs::write(&flight.transcript, format!("{injection}\n{marker}\n")).unwrap();
+        assert!(
+            matches!(
+                expired_requeue_outcome(&mut flight, deadline, deadline),
+                Some(Outcome::Dropped(_))
+            ),
+            "expiry must not accept an injection removed by rewind"
+        );
+    }
+
     #[test]
     fn detached_lost_response_requires_strict_durable_injection() {
         let dir = tempfile::tempdir().unwrap();

From 4f49f4783aa9253cb583eaa37349877202341e12 Mon Sep 17 00:00:00 2001
From: aannoo <aannoo@users.noreply.github.com>
Date: Tue, 20 Oct 2026 04:01:58 +0100
Subject: [PATCH 4/4] Grok: acknowledge a batch steer injected into the running
 turn

With follow-up behavior Steer, Grok moves a queued prompt into the
running turn when both have the same owner, so an hcom batch is only
steered into a turn hcom started (an agent still working on earlier
mail), never into the user's own turn. Grok announces it with
x.ai/session/interjection under our promptId before answering
removedFromQueue. Treat that as delivery instead of matching
interjection text in the transcript.
---
 src/delivery/grok.rs | 869 +++----------------------------------------
 1 file changed, 52 insertions(+), 817 deletions(-)

diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs
index 8fe60118..42bfe794 100644
--- a/src/delivery/grok.rs
+++ b/src/delivery/grok.rs
@@ -23,8 +23,9 @@
 //!
 //! Grok uses our `promptId` as the queue entry id. The batch is acknowledged
 //! when Grok reports that entry running (`x.ai/queue/changed`), i.e. once it
-//! is part of the model's turn, or when the prompt request returns a result.
-//! A batch that never started (removed from the queue, transport lost) stays
+//! is part of the model's turn, when Grok injects it into the running turn
+//! (`x.ai/session/interjection`, follow-up behavior Steer), or when the prompt
+//! request returns a result. A batch that never started (removed from the queue, transport lost) stays
 //! unread and is queued again later: at-least-once, never silently dropped.
 
 use std::collections::{HashMap, HashSet};
@@ -739,17 +740,9 @@ struct InFlight {
     /// The session's `updates.jsonl`, checked before re-queueing.
     transcript: std::path::PathBuf,
     /// When Grok answered `removedFromQueue`: either removed, or merged into
-    /// the prompt ahead of it (which then runs its text), or injected by
-    /// steer. The confirming queue/update evidence may arrive later.
+    /// the prompt ahead of it (which then runs its text). Only a combined-turn
+    /// snapshot tells them apart, and it may arrive after the response.
     removed_at: Option<Instant>,
-    /// Live steer evidence, pending its removal response. A rewind invalidates
-    /// the cache until its marker is durable and the live branch is rechecked.
-    interjection: SteerEvidence,
-    /// Limit full transcript scans while the live evidence is pending.
-    transcript_check_at: Option<Instant>,
-    /// Keep tracking this batch during the existing re-queue pause: an
-    /// injection or its transcript can arrive after the merge wait.
-    requeue_pending: bool,
     /// When the ACP connection that queued it dropped. Grok keeps a queued
     /// prompt when its client disconnects, so the batch is followed through
     /// the reconnected client's queue snapshots, not treated as dropped.
@@ -758,7 +751,6 @@ struct InFlight {
 
 /// How long to wait after `removedFromQueue` for the snapshot showing a merge.
 const MERGE_EVIDENCE_WAIT: Duration = Duration::from_secs(2);
-const INTERJECTION_TRANSCRIPT_POLL: Duration = Duration::from_secs(1);
 /// How long a batch whose connection dropped waits for a queue snapshot
 /// before the transcript decides.
 const DETACHED_WAIT: Duration = Duration::from_secs(30);
@@ -771,102 +763,6 @@ enum Outcome {
     Dropped(String),
 }
 
-enum SteerEvidence {
-    Unseen,
-    Injected,
-    Rewound(Value),
-}
-
-/// Steer runs a queued prompt inside the current turn. Its display text is
-/// the original prompt, while content.text wraps it in Grok's instructions.
-fn is_interjection(params: &Value, flight: &InFlight) -> bool {
-    let update = &params["update"];
-    params["sessionId"].as_str() == Some(flight.session.as_str())
-        && update["sessionUpdate"] == "user_message_chunk"
-        && update["_meta"]["interjection"] == true
-        && update["content"]["type"] == "text"
-        && update["content"]["_meta"]["displayText"].as_str() == Some(flight.text.as_str())
-        && update["content"]["text"]
-            .as_str()
-            .is_some_and(|text| text.contains(&flight.text))
-}
-
-fn observe_interjection(method: &str, params: &Value, flight: &mut InFlight) {
-    if method == "session/update" && is_interjection(params, flight) {
-        flight.interjection = SteerEvidence::Injected;
-    } else if method == "x.ai/session/update"
-        && params["sessionId"].as_str() == Some(flight.session.as_str())
-        && params["update"]["sessionUpdate"] == "rewind_marker"
-    {
-        // A rewind can remove the injected turn before its file is flushed.
-        // Keep the exact marker, including its event metadata, so an older
-        // disk branch cannot restore the invalidated cache.
-        flight.interjection = SteerEvidence::Rewound(params.clone());
-        flight.transcript_check_at = None;
-    }
-}
-
-fn interjected_in_transcript(flight: &InFlight) -> bool {
-    let Ok(content) = std::fs::read_to_string(&flight.transcript) else {
-        return false;
-    };
-    if let SteerEvidence::Rewound(marker) = &flight.interjection
-        && !content.lines().any(|line| {
-            serde_json::from_str::<Value>(line).is_ok_and(|value| {
-                value["method"] == "_x.ai/session/update" && value["params"] == *marker
-            })
-        })
-    {
-        return false;
-    }
-    crate::transcript::grok::live_lines(&content)
-        .into_iter()
-        .rev()
-        .any(|line| {
-            serde_json::from_str::<Value>(line).is_ok_and(|value| {
-                value["method"] == "session/update" && is_interjection(&value["params"], flight)
-            })
-        })
-}
-
-fn interjection_outcome(flight: &mut InFlight, force_transcript: bool) -> Option<Outcome> {
-    // A detached request may have lost its removal response. Keep the
-    // original transcript recovery, but require the exact steer identity.
-    if flight.removed_at.is_none() && (flight.detached_at.is_none() || flight.request_id.is_some())
-    {
-        return None;
-    }
-    let injected =
-        if flight.removed_at.is_some() && matches!(flight.interjection, SteerEvidence::Injected) {
-            true
-        } else {
-            let now = Instant::now();
-            if !force_transcript && flight.transcript_check_at.is_some_and(|at| now < at) {
-                return None;
-            }
-            flight.transcript_check_at = Some(now + INTERJECTION_TRANSCRIPT_POLL);
-            interjected_in_transcript(flight)
-        };
-    injected.then_some(Outcome::Delivered)
-}
-
-/// The final transition out of the retry pause, evaluated at its deadline.
-fn expired_requeue_outcome(
-    flight: &mut InFlight,
-    deadline: Instant,
-    now: Instant,
-) -> Option<Outcome> {
-    if !flight.requeue_pending || now < deadline {
-        return None;
-    }
-    // Decide expiry and consume its last evidence in one transition. A
-    // deadline crossed after an earlier poll cannot reuse that cached miss.
-    Some(
-        interjection_outcome(flight, true)
-            .unwrap_or_else(|| Outcome::Dropped("removed from Grok's queue".into())),
-    )
-}
-
 /// What a queue snapshot proves about our prompt. Only identity counts: our
 /// prompt id running, or our exact text among the running combined texts
 /// (Grok merges queued plain prompts into the front one).
@@ -892,6 +788,15 @@ fn queue_outcome(params: &Value, flight: &InFlight) -> Option<Outcome> {
     None
 }
 
+/// Whether an `x.ai/session/interjection` is our prompt. With follow-up
+/// behavior Steer, Grok moves a queued prompt into a running turn with the
+/// same owner, so only into a turn hcom started: it announces it under our
+/// prompt id, then answers the prompt `removedFromQueue`.
+fn interjected(params: &Value, flight: &InFlight) -> bool {
+    params["sessionId"].as_str() == Some(flight.session.as_str())
+        && params["interjectionId"].as_str() == Some(flight.prompt_id.as_str())
+}
+
 /// Whether the session transcript shows the prompt text as sent: durable
 /// proof it ran when the live evidence was lost (bridge disconnect, a merge
 /// snapshot that never arrived).
@@ -908,9 +813,6 @@ fn ran_in_transcript(path: &std::path::Path, text: &str) -> bool {
                 && serde_json::from_str::<Value>(line).is_ok_and(|value| {
                     let update = &value["params"]["update"];
                     update["sessionUpdate"] == "user_message_chunk"
-                        // A steer wrapper needs the stricter display-text
-                        // identity check above, not just a body substring.
-                        && update["_meta"]["interjection"] != true
                         && update["content"]["text"]
                             .as_str()
                             .is_some_and(|sent| sent.contains(text))
@@ -1223,15 +1125,17 @@ pub(super) fn run(
                     }
                 }
                 Event::Notification { method, params } => {
-                    if let Some(flight) = in_flight.as_mut() {
-                        observe_interjection(&method, &params, flight);
-                    }
                     if method == "x.ai/queue/changed"
                         && let Some(flight) = in_flight.as_ref()
                         && params["sessionId"].as_str() == Some(flight.session.as_str())
                     {
                         outcome = queue_outcome(&params, flight);
                     }
+                    if method == "x.ai/session/interjection"
+                        && in_flight.as_ref().is_some_and(|f| interjected(&params, f))
+                    {
+                        outcome = Some(Outcome::Delivered);
+                    }
                     effects.extend(tracker.on_notification(&method, &params));
                 }
                 Event::Request { id, method, params } => {
@@ -1333,15 +1237,9 @@ pub(super) fn run(
             }
             log_warn("native", "grok.acp.disconnected", &error);
         }
-        if outcome.is_none()
-            && let Some(flight) = in_flight.as_mut()
-        {
-            outcome = interjection_outcome(flight, false);
-        }
         if outcome.is_none()
             && in_flight
                 .as_ref()
-                .filter(|flight| !flight.requeue_pending)
                 .and_then(|flight| flight.removed_at)
                 .is_some_and(|at| at.elapsed() >= MERGE_EVIDENCE_WAIT)
         {
@@ -1350,7 +1248,6 @@ pub(super) fn run(
         if outcome.is_none()
             && in_flight
                 .as_ref()
-                .filter(|flight| !flight.requeue_pending)
                 .and_then(|flight| flight.detached_at)
                 .is_some_and(|at| at.elapsed() >= DETACHED_WAIT)
         {
@@ -1358,25 +1255,15 @@ pub(super) fn run(
                 "no queue evidence after reconnecting".into(),
             ));
         }
-        if matches!(outcome, Some(Outcome::Dropped(_)))
-            && let Some(flight) = in_flight.as_mut()
-            && (interjection_outcome(flight, false).is_some()
-                || ran_in_transcript(&flight.transcript, &flight.text))
-        {
-            outcome = Some(Outcome::Delivered);
-        }
-        // This is the only expiry decision. Do it after potentially slow
-        // transcript work, then apply its outcome without a second clock read.
-        let expired = in_flight
-            .as_mut()
-            .and_then(|flight| expired_requeue_outcome(flight, requeue_at, Instant::now()));
-        let requeue_expired = expired.is_some();
-        if outcome != Some(Outcome::Delivered) && expired.is_some() {
-            outcome = expired;
-        }
         if let Some(outcome) = outcome
-            && let Some(mut flight) = in_flight.take()
+            && let Some(flight) = in_flight.take()
         {
+            let outcome = match outcome {
+                Outcome::Dropped(_) if ran_in_transcript(&flight.transcript, &flight.text) => {
+                    Outcome::Delivered
+                }
+                other => other,
+            };
             match outcome {
                 Outcome::Delivered if flight.ack.instance_name != *current_name => {
                     // Queued under an identity a session switch retired: its
@@ -1404,21 +1291,12 @@ pub(super) fn run(
                 }
                 Outcome::Dropped(reason) => {
                     // Still unread; queue it again after a pause.
-                    if !flight.requeue_pending {
-                        requeue_at = Instant::now() + REQUEUE_DELAY;
-                    }
-                    if !requeue_expired
-                        && (flight.removed_at.is_some() || flight.detached_at.is_some())
-                    {
-                        flight.requeue_pending = true;
-                        in_flight = Some(flight);
-                    } else {
-                        log_warn(
-                            "native",
-                            "grok.acp.requeue",
-                            &format!("prompt={} {reason}; kept unread", flight.prompt_id),
-                        );
-                    }
+                    requeue_at = Instant::now() + REQUEUE_DELAY;
+                    log_warn(
+                        "native",
+                        "grok.acp.requeue",
+                        &format!("prompt={} {reason}; kept unread", flight.prompt_id),
+                    );
                 }
             }
         }
@@ -1502,9 +1380,6 @@ pub(super) fn run(
                         text: prepared.formatted,
                         ack: prepared.ack,
                         removed_at: None,
-                        interjection: SteerEvidence::Unseen,
-                        transcript_check_at: None,
-                        requeue_pending: false,
                         detached_at: None,
                     });
                 }
@@ -1527,360 +1402,6 @@ mod tests {
 
     const BATCH: &str = "<hcom>[request #7] luna → nova: hi</hcom>";
 
-    // Synthetic packet shaped like an observed Grok steer injection. The
-    // original session/prompt response was not captured; removal responses
-    // below are inferred fixtures, not original-response replays.
-    const STEER_INJECTION_FIXTURE: &str = r#"{"timestamp":1000,"method":"session/update","params":{"sessionId":"fixture-session","update":{"sessionUpdate":"user_message_chunk","content":{"type":"text","text":"The user sent a message while you were working:\n<user_query>\n<hcom>[request #7] luna → nova: Please continue the current task and acknowledge this message.</hcom>\n</user_query>\nIf the user is asking for a response, address the user first. After replying, complete any unfinished tasks from previous turns.","_meta":{"displayText":"<hcom>[request #7] luna → nova: Please continue the current task and acknowledge this message.</hcom>"}},"_meta":{"modelId":"grok-4.7-build-fast","interjection":true}},"_meta":{"eventId":"fixture-session-injection","agentTimestampMs":1000000}}}"#;
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn steer_removed_batch_is_not_requeued_when_injection_arrives_late() {
-        exercise_removed_steer_fixture(true, None, 0);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn removed_without_injection_still_requeues_and_stays_unread() {
-        exercise_removed_steer_fixture(false, None, 0);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn rewind_preserving_earlier_steer_does_not_requeue() {
-        exercise_removed_steer_fixture(true, Some((1, 0)), 0);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn rewind_before_disk_rechecks_preserved_steer() {
-        exercise_removed_steer_fixture(true, Some((1, 3)), 0);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn rewind_before_disk_removing_steer_stays_unread() {
-        exercise_removed_steer_fixture(true, Some((0, 3)), 0);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn reconnect_with_lost_response_acknowledges_durable_steer() {
-        exercise_removed_steer_fixture(true, None, 1);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn reconnect_repeated_missing_snapshots_do_not_extend_retry_pause() {
-        exercise_removed_steer_fixture(false, None, 2);
-    }
-
-    #[cfg(unix)]
-    #[test]
-    #[serial_test::serial]
-    fn reconnect_lost_response_tracks_injection_written_during_retry_pause() {
-        exercise_removed_steer_fixture(true, None, 3);
-    }
-
-    #[cfg(unix)]
-    fn exercise_removed_steer_fixture(inject: bool, rewind: Option<(u64, u64)>, reconnect: u8) {
-        // Exercise the production run/Client/ACK path against fixture ACP,
-        // including an injection after the original two-second decision.
-        let (_home_dir, _hcom, home, _env) = crate::hooks::test_helpers::isolated_test_env();
-        // EnvGuard does not manage GROK_HOME. Restore this override before
-        // releasing its environment lock, including during an assertion unwind.
-        struct GrokHome(Option<std::ffi::OsString>);
-        impl Drop for GrokHome {
-            fn drop(&mut self) {
-                unsafe {
-                    match &self.0 {
-                        Some(value) => std::env::set_var("GROK_HOME", value),
-                        None => std::env::remove_var("GROK_HOME"),
-                    }
-                }
-            }
-        }
-        let _grok_home = GrokHome(std::env::var_os("GROK_HOME"));
-        unsafe { std::env::set_var("GROK_HOME", home.join(".grok")) };
-        let dir = tempfile::tempdir().unwrap();
-        let capture = dir.path().join("prompts.jsonl");
-        let script = dir.path().join("acp.sh");
-        std::fs::write(&script, r#"capture=$1
-roster=$2
-queue=$3
-injection=$4
-rewind=$5
-later=$6
-transcript=$7
-disk_delay=$8
-reconnect=$9
-while IFS= read -r line; do
-    id=${line#*'"id":'}
-    id=${id%%,*}
-    case "$line" in
-        *'"method":"initialize"'*)
-            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id" ;;
-        *'"method":"_x.ai/sessions/list"'*)
-            printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$id" "$roster"
-            if [ "$reconnect" != 0 ] && [ -f "$capture" ]; then
-                # Keep reporting the absent entry; each snapshot must not
-                # restart the retry pause. The hint keeps roster polls fast.
-                printf '%s\n' "$queue"
-                printf '{"method":"_x.ai/models/update","params":{}}\n'
-            fi ;;
-        *'"method":"session/load"'*)
-            printf '%s\n' "$queue"
-            printf '{"jsonrpc":"2.0","id":%s,"result":{}}\n' "$id"
-            if [ "$reconnect" = 3 ] && [ -f "$capture" ] && [ "$sent" != yes ]; then
-                # The missing snapshot reaches run before the injection is
-                # durable. There is no session/prompt response to rescue it.
-                sleep 3
-                printf '%s\n' "$injection" > "$transcript"
-                sent=yes
-            fi ;;
-        *'"method":"session/prompt"'*)
-            first=no
-            if [ ! -f "$capture" ]; then first=yes; fi
-            printf '%s\n' "$line" >> "$capture"
-            if [ "$reconnect" != 0 ] && [ "$first" = yes ]; then
-                if [ "$injection" != '{}' ] && [ "$reconnect" != 3 ]; then printf '%s\n' "$injection" > "$transcript"; fi
-                if [ "$reconnect" = 2 ]; then
-                    # INFERRED response, not a captured production packet.
-                    printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
-                fi
-                exit 0
-            fi
-            if [ "$rewind" != '{}' ] && [ "$sent" != yes ]; then
-                # Inject prompt 0, then rewind only the later prompt 1 (or
-                # delete prompt 0 for the negative). The response is inferred.
-                printf '%s\n%s\n' "$injection" "$later" > "$transcript"
-                printf '%s\n%s\n%s\n' "$injection" "$later" "$rewind"
-                if [ "$disk_delay" = 0 ]; then printf '%s\n' "$rewind" >> "$transcript"; fi
-                printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
-                if [ "$disk_delay" != 0 ]; then
-                    sleep "$disk_delay"
-                    printf '%s\n' "$rewind" >> "$transcript"
-                fi
-                sent=yes
-            else
-                # INFERRED response, not a captured production packet.
-                printf '{"jsonrpc":"2.0","id":%s,"result":{"_meta":{"completionKind":"removedFromQueue"}}}\n' "$id"
-            fi
-            if [ "$rewind" = '{}' ] && [ "$sent" != yes ]; then
-                sleep 3
-                printf '%s\n' "$injection"
-                sent=yes
-            fi ;;
-    esac
-done
-"#).unwrap();
-        let injected: Value = serde_json::from_str(STEER_INJECTION_FIXTURE).unwrap();
-        let session = injected["params"]["sessionId"].as_str().unwrap();
-        let exact = injected["params"]["update"]["content"]["_meta"]["displayText"]
-            .as_str()
-            .unwrap();
-        let text = exact
-            .strip_prefix("<hcom>[request #7] luna → nova: ")
-            .unwrap()
-            .strip_suffix("</hcom>")
-            .unwrap();
-        let cwd = dir.path().to_str().unwrap();
-        let transcript = crate::transcript::grok::session_transcript_path(cwd, session);
-        std::fs::create_dir_all(transcript.parent().unwrap()).unwrap();
-        // Synthetic later-turn/rewind and response fixtures. The injection
-        // has the shape observed in a real steer delivery.
-        let marker = rewind.map(|(target, _)| {
-            json!({
-            "method": "_x.ai/session/update", "params": {
-                "sessionId": session, "update": {"sessionUpdate": "rewind_marker",
-                    "target_prompt_index": target},
-                "_meta": {"eventId": "fixture-rewind-500"}}})
-        });
-        let later = json!({"method": "session/update", "params": {
-            "sessionId": session, "update": {"sessionUpdate": "user_message_chunk",
-                "content": {"type": "text", "text": "later prompt"},
-                "_meta": {"promptIndex": 1}}}});
-        let roster = json!({"sessions": [{"sessionId": session, "cwd": cwd,
-            "activity": "idle", "resident": true, "lastActivityAt": 1}]});
-        let queue = json!({"method": "_x.ai/queue/changed", "params": {
-            "sessionId": session, "runningPromptId": "user-turn", "entries": []}});
-        let launch = Launch::new(
-            "/bin/sh",
-            &[
-                script.to_string_lossy().into_owned(),
-                capture.to_string_lossy().into_owned(),
-                roster.to_string(),
-                queue.to_string(),
-                if inject {
-                    STEER_INJECTION_FIXTURE
-                } else {
-                    "{}"
-                }
-                .to_string(),
-                marker
-                    .as_ref()
-                    .map_or_else(|| "{}".into(), Value::to_string),
-                later.to_string(),
-                transcript.to_string_lossy().into_owned(),
-                rewind.map_or(0, |(_, delay)| delay).to_string(),
-                reconnect.to_string(),
-            ],
-            &[],
-        )
-        .unwrap();
-        let mut db = HcomDb::open_raw(&dir.path().join("test.db")).unwrap();
-        db.init_db().unwrap();
-        db.conn()
-            .execute(
-                "INSERT INTO instances (name, tool, session_id, directory, status, created_at)
-             VALUES ('nova', 'grok', ?1, ?2, 'listening', 0)",
-                rusqlite::params![session, cwd],
-            )
-            .unwrap();
-        db.set_session_binding(session, "nova").unwrap();
-        db.set_process_binding("fixture-process", session, "nova")
-            .unwrap();
-        let event = db
-            .insert_event_row(
-                "message",
-                "luna",
-                &json!({
-            "from": "luna", "text": text, "intent": "request", "scope": "mentions",
-            "mentions": ["nova"], "delivered_to": ["nova"]}),
-                None,
-            )
-            .unwrap();
-        db.conn()
-            .execute("UPDATE events SET id=7 WHERE id=?1", [event])
-            .unwrap();
-        // Consume the first-receipt tip, so the next production batch is
-        // byte-for-byte identical to the synthetic displayText.
-        common::prepare_pending_messages(&db, "nova").unwrap();
-        assert_eq!(
-            common::prepare_pending_messages(&db, "nova")
-                .unwrap()
-                .formatted,
-            exact
-        );
-        let running = Arc::new(AtomicBool::new(true));
-        let stop = Arc::clone(&running);
-        let delivered = inject && rewind.is_none_or(|(target, _)| target == 1);
-        let check_stale_file = rewind.is_some_and(|(_, delay)| delay > 0);
-        let test_db = dir.path().join("test.db");
-        let observed_capture = capture.clone();
-        let observer = std::thread::spawn(move || {
-            let deadline = Instant::now() + Duration::from_secs(22);
-            let mut first_prompt = None;
-            let mut early_cursor = None;
-            let mut covered_retry_window = false;
-            loop {
-                let prompts = std::fs::read_to_string(&observed_capture)
-                    .map_or(0, |content| content.lines().count());
-                if prompts > 0 {
-                    let first = *first_prompt.get_or_insert_with(Instant::now);
-                    if check_stale_file
-                        && early_cursor.is_none()
-                        && first.elapsed() >= Duration::from_secs(2)
-                    {
-                        let conn = rusqlite::Connection::open_with_flags(
-                            &test_db,
-                            rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY,
-                        )
-                        .unwrap();
-                        early_cursor = Some(
-                            conn.query_row(
-                                "SELECT last_event_id FROM instances WHERE name='nova'",
-                                [],
-                                |row| row.get::<_, i64>(0),
-                            )
-                            .unwrap(),
-                        );
-                    }
-                    // Delivered cases must survive a complete retry period
-                    // measured from submission, not from thread startup.
-                    if !delivered && prompts >= 2 {
-                        break;
-                    }
-                    if first.elapsed()
-                        >= MERGE_EVIDENCE_WAIT + REQUEUE_DELAY + Duration::from_secs(2)
-                    {
-                        covered_retry_window = true;
-                        break;
-                    }
-                }
-                if Instant::now() >= deadline {
-                    break;
-                }
-                std::thread::sleep(Duration::from_millis(50));
-            }
-            stop.store(false, Ordering::Release);
-            (early_cursor, covered_retry_window)
-        });
-        let state = DeliveryState {
-            screen: Arc::new(RwLock::new(super::super::ScreenState {
-                ready: true,
-                input_text: Some("PRIVATE_DRAFT".into()),
-                prompt_empty: false,
-                ..Default::default()
-            })),
-            grok_acp: None,
-            launch_phase_active: Arc::new(AtomicBool::new(false)),
-            inject_port: 0,
-            user_activity_cooldown_ms: 0,
-        };
-        let notify = NotifyServer::new().unwrap();
-        let mut name = "nova".to_string();
-        run(
-            &launch,
-            &running,
-            &mut db,
-            &notify,
-            &state,
-            "fixture-process",
-            &mut name,
-            &ToolConfig::for_tool(crate::tool::Tool::Grok),
-            &None,
-            &None,
-            &None,
-            &mut super::super::host_label::HostLabel::resolve(),
-            &mut LaunchOutcome::Ready,
-        );
-        let (early_cursor, covered_retry_window) = observer.join().unwrap();
-        assert!(
-            !delivered || covered_retry_window,
-            "the deadline expired before a complete post-submission retry window"
-        );
-        if let Some(early_cursor) = early_cursor {
-            assert_eq!(
-                early_cursor, 0,
-                "a live rewind must veto the not-yet-updated disk branch"
-            );
-        }
-        let packets: Vec<Value> = std::fs::read_to_string(&capture)
-            .unwrap()
-            .lines()
-            .map(|line| serde_json::from_str(line).unwrap())
-            .collect();
-        eprintln!(
-            "late steer: prompts={} cursor={}",
-            packets.len(),
-            db.get_cursor("nova")
-        );
-        assert_eq!(packets.len(), if delivered { 1 } else { 2 });
-        assert_eq!(packets[0]["params"]["_meta"]["sendNow"], false);
-        assert_eq!(db.get_cursor("nova"), if delivered { 7 } else { 0 });
-        assert_eq!(
-            state.screen.read().unwrap().input_text.as_deref(),
-            Some("PRIVATE_DRAFT")
-        );
-    }
-
     fn flight(removed: bool) -> InFlight {
         InFlight {
             request_id: Some(4),
@@ -1897,316 +1418,10 @@ done
                 mark_announced: false,
             },
             removed_at: removed.then(Instant::now),
-            interjection: SteerEvidence::Unseen,
-            transcript_check_at: None,
-            requeue_pending: false,
             detached_at: None,
         }
     }
 
-    fn steer_flight(removed: bool) -> (InFlight, Value) {
-        let injection: Value = serde_json::from_str(STEER_INJECTION_FIXTURE).unwrap();
-        let mut flight = flight(removed);
-        flight.session = injection["params"]["sessionId"].as_str().unwrap().into();
-        flight.text = injection["params"]["update"]["content"]["_meta"]["displayText"]
-            .as_str()
-            .unwrap()
-            .into();
-        (flight, injection)
-    }
-
-    #[test]
-    fn steer_injection_and_inferred_removal_can_arrive_in_either_order() {
-        // Removal is inferred from the observed requeue branch; its original
-        // response was not captured. The notification shape is synthetic.
-        let inferred = json!({"id": 4, "result": {
-            "_meta": {"completionKind": "removedFromQueue"}}});
-        for inject_first in [false, true] {
-            let (mut flight, injection) = steer_flight(false);
-            if inject_first {
-                observe_interjection("session/update", &injection["params"], &mut flight);
-                assert_eq!(
-                    interjection_outcome(&mut flight, true),
-                    None,
-                    "injection alone cannot claim a batch"
-                );
-            }
-            assert_eq!(response_outcome(&inferred), None);
-            flight.removed_at = Some(Instant::now());
-            if !inject_first {
-                assert_eq!(
-                    interjection_outcome(&mut flight, true),
-                    None,
-                    "manual queue removal stays unread"
-                );
-                observe_interjection("session/update", &injection["params"], &mut flight);
-            }
-            assert_eq!(
-                interjection_outcome(&mut flight, true),
-                Some(Outcome::Delivered)
-            );
-        }
-    }
-
-    #[test]
-    fn steer_evidence_requires_the_exact_batch_session_and_user_injection() {
-        let (flight, injection) = steer_flight(true);
-        assert!(is_interjection(&injection["params"], &flight));
-        let variants = [
-            ("/params/sessionId", json!("another-session")),
-            ("/params/update/sessionUpdate", json!("agent_message_chunk")),
-            ("/params/update/_meta/interjection", json!(false)),
-            ("/params/update/content/type", json!("image")),
-            ("/params/update/content/text", json!("different body")),
-            (
-                "/params/update/content/_meta/displayText",
-                json!(format!("{} ", flight.text)),
-            ),
-            (
-                "/params/update/content/_meta/displayText",
-                json!("partial batch text"),
-            ),
-        ];
-        for (pointer, replacement) in variants {
-            let mut wrong = injection.clone();
-            *wrong.pointer_mut(pointer).unwrap() = replacement;
-            let mut pending = steer_flight(true).0;
-            observe_interjection("session/update", &wrong["params"], &mut pending);
-            assert_eq!(interjection_outcome(&mut pending, true), None, "{pointer}");
-        }
-        let mut wrong_method = steer_flight(true).0;
-        observe_interjection(
-            "x.ai/session_notification",
-            &injection["params"],
-            &mut wrong_method,
-        );
-        assert_eq!(interjection_outcome(&mut wrong_method, true), None);
-    }
-
-    #[test]
-    fn retry_expiry_forces_strict_evidence_after_an_earlier_cached_miss() {
-        let dir = tempfile::tempdir().unwrap();
-        let (mut flight, injection) = steer_flight(true);
-        flight.transcript = dir.path().join("updates.jsonl");
-        flight.requeue_pending = true;
-        assert_eq!(interjection_outcome(&mut flight, false), None);
-        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
-        let deadline = Instant::now() + Duration::from_secs(60);
-        assert_eq!(
-            expired_requeue_outcome(&mut flight, deadline, deadline - Duration::from_nanos(1)),
-            None
-        );
-        // A scheduler pause crosses the deadline after the earlier poll.
-        // Explicit times make the boundary deterministic; no sleep race.
-        assert_eq!(
-            expired_requeue_outcome(&mut flight, deadline, deadline),
-            Some(Outcome::Delivered)
-        );
-        let marker = json!({"method":"_x.ai/session/update", "params": {
-            "sessionId":flight.session, "update":{"sessionUpdate":"rewind_marker",
-                "target_prompt_index":0}, "_meta":{"eventId":"expiry-rewind"}}});
-        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
-        assert!(
-            matches!(
-                expired_requeue_outcome(&mut flight, deadline, deadline),
-                Some(Outcome::Dropped(_))
-            ),
-            "expiry must not accept the stale branch before rewind is durable"
-        );
-        std::fs::write(&flight.transcript, format!("{injection}\n{marker}\n")).unwrap();
-        assert!(
-            matches!(
-                expired_requeue_outcome(&mut flight, deadline, deadline),
-                Some(Outcome::Dropped(_))
-            ),
-            "expiry must not accept an injection removed by rewind"
-        );
-    }
-
-    #[test]
-    fn detached_lost_response_requires_strict_durable_injection() {
-        let dir = tempfile::tempdir().unwrap();
-        for invalid in [
-            None,
-            Some("/params/sessionId"),
-            Some("/params/update/content/_meta/displayText"),
-        ] {
-            let (mut flight, mut injection) = steer_flight(false);
-            flight.transcript = dir.path().join("updates.jsonl");
-            observe_interjection("session/update", &injection["params"], &mut flight);
-            assert_eq!(
-                interjection_outcome(&mut flight, true),
-                None,
-                "a connected request still needs its removal response"
-            );
-            flight.detached_at = Some(Instant::now());
-            assert_eq!(
-                interjection_outcome(&mut flight, true),
-                None,
-                "the old request id has not been detached yet"
-            );
-            flight.request_id = None;
-            assert_eq!(
-                interjection_outcome(&mut flight, true),
-                None,
-                "a cached live injection alone cannot replace the lost response"
-            );
-            if let Some(pointer) = invalid {
-                *injection.pointer_mut(pointer).unwrap() = json!("wrong identity");
-            }
-            std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
-            assert_eq!(
-                interjection_outcome(&mut flight, true),
-                invalid.is_none().then_some(Outcome::Delivered)
-            );
-            std::fs::remove_file(&flight.transcript).unwrap();
-        }
-    }
-
-    #[test]
-    fn rewind_invalidates_the_transcript_poll_deadline() {
-        let dir = tempfile::tempdir().unwrap();
-        let (mut flight, injection) = steer_flight(true);
-        flight.transcript = dir.path().join("updates.jsonl");
-        assert_eq!(interjection_outcome(&mut flight, false), None);
-        let later = json!({"method":"session/update", "params": {
-            "sessionId": flight.session, "update": {"sessionUpdate":"user_message_chunk",
-                "content":{"type":"text","text":"later"}, "_meta":{"promptIndex":1}}}});
-        let marker = json!({"method":"_x.ai/session/update", "params": {
-            "sessionId":flight.session, "update":{"sessionUpdate":"rewind_marker",
-                "target_prompt_index":1}, "_meta":{"eventId":"new-rewind"}}});
-        std::fs::write(
-            &flight.transcript,
-            format!("{injection}\n{later}\n{marker}\n"),
-        )
-        .unwrap();
-        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
-        assert_eq!(
-            interjection_outcome(&mut flight, false),
-            Some(Outcome::Delivered),
-            "a new rewind must invalidate a cached miss immediately"
-        );
-    }
-
-    #[test]
-    fn interjection_transcript_checks_are_throttled_between_polls() {
-        let dir = tempfile::tempdir().unwrap();
-        let (mut flight, injection) = steer_flight(true);
-        flight.transcript = dir.path().join("updates.jsonl");
-        assert_eq!(interjection_outcome(&mut flight, false), None);
-        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
-        assert_eq!(
-            interjection_outcome(&mut flight, false),
-            None,
-            "a second poll must not rescan the transcript within one second"
-        );
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            Some(Outcome::Delivered),
-            "the last check before retry must bypass a cached miss"
-        );
-    }
-
-    #[test]
-    fn late_steer_transcript_evidence_is_consumed_only_on_its_live_branch() {
-        let dir = tempfile::tempdir().unwrap();
-        let (mut flight, injection) = steer_flight(true);
-        flight.transcript = dir.path().join("updates.jsonl");
-        flight.removed_at = Some(Instant::now() - MERGE_EVIDENCE_WAIT);
-        flight.requeue_pending = true;
-        assert_eq!(interjection_outcome(&mut flight, true), None);
-        std::fs::write(&flight.transcript, format!("{injection}\n")).unwrap();
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            Some(Outcome::Delivered)
-        );
-        // Generic substring proof must not bypass the strict steer check.
-        assert!(!ran_in_transcript(&flight.transcript, &flight.text));
-        let rewind = json!({"method": "_x.ai/session/update", "params": {
-            "sessionId": flight.session, "update": {
-                "sessionUpdate": "rewind_marker", "target_prompt_index": 0}}});
-        observe_interjection("session/update", &injection["params"], &mut flight);
-        observe_interjection("x.ai/session/update", &rewind["params"], &mut flight);
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            None,
-            "a live rewind vetoes a stale file"
-        );
-        std::fs::write(&flight.transcript, format!("{injection}\n{rewind}\n")).unwrap();
-        flight.interjection = SteerEvidence::Unseen;
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            None,
-            "rewound injection is not delivery evidence"
-        );
-        for (pointer, replacement) in [
-            ("/params/sessionId", json!("another-session")),
-            ("/params/update/content/text", json!("different body")),
-            (
-                "/params/update/content/_meta/displayText",
-                json!(format!("{} ", flight.text)),
-            ),
-            ("/params/update/sessionUpdate", json!("agent_message_chunk")),
-        ] {
-            let mut wrong = injection.clone();
-            *wrong.pointer_mut(pointer).unwrap() = replacement;
-            std::fs::write(&flight.transcript, format!("{wrong}\n")).unwrap();
-            assert_eq!(interjection_outcome(&mut flight, true), None, "{pointer}");
-            assert!(!ran_in_transcript(&flight.transcript, &flight.text));
-        }
-    }
-
-    #[test]
-    fn matching_rewind_must_be_durable_before_rechecking_the_surviving_branch() {
-        let dir = tempfile::tempdir().unwrap();
-        let (mut flight, injection) = steer_flight(true);
-        flight.transcript = dir.path().join("updates.jsonl");
-        let later = json!({"method": "session/update", "params": {
-            "sessionId": flight.session, "update": {"sessionUpdate": "user_message_chunk",
-                "content": {"type": "text", "text": "later"}, "_meta": {"promptIndex": 1}}}});
-        let marker = json!({"method": "_x.ai/session/update", "params": {
-            "sessionId": flight.session, "update": {"sessionUpdate": "rewind_marker",
-                "target_prompt_index": 1}, "_meta": {"eventId": "rewind-current"}}});
-        let mut older = marker.clone();
-        older["params"]["_meta"]["eventId"] = json!("rewind-older");
-        observe_interjection("session/update", &injection["params"], &mut flight);
-        observe_interjection("x.ai/session/update", &marker["params"], &mut flight);
-        std::fs::write(
-            &flight.transcript,
-            format!("{injection}\n{later}\n{older}\n"),
-        )
-        .unwrap();
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            None,
-            "an older rewind cannot unlock the stale file"
-        );
-        std::fs::write(
-            &flight.transcript,
-            format!("{injection}\n{later}\n{marker}\n"),
-        )
-        .unwrap();
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            Some(Outcome::Delivered)
-        );
-        let mut deleting = marker.clone();
-        deleting["params"]["update"]["target_prompt_index"] = json!(0);
-        deleting["params"]["_meta"]["eventId"] = json!("rewind-delete");
-        observe_interjection("x.ai/session/update", &deleting["params"], &mut flight);
-        assert_eq!(interjection_outcome(&mut flight, true), None);
-        std::fs::write(
-            &flight.transcript,
-            format!("{injection}\n{later}\n{marker}\n{deleting}\n"),
-        )
-        .unwrap();
-        assert_eq!(
-            interjection_outcome(&mut flight, true),
-            None,
-            "a removed injection stays unread"
-        );
-    }
-
     #[test]
     fn running_prompt_is_delivered() {
         let params = json!({"sessionId": "s", "runningPromptId": "hcom-1", "entries": []});
@@ -2248,6 +1463,20 @@ done
         );
     }
 
+    #[test]
+    fn steer_injection_of_our_prompt_is_delivered() {
+        let ours = json!({"sessionId": "s", "text": BATCH, "interjectionId": "hcom-1"});
+        assert!(interjected(&ours, &flight(false)));
+        assert!(interjected(&ours, &flight(true)));
+        for other in [
+            json!({"sessionId": "s", "text": BATCH, "interjectionId": "user-1"}),
+            json!({"sessionId": "s", "text": BATCH}),
+            json!({"sessionId": "other", "text": BATCH, "interjectionId": "hcom-1"}),
+        ] {
+            assert!(!interjected(&other, &flight(true)), "{other}");
+        }
+    }
+
     #[test]
     fn detached_batch_follows_the_reconnected_queue() {
         let mut detached = flight(false);
@@ -2278,6 +1507,12 @@ done
         // Merged into a combined turn.
         std::fs::write(&path, format!("{}\n", chunk(&format!("fix it\n\n{BATCH}")))).unwrap();
         assert!(ran_in_transcript(&path, BATCH));
+        // Injected by steer: the persisted chunk wraps the text.
+        let steer = format!(
+            "The user sent a message while you were working:\n<user_query>\n{BATCH}\n</user_query>\n..."
+        );
+        std::fs::write(&path, format!("{}\n", chunk(&steer))).unwrap();
+        assert!(ran_in_transcript(&path, BATCH));
         assert!(!ran_in_transcript(&dir.path().join("missing"), BATCH));
     }