feat(infinity-daemon): add input-less Interrupt for in-flight generation

Adds a real interrupt capability to the protocol and daemon: clients can
stop a thread's in-flight work without queuing input and without shutting
the session down.

## Protocol (`infinity-protocol`)

* New `ClientMessage::Interrupt { thread_id }`: interrupts the thread and
  its descendant subtree. Additive enum variant; old daemons silently skip
  unparsable frames (both the unix-socket and websocket handlers
  `continue` on deserialization failure), so sending to an old daemon is a
  safe no-op.

## Core (`infinity-agent-core`)

* New `TaggedSyntheticKind::Interrupt` input marker (plus
  `SyntheticKind::is_interrupt` / `thread::is_interrupt_input`). The marker
  is consumed entirely by the thread driver — never folded into history or
  queued for the model.
* Driver: an interrupt marker cancels the in-flight completion exactly like
  user text does (the cancelled step flushes partial output and emits its
  completion notification); markers reaching a parked or respawned driver
  settle the thread directly. User text in the same batch supersedes the
  marker.
* New `Thread::interrupt()`: settles a trailing unanswered tool call with
  the synthetic "interrupted by user" result (so a late real result is
  matched as already-answered and dropped instead of restarting
  generation), sends best-effort RAP cancellation notifications, dismisses
  pending user choices (broadcast as `UserChoiceComplete`), and persists.
  The interrupted-call settlement logic is factored out of `step_no_defer`
  and shared. Exactly one `CompletionFinished` (→ `ResponseDone`) is
  emitted per interrupted round so clients settle their indicators.

## Daemon (`infinity-daemon`)

* `SessionManager::interrupt_thread()`: walks the target's subtree
  (children tree) and delivers interrupt markers to the target plus every
  active descendant (live or parked drivers; idle descendants are skipped —
  nothing in flight to cancel). Threads stay alive and accept new input,
  like after a normal `ResponseDone`.
* `ClientMessage::Interrupt` handled in the client handler (local and
  remote-proxy paths; thread refs are stripped/prefixed like `UserInput`).
  Web clients get the message for free via the existing ws passthrough.

## Tests

* core: interrupt cancels an in-flight completion without queuing input;
  interrupt settles a pending async tool call so its late result is
  dropped; threads accept input afterwards.
* daemon: `interrupt_thread` targets exactly the requested subtree —
  interrupting a child leaves the root streaming, and both threads accept
  new input after interruption.

Author: akainth015

crates/infinity-agent-core/src/event_processor.rs

@@ -321,7 +321,11 @@ impl<C: ConversationStore, S: StateStore> HistoryManager<C, S> {
     /// If the last history entry is an unanswered tool call, inject a
     /// synthetic "interrupted" result. (A tool call at the tail is unanswered
     /// by construction: its result would have been appended after it.)
-    fn interrupt_pending_tool_call(&self) {
+    ///
+    /// Exposed to the crate for input-less interrupts
+    /// ([`Thread::interrupt`][crate::system::thread::Thread::interrupt]),
+    /// which settle the trailing call without folding any input into history.
+    pub(crate) fn interrupt_pending_tool_call(&self) {
         let last_call = self.history.borrow().last().and_then(|m| {
             if let InfinityMessage::ToolCall { call, .. } = m {
                 Some(call.clone())

crates/infinity-agent-core/src/message.rs

@@ -68,6 +68,12 @@ pub enum TaggedSyntheticKind {
     Compaction,
     #[serde(rename = "compaction_complete")]
     CompactionComplete,
+    /// Input-less interrupt: cancels the thread's in-flight completion and
+    /// settles a trailing unanswered tool call, without queuing anything for
+    /// the model. Consumed entirely by the thread driver — it is never
+    /// folded into history.
+    #[serde(rename = "interrupt")]
+    Interrupt,
 }
 
 impl SyntheticKind {
@@ -86,6 +92,7 @@ impl SyntheticKind {
             SyntheticKind::Tagged(TaggedSyntheticKind::CompactionComplete) => {
                 ToolCallId::from_ref("")
             }
+            SyntheticKind::Tagged(TaggedSyntheticKind::Interrupt) => ToolCallId::from_ref(""),
             SyntheticKind::SubscriptionEvent(id) => id,
         }
     }
@@ -115,6 +122,12 @@ impl SyntheticKind {
         )
     }
 
+    /// Whether this is an input-less interrupt (see
+    /// [`TaggedSyntheticKind::Interrupt`]).
+    pub fn is_interrupt(&self) -> bool {
+        matches!(self, SyntheticKind::Tagged(TaggedSyntheticKind::Interrupt))
+    }
+
     /// Associative subscription events are injected inline into the subscribing
     /// thread's history (like thread reports) rather than spawning a child thread.
     pub fn is_associative(&self) -> bool {

crates/infinity-agent-core/src/system/local/driver.rs

@@ -26,7 +26,7 @@ use crate::system::builder::SystemInner;
 use crate::system::defer::InMemoryDeferQueue;
 use crate::system::events::AgentEvent;
 use crate::system::observer::ThreadObserver;
-use crate::system::thread::{StepOutcome, Thread, is_user_text_input};
+use crate::system::thread::{StepOutcome, Thread, is_interrupt_input, is_user_text_input};
 
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
 
@@ -175,8 +175,11 @@ pub(crate) async fn drive_thread<C, S, H, O>(
                         batch.push(item);
                     }
 
-                    if batch.iter().any(|(msg, _)| is_user_text_input(msg)) {
-                        // User text interrupts the in-flight completion.
+                    let has_user_text = batch.iter().any(|(msg, _)| is_user_text_input(msg));
+                    // Input-less interrupts cancel the in-flight completion
+                    // just like user text; the markers themselves are consumed
+                    // by the shared stripping below (never stepped).
+                    if has_user_text || batch.iter().any(|(msg, _)| is_interrupt_input(msg)) {
                         let res = in_flight
                             .take()
                             .expect("bug: in-flight step missing during interrupt")
@@ -191,20 +194,41 @@ pub(crate) async fn drive_thread<C, S, H, O>(
                             &mut pending_batch,
                         );
 
-                        let (mut user_inputs, non_user_inputs): (Vec<_>, Vec<_>) = batch
-                            .into_iter()
-                            .partition(|(msg, _)| is_user_text_input(msg));
+                        if has_user_text {
+                            // User text supersedes an interrupt marker in the
+                            // same batch: folding the text already settles a
+                            // trailing tool call, so the marker is redundant.
+                            batch.retain(|(msg, _)| !is_interrupt_input(msg));
+                            let (mut user_inputs, non_user_inputs): (Vec<_>, Vec<_>) = batch
+                                .into_iter()
+                                .partition(|(msg, _)| is_user_text_input(msg));
+
+                            if let InputMessageContent::User(UserContent::Text(text)) =
+                                &mut user_inputs[0].0.content
+                            {
+                                text.text = format!("<interrupt>{}", text.text);
+                            } else {
+                                panic!("bug: user_inputs should only contain user text");
+                            }
 
-                        if let InputMessageContent::User(UserContent::Text(text)) =
-                            &mut user_inputs[0].0.content
-                        {
-                            text.text = format!("<interrupt>{}", text.text);
+                            pending_batch.extend(non_user_inputs);
+                            user_inputs
                         } else {
-                            panic!("bug: user_inputs should only contain user text");
+                            // Consume the markers: settle the thread without
+                            // emitting another completion notification (the
+                            // cancelled step already emitted one).
+                            batch.retain(|(msg, _)| !is_interrupt_input(msg));
+                            if let Err(e) = thread.interrupt(&observer, false).await {
+                                observer.on_event(
+                                    &thread_id,
+                                    &AgentEvent::Info {
+                                        text: format!("Error: {e}"),
+                                    },
+                                );
+                            }
+                            pending_batch.extend(batch);
+                            continue;
                         }
-
-                        pending_batch.extend(non_user_inputs);
-                        user_inputs
                     } else {
                         pending_batch.extend(batch);
                         continue;
@@ -293,6 +317,26 @@ pub(crate) async fn drive_thread<C, S, H, O>(
             }
         }
 
+        // Consume input-less interrupt markers: settle the thread (synthetic
+        // "interrupted" result for a trailing tool call, RAP cancellation
+        // notifications, pending-choice dismissal) without queueing anything
+        // for the model. Markers that cancelled an in-flight step were
+        // consumed in the select arm above; these reached a parked or
+        // respawned driver with nothing in flight, so the settlement itself
+        // emits the completion notification that settles client indicators.
+        let mut new_inputs = new_inputs;
+        if new_inputs.iter().any(|(msg, _)| is_interrupt_input(msg)) {
+            new_inputs.retain(|(msg, _)| !is_interrupt_input(msg));
+            if let Err(e) = thread.interrupt(&observer, true).await {
+                observer.on_event(
+                    &thread_id,
+                    &AgentEvent::Info {
+                        text: format!("Error: {e}"),
+                    },
+                );
+            }
+        }
+
         let batch = match thread.filter_deferrable(new_inputs, &mut defer).await {
             Ok(batch) => batch,
             Err(e) => {

crates/infinity-agent-core/src/system/test_support.rs

@@ -83,6 +83,23 @@ pub(crate) fn user_text_input(group_id: &str, text: &str) -> (InputMessage, Stri
     )
 }
 
+/// An input-less interrupt marker (see
+/// [`TaggedSyntheticKind::Interrupt`]): cancels the thread's in-flight
+/// completion without queuing anything for the model.
+pub(crate) fn interrupt_input(group_id: &str) -> (InputMessage, String) {
+    (
+        InputMessage {
+            content: InputMessageContent::User(UserContent::text("")),
+            group_id: group_id.into(),
+            metadata: None,
+            synthetic: Some(SyntheticKind::Tagged(TaggedSyntheticKind::Interrupt)),
+            display_as: None,
+            subscription: false,
+        },
+        uuid::Uuid::new_v4().to_string(),
+    )
+}
+
 pub(crate) fn tool_result_input(group_id: &str, id: &str, text: &str) -> (InputMessage, String) {
     (
         InputMessage {

crates/infinity-agent-core/src/system/tests.rs

@@ -221,6 +221,108 @@ async fn user_text_interrupts_active_completion() {
         .await;
 }
 
+/// An input-less interrupt marker cancels the in-flight completion without
+/// queuing anything for the model, and the thread accepts new input
+/// afterwards.
+#[tokio::test(flavor = "current_thread")]
+async fn interrupt_input_cancels_completion_without_queuing_input() {
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let (mut running, mut rx, mut ctrl, _conv) = start_system(vec![], None);
+            running
+                .send_user_text(ThreadId::from_ref("t1"), "work on something")
+                .await;
+            let _req = ctrl.next_request().await;
+            ctrl.send_text("partial...");
+            // Wait until the chunk is observed so the completion is in flight.
+            loop {
+                if let Evt::E(AgentEvent::TextChunk { .. }) = next_evt(&mut rx).await {
+                    break;
+                }
+            }
+
+            // The interrupt cancels the stream: the round finishes without
+            // the model ever sending a Final marker.
+            let (interrupt, dedup) = interrupt_input("t1");
+            running.send(interrupt, &dedup).await;
+            loop {
+                if let Evt::E(AgentEvent::CompletionFinished { .. }) = next_evt(&mut rx).await {
+                    break;
+                }
+            }
+            wait_idle(&mut running).await;
+
+            // Nothing was queued for the model: no new completion round ran.
+            assert!(
+                ctrl.try_next_request().is_none(),
+                "an interrupt must not trigger a completion round"
+            );
+
+            // The thread stays alive and accepts new input normally.
+            running
+                .send_user_text(ThreadId::from_ref("t1"), "follow up")
+                .await;
+            let _req2 = ctrl.next_request().await;
+            ctrl.send_text("sure");
+            ctrl.finish();
+            collect_until_finished(&mut rx).await;
+        })
+        .await;
+}
+
+/// An interrupt arriving while the thread waits for an async tool result
+/// settles the pending call with a synthetic "interrupted" result, so the
+/// late real result is dropped instead of restarting generation.
+#[tokio::test(flavor = "current_thread")]
+async fn interrupt_settles_pending_tool_call_so_late_result_is_dropped() {
+    let local = tokio::task::LocalSet::new();
+    local
+        .run_until(async {
+            let (mut running, mut rx, mut ctrl, _conv) =
+                start_system(vec![Box::new(AsyncTool)], None);
+            running
+                .send_user_text(ThreadId::from_ref("t1"), "run the tool")
+                .await;
+            let _req = ctrl.next_request().await;
+            ctrl.send_tool_call("tc-1", "async_tool", serde_json::json!({}));
+            ctrl.finish();
+            collect_until_finished(&mut rx).await;
+
+            // The thread is now parked waiting for the tool result; the
+            // interrupt settles the pending call instead.
+            let (interrupt, dedup) = interrupt_input("t1");
+            running.send(interrupt, &dedup).await;
+            loop {
+                if let Evt::E(AgentEvent::CompletionFinished { .. }) = next_evt(&mut rx).await {
+                    break;
+                }
+            }
+            wait_idle(&mut running).await;
+
+            // The late real result matches an already-settled call: it must
+            // not wake the model.
+            running
+                .send(tool_result_input("t1", "tc-1", "late output").0, "res-late")
+                .await;
+            wait_idle(&mut running).await;
+            assert!(
+                ctrl.try_next_request().is_none(),
+                "a settled tool call's late result must not restart generation"
+            );
+
+            // The thread still accepts new input.
+            running
+                .send_user_text(ThreadId::from_ref("t1"), "next task")
+                .await;
+            let _req2 = ctrl.next_request().await;
+            ctrl.send_text("on it");
+            ctrl.finish();
+            collect_until_finished(&mut rx).await;
+        })
+        .await;
+}
+
 /// A non-user-text input (e.g. a stale tool result) arriving while a
 /// completion is streaming must not interrupt it.
 #[tokio::test(flavor = "current_thread")]