Skip to content
114 changes: 107 additions & 7 deletions src/openhuman/security/approval/gate_tests.rs
Original file line number Diff line number Diff line change
@@ -1,8 +1,39 @@
use super::*;
use tempfile::TempDir;

/// TTL for the tests that assert the *timeout* path — the only ones that
/// need a park to expire while the test is still running.
const EXPIRY_TEST_TTL: Duration = Duration::from_secs(2);

/// Boot-time TTL for the `effective_ttl` fallback tests, which need a
/// window they can name rather than one that has to expire.
///
/// Deliberately a value nothing else in this module uses: those tests read
/// the TTL back out of the gate, so a number shared with the default (or
/// with [`EXPIRY_TEST_TTL`]) would let them pass against the wrong source.
const BOOT_TTL_UNDER_TEST: Duration = Duration::from_secs(7);

/// A gate whose park window is the production [`DEFAULT_APPROVAL_TTL`].
///
/// Most tests here park a call, poll for the row, then decide it; none of
/// them wait for expiry, so the window only has to outlast the poll. It
/// twice did not. #2367 raised it 500ms → 2s after the row expired before
/// `decide` could fire, and 2s then lost the same race under
/// `cargo-llvm-cov`: the poll loops budget 50×10ms, but each
/// `sleep(10ms)` stretches on a contended runner, and once the row is past
/// `expires_at` the `expire_stale_with_now` pass inside [`store::decide`]
/// denies it before that call's own `UPDATE ... WHERE decided_at IS NULL`
/// can match — so `decide` returns `Ok(None)`, the waiter is never woken,
/// and the park resolves as a TTL `Deny`.
///
/// Raising the number a third time would only move the threshold, so the
/// coupling is gone instead: tests that do not exercise expiry cannot
/// reach it, and the five that do ask for [`EXPIRY_TEST_TTL`] explicitly
/// via [`expiry_gate`]. That also removes an undocumented contract
/// between this helper and those distant call sites — two of them still
/// carried a `// TTL = 500ms` comment three raises later.
fn test_gate() -> (ApprovalGate, TempDir) {
test_gate_with_ttl(Duration::from_secs(2))
test_gate_with_ttl(DEFAULT_APPROVAL_TTL)
}

/// Build an approval gate with an explicit park deadline for tests that must
Expand All @@ -18,16 +49,85 @@ fn test_gate_with_ttl(ttl: Duration) -> (ApprovalGate, TempDir) {
// `debug_assert!` regression guard in `ApprovalGate::new`
// doesn't trip in tests.
let session = format!("session-{}", uuid::Uuid::new_v4());
// 500ms TTL was racing the 50×10ms poll loop on slow CI
// runners — the row would expire (and get denied by
// list_pending's lazy-expire) before `decide` could fire,
// surfacing as "pending row never appeared". 2s gives the
// most polling tests enough headroom while keeping
// `timeout_returns_deny` fast (PR #2367 CI flake).
let gate = ApprovalGate::new(config, session, ttl);
(gate, dir)
}

/// Return a request id after the approval row has been persisted and its
/// waiter has been registered. Unlike `list_pending`, this observation does
/// not lazily expire the row while a short-TTL test is waiting for insertion.
fn parked_request_id(gate: &ApprovalGate) -> Option<String> {
gate.waiters.lock().keys().next().cloned()
}

/// A gate for the tests that let a park expire, plus the env lock they
/// hold until the pending row captures its effective TTL.
///
/// [`ApprovalGate::effective_ttl`] reads `OPENHUMAN_APPROVAL_TTL_SECS` at
/// park time, not at construction, and this whole suite is a debug build,
/// so the override is live. The `effective_ttl_*` tests set that variable
/// for the length of their assertion. Rust runs tests as threads in one
/// process, so without sharing their lock an expiry test can park under
/// their `42`, and a row that was supposed to die in two seconds outlives
/// the test that is waiting for it instead.
///
/// Callers drop the returned guard after observing the pending row. Holding
/// it only while the gate is built would leave the window open for the park's
/// initial `effective_ttl` read, while holding it through expiry serializes
/// these tests unnecessarily.
/// `test_gate_with_ttl` must not take the lock, because the
/// `effective_ttl_*` tests call it while already holding it.
struct ExpiryEnvGuard {
previous_ttl: Option<String>,
_env_lock: std::sync::MutexGuard<'static, ()>,
}

impl Drop for ExpiryEnvGuard {
fn drop(&mut self) {
match self.previous_ttl.take() {
Some(value) => unsafe { std::env::set_var("OPENHUMAN_APPROVAL_TTL_SECS", value) },
None => unsafe { std::env::remove_var("OPENHUMAN_APPROVAL_TTL_SECS") },
}
}
}

fn expiry_gate() -> (ApprovalGate, TempDir, ExpiryEnvGuard) {
let env = crate::openhuman::config::TEST_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
// The lock keeps a sibling test from setting the override, but not a
// developer who exported it in their shell. effective_ttl would then
// replace EXPIRY_TEST_TTL at park time and the wait would be measuring
// their value, so clear it while the lock is held.
let previous_ttl = std::env::var("OPENHUMAN_APPROVAL_TTL_SECS").ok();
unsafe { std::env::remove_var("OPENHUMAN_APPROVAL_TTL_SECS") };
Comment thread
coderabbitai[bot] marked this conversation as resolved.
let (gate, dir) = test_gate_with_ttl(EXPIRY_TEST_TTL);
(
gate,
dir,
ExpiryEnvGuard {
previous_ttl,
_env_lock: env,
},
Comment thread
senamakel marked this conversation as resolved.
)
}

/// Decide a row that the test has just seen parked, failing on the expiry
/// race rather than through it.
///
/// [`ApprovalGate::decide`] returns `Ok(None)` when the row was already
/// resolved — which, per [`DecideMiss::AlreadyResolved`], includes *lazily
/// expired*. A bare `.unwrap()` there unwraps the `Result`, not the
/// `Option`, so that case passes silently and the test instead fails a few
/// lines later on "the outcome was not `Allow`", naming the wrong event.
fn decide_parked(gate: &ApprovalGate, request_id: &str, decision: ApprovalDecision) {
assert!(
gate.decide(request_id, decision).unwrap().is_some(),
"the parked row {request_id} was already resolved before the decision \
landed — it expired mid-test rather than being decided"
);
}

/// A chat context — the gate only parks within a live chat turn now, so
/// tests that exercise parking must run intercept inside this scope.
fn chat_ctx() -> ApprovalChatContext {
Expand Down
64 changes: 43 additions & 21 deletions src/openhuman/security/approval/gate_tests_part_01_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -511,16 +511,27 @@ async fn auto_approve_all_allows_a_remote_triage_dispatch_without_an_audit_row()

#[tokio::test]
async fn timeout_returns_deny() {
let (gate, _dir) = test_gate(); // TTL = 500ms
let (gate, _dir, env) = expiry_gate();
let gate = Arc::new(gate);
let outcome = turn_origin::with_origin(
web_origin(),
APPROVAL_CHAT_CONTEXT.scope(
chat_ctx(),
gate.intercept("composio", "timed out", serde_json::json!({})),
),
)
.await;
let g = gate.clone();
let handle = tokio::spawn(async move {
turn_origin::with_origin(
web_origin(),
APPROVAL_CHAT_CONTEXT.scope(
chat_ctx(),
g.intercept("composio", "timed out", serde_json::json!({})),
),
)
.await
});
let mut tries = 0;
while gate.list_pending().unwrap().is_empty() {
Comment thread
senamakel marked this conversation as resolved.
tries += 1;
assert!(tries < 50, "audit row never appeared for timeout test");
tokio::time::sleep(Duration::from_millis(10)).await;
}
drop(env);
let outcome = handle.await.unwrap();
match outcome {
GateOutcome::Deny { reason } => assert!(reason.contains("timed out")),
other => panic!("expected deny, got {other:?}"),
Expand All @@ -538,20 +549,31 @@ async fn timeout_returns_deny() {
/// unapproved, mirroring `timeout_returns_deny` above.
#[tokio::test]
async fn cancel_flow_run_parks_for_approval_when_a_gate_is_present() {
let (gate, _dir) = test_gate(); // TTL = 500ms
let (gate, _dir, env) = expiry_gate();
let gate = Arc::new(gate);
let outcome = turn_origin::with_origin(
web_origin(),
APPROVAL_CHAT_CONTEXT.scope(
chat_ctx(),
gate.intercept(
"cancel_flow_run",
"cancel run r-1 of flow f-1",
serde_json::json!({ "flow_id": "f-1", "run_id": "r-1" }),
let g = gate.clone();
let handle = tokio::spawn(async move {
turn_origin::with_origin(
web_origin(),
APPROVAL_CHAT_CONTEXT.scope(
chat_ctx(),
g.intercept(
"cancel_flow_run",
"cancel run r-1 of flow f-1",
serde_json::json!({ "flow_id": "f-1", "run_id": "r-1" }),
),
),
),
)
.await;
)
.await
});
let mut tries = 0;
while gate.list_pending().unwrap().is_empty() {
tries += 1;
assert!(tries < 50, "audit row never appeared for cancel_flow_run");
tokio::time::sleep(Duration::from_millis(10)).await;
}
drop(env);
let outcome = handle.await.unwrap();
// No decision ever arrives — the call must NOT auto-execute. It
// parks until the gate's TTL elapses, then denies (never `Allow`).
match outcome {
Expand Down
36 changes: 16 additions & 20 deletions src/openhuman/security/approval/gate_tests_part_02_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ async fn pending_for_thread_tracks_request_under_chat_context_and_clears() {
// This test polls a parked task while the full Rust suite runs thousands
// of other tests concurrently. Keep its coordination deadline independent
// of the short TTL used by the timeout behavior tests.
let (gate, _dir) = test_gate_with_ttl(Duration::from_secs(10));
let (gate, _dir) = test_gate();
let gate = Arc::new(gate);

// Run intercept inside a scoped chat context + matching WebChat
Expand Down Expand Up @@ -40,8 +40,7 @@ async fn pending_for_thread_tracks_request_under_chat_context_and_clears() {
};

// Decide via the mapped request_id (as the chat ingress router will).
gate.decide(&request_id, ApprovalDecision::ApproveOnce)
.unwrap();
decide_parked(&gate, &request_id, ApprovalDecision::ApproveOnce);
assert!(matches!(handle.await.unwrap(), GateOutcome::Allow));

// Mapping is cleared once intercept returns.
Expand All @@ -61,7 +60,7 @@ async fn pending_for_thread_tracks_request_under_chat_context_and_clears() {
async fn webchat_origin_routes_park_when_approval_chat_context_absent() {
// See the companion chat-context routing test above: this is a
// coordination deadline, not the timeout behavior under test.
let (gate, _dir) = test_gate_with_ttl(Duration::from_secs(10));
let (gate, _dir) = test_gate();
let gate = Arc::new(gate);

// WebChat origin scoped, but NO `APPROVAL_CHAT_CONTEXT` — exactly the
Expand Down Expand Up @@ -100,8 +99,7 @@ async fn webchat_origin_routes_park_when_approval_chat_context_absent() {

// A decision on that mapped request resolves the park (the card can
// surface and be approved), instead of silently TTL-denying.
gate.decide(&request_id, ApprovalDecision::ApproveOnce)
.unwrap();
decide_parked(&gate, &request_id, ApprovalDecision::ApproveOnce);
assert!(matches!(handle.await.unwrap(), GateOutcome::Allow));
assert!(gate.pending_for_thread("thread-async").is_none());
}
Expand Down Expand Up @@ -187,7 +185,7 @@ async fn waiter_future_dropped_mid_park_evicts_waiter_clears_routing_and_denies_
// card-click still resolves it in the DB.
#[tokio::test]
async fn intercept_audited_bounded_abandons_park_and_leaves_row_pending() {
let (gate, _dir) = test_gate(); // boot-time TTL = 2s
let (gate, _dir) = test_gate();
let gate = Arc::new(gate);

let g = gate.clone();
Expand All @@ -200,7 +198,7 @@ async fn intercept_audited_bounded_abandons_park_and_leaves_row_pending() {
client_id: "client-1".into(),
request_id: Some("req-bound".into()),
};
// 100ms caller bound — far below the 2s gate TTL — so the bound is what
// 100ms caller bound — far below the gate TTL — so the bound is what
// elapses, not the gate's own timeout.
let handle = tokio::spawn(async move {
turn_origin::with_origin(
Expand Down Expand Up @@ -269,7 +267,7 @@ fn effective_ttl_uses_env_override_when_valid() {
let _env = crate::openhuman::config::TEST_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let (gate, _dir) = test_gate(); // boot-time TTL = 2s
let (gate, _dir) = test_gate_with_ttl(BOOT_TTL_UNDER_TEST);
unsafe { std::env::set_var("OPENHUMAN_APPROVAL_TTL_SECS", "42") };
assert_eq!(
gate.effective_ttl(),
Expand All @@ -284,11 +282,11 @@ fn effective_ttl_falls_back_to_boot_ttl_for_garbage_value() {
let _env = crate::openhuman::config::TEST_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let (gate, _dir) = test_gate(); // boot-time TTL = 2s
let (gate, _dir) = test_gate_with_ttl(BOOT_TTL_UNDER_TEST);
unsafe { std::env::set_var("OPENHUMAN_APPROVAL_TTL_SECS", "not-a-number") };
assert_eq!(
gate.effective_ttl(),
Duration::from_secs(2),
BOOT_TTL_UNDER_TEST,
"garbage OPENHUMAN_APPROVAL_TTL_SECS must fall back to boot-time TTL"
);
unsafe { std::env::remove_var("OPENHUMAN_APPROVAL_TTL_SECS") };
Expand All @@ -299,11 +297,11 @@ fn effective_ttl_falls_back_to_boot_ttl_when_unset() {
let _env = crate::openhuman::config::TEST_ENV_LOCK
.lock()
.unwrap_or_else(|e| e.into_inner());
let (gate, _dir) = test_gate(); // boot-time TTL = 2s
let (gate, _dir) = test_gate_with_ttl(BOOT_TTL_UNDER_TEST);
unsafe { std::env::remove_var("OPENHUMAN_APPROVAL_TTL_SECS") };
assert_eq!(
gate.effective_ttl(),
Duration::from_secs(2),
BOOT_TTL_UNDER_TEST,
"unset OPENHUMAN_APPROVAL_TTL_SECS must fall back to boot-time TTL"
);
}
Expand Down Expand Up @@ -359,7 +357,7 @@ mod resolve_park_ttl_tests {
/// above, but that alone doesn't prove `intercept_audited_inner` actually
/// persists the clamped TTL when the copilot-streaming context is scoped.
/// Builds a gate with the full `DEFAULT_APPROVAL_TTL` boot TTL (unlike
/// `test_gate()`'s 2s, which is already shorter than either clamp and
/// `test_gate()` before this suite stopped hard-coding a 2s window, which
/// would make this assertion vacuous), scopes
/// `APPROVAL_COPILOT_STREAM_CONTEXT` alongside the chat context + WebChat
/// origin the way `flows::ops::flows_build` does in production, and
Expand Down Expand Up @@ -596,7 +594,7 @@ async fn intercept_with_workflow_require_approval_persists_and_ttl_denies() {
// trust root — same conservative park-and-audit shape as
// `GoalContinuation` / `ExternalChannel`, since there is no flow
// review surface to route the prompt to yet (B3).
let (gate, _dir) = test_gate(); // 2s TTL
let (gate, _dir, env) = expiry_gate();
let gate = Arc::new(gate);
let origin = AgentTurnOrigin::TrustedAutomation {
job_id: "flow-2".into(),
Expand All @@ -615,18 +613,16 @@ async fn intercept_with_workflow_require_approval_persists_and_ttl_denies() {
});

let mut tries = 0;
loop {
if !gate.list_pending().unwrap().is_empty() {
break;
}
while parked_request_id(&gate).is_none() {
tries += 1;
assert!(
tries < 50,
"audit row never appeared for require_approval workflow origin"
"approval waiter never appeared for require_approval workflow origin"
);
tokio::time::sleep(Duration::from_millis(10)).await;
}

drop(env);
let outcome = handle.await.unwrap();
match outcome {
GateOutcome::Deny { reason } => assert!(reason.contains("timed out")),
Expand Down
Loading
Loading