feat(cli): soft interrupt entry points — serve and MCP (#377, 2/2)

CHANGELOG.md

@@ -8,7 +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.
+- Soft interrupts (#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: `POST /sessions/{id}/interrupt` (`tole serve`) and the `tole_session_interrupt` MCP tool (`409`/error when no turn is running).
 - `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.
 

README.md

@@ -59,9 +59,13 @@ tole is the agent harness of the CodeCora ecosystem — the hands:
   turns, and poll status without SSH-ing into the box. Binds 127.0.0.1 by
   default. `POST /sessions/{id}/cancel` (#178, REST face) stops an
   in-flight turn — it settles `stopReason: "cancelled"`.
+  `POST /sessions/{id}/interrupt` (#377) steers the running turn instead,
+  without losing its work: the `{"text", "urgent"}` message is injected at
+  the turn's next safe point as an attributed entry (`409` when no turn
+  is running).
   `--transport mcp` (issue #137) serves **multi-session MCP over
   Streamable HTTP** instead: one authenticated MCP connection addresses N
-  durable tole sessions (`tole_session_new/prompt/cancel/status/list` +
+  durable tole sessions (`tole_session_new/prompt/cancel/interrupt/status/list` +
   the registry tools routed to each session's workspace jail).
 
 Every ecosystem integration is probe-first: a missing binary degrades to a

crates/tole-cli/src/serve.rs

@@ -8,6 +8,8 @@
 //! - `GET  /sessions`              → list session ids + busy flags
 //! - `GET  /sessions/{id}`         → session status
 //! - `POST /sessions/{id}/prompt`  → run one turn (`{"text": "..."}`)
+//! - `POST /sessions/{id}/interrupt` → steer the running turn without
+//!   cancelling it (`{"text": "...", "urgent": false}`, #377)
 //!
 //! v1 scope: REST only (MCP-over-HTTP is the follow-up; the engine calls
 //! are identical). Auth is a static bearer token (env `TOLE_SERVE_TOKEN`
@@ -729,6 +731,41 @@ fn route(state: &State, method: &str, path: &str, body: &str) -> (u16, serde_jso
                         (404, json!({"error": "unknown session"}))
                     }
                 }
+                ("POST", id, Some("interrupt")) => {
+                    // Soft interrupt (#377): steer the in-flight turn
+                    // without cancelling it. 409 when no turn is running.
+                    let Ok(req) = serde_json::from_str::<serde_json::Value>(body) else {
+                        return (400, json!({"error": "invalid JSON body"}));
+                    };
+                    let Some(text) = req.get("text").and_then(|v| v.as_str()) else {
+                        return (400, json!({"error": "missing text"}));
+                    };
+                    let urgent = req.get("urgent").and_then(|v| v.as_bool()).unwrap_or(false);
+                    let st = {
+                        let sessions = lock_sessions(&state.sessions);
+                        sessions.map.get(id).cloned()
+                    };
+                    let Some(st) = st else {
+                        return (404, json!({"error": "unknown session"}));
+                    };
+                    match tole_cli::session_host::interrupt_session(&st, text, urgent) {
+                        Ok(()) => {
+                            eprintln!(
+                                "tole: session/{id} interrupt queued (serve, urgent={urgent})"
+                            );
+                            (
+                                202,
+                                json!({"queued": true, "urgent": urgent, "sessionId": id}),
+                            )
+                        }
+                        Err(tole_cli::session_host::InterruptRefusal::NotBusy) => {
+                            (409, json!({"error": "no turn is running in this session"}))
+                        }
+                        Err(tole_cli::session_host::InterruptRefusal::Invalid(e)) => {
+                            (400, json!({"error": e}))
+                        }
+                    }
+                }
                 _ => (404, json!({"error": "not found"})),
             }
         }

crates/tole-cli/src/session_host.rs

@@ -39,6 +39,34 @@ pub struct SessionState {
     pub cancel: tole_core::cancel::CancelToken,
 }
 
+/// Why a soft interrupt was not queued (#377).
+#[derive(Debug, PartialEq, Eq)]
+pub enum InterruptRefusal {
+    /// No turn is running: there is nothing to steer, and a stale message
+    /// must not wait around to hijack the next, unrelated turn.
+    NotBusy,
+    /// Blank, oversized, or the queue is full (see `tole_core::interrupt`).
+    Invalid(String),
+}
+
+/// Queue a soft interrupt for the session's in-flight turn (#377). Shared
+/// by the REST and MCP faces. The turn loop delivers it at its next safe
+/// point; a turn that ends first leaves it queued for the next turn's
+/// first safe point (the check below is a best-effort guard, not a lock).
+pub fn interrupt_session(
+    st: &SessionState,
+    text: &str,
+    urgent: bool,
+) -> Result<(), InterruptRefusal> {
+    if !*st.busy.lock().expect("busy lock") {
+        return Err(InterruptRefusal::NotBusy);
+    }
+    st.cancel
+        .interrupts()
+        .push(text, urgent)
+        .map_err(|e| InterruptRefusal::Invalid(e.to_string()))
+}
+
 /// Marks a session busy for its whole lifetime; Drop un-marks even on
 /// panic, so one failed turn cannot brick the session.
 pub struct BusyGuard(StdArc<Mutex<bool>>);

crates/tole-cli/src/session_tools.rs

@@ -87,6 +87,7 @@ impl SessionToolState {
             Box::new(SessionNewTool(self.clone())),
             Box::new(SessionPromptTool(self.clone())),
             Box::new(SessionCancelTool(self.clone())),
+            Box::new(SessionInterruptTool(self.clone())),
             Box::new(SessionStatusTool(self.clone())),
             Box::new(SessionListTool(self.clone())),
         ]
@@ -271,6 +272,71 @@ impl Tool for SessionPromptTool {
     }
 }
 
+// ---------------------------------------------------------------------------
+// tole_session_interrupt — soft interrupt (#377)
+// ---------------------------------------------------------------------------
+
+/// Steers the target session's in-flight turn WITHOUT cancelling it: the
+/// message lands at the turn's next safe point as an attributed user-role
+/// entry. Same classification as `tole_session_cancel` (callable while a
+/// `tole_session_prompt` runs; the session's own approver still gates every
+/// tool the steered turn runs).
+struct SessionInterruptTool(SessionToolState);
+
+impl Tool for SessionInterruptTool {
+    fn name(&self) -> &str {
+        "tole_session_interrupt"
+    }
+    fn risk(&self) -> Risk {
+        Risk::ReadOnly
+    }
+    fn summary(&self) -> String {
+        "Steer the in-flight turn of a tole session with a message, without cancelling it.".into()
+    }
+    fn describe(&self, input: &Value) -> String {
+        format!(
+            "interrupt session {:?}",
+            input
+                .get("session_id")
+                .and_then(Value::as_str)
+                .unwrap_or("<implicit>")
+        )
+    }
+    fn spec(&self) -> Option<Value> {
+        Some(json!({
+            "type": "object",
+            "properties": {
+                "session_id": { "type": "string", "description": "Target session (optional when exactly one is open)" },
+                "text": { "type": "string", "description": "The steering message" },
+                "urgent": { "type": "boolean", "description": "Also withdraw a tool call that has not started yet (default false)" }
+            },
+            "required": ["text"]
+        }))
+    }
+    fn execute(&self, input: Value) -> Result<Value, String> {
+        let text = input
+            .get("text")
+            .and_then(Value::as_str)
+            .ok_or("missing 'text'")?;
+        let urgent = input
+            .get("urgent")
+            .and_then(Value::as_bool)
+            .unwrap_or(false);
+        let st = self.0.resolve(&input)?;
+        match crate::session_host::interrupt_session(&st, text, urgent) {
+            Ok(()) => Ok(json!({
+                "queued": true,
+                "urgent": urgent,
+                "note": "delivered at the turn's next safe point (after the running tool, before the next model call)"
+            })),
+            Err(crate::session_host::InterruptRefusal::NotBusy) => {
+                Err("no turn is running in this session".into())
+            }
+            Err(crate::session_host::InterruptRefusal::Invalid(e)) => Err(e),
+        }
+    }
+}
+
 // ---------------------------------------------------------------------------
 // tole_session_cancel — the MCP face of #178
 // ---------------------------------------------------------------------------

crates/tole-cli/tests/serve_session_cancel.rs

@@ -232,11 +232,19 @@ fn spawn_mock() -> String {
                 if last_user.contains("DELAYED") {
                     std::thread::sleep(Duration::from_secs(2));
                 }
+                // #377: a transcript carrying the owner's soft interrupt
+                // is answered differently, so the test can see the
+                // steering reached the provider.
+                let steered = msgs.iter().any(|m| {
+                    m["content"]
+                        .as_str()
+                        .is_some_and(|c| c.contains("owner interrupt"))
+                });
                 let reply = json!({
                     "id": "chatcmpl-mock", "object": "chat.completion", "created": 1,
                     "model": "mock",
                     "choices": [{"index": 0, "message": {"role": "assistant",
-                        "content": "DONE"}, "finish_reason": "stop"}],
+                        "content": if steered { "STEERED" } else { "DONE" }}, "finish_reason": "stop"}],
                     "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}
                 });
                 let resp = sse_response(&reply, streamed);
@@ -299,6 +307,57 @@ fn serve_cancel_mid_provider_settles_cancelled_and_resumable() {
     assert_eq!(resp["stopReason"].as_str(), Some("end_turn"), "{resp}");
 }
 
+/// #377: `POST /sessions/{id}/interrupt` steers the in-flight turn without
+/// cancelling it. The first (stale) answer is superseded, the next provider
+/// call carries the attributed interrupt, and the prompt returns the
+/// steered answer. Idle / unknown / malformed requests are refused.
+#[test]
+fn serve_soft_interrupt_steers_the_running_turn() {
+    let base = spawn_mock();
+    let serve = ServeProcess::spawn(&base);
+    let (code, created) = serve.request("POST", "/sessions", "{}").expect("create");
+    assert_eq!(code, 200, "{created}");
+    let sid = created["sessionId"].as_str().unwrap().to_string();
+    let path = format!("/sessions/{sid}/interrupt");
+
+    // Refusals: unknown session, idle session, bad bodies.
+    let (code, _) = serve
+        .request("POST", "/sessions/nope/interrupt", r#"{"text":"x"}"#)
+        .unwrap();
+    assert_eq!(code, 404);
+    let (code, body) = serve.request("POST", &path, r#"{"text":"x"}"#).unwrap();
+    assert_eq!(code, 409, "no turn running: {body}");
+    let (code, _) = serve.request("POST", &path, "not json").unwrap();
+    assert_eq!(code, 400);
+    let (code, _) = serve.request("POST", &path, r#"{"nope":1}"#).unwrap();
+    assert_eq!(code, 400);
+
+    // A running turn (the first provider call takes ~2 s).
+    let thread_serve = Arc::clone(&serve);
+    let sid_clone = sid.clone();
+    let prompt = std::thread::spawn(move || {
+        thread_serve.request(
+            "POST",
+            &format!("/sessions/{sid_clone}/prompt"),
+            r#"{"text": "DELAYED mission"}"#,
+        )
+    });
+    std::thread::sleep(Duration::from_millis(500));
+    let (code, _) = serve.request("POST", &path, r#"{"text":"   "}"#).unwrap();
+    assert_eq!(code, 400, "blank text is refused, not queued");
+    let (code, ack) = serve
+        .request("POST", &path, r#"{"text":"prefer approach Y"}"#)
+        .unwrap();
+    assert_eq!(code, 202, "{ack}");
+    assert_eq!(ack["queued"], json!(true));
+    assert_eq!(ack["urgent"], json!(false));
+
+    let (pcode, presp) = prompt.join().unwrap().expect("prompt response");
+    assert_eq!(pcode, 200, "{presp}");
+    assert_eq!(presp["stopReason"].as_str(), Some("end_turn"), "{presp}");
+    assert_eq!(presp["text"].as_str(), Some("STEERED"), "{presp}");
+}
+
 /// #211: `GET /sessions/{id}/status` gains an ADDITIVE `usage_report`
 /// object; the pre-existing keys stay. A fresh session reports zero
 /// steps and `null` for what was never measured; after a turn the real

crates/tole-cli/tests/session_tools.rs

@@ -237,3 +237,65 @@ fn session_is_rooted_at_the_canonical_cwd_the_jail_check_validated() {
         want.as_path()
     );
 }
+
+/// #377 MCP face: `tole_session_interrupt` queues a soft interrupt on the
+/// session's turn handle only while a turn is running; idle, unknown,
+/// blank and missing-text calls are clear errors.
+#[test]
+fn session_interrupt_tool_queues_only_for_a_running_turn() {
+    let dir = ws("interrupt-tool");
+    let state = SessionToolState::new(
+        vec![],
+        false,
+        None,
+        std::path::PathBuf::from(&dir),
+        None,
+        vec![],
+    );
+    let tools = state.tools();
+    let tool = tool_by_name(&tools, "tole_session_interrupt");
+
+    let err = tool
+        .execute(json!({"session_id": "nope", "text": "x"}))
+        .unwrap_err();
+    assert!(err.contains("unknown session"), "{err}");
+
+    let a = std::path::Path::new(&dir).join("a");
+    let sid = tool_by_name(&tools, "tole_session_new")
+        .execute(json!({"cwd": a.to_string_lossy()}))
+        .unwrap()["session_id"]
+        .as_str()
+        .unwrap()
+        .to_string();
+    let (token, busy) = {
+        let sessions = tole_cli::session_host::lock_sessions(&state.sessions);
+        let st = sessions.map.get(&sid).unwrap();
+        (st.cancel.clone(), std::sync::Arc::clone(&st.busy))
+    };
+
+    // Idle: nothing to steer.
+    let err = tool.execute(json!({"text": "x"})).unwrap_err();
+    assert!(err.contains("no turn is running"), "{err}");
+    assert!(token.interrupts().is_empty());
+
+    // Missing / blank text.
+    assert!(tool
+        .execute(json!({}))
+        .unwrap_err()
+        .contains("missing 'text'"));
+    *busy.lock().unwrap() = true;
+    assert!(tool.execute(json!({"text": "  "})).is_err());
+    assert!(token.interrupts().is_empty());
+
+    // Running: queued on THE session's handle, implicit session resolved.
+    let out = tool
+        .execute(json!({"text": "switch to Y", "urgent": true}))
+        .unwrap();
+    assert_eq!(out["queued"], json!(true));
+    assert_eq!(out["urgent"], json!(true));
+    let q = token.interrupts().drain();
+    assert_eq!(q.len(), 1);
+    assert_eq!((q[0].text.as_str(), q[0].urgent), ("switch to Y", true));
+    *busy.lock().unwrap() = false;
+    let _ = std::fs::remove_dir_all(&dir);
+}

docs/mobile-control.md

@@ -20,6 +20,7 @@ rate-limited per source IP).
 | **Pending approvals** | `GET /approvals` | queue entries: id, session, tool, status; expires to denied after 15 min |
 | **Approve / deny** | `POST /approvals/{id}/decision` `{"decision":"allow"\|"deny"}` | allow = one-shot for exactly that (session, tool+input) + automatic resume; deny = recorded verdict |
 | Cancel a turn | `POST /sessions/{id}/cancel` (#178) | the turn settles `cancelled`, durably |
+| Steer a turn | `POST /sessions/{id}/interrupt` `{text, urgent?}` (#377) | `202` queued; delivered at the turn's next safe point (after the running tool, before the next model call) as an attributed user-role entry, nothing is lost. `urgent` also withdraws a tool call that has not started. `409` when no turn is running, `400` for blank/oversized text or a full queue (16 pending) |
 | Cost report | `GET /sessions/{id}/status` | the `mission` register: turns, steps, tokens, wall time, per-risk-tier tool calls |
 
 The round trip the app must support (the cross-repo acceptance):

docs/threat-model.md

@@ -173,7 +173,7 @@ failing gate cannot livelock the loop — the turn settles durably as
 
 All three server faces expose cooperative turn cancellation: ACP
 clients send `session/cancel`, REST callers `POST /sessions/{id}/cancel`,
-MCP clients call `tole_session_cancel`. Every path sets the same
+MCP clients call `tole_session_cancel`. Steering (soft interrupt, #377) goes through the same authenticated faces (`POST /sessions/{id}/interrupt`, `tole_session_interrupt`) and adds no authority: the message is ordinary user-role input to the session, so every tool the steered turn runs still passes the session's approver. Every path sets the same
 per-session token; a blocking turn observes it at checkpoints (between
 provider calls and tool executions) and settles
 `stopReason: "cancelled"` as a normal, durable turn end — the session