fix(approval): RAII guard so a parked waiter is cleaned up on external turn teardown (#4819)

src/openhuman/approval/gate.rs

@@ -203,6 +203,72 @@ pub struct ApprovalGate {
     meeting_to_request: Mutex<HashMap<String, String>>,
 }
 
+/// 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<String>,
+    meeting_key: Option<String>,
+    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