diff --git a/src/delivery/grok.rs b/src/delivery/grok.rs index 277be5a9..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}; @@ -787,6 +788,15 @@ fn queue_outcome(params: &Value, flight: &InFlight) -> Option { 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). @@ -1121,6 +1131,11 @@ pub(super) fn run( { outcome = queue_outcome(¶ms, flight); } + if method == "x.ai/session/interjection" + && in_flight.as_ref().is_some_and(|f| interjected(¶ms, f)) + { + outcome = Some(Outcome::Delivered); + } effects.extend(tracker.on_notification(&method, ¶ms)); } Event::Request { id, method, params } => { @@ -1448,6 +1463,20 @@ mod tests { ); } + #[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); @@ -1478,6 +1507,12 @@ mod tests { // 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\n{BATCH}\n\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)); }