From 7d81cf37327b7a83f72e15f5e868a35ad3340f70 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 9 Jul 2026 23:17:18 +0530 Subject: [PATCH 1/5] fix(web-channel): scope turn cancellation to request_id so a stale cancel can't kill the next turn (#4760) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `cancel_chat` / `channel_web_cancel` cancelled whatever turn was in `IN_FLIGHT[thread]` with no `request_id` scoping (the RPC + socket cancel API carried only client_id + thread_id). So a client that timed out on request A and then sent request B — which supersedes A on the same thread — had A's late, thread-wide cancel tear down B, killing the newer turn at t=0. `channel_web_cancel` also escalated to the thread-wide `cancel_session` when nothing matched, which would tear down a newer turn too. Make cancellation request-scoped: - Add the pure predicate `cancel_should_target(requested, in_flight)`: a scoped cancel fires only when it targets the request actually in flight; an unscoped cancel (None) still stops whatever is running. - Add `cancel_chat_scoped(client_id, thread_id, request_id)` built on it (primary turn + the request-keyed parallel lane). Keep `cancel_chat` (2-arg) as the unscoped "stop everything" back-compat entry, so existing callers/tests are untouched. - `channel_web_cancel` no longer falls through to `cancel_session` on a scoped no-op — a stale cancel that matched nothing is a clean no-op. - Plumb an optional `request_id` from the client through: socket `chat:cancel` payload -> `channel.web_cancel` RPC schema/params -> `cancel_chat_scoped`. Regression: tests/web_cancel_request_scoping.rs. --- src/core/socketio.rs | 8 +- src/openhuman/channels/providers/web/mod.rs | 5 +- src/openhuman/channels/providers/web/ops.rs | 110 +++++++++++++++++- .../channels/providers/web/schemas.rs | 6 +- src/openhuman/channels/providers/web/types.rs | 5 + .../channels_web_startup_raw_coverage_e2e.rs | 4 +- ...tools_network_channels_raw_coverage_e2e.rs | 2 +- tests/web_cancel_request_scoping.rs | 42 +++++++ 8 files changed, 169 insertions(+), 13 deletions(-) create mode 100644 tests/web_cancel_request_scoping.rs diff --git a/src/core/socketio.rs b/src/core/socketio.rs index 82b01dc1e3..1931aa198b 100644 --- a/src/core/socketio.rs +++ b/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, } #[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; }, diff --git a/src/openhuman/channels/providers/web/mod.rs b/src/openhuman/channels/providers/web/mod.rs index 75c340b955..c7f4ae4fbf 100644 --- a/src/openhuman/channels/providers/web/mod.rs +++ b/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; diff --git a/src/openhuman/channels/providers/web/ops.rs b/src/openhuman/channels/providers/web/ops.rs index e82b2b7e5e..05c8ea9651 100644 --- a/src/openhuman/channels/providers/web/ops.rs +++ b/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 { + 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, 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, String> { let client_id = client_id.trim(); let thread_id = thread_id.trim(); @@ -948,14 +1019,34 @@ pub async fn cancel_chat(client_id: &str, thread_id: &str) -> Result cancel_parallel_turn_by_request_id(thread_id, rid).await, + None => cancel_parallel_turns_for_thread(thread_id).await, + }; // Emit a cancelled chat_error for each cancelled turn (primary + parallels) // so every interleaved branch's UI is resolved. @@ -1078,11 +1169,18 @@ pub async fn channel_web_queue_clear(thread_id: &str) -> Result, ) -> Result, 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?; + // A scoped cancel that matched nothing (the request already finished or was + // superseded) is a clean no-op — do NOT fall through to the thread-wide + // `cancel_session`, which would tear down a newer turn on the thread (#4760). + // Only the unscoped stop escalates to the session-level teardown. let cancelled = if cancelled_request_id.is_some() { true + } else if request_id.is_some() { + false } else { crate::openhuman::agent::task_dispatcher::cancel_session(thread_id.trim()).await }; diff --git a/src/openhuman/channels/providers/web/schemas.rs b/src/openhuman/channels/providers/web/schemas.rs index dea632c1ac..b1fa356c64 100644 --- a/src/openhuman/channels/providers/web/schemas.rs +++ b/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) -> ControllerFuture { fn handle_cancel(params: Map) -> ControllerFuture { Box::pin(async move { let p = deserialize_params::(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?) }) } diff --git a/src/openhuman/channels/providers/web/types.rs b/src/openhuman/channels/providers/web/types.rs index 3b6dca9981..9fe1f9100f 100644 --- a/src/openhuman/channels/providers/web/types.rs +++ b/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, } diff --git a/tests/raw_coverage/channels_web_startup_raw_coverage_e2e.rs b/tests/raw_coverage/channels_web_startup_raw_coverage_e2e.rs index 0468759cae..a6d57ad4c9 100644 --- a/tests/raw_coverage/channels_web_startup_raw_coverage_e2e.rs +++ b/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() diff --git a/tests/raw_coverage/tools_network_channels_raw_coverage_e2e.rs b/tests/raw_coverage/tools_network_channels_raw_coverage_e2e.rs index 1577235c10..97cb277dd2 100644 --- a/tests/raw_coverage/tools_network_channels_raw_coverage_e2e.rs +++ b/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); diff --git a/tests/web_cancel_request_scoping.rs b/tests/web_cancel_request_scoping.rs new file mode 100644 index 0000000000..500c07e59f --- /dev/null +++ b/tests/web_cancel_request_scoping.rs @@ -0,0 +1,42 @@ +//! 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")); +} From c9484b920648d9521d6dc4f03ba34f1e52d5b096 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 10 Jul 2026 00:29:09 +0530 Subject: [PATCH 2/5] style: rustfmt web_cancel_request_scoping test --- tests/web_cancel_request_scoping.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/web_cancel_request_scoping.rs b/tests/web_cancel_request_scoping.rs index 500c07e59f..4b79df0fab 100644 --- a/tests/web_cancel_request_scoping.rs +++ b/tests/web_cancel_request_scoping.rs @@ -38,5 +38,8 @@ fn stale_scoped_cancel_does_not_kill_a_newer_turn() { // 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")); + assert!(!cancel_should_target( + Some("old-request"), + "current-request" + )); } From e2dd1d864c87588e3ebae93c33102debb327e733 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 10 Jul 2026 01:08:15 +0530 Subject: [PATCH 3/5] fix(web-channel): report parallel-turn cancellations in scoped-cancel result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cancel_chat_scoped returned only the primary in-flight request id, so a scoped cancel that matched only a parallel (forked) turn — which still cancelled a turn and emitted its cancelled event — returned Ok(None). channel_web_cancel then misreported cancelled: false with a null id even though a turn was genuinely torn down. Fold the cancelled parallel id into the returned value so the RPC ack matches the emitted event. Addresses @coderabbitai on src/openhuman/channels/providers/web/ops.rs. --- src/openhuman/channels/providers/web/ops.rs | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/openhuman/channels/providers/web/ops.rs b/src/openhuman/channels/providers/web/ops.rs index 05c8ea9651..9111b059d7 100644 --- a/src/openhuman/channels/providers/web/ops.rs +++ b/src/openhuman/channels/providers/web/ops.rs @@ -1048,9 +1048,18 @@ pub async fn cancel_chat_scoped( 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(), @@ -1062,7 +1071,7 @@ pub async fn cancel_chat_scoped( }); } - Ok(removed_request_id) + Ok(cancelled_any) } pub async fn channel_web_chat( From 22b89fbd18fcb5f92318f80d265a0c49fe4105aa Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Fri, 10 Jul 2026 01:08:23 +0530 Subject: [PATCH 4/5] fix(task-dispatcher): request-scope the autonomous-run cancel fallback A scoped web cancel (request_id set) that targeted an autonomous task-dispatcher run was dropped: those runs aren't web-channel turns, so cancel_chat_scoped returned None and channel_web_cancel's `else if request_id.is_some() => false` branch suppressed the cancel_session fallback, leaving the run alive until completion. Add cancel_session_scoped: when a request_id is given it aborts the active run only if its run_id matches, so scoped cancels now reach detached runs while a stale cancel for a superseded request stays a no-op (#4760). An unscoped stop is unchanged. channel_web_cancel routes its fallback through it. Adds unit tests for the match / mismatch / no-run paths. Addresses @chatgpt-codex-connector on src/openhuman/channels/providers/web/ops.rs. --- src/openhuman/agent/task_dispatcher/mod.rs | 2 +- .../agent/task_dispatcher/registry.rs | 25 ++++++ src/openhuman/agent/task_dispatcher/tests.rs | 87 ++++++++++++++++++- src/openhuman/channels/providers/web/ops.rs | 18 ++-- 4 files changed, 123 insertions(+), 9 deletions(-) diff --git a/src/openhuman/agent/task_dispatcher/mod.rs b/src/openhuman/agent/task_dispatcher/mod.rs index 75fbabe1f9..ffefcf52b9 100644 --- a/src/openhuman/agent/task_dispatcher/mod.rs +++ b/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 diff --git a/src/openhuman/agent/task_dispatcher/registry.rs b/src/openhuman/agent/task_dispatcher/registry.rs index abc522fb4f..4dba63ccda 100644 --- a/src/openhuman/agent/task_dispatcher/registry.rs +++ b/src/openhuman/agent/task_dispatcher/registry.rs @@ -72,3 +72,28 @@ pub async fn cancel_session(thread_id: &str) -> bool { ); true } + +/// 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 { + if let Some(rid) = request_id { + // Peek (don't remove) so a non-matching scoped cancel leaves the run + // intact for the request that actually owns it. + let matches = active_runs() + .lock() + .expect("active_runs mutex poisoned") + .get(thread_id) + .map(|run| run.run_id == rid) + .unwrap_or(false); + if !matches { + return false; + } + } + cancel_session(thread_id).await +} diff --git a/src/openhuman/agent/task_dispatcher/tests.rs b/src/openhuman/agent/task_dispatcher/tests.rs index 398494164f..f82715a7ad 100644 --- a/src/openhuman/agent/task_dispatcher/tests.rs +++ b/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(), diff --git a/src/openhuman/channels/providers/web/ops.rs b/src/openhuman/channels/providers/web/ops.rs index 9111b059d7..1c838cb339 100644 --- a/src/openhuman/channels/providers/web/ops.rs +++ b/src/openhuman/channels/providers/web/ops.rs @@ -1182,16 +1182,20 @@ pub async fn channel_web_cancel( ) -> Result, String> { let cancelled_request_id = cancel_chat_scoped(client_id, thread_id, request_id).await?; - // A scoped cancel that matched nothing (the request already finished or was - // superseded) is a clean no-op — do NOT fall through to the thread-wide - // `cancel_session`, which would tear down a newer turn on the thread (#4760). - // Only the unscoped stop escalates to the session-level teardown. + // 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 if request_id.is_some() { - false } 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( From 6075e07f0966a1f955a6f6a8299d74b230aa08e4 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Mon, 13 Jul 2026 17:54:49 +0530 Subject: [PATCH 5/5] fix(task_dispatcher): make scoped cancel match+remove atomic and log no-op branches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses two review findings on the request-scoped cancel path: - Critical TOCTOU: the run_id match and the removal ran under two separate lock acquisitions, so a matched run could complete and be replaced by a newer run in between — cancel_session_scoped would then cancel the *new* run, the exact #4760 bug this PR prevents. New take_active_run_if holds the lock across the match and the remove; cancel_session's side effects are factored into cancel_taken_run so the scoped path cancels the exact run it removed instead of re-acquiring the lock and racing a replacement. - Diagnostics: both scoped no-op cases (no active run / run_id mismatch) now emit grep-friendly debug logs, matching the success path's tracing. Existing tests cover all three branches (no run / mismatch / match). --- .../agent/task_dispatcher/registry.rs | 78 +++++++++++++++---- 1 file changed, 63 insertions(+), 15 deletions(-) diff --git a/src/openhuman/agent/task_dispatcher/registry.rs b/src/openhuman/agent/task_dispatcher/registry.rs index 4dba63ccda..375f369397 100644 --- a/src/openhuman/agent/task_dispatcher/registry.rs +++ b/src/openhuman/agent/task_dispatcher/registry.rs @@ -32,6 +32,46 @@ pub(super) fn take_active_run(thread_id: &str) -> Option { .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 { + 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,7 +123,6 @@ pub async fn cancel_session(thread_id: &str) -> bool { run_id = %run.run_id, "[task_dispatcher] cancelled autonomous run via chat cancel" ); - true } /// Request-scoped variant of [`cancel_session`]. @@ -82,18 +134,14 @@ pub async fn cancel_session(thread_id: &str) -> bool { /// 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 { - if let Some(rid) = request_id { - // Peek (don't remove) so a non-matching scoped cancel leaves the run - // intact for the request that actually owns it. - let matches = active_runs() - .lock() - .expect("active_runs mutex poisoned") - .get(thread_id) - .map(|run| run.run_id == rid) - .unwrap_or(false); - if !matches { - return false; - } - } - cancel_session(thread_id).await + // 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 }