From 8ca1d875e5451b8a50b5c5068794b2338150604f Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Mon, 13 Jul 2026 17:26:51 +0530 Subject: [PATCH 1/3] fix(approval): RAII guard so a parked waiter is cleaned up on external turn teardown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ApprovalGate::intercept_audited_inner registers an in-memory waiter and persists a pending_approvals row, then parks on tokio::time::timeout. All cleanup (evict_waiter / store::decide(Deny) / thread+meeting routing-map removal) lived only inside the timeout match arms, so it ran only on a normal park resolution. Since #4751 (the #4746 wall-clock backstop), a turn future can be torn down externally while a tool call is parked. Dropping the future skips the match arms, leaking the waiter, the routing mappings, and the pending row until the store TTL sweeps them — a later yes/no then routes to a dead request and returns without starting a fresh turn. Add a WaiterGuard RAII guard created just before the park await: on Drop (external cancellation) it evicts the waiter, clears routing, and denies the still-open row (store::decide is WHERE decided_at IS NULL, so a same-instant real decision is honored). Disarmed on every normal exit so the match arms stay authoritative. Regression test tears the parked future down mid-park and asserts waiter eviction + routing clear + row denied. Closes #4774 --- src/openhuman/approval/gate.rs | 146 +++++++++++++++++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index e074b3dd5b..1d8fb6bedb 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -203,6 +203,64 @@ pub struct ApprovalGate { meeting_to_request: Mutex>, } +/// RAII guard that tears the parked waiter down even when the surrounding turn +/// future is dropped mid-park. +/// +/// `intercept_audited_inner` only runs its cleanup (`evict_waiter` / +/// `store::decide(Deny)` / routing-map removal) inside the +/// `tokio::time::timeout(...).await` match arms — i.e. only when the park +/// resolves *normally*. Once a turn future can be torn down *externally* — the +/// harness `max_wall_clock_ms` backstop (#4746) or the outer web backstop +/// (#4751) firing while a tool call is parked — dropping the future skips those +/// arms entirely, leaving the in-memory waiter, the thread/meeting routing +/// mappings, and the `pending_approvals` row dangling until the store TTL +/// sweeps them. A later yes/no arriving before that expiry would then route to a +/// dead request and return without starting a fresh turn (#4774). +/// +/// The guard is created just before the park await and [`disarm`](Self::disarm)ed +/// on every normal exit (the match arm already ran the exact teardown for its +/// outcome), so its `Drop` fires *only* on external cancellation. +struct WaiterGuard<'a> { + gate: &'a ApprovalGate, + request_id: String, + thread_id: Option, + meeting_key: Option, + armed: bool, +} + +impl WaiterGuard<'_> { + /// Mark the park as resolved normally so `Drop` becomes a no-op. + fn disarm(&mut self) { + self.armed = false; + } +} + +impl Drop for WaiterGuard<'_> { + fn drop(&mut self) { + if !self.armed { + return; + } + // External teardown: the normal cleanup was skipped. Evict the waiter, + // drop the routing mappings so a later chat/voice reply is not + // mis-routed to this now-dead request, and deny the still-open pending + // row. `store::decide` is `WHERE decided_at IS NULL`, so a decision that + // committed in the same instant is honored rather than overwritten. + self.gate.evict_waiter(&self.request_id); + if let Some(thread_id) = &self.thread_id { + self.gate.thread_to_request.lock().remove(thread_id); + } + if let Some(meeting_key) = &self.meeting_key { + self.gate.meeting_to_request.lock().remove(meeting_key); + } + let _ = store::decide(&self.gate.config, &self.request_id, ApprovalDecision::Deny); + tracing::warn!( + request_id = %self.request_id, + "[approval::gate] parked approval future dropped mid-park (external turn teardown) — \ + evicted waiter, cleared routing, denied pending row (#4774)" + ); + } +} + impl ApprovalGate { /// Install the process-global gate. Returns the existing gate if /// one was already installed (re-install is a no-op so repeated @@ -816,6 +874,19 @@ impl ApprovalGate { None => effective_ttl, }; + // RAII cleanup for external teardown (#4774): if the turn future is + // dropped while parked on the await below (the #4746/#4751 wall-clock + // backstop firing), the match arms never run, so this guard evicts the + // waiter, clears routing, and denies the pending row on drop. Disarmed + // right after the match on every normal exit. + let mut waiter_guard = WaiterGuard { + gate: self, + request_id: request_id.clone(), + thread_id: chat_thread_id.clone(), + meeting_key: in_call_ctx.as_ref().map(|ic| ic.meeting_key.clone()), + armed: true, + }; + let outcome = match tokio::time::timeout(wait, rx).await { Ok(Ok(decision)) => { tracing::info!( @@ -941,6 +1012,10 @@ impl ApprovalGate { } } }; + // Reached only on a normal park resolution: the match arm above already + // ran the exact teardown for its outcome, so disarm the RAII guard (its + // Drop is reserved for external cancellation — see `WaiterGuard`). + waiter_guard.disarm(); // The routing mappings are only needed while parked; clear them on // every exit (decision, channel drop, or timeout). self.clear_thread(&chat_thread_id); @@ -1610,6 +1685,77 @@ mod tests { assert!(gate.pending_for_thread("thread-42").is_none()); } + #[tokio::test] + async fn waiter_future_dropped_mid_park_evicts_waiter_clears_routing_and_denies_row() { + // #4774: once a turn future can be torn down *externally* (the #4746 + // harness wall-clock backstop / #4751 outer web backstop firing while a + // tool call is parked), dropping the intercept future must not leak the + // waiter, the thread→request routing mapping, or the still-open pending + // row. The `WaiterGuard` Drop impl runs the cleanup the timeout match + // arms would otherwise own. + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + // Build the parked future with the WebChat origin + chat context scoped, + // exactly like the production web channel caller — but drive it locally + // so we can drop it mid-park instead of resolving it. + let g = gate.clone(); + // `Box::pin` (not `tokio::pin!`) so `drop(fut)` below drops the *future + // itself* — and thus the `WaiterGuard` saved in its async state — rather + // than just a `Pin<&mut _>` reference. + let mut fut = Box::pin(turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( + chat_ctx(), + g.intercept("shell", "run rm", serde_json::json!({})), + ), + )); + + // Poll it just long enough to register the waiter, persist the pending + // row, and park on the TTL timeout. Nothing resolves it, so the outer + // timeout must elapse with the future still pending. + let parked = tokio::time::timeout(Duration::from_millis(200), &mut fut).await; + assert!( + parked.is_err(), + "future should still be parked, not resolved" + ); + + // Capture the request_id from the routing mapping while parked, and + // confirm the waiter + pending row exist before teardown. + let request_id = gate + .pending_for_thread("t-test") + .expect("thread→request mapping must exist while parked"); + assert!( + gate.waiters.lock().contains_key(&request_id), + "waiter must be registered while parked" + ); + assert!( + matches!(store::get_decision(&gate.config, &request_id), Ok(None)), + "pending row must be open (undecided) while parked" + ); + + // External teardown: the wall-clock backstop tears the turn future down + // mid-park. This skips the timeout match arms entirely. + drop(fut); + + // The RAII guard must have run the cleanup on drop. + assert!( + !gate.waiters.lock().contains_key(&request_id), + "waiter must be evicted when the parked future is dropped" + ); + assert!( + gate.pending_for_thread("t-test").is_none(), + "thread→request routing must be cleared on external teardown" + ); + assert!( + matches!( + store::get_decision(&gate.config, &request_id), + Ok(Some(ApprovalDecision::Deny)) + ), + "pending row must be denied when the parked future is dropped" + ); + } + // ── caller park bound (issue #4756) ────────────────────────────── // // A caller (composio_connect) can cap the park via From c0764919e26d9d703992e2f0f74ae84bc308df34 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Mon, 13 Jul 2026 17:29:49 +0530 Subject: [PATCH 2/3] fix(GH-4774): finalize changes --- vendor/tinyagents | 2 +- vendor/tinycortex | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/vendor/tinyagents b/vendor/tinyagents index b34b2f83d7..b1f0d82d18 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit b34b2f83d7af492c9b7ccaf558c38bece4182dde +Subproject commit b1f0d82d1845d5e3de020882b8817565f529cad9 diff --git a/vendor/tinycortex b/vendor/tinycortex index 671e78a014..33dda94305 160000 --- a/vendor/tinycortex +++ b/vendor/tinycortex @@ -1 +1 @@ -Subproject commit 671e78a01411de5bcda8f3d1816ac6c67485d694 +Subproject commit 33dda943053e61ef585fc39647cf1854344b6323 From fabfb17c25856e3ecb4e41c65b69e6f2c6c449a7 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Mon, 13 Jul 2026 17:43:40 +0530 Subject: [PATCH 3/3] fix(approval): scope waiter-guard routing cleanup by request_id + restore vendor pins On external turn teardown the WaiterGuard drop cleared the thread/meeting routing maps unconditionally. If a replacement turn had already parked a new approval on the same thread and overwritten the entry, this deleted the *new* request's routing, so the next typed yes/no fell through as a fresh chat turn instead of resolving the live gate. Only remove the mapping when it still points at this guard's request_id (both thread and meeting maps). Also restores vendor/tinyagents and vendor/tinycortex to the main pins; the branch had picked up stray submodule bumps that broke the build (tinycortex 33dda94 lacks the git-diff feature openhuman requires). --- src/openhuman/approval/gate.rs | 66 ++++++++++++++++++++++++++++++++-- vendor/tinyagents | 2 +- vendor/tinycortex | 2 +- 3 files changed, 66 insertions(+), 4 deletions(-) diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index 1d8fb6bedb..3b11a213b0 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -246,11 +246,19 @@ impl Drop for WaiterGuard<'_> { // row. `store::decide` is `WHERE decided_at IS NULL`, so a decision that // committed in the same instant is honored rather than overwritten. self.gate.evict_waiter(&self.request_id); + // Only clear the routing entry when it still points at *this* request. + // On external teardown a replacement turn can park a new approval on the + // same thread/meeting and overwrite the mapping before this guard drops; + // an unconditional `remove` would delete the *new* request's routing, so + // the next typed yes/no would fall through as a fresh chat turn instead + // of resolving the live gate (#4774). if let Some(thread_id) = &self.thread_id { - self.gate.thread_to_request.lock().remove(thread_id); + self.gate + .clear_thread_route_if_owned(thread_id, &self.request_id); } if let Some(meeting_key) = &self.meeting_key { - self.gate.meeting_to_request.lock().remove(meeting_key); + self.gate + .clear_meeting_route_if_owned(meeting_key, &self.request_id); } let _ = store::decide(&self.gate.config, &self.request_id, ApprovalDecision::Deny); tracing::warn!( @@ -1201,6 +1209,26 @@ impl ApprovalGate { self.meeting_to_request.lock().remove(&ic.meeting_key); } } + + /// Drop the thread → request mapping **only if** it still points at + /// `request_id`. Used by [`WaiterGuard::drop`] on external teardown, where a + /// replacement turn may have already parked a new approval on the same + /// thread and overwritten the entry; clearing unconditionally would delete + /// the *new* request's routing (#4774). + fn clear_thread_route_if_owned(&self, thread_id: &str, request_id: &str) { + let mut map = self.thread_to_request.lock(); + if map.get(thread_id).is_some_and(|rid| rid == request_id) { + map.remove(thread_id); + } + } + + /// Meeting-map analogue of [`Self::clear_thread_route_if_owned`]. + fn clear_meeting_route_if_owned(&self, meeting_key: &str, request_id: &str) { + let mut map = self.meeting_to_request.lock(); + if map.get(meeting_key).is_some_and(|rid| rid == request_id) { + map.remove(meeting_key); + } + } } /// Wall-clock milliseconds since the Unix epoch, for `CoreNotificationEvent::timestamp_ms`. @@ -1377,6 +1405,40 @@ mod tests { ); } + #[test] + fn guard_cleanup_only_clears_routing_it_still_owns() { + // Regression for #4774: on external turn teardown a replacement turn may + // have already parked a new approval on the same thread/meeting and + // overwritten the routing entry. The dropped guard for the *old* request + // must not clobber the *new* request's mapping. + let (gate, _dir) = test_gate(); + + gate.thread_to_request + .lock() + .insert("thread-1".into(), "req-new".into()); + gate.meeting_to_request + .lock() + .insert("meet-1".into(), "req-new".into()); + + // Stale guard for the superseded request is a no-op. + gate.clear_thread_route_if_owned("thread-1", "req-old"); + gate.clear_meeting_route_if_owned("meet-1", "req-old"); + assert_eq!( + gate.pending_for_thread("thread-1").as_deref(), + Some("req-new") + ); + assert_eq!( + gate.pending_for_meeting("meet-1").as_deref(), + Some("req-new") + ); + + // The owning request's guard clears its own routing. + gate.clear_thread_route_if_owned("thread-1", "req-new"); + gate.clear_meeting_route_if_owned("meet-1", "req-new"); + assert!(gate.pending_for_thread("thread-1").is_none()); + assert!(gate.pending_for_meeting("meet-1").is_none()); + } + #[tokio::test] async fn in_call_voice_deny_resolves_parked_approval_with_deny() { let (gate, _dir) = test_gate(); diff --git a/vendor/tinyagents b/vendor/tinyagents index b1f0d82d18..b34b2f83d7 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit b1f0d82d1845d5e3de020882b8817565f529cad9 +Subproject commit b34b2f83d7af492c9b7ccaf558c38bece4182dde diff --git a/vendor/tinycortex b/vendor/tinycortex index 33dda94305..671e78a014 160000 --- a/vendor/tinycortex +++ b/vendor/tinycortex @@ -1 +1 @@ -Subproject commit 33dda943053e61ef585fc39647cf1854344b6323 +Subproject commit 671e78a01411de5bcda8f3d1816ac6c67485d694