fix(auth): implement deny-protected fail-safe mode

Signed-off-by: Cea Stapleton Cordasco <261786559+cea-block@users.noreply.github.com>

crates/buzz-relay/src/api/invites.rs

@@ -284,11 +284,16 @@ async fn authenticate(
         true, // POST bodies must be covered by a payload tag
         tenant.community(),
     )?;
-    if state
+    let authorization_mode = state
         .protected_transport()
-        .and_then(|runtime| runtime.mode_for_domain(tenant.community()))
-        != Some(AuthorizationMode::Enforce)
-    {
+        .and_then(|runtime| runtime.mode_for_domain(tenant.community()));
+    if authorization_mode == Some(AuthorizationMode::DenyProtected) {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            "protected authorization denied",
+        ));
+    }
+    if authorization_mode != Some(AuthorizationMode::Enforce) {
         bridge::check_nip98_replay(state, &tenant, event_id_bytes).await?;
     }
 
@@ -556,10 +561,16 @@ pub async fn claim_invite(
     let (tenant, pubkey, identity_proof, verified_proof, enrollment_assertion) =
         authenticate(&state, &headers, "/api/invites/claim", &body).await?;
 
-    let enforcing = state
+    let authorization_mode = state
         .protected_transport()
-        .and_then(|runtime| runtime.mode_for_domain(tenant.community()))
-        == Some(AuthorizationMode::Enforce);
+        .and_then(|runtime| runtime.mode_for_domain(tenant.community()));
+    if authorization_mode == Some(AuthorizationMode::DenyProtected) {
+        return Err(api_error(
+            StatusCode::FORBIDDEN,
+            "protected authorization denied",
+        ));
+    }
+    let enforcing = authorization_mode == Some(AuthorizationMode::Enforce);
 
     if claim_rate_limited(&state, tenant.community(), &pubkey) {
         return Err(api_error(

crates/buzz-relay/src/authorization_runtime/finalization.rs

@@ -0,0 +1,632 @@
+//! Relay adapter for exact-domain authorization policy and finalization.
+//!
+//! A request never selects its provider profile or activation mode. The relay
+//! resolves one immutable policy from the row-zero [`TenantContext`] domain.
+//! Missing and duplicate domain configuration fail closed without a global or
+//! Nostr-only fallback.
+
+use std::{collections::HashMap, fmt, sync::Arc};
+
+use buzz_auth::{
+    resolve_authorization, AccessLeasePolicy, AuthorizationFinalizer, AuthorizationOutcome,
+    AuthorizationProfileId, AuthorizationProvider, AuthorizationRequest, BindingLeaseBound,
+    CapabilitySet, DecisionSource, EnrollmentMode, FederatedAuthorization, FederatedPrincipal,
+    FinalizationError, LeaseVersion, PolicyVersion, ProviderAuthorizationClock,
+    ProviderContractError, ProviderTimeout, ResolvedFederatedPolicy, SharedAuthorizationClock,
+    VerificationOnlyDisposition, VerificationStatusPolicy, VerifiedFederatedAssertion,
+    VerifiedNostrProof, VersionedBindingRef,
+};
+use buzz_core::{tenant::TenantContext, CommunityId};
+use thiserror::Error;
+use uuid::Uuid;
+
+/// Server-owned activation mode for one exact authorization domain.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AuthorizationMode {
+    /// Do not evaluate federated identity or provider policy.
+    Off,
+    /// Evaluate read-only provider policy without binding or authority changes.
+    Shadow,
+    /// Produce a short-lived display-only result after full direct finalization.
+    VerifyOnly,
+    /// Issue bounded access leases after full direct or delegated finalization.
+    Enforce,
+    /// Keep every protected surface active while denying all protected access.
+    DenyProtected,
+}
+
+impl AuthorizationMode {
+    /// Whether this mode may evaluate the configured admission provider.
+    pub const fn evaluates_provider(self) -> bool {
+        matches!(self, Self::Shadow | Self::VerifyOnly | Self::Enforce)
+    }
+
+    /// Whether this mode keeps the protected-surface inventory authoritative.
+    pub const fn protects_surfaces(self) -> bool {
+        matches!(self, Self::Enforce | Self::DenyProtected)
+    }
+}
+
+/// Immutable server configuration for one exact authorization domain.
+#[derive(Clone)]
+pub struct DomainAuthorizationPolicy {
+    authorization_domain: CommunityId,
+    profile_id: AuthorizationProfileId,
+    provider: Arc<dyn AuthorizationProvider>,
+    enrollment_mode: EnrollmentMode,
+    mode: AuthorizationMode,
+    provider_timeout: ProviderTimeout,
+    access_lease_policy: AccessLeasePolicy,
+    verification_status_policy: VerificationStatusPolicy,
+}
+
+impl DomainAuthorizationPolicy {
+    /// Build policy exclusively from trusted server configuration.
+    #[allow(clippy::too_many_arguments)]
+    pub fn from_server_configuration(
+        authorization_domain: CommunityId,
+        profile_id: impl Into<String>,
+        provider: Arc<dyn AuthorizationProvider>,
+        enrollment_mode: EnrollmentMode,
+        mode: AuthorizationMode,
+        provider_timeout: ProviderTimeout,
+        access_lease_policy: AccessLeasePolicy,
+        verification_status_policy: VerificationStatusPolicy,
+    ) -> Result<Self, ProviderContractError> {
+        Ok(Self {
+            authorization_domain,
+            profile_id: AuthorizationProfileId::from_server_configuration(profile_id)?,
+            provider,
+            enrollment_mode,
+            mode,
+            provider_timeout,
+            access_lease_policy,
+            verification_status_policy,
+        })
+    }
+
+    /// Exact server-owned authorization domain.
+    pub const fn authorization_domain(&self) -> CommunityId {
+        self.authorization_domain
+    }
+
+    /// Exact server-owned activation mode.
+    pub const fn mode(&self) -> AuthorizationMode {
+        self.mode
+    }
+
+    /// Server-resolved provider profile.
+    pub const fn profile_id(&self) -> &AuthorizationProfileId {
+        &self.profile_id
+    }
+
+    /// Server-resolved binding enrollment mode.
+    pub const fn enrollment_mode(&self) -> EnrollmentMode {
+        self.enrollment_mode
+    }
+}
+
+impl fmt::Debug for DomainAuthorizationPolicy {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("DomainAuthorizationPolicy")
+            .field("authorization_domain", &"[redacted]")
+            .field("profile_id", &"[redacted]")
+            .field("provider", &"[configured]")
+            .field("enrollment_mode", &"[redacted]")
+            .field("mode", &self.mode)
+            .field("provider_timeout", &"[redacted]")
+            .field("access_lease_policy", &"[redacted]")
+            .field("verification_status_policy", &"[redacted]")
+            .finish()
+    }
+}
+
+/// Immutable exact-domain provider selector.
+#[derive(Clone)]
+pub struct DomainProviderSelector {
+    policies: HashMap<CommunityId, DomainAuthorizationPolicy>,
+}
+
+impl DomainProviderSelector {
+    /// Build an exact-domain selector, rejecting every ambiguous duplicate.
+    pub fn new(
+        policies: impl IntoIterator<Item = DomainAuthorizationPolicy>,
+    ) -> Result<Self, DomainPolicyError> {
+        let mut by_domain = HashMap::new();
+        for policy in policies {
+            let domain = policy.authorization_domain;
+            if by_domain.insert(domain, policy).is_some() {
+                return Err(DomainPolicyError::AmbiguousDomainPolicy);
+            }
+        }
+        Ok(Self {
+            policies: by_domain,
+        })
+    }
+
+    /// Resolve policy only from the row-zero server tenant.
+    ///
+    /// No default provider exists. A federated authorization attempt for an
+    /// unconfigured domain is denied as missing policy.
+    pub fn resolve(
+        &self,
+        tenant: &TenantContext,
+    ) -> Result<&DomainAuthorizationPolicy, DomainPolicyError> {
+        self.policies
+            .get(&tenant.community())
+            .ok_or(DomainPolicyError::MissingDomainPolicy)
+    }
+}
+
+impl fmt::Debug for DomainProviderSelector {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("DomainProviderSelector")
+            .field("policies", &"[redacted]")
+            .finish()
+    }
+}
+
+/// Fail-closed exact-domain policy resolution error.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+pub enum DomainPolicyError {
+    /// More than one authorization policy named the same exact domain.
+    #[error("federated authorization policy is ambiguous for this domain")]
+    AmbiguousDomainPolicy,
+    /// No federated policy was configured for this exact domain.
+    #[error("federated authorization policy is missing for this domain")]
+    MissingDomainPolicy,
+}
+
+/// Provider-neutral relay finalizer with immutable policy and injected time.
+#[derive(Clone)]
+pub struct RelayAuthorizationFinalizer {
+    selector: DomainProviderSelector,
+    finalizer: AuthorizationFinalizer,
+    clock: SharedAuthorizationClock,
+    runtime_binding: Uuid,
+}
+
+struct RelayProviderClock<'a>(&'a dyn buzz_auth::AuthorizationClock);
+
+impl ProviderAuthorizationClock for RelayProviderClock<'_> {
+    fn now_unix_seconds(&self) -> Option<u64> {
+        self.0.now().ok().map(|value| value.unix_seconds())
+    }
+}
+
+impl RelayAuthorizationFinalizer {
+    /// Build a runtime that shares one central clock across evaluation and finalization.
+    pub fn new(selector: DomainProviderSelector, clock: SharedAuthorizationClock) -> Self {
+        Self {
+            selector,
+            finalizer: AuthorizationFinalizer::new(Arc::clone(&clock)),
+            clock,
+            runtime_binding: Uuid::new_v4(),
+        }
+    }
+
+    /// Evaluate current direct provider admission for a server-resolved domain.
+    ///
+    /// Off mode performs no provider call. All other modes preserve provider
+    /// deny and unavailable outcomes without falling back to another policy.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn evaluate_direct(
+        &self,
+        tenant: &TenantContext,
+        proof: &VerifiedNostrProof,
+        assertion: &VerifiedFederatedAssertion,
+        federated_policy: ResolvedFederatedPolicy,
+        requested_capabilities: CapabilitySet,
+        correlation_id: Uuid,
+    ) -> Result<AuthorizationOutcome, RelayFinalizationError> {
+        let policy = self.selector.resolve(tenant)?;
+        if !policy.mode.evaluates_provider() {
+            return Err(RelayFinalizationError::ModeDoesNotEvaluate);
+        }
+        let now = self.finalizer.now()?;
+        let request = AuthorizationRequest::direct(
+            proof,
+            assertion,
+            federated_policy,
+            requested_capabilities,
+            correlation_id,
+            now.unix_seconds(),
+        )?;
+        Ok(resolve_authorization(
+            policy.provider.as_ref(),
+            &request,
+            &RelayProviderClock(self.clock.as_ref()),
+            policy.provider_timeout,
+            self.runtime_binding,
+        )
+        .await)
+    }
+
+    /// Evaluate current delegated-owner provider admission for an exact domain.
+    pub async fn evaluate_delegated(
+        &self,
+        tenant: &TenantContext,
+        proof: &VerifiedNostrProof,
+        owner: &VersionedBindingRef,
+        federated_policy: ResolvedFederatedPolicy,
+        requested_capabilities: CapabilitySet,
+        correlation_id: Uuid,
+    ) -> Result<AuthorizationOutcome, RelayFinalizationError> {
+        let policy = self.selector.resolve(tenant)?;
+        if !policy.mode.evaluates_provider() {
+            return Err(RelayFinalizationError::ModeDoesNotEvaluate);
+        }
+        let now = self.finalizer.now()?;
+        let request = AuthorizationRequest::delegated_from_active_binding(
+            proof,
+            owner,
+            federated_policy,
+            requested_capabilities,
+            correlation_id,
+            now.unix_seconds(),
+        )?;
+        Ok(resolve_authorization(
+            policy.provider.as_ref(),
+            &request,
+            &RelayProviderClock(self.clock.as_ref()),
+            policy.provider_timeout,
+            self.runtime_binding,
+        )
+        .await)
+    }
+
+    /// Finalize one validated allow snapshot according to server-owned mode.
+    ///
+    /// Off and shadow modes cannot finalize a binding, status, or access
+    /// context. Verify-only returns a distinct display type; enforce is the
+    /// only branch capable of returning an access context with a lease.
+    pub fn finalize_allowed(
+        &self,
+        input: buzz_auth::AuthContextInput,
+        federated_policy: ResolvedFederatedPolicy,
+        authorization: FederatedAuthorization,
+        snapshot: Box<buzz_auth::CapabilitySnapshot>,
+        binding_bound: BindingLeaseBound,
+        lease_version: LeaseVersion,
+    ) -> Result<RuntimeAuthorizationDisposition, RelayFinalizationError> {
+        let policy = self.selector.resolve(input.tenant())?;
+        match policy.mode {
+            AuthorizationMode::Off
+            | AuthorizationMode::Shadow
+            | AuthorizationMode::DenyProtected => Err(RelayFinalizationError::ModeDoesNotFinalize),
+            AuthorizationMode::VerifyOnly => self
+                .finalizer
+                .finalize_verification_only(
+                    input,
+                    federated_policy,
+                    authorization,
+                    snapshot,
+                    &policy.profile_id,
+                    binding_bound,
+                    policy.verification_status_policy,
+                )
+                .map(RuntimeAuthorizationDisposition::VerificationOnly)
+                .map_err(Into::into),
+            AuthorizationMode::Enforce => self
+                .finalizer
+                .finalize_access(
+                    input,
+                    federated_policy,
+                    authorization,
+                    snapshot,
+                    &policy.profile_id,
+                    binding_bound,
+                    policy.access_lease_policy,
+                    lease_version,
+                )
+                .map(|context| RuntimeAuthorizationDisposition::Access(Box::new(context)))
+                .map_err(Into::into),
+        }
+    }
+
+    /// Finalize the same current direct evidence into a short-lived,
+    /// display-only status. The caller separately proves the presentation
+    /// gate; this method cannot issue access or a lease and performs no writes.
+    pub fn finalize_client_status(
+        &self,
+        input: buzz_auth::AuthContextInput,
+        federated_policy: ResolvedFederatedPolicy,
+        authorization: FederatedAuthorization,
+        snapshot: Box<buzz_auth::CapabilitySnapshot>,
+        binding_bound: BindingLeaseBound,
+    ) -> Result<VerificationOnlyDisposition, RelayFinalizationError> {
+        let policy = self.selector.resolve(input.tenant())?;
+        if !matches!(
+            policy.mode,
+            AuthorizationMode::VerifyOnly | AuthorizationMode::Enforce
+        ) {
+            return Err(RelayFinalizationError::ModeDoesNotFinalize);
+        }
+        self.finalizer
+            .finalize_verification_only(
+                input,
+                federated_policy,
+                authorization,
+                snapshot,
+                &policy.profile_id,
+                binding_bound,
+                policy.verification_status_policy,
+            )
+            .map_err(Into::into)
+    }
+
+    /// Finalize direct first-enrollment evidence without creating a binding.
+    pub fn finalize_enrollment(
+        &self,
+        tenant: &TenantContext,
+        proof: &VerifiedNostrProof,
+        assertion: &VerifiedFederatedAssertion,
+        federated_policy: ResolvedFederatedPolicy,
+        snapshot: Box<buzz_auth::CapabilitySnapshot>,
+        correlation_id: Uuid,
+    ) -> Result<EnrollmentDisposition, RelayFinalizationError> {
+        let policy = self.selector.resolve(tenant)?;
+        if policy.mode != AuthorizationMode::Enforce {
+            return Err(RelayFinalizationError::ModeDoesNotFinalize);
+        }
+        if policy.enrollment_mode != EnrollmentMode::AttestedKey {
+            return Err(RelayFinalizationError::EnrollmentModeUnsupported);
+        }
+        let now = self.finalizer.now()?;
+        let key = assertion
+            .key_attestation()
+            .ok_or(RelayFinalizationError::EnrollmentEvidenceMismatch)?;
+        if proof.verified_delegation().is_some()
+            || proof.authorization_domain() != tenant.community()
+            || federated_policy.authorization_domain() != tenant.community()
+            || !snapshot.is_bound_to_federated_policy(&federated_policy)
+            || assertion.authorization_domain() != tenant.community()
+            || snapshot.authorization_domain() != tenant.community()
+            || proof.authorized_transport() != assertion.authorized_transport()
+            || snapshot.transport() != proof.authorized_transport()
+            || snapshot.actor_pubkey() != proof.actor_pubkey()
+            || key.pubkey() != proof.actor_pubkey()
+            || snapshot.owner_pubkey().is_some()
+            || snapshot.binding_id().is_some()
+            || snapshot.binding_version().is_some()
+            || snapshot.proof_method() != proof.proof_method()
+            || snapshot.principal() != assertion.principal()
+            || snapshot.profile_id() != &policy.profile_id
+            || snapshot.decision_source() != DecisionSource::DirectAssertion
+            || snapshot.correlation_id() != correlation_id
+            || !snapshot
+                .capabilities()
+                .contains(buzz_auth::AuthorizationCapability::InviteClaim)
+            || snapshot.issued_at() > now.unix_seconds()
+            || snapshot.fresh_until() <= now.unix_seconds()
+            || snapshot.effective_until() <= now.unix_seconds()
+            || assertion
+                .not_before()
+                .is_some_and(|bound| bound.is_not_yet_valid_at(now.unix_seconds()))
+            || assertion.expires_at().is_expired_at(now.unix_seconds())
+        {
+            return Err(RelayFinalizationError::EnrollmentEvidenceMismatch);
+        }
+        let application_until = now
+            .unix_seconds()
+            .checked_add(policy.access_lease_policy.application_limit().seconds())
+            .ok_or(RelayFinalizationError::EnrollmentEvidenceMismatch)?;
+        let expires_at = snapshot
+            .effective_until()
+            .min(assertion.expires_at().unix_seconds())
+            .min(application_until)
+            .saturating_sub(policy.access_lease_policy.clock_skew().seconds());
+        if expires_at <= now.unix_seconds() {
+            return Err(RelayFinalizationError::EnrollmentEvidenceMismatch);
+        }
+        Ok(EnrollmentDisposition {
+            authorization_domain: tenant.community(),
+            actor_pubkey: proof.actor_pubkey(),
+            principal: assertion.principal().clone(),
+            profile_id: policy.profile_id.clone(),
+            policy_version: snapshot.policy_version().clone(),
+            correlation_id,
+            expires_at,
+        })
+    }
+}
+
+/// Direct provider decision sealed for atomic first enrollment.
+#[must_use]
+pub struct EnrollmentDisposition {
+    authorization_domain: CommunityId,
+    actor_pubkey: nostr::PublicKey,
+    principal: FederatedPrincipal,
+    profile_id: AuthorizationProfileId,
+    policy_version: PolicyVersion,
+    correlation_id: Uuid,
+    expires_at: u64,
+}
+
+impl EnrollmentDisposition {
+    /// Exact server-resolved authorization domain.
+    pub const fn authorization_domain(&self) -> CommunityId {
+        self.authorization_domain
+    }
+    /// Direct actor whose key is attested by the assertion.
+    pub const fn actor_pubkey(&self) -> nostr::PublicKey {
+        self.actor_pubkey
+    }
+    /// Literal issuer-qualified principal staged for enrollment.
+    pub const fn principal(&self) -> &FederatedPrincipal {
+        &self.principal
+    }
+    /// Server-selected authorization profile.
+    pub const fn profile_id(&self) -> &AuthorizationProfileId {
+        &self.profile_id
+    }
+    /// Provider policy version that authorized the enrollment.
+    pub const fn policy_version(&self) -> &PolicyVersion {
+        &self.policy_version
+    }
+    /// Exact decision correlation identifier.
+    pub const fn correlation_id(&self) -> Uuid {
+        self.correlation_id
+    }
+    /// Earliest authoritative expiry bound for the enrollment transaction.
+    pub const fn expires_at(&self) -> u64 {
+        self.expires_at
+    }
+}
+
+impl fmt::Debug for EnrollmentDisposition {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("EnrollmentDisposition")
+            .field("evidence", &"[redacted]")
+            .finish()
+    }
+}
+
+impl fmt::Debug for RelayAuthorizationFinalizer {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("RelayAuthorizationFinalizer")
+            .field("selector", &self.selector)
+            .field("finalizer", &self.finalizer)
+            .finish()
+    }
+}
+
+/// Typed result of server-mode finalization.
+#[must_use]
+pub enum RuntimeAuthorizationDisposition {
+    /// Enforcing authority carrying a bounded access lease.
+    Access(Box<buzz_auth::AuthContext>),
+    /// Display-only verification carrying no access authority.
+    VerificationOnly(VerificationOnlyDisposition),
+}
+
+impl fmt::Debug for RuntimeAuthorizationDisposition {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Access(_) => formatter
+                .debug_tuple("Access")
+                .field(&"[redacted]")
+                .finish(),
+            Self::VerificationOnly(_) => formatter
+                .debug_tuple("VerificationOnly")
+                .field(&"[redacted]")
+                .finish(),
+        }
+    }
+}
+
+/// Fail-closed relay finalization adapter error.
+#[derive(Debug, Error)]
+pub enum RelayFinalizationError {
+    /// Exact-domain server policy could not be resolved.
+    #[error(transparent)]
+    DomainPolicy(#[from] DomainPolicyError),
+    /// Central authorization time was unavailable.
+    #[error(transparent)]
+    Clock(#[from] buzz_auth::AuthorizationClockError),
+    /// Provider request evidence was inconsistent.
+    #[error(transparent)]
+    ProviderContract(#[from] ProviderContractError),
+    /// Full provider/binding finalization failed.
+    #[error(transparent)]
+    Finalization(#[from] FinalizationError),
+    /// A non-evaluating mode was asked to evaluate federated policy.
+    #[error("server-resolved authorization mode does not evaluate federated policy")]
+    ModeDoesNotEvaluate,
+    /// A non-finalizing mode was asked to create status or authority.
+    #[error("server-resolved authorization mode does not permit finalization")]
+    ModeDoesNotFinalize,
+    /// First enrollment was configured for a non-attested mode.
+    #[error("server-resolved authorization policy does not permit direct enrollment")]
+    EnrollmentModeUnsupported,
+    /// Direct assertion, provider, proof, or expiry evidence did not match.
+    #[error("direct enrollment evidence is inconsistent or stale")]
+    EnrollmentEvidenceMismatch,
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{future::ready, time::Duration};
+
+    use buzz_auth::{
+        ApplicationLeaseLimit, AuthorizationClockSkew, AuthorizationDenial,
+        AuthorizationDenialReason, AuthorizationProviderFuture, ProviderDecision,
+    };
+    use uuid::Uuid;
+
+    use super::*;
+
+    struct DenyProvider;
+
+    impl AuthorizationProvider for DenyProvider {
+        fn authorize<'a>(
+            &'a self,
+            _request: &'a AuthorizationRequest,
+        ) -> AuthorizationProviderFuture<'a> {
+            Box::pin(ready(ProviderDecision::Deny(AuthorizationDenial::new(
+                AuthorizationDenialReason::ProviderDenied,
+            ))))
+        }
+    }
+
+    fn domain(value: u128) -> CommunityId {
+        CommunityId::from_uuid(Uuid::from_u128(value))
+    }
+
+    fn policy(domain: CommunityId) -> DomainAuthorizationPolicy {
+        let application_limit =
+            ApplicationLeaseLimit::from_seconds(300).expect("synthetic limit is valid");
+        let skew = AuthorizationClockSkew::from_seconds(5).expect("synthetic skew is valid");
+        DomainAuthorizationPolicy::from_server_configuration(
+            domain,
+            "synthetic-provider.example",
+            Arc::new(DenyProvider),
+            EnrollmentMode::Provisioned,
+            AuthorizationMode::Enforce,
+            ProviderTimeout::new(Duration::from_secs(1)).expect("synthetic timeout is valid"),
+            AccessLeasePolicy::new(application_limit, skew),
+            VerificationStatusPolicy::new(application_limit, skew),
+        )
+        .expect("synthetic policy is valid")
+    }
+
+    #[test]
+    fn duplicate_domain_policy_is_rejected_as_ambiguous() {
+        let exact_domain = domain(1);
+        let result = DomainProviderSelector::new([policy(exact_domain), policy(exact_domain)]);
+        assert!(matches!(
+            result,
+            Err(DomainPolicyError::AmbiguousDomainPolicy)
+        ));
+    }
+
+    #[test]
+    fn missing_domain_has_no_default_provider_fallback() {
+        let configured = domain(1);
+        let missing = domain(2);
+        let selector = DomainProviderSelector::new([policy(configured)])
+            .expect("one exact policy is unambiguous");
+        let tenant = TenantContext::resolved(missing, "missing.authorization.example");
+        assert!(matches!(
+            selector.resolve(&tenant),
+            Err(DomainPolicyError::MissingDomainPolicy)
+        ));
+    }
+
+    #[test]
+    fn exact_server_tenant_selects_its_policy() {
+        let configured = domain(1);
+        let selector = DomainProviderSelector::new([policy(configured)])
+            .expect("one exact policy is unambiguous");
+        let tenant = TenantContext::resolved(configured, "configured.authorization.example");
+        let resolved = selector
+            .resolve(&tenant)
+            .expect("exact domain is configured");
+        assert_eq!(resolved.authorization_domain(), configured);
+        assert_eq!(resolved.mode(), AuthorizationMode::Enforce);
+    }
+}

crates/buzz-relay/src/authorization_runtime/production.rs

@@ -732,14 +732,13 @@ pub async fn build_from_environment_with_providers(
         return Ok(None);
     }
     validate_provider_coverage(&configured, &providers)?;
-    if configured
-        .values()
-        .any(|mode| *mode != AuthorizationMode::Off)
+    if configured.values().any(|mode| mode.evaluates_provider())
         && state.identity_assertion_provenance().is_none()
     {
         return Err(ProductionRuntimeError::AssertionProvenanceMissing);
     }
-    let enforcing_domains = authoritative_domains(&configured);
+    let protected_domains = protected_domains(&configured);
+    let enforcing_domains = enforcing_domains(&configured);
     let projection_domains = projection_reconciliation_domains(&configured);
     if !enforcing_domains.is_empty() && state.corporate_identity.is_none() {
         return Err(ProductionRuntimeError::VerifierMissing);
@@ -747,7 +746,7 @@ pub async fn build_from_environment_with_providers(
     let clock: SharedAuthorizationClock = Arc::new(SystemAuthorizationClock);
     let restore_bootstraps = parse_restore_bootstraps(
         &env::var(RESTORE_BOOTSTRAPS_ENV).unwrap_or_default(),
-        &enforcing_domains,
+        &protected_domains,
     )?;
     let profile = env::var(PROFILE_ENV).unwrap_or_else(|_| "current-membership-v1".to_owned());
     let lease_seconds = parse_positive_seconds(LEASE_SECONDS_ENV, 300)?;
@@ -757,11 +756,13 @@ pub async fn build_from_environment_with_providers(
     let mut policies = Vec::with_capacity(configured.len());
     let mut transports = Vec::with_capacity(configured.len());
     for (domain, mode) in &configured {
-        let provider = if *mode == AuthorizationMode::Off {
+        transports.push(DomainTransportPolicy::from_server_configuration(
+            *domain, *mode,
+        ));
+        if !mode.evaluates_provider() {
             continue;
-        } else {
-            providers.provider_for(*domain)?
-        };
+        }
+        let provider = providers.provider_for(*domain)?;
         policies.push(DomainAuthorizationPolicy::from_server_configuration(
             *domain,
             profile.clone(),
@@ -772,16 +773,6 @@ pub async fn build_from_environment_with_providers(
             AccessLeasePolicy::new(lease_limit, skew),
             VerificationStatusPolicy::new(status_limit, skew),
         )?);
-        transports.push(DomainTransportPolicy::from_server_configuration(
-            *domain, *mode,
-        ));
-    }
-    for (domain, mode) in &configured {
-        if *mode == AuthorizationMode::Off {
-            transports.push(DomainTransportPolicy::from_server_configuration(
-                *domain, *mode,
-            ));
-        }
     }
     let hosts = state.db.usage_community_hosts().await?;
     let host_map = hosts
@@ -804,7 +795,7 @@ pub async fn build_from_environment_with_providers(
         restore_bootstraps,
     )
     .await?;
-    activate_enforcing_domains(&state.db, &restore, enforcing_domains.iter().copied()).await?;
+    activate_protected_domains(&state.db, &restore, protected_domains.iter().copied()).await?;
     reconcile_audio_admissions_once(&state.db, &restore, enforcing_domains.iter().copied()).await?;
     let invalidation = AuthorizationInvalidationRuntime::new_with_restore(
         state.db.clone(),
@@ -813,7 +804,7 @@ pub async fn build_from_environment_with_providers(
         Arc::clone(&restore),
     );
     invalidation
-        .initialize_domains(enforcing_domains.iter().copied())
+        .initialize_domains(protected_domains.iter().copied())
         .await?;
     crate::corporate_identity::reconcile_public_projection_retirements_startup(
         state,
@@ -843,7 +834,7 @@ pub async fn build_from_environment_with_providers(
     }))
 }
 
-async fn activate_enforcing_domains(
+async fn activate_protected_domains(
     db: &buzz_db::Db,
     restore: &Arc<super::restore::RestoreProtectionRuntime>,
     domains: impl IntoIterator<Item = CommunityId>,
@@ -886,7 +877,7 @@ fn validate_provider_coverage(
     providers: &ProductionProviderRegistry,
 ) -> Result<(), ProductionRuntimeError> {
     for (domain, mode) in configured {
-        if *mode != AuthorizationMode::Off {
+        if mode.evaluates_provider() {
             providers.provider_for(*domain)?;
         }
     }
@@ -897,10 +888,11 @@ fn validate_activated_domain_configuration(
     configured: &HashMap<CommunityId, AuthorizationMode>,
     activated: impl IntoIterator<Item = CommunityId>,
 ) -> Result<(), ProductionRuntimeError> {
-    if activated
-        .into_iter()
-        .any(|domain| configured.get(&domain) != Some(&AuthorizationMode::Enforce))
-    {
+    if activated.into_iter().any(|domain| {
+        !configured
+            .get(&domain)
+            .is_some_and(|mode| mode.protects_surfaces())
+    }) {
         return Err(ProductionRuntimeError::ActivatedDomainDowngrade);
     }
     Ok(())
@@ -1083,6 +1075,7 @@ fn parse_domains(
             "shadow" => AuthorizationMode::Shadow,
             "verify_only" => AuthorizationMode::VerifyOnly,
             "enforce" => AuthorizationMode::Enforce,
+            "deny_protected" => AuthorizationMode::DenyProtected,
             _ => return Err(ProductionRuntimeError::InvalidConfiguration),
         };
         if domains.insert(domain, mode).is_some() {
@@ -1092,7 +1085,14 @@ fn parse_domains(
     Ok(domains)
 }
 
-fn authoritative_domains(configured: &HashMap<CommunityId, AuthorizationMode>) -> Vec<CommunityId> {
+fn protected_domains(configured: &HashMap<CommunityId, AuthorizationMode>) -> Vec<CommunityId> {
+    configured
+        .iter()
+        .filter_map(|(domain, mode)| mode.protects_surfaces().then_some(*domain))
+        .collect()
+}
+
+fn enforcing_domains(configured: &HashMap<CommunityId, AuthorizationMode>) -> Vec<CommunityId> {
     configured
         .iter()
         .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain))
@@ -1112,7 +1112,7 @@ fn projection_reconciliation_domains(
 
 fn parse_restore_bootstraps(
     raw: &str,
-    enforcing_domains: &[CommunityId],
+    protected_domains: &[CommunityId],
 ) -> Result<Vec<(CommunityId, uuid::Uuid)>, ProductionRuntimeError> {
     let mut anchors = HashMap::new();
     for item in raw
@@ -1133,8 +1133,8 @@ fn parse_restore_bootstraps(
             return Err(ProductionRuntimeError::InvalidConfiguration);
         }
     }
-    let mut result = Vec::with_capacity(enforcing_domains.len());
-    for domain in enforcing_domains {
+    let mut result = Vec::with_capacity(protected_domains.len());
+    for domain in protected_domains {
         let bootstrap = anchors
             .remove(domain)
             .ok_or(ProductionRuntimeError::RestoreBootstrapMissing)?;
@@ -1240,8 +1240,11 @@ mod tests {
     fn exact_modes_parse_without_a_default() {
         let first = uuid::Uuid::new_v4();
         let second = uuid::Uuid::new_v4();
-        let parsed = parse_domains(&format!("{first}:enforce,{second}:verify_only"))
-            .expect("valid exact domains");
+        let third = uuid::Uuid::new_v4();
+        let parsed = parse_domains(&format!(
+            "{first}:enforce,{second}:verify_only,{third}:deny_protected"
+        ))
+        .expect("valid exact domains");
         assert_eq!(
             parsed.get(&CommunityId::from_uuid(first)),
             Some(&AuthorizationMode::Enforce)
@@ -1250,6 +1253,10 @@ mod tests {
             parsed.get(&CommunityId::from_uuid(second)),
             Some(&AuthorizationMode::VerifyOnly)
         );
+        assert_eq!(
+            parsed.get(&CommunityId::from_uuid(third)),
+            Some(&AuthorizationMode::DenyProtected)
+        );
     }
 
     #[test]
@@ -1265,12 +1272,19 @@ mod tests {
         let shadow = uuid::Uuid::new_v4();
         let verify = uuid::Uuid::new_v4();
         let enforce = uuid::Uuid::new_v4();
+        let deny = uuid::Uuid::new_v4();
         let parsed = parse_domains(&format!(
-            "{off}:off,{shadow}:shadow,{verify}:verify_only,{enforce}:enforce"
+            "{off}:off,{shadow}:shadow,{verify}:verify_only,{enforce}:enforce,{deny}:deny_protected"
         ))
         .expect("valid exact modes");
+        let protected = protected_domains(&parsed)
+            .into_iter()
+            .collect::<std::collections::HashSet<_>>();
+        assert_eq!(protected.len(), 2);
+        assert!(protected.contains(&CommunityId::from_uuid(enforce)));
+        assert!(protected.contains(&CommunityId::from_uuid(deny)));
         assert_eq!(
-            authoritative_domains(&parsed),
+            enforcing_domains(&parsed),
             vec![CommunityId::from_uuid(enforce)]
         );
         let projection_domains = projection_reconciliation_domains(&parsed)
@@ -1281,6 +1295,7 @@ mod tests {
         assert!(!projection_domains.contains(&CommunityId::from_uuid(shadow)));
         assert!(!projection_domains.contains(&CommunityId::from_uuid(verify)));
         assert!(projection_domains.contains(&CommunityId::from_uuid(enforce)));
+        assert!(!projection_domains.contains(&CommunityId::from_uuid(deny)));
     }
 
     #[test]
@@ -1304,6 +1319,9 @@ mod tests {
         let configured = HashMap::from([(activated, AuthorizationMode::Enforce)]);
         validate_activated_domain_configuration(&configured, [activated])
             .expect("exact Enforce configuration preserves one-way activation");
+        let configured = HashMap::from([(activated, AuthorizationMode::DenyProtected)]);
+        validate_activated_domain_configuration(&configured, [activated])
+            .expect("deny-protected preserves the protected inventory after activation");
     }
 
     #[test]
@@ -1324,8 +1342,11 @@ mod tests {
     fn exact_provider_coverage_makes_enforce_constructible_without_fallback() {
         let enforce = CommunityId::from_uuid(uuid::Uuid::new_v4());
         let off = CommunityId::from_uuid(uuid::Uuid::new_v4());
-        let configured = parse_domains(&format!("{enforce}:enforce,{off}:off"))
-            .expect("exact production configuration");
+        let deny = CommunityId::from_uuid(uuid::Uuid::new_v4());
+        let configured = parse_domains(&format!(
+            "{enforce}:enforce,{off}:off,{deny}:deny_protected"
+        ))
+        .expect("exact production configuration");
         assert!(matches!(
             validate_provider_coverage(&configured, &ProductionProviderRegistry::default()),
             Err(ProductionRuntimeError::ProviderMissing)
@@ -1356,10 +1377,10 @@ mod tests {
     fn enforce_domain_activation_precedes_snapshot_and_transport_reachability() {
         let source = include_str!("production.rs");
         let activation = source
-            .find("activate_enforcing_domains(&state.db")
+            .find("activate_protected_domains(&state.db")
             .expect("durable activation is part of construction");
         let snapshot = source
-            .find(".initialize_domains(enforcing_domains.iter().copied())")
+            .find(".initialize_domains(protected_domains.iter().copied())")
             .expect("invalidation snapshot is initialized");
         let transport = source
             .find("ProtectedTransportRuntime::new(transports, resolver, clock)")

crates/buzz-relay/src/authorization_runtime/transport.rs

@@ -33,8 +33,8 @@ pub enum LegacyIdentityLane {
     /// immutable. This lane can never enroll, reactivate, strengthen, update,
     /// or retire a binding.
     ObserveOnly,
-    /// Enforce: the protected resolver owns admission/finalization and legacy
-    /// projection must not run.
+    /// Enforce or DenyProtected: the protected runtime owns the surface and
+    /// legacy projection must not run.
     ProtectedEnforce,
 }
 
@@ -57,7 +57,9 @@ pub const fn legacy_identity_lane_for_mode(mode: Option<AuthorizationMode>) -> L
         Some(AuthorizationMode::Shadow) | Some(AuthorizationMode::VerifyOnly) => {
             LegacyIdentityLane::ObserveOnly
         }
-        Some(AuthorizationMode::Enforce) => LegacyIdentityLane::ProtectedEnforce,
+        Some(AuthorizationMode::Enforce | AuthorizationMode::DenyProtected) => {
+            LegacyIdentityLane::ProtectedEnforce
+        }
     }
 }
 
@@ -532,7 +534,7 @@ impl ProtectedTransportRuntime {
     pub fn enforcing_domains(&self) -> Vec<CommunityId> {
         self.domains
             .iter()
-            .filter_map(|(domain, mode)| (*mode == AuthorizationMode::Enforce).then_some(*domain))
+            .filter_map(|(domain, mode)| mode.protects_surfaces().then_some(*domain))
             .collect()
     }
 
@@ -555,6 +557,7 @@ impl ProtectedTransportRuntime {
                 let _ = self.resolver.observe(request).await;
                 Ok(ProtectedAuthorization::Legacy)
             }
+            AuthorizationMode::DenyProtected => deny_protected_request(request),
             AuthorizationMode::Enforce => {
                 let resolution = self
                     .resolver
@@ -597,6 +600,7 @@ impl ProtectedTransportRuntime {
                 .await
                 .map(Some)
                 .map_err(ProtectedTransportError::Resolution),
+            Some(AuthorizationMode::DenyProtected) => Err(ProtectedTransportError::DenyProtected),
             None | Some(AuthorizationMode::Off | AuthorizationMode::Shadow) => Ok(None),
         }
     }
@@ -613,6 +617,7 @@ impl ProtectedTransportRuntime {
             AuthorizationMode::Off | AuthorizationMode::Shadow | AuthorizationMode::VerifyOnly => {
                 Ok(ProtectedEnrollmentAuthorization::Legacy)
             }
+            AuthorizationMode::DenyProtected => Err(ProtectedTransportError::DenyProtected),
             AuthorizationMode::Enforce => {
                 if request.capability() != AuthorizationCapability::InviteClaim
                     || request.enrollment_assertion().is_none()
@@ -761,9 +766,19 @@ fn authorize_unwired_for_mode(
         | Some(AuthorizationMode::Shadow)
         | Some(AuthorizationMode::VerifyOnly) => Ok(ProtectedAuthorization::Legacy),
         Some(AuthorizationMode::Enforce) => Err(ProtectedTransportError::MissingVerifiedProof),
+        Some(AuthorizationMode::DenyProtected) => Err(ProtectedTransportError::DenyProtected),
     }
 }
 
+fn deny_protected_request(
+    request: &ProtectedOperationRequest,
+) -> Result<ProtectedAuthorization, ProtectedTransportError> {
+    if let Some(cancellation) = request.cancellation() {
+        cancellation.cancel();
+    }
+    Err(ProtectedTransportError::DenyProtected)
+}
+
 impl fmt::Debug for ProtectedTransportRuntime {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         formatter
@@ -1093,6 +1108,9 @@ pub enum ProtectedTransportError {
     /// An enforcing surface did not retain sealed verifier evidence.
     #[error("protected authorization requires verified transport evidence")]
     MissingVerifiedProof,
+    /// The exact domain is in the explicit fail-safe protected-denial mode.
+    #[error("protected authorization is unavailable in deny-protected mode")]
+    DenyProtected,
     /// Resolver denied or could not evaluate current policy.
     #[error(transparent)]
     Resolution(#[from] ProtectedResolutionError),

crates/buzz-relay/src/connection.rs

@@ -6,6 +6,7 @@ use std::sync::atomic::{AtomicU8, Ordering};
 use std::sync::Arc;
 use std::time::Duration;
 
+use async_trait::async_trait;
 use axum::extract::ws::{Message as WsMessage, WebSocket};
 use futures_util::{Sink, SinkExt, StreamExt};
 use tokio::sync::{mpsc, Mutex, RwLock};
@@ -23,6 +24,229 @@ use crate::protocol::{ClientMessage, RelayMessage};
 use crate::state::{run_registered_community_connection, AppState};
 use buzz_pubsub::EventTopic;
 
+/// Fail-closed release check evaluated at the socket acceptance boundary.
+pub(crate) trait OutboundReleaseFence: Send + Sync {
+    fn release(&self) -> bool;
+}
+
+/// Potentially asynchronous release fence retained until socket drain.
+#[async_trait]
+pub(crate) trait QueuedOutboundReleaseFence: Send + Sync {
+    async fn release(&self) -> bool;
+}
+
+struct SyncQueuedReleaseFence {
+    authority: Arc<dyn OutboundReleaseFence>,
+}
+
+#[async_trait]
+impl QueuedOutboundReleaseFence for SyncQueuedReleaseFence {
+    async fn release(&self) -> bool {
+        self.authority.release()
+    }
+}
+
+pub(crate) fn queued_local_authority(
+    authority: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+) -> Arc<dyn QueuedOutboundReleaseFence> {
+    Arc::new(SyncQueuedReleaseFence { authority })
+}
+
+#[async_trait]
+trait ChannelReadAuthoritySource: Send + Sync {
+    async fn channel_set_read_authorized(
+        &self,
+        community_id: buzz_core::tenant::CommunityId,
+        channel_ids: &[Uuid],
+        actor: &[u8],
+    ) -> bool;
+}
+
+#[async_trait]
+impl ChannelReadAuthoritySource for buzz_db::Db {
+    async fn channel_set_read_authorized(
+        &self,
+        community_id: buzz_core::tenant::CommunityId,
+        channel_ids: &[Uuid],
+        actor: &[u8],
+    ) -> bool {
+        buzz_db::Db::channel_set_read_authorized(self, community_id, channel_ids, actor)
+            .await
+            .unwrap_or(false)
+    }
+}
+
+struct ChannelReadReleaseFence {
+    source: Arc<dyn ChannelReadAuthoritySource>,
+    community_id: buzz_core::tenant::CommunityId,
+    channel_ids: Vec<Uuid>,
+    actor: Vec<u8>,
+    protected: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+}
+
+#[async_trait]
+impl QueuedOutboundReleaseFence for ChannelReadReleaseFence {
+    async fn release(&self) -> bool {
+        if self
+            .protected
+            .as_ref()
+            .is_some_and(|authority| authority.revalidate().is_err())
+        {
+            return false;
+        }
+        if !self
+            .source
+            .channel_set_read_authorized(self.community_id, &self.channel_ids, &self.actor)
+            .await
+        {
+            return false;
+        }
+        self.protected
+            .as_ref()
+            .is_none_or(|authority| authority.revalidate().is_ok())
+    }
+}
+
+/// Retain uncached channel access, plus optional protected identity authority,
+/// until the socket writer accepts the queued frame.
+pub(crate) fn queued_channel_read_authority(
+    db: buzz_db::Db,
+    community_id: buzz_core::tenant::CommunityId,
+    channel_id: Uuid,
+    actor: Vec<u8>,
+    protected: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+) -> Arc<dyn QueuedOutboundReleaseFence> {
+    queued_channel_set_read_authority(db, community_id, vec![channel_id], actor, protected)
+}
+
+/// Retain uncached access to every channel that can contribute to one
+/// aggregate response until the response is released.
+pub(crate) fn queued_channel_set_read_authority(
+    db: buzz_db::Db,
+    community_id: buzz_core::tenant::CommunityId,
+    mut channel_ids: Vec<Uuid>,
+    actor: Vec<u8>,
+    protected: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+) -> Arc<dyn QueuedOutboundReleaseFence> {
+    channel_ids.sort_unstable();
+    channel_ids.dedup();
+    Arc::new(ChannelReadReleaseFence {
+        source: Arc::new(db),
+        community_id,
+        channel_ids,
+        actor,
+        protected,
+    })
+}
+
+/// Evaluate the aggregate read fence synchronously with an HTTP response
+/// release. WebSocket callers retain the same fence in their outbound queue.
+pub(crate) async fn release_channel_set_read_authority(
+    db: buzz_db::Db,
+    community_id: buzz_core::tenant::CommunityId,
+    channel_ids: Vec<Uuid>,
+    actor: Vec<u8>,
+    protected: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+) -> bool {
+    queued_channel_set_read_authority(db, community_id, channel_ids, actor, protected)
+        .release()
+        .await
+}
+
+impl OutboundReleaseFence for crate::authorization_runtime::transport::ProtectedAuthorization {
+    fn release(&self) -> bool {
+        self.release_fetched(()).is_ok()
+    }
+}
+
+struct CombinedReleaseFence {
+    sender: Arc<dyn QueuedOutboundReleaseFence>,
+    recipient: Arc<dyn QueuedOutboundReleaseFence>,
+}
+
+#[async_trait]
+impl QueuedOutboundReleaseFence for CombinedReleaseFence {
+    async fn release(&self) -> bool {
+        self.sender.release().await
+            && self.recipient.release().await
+            && self.sender.release().await
+            && self.recipient.release().await
+    }
+}
+
+/// One queued data frame with optional authority retained until socket drain.
+pub struct OutboundData {
+    pub(crate) message: WsMessage,
+    authority: Option<Arc<dyn QueuedOutboundReleaseFence>>,
+}
+
+impl OutboundData {
+    pub(crate) fn plain(message: WsMessage) -> Self {
+        Self {
+            message,
+            authority: None,
+        }
+    }
+
+    pub(crate) fn protected(
+        message: WsMessage,
+        authority: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+    ) -> Self {
+        Self {
+            message,
+            authority: Some(queued_local_authority(authority)),
+        }
+    }
+
+    pub(crate) fn protected_pair(
+        message: WsMessage,
+        sender: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+        recipient: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+    ) -> Self {
+        Self {
+            message,
+            authority: Some(Arc::new(CombinedReleaseFence {
+                sender: queued_local_authority(sender),
+                recipient: queued_local_authority(recipient),
+            })),
+        }
+    }
+
+    pub(crate) fn guarded(
+        message: WsMessage,
+        authority: Arc<dyn QueuedOutboundReleaseFence>,
+    ) -> Self {
+        Self {
+            message,
+            authority: Some(authority),
+        }
+    }
+
+    pub(crate) fn guarded_pair(
+        message: WsMessage,
+        sender: Arc<dyn QueuedOutboundReleaseFence>,
+        recipient: Arc<dyn QueuedOutboundReleaseFence>,
+    ) -> Self {
+        Self {
+            message,
+            authority: Some(Arc::new(CombinedReleaseFence { sender, recipient })),
+        }
+    }
+
+    #[cfg(test)]
+    fn guarded_for_test(message: WsMessage, authority: Arc<dyn OutboundReleaseFence>) -> Self {
+        Self::guarded(message, Arc::new(SyncQueuedReleaseFence { authority }))
+    }
+
+    async fn release(self) -> Option<WsMessage> {
+        match self.authority {
+            Some(authority) if authority.release().await => Some(self.message),
+            Some(_) => None,
+            None => Some(self.message),
+        }
+    }
+}
+
 /// Maximum time a new socket may hold a connection slot without completing NIP-42 auth.
 const AUTH_TIMEOUT: Duration = Duration::from_secs(5);
 
@@ -59,14 +283,14 @@ pub struct ConnectionState {
     pub tenant: TenantContext,
     /// Remote socket address of the client.
     pub remote_addr: SocketAddr,
-    /// Optional corporate identity JWT captured from the WebSocket upgrade request.
-    pub corporate_identity_jwt: Option<String>,
+    /// Optional direct identity assertion captured with verified provenance.
+    pub corporate_identity_assertion: Option<crate::corporate_identity::IdentityAssertionInput>,
     /// Current NIP-42 authentication state.
     pub auth_state: RwLock<AuthState>,
     /// Active subscriptions keyed by subscription ID.
     pub subscriptions: ConnectionSubscriptions,
     /// Sender for outbound data messages (EVENT, NOTICE, OK, etc.).
-    pub send_tx: mpsc::Sender<WsMessage>,
+    pub send_tx: mpsc::Sender<OutboundData>,
     /// Sender for outbound control frames (Pong, Close).
     /// Separate channel with priority drain — if this channel fills too,
     /// the connection is closed (writer is completely stalled).
@@ -88,7 +312,43 @@ impl ConnectionState {
     /// `grace_limit` occurrences log a warning; sustained backpressure
     /// cancels the connection to prevent unbounded memory growth.
     pub fn send(&self, msg: String) -> bool {
-        match self.send_tx.try_send(WsMessage::Text(msg.into())) {
+        self.send_data(OutboundData::plain(WsMessage::Text(msg.into())))
+    }
+
+    /// Queue a terminal text frame on the priority control channel.
+    ///
+    /// Callers may cancel immediately after this returns: the send loop drains
+    /// control frames before emitting the WebSocket close frame.
+    pub(crate) fn send_terminal(&self, msg: String) -> bool {
+        self.ctrl_tx.try_send(WsMessage::Text(msg.into())).is_ok()
+    }
+
+    /// Queue protected output while retaining its guard through socket drain.
+    pub fn send_protected(
+        &self,
+        msg: String,
+        authority: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+    ) -> bool {
+        self.send_data(OutboundData::protected(
+            WsMessage::Text(msg.into()),
+            authority,
+        ))
+    }
+
+    /// Queue output behind an arbitrary asynchronous release fence.
+    pub(crate) fn send_guarded(
+        &self,
+        msg: String,
+        authority: Arc<dyn QueuedOutboundReleaseFence>,
+    ) -> bool {
+        self.send_data(OutboundData::guarded(
+            WsMessage::Text(msg.into()),
+            authority,
+        ))
+    }
+
+    fn send_data(&self, msg: OutboundData) -> bool {
+        match self.send_tx.try_send(msg) {
             Ok(_) => {
                 // Successful send resets the grace counter.
                 self.backpressure_count.store(0, Ordering::Relaxed);
@@ -122,7 +382,7 @@ pub async fn handle_connection(
     state: Arc<AppState>,
     addr: SocketAddr,
     tenant: TenantContext,
-    corporate_identity_jwt: Option<String>,
+    corporate_identity_assertion: Option<crate::corporate_identity::IdentityAssertionInput>,
 ) {
     let conn_id = Uuid::new_v4();
     let cancel = CancellationToken::new();
@@ -144,7 +404,7 @@ pub async fn handle_connection(
                 tenant,
                 conn_id,
                 cancel,
-                corporate_identity_jwt,
+                corporate_identity_assertion,
             )
         },
     )
@@ -158,7 +418,7 @@ async fn handle_active_connection(
     tenant: TenantContext,
     conn_id: Uuid,
     cancel: CancellationToken,
-    corporate_identity_jwt: Option<String>,
+    corporate_identity_assertion: Option<crate::corporate_identity::IdentityAssertionInput>,
 ) {
     let permit = match state.conn_semaphore.clone().try_acquire_owned() {
         Ok(p) => p,
@@ -170,7 +430,7 @@ async fn handle_active_connection(
 
     let challenge = generate_challenge();
 
-    let (tx, rx) = mpsc::channel::<WsMessage>(state.config.send_buffer_size);
+    let (tx, rx) = mpsc::channel::<OutboundData>(state.config.send_buffer_size);
     // Control channel for Pong/Close — small capacity, guaranteed delivery
     // even when the data buffer is full.
     let (ctrl_tx, ctrl_rx) = mpsc::channel::<WsMessage>(8);
@@ -182,7 +442,7 @@ async fn handle_active_connection(
         conn_id,
         tenant,
         remote_addr: addr,
-        corporate_identity_jwt,
+        corporate_identity_assertion,
         auth_state: RwLock::new(AuthState::Pending {
             challenge: challenge.clone(),
         }),
@@ -197,13 +457,13 @@ async fn handle_active_connection(
     info!(conn_id = %conn_id, addr = %addr, "WebSocket connection established");
     metrics::counter!(
         "buzz_ws_connections_total",
-        "community" => conn.tenant.host().to_owned()
+        "community" => crate::metrics::community_label(conn.tenant.community())
     )
     .increment(1);
 
     let challenge_msg = RelayMessage::auth_challenge(&challenge);
     if tx
-        .send(WsMessage::Text(challenge_msg.into()))
+        .send(OutboundData::plain(WsMessage::Text(challenge_msg.into())))
         .await
         .is_err()
     {
@@ -310,7 +570,7 @@ async fn handle_active_connection(
 /// treat a full control channel as terminal (Bug 7 fix).
 async fn send_loop(
     ws_send: futures_util::stream::SplitSink<WebSocket, WsMessage>,
-    data_rx: mpsc::Receiver<WsMessage>,
+    data_rx: mpsc::Receiver<OutboundData>,
     ctrl_rx: mpsc::Receiver<WsMessage>,
     cancel: CancellationToken,
 ) {
@@ -319,7 +579,7 @@ async fn send_loop(
 
 async fn send_loop_inner<S>(
     mut ws_send: S,
-    mut data_rx: mpsc::Receiver<WsMessage>,
+    mut data_rx: mpsc::Receiver<OutboundData>,
     mut ctrl_rx: mpsc::Receiver<WsMessage>,
     cancel: CancellationToken,
 ) where
@@ -357,16 +617,30 @@ async fn send_loop_inner<S>(
                     break;
                 }
             }
-            Some(msg) = data_rx.recv() => {
+            Some(queued) = data_rx.recv() => {
                 let mut batched = 1usize;
-                if ws_send.feed(msg).await.is_err() {
+                if !sink_ready_before_cancellation(&mut ws_send, &cancel).await {
+                    break;
+                }
+                let Some(msg) = queued.release().await else {
+                    cancel.cancel();
+                    break;
+                };
+                if std::pin::Pin::new(&mut ws_send).start_send(msg).is_err() {
                     break;
                 }
 
                 while batched < MAX_WS_SEND_BATCH {
                     match data_rx.try_recv() {
                         Ok(next) => {
-                            if ws_send.feed(next).await.is_err() {
+                            if !sink_ready_before_cancellation(&mut ws_send, &cancel).await {
+                                return;
+                            }
+                            let Some(next) = next.release().await else {
+                                cancel.cancel();
+                                return;
+                            };
+                            if std::pin::Pin::new(&mut ws_send).start_send(next).is_err() {
                                 return;
                             }
                             batched += 1;
@@ -385,6 +659,19 @@ async fn send_loop_inner<S>(
     }
 }
 
+async fn sink_ready_before_cancellation<S>(ws_send: &mut S, cancel: &CancellationToken) -> bool
+where
+    S: Sink<WsMessage> + Unpin,
+{
+    tokio::select! {
+        biased;
+        _ = cancel.cancelled() => false,
+        result = std::future::poll_fn(|cx| std::pin::Pin::new(&mut *ws_send).poll_ready(cx)) => {
+            result.is_ok()
+        }
+    }
+}
+
 /// 3 missed pongs → disconnect.
 ///
 /// Sends Ping through the control channel so it isn't blocked by a full
@@ -703,6 +990,7 @@ fn topic_for_subscription(channel_id: Option<Uuid>) -> EventTopic {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::sync::atomic::{AtomicBool, AtomicUsize};
     use std::sync::{Arc, Mutex};
 
     #[derive(Debug, Default)]
@@ -777,6 +1065,106 @@ mod tests {
         }
     }
 
+    struct ScriptedFence(AtomicBool);
+
+    impl OutboundReleaseFence for ScriptedFence {
+        fn release(&self) -> bool {
+            self.0.load(Ordering::SeqCst)
+        }
+    }
+
+    struct CountingQueuedFence(AtomicUsize);
+
+    #[async_trait]
+    impl QueuedOutboundReleaseFence for CountingQueuedFence {
+        async fn release(&self) -> bool {
+            self.0.fetch_add(1, Ordering::SeqCst);
+            true
+        }
+    }
+
+    #[tokio::test]
+    async fn combined_release_rechecks_both_sides_after_async_boundaries() {
+        let sender = Arc::new(CountingQueuedFence(AtomicUsize::new(0)));
+        let recipient = Arc::new(CountingQueuedFence(AtomicUsize::new(0)));
+        let fence = CombinedReleaseFence {
+            sender: sender.clone(),
+            recipient: recipient.clone(),
+        };
+
+        assert!(fence.release().await);
+        assert_eq!(sender.0.load(Ordering::SeqCst), 2);
+        assert_eq!(recipient.0.load(Ordering::SeqCst), 2);
+    }
+
+    struct ScriptedChannelAuthority {
+        allowed: AtomicBool,
+        checked: Mutex<Vec<Uuid>>,
+    }
+
+    #[async_trait]
+    impl ChannelReadAuthoritySource for ScriptedChannelAuthority {
+        async fn channel_set_read_authorized(
+            &self,
+            _community_id: buzz_core::tenant::CommunityId,
+            channel_ids: &[Uuid],
+            _actor: &[u8],
+        ) -> bool {
+            self.checked
+                .lock()
+                .expect("scripted channel checks poisoned")
+                .extend_from_slice(channel_ids);
+            self.allowed.load(Ordering::SeqCst)
+        }
+    }
+
+    struct ReadinessBarrierSink {
+        ready: Arc<AtomicBool>,
+        polled: Arc<tokio::sync::Notify>,
+        waker: Arc<Mutex<Option<std::task::Waker>>>,
+        state: Arc<Mutex<MockSinkState>>,
+    }
+
+    impl Sink<WsMessage> for ReadinessBarrierSink {
+        type Error = std::io::Error;
+
+        fn poll_ready(
+            self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            if self.ready.load(Ordering::SeqCst) {
+                std::task::Poll::Ready(Ok(()))
+            } else {
+                *self.waker.lock().expect("barrier waker poisoned") = Some(cx.waker().clone());
+                self.polled.notify_one();
+                std::task::Poll::Pending
+            }
+        }
+
+        fn start_send(self: std::pin::Pin<&mut Self>, item: WsMessage) -> Result<(), Self::Error> {
+            self.state
+                .lock()
+                .expect("barrier sink poisoned")
+                .messages
+                .push(item);
+            Ok(())
+        }
+
+        fn poll_flush(
+            self: std::pin::Pin<&mut Self>,
+            _cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            std::task::Poll::Ready(Ok(()))
+        }
+
+        fn poll_close(
+            self: std::pin::Pin<&mut Self>,
+            cx: &mut std::task::Context<'_>,
+        ) -> std::task::Poll<Result<(), Self::Error>> {
+            self.poll_flush(cx)
+        }
+    }
+
     fn text_payloads(messages: &[WsMessage]) -> Vec<String> {
         messages
             .iter()
@@ -806,7 +1194,9 @@ mod tests {
         let (_ctrl_tx, ctrl_rx) = mpsc::channel(1);
         for i in 0..5 {
             data_tx
-                .send(WsMessage::Text(format!("data-{i}").into()))
+                .send(OutboundData::plain(WsMessage::Text(
+                    format!("data-{i}").into(),
+                )))
                 .await
                 .expect("queue data frame");
         }
@@ -822,12 +1212,208 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn protected_frame_revalidates_after_sink_readiness() {
+        let ready = Arc::new(AtomicBool::new(false));
+        let polled = Arc::new(tokio::sync::Notify::new());
+        let waker = Arc::new(Mutex::new(None));
+        let state = Arc::new(Mutex::new(MockSinkState::default()));
+        let sink = ReadinessBarrierSink {
+            ready: Arc::clone(&ready),
+            polled: Arc::clone(&polled),
+            waker: Arc::clone(&waker),
+            state: Arc::clone(&state),
+        };
+        let fence = Arc::new(ScriptedFence(AtomicBool::new(true)));
+        let (data_tx, data_rx) = mpsc::channel(1);
+        let (_ctrl_tx, ctrl_rx) = mpsc::channel(1);
+        let cancel = CancellationToken::new();
+        data_tx
+            .send(OutboundData::guarded_for_test(
+                WsMessage::Text("protected".into()),
+                fence.clone(),
+            ))
+            .await
+            .expect("queue protected frame");
+        drop(data_tx);
+
+        let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone()));
+        polled.notified().await;
+        fence.0.store(false, Ordering::SeqCst);
+        ready.store(true, Ordering::SeqCst);
+        waker
+            .lock()
+            .expect("barrier waker poisoned")
+            .take()
+            .expect("poll_ready registered a waker")
+            .wake();
+        task.await.expect("send loop joins");
+
+        assert!(cancel.is_cancelled());
+        assert!(
+            state
+                .lock()
+                .expect("barrier sink poisoned")
+                .messages
+                .is_empty(),
+            "authority loss while readiness is pending must prevent start_send"
+        );
+    }
+
+    #[tokio::test]
+    async fn count_release_rechecks_every_contributing_channel() {
+        let first = Uuid::new_v4();
+        let second = Uuid::new_v4();
+        let source = Arc::new(ScriptedChannelAuthority {
+            allowed: AtomicBool::new(true),
+            checked: Mutex::new(Vec::new()),
+        });
+        let fence = ChannelReadReleaseFence {
+            source: source.clone(),
+            community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()),
+            channel_ids: vec![first, second],
+            actor: vec![7; 32],
+            protected: None,
+        };
+
+        assert!(fence.release().await);
+        assert_eq!(
+            *source
+                .checked
+                .lock()
+                .expect("scripted channel checks poisoned"),
+            vec![first, second]
+        );
+    }
+
+    #[tokio::test]
+    async fn net_http_004_count_release_denies_authority_loss_after_fetch() {
+        let source = Arc::new(ScriptedChannelAuthority {
+            allowed: AtomicBool::new(true),
+            checked: Mutex::new(Vec::new()),
+        });
+        let fence = ChannelReadReleaseFence {
+            source: source.clone(),
+            community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()),
+            channel_ids: vec![Uuid::new_v4()],
+            actor: vec![9; 32],
+            protected: None,
+        };
+
+        // The query has completed. A membership removal or an open-to-private
+        // transition now makes the authoritative DB check return false.
+        source.allowed.store(false, Ordering::SeqCst);
+        assert!(!fence.release().await);
+    }
+
+    /// NET-WS-009: a COUNT queued while access is valid must not become
+    /// visible if membership or channel visibility changes while the socket
+    /// is waiting for sink readiness.
+    #[tokio::test]
+    async fn net_ws_009_count_release_denies_authority_loss_before_socket_acceptance() {
+        let ready = Arc::new(AtomicBool::new(false));
+        let polled = Arc::new(tokio::sync::Notify::new());
+        let waker = Arc::new(Mutex::new(None));
+        let state = Arc::new(Mutex::new(MockSinkState::default()));
+        let sink = ReadinessBarrierSink {
+            ready: Arc::clone(&ready),
+            polled: Arc::clone(&polled),
+            waker: Arc::clone(&waker),
+            state: Arc::clone(&state),
+        };
+        let source = Arc::new(ScriptedChannelAuthority {
+            allowed: AtomicBool::new(true),
+            checked: Mutex::new(Vec::new()),
+        });
+        let release = Arc::new(ChannelReadReleaseFence {
+            source: source.clone(),
+            community_id: buzz_core::tenant::CommunityId::from_uuid(Uuid::new_v4()),
+            channel_ids: vec![Uuid::new_v4()],
+            actor: vec![10; 32],
+            protected: None,
+        });
+        let (data_tx, data_rx) = mpsc::channel(1);
+        let (_ctrl_tx, ctrl_rx) = mpsc::channel(1);
+        let cancel = CancellationToken::new();
+        data_tx
+            .send(OutboundData::guarded(
+                WsMessage::Text(RelayMessage::count("count-race", 1).into()),
+                release,
+            ))
+            .await
+            .expect("queue protected COUNT");
+        drop(data_tx);
+
+        let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone()));
+        polled.notified().await;
+        source.allowed.store(false, Ordering::SeqCst);
+        ready.store(true, Ordering::SeqCst);
+        waker
+            .lock()
+            .expect("barrier waker poisoned")
+            .take()
+            .expect("poll_ready registered a waker")
+            .wake();
+        task.await.expect("send loop joins");
+
+        assert!(cancel.is_cancelled());
+        assert!(
+            state
+                .lock()
+                .expect("barrier sink poisoned")
+                .messages
+                .is_empty(),
+            "COUNT must not reach start_send after its channel authority is lost"
+        );
+    }
+
+    /// O4-EXP-SESSION-001: session expiry wins over a COUNT that has been
+    /// computed and queued but has not yet crossed the socket boundary.
+    #[tokio::test(start_paused = true)]
+    async fn o4_exp_session_001_count_is_suppressed_when_deadline_precedes_emission() {
+        let ready = Arc::new(AtomicBool::new(false));
+        let polled = Arc::new(tokio::sync::Notify::new());
+        let waker = Arc::new(Mutex::new(None));
+        let state = Arc::new(Mutex::new(MockSinkState::default()));
+        let sink = ReadinessBarrierSink {
+            ready: Arc::clone(&ready),
+            polled: Arc::clone(&polled),
+            waker: Arc::clone(&waker),
+            state: Arc::clone(&state),
+        };
+        let (data_tx, data_rx) = mpsc::channel(1);
+        let (_ctrl_tx, ctrl_rx) = mpsc::channel(1);
+        let cancel = CancellationToken::new();
+        data_tx
+            .send(OutboundData::plain(WsMessage::Text(
+                RelayMessage::count("expiry-race", 1).into(),
+            )))
+            .await
+            .expect("queue COUNT before expiry");
+        drop(data_tx);
+
+        let task = tokio::spawn(send_loop_inner(sink, data_rx, ctrl_rx, cancel.clone()));
+        polled.notified().await;
+        tokio::time::advance(Duration::from_millis(500)).await;
+        cancel.cancel();
+        task.await.expect("send loop joins");
+
+        assert!(cancel.is_cancelled());
+        let messages = &state.lock().expect("barrier sink poisoned").messages;
+        assert!(
+            messages
+                .iter()
+                .all(|message| !matches!(message, WsMessage::Text(_))),
+            "expired session must not emit its queued COUNT"
+        );
+    }
+
     #[tokio::test]
     async fn send_loop_batch_one_preserves_single_frame_flush_behavior() {
         let (data_tx, data_rx) = mpsc::channel(1);
         let (_ctrl_tx, ctrl_rx) = mpsc::channel(1);
         data_tx
-            .send(WsMessage::Text("single".into()))
+            .send(OutboundData::plain(WsMessage::Text("single".into())))
             .await
             .expect("queue data frame");
 
@@ -844,11 +1430,11 @@ mod tests {
         let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH);
         let (ctrl_tx, ctrl_rx) = mpsc::channel(1);
         data_tx
-            .send(WsMessage::Text("data-0".into()))
+            .send(OutboundData::plain(WsMessage::Text("data-0".into())))
             .await
             .expect("queue data frame");
         data_tx
-            .send(WsMessage::Text("data-1".into()))
+            .send(OutboundData::plain(WsMessage::Text("data-1".into())))
             .await
             .expect("queue data frame");
         ctrl_tx
@@ -905,4 +1491,35 @@ mod tests {
             "Close is sent only after the reason frame is flushed"
         );
     }
+
+    #[tokio::test]
+    async fn protected_count_denial_is_visible_before_terminal_close() {
+        let (_data_tx, data_rx) = mpsc::channel(1);
+        let (ctrl_tx, ctrl_rx) = mpsc::channel(1);
+        ctrl_tx
+            .send(WsMessage::Text(
+                RelayMessage::closed(
+                    "deny-count",
+                    "auth-required: protected authorization denied",
+                )
+                .into(),
+            ))
+            .await
+            .expect("queue protected COUNT denial");
+
+        let cancel = CancellationToken::new();
+        cancel.cancel();
+        let (sink, state) = MockSink::new(None);
+        send_loop_inner(sink, data_rx, ctrl_rx, cancel).await;
+
+        let state = state.lock().expect("mock sink poisoned");
+        assert_eq!(state.messages.len(), 2);
+        assert!(matches!(
+            &state.messages[0],
+            WsMessage::Text(text)
+                if text.as_str().contains("deny-count")
+                    && text.as_str().contains("protected authorization denied")
+        ));
+        assert!(matches!(state.messages[1], WsMessage::Close(_)));
+    }
 }

crates/buzz-relay/src/handlers/auth.rs

@@ -91,6 +91,28 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc<ConnectionState>, state:
         Ok(mut auth_ctx) => {
             let pubkey = auth_ctx.pubkey;
 
+            if state
+                .protected_transport()
+                .and_then(|runtime| runtime.mode_for_domain(conn.tenant.community()))
+                == Some(
+                    crate::authorization_runtime::finalization::AuthorizationMode::DenyProtected,
+                )
+            {
+                metrics::counter!(
+                    "buzz_auth_failures_total",
+                    "reason" => "deny_protected"
+                )
+                .increment(1);
+                *conn.auth_state.write().await = AuthState::Failed;
+                conn.send(RelayMessage::ok(
+                    &event_id_hex,
+                    false,
+                    "auth-required: protected authorization unavailable",
+                ));
+                conn.cancel.cancel();
+                return;
+            }
+
             // Community ban gate (NIP-42 seam). Runs immediately after auth
             // verification succeeds and before the allowlist and relay-membership
             // gates, per COMMUNITY_MODERATION_PLAN.md §0 decision 4 and the

crates/buzz-relay/src/protected_surface.rs

@@ -755,8 +755,8 @@ pub enum EffectPermitError {
     /// The closed identifier has no registry row.
     #[error("protected effect is not classified")]
     Unclassified,
-    /// The registered effect is deliberately unavailable in Enforce.
-    #[error("protected effect is unavailable in enforce mode")]
+    /// The registered effect is deliberately unavailable in protected mode.
+    #[error("protected effect is unavailable in protected mode")]
     Unavailable(UnavailableReason),
 }
 
@@ -769,7 +769,7 @@ pub fn require_effect_permit(
         .iter()
         .find(|surface| surface.id == id)
         .ok_or(EffectPermitError::Unclassified)?;
-    if mode == Some(AuthorizationMode::Enforce) {
+    if mode.is_some_and(AuthorizationMode::protects_surfaces) {
         if let EnforceDisposition::DenyBeforeEffect(reason) = surface.enforce {
             return Err(EffectPermitError::Unavailable(reason));
         }

crates/buzz-workflow/src/lib.rs

@@ -33,11 +33,13 @@
 pub mod action_sink;
 pub mod error;
 pub mod executor;
+pub mod mutation_gate;
 pub mod schema;
 
 pub use action_sink::{ActionSink, ActionSinkError};
 pub use error::{PartialProgress, WorkflowError};
 pub use executor::ExecutionResult;
+pub use mutation_gate::MutationGate;
 pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef};
 
 use std::collections::HashMap;
@@ -87,6 +89,9 @@ pub struct WorkflowEngine {
     /// Action sink for executing side-effects (SendMessage, etc.).
     /// Late-initialized via [`set_action_sink`] after `AppState` construction.
     pub(crate) action_sink: OnceLock<Arc<dyn ActionSink>>,
+    /// Provider-neutral gate evaluated before every mutation or external effect.
+    /// Late-initialized by the embedding relay after `AppState` construction.
+    pub(crate) mutation_gate: OnceLock<Arc<dyn MutationGate>>,
     /// Short-TTL cache for the per-event enabled-workflow lookup, keyed
     /// `(community_id, channel_id)`. Most channels have no workflows, so this
     /// removes one SELECT from nearly every ingested event.
@@ -115,6 +120,7 @@ impl WorkflowEngine {
             run_semaphore,
             last_fired: DashMap::new(),
             action_sink: OnceLock::new(),
+            mutation_gate: OnceLock::new(),
             workflow_cache: moka::sync::Cache::builder()
                 .max_capacity(10_000)
                 .time_to_live(std::time::Duration::from_secs(10))
@@ -180,6 +186,38 @@ impl WorkflowEngine {
         }
     }
 
+    /// Set the workflow mutation gate. Called once by an embedding relay.
+    ///
+    /// # Panics
+    /// Panics if called more than once.
+    pub fn set_mutation_gate(&self, gate: Arc<dyn MutationGate>) {
+        if self.mutation_gate.set(gate).is_err() {
+            panic!("mutation_gate already initialized");
+        }
+    }
+
+    /// Require current authority before a workflow mutation or external effect.
+    ///
+    /// A standalone engine with no installed gate preserves legacy behavior.
+    /// Once an embedding relay installs a gate, every engine-owned mutation
+    /// door calls this method before touching durable or external state.
+    pub(crate) fn require_mutation(&self, community_id: CommunityId) -> Result<(), WorkflowError> {
+        mutation_gate::require_configured_mutation(
+            self.mutation_gate.get().map(AsRef::as_ref),
+            community_id,
+        )
+    }
+
+    pub(crate) fn require_outbound_webhook(
+        &self,
+        community_id: CommunityId,
+    ) -> Result<(), WorkflowError> {
+        mutation_gate::require_configured_outbound_webhook(
+            self.mutation_gate.get().map(AsRef::as_ref),
+            community_id,
+        )
+    }
+
     /// Get the action sink reference.
     ///
     /// Returns `Err(WorkflowError)` if the sink has not been initialized via
@@ -217,6 +255,13 @@ impl WorkflowEngine {
         result: Result<ExecutionResult, (WorkflowError, PartialProgress)>,
         existing_trace: Option<Vec<serde_json::Value>>,
     ) {
+        if let Err(error) = self.require_mutation(community_id) {
+            tracing::warn!(
+                run_id = %run_id,
+                "Skipping workflow finalization because mutation authority is unavailable: {error}"
+            );
+            return;
+        }
         let prefix = existing_trace.unwrap_or_default();
 
         match result {
@@ -395,6 +440,14 @@ impl WorkflowEngine {
                 continue;
             }
 
+            if let Err(error) = self.require_mutation(community_id) {
+                tracing::warn!(
+                    workflow_id = %workflow.id,
+                    "Skipping workflow because mutation authority is unavailable: {error}"
+                );
+                continue;
+            }
+
             let trigger_event_id_bytes = event.event.id.as_bytes().to_vec();
             let run_id = match self
                 .db
@@ -608,6 +661,14 @@ impl WorkflowEngine {
                     continue;
                 }
 
+                if let Err(error) = self.require_mutation(community_id) {
+                    tracing::warn!(
+                        workflow_id = %workflow.id,
+                        "Cron tick: skipping workflow because mutation authority is unavailable: {error}"
+                    );
+                    continue;
+                }
+
                 // Durable at-most-once claim — the cross-pod fire boundary.
                 // The loser receives `None` and skips BEFORE any run creation or
                 // side effect. `community_id` is the workflow row's own

crates/buzz-workflow/src/mutation_gate.rs

@@ -0,0 +1,77 @@
+//! Provider-neutral admission gate for workflow mutations and external effects.
+
+use buzz_core::tenant::CommunityId;
+
+use crate::WorkflowError;
+
+/// Server-owned gate evaluated before every workflow mutation or external effect.
+///
+/// The workflow engine deliberately knows nothing about identity providers,
+/// leases, or deployment configuration. A relay can install a gate that denies
+/// an authorization domain until it has a transaction-owning executor. When no
+/// gate is installed, the standalone engine preserves its legacy behavior.
+pub trait MutationGate: Send + Sync {
+    /// Require current authority for one server-resolved authorization domain.
+    fn require_mutation(&self, community_id: CommunityId) -> Result<(), WorkflowError>;
+
+    /// Require authority for an outbound network effect.
+    fn require_outbound_webhook(&self, community_id: CommunityId) -> Result<(), WorkflowError> {
+        self.require_mutation(community_id)
+    }
+}
+
+pub(crate) fn require_configured_mutation(
+    gate: Option<&dyn MutationGate>,
+    community_id: CommunityId,
+) -> Result<(), WorkflowError> {
+    match gate {
+        Some(gate) => gate.require_mutation(community_id),
+        None => Ok(()),
+    }
+}
+
+pub(crate) fn require_configured_outbound_webhook(
+    gate: Option<&dyn MutationGate>,
+    community_id: CommunityId,
+) -> Result<(), WorkflowError> {
+    match gate {
+        Some(gate) => gate.require_outbound_webhook(community_id),
+        None => Ok(()),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    use super::*;
+
+    struct DenyGate(AtomicUsize);
+
+    impl MutationGate for DenyGate {
+        fn require_mutation(&self, _community_id: CommunityId) -> Result<(), WorkflowError> {
+            self.0.fetch_add(1, Ordering::SeqCst);
+            Err(WorkflowError::Unauthorized(
+                "synthetic mutation denial".into(),
+            ))
+        }
+    }
+
+    #[test]
+    fn absent_gate_preserves_legacy_and_configured_denial_is_authoritative() {
+        let community_id = CommunityId::from_uuid(uuid::Uuid::from_u128(1));
+        assert!(require_configured_mutation(None, community_id).is_ok());
+
+        let gate = DenyGate(AtomicUsize::new(0));
+        assert!(matches!(
+            require_configured_mutation(Some(&gate), community_id),
+            Err(WorkflowError::Unauthorized(_))
+        ));
+        assert_eq!(gate.0.load(Ordering::SeqCst), 1);
+        assert!(matches!(
+            require_configured_outbound_webhook(Some(&gate), community_id),
+            Err(WorkflowError::Unauthorized(_))
+        ));
+        assert_eq!(gate.0.load(Ordering::SeqCst), 2);
+    }
+}