diff --git a/crates/ironclaw_reborn_composition/src/auth.rs b/crates/ironclaw_reborn_composition/src/auth.rs index 43001a7192c..a9a8d4c2fe3 100644 --- a/crates/ironclaw_reborn_composition/src/auth.rs +++ b/crates/ironclaw_reborn_composition/src/auth.rs @@ -36,7 +36,7 @@ use crate::product_auth_runtime_credentials::{ RuntimeCredentialAccountRefreshPort, RuntimeCredentialAccountRefreshService, RuntimeCredentialAccountSelectionService, }; -use crate::{AuthChallengeProvider, AuthChallengeView}; +use crate::{AuthChallengeProvider, AuthChallengeView, BlockedAuthFlowCanceller}; pub(crate) const AUTH_CONTINUATION_DISPATCH_FAILED_CODE: &str = "auth_continuation_dispatch_failed"; @@ -738,11 +738,34 @@ impl RebornProductAuthServices { /// 4-field view in that case, which is backward-compatible. #[doc(hidden)] pub fn as_auth_challenge_provider(self: &Arc) -> Option> { - if self.flow_record_source.is_some() { - Some(Arc::clone(self) as Arc) - } else { - None - } + self.has_flow_record_source() + .then(|| Arc::clone(self) as Arc) + } + + /// Expose this service as an `Arc` so the Slack + /// delivery path can cancel the durable `AuthFlow` record alongside the run + /// when it auto-denies a non-OAuth auth challenge (issue #4952). + /// + /// Returns `None` under the same condition as + /// [`Self::as_auth_challenge_provider`] — both flow-backed facades gate on + /// [`Self::has_flow_record_source`]. They stay separate accessors because they + /// expose distinct capability ports (`AuthChallengeProvider` vs + /// `BlockedAuthFlowCanceller`), but share the one wiring precondition. + #[doc(hidden)] + pub fn as_blocked_auth_flow_canceller( + self: &Arc, + ) -> Option> { + self.has_flow_record_source() + .then(|| Arc::clone(self) as Arc) + } + + /// Shared precondition for the flow-backed facades: both + /// [`Self::as_auth_challenge_provider`] and + /// [`Self::as_blocked_auth_flow_canceller`] are only available when an + /// `AuthFlowRecordSource` projection is wired in. Defined once so the gate + /// cannot drift between the two accessors. + fn has_flow_record_source(&self) -> bool { + self.flow_record_source.is_some() } /// Refresh a credential account through the injected product-auth port. @@ -1394,6 +1417,62 @@ impl AuthChallengeProvider for RebornProductAuthServices { } } +#[async_trait] +impl BlockedAuthFlowCanceller for RebornProductAuthServices { + async fn cancel_blocked_auth_flow( + &self, + scope: &TurnScope, + owner_user_id: &UserId, + run_id: TurnRunId, + gate_ref: &str, + ) -> Result<(), AuthProductError> { + let gate_ref = AuthGateRef::new(gate_ref.to_string()).map_err(|err| { + AuthProductError::InvalidRequest { + reason: format!("invalid gate ref for auth-flow cancel: {err}"), + } + })?; + let Some(source) = self.flow_record_source.as_ref() else { + // No projection source wired in: nothing to cancel here. + return Ok(()); + }; + // `include_terminal: false` means an already-terminal flow (or a missing + // one) resolves to `None`, so the OAuth-callback race — where the flow + // completes just before auto-deny — is a graceful no-op rather than an + // error. We only ever cancel a flow that is still non-terminal. + let flow = source + .flow_for_turn_gate(TurnGateAuthFlowQuery { + owner: AuthFlowOwnerScope { + tenant_id: scope.tenant_id.clone(), + user_id: owner_user_id.clone(), + agent_id: scope.agent_id.clone(), + project_id: scope.project_id.clone(), + thread_id: scope.thread_id.clone(), + }, + turn_run_ref: TurnRunRef::new(run_id.to_string()).map_err(|err| { + AuthProductError::InvalidRequest { + reason: format!("invalid turn run ref for auth-flow cancel: {err}"), + } + })?, + gate_ref, + include_terminal: false, + }) + .await?; + let Some(flow) = flow else { + return Ok(()); + }; + match self.flow_manager.cancel_flow(&flow.scope, flow.id).await { + Ok(_) => Ok(()), + // The flow terminalized between our non-terminal read above and this + // cancel (a concurrent OAuth callback or another canceller). Already + // terminal is the desired end state, so honor the documented graceful + // no-op contract instead of surfacing the race as an error. Real + // lookup/scope/backend errors still propagate. + Err(AuthProductError::Canceled | AuthProductError::FlowAlreadyTerminal) => Ok(()), + Err(err) => Err(err), + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -1755,4 +1834,538 @@ mod tests { unreachable!("constructor tests do not call cleanup methods") } } + + // ── cancel_blocked_auth_flow facade tests ───────────────────────────────── + + /// Build a minimal `RebornProductAuthServices` for `cancel_blocked_auth_flow` + /// tests. The `flow_manager` is backed by `InMemoryAuthProductServices` so + /// callers can inspect whether `cancel_flow` was actually invoked (by checking + /// the flow's status after the call). All other ports use `SharedAuthTestDouble` + /// (they are never called by `cancel_blocked_auth_flow`). + fn make_auth_services_with_flow_source( + auth_svc: Arc, + ) -> RebornProductAuthServices { + let double = Arc::new(SharedAuthTestDouble); + RebornProductAuthServices::new( + auth_svc.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + Arc::new(NoopAuthContinuationDispatcher), + ) + .with_flow_record_source(auth_svc as Arc) + } + + /// Build an `AuthProductScope` that is consistent with a `personal_turn_scope`-like + /// `TurnScope` used by `cancel_blocked_auth_flow`. + fn test_auth_product_scope() -> AuthProductScope { + use ironclaw_auth::AuthSurface; + use ironclaw_host_api::{AgentId, ResourceScope, TenantId, ThreadId, UserId}; + + let resource = ResourceScope { + tenant_id: TenantId::new("test-tenant").expect("tenant"), + user_id: UserId::new("creator-user").expect("user"), + agent_id: Some(AgentId::new("test-agent").expect("agent")), + project_id: None, + mission_id: None, + thread_id: Some(ThreadId::new("test-thread").expect("thread")), + invocation_id: ironclaw_host_api::InvocationId::new(), + }; + AuthProductScope::new(resource, AuthSurface::Chat) + } + + /// Build a minimal non-terminal `AuthFlowRecord` whose continuation matches + /// a `TurnGateAuthFlowQuery` for `run_id` / `gate_ref`. + async fn create_test_flow( + auth_svc: &InMemoryAuthProductServices, + scope: AuthProductScope, + run_id: TurnRunId, + gate_ref_str: &str, + ) -> AuthFlowRecord { + let gate_ref = AuthGateRef::new(gate_ref_str.to_string()).expect("gate ref"); + let turn_run_ref = TurnRunRef::new(run_id.to_string()).expect("turn run ref"); + auth_svc + .create_flow(NewAuthFlow { + id: None, + scope, + kind: AuthFlowKind::IntegrationCredential, + provider: AuthProviderId::new("test-provider").expect("provider"), + challenge: AuthChallenge::SetupRequired { + provider: AuthProviderId::new("test-provider").expect("provider"), + message: "test".to_string(), + }, + continuation: AuthContinuationRef::TurnGateResume { + turn_run_ref, + gate_ref, + }, + update_binding: None, + opaque_state_hash: None, + pkce_verifier_hash: None, + expires_at: chrono::Utc::now() + chrono::Duration::hours(1), + }) + .await + .expect("create test flow") + } + + /// `cancel_blocked_auth_flow` must cancel a non-terminal flow via `flow_manager` + /// when `flow_record_source` returns one for the queried run/gate. + #[tokio::test] + async fn cancel_blocked_auth_flow_cancels_non_terminal_flow() { + use ironclaw_host_api::UserId; + use ironclaw_turns::TurnScope; + + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + let services = Arc::new(make_auth_services_with_flow_source(Arc::clone(&auth_svc))); + + let run_id = TurnRunId::new(); + let gate_ref_str = "gate:cancel-test"; + let scope_resource = test_auth_product_scope(); + let flow = create_test_flow(&auth_svc, scope_resource, run_id, gate_ref_str).await; + + // Sanity: flow is non-terminal before the call. + assert_eq!( + flow.status, + AuthFlowStatus::AwaitingUser, + "pre-condition: flow must be non-terminal" + ); + + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + + services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, gate_ref_str) + .await + .expect("cancel_blocked_auth_flow must succeed"); + + // The flow must now be terminal (Canceled). + let flows = auth_svc.flow_records_snapshot(); + let updated = flows + .iter() + .find(|f| f.id == flow.id) + .expect("flow must still exist after cancel"); + assert_eq!( + updated.status, + AuthFlowStatus::Canceled, + "cancel_blocked_auth_flow must have cancelled the flow via flow_manager" + ); + } + + /// `cancel_blocked_auth_flow` is a no-op (returns `Ok`) when the + /// `flow_record_source` returns `None` for the queried run/gate (flow absent + /// or already terminal). + #[tokio::test] + async fn cancel_blocked_auth_flow_is_noop_when_flow_absent() { + use ironclaw_host_api::UserId; + use ironclaw_turns::{TurnRunId, TurnScope}; + + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + let services = Arc::new(make_auth_services_with_flow_source(Arc::clone(&auth_svc))); + + // No flow is seeded — `flow_for_turn_gate` returns None. + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + let run_id = TurnRunId::new(); + + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, "gate:absent") + .await; + + assert!( + result.is_ok(), + "cancel_blocked_auth_flow must return Ok when flow is absent; got: {result:?}" + ); + // No flows were created, so nothing to check in auth_svc. + assert!( + auth_svc.flow_records_snapshot().is_empty(), + "no flow must exist after a no-op cancel" + ); + } + + /// `cancel_blocked_auth_flow` must treat `Err(AuthProductError::Canceled)` and + /// `Err(AuthProductError::FlowAlreadyTerminal)` from `flow_manager.cancel_flow` + /// as `Ok(())` — these represent a concurrent terminal race where the flow + /// completed between the non-terminal read and the cancel call. + /// + /// Also asserts a negative case: a real backend error (e.g. `BackendUnavailable`) + /// still propagates as `Err` to confirm the normalization is not over-broad. + #[tokio::test] + async fn cancel_blocked_auth_flow_treats_terminal_race_as_ok() { + use ironclaw_auth::{ + AuthFlowId, AuthFlowRecord, AuthProductError, AuthProductScope, + OAuthCallbackClaimRequest, OAuthCallbackFailureInput, OAuthCallbackInput, Timestamp, + }; + use ironclaw_host_api::UserId; + use ironclaw_turns::TurnScope; + + /// A `AuthFlowManager` whose `cancel_flow` returns a caller-supplied error + /// while all other methods forward to the real in-memory store. Used to + /// simulate the terminal race without needing to actually put the flow in + /// a terminal state before the call. + struct TerminalRaceFlowManager { + inner: Arc, + cancel_error: tokio::sync::Mutex>, + } + + impl TerminalRaceFlowManager { + fn returning( + inner: Arc, + error: AuthProductError, + ) -> Arc { + Arc::new(Self { + inner, + cancel_error: tokio::sync::Mutex::new(Some(error)), + }) + } + } + + #[async_trait::async_trait] + impl AuthFlowManager for TerminalRaceFlowManager { + async fn create_flow( + &self, + request: NewAuthFlow, + ) -> Result { + self.inner.create_flow(request).await + } + + async fn get_flow( + &self, + scope: &AuthProductScope, + flow_id: AuthFlowId, + ) -> Result, AuthProductError> { + self.inner.get_flow(scope, flow_id).await + } + + async fn cancel_flow( + &self, + _scope: &AuthProductScope, + _flow_id: AuthFlowId, + ) -> Result { + let err = self + .cancel_error + .lock() + .await + .take() + .expect("cancel_flow called more than once on TerminalRaceFlowManager"); + Err(err) + } + + async fn claim_oauth_callback( + &self, + _scope: &AuthProductScope, + _request: OAuthCallbackClaimRequest, + ) -> Result { + unreachable!("terminal-race test does not call claim_oauth_callback") + } + + async fn complete_oauth_callback( + &self, + _scope: &AuthProductScope, + _input: OAuthCallbackInput, + ) -> Result { + unreachable!("terminal-race test does not call complete_oauth_callback") + } + + async fn complete_credential_selection( + &self, + _scope: &AuthProductScope, + _input: ironclaw_auth::CredentialSelectionInput, + ) -> Result { + unreachable!("terminal-race test does not call complete_credential_selection") + } + + async fn complete_manual_token( + &self, + _scope: &AuthProductScope, + _input: ironclaw_auth::ManualTokenCompletionInput, + ) -> Result { + unreachable!("terminal-race test does not call complete_manual_token") + } + + async fn cancel_manual_token( + &self, + _scope: &AuthProductScope, + _interaction_id: AuthInteractionId, + ) -> Result, AuthProductError> { + unreachable!("terminal-race test does not call cancel_manual_token") + } + + async fn fail_oauth_callback( + &self, + _scope: &AuthProductScope, + _input: OAuthCallbackFailureInput, + ) -> Result { + unreachable!("terminal-race test does not call fail_oauth_callback") + } + + async fn mark_continuation_dispatched( + &self, + _scope: &AuthProductScope, + _flow_id: AuthFlowId, + _emitted_at: Timestamp, + ) -> Result { + unreachable!("terminal-race test does not call mark_continuation_dispatched") + } + } + + // Helper: build services with a custom flow_manager but real flow_record_source. + let build_services_with_manager = + |auth_svc: Arc, manager: Arc| { + let double = Arc::new(SharedAuthTestDouble); + RebornProductAuthServices::new( + manager, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + Arc::new(NoopAuthContinuationDispatcher), + ) + .with_flow_record_source(auth_svc as Arc) + }; + + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + let run_id = TurnRunId::new(); + let gate_ref_str = "gate:terminal-race-test"; + let scope_resource = test_auth_product_scope(); + + // ── Case 1: cancel_flow returns Err(FlowAlreadyTerminal) → Ok(()) ─────────── + { + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + // Seed a non-terminal flow so flow_record_source returns Some(…). + create_test_flow(&auth_svc, scope_resource.clone(), run_id, gate_ref_str).await; + let manager = TerminalRaceFlowManager::returning( + Arc::clone(&auth_svc), + AuthProductError::FlowAlreadyTerminal, + ); + let services = Arc::new(build_services_with_manager(auth_svc, manager)); + + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, gate_ref_str) + .await; + assert!( + result.is_ok(), + "FlowAlreadyTerminal from cancel_flow must be normalized to Ok(()); got: {result:?}" + ); + } + + // ── Case 2: cancel_flow returns Err(Canceled) → Ok(()) ────────────────────── + { + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + create_test_flow(&auth_svc, scope_resource.clone(), run_id, gate_ref_str).await; + let manager = TerminalRaceFlowManager::returning( + Arc::clone(&auth_svc), + AuthProductError::Canceled, + ); + let services = Arc::new(build_services_with_manager(auth_svc, manager)); + + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, gate_ref_str) + .await; + assert!( + result.is_ok(), + "Canceled from cancel_flow must be normalized to Ok(()); got: {result:?}" + ); + } + + // ── Negative case: cancel_flow returns a real error → Err propagates ───────── + { + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + create_test_flow(&auth_svc, scope_resource, run_id, gate_ref_str).await; + let manager = TerminalRaceFlowManager::returning( + Arc::clone(&auth_svc), + AuthProductError::BackendUnavailable, + ); + let services = Arc::new(build_services_with_manager(auth_svc, manager)); + + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, gate_ref_str) + .await; + assert!( + matches!(result, Err(AuthProductError::BackendUnavailable)), + "BackendUnavailable from cancel_flow must propagate as Err; got: {result:?}" + ); + } + } + + /// `cancel_blocked_auth_flow` is a no-op (returns `Ok`) when the service + /// was built without a `flow_record_source`. + #[tokio::test] + async fn cancel_blocked_auth_flow_is_noop_without_flow_record_source() { + use ironclaw_host_api::UserId; + use ironclaw_turns::{TurnRunId, TurnScope}; + + let double = Arc::new(SharedAuthTestDouble); + // Build WITHOUT `.with_flow_record_source` — `flow_record_source` is None. + let services = Arc::new(RebornProductAuthServices::new( + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + Arc::new(NoopAuthContinuationDispatcher), + )); + + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + let run_id = TurnRunId::new(); + + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, "gate:no-source") + .await; + + assert!( + result.is_ok(), + "cancel_blocked_auth_flow must return Ok when flow_record_source is absent; got: {result:?}" + ); + // SharedAuthTestDouble's cancel_flow panics with unreachable! — if we reach + // here without panic, cancel_flow was never called (as required). + } + + /// `cancel_blocked_auth_flow` must return `Err(AuthProductError::InvalidRequest)` + /// when the supplied `gate_ref` string fails `AuthGateRef::new` validation. + /// + /// `AuthGateRef` delegates to `validate_public_text`, which rejects empty + /// strings ("must not be empty"). An empty `gate_ref` is therefore the + /// simplest value that always fails at the facade boundary — regardless of + /// whether any flow or source is present. + #[tokio::test] + async fn cancel_blocked_auth_flow_rejects_invalid_gate_ref() { + use ironclaw_host_api::UserId; + use ironclaw_turns::{TurnRunId, TurnScope}; + + let auth_svc = Arc::new(InMemoryAuthProductServices::new()); + let services = Arc::new(make_auth_services_with_flow_source(Arc::clone(&auth_svc))); + + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + let run_id = TurnRunId::new(); + + // Empty string is rejected by `validate_public_text` ("must not be empty"). + let result = services + .cancel_blocked_auth_flow(&turn_scope, &owner_user_id, run_id, "") + .await; + + match result { + Err(AuthProductError::InvalidRequest { reason }) => { + assert!( + !reason.is_empty(), + "InvalidRequest reason must be non-empty for an invalid gate ref" + ); + assert!( + reason.contains("invalid gate ref for auth-flow cancel"), + "reason must include the caller-supplied context string; got: {reason}" + ); + } + other => panic!("expected Err(InvalidRequest) for empty gate_ref, got: {other:?}"), + } + } + + /// `cancel_blocked_auth_flow` must propagate `Err` returned by the + /// `flow_record_source` — a backend lookup failure must not be silently + /// swallowed. + /// + /// Uses a minimal local stub whose `flow_for_turn_gate` always returns + /// `Err(AuthProductError::BackendUnavailable)`. This exercises the `?` + /// on the `source.flow_for_turn_gate(…).await?` call site. + #[tokio::test] + async fn cancel_blocked_auth_flow_propagates_flow_source_error() { + use ironclaw_host_api::UserId; + use ironclaw_turns::{TurnRunId, TurnScope}; + + /// A flow record source that always errors out. + struct AlwaysFailingFlowSource; + + #[async_trait::async_trait] + impl AuthFlowRecordSource for AlwaysFailingFlowSource { + async fn flow_for_turn_gate( + &self, + _query: ironclaw_auth::TurnGateAuthFlowQuery, + ) -> Result, AuthProductError> { + Err(AuthProductError::BackendUnavailable) + } + + async fn flows_for_owner( + &self, + _owner: ironclaw_auth::AuthFlowOwnerScope, + ) -> Result, AuthProductError> { + unreachable!("flow-source-error test does not call flows_for_owner") + } + } + + let double = Arc::new(SharedAuthTestDouble); + let services = Arc::new( + RebornProductAuthServices::new( + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + double.clone() as Arc, + Arc::new(NoopAuthContinuationDispatcher), + ) + .with_flow_record_source( + Arc::new(AlwaysFailingFlowSource) as Arc + ), + ); + + let turn_scope = TurnScope::new_with_owner( + ironclaw_host_api::TenantId::new("test-tenant").expect("tenant"), + Some(ironclaw_host_api::AgentId::new("test-agent").expect("agent")), + None, + ironclaw_host_api::ThreadId::new("test-thread").expect("thread"), + Some(UserId::new("creator-user").expect("owner")), + ); + let owner_user_id = UserId::new("creator-user").expect("owner"); + let run_id = TurnRunId::new(); + + // A valid gate_ref so the validation step is not the rejection point — + // the error must come from the source lookup. + let result = services + .cancel_blocked_auth_flow( + &turn_scope, + &owner_user_id, + run_id, + "gate:source-error-test", + ) + .await; + + assert!( + matches!(result, Err(AuthProductError::BackendUnavailable)), + "BackendUnavailable from flow_record_source must propagate; got: {result:?}" + ); + } } diff --git a/crates/ironclaw_reborn_composition/src/auth_prompt.rs b/crates/ironclaw_reborn_composition/src/auth_prompt.rs index 2a5ba9b67de..0b44a2c10d9 100644 --- a/crates/ironclaw_reborn_composition/src/auth_prompt.rs +++ b/crates/ironclaw_reborn_composition/src/auth_prompt.rs @@ -59,6 +59,35 @@ pub trait AuthChallengeProvider: Send + Sync { ) -> Result, AuthProductError>; } +/// Cancels the durable `AuthFlow` record behind a blocked-auth turn gate. +/// +/// When a Slack run blocked on interactive auth is auto-denied (a non-OAuth +/// challenge the Slack surface can't satisfy), the delivery path cancels the run +/// directly via `TurnCoordinator` rather than through the canonical +/// `AuthInteractionService` deny path (which *resumes* the run with a denied +/// disposition instead of cancelling it). Without this port the underlying +/// `AuthFlow` record lingers non-terminal (`Pending`/`AwaitingUser`) until it +/// expires — see issue #4952. Implemented by `RebornProductAuthServices` when a +/// `flow_record_source` is wired in; a no-op when it isn't. +/// +/// Implementations MUST scope the lookup by caller user, run id, gate ref, and +/// tenant/agent/project/thread, and MUST treat an already-terminal (or absent) +/// flow as a graceful no-op so the OAuth-callback race — where the flow completes +/// just before auto-deny — does not surface an error. +#[async_trait] +pub trait BlockedAuthFlowCanceller: Send + Sync { + /// Cancel the non-terminal auth flow backing `(scope, run_id, gate_ref)`. + /// Returns `Ok(())` when the flow was cancelled, was already terminal, or + /// could not be found (nothing to cancel). + async fn cancel_blocked_auth_flow( + &self, + scope: &TurnScope, + owner_user_id: &UserId, + run_id: TurnRunId, + gate_ref: &str, + ) -> Result<(), AuthProductError>; +} + pub(crate) async fn auth_prompt_view_for_blocked_auth( fallback_owner_user_id: &UserId, scope: &TurnScope, diff --git a/crates/ironclaw_reborn_composition/src/lib.rs b/crates/ironclaw_reborn_composition/src/lib.rs index 4aafad77260..732e4e7323a 100644 --- a/crates/ironclaw_reborn_composition/src/lib.rs +++ b/crates/ironclaw_reborn_composition/src/lib.rs @@ -91,7 +91,7 @@ mod profile_approval_authorization; mod project_filesystem_reader; mod projection; mod trajectory_observer; -pub use auth_prompt::{AuthChallengeProvider, AuthChallengeView}; +pub use auth_prompt::{AuthChallengeProvider, AuthChallengeView, BlockedAuthFlowCanceller}; #[cfg(feature = "slack-v2-host-beta")] mod delivered_gate_routing; #[cfg(feature = "root-llm-provider")] diff --git a/crates/ironclaw_reborn_composition/src/runtime.rs b/crates/ironclaw_reborn_composition/src/runtime.rs index cd84fd18b36..7dcc4a2cf95 100644 --- a/crates/ironclaw_reborn_composition/src/runtime.rs +++ b/crates/ironclaw_reborn_composition/src/runtime.rs @@ -1075,6 +1075,16 @@ impl RebornRuntime { .and_then(|product_auth| product_auth.as_auth_challenge_provider()) } + #[cfg(feature = "slack-v2-host-beta")] + pub(crate) fn blocked_auth_flow_canceller( + &self, + ) -> Option> { + self.services + .product_auth + .as_ref() + .and_then(|product_auth| product_auth.as_blocked_auth_flow_canceller()) + } + pub(crate) fn webui_event_stream(&self) -> Arc { self.projection_services.webui_event_stream() } diff --git a/crates/ironclaw_reborn_composition/src/slack_delivery.rs b/crates/ironclaw_reborn_composition/src/slack_delivery.rs index 3bc9fc1df1d..5d8689119d2 100644 --- a/crates/ironclaw_reborn_composition/src/slack_delivery.rs +++ b/crates/ironclaw_reborn_composition/src/slack_delivery.rs @@ -50,11 +50,11 @@ use serde::{Deserialize, Serialize}; use std::collections::{HashSet, VecDeque}; use tokio::sync::Semaphore; -use crate::AuthChallengeProvider; use crate::auth_prompt::auth_prompt_view_for_blocked_auth; use crate::slack_outbound_targets::{ slack_conversation_id_from_reply_target_binding_ref, slack_reply_target_is_personal_dm, }; +use crate::{AuthChallengeProvider, BlockedAuthFlowCanceller}; const MAX_SLACK_RUN_POLL_INTERVAL: Duration = Duration::from_secs(5); const DEFAULT_TRIGGERED_RUN_DELIVERY_MAX_WAIT: Duration = Duration::from_secs(30 * 60); @@ -127,6 +127,17 @@ pub struct SlackFinalReplyDeliveryServices { /// challenges are surfaced in Slack; other challenge kinds are denied (see the /// `BlockedAuth` arm of `notification_for_actionable_state`). pub auth_challenges: Option>, + /// Cancels the durable `AuthFlow` record whenever a `BlockedAuth` run is + /// auto-cancelled by the Slack delivery path. Threaded through the shared + /// `cancel_auth_blocked_run` helper, so it covers every caller that cancels a + /// blocked-auth run: the live observer non-OAuth deny arm, the triggered + /// non-OAuth deny arm, and the OAuth send-time DM backstop. The Slack path + /// cancels the run directly via `TurnCoordinator` (it does not go through the + /// canonical `AuthInteractionService` deny path), which would otherwise leave + /// the flow record non-terminal (#4952); this cancels the flow alongside the + /// run, after the run cancel succeeds. `None` (e.g. no `flow_record_source` + /// wired in) skips the flow cancel and still cancels the run — backward-compatible. + pub auth_flow_canceller: Option>, /// Store used to resolve an approval gate's request details (tool/action/reason) /// so the Slack approval prompt can say WHAT is being approved — the same /// source the WebUI projection reads. `None` disables the enrichment. @@ -486,6 +497,7 @@ impl SlackFinalReplyDeliveryObserver { scope, TurnActor::new(binding.actor_user_id.clone()), run_id, + gate_ref.as_str(), ) .await?; if let Err(error) = post_slack_message( @@ -517,12 +529,15 @@ impl SlackFinalReplyDeliveryObserver { scope: &TurnScope, actor: TurnActor, run_id: TurnRunId, + gate_ref: &str, ) -> Result<(), SlackFinalReplyDeliveryError> { cancel_auth_blocked_run( self.services.turn_coordinator.as_ref(), + self.services.auth_flow_canceller.as_deref(), scope, actor, run_id, + Some(gate_ref), ) .await } @@ -1140,14 +1155,38 @@ fn slack_approval_gate_prompt_view( /// cancellation contract cannot drift between them. async fn cancel_auth_blocked_run( coordinator: &dyn TurnCoordinator, + auth_flow_canceller: Option<&dyn BlockedAuthFlowCanceller>, scope: &TurnScope, actor: TurnActor, run_id: TurnRunId, + gate_ref: Option<&str>, ) -> Result<(), SlackFinalReplyDeliveryError> { + // Resolve the flow-cancel target BEFORE `cancel_run` consumes `actor`. Owner + // resolution mirrors `auth_prompt_view_for_blocked_auth`: an explicit turn owner + // (shared/team subject) wins, else the acting user. When `gate_ref` is absent + // there is no flow to resolve, so the flow cancel is skipped entirely (not + // encoded as an empty ref). + let flow_cancel_target = match (auth_flow_canceller, gate_ref) { + (Some(canceller), Some(gate_ref)) => { + let owner_user_id = scope + .explicit_owner_user_id() + .unwrap_or(&actor.user_id) + .clone(); + Some((canceller, owner_user_id, gate_ref)) + } + _ => None, + }; + let idempotency_key = ironclaw_turns::IdempotencyKey::new(format!("slack-auth-block:{run_id}")) .map_err(|err| SlackFinalReplyDeliveryError::SlackWebApi { reason: format!("invalid idempotency key for slack auth block: {err}"), })?; + // Cancel the run FIRST — it is the user-visible terminal action. `cancel_run` is + // idempotent (`slack-auth-block:{run_id}`), so repeated passes are safe. If it + // fails we return here and leave the durable `AuthFlow` (and the still-usable + // auth prompt) intact: marking the flow terminal while the run is still + // `BlockedAuth` would be the inverse state drift this fix is meant to prevent, + // and the OAuth backstop relies on a failed cancel leaving the prompt usable. coordinator .cancel_run(ironclaw_turns::CancelRunRequest { scope: scope.clone(), @@ -1157,6 +1196,22 @@ async fn cancel_auth_blocked_run( idempotency_key, }) .await?; + + // Run is now terminal — cancel the stale `AuthFlow` record alongside it (#4952). + // Best-effort cleanliness: a flow-cancel failure does not surface, since the + // run (the user-visible action) has already been cancelled. + if let Some((canceller, owner_user_id, gate_ref)) = flow_cancel_target + && let Err(error) = canceller + .cancel_blocked_auth_flow(scope, &owner_user_id, run_id, gate_ref) + .await + { + tracing::debug!( + target = "ironclaw::reborn::slack_delivery", + %run_id, + %error, + "failed to cancel stale auth flow on Slack auth auto-deny (best-effort)" + ); + } Ok(()) } @@ -1898,6 +1953,7 @@ impl PostSubmitDeliveryHook for TriggeredRunDeliveryDriver { egress: Arc::clone(&self.services.egress), delivery_sink: Arc::clone(&self.services.delivery_sink), auth_challenges: self.services.auth_challenges.clone(), + auth_flow_canceller: self.services.auth_flow_canceller.clone(), approval_requests: self.services.approval_requests.clone(), }; let settings = self.settings; @@ -2161,9 +2217,11 @@ async fn deliver_triggered_run( // deleting anything. if let Err(err) = cancel_auth_blocked_run( services.turn_coordinator.as_ref(), + services.auth_flow_canceller.as_deref(), &scope, actor.clone(), run_id, + state.gate_ref.as_ref().map(|gate_ref| gate_ref.as_str()), ) .await { @@ -2457,9 +2515,11 @@ async fn triggered_notification_for_state( // parked run and post the auth-unavailable notice directly. cancel_auth_blocked_run( services.turn_coordinator.as_ref(), + services.auth_flow_canceller.as_deref(), scope, actor.clone(), run_id, + Some(gate_ref.as_str()), ) .await?; Ok(Some(SlackActionableNotification { @@ -3209,6 +3269,27 @@ mod tests { egress: Arc, outbound: Arc, installation_id: &str, + ) -> SlackFinalReplyDeliveryServices { + make_services_with_canceller( + coordinator, + thread_service, + egress, + outbound, + installation_id, + None, + ) + } + + /// Like [`make_services`] but threads in an explicit `auth_flow_canceller`. + /// Used by triggered-path tests that need to assert `BlockedAuthFlowCanceller` + /// is called (or not called) when the triggered delivery hits a `BlockedAuth` state. + fn make_services_with_canceller( + coordinator: Arc, + thread_service: Arc, + egress: Arc, + outbound: Arc, + installation_id: &str, + auth_flow_canceller: Option>, ) -> SlackFinalReplyDeliveryServices { SlackFinalReplyDeliveryServices { binding_service: Arc::new(TestNoopConversationBindingService), @@ -3221,6 +3302,7 @@ mod tests { egress, delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller, approval_requests: None, } } @@ -3877,6 +3959,16 @@ mod tests { egress: Arc, outbound: Arc, installation_id: &str, + ) -> SlackFinalReplyDeliveryObserver { + make_observer_with_canceller(coordinator, egress, outbound, installation_id, None) + } + + fn make_observer_with_canceller( + coordinator: Arc, + egress: Arc, + outbound: Arc, + installation_id: &str, + auth_flow_canceller: Option>, ) -> SlackFinalReplyDeliveryObserver { use ironclaw_product_workflow::FakeConversationBindingService; @@ -3892,6 +3984,7 @@ mod tests { egress, delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -4390,6 +4483,319 @@ mod tests { ); } + /// Records every `cancel_blocked_auth_flow` call so tests can assert the Slack + /// auto-deny path cancels the durable auth-flow record alongside the run (#4952). + /// + /// Captures all four arguments of `cancel_blocked_auth_flow` so tests can assert + /// that both the wiring (run_id/gate_ref) and the owner-resolution logic + /// (scope/owner_user_id) are correct. Asserting against concrete fixture values + /// catches a wrong-owner regression at production line 1167 that a tuple of + /// `(TurnRunId, String)` would silently miss. + #[derive(Clone)] + struct RecordedFlowCancel { + scope: TurnScope, + owner_user_id: ironclaw_host_api::UserId, + run_id: TurnRunId, + gate_ref: String, + } + + #[derive(Default)] + struct RecordingBlockedAuthFlowCanceller { + calls: std::sync::Mutex>, + } + + #[async_trait] + impl BlockedAuthFlowCanceller for RecordingBlockedAuthFlowCanceller { + async fn cancel_blocked_auth_flow( + &self, + scope: &TurnScope, + owner_user_id: &ironclaw_host_api::UserId, + run_id: TurnRunId, + gate_ref: &str, + ) -> Result<(), ironclaw_auth::AuthProductError> { + self.calls + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .push(RecordedFlowCancel { + scope: scope.clone(), + owner_user_id: owner_user_id.clone(), + run_id, + gate_ref: gate_ref.to_string(), + }); + Ok(()) + } + } + + /// Accepted ack + BlockedAuth (non-OAuth) → the auto-deny cancels the stale + /// auth-flow record (via `BlockedAuthFlowCanceller`) for the blocked gate, not + /// just the run. Drives the live observer caller so a wiring regression — the + /// canceller no longer threaded into `cancel_auth_blocked_run` — is caught. + #[tokio::test] + async fn blocked_auth_cancels_stale_auth_flow() { + let install = "test-install"; + let egress = Arc::new(FakeProtocolHttpEgress::new(vec!["slack.com".to_string()])); + egress.allow_credential_handle("slack_bot_token"); + egress.program_response( + "slack.com", + Ok(EgressResponse::new( + 200, + slack_post_ok_json("D123", "6006.1"), + )), + ); + + let outbound = Arc::new(InMemoryOutboundStateStore::default()); + let coordinator = Arc::new(ScriptedTurnCoordinator::with_states(vec![scripted_state( + TurnStatus::BlockedAuth, + Some("gate:auth-cancel-test"), + )])); + let recorder = Arc::new(RecordingBlockedAuthFlowCanceller::default()); + let observer = make_observer_with_canceller( + Arc::clone(&coordinator) as Arc, + egress.clone(), + outbound, + install, + Some(Arc::clone(&recorder) as Arc), + ); + let env = envelope(user_message_payload()); + let submitted_run_id = TurnRunId::new(); + let ack = ProductInboundAck::Accepted { + accepted_message_ref: AcceptedMessageRef::new("slack:blocked-auth-flow-cancel-test") + .expect("ref"), + submitted_run_id, + }; + + observer.observe_workflow_ack(env, ack).await; + + // Run is still cancelled exactly once... + assert_eq!( + coordinator.cancel_call_count(), + 1, + "BlockedAuth must cancel the run exactly once" + ); + // ...and the stale auth flow is cancelled for the same blocked gate. + let calls = recorder + .calls + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert_eq!( + calls.len(), + 1, + "auto-deny must cancel the stale auth flow exactly once" + ); + assert_eq!( + calls[0].run_id, submitted_run_id, + "canceller must receive the same run_id as the submitted ack" + ); + assert_eq!( + calls[0].gate_ref, "gate:auth-cancel-test", + "must cancel the auth flow for the blocked gate" + ); + // FIX 2: Assert the resolved owner_user_id and scope match the fixture values. + // + // In the live-observer path, `FakeConversationBindingService` derives: + // actor_user_id = "user:{external_actor_ref.id()}" = "user:U123" + // subject_user_id = Some("user:U123") + // That subject_user_id becomes thread_scope.owner_user_id → passed as the + // explicit owner to `TurnScope::new_with_owner`, so + // `scope.explicit_owner_user_id() = Some("user:U123")` which wins over + // actor.user_id in `cancel_auth_blocked_run` (production line 1167). + let expected_owner = + ironclaw_host_api::UserId::new("user:U123").expect("expected owner fixture"); + assert_eq!( + calls[0].owner_user_id, expected_owner, + "owner_user_id must be the subject user derived from the external actor ref (U123)" + ); + // Scope tenant must match what FakeConversationBindingService builds from + // installation_id "install_alpha". + let expected_tenant = + ironclaw_host_api::TenantId::new("tenant:install_alpha").expect("expected tenant"); + assert_eq!( + calls[0].scope.tenant_id, expected_tenant, + "scope.tenant_id must match the tenant derived from the installation" + ); + } + + /// FIX 3: A failed `cancel_run` must leave the `AuthFlow` record intact. + /// + /// `cancel_auth_blocked_run` was reordered so the run is cancelled FIRST and + /// the durable `AuthFlow` is only marked terminal AFTER a successful cancel. + /// This test proves the invariant: when `cancel_run` returns `Err`, the + /// `BlockedAuthFlowCanceller` is NOT invoked — preventing inverse state drift + /// (a terminal `AuthFlow` whose corresponding run is still `BlockedAuth`). + /// + /// Drives the live-observer path (`SlackFinalReplyDeliveryObserver`) with a + /// `ScriptedTurnCoordinator` whose `cancel_should_fail` flag is set, mirroring + /// the mechanism used in `triggered_oauth_auth_backstop_cancel_failure_records_failed`. + #[tokio::test] + async fn blocked_auth_cancel_run_failure_leaves_auth_flow_intact() { + let install = "test-install"; + let egress = Arc::new(FakeProtocolHttpEgress::new(vec!["slack.com".to_string()])); + egress.allow_credential_handle("slack_bot_token"); + // No HTTP response programmed: the cancel fails before any post is made. + + let outbound = Arc::new(InMemoryOutboundStateStore::default()); + let coordinator = Arc::new(ScriptedTurnCoordinator::with_states(vec![scripted_state( + TurnStatus::BlockedAuth, + Some("gate:cancel-fail-intact"), + )])); + // Make cancel_run fail — mirrors the mechanism in + // `triggered_oauth_auth_backstop_cancel_failure_records_failed`. + coordinator + .cancel_should_fail + .store(true, std::sync::atomic::Ordering::Release); + + let recorder = Arc::new(RecordingBlockedAuthFlowCanceller::default()); + let observer = make_observer_with_canceller( + Arc::clone(&coordinator) as Arc, + egress.clone(), + outbound, + install, + Some(Arc::clone(&recorder) as Arc), + ); + let env = envelope(user_message_payload()); + let submitted_run_id = TurnRunId::new(); + let ack = ProductInboundAck::Accepted { + accepted_message_ref: AcceptedMessageRef::new("slack:cancel-fail-intact-test") + .expect("ref"), + submitted_run_id, + }; + + observer.observe_workflow_ack(env, ack).await; + + // cancel_run was attempted (it just failed). + assert_eq!( + coordinator.cancel_call_count(), + 1, + "cancel_run must be attempted exactly once even when it fails" + ); + // The flow canceller must NOT have been called: a failed run-cancel must + // leave the durable AuthFlow record intact so the auth prompt remains usable. + let calls = recorder + .calls + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert!( + calls.is_empty(), + "BlockedAuthFlowCanceller must NOT be called when cancel_run fails; got {} call(s)", + calls.len() + ); + } + + /// A `BlockedAuthFlowCanceller` that always returns `Err(BackendUnavailable)`. + /// Used to assert that a flow-cancel error is swallowed and does not break + /// Slack auto-denial delivery. + /// + /// `call_count` is incremented atomically on every `cancel_blocked_auth_flow` + /// invocation so tests can assert the canceller was actually wired and called. + struct FailingBlockedAuthFlowCanceller { + call_count: std::sync::atomic::AtomicUsize, + } + + impl FailingBlockedAuthFlowCanceller { + fn new() -> Self { + Self { + call_count: std::sync::atomic::AtomicUsize::new(0), + } + } + + fn call_count(&self) -> usize { + self.call_count.load(std::sync::atomic::Ordering::SeqCst) + } + } + + #[async_trait] + impl BlockedAuthFlowCanceller for FailingBlockedAuthFlowCanceller { + async fn cancel_blocked_auth_flow( + &self, + _scope: &ironclaw_turns::TurnScope, + _owner_user_id: &ironclaw_host_api::UserId, + _run_id: TurnRunId, + _gate_ref: &str, + ) -> Result<(), ironclaw_auth::AuthProductError> { + self.call_count + .fetch_add(1, std::sync::atomic::Ordering::SeqCst); + Err(ironclaw_auth::AuthProductError::BackendUnavailable) + } + } + + /// A flow-cancel failure must be swallowed: a failing `BlockedAuthFlowCanceller` + /// must not break Slack auto-denial. + /// + /// After `cancel_run` succeeds, `cancel_auth_blocked_run` attempts a best-effort + /// `cancel_blocked_auth_flow`. When that returns `Err`, the error is debug-logged + /// and the function still returns `Ok(())` — so the `SLACK_AUTH_UNAVAILABLE_MESSAGE` + /// post still goes out and the coordinator cancel count is still 1. + #[tokio::test] + async fn blocked_auth_canceller_failure_is_swallowed() { + let install = "test-install"; + let egress = Arc::new(FakeProtocolHttpEgress::new(vec!["slack.com".to_string()])); + egress.allow_credential_handle("slack_bot_token"); + egress.program_response( + "slack.com", + Ok(EgressResponse::new( + 200, + slack_post_ok_json("D123", "6007.1"), + )), + ); + + let outbound = Arc::new(InMemoryOutboundStateStore::default()); + // cancel_run SUCCEEDS (cancel_should_fail is NOT set, matching the default). + let coordinator = Arc::new(ScriptedTurnCoordinator::with_states(vec![scripted_state( + TurnStatus::BlockedAuth, + Some("gate:auth-cancel-test"), + )])); + // Wire in a canceller that always fails — swallow path under test. + // Hold a clone of the Arc so we can inspect call_count after the observer runs. + let failing_canceller = Arc::new(FailingBlockedAuthFlowCanceller::new()); + let observer = make_observer_with_canceller( + Arc::clone(&coordinator) as Arc, + egress.clone(), + outbound, + install, + Some(Arc::clone(&failing_canceller) as Arc), + ); + let env = envelope(user_message_payload()); + let ack = ProductInboundAck::Accepted { + accepted_message_ref: AcceptedMessageRef::new("slack:canceller-fail-swallowed-test") + .expect("ref"), + submitted_run_id: TurnRunId::new(), + }; + + observer.observe_workflow_ack(env, ack).await; + + // The canceller must have been invoked exactly once — proving it is wired up. + assert_eq!( + failing_canceller.call_count(), + 1, + "cancel_blocked_auth_flow must be called exactly once on the failing canceller" + ); + + // The run was still cancelled exactly once — flow-cancel failure does not + // prevent run cancellation or the auto-denial post. + assert_eq!( + coordinator.cancel_call_count(), + 1, + "cancel_run must be called exactly once even when flow-cancel fails" + ); + + // The SLACK_AUTH_UNAVAILABLE_MESSAGE post must still go out. + let calls = egress.calls(); + let post_calls: Vec<_> = calls + .iter() + .filter(|c| c.path == "/api/chat.postMessage") + .collect(); + assert_eq!( + post_calls.len(), + 1, + "expected exactly one chat.postMessage despite flow-cancel failure" + ); + let body = std::str::from_utf8(&post_calls[0].body).expect("utf8 body"); + assert!( + body.contains(SLACK_AUTH_UNAVAILABLE_MESSAGE), + "body must contain SLACK_AUTH_UNAVAILABLE_MESSAGE text, got: {body}" + ); + } + /// DeferredBusy + UserMessage + BlockedApproval with no gate_ref → fallback wording /// without a specific gate command. #[tokio::test] @@ -4507,6 +4913,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -4968,6 +5375,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -5046,6 +5454,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -5324,6 +5733,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -5907,6 +6317,114 @@ mod tests { ); } + /// Triggered non-OAuth `BlockedAuth` → `cancel_auth_blocked_run` invokes the + /// `BlockedAuthFlowCanceller` for the blocked gate (#4952). + /// + /// Drives the same `triggered_notification_for_state` non-OAuth branch as + /// `triggered_non_oauth_auth_denial_records_delivered`, but this time a + /// `RecordingBlockedAuthFlowCanceller` is wired so we can assert the stale + /// auth-flow record is cancelled. + #[tokio::test] + async fn triggered_non_oauth_auth_cancels_stale_auth_flow() { + let install = "test-install"; + let gate_ref_str = "gate:triggered-non-oauth-stale-flow"; + let scope = personal_turn_scope(); + let run_id = TurnRunId::new(); + let binding_ref = + test_slack_binding_ref(install, scope.agent_id.as_ref().expect("agent").as_str()); + + // First poll → BlockedAuth (non-OAuth: no auth_challenges, so no + // authorization_url → deny branch in triggered_notification_for_state). + // Second poll → Cancelled (terminal, no message → Ok(None)). + let coordinator = Arc::new(ScriptedTurnCoordinator::with_states(vec![ + scripted_state(TurnStatus::BlockedAuth, Some(gate_ref_str)), + scripted_state(TurnStatus::Cancelled, None), + ])); + + let outbound = Arc::new(InMemoryOutboundStateStore::default()); + seed_personal_preference(&outbound, &scope, binding_ref).await; + + let egress = Arc::new(FakeProtocolHttpEgress::new(vec!["slack.com".to_string()])); + egress.allow_credential_handle("slack_bot_token"); + // One postMessage for the auth-unavailable notice. + egress.program_response( + "slack.com", + Ok(EgressResponse::new( + 200, + slack_post_ok_json("D456", "noa1.1"), + )), + ); + + let recorder = Arc::new(RecordingBlockedAuthFlowCanceller::default()); + let delivery_store = Arc::new(InMemoryTriggeredRunDeliveryStore::default()); + let route_store = Arc::new(InMemoryDeliveredGateRouteStore::default()); + let services = make_services_with_canceller( + coordinator.clone(), + Arc::new(InMemorySessionThreadService::default()), + egress.clone(), + outbound, + install, + Some(Arc::clone(&recorder) as Arc), + ); + let settings = SlackFinalReplyDeliverySettings { + poll_interval: std::time::Duration::ZERO, + max_wait: std::time::Duration::from_secs(5), + max_concurrent_deliveries: NonZeroUsize::new(1).unwrap(), + max_pending_deliveries: NonZeroUsize::new(8).unwrap(), + }; + let driver = TriggeredRunDeliveryDriver::with_settings( + services, + settings, + delivery_store.clone(), + route_store.clone(), + scope.agent_id.clone().expect("test scope has agent"), + ); + + let fire = minimal_trigger_fire(None); + driver + .on_trigger_submitted(fire, run_id, scope.clone()) + .await; + wait_for_delivery_record(&delivery_store, run_id).await; + + // The stale auth flow must have been cancelled exactly once. + let calls = recorder + .calls + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert_eq!( + calls.len(), + 1, + "triggered non-OAuth auth deny must cancel the stale auth flow exactly once; got {} calls", + calls.len() + ); + assert_eq!( + calls[0].run_id, run_id, + "canceller must receive the triggered run's run_id" + ); + assert_eq!( + calls[0].gate_ref, gate_ref_str, + "canceller must receive the blocked gate_ref" + ); + // FIX 2: Assert the resolved owner_user_id and scope match the fixture values. + // + // In the triggered path, `deliver_triggered_run` builds: + // actor = TurnActor::new(fire.creator_user_id) = "creator-user" + // `personal_turn_scope()` sets explicit owner = "creator-user", so + // `scope.explicit_owner_user_id() = Some("creator-user")` wins at + // production line 1167 (`cancel_auth_blocked_run` owner resolution). + let expected_owner = + ironclaw_host_api::UserId::new("creator-user").expect("expected owner fixture"); + assert_eq!( + calls[0].owner_user_id, expected_owner, + "owner_user_id must be the scope's explicit owner (creator-user from personal_turn_scope)" + ); + // Scope tenant must match personal_turn_scope(). + assert_eq!( + calls[0].scope.tenant_id, scope.tenant_id, + "scope.tenant_id must match the personal_turn_scope tenant" + ); + } + /// OAuth backstop cancel-failure path: when `cancel_auth_blocked_run` fails in /// the `OAuthTargetNotDm` error arm, the outcome must be `Failed` and NO /// `/api/chat.delete` calls must be made (we must not strip the auth prompt @@ -6012,6 +6530,108 @@ mod tests { ); } + /// OAuth `OAuthTargetNotDm` backstop → `BlockedAuthFlowCanceller` is invoked + /// to cancel the stale auth-flow record alongside the run (#4952). + /// + /// Models on `triggered_oauth_auth_backstop_cancel_failure_records_failed` but + /// wires a `RecordingBlockedAuthFlowCanceller` (no cancel_run failure) so the + /// backstop succeeds and we can assert the canceller was called for the correct + /// gate_ref. + #[tokio::test] + async fn triggered_oauth_backstop_cancels_stale_auth_flow() { + let install = "test-install"; + let gate_ref_str = "gate:oauth-backstop-stale-flow"; + let scope = personal_turn_scope(); + let run_id = TurnRunId::new(); + + // Use a SHARED CHANNEL binding ref so the OAuth backstop trips + // (OAuthTargetNotDm is returned before any HTTP post is made). + let binding_ref = test_slack_shared_channel_binding_ref( + install, + scope.agent_id.as_ref().expect("agent").as_str(), + ); + + // First poll → BlockedAuth; second poll → Cancelled (after cancel_run). + let coordinator = Arc::new(ScriptedTurnCoordinator::with_states(vec![ + scripted_state(TurnStatus::BlockedAuth, Some(gate_ref_str)), + scripted_state(TurnStatus::Cancelled, None), + ])); + // cancel_run must succeed so the backstop posts the unavailable notice. + + let outbound = Arc::new(InMemoryOutboundStateStore::default()); + // Seed preference with the shared-channel binding so the OAuth guard trips. + seed_personal_preference(&outbound, &scope, binding_ref).await; + + let egress = Arc::new(FakeProtocolHttpEgress::new(vec!["slack.com".to_string()])); + egress.allow_credential_handle("slack_bot_token"); + // One postMessage for the auth-unavailable notice (the backstop posts + // after a successful cancel_run). + egress.program_response( + "slack.com", + Ok(EgressResponse::new( + 200, + slack_post_ok_json("C0SHARED", "obs2.1"), + )), + ); + + let recorder = Arc::new(RecordingBlockedAuthFlowCanceller::default()); + let delivery_store = Arc::new(InMemoryTriggeredRunDeliveryStore::default()); + let route_store = Arc::new(InMemoryDeliveredGateRouteStore::default()); + let mut services = make_services_with_canceller( + coordinator.clone(), + Arc::new(InMemorySessionThreadService::default()), + egress.clone(), + outbound, + install, + Some(Arc::clone(&recorder) as Arc), + ); + // Wire up an OAuth challenge provider so the BlockedAuth state generates + // an authorization_url, triggering the DM-only guard. + services.auth_challenges = Some(Arc::new(OAuthAuthChallengeProvider { + url: "https://provider.example/oauth-backstop-stale".to_string(), + })); + + let settings = SlackFinalReplyDeliverySettings { + poll_interval: std::time::Duration::ZERO, + max_wait: std::time::Duration::from_secs(5), + max_concurrent_deliveries: NonZeroUsize::new(1).unwrap(), + max_pending_deliveries: NonZeroUsize::new(8).unwrap(), + }; + let driver = TriggeredRunDeliveryDriver::with_settings( + services, + settings, + delivery_store.clone(), + route_store.clone(), + scope.agent_id.clone().expect("test scope has agent"), + ); + + let fire = minimal_trigger_fire(None); + driver + .on_trigger_submitted(fire, run_id, scope.clone()) + .await; + wait_for_delivery_record(&delivery_store, run_id).await; + + // The stale auth flow must have been cancelled exactly once. + let calls = recorder + .calls + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + assert_eq!( + calls.len(), + 1, + "OAuth backstop must cancel the stale auth flow exactly once; got {} calls", + calls.len() + ); + assert_eq!( + calls[0].run_id, run_id, + "canceller must receive the triggered run's run_id" + ); + assert_eq!( + calls[0].gate_ref, gate_ref_str, + "canceller must receive the blocked gate_ref" + ); + } + // ── DM-gate security tests ───────────────────────────────────────────────── // // These tests verify the fail-closed gate that prevents OAuth @@ -6660,6 +7280,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -6726,6 +7347,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -7055,6 +7677,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { @@ -7139,6 +7762,7 @@ mod tests { egress: egress.clone(), delivery_sink: Arc::new(FakeOutboundDeliverySink::default()), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }; let settings = SlackFinalReplyDeliverySettings { diff --git a/crates/ironclaw_reborn_composition/src/slack_host_beta.rs b/crates/ironclaw_reborn_composition/src/slack_host_beta.rs index a4514dca8f1..b41c32996fd 100644 --- a/crates/ironclaw_reborn_composition/src/slack_host_beta.rs +++ b/crates/ironclaw_reborn_composition/src/slack_host_beta.rs @@ -419,6 +419,7 @@ pub fn build_triggered_run_delivery_hook( egress, delivery_sink, auth_challenges: runtime.auth_challenge_provider(), + auth_flow_canceller: runtime.blocked_auth_flow_canceller(), approval_requests: Some(Arc::clone(&local_runtime.approval_requests) as Arc), }; @@ -753,6 +754,7 @@ fn build_slack_events_route_mount_with_resolvers( egress, delivery_sink, auth_challenges: runtime.auth_challenge_provider(), + auth_flow_canceller: runtime.blocked_auth_flow_canceller(), approval_requests: Some(Arc::clone(&local_runtime.approval_requests) as Arc), }, diff --git a/crates/ironclaw_reborn_composition/src/slack_serve/e2e_tests.rs b/crates/ironclaw_reborn_composition/src/slack_serve/e2e_tests.rs index 00e477dd980..91b814b1902 100644 --- a/crates/ironclaw_reborn_composition/src/slack_serve/e2e_tests.rs +++ b/crates/ironclaw_reborn_composition/src/slack_serve/e2e_tests.rs @@ -331,6 +331,7 @@ async fn build_harness_with_full_settings( egress: Arc::new(egress.clone()), delivery_sink: Arc::new(sink), auth_challenges, + auth_flow_canceller: None, approval_requests: None, }, SlackFinalReplyDeliverySettings { @@ -546,6 +547,7 @@ async fn build_harness_for_delivered_route_tests_with_store_mode( egress: Arc::new(egress.clone()), delivery_sink: Arc::new(sink), auth_challenges: None, + auth_flow_canceller: None, approval_requests: None, }, SlackFinalReplyDeliverySettings { @@ -2504,6 +2506,7 @@ async fn build_harness_for_auth_fanout_test( egress: Arc::new(egress.clone()), delivery_sink: Arc::new(sink), auth_challenges: Some(auth_challenges), + auth_flow_canceller: None, approval_requests: None, }, SlackFinalReplyDeliverySettings {