From 86e1a4c524bcd3cc3396cffab90af27693fde441 Mon Sep 17 00:00:00 2001 From: Sam Date: Tue, 21 Jul 2026 19:00:30 +0200 Subject: [PATCH] fix(approval): preserve replacement routes during cleanup (#4786) Co-authored-by: Sami Rusani <14844597+samrusani@users.noreply.github.com> --- src/openhuman/approval/gate.rs | 237 +++++++++++++++++++++++++++++++-- 1 file changed, 226 insertions(+), 11 deletions(-) diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index afde88c84..1a46e3676 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -779,7 +779,8 @@ impl ApprovalGate { if let Err(err) = store::insert_pending(&self.config, &pending, &self.session_id) { self.evict_waiter(&request_id); - self.clear_thread(&chat_thread_id); + self.clear_thread(&chat_thread_id, &request_id); + self.clear_meeting(&in_call_ctx, &request_id); tracing::error!( error = %err, tool = tool_name, @@ -904,7 +905,7 @@ impl ApprovalGate { "[approval::gate] decision received" ); if decision.is_approve() { - (GateOutcome::Allow, Some(request_id)) + (GateOutcome::Allow, Some(request_id.clone())) } else { ( GateOutcome::Deny { @@ -998,7 +999,7 @@ impl ApprovalGate { // on this path too — otherwise the stale thread→request // mapping survives and the next yes/no on the thread could be // routed to this already-finished request. - (GateOutcome::Allow, Some(request_id)) + (GateOutcome::Allow, Some(request_id.clone())) } else { tracing::warn!( request_id = %request_id, @@ -1026,8 +1027,8 @@ impl ApprovalGate { 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); - self.clear_meeting(&in_call_ctx); + self.clear_thread(&chat_thread_id, &request_id); + self.clear_meeting(&in_call_ctx, &request_id); outcome } @@ -1196,17 +1197,17 @@ impl ApprovalGate { self.meeting_to_request.lock().get(meeting_key).cloned() } - /// Drop the thread → request mapping (best-effort; no-op when absent). - fn clear_thread(&self, thread_id: &Option) { + /// Drop the thread → request mapping when it still belongs to this request. + fn clear_thread(&self, thread_id: &Option, request_id: &str) { if let Some(t) = thread_id { - self.thread_to_request.lock().remove(t); + self.clear_thread_route_if_owned(t, request_id); } } - /// Drop the meeting → request mapping (best-effort; no-op when absent). - fn clear_meeting(&self, ctx: &Option) { + /// Drop the meeting → request mapping when it still belongs to this request. + fn clear_meeting(&self, ctx: &Option, request_id: &str) { if let Some(ic) = ctx { - self.meeting_to_request.lock().remove(&ic.meeting_key); + self.clear_meeting_route_if_owned(&ic.meeting_key, request_id); } } @@ -1574,6 +1575,220 @@ mod tests { } } + #[tokio::test] + async fn aborting_older_chat_waiter_preserves_newer_thread_route() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let old_gate = gate.clone(); + let old_handle = tokio::spawn(async move { + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( + chat_ctx(), + old_gate.intercept("composio", "old action", serde_json::json!({})), + ), + ) + .await + }); + + let mut tries = 0; + let old_request_id = loop { + if let Some(request_id) = gate.pending_for_thread("t-test") { + break request_id; + } + tries += 1; + assert!(tries < 1_000, "old chat approval route never appeared"); + tokio::task::yield_now().await; + }; + + let new_gate = gate.clone(); + let new_handle = tokio::spawn(async move { + turn_origin::with_origin( + web_origin(), + APPROVAL_CHAT_CONTEXT.scope( + chat_ctx(), + new_gate.intercept("composio", "new action", serde_json::json!({})), + ), + ) + .await + }); + + let mut tries = 0; + let new_request_id = loop { + if let Some(request_id) = gate.pending_for_thread("t-test") { + if request_id != old_request_id { + break request_id; + } + } + tries += 1; + assert!(tries < 1_000, "new chat approval route never appeared"); + tokio::task::yield_now().await; + }; + + old_handle.abort(); + assert!(old_handle.await.unwrap_err().is_cancelled()); + + assert_eq!( + gate.pending_for_thread("t-test").as_deref(), + Some(new_request_id.as_str()) + ); + assert!(!gate.waiters.lock().contains_key(&old_request_id)); + assert!(gate.waiters.lock().contains_key(&new_request_id)); + assert_eq!( + store::get_decision(&gate.config, &old_request_id).unwrap(), + Some(ApprovalDecision::Deny) + ); + + gate.decide(&new_request_id, ApprovalDecision::ApproveOnce) + .unwrap(); + assert!(matches!(new_handle.await.unwrap(), GateOutcome::Allow)); + assert!(gate.pending_for_thread("t-test").is_none()); + } + + #[tokio::test] + async fn pending_store_failure_clears_in_call_meeting_route() { + let dir = TempDir::new().unwrap(); + let blocked_workspace = dir.path().join("workspace-file"); + std::fs::write(&blocked_workspace, b"not a directory").unwrap(); + let config = Config { + workspace_dir: blocked_workspace, + ..Config::default() + }; + let gate = ApprovalGate::new( + config, + format!("session-{}", uuid::Uuid::new_v4()), + Duration::from_secs(2), + ); + + let outcome = turn_origin::with_origin( + meet_origin(), + APPROVAL_IN_CALL_CONTEXT.scope( + in_call_ctx(), + gate.intercept("composio", "send email", serde_json::json!({})), + ), + ) + .await; + + assert!(matches!( + outcome, + GateOutcome::Deny { reason } if reason.contains("could not persist") + )); + assert!(gate.waiters.lock().is_empty()); + assert!(gate.pending_for_meeting("meet-1").is_none()); + } + + #[tokio::test] + async fn externally_aborted_in_call_waiter_cleans_meeting_route() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let g = gate.clone(); + let handle = tokio::spawn(async move { + turn_origin::with_origin( + meet_origin(), + APPROVAL_IN_CALL_CONTEXT.scope( + in_call_ctx(), + g.intercept("composio", "send email", serde_json::json!({})), + ), + ) + .await + }); + + let mut tries = 0; + let request_id = loop { + if let Some(request_id) = gate.pending_for_meeting("meet-1") { + break request_id; + } + tries += 1; + assert!(tries < 1_000, "meeting approval route never appeared"); + tokio::task::yield_now().await; + }; + assert!(gate.waiters.lock().contains_key(&request_id)); + + handle.abort(); + assert!(handle.await.unwrap_err().is_cancelled()); + + assert!(gate.pending_for_meeting("meet-1").is_none()); + assert!(!gate.waiters.lock().contains_key(&request_id)); + assert!(gate.list_pending().unwrap().is_empty()); + assert_eq!( + store::get_decision(&gate.config, &request_id).unwrap(), + Some(ApprovalDecision::Deny) + ); + } + + #[tokio::test] + async fn aborting_older_in_call_waiter_preserves_newer_meeting_route() { + let (gate, _dir) = test_gate(); + let gate = Arc::new(gate); + + let old_gate = gate.clone(); + let old_handle = tokio::spawn(async move { + turn_origin::with_origin( + meet_origin(), + APPROVAL_IN_CALL_CONTEXT.scope( + in_call_ctx(), + old_gate.intercept("composio", "old action", serde_json::json!({})), + ), + ) + .await + }); + + let mut tries = 0; + let old_request_id = loop { + if let Some(request_id) = gate.pending_for_meeting("meet-1") { + break request_id; + } + tries += 1; + assert!(tries < 1_000, "old meeting approval route never appeared"); + tokio::task::yield_now().await; + }; + + let new_gate = gate.clone(); + let new_handle = tokio::spawn(async move { + turn_origin::with_origin( + meet_origin(), + APPROVAL_IN_CALL_CONTEXT.scope( + in_call_ctx(), + new_gate.intercept("composio", "new action", serde_json::json!({})), + ), + ) + .await + }); + + let mut tries = 0; + let new_request_id = loop { + if let Some(request_id) = gate.pending_for_meeting("meet-1") { + if request_id != old_request_id { + break request_id; + } + } + tries += 1; + assert!(tries < 1_000, "new meeting approval route never appeared"); + tokio::task::yield_now().await; + }; + + old_handle.abort(); + assert!(old_handle.await.unwrap_err().is_cancelled()); + + assert_eq!( + gate.pending_for_meeting("meet-1").as_deref(), + Some(new_request_id.as_str()) + ); + assert!(!gate.waiters.lock().contains_key(&old_request_id)); + assert!(gate.waiters.lock().contains_key(&new_request_id)); + assert_eq!( + store::get_decision(&gate.config, &old_request_id).unwrap(), + Some(ApprovalDecision::Deny) + ); + + gate.decide(&new_request_id, ApprovalDecision::ApproveOnce) + .unwrap(); + assert!(matches!(new_handle.await.unwrap(), GateOutcome::Allow)); + assert!(gate.pending_for_meeting("meet-1").is_none()); + } + #[tokio::test] async fn auto_approve_tool_skips_prompt() { // The gate reads the "Always allow" allowlist from the process-global