From 60459bb30f4eed059c5d530bc36d4b40e3f3558a Mon Sep 17 00:00:00 2001 From: ajianaz Date: Fri, 16 Oct 2026 11:22:37 +0700 Subject: [PATCH 1/3] =?UTF-8?q?feat(core):=20soft=20interrupts=20=E2=80=94?= =?UTF-8?q?=20steer=20a=20running=20turn=20without=20cancelling=20it=20(#3?= =?UTF-8?q?77,=201/2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 1 + crates/tole-core/src/cancel.rs | 9 + crates/tole-core/src/interrupt.rs | 150 +++++++++++ crates/tole-core/src/lib.rs | 1 + crates/tole-core/src/turn.rs | 56 ++++ crates/tole-core/tests/soft_interrupt.rs | 314 +++++++++++++++++++++++ docs/architecture.md | 2 + 7 files changed, 533 insertions(+) create mode 100644 crates/tole-core/src/interrupt.rs create mode 100644 crates/tole-core/tests/soft_interrupt.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c55f3e..06b1c28 100644 --- a/CHANGELOG.md +++ b/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) and `catalog` (live `/models`, the model must be served); each blocker carries its fix command, the exit code is non-zero when the requested tier is not reached, and keys / URL credentials are never printed. ### Changed diff --git a/crates/tole-core/src/cancel.rs b/crates/tole-core/src/cancel.rs index 687e068..ee50ec4 100644 --- a/crates/tole-core/src/cancel.rs +++ b/crates/tole-core/src/cancel.rs @@ -21,6 +21,10 @@ use std::sync::Arc; #[derive(Clone, Default)] pub struct CancelToken { flag: Arc, + /// 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 { @@ -28,6 +32,11 @@ impl CancelToken { 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) { diff --git a/crates/tole-core/src/interrupt.rs b/crates/tole-core/src/interrupt.rs new file mode 100644 index 0000000..b8bf0fb --- /dev/null +++ b/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>>, +} + +impl InterruptQueue { + fn lock(&self) -> std::sync::MutexGuard<'_, VecDeque> { + // 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 { + 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); + } +} diff --git a/crates/tole-core/src/lib.rs b/crates/tole-core/src/lib.rs index 48105e3..e39b11e 100644 --- a/crates/tole-core/src/lib.rs +++ b/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; diff --git a/crates/tole-core/src/turn.rs b/crates/tole-core/src/turn.rs index 9922172..4ee9e1a 100644 --- a/crates/tole-core/src/turn.rs +++ b/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() diff --git a/crates/tole-core/tests/soft_interrupt.rs b/crates/tole-core/tests/soft_interrupt.rs new file mode 100644 index 0000000..ae5c079 --- /dev/null +++ b/crates/tole-core/tests/soft_interrupt.rs @@ -0,0 +1,314 @@ +//! #377 — soft interrupts: steer a running turn without cancelling it. +//! +//! The mock provider/tool play the "second client": they push into the +//! session's `CancelToken` interrupt queue at the moments a real client +//! could (while a tool runs, while the provider call is in flight). + +use serde_json::{json, Value}; +use std::collections::VecDeque; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; +use tole_core::approval::AllowlistApprover; +use tole_core::cancel::CancelToken; +use tole_core::entry::Entry; +use tole_core::provider::{Provider, ProviderError, ProviderOutput}; +use tole_core::storage::{JsonlStorage, Storage}; +use tole_core::tool::{Risk, Tool, ToolRegistry}; +use tole_core::turn::{run_turn_with_cancel, TurnOutcome}; + +fn tmpdir(name: &str) -> std::path::PathBuf { + static SEQ: AtomicUsize = AtomicUsize::new(0); + let n = SEQ.fetch_add(1, Ordering::Relaxed); + let d = std::env::temp_dir().join(format!("tole-int-{name}-{}-{n}", std::process::id())); + let _ = std::fs::remove_dir_all(&d); + std::fs::create_dir_all(&d).unwrap(); + d +} + +/// ReadOnly tool that counts executions and, when armed, pushes an +/// interrupt while it "runs". +struct WorkTool { + runs: Arc, + push_while_running: Option<(CancelToken, &'static str, bool)>, +} + +impl Tool for WorkTool { + fn name(&self) -> &str { + "work" + } + fn risk(&self) -> Risk { + Risk::ReadOnly + } + fn execute(&self, _input: Value) -> Result { + self.runs.fetch_add(1, Ordering::SeqCst); + if let Some((tok, text, urgent)) = &self.push_while_running { + tok.interrupts().push(text, *urgent).unwrap(); + } + Ok(json!({"done": true})) + } +} + +fn registry(tool: WorkTool) -> ToolRegistry { + let mut r = ToolRegistry::with_approver(AllowlistApprover::allow_only(vec![])); + r.register(Box::new(tool)).unwrap(); + r +} + +type Hook = Box; + +/// Scripted provider that records each transcript it was shown and can +/// run a hook at the start of a call (a client pushing mid-call). +struct Steered { + outputs: VecDeque, + hook: Option, + seen: Vec>, +} + +impl Provider for Steered { + fn complete(&mut self, transcript: &[Entry]) -> Result { + let n = self.seen.len(); + self.seen.push(transcript.to_vec()); + if let Some(h) = self.hook.as_mut() { + h(n); + } + self.outputs + .pop_front() + .ok_or_else(|| ProviderError("script exhausted".into())) + } +} + +fn call_work() -> ProviderOutput { + ProviderOutput::ToolCall { + tool: "work".into(), + input: json!({}), + } +} + +fn final_text(t: &str) -> ProviderOutput { + ProviderOutput::Final { text: t.into() } +} + +fn is_interrupt(e: &Entry) -> bool { + e.kind.as_str() == "message" && e.payload["interrupt"] == json!(true) +} + +/// No `tool_use` without its `tool_result`: every intent has one settlement. +fn assert_no_orphan_intents(entries: &[Entry]) { + let intents = entries + .iter() + .filter(|e| e.kind.as_str() == "intent") + .count(); + let results = entries + .iter() + .filter(|e| matches!(e.kind.as_str(), "tool_result" | "error") && e.parent_id.is_some()) + .filter(|e| { + entries + .iter() + .any(|i| i.kind.as_str() == "intent" && Some(&i.id) == e.parent_id.as_ref()) + }) + .count(); + assert_eq!(intents, results, "an intent was left without a settlement"); +} + +#[test] +fn normal_interrupt_is_delivered_after_the_tool_settles_and_before_the_next_call() { + let dir = tmpdir("safe-point"); + let mut s = JsonlStorage::create(&dir, "s", None).unwrap(); + let cancel = CancelToken::default(); + let runs = Arc::new(AtomicUsize::new(0)); + let reg = registry(WorkTool { + runs: Arc::clone(&runs), + push_while_running: Some((cancel.clone(), "focus on the tests", false)), + }); + let mut p = Steered { + outputs: VecDeque::from([call_work(), final_text("done")]), + hook: None, + seen: vec![], + }; + let out = run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + assert!(matches!(out, TurnOutcome::Final { .. }), "{out:?}"); + assert_eq!(runs.load(Ordering::SeqCst), 1); + + // The second provider call sees: user, intent, tool_result, interrupt. + let second = &p.seen[1]; + let kinds: Vec<&str> = second.iter().map(|e| e.kind.as_str()).collect(); + let at = second + .iter() + .position(is_interrupt) + .expect("interrupt visible"); + assert!(kinds[..at].contains(&"tool_result"), "{kinds:?}"); + assert_eq!(at, second.len() - 1, "delivered right before the call"); + + // Attributed, and not a turn-opening prompt. + let e = &second[at]; + assert_eq!(e.payload["role"], json!("user")); + assert_eq!(e.payload["urgent"], json!(false)); + let text = e.payload["text"].as_str().unwrap(); + assert!( + text.contains("owner interrupt") && text.contains("focus on the tests"), + "{text}" + ); + assert!(cancel.interrupts().is_empty()); + assert_no_orphan_intents(s.entries()); +} + +#[test] +fn urgent_interrupt_withdraws_an_unstarted_tool_call_without_orphaning_it() { + let dir = tmpdir("urgent"); + let mut s = JsonlStorage::create(&dir, "u", None).unwrap(); + let cancel = CancelToken::default(); + let runs = Arc::new(AtomicUsize::new(0)); + let reg = registry(WorkTool { + runs: Arc::clone(&runs), + push_while_running: None, + }); + let c = cancel.clone(); + let mut p = Steered { + outputs: VecDeque::from([call_work(), final_text("changed course")]), + // The client interrupts while the first provider call is in flight. + hook: Some(Box::new(move |n| { + if n == 0 { + c.interrupts().push("stop, do X instead", true).unwrap(); + } + })), + seen: vec![], + }; + let out = run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + assert!( + matches!(out, TurnOutcome::Final { ref text, .. } if text == "changed course"), + "{out:?}" + ); + assert_eq!( + runs.load(Ordering::SeqCst), + 0, + "the withdrawn tool never ran" + ); + let entries = s.entries(); + assert!( + entries.iter().all(|e| e.kind.as_str() != "intent"), + "no intent was written for the withdrawn call" + ); + assert_no_orphan_intents(entries); + // The drop is recorded, not silent. + assert!(entries.iter().any(|e| e.kind.as_str() == "error" + && e.payload["error"] == json!("superseded by owner interrupt") + && e.payload["detail"].as_str().unwrap().contains("work"))); + // And the model saw the steering on its next call. + assert!(p.seen[1].iter().any(is_interrupt)); + assert!(p.seen[1].iter().any(|e| e.payload["urgent"] == json!(true))); +} + +#[test] +fn a_normal_interrupt_does_not_withdraw_a_tool_call() { + let dir = tmpdir("normal-keeps-tool"); + let mut s = JsonlStorage::create(&dir, "n", None).unwrap(); + let cancel = CancelToken::default(); + let runs = Arc::new(AtomicUsize::new(0)); + let reg = registry(WorkTool { + runs: Arc::clone(&runs), + push_while_running: None, + }); + let c = cancel.clone(); + let mut p = Steered { + outputs: VecDeque::from([call_work(), final_text("done")]), + hook: Some(Box::new(move |n| { + if n == 0 { + c.interrupts().push("by the way, prefer Y", false).unwrap(); + } + })), + seen: vec![], + }; + let out = run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + assert!(matches!(out, TurnOutcome::Final { .. })); + assert_eq!(runs.load(Ordering::SeqCst), 1, "the call still ran"); + // Delivered at the next safe point, after the tool settled. + let second = &p.seen[1]; + let result_at = second + .iter() + .position(|e| e.kind.as_str() == "tool_result") + .unwrap(); + let int_at = second.iter().position(is_interrupt).unwrap(); + assert!(int_at > result_at); + assert_no_orphan_intents(s.entries()); +} + +#[test] +fn a_final_answer_is_superseded_by_a_pending_interrupt() { + let dir = tmpdir("final"); + let mut s = JsonlStorage::create(&dir, "f", None).unwrap(); + let cancel = CancelToken::default(); + let reg = registry(WorkTool { + runs: Arc::new(AtomicUsize::new(0)), + push_while_running: None, + }); + let c = cancel.clone(); + let mut p = Steered { + outputs: VecDeque::from([final_text("stale answer"), final_text("steered answer")]), + hook: Some(Box::new(move |n| { + if n == 0 { + c.interrupts().push("also cover Z", false).unwrap(); + } + })), + seen: vec![], + }; + let out = run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + assert!( + matches!(out, TurnOutcome::Final { ref text, .. } if text == "steered answer"), + "{out:?}" + ); + let entries = s.entries(); + assert!( + !entries + .iter() + .any(|e| e.payload["text"] == json!("stale answer")), + "the withdrawn answer was never committed" + ); + assert!(entries + .iter() + .any(|e| e.payload["error"] == json!("superseded by owner interrupt"))); + assert!(entries.iter().any(is_interrupt)); +} + +#[test] +fn an_interrupt_queued_before_the_turn_lands_after_the_prompt() { + let dir = tmpdir("queued"); + let mut s = JsonlStorage::create(&dir, "q", None).unwrap(); + let cancel = CancelToken::default(); + cancel.interrupts().push("queued steering", false).unwrap(); + let reg = registry(WorkTool { + runs: Arc::new(AtomicUsize::new(0)), + push_while_running: None, + }); + let mut p = Steered { + outputs: VecDeque::from([final_text("ok")]), + hook: None, + seen: vec![], + }; + run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + let first = &p.seen[0]; + let user_at = first + .iter() + .position(|e| e.payload["text"] == json!("go")) + .unwrap(); + let int_at = first.iter().position(is_interrupt).unwrap(); + assert!(int_at > user_at, "the original prompt still comes first"); +} + +#[test] +fn without_interrupts_nothing_changes() { + let dir = tmpdir("none"); + let mut s = JsonlStorage::create(&dir, "z", None).unwrap(); + let cancel = CancelToken::default(); + let reg = registry(WorkTool { + runs: Arc::new(AtomicUsize::new(0)), + push_while_running: None, + }); + let mut p = Steered { + outputs: VecDeque::from([call_work(), final_text("done")]), + hook: None, + seen: vec![], + }; + run_turn_with_cancel(&mut s, &mut p, ®, "go", &cancel).unwrap(); + assert!(!s.entries().iter().any(is_interrupt)); + assert!(!s.entries().iter().any(|e| e.kind.as_str() == "error")); +} diff --git a/docs/architecture.md b/docs/architecture.md index fafe5da..3a08379 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -153,6 +153,8 @@ sequenceDiagram All three tool-authorization paths — a fresh provider call (`drive`), a crash-replayed intent (`resume_turn`) and the MCP server (`RegistryServer::execute_checked`, `Mode::Fresh`) — share one authorization module, `crates/tole-core/src/gate.rs` (crate-internal; see [GLOSSARY.md](../GLOSSARY.md)). `gate::authorize(registry, name, input, Mode)` is the policy decision point: it reads the tool's risk once, then consults the approver and the opt-in pre-hooks in that order, returning an `Authorized` call or a typed `Denied` (`Approver`, `PreHook`, `UnknownTool`). `Mode::Fresh` gates when risk is not `ReadOnly`; `Mode::Replay { recorded }` gates when the recorded safety is `Guarded` **or** the current risk is not `ReadOnly` (#249). +Soft interrupts (#377) steer a running turn without cancelling it. The session's `CancelToken` also carries a bounded `InterruptQueue` (16 pending, 8 KiB each; blank or oversized text is refused, never truncated). The loop drains it at its one safe point — the tool has settled, the next provider call has not started — and commits each message as a user-role entry marked `interrupt: true` (so it never opens a turn, like the stop-gate feedback) with the attribution in the text, since that is all the provider sees. An interrupt that lands while a provider call is in flight supersedes that output before it takes effect: a final answer yields to any pending interrupt, a not-yet-started tool call only to an `urgent` one. The withdrawn output was never committed (no intent is written), so no `tool_use` can be left without its `tool_result`; a durable `superseded by owner interrupt` error entry records the drop. An interrupt that arrives after a turn's last safe point waits for the next turn's first. + The turn loop remains the enforcement point: it maps each `Denied` to the existing durable records and `TurnOutcome`s, and owns the effect sandwich, cancel checkpoints, observer, post-hooks and the `wrote_this_turn` fact. A non-`ReadOnly` `Authorized` carries a `Permit` (private constructor) and `Authorized::execute` is the only way the loop runs a tool, so executing a Write/Destructive tool without passing the gate does not compile inside the crate. The MCP server keeps its own server-mode policy outside the gate (Destructive is refused before the gate is called, approver never consulted); pre-hooks configured on a registry passed to `RegistryServer::new` are enforced by the gate there, though no in-repo server face attaches any. The public `Tool` trait and `ToolRegistry::decide` are unchanged. ## 8. Provider From 1bf5c35401c31f8ae42c040bb330bc8ade481570 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Fri, 16 Oct 2026 11:26:34 +0700 Subject: [PATCH 2/3] fix(cli): a soft interrupt does not count as a user turn --- crates/tole-cli/src/main.rs | 10 ++++++++-- crates/tole-cli/src/usage_report.rs | 19 ++++++++++++++++++- 2 files changed, 26 insertions(+), 3 deletions(-) diff --git a/crates/tole-cli/src/main.rs b/crates/tole-cli/src/main.rs index 96ebc15..3ad6a0a 100644 --- a/crates/tole-cli/src/main.rs +++ b/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, diff --git a/crates/tole-cli/src/usage_report.rs b/crates/tole-cli/src/usage_report.rs index 635002e..2ca7053 100644 --- a/crates/tole-cli/src/usage_report.rs +++ b/crates/tole-cli/src/usage_report.rs @@ -207,11 +207,12 @@ fn add(acc: &mut Option, x: Option) { } /// 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)]; From 76fb9d5c2f8325699a9de561ad383237f83c7aa9 Mon Sep 17 00:00:00 2001 From: ajianaz Date: Fri, 16 Oct 2026 11:49:50 +0700 Subject: [PATCH 3/3] =?UTF-8?q?feat(cli):=20soft=20interrupt=20entry=20poi?= =?UTF-8?q?nts=20=E2=80=94=20POST=20/sessions/{id}/interrupt=20and=20tole?= =?UTF-8?q?=5Fsession=5Finterrupt=20(#377,=202/2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 2 +- README.md | 6 +- crates/tole-cli/src/serve.rs | 37 +++++++++++ crates/tole-cli/src/session_host.rs | 28 ++++++++ crates/tole-cli/src/session_tools.rs | 66 +++++++++++++++++++ crates/tole-cli/tests/serve_session_cancel.rs | 61 ++++++++++++++++- crates/tole-cli/tests/session_tools.rs | 62 +++++++++++++++++ docs/mobile-control.md | 1 + docs/threat-model.md | 2 +- 9 files changed, 261 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f16b9a..e4287fe 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index dcf79ff..767fdc4 100644 --- a/README.md +++ b/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 diff --git a/crates/tole-cli/src/serve.rs b/crates/tole-cli/src/serve.rs index cef4654..5fd384d 100644 --- a/crates/tole-cli/src/serve.rs +++ b/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::(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"})), } } diff --git a/crates/tole-cli/src/session_host.rs b/crates/tole-cli/src/session_host.rs index a289b8c..ceafe16 100644 --- a/crates/tole-cli/src/session_host.rs +++ b/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>); diff --git a/crates/tole-cli/src/session_tools.rs b/crates/tole-cli/src/session_tools.rs index a3152e7..560c610 100644 --- a/crates/tole-cli/src/session_tools.rs +++ b/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("") + ) + } + fn spec(&self) -> Option { + 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 { + 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 // --------------------------------------------------------------------------- diff --git a/crates/tole-cli/tests/serve_session_cancel.rs b/crates/tole-cli/tests/serve_session_cancel.rs index bb8260c..7453d15 100644 --- a/crates/tole-cli/tests/serve_session_cancel.rs +++ b/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 diff --git a/crates/tole-cli/tests/session_tools.rs b/crates/tole-cli/tests/session_tools.rs index 1aca8c9..93dee59 100644 --- a/crates/tole-cli/tests/session_tools.rs +++ b/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); +} diff --git a/docs/mobile-control.md b/docs/mobile-control.md index c8c0b9c..251dd27 100644 --- a/docs/mobile-control.md +++ b/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): diff --git a/docs/threat-model.md b/docs/threat-model.md index 5ec3641..7799f30 100644 --- a/docs/threat-model.md +++ b/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