diff --git a/src/openhuman/approval/gate.rs b/src/openhuman/approval/gate.rs index e074b3dd5..3b11a213b 100644 --- a/src/openhuman/approval/gate.rs +++ b/src/openhuman/approval/gate.rs @@ -203,6 +203,72 @@ 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); + // 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 + .clear_thread_route_if_owned(thread_id, &self.request_id); + } + if let Some(meeting_key) = &self.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!( + 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 +882,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 +1020,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); @@ -1126,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`. @@ -1302,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(); @@ -1610,6 +1747,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