Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
208 changes: 208 additions & 0 deletions src/openhuman/approval/gate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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!(
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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`.
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand Down
Loading