feat(core): soft interrupts — steer a running turn without cancelling it (#377, 1/2)

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- Soft interrupts, core (#377): a bounded queue on the session's `CancelToken` that the turn loop drains at the safe point between a settled tool and the next provider call, committing each message as an attributed user-role entry (`interrupt: true`). An `urgent` interrupt (or any interrupt, for a final answer) withdraws an output that has not taken effect, so no tool call is left without its result, and records the drop durably. Host entry points (serve/MCP) follow separately.
 - `tole doctor` (#378): tiered provider diagnostics — `offline` (wiring), `catalog` (live `/models`, model must be served) and `full` (chat, streaming and a real tool-call loop) — each blocker carries its fix command, and keys / URL credentials are never printed. Exits non-zero when the requested tier is not reached; `--tier full` is the release-validation gate.
 - Per-user-turn token stats (#379): `tole status` lists input / output / cache-read / cache-write / reasoning tokens per turn and the status JSON (`GET /sessions/{id}/status`, `tole_session_status`) gains an additive `usage_report.turns` array. Derived from the durable usage ledger, so old sessions read the same way and nothing is stored twice; a value no step reported is `n/a` / `null`, never 0, and no elapsed time is invented.
 

crates/tole-cli/src/main.rs

@@ -2111,7 +2111,10 @@ fn status_command(sessions_dir: &Path, id: &str) -> Result<()> {
     let turns: usize = storage
         .entries()
         .iter()
-        .filter(|e| e.payload.get("role") == Some(&serde_json::json!("user")))
+        .filter(|e| {
+            e.payload.get("role") == Some(&serde_json::json!("user"))
+                && e.payload.get("interrupt") != Some(&serde_json::json!(true))
+        })
         .count();
     println!("turns:   {turns} (user messages)");
     println!(
@@ -2193,7 +2196,10 @@ fn sessions_command(sessions_dir: &Path) -> Result<()> {
         let turns: usize = s
             .entries()
             .iter()
-            .filter(|e| e.payload.get("role") == Some(&serde_json::json!("user")))
+            .filter(|e| {
+                e.payload.get("role") == Some(&serde_json::json!("user"))
+                    && e.payload.get("interrupt") != Some(&serde_json::json!(true))
+            })
             .count();
         rows.push((
             mtime,

crates/tole-cli/src/usage_report.rs

@@ -207,11 +207,12 @@ fn add(acc: &mut Option<u64>, x: Option<u64>) {
 }
 
 /// A turn opens at a user message, except the stop-gate's own feedback
-/// message (the turn continues after it).
+/// and an owner's soft interrupt (#377): the turn continues after them.
 fn opens_turn(e: &Entry) -> bool {
     e.kind.as_str() == "message"
         && e.payload["role"] == json!("user")
         && e.payload["stop_gate_feedback"] != json!(true)
+        && e.payload["interrupt"] != json!(true)
 }
 
 /// Fold the usage ledger into per-user-turn stats. A usage row belongs to
@@ -658,6 +659,22 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_soft_interrupt_does_not_open_a_turn() {
+        let entries = vec![
+            user(1),
+            entry(
+                2,
+                "message",
+                json!({"role": "user", "text": "steer", "interrupt": true}),
+            ),
+            assistant(3),
+        ];
+        let t = turn_stats(&entries, &[rec_at("e_3", json!({"prompt_tokens": 4}))]);
+        assert_eq!(t.len(), 1);
+        assert_eq!(t[0].input_tokens, Some(4));
+    }
+
     #[test]
     fn unknown_stays_unknown_never_zero() {
         let entries = vec![user(1), assistant(2), user(3), assistant(4)];

crates/tole-core/src/cancel.rs

@@ -21,13 +21,22 @@ use std::sync::Arc;
 #[derive(Clone, Default)]
 pub struct CancelToken {
     flag: Arc<AtomicBool>,
+    /// Soft-interrupt queue (#377): the same per-session turn handle that
+    /// carries the cancel flag also carries steering messages. Never
+    /// cleared by `reset`: an interrupt waits for the next safe point.
+    interrupts: crate::interrupt::InterruptQueue,
 }
 
 impl CancelToken {
     pub fn new() -> Self {
         Self::default()
     }
 
+    /// The session's soft-interrupt queue (#377).
+    pub fn interrupts(&self) -> &crate::interrupt::InterruptQueue {
+        &self.interrupts
+    }
+
     /// Set by the transport when the client cancels the session's
     /// current turn. Idempotent; safe from any thread.
     pub fn cancel(&self) {

crates/tole-core/src/interrupt.rs

@@ -0,0 +1,150 @@
+//! Soft interrupts: steer a running turn without cancelling it (#377).
+//!
+//! The owner (a second client of a `serve`/MCP session) pushes a message
+//! while a turn is running. The turn loop drains the queue at its safe
+//! point — after the current tool has settled and before the next provider
+//! call — and commits each message as an attributed user-role entry, so
+//! nothing done so far is lost (unlike a cancel).
+//!
+//! `urgent` additionally withdraws a tool call the model has just asked for
+//! but that has not started: the call is dropped BEFORE its intent is
+//! written, so no `tool_use` is ever left without a `tool_result`, and a
+//! durable error entry records the drop (no call vanishes silently).
+//!
+//! The queue rides inside [`crate::cancel::CancelToken`], the per-session
+//! turn handle every host already shares; a default token has an empty
+//! queue nothing pushes to, so existing callers are unchanged.
+
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+
+/// Most interrupts that may wait for a safe point.
+pub const MAX_PENDING: usize = 16;
+/// Largest accepted interrupt text, in bytes.
+pub const MAX_TEXT_BYTES: usize = 8 * 1024;
+
+/// One owner message injected mid-turn.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Interrupt {
+    pub text: String,
+    /// Also withdraw a tool call that has not started yet.
+    pub urgent: bool,
+}
+
+/// Why a push was refused.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PushError {
+    Empty,
+    TooLong,
+    Full,
+}
+
+impl std::fmt::Display for PushError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            PushError::Empty => f.write_str("interrupt text is empty"),
+            PushError::TooLong => write!(f, "interrupt text exceeds {MAX_TEXT_BYTES} bytes"),
+            PushError::Full => write!(f, "{MAX_PENDING} interrupts are already pending"),
+        }
+    }
+}
+
+/// Shared, bounded queue of pending interrupts. `Clone` shares the queue.
+#[derive(Clone, Default)]
+pub struct InterruptQueue {
+    q: Arc<Mutex<VecDeque<Interrupt>>>,
+}
+
+impl InterruptQueue {
+    fn lock(&self) -> std::sync::MutexGuard<'_, VecDeque<Interrupt>> {
+        // A panicking pusher must not wedge steering for the session.
+        self.q.lock().unwrap_or_else(|p| p.into_inner())
+    }
+
+    /// Queue a message. Bounded: blank, oversized, or beyond
+    /// [`MAX_PENDING`] pending are refused, never truncated.
+    pub fn push(&self, text: &str, urgent: bool) -> Result<(), PushError> {
+        let text = text.trim();
+        if text.is_empty() {
+            return Err(PushError::Empty);
+        }
+        if text.len() > MAX_TEXT_BYTES {
+            return Err(PushError::TooLong);
+        }
+        let mut q = self.lock();
+        if q.len() >= MAX_PENDING {
+            return Err(PushError::Full);
+        }
+        q.push_back(Interrupt {
+            text: text.to_string(),
+            urgent,
+        });
+        Ok(())
+    }
+
+    /// Take everything pending, oldest first.
+    pub fn drain(&self) -> Vec<Interrupt> {
+        self.lock().drain(..).collect()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.lock().is_empty()
+    }
+
+    /// Is an urgent interrupt waiting?
+    pub fn has_urgent(&self) -> bool {
+        self.lock().iter().any(|i| i.urgent)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn drains_in_order_and_empties() {
+        let q = InterruptQueue::default();
+        q.push("first", false).unwrap();
+        q.push("  second  ", true).unwrap();
+        assert!(q.has_urgent());
+        let d = q.drain();
+        assert_eq!(
+            d,
+            vec![
+                Interrupt {
+                    text: "first".into(),
+                    urgent: false
+                },
+                Interrupt {
+                    text: "second".into(),
+                    urgent: true
+                },
+            ]
+        );
+        assert!(q.is_empty() && !q.has_urgent());
+        assert!(q.drain().is_empty());
+    }
+
+    #[test]
+    fn clones_share_the_queue() {
+        let a = InterruptQueue::default();
+        let b = a.clone();
+        a.push("x", false).unwrap();
+        assert_eq!(b.drain().len(), 1);
+    }
+
+    #[test]
+    fn push_is_bounded_never_truncated() {
+        let q = InterruptQueue::default();
+        assert_eq!(q.push("   ", false), Err(PushError::Empty));
+        assert_eq!(
+            q.push(&"x".repeat(MAX_TEXT_BYTES + 1), false),
+            Err(PushError::TooLong)
+        );
+        for _ in 0..MAX_PENDING {
+            q.push("m", false).unwrap();
+        }
+        assert_eq!(q.push("m", false), Err(PushError::Full));
+        assert_eq!(q.drain().len(), MAX_PENDING);
+    }
+}

crates/tole-core/src/lib.rs

@@ -41,6 +41,7 @@ pub mod gh;
 #[cfg(feature = "shell-tools")]
 pub mod git;
 pub mod gitea;
+pub mod interrupt;
 pub mod systemone;
 pub mod todo;
 pub mod web;

crates/tole-core/src/turn.rs

@@ -482,6 +482,9 @@ fn drive(
         if cancel.is_cancelled() {
             return settle_cancelled(s);
         }
+        // Soft-interrupt safe point (#377): the previous tool has settled
+        // and the next provider call has not started.
+        deliver_interrupts(s, cancel)?;
         // No clone: complete() borrows the storage slice. (The old
         // to_vec() allocated the whole transcript every provider step.)
         let next = match p.complete(s.entries()) {
@@ -563,6 +566,33 @@ fn drive(
         if cancel.is_cancelled() {
             return settle_cancelled(s);
         }
+        // Soft interrupt that landed while `complete()` was in flight
+        // (#377). Like the cancel above, the output is a pure read with no
+        // pending intent, so dropping it is crash-safe — and because the
+        // intent is never written, no `tool_use` can be left without its
+        // `tool_result`. A final answer is superseded by ANY pending
+        // interrupt (the owner is steering it); a not-yet-started tool call
+        // only by an URGENT one. Each drop is recorded durably, and the
+        // next loop pass delivers the message at the safe point.
+        let superseded = match &next {
+            ProviderOutput::Final { .. } => !cancel.interrupts().is_empty(),
+            ProviderOutput::ToolCall { .. } | ProviderOutput::InvalidToolArgs { .. } => {
+                cancel.interrupts().has_urgent()
+            }
+        };
+        if superseded {
+            let what = match &next {
+                ProviderOutput::Final { .. } => "final answer".to_string(),
+                ProviderOutput::ToolCall { tool, .. }
+                | ProviderOutput::InvalidToolArgs { tool, .. } => format!("tool call {tool:?}"),
+            };
+            append_turn_error(
+                s,
+                "superseded by owner interrupt",
+                &format!("the model's {what} was withdrawn before it took effect"),
+            )?;
+            continue;
+        }
         // Issue #196 phase 2: surface the model's reasoning for this
         // step (when the provider supplies one) BEFORE the step's
         // effect lands — thought first, action after.
@@ -599,6 +629,7 @@ fn drive(
                             e.kind.as_str() == "message"
                                 && e.payload["role"] == json!("user")
                                 && e.payload["stop_gate_feedback"] != json!(true)
+                                && e.payload["interrupt"] != json!(true)
                         })
                         .unwrap_or(0);
                     let tools_seen: Vec<(String, Risk)> = s
@@ -853,6 +884,31 @@ fn drive(
 /// Appends a durable ERROR entry under the turn's user message. No
 /// transition: the machine stays parked in Planning for the host to
 /// resolve (resume/finish) — the record exists so replay can see why.
+/// Commit every pending soft interrupt as an attributed user-role entry
+/// (#377). The marker fields keep it from opening a new turn (like the
+/// stop-gate feedback) and let hosts and readers tell an owner's steering
+/// from the original prompt; the text carries the attribution too, since
+/// that is all the provider sees.
+fn deliver_interrupts(s: &mut dyn Storage, cancel: &CancelToken) -> Result<(), StorageError> {
+    for i in cancel.interrupts().drain() {
+        let tag = if i.urgent {
+            "owner interrupt · urgent"
+        } else {
+            "owner interrupt"
+        };
+        s.commit(Commit::new().entry(NewEntry::root(
+            EntryType::new(EntryType::MESSAGE),
+            json!({
+                "role": "user",
+                "text": format!("[{tag} — injected mid-turn, not part of the original request] {}", i.text),
+                "interrupt": true,
+                "urgent": i.urgent,
+            }),
+        )))?;
+    }
+    Ok(())
+}
+
 fn append_turn_error(s: &mut dyn Storage, error: &str, detail: &str) -> Result<(), StorageError> {
     let parent = s
         .entries()