fix(web-channel): scope turn cancellation to request_id so a stale cancel cannot kill the next turn (#4760) (#4765)

src/core/socketio.rs

@@ -363,6 +363,11 @@ struct ChatStartPayload {
 #[derive(Debug, Deserialize)]
 struct ChatCancelPayload {
     thread_id: String,
+    /// The request this cancel targets. When the client passes the id of the
+    /// turn it started, the cancel is scoped to that turn so a late cancel for a
+    /// timed-out request can't kill the next turn on the thread (#4760).
+    #[serde(default)]
+    request_id: Option<String>,
 }
 
 #[derive(Debug, Deserialize)]
@@ -551,9 +556,10 @@ pub fn attach_socketio() -> (socketioxide::layer::SocketIoLayer, SocketIo) {
                         client_id,
                         payload.thread_id
                     );
-                    let _ = crate::openhuman::channels::providers::web::cancel_chat(
+                    let _ = crate::openhuman::channels::providers::web::cancel_chat_scoped(
                         &client_id,
                         &payload.thread_id,
+                        payload.request_id.as_deref(),
                     )
                     .await;
                 },

src/openhuman/agent/task_dispatcher/mod.rs

@@ -33,7 +33,7 @@ mod tests;
 pub use dispatch::dispatch_card;
 pub use poller::start_board_poller;
 pub use prompt::build_task_prompt;
-pub use registry::cancel_session;
+pub use registry::{cancel_session, cancel_session_scoped};
 pub use types::DispatchOutcome;
 
 /// Run a one-off **system** agent turn on an existing chat thread, streaming the

src/openhuman/agent/task_dispatcher/registry.rs

@@ -32,6 +32,46 @@ pub(super) fn take_active_run(thread_id: &str) -> Option<ActiveRun> {
         .remove(thread_id)
 }
 
+/// Atomically remove the active-run entry for `thread_id`, but only when it
+/// matches `request_id` (when `Some`).
+///
+/// The match check and the removal happen under a single lock acquisition, so
+/// there is no window in which the matched run could complete and be replaced
+/// by a newer run before removal — the "stale cancel kills a newer turn" race a
+/// separate peek-then-`take_active_run` would reopen (#4760). A `None`
+/// `request_id` removes whatever run is on the thread (unscoped Stop /
+/// teardown).
+///
+/// Both scoped no-op cases (no active run, or a `run_id` mismatch from a
+/// superseded/unrelated request) emit grep-friendly `debug` diagnostics so an
+/// intentional no-op cancel is still traceable.
+pub(super) fn take_active_run_if(thread_id: &str, request_id: Option<&str>) -> Option<ActiveRun> {
+    let mut guard = active_runs().lock().expect("active_runs mutex poisoned");
+    if let Some(rid) = request_id {
+        match guard.get(thread_id) {
+            None => {
+                tracing::debug!(
+                    thread_id = %thread_id,
+                    request_id = %rid,
+                    "[task_dispatcher] scoped cancel ignored: no active run on thread"
+                );
+                return None;
+            }
+            Some(run) if run.run_id != rid => {
+                tracing::debug!(
+                    thread_id = %thread_id,
+                    request_id = %rid,
+                    active_run_id = %run.run_id,
+                    "[task_dispatcher] scoped cancel ignored: run_id mismatch (superseded/unrelated request)"
+                );
+                return None;
+            }
+            _ => {}
+        }
+    }
+    guard.remove(thread_id)
+}
+
 /// Cancel the in-flight autonomous run streaming into session `thread_id`.
 ///
 /// Aborts the detached run task, stops its heartbeat, marks the card `blocked`
@@ -43,6 +83,19 @@ pub async fn cancel_session(thread_id: &str) -> bool {
     let Some(run) = take_active_run(thread_id) else {
         return false;
     };
+    cancel_taken_run(thread_id, run);
+    true
+}
+
+/// Drive the cancellation side effects for a run that has **already** been
+/// removed from the registry: abort the task, stop its heartbeat, write the
+/// card back to a terminal state (the aborted task never reaches its own
+/// write-back), and emit the terminal `chat_error` event.
+///
+/// Split out of [`cancel_session`] so [`cancel_session_scoped`] can cancel the
+/// exact run it atomically removed via [`take_active_run_if`], rather than
+/// re-acquiring the lock and racing a replacement run (#4760).
+fn cancel_taken_run(thread_id: &str, run: ActiveRun) {
     run.abort.abort();
     let _ = run.hb_cancel.send(true);
     // The aborted task never reaches its own write-back — do it here so the
@@ -70,5 +123,25 @@ pub async fn cancel_session(thread_id: &str) -> bool {
         run_id = %run.run_id,
         "[task_dispatcher] cancelled autonomous run via chat cancel"
     );
+}
+
+/// Request-scoped variant of [`cancel_session`].
+///
+/// When `request_id` is `Some`, the active run is aborted only if its `run_id`
+/// matches — a scoped cancel for a superseded or unrelated request is a no-op so
+/// it can't tear down a newer autonomous run on the thread (#4760). When
+/// `request_id` is `None`, this behaves exactly like [`cancel_session`] (stop
+/// whatever run is on the thread — the Stop button / session-teardown path).
+/// Returns `true` if a run was found and cancelled.
+pub async fn cancel_session_scoped(thread_id: &str, request_id: Option<&str>) -> bool {
+    // Atomic match + remove: holding the lock across both closes the TOCTOU
+    // window where the matched run could complete and be replaced by a newer run
+    // before we remove it, which would cancel the *new* run — the exact #4760
+    // bug this path guards against. `take_active_run_if` also logs the scoped
+    // no-op cases (no active run / run_id mismatch).
+    let Some(run) = take_active_run_if(thread_id, request_id) else {
+        return false;
+    };
+    cancel_taken_run(thread_id, run);
     true
 }

src/openhuman/agent/task_dispatcher/tests.rs

@@ -8,7 +8,7 @@ use crate::openhuman::todos::ops::{self, BoardLocation, CardPatch};
 use super::executor::{truncate_chars, write_back, EVIDENCE_MAX_CHARS};
 use super::poller::{pick_next_todo, requires_plan_approval};
 use super::prompt::{build_progress_instruction, build_task_prompt};
-use super::registry::{register_active_run, take_active_run};
+use super::registry::{cancel_session_scoped, register_active_run, take_active_run};
 use super::types::ActiveRun;
 
 #[tokio::test]
@@ -36,6 +36,91 @@ async fn active_run_registry_take_is_once() {
     handle.abort();
 }
 
+#[tokio::test]
+async fn cancel_session_scoped_returns_false_without_a_run() {
+    // No autonomous run on the thread — scoped and unscoped both no-op.
+    assert!(!cancel_session_scoped("no-such-thread-xyz", Some("r1")).await);
+    assert!(!cancel_session_scoped("no-such-thread-xyz", None).await);
+}
+
+#[tokio::test]
+async fn cancel_session_scoped_ignores_a_mismatched_request() {
+    // #4760: a scoped cancel whose request_id does not match the in-flight run's
+    // run_id must NOT abort it — the run survives for the request that owns it,
+    // so a stale cancel for a superseded request can't kill a newer run.
+    let (tx, _rx) = tokio::sync::watch::channel(false);
+    let handle = tokio::spawn(async { std::future::pending::<()>().await });
+    let key = "task-cancel-scoped-mismatch-test";
+    register_active_run(
+        key.to_string(),
+        ActiveRun {
+            abort: handle.abort_handle(),
+            hb_cancel: tx,
+            location: BoardLocation::Scratch,
+            card_id: "c1".to_string(),
+            run_id: "r1".to_string(),
+        },
+    );
+    assert!(
+        !cancel_session_scoped(key, Some("some-other-request")).await,
+        "a scoped cancel naming a different request must be a no-op"
+    );
+    assert!(
+        take_active_run(key).is_some(),
+        "the mismatched scoped cancel must leave the run in flight"
+    );
+    handle.abort();
+}
+
+#[tokio::test]
+async fn cancel_session_scoped_aborts_the_run_when_the_request_matches() {
+    // The matching-request path: a scoped cancel that names the running turn
+    // tears it down and lands the card in a terminal (blocked) state.
+    let dir = tempfile::tempdir().unwrap();
+    let loc = board_loc(dir.path());
+    let id = ops::add(&loc, "run me", CardPatch::default())
+        .unwrap()
+        .cards[0]
+        .id
+        .clone();
+    ops::update_status(&loc, &id, TaskCardStatus::InProgress).unwrap();
+
+    let (tx, _rx) = tokio::sync::watch::channel(false);
+    let handle = tokio::spawn(async { std::future::pending::<()>().await });
+    let key = "task-cancel-scoped-match-test";
+    register_active_run(
+        key.to_string(),
+        ActiveRun {
+            abort: handle.abort_handle(),
+            hb_cancel: tx,
+            location: loc.clone(),
+            card_id: id.clone(),
+            run_id: "r1".to_string(),
+        },
+    );
+
+    assert!(
+        cancel_session_scoped(key, Some("r1")).await,
+        "a scoped cancel naming the running request must abort it"
+    );
+    assert!(
+        take_active_run(key).is_none(),
+        "the cancelled run is removed from the registry"
+    );
+    let card = ops::list(&loc)
+        .unwrap()
+        .cards
+        .into_iter()
+        .find(|c| c.id == id)
+        .unwrap();
+    assert_eq!(
+        card.status,
+        TaskCardStatus::Blocked,
+        "cancellation lands the card in a terminal state, not stale in_progress"
+    );
+    handle.abort();
+}
+
 fn card(objective: Option<&str>) -> TaskBoardCard {
     TaskBoardCard {
         id: "task-1".into(),

src/openhuman/channels/providers/web/mod.rs

@@ -35,8 +35,9 @@ pub use event_bus::fresh_approval_surface_subscription;
 #[cfg(any(test, debug_assertions))]
 pub use ops::parallel_in_flight_entries_for_test;
 pub use ops::{
-    cancel_chat, channel_web_cancel, channel_web_chat, channel_web_queue_clear,
-    channel_web_queue_status, in_flight_entries_for_test, invalidate_thread_sessions, start_chat,
+    cancel_chat, cancel_chat_scoped, cancel_should_target, channel_web_cancel, channel_web_chat,
+    channel_web_queue_clear, channel_web_queue_status, in_flight_entries_for_test,
+    invalidate_thread_sessions, start_chat,
 };
 pub use types::ChatRequestMetadata;
 

src/openhuman/channels/providers/web/ops.rs

@@ -932,7 +932,78 @@ pub async fn parallel_in_flight_entries_for_test() -> Vec<(String, String)> {
         .collect()
 }
 
+/// Whether a cancel request should tear down the turn currently in flight for a
+/// thread.
+///
+/// `requested` is the `request_id` the caller is cancelling; `None` means an
+/// unscoped stop ("cancel whatever is running", e.g. a Stop button or a session
+/// teardown). `in_flight` is the `request_id` currently registered for the
+/// thread.
+///
+/// A *scoped* cancel matches only its own request. This is the fix for #4760: a
+/// client that times out on request A and then sends request B — which
+/// supersedes A on the same thread — must not have A's late-arriving cancel tear
+/// down B. Scoping the cancel to A makes it a no-op once B is in flight, so the
+/// newer turn survives instead of being killed at t=0.
+pub fn cancel_should_target(requested: Option<&str>, in_flight: &str) -> bool {
+    match requested {
+        Some(rid) => rid == in_flight,
+        None => true,
+    }
+}
+
+/// Cancel a single parallel (forked) turn identified by `request_id`, but only
+/// when it belongs to `thread_id`. Returns the cancelled id (as a one-element
+/// vec, mirroring [`cancel_parallel_turns_for_thread`]) or empty when no such
+/// parallel turn exists. Request-scoped cancel path (#4760).
+async fn cancel_parallel_turn_by_request_id(thread_id: &str, request_id: &str) -> Vec<String> {
+    let mut parallel = PARALLEL_IN_FLIGHT.lock().await;
+    let matches = parallel
+        .get(request_id)
+        .map(|entry| entry.thread_id == thread_id)
+        .unwrap_or(false);
+    if !matches {
+        return Vec::new();
+    }
+    if let Some(entry) = parallel.remove(request_id) {
+        entry.cancel_token.cancel();
+        let mut handle = entry.handle;
+        tokio::spawn(async move {
+            tokio::select! {
+                _ = &mut handle => {}
+                _ = tokio::time::sleep(Duration::from_secs(5)) => {
+                    handle.abort();
+                }
+            }
+        });
+        return vec![request_id.to_string()];
+    }
+    Vec::new()
+}
+
+/// Cancel whatever turn is currently running on a thread (unscoped stop).
+///
+/// Back-compat entry point (Stop button / session teardown). For a cancel that
+/// must only affect a specific turn — so a stale cancel can't kill a newer turn
+/// on the same thread — use [`cancel_chat_scoped`] with the target `request_id`
+/// (#4760).
 pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<String>, String> {
+    cancel_chat_scoped(client_id, thread_id, None).await
+}
+
+/// Cancel the in-flight turn(s) for a thread.
+///
+/// When `request_id` is `Some`, the cancel is **scoped**: it only tears down the
+/// primary turn if that exact request is still running (and only the matching
+/// parallel turn), so a stale cancel for a superseded request can't kill the
+/// newer turn that replaced it (#4760). When `request_id` is `None`, it stops
+/// whatever is running on the thread (primary + every parallel) — the "stop
+/// everything" behaviour used by session teardown / a Stop button.
+pub async fn cancel_chat_scoped(
+    client_id: &str,
+    thread_id: &str,
+    request_id: Option<&str>,
+) -> Result<Option<String>, String> {
     let client_id = client_id.trim();
     let thread_id = thread_id.trim();
 
@@ -948,18 +1019,47 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<Stri
 
     {
         let mut in_flight = IN_FLIGHT.lock().await;
-        if let Some(existing) = in_flight.remove(&map_key) {
-            removed_request_id = Some(cancel_in_flight_gracefully(existing));
+        // #4760: only tear down the primary turn when the cancel is unscoped OR
+        // targets exactly the request that is running. A stale cancel for an
+        // already-superseded request must be a no-op so the newer turn lives.
+        let should_cancel_primary = in_flight
+            .get(&map_key)
+            .map(|entry| cancel_should_target(request_id, &entry.request_id))
+            .unwrap_or(false);
+        if should_cancel_primary {
+            if let Some(existing) = in_flight.remove(&map_key) {
+                removed_request_id = Some(cancel_in_flight_gracefully(existing));
+            }
+        } else if let Some(rid) = request_id {
+            log::info!(
+                "[web-channel] ignoring stale cancel request_id={} for thread_id={} — current in-flight is {:?}; newer turn preserved",
+                rid,
+                thread_id,
+                in_flight.get(&map_key).map(|e| e.request_id.as_str())
+            );
         }
     }
 
-    // Also tear down any concurrent parallel (forked) turns on the thread so a
-    // cancel/stop covers every in-flight turn, not just the primary one.
-    let cancelled_parallel = cancel_parallel_turns_for_thread(thread_id).await;
+    // Also tear down concurrent parallel (forked) turns. A scoped cancel targets
+    // only the named parallel turn (if it is one); an unscoped cancel/stop
+    // covers every parallel turn on the thread, not just the primary one.
+    let cancelled_parallel = match request_id {
+        Some(rid) => cancel_parallel_turn_by_request_id(thread_id, rid).await,
+        None => cancel_parallel_turns_for_thread(thread_id).await,
+    };
+
+    // #4760: a scoped cancel that matched only a parallel (forked) turn — not the
+    // primary — still genuinely tore a turn down and emitted its cancelled event.
+    // Surface that id so `channel_web_cancel` reports `cancelled: true` with the
+    // right request_id instead of misreporting a no-op just because the primary
+    // turn wasn't the one cancelled.
+    let cancelled_any = removed_request_id
+        .clone()
+        .or_else(|| cancelled_parallel.first().cloned());
 
     // Emit a cancelled chat_error for each cancelled turn (primary + parallels)
     // so every interleaved branch's UI is resolved.
-    for request_id in removed_request_id.iter().cloned().chain(cancelled_parallel) {
+    for request_id in removed_request_id.into_iter().chain(cancelled_parallel) {
         publish_web_channel_event(WebChannelEvent {
             event: "chat_error".to_string(),
             client_id: client_id.to_string(),
@@ -971,7 +1071,7 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result<Option<Stri
         });
     }
 
-    Ok(removed_request_id)
+    Ok(cancelled_any)
 }
 
 pub async fn channel_web_chat(
@@ -1078,13 +1178,24 @@ pub async fn channel_web_queue_clear(thread_id: &str) -> Result<RpcOutcome<Value
 pub async fn channel_web_cancel(
     client_id: &str,
     thread_id: &str,
+    request_id: Option<&str>,
 ) -> Result<RpcOutcome<Value>, String> {
-    let cancelled_request_id = cancel_chat(client_id, thread_id).await?;
+    let cancelled_request_id = cancel_chat_scoped(client_id, thread_id, request_id).await?;
 
+    // No web-channel turn matched. Fall through to the task-dispatcher registry,
+    // which holds autonomous runs that are NOT web-channel turns (so they never
+    // appear in IN_FLIGHT and can only be reached here). The fallback is itself
+    // request-scoped: a scoped cancel aborts the run only when its run_id
+    // matches, so a stale cancel for a superseded request can't tear down a newer
+    // run on the thread (#4760); an unscoped stop aborts whatever run is running.
     let cancelled = if cancelled_request_id.is_some() {
         true
     } else {
-        crate::openhuman::agent::task_dispatcher::cancel_session(thread_id.trim()).await
+        crate::openhuman::agent::task_dispatcher::cancel_session_scoped(
+            thread_id.trim(),
+            request_id,
+        )
+        .await
     };
 
     Ok(RpcOutcome::single_log(

src/openhuman/channels/providers/web/schemas.rs

@@ -73,6 +73,10 @@ pub fn schemas(function: &str) -> ControllerSchema {
             inputs: vec![
                 required_string("client_id", "Client stream identifier."),
                 required_string("thread_id", "Thread identifier."),
+                optional_string(
+                    "request_id",
+                    "Request id to cancel. When set, only that turn is cancelled (a stale cancel for a superseded request is ignored so the newer turn survives). Omit to stop whatever is running on the thread.",
+                ),
             ],
             outputs: vec![json_output("ack", "Cancellation payload.")],
         },
@@ -149,7 +153,7 @@ fn handle_queue_clear(params: Map<String, Value>) -> ControllerFuture {
 fn handle_cancel(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move {
         let p = deserialize_params::<WebCancelParams>(params)?;
-        to_json(channel_web_cancel(&p.client_id, &p.thread_id).await?)
+        to_json(channel_web_cancel(&p.client_id, &p.thread_id, p.request_id.as_deref()).await?)
     })
 }
 

src/openhuman/channels/providers/web/types.rs

@@ -148,4 +148,9 @@ pub(super) struct WebQueueParams {
 pub(super) struct WebCancelParams {
     pub(super) client_id: String,
     pub(super) thread_id: String,
+    /// The `request_id` this cancel targets. When present, the cancel is scoped
+    /// to that exact turn, so a stale cancel for a superseded request can't kill
+    /// the newer turn on the thread (#4760). Absent = stop whatever is running.
+    #[serde(default)]
+    pub(super) request_id: Option<String>,
 }

tests/raw_coverage/channels_web_startup_raw_coverage_e2e.rs

@@ -128,7 +128,7 @@ async fn web_controllers_validate_inputs_and_emit_structured_forced_errors() {
     .expect_err("blank messages are rejected");
     assert!(err.contains("message is required"));
 
-    let cancel = channel_web_cancel("client", "missing-thread")
+    let cancel = channel_web_cancel("client", "missing-thread", None)
         .await
         .expect("cancel without in-flight request is ok")
         .into_cli_compatible_json()
@@ -201,7 +201,7 @@ async fn web_chat_cancel_aborts_in_flight_thread_without_real_provider() {
     .await
     .expect("start chat");
 
-    let cancel = channel_web_cancel("cancel-client", "cancel-thread")
+    let cancel = channel_web_cancel("cancel-client", "cancel-thread", None)
         .await
         .expect("cancel")
         .into_cli_compatible_json()

tests/raw_coverage/tools_network_channels_raw_coverage_e2e.rs

@@ -578,7 +578,7 @@ async fn web_channel_public_paths_cover_validation_cancel_schema_and_event_bus()
         .expect("no in-flight cancel");
     assert_eq!(none, None);
 
-    let outcome = channel_web_cancel(" client ", " round15-thread ")
+    let outcome = channel_web_cancel(" client ", " round15-thread ", None)
         .await
         .expect("cancel rpc outcome");
     assert_eq!(outcome.value["cancelled"], false);

tests/web_cancel_request_scoping.rs

@@ -0,0 +1,45 @@
+//! Request-scoped web-chat cancellation (#4760).
+//!
+//! Regression for a real in-flight-teardown bug: a client that timed out on
+//! request A and then sent request B — which supersedes A on the same thread —
+//! had A's late-arriving cancel tear down B, killing the newer turn at t=0
+//! ("Cancelled by newer request"). The root cause was that `cancel_chat`
+//! cancelled *whatever* turn was in flight for the thread, unscoped by
+//! `request_id`.
+//!
+//! The fix routes the cancel decision through `cancel_should_target`: a scoped
+//! cancel (`Some(request_id)`) only fires when it matches the turn actually in
+//! flight, so a stale cancel for a superseded request becomes a no-op and the
+//! newer turn survives. An unscoped cancel (`None` — Stop button / session
+//! teardown) still stops whatever is running.
+//!
+//! These assertions pin that decision. `cancel_should_target` is a pure
+//! predicate, so this is a fast, deterministic unit test with no runtime setup.
+
+use openhuman_core::openhuman::channels::web::cancel_should_target;
+
+#[test]
+fn unscoped_cancel_always_targets_the_in_flight_turn() {
+    // No request_id => "stop whatever is running on this thread" (the Stop
+    // button / session-teardown path). It always targets the in-flight turn.
+    assert!(cancel_should_target(None, "req-A"));
+    assert!(cancel_should_target(None, "req-B"));
+}
+
+#[test]
+fn scoped_cancel_fires_for_its_own_request() {
+    // A client cancelling exactly the turn it started tears that turn down.
+    assert!(cancel_should_target(Some("req-A"), "req-A"));
+}
+
+#[test]
+fn stale_scoped_cancel_does_not_kill_a_newer_turn() {
+    // The #4760 bug: request A timed out client-side and B is now the in-flight
+    // turn on this thread. A's late, request-scoped cancel must NOT tear down B.
+    assert!(!cancel_should_target(Some("req-A"), "req-B"));
+    // Symmetric: a cancel naming an already-finished request is inert.
+    assert!(!cancel_should_target(
+        Some("old-request"),
+        "current-request"
+    ));
+}