feat(auth): install typed provider provenance

Canonical O4 slice: 09_provider_installation
Final-tree source: e45553ea65b66631771c16908513f983d3c0b881

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

crates/buzz-auth/src/lib.rs

@@ -17,10 +17,18 @@
 
 /// Channel access checking trait and helpers.
 pub mod access;
+/// Complete Blossom operation authentication verification.
+pub mod blossom;
 /// Versioned, transport-neutral authorization context.
 pub mod context;
 /// Authentication error types.
 pub mod error;
+/// Trusted-workspace adapter for sealed verifier and binding evidence.
+pub mod evidence_adapter;
+/// Federated provider-evidence finalization.
+pub mod finalization;
+/// Bounded, versioned authorization leases.
+pub mod lease;
 /// NIP-42 challenge–response authentication.
 pub mod nip42;
 /// NIP-98 HTTP Auth verification (kind:27235).
@@ -44,10 +52,25 @@ pub use context::{
     DelegationCapability, DelegationExpiry, DirectBindingResolutionSink, EnrollmentMode,
     ExistingBindingResolutionSink, FederatedAuthorityAdapter, FederatedAuthorization,
     FederatedIdentityRequirement, FederatedPrincipal, NostrAuthority, ResolvedFederatedPolicy,
-    VerifiedFederatedAssertion, VerifiedKeyAttestation, VerifiedNostrProof, VerifiedOwnerAdmission,
+    VerifiedFederatedAssertion, VerifiedKeyAttestation, VerifiedNostrProof,
+    VerifiedOperationBinding, VerifiedOperationBindingKind, VerifiedOwnerAdmission,
     VerifiedTransportDelegation, VersionedBindingRef,
 };
 pub use error::AuthError;
+pub use evidence_adapter::{
+    ActiveBindingResolution, EvidenceAdapterError, VerifiedDelegationOutput,
+    VerifiedEvidenceAdapter,
+};
+pub use finalization::{AuthorizationFinalizer, FinalizationError, VerificationOnlyDisposition};
+pub use lease::{
+    AccessLeasePolicy, ApplicationLeaseLimit, AuthorizationClock, AuthorizationClockError,
+    AuthorizationClockSkew, AuthorizationLease, AuthorizationLeaseValidator,
+    AuthorizationOperationGuard, AuthorizationTime, BindingLeaseBound, LeaseIssueError,
+    LeasePolicyError, LeaseRenewalAction, LeaseRenewalLeadTime, LeaseRenewalSchedule,
+    LeaseUseRequirement, LeaseValidationError, LeaseVersion, SharedAuthorizationClock,
+    SystemAuthorizationClock, VerificationStatusPolicy, MAX_APPLICATION_LEASE_SECONDS,
+    MAX_AUTHORIZATION_CLOCK_SKEW_SECONDS, MAX_LEASE_RENEWAL_LEAD_SECONDS,
+};
 pub use nip42::{generate_challenge, verify_nip42_event};
 pub use nip98::verify_nip98_event;
 pub use nip98_replay::{
@@ -55,13 +78,14 @@ pub use nip98_replay::{
     MAX_REPLAY_TTL_SECS,
 };
 pub use provider::{
-    AuthorizationAuthority, AuthorizationCapability, AuthorizationClock, AuthorizationDenial,
+    resolve_authorization, AuthorizationAuthority, AuthorizationCapability,
+    AuthorizationClock as ProviderAuthorizationClock, AuthorizationDenial,
     AuthorizationDenialReason, AuthorizationOutcome, AuthorizationProfileId, AuthorizationProvider,
     AuthorizationProviderFuture, AuthorizationRequest, AuthorizationRuntime, CapabilitySet,
-    CapabilitySnapshot, DecisionSource, PolicyVersion, ProviderAllow, ProviderAllowReason,
-    ProviderAuthorizationError, ProviderContractError, ProviderDecision, ProviderTimeout,
-    ProviderUnavailable, ProviderUnavailableReason, RetryAfter, MAX_PROVIDER_FRESHNESS_SECONDS,
-    MAX_PROVIDER_TIMEOUT,
+    CapabilitySnapshot, DecisionSource, OwnerAdmissionError, PolicyVersion, ProviderAllow,
+    ProviderAllowReason, ProviderAuthorizationError, ProviderContractError, ProviderDecision,
+    ProviderTimeout, ProviderUnavailable, ProviderUnavailableReason, RetryAfter,
+    MAX_PROVIDER_FRESHNESS_SECONDS, MAX_PROVIDER_TIMEOUT,
 };
 pub use rate_limit::{
     ip_rate_limit_key, rate_limit_key, LimitType, RateLimitConfig, RateLimitResult, RateLimiter,

crates/buzz-auth/src/provider/mod.rs

@@ -15,10 +15,10 @@ use crate::context::{
     authority::{resolve_direct_binding, resolve_existing_binding},
     resolve_current_federated_policy, AdmissionExpiry, AssertionTransport, AuthContext,
     AuthContextError, AuthContextInput, AuthMethod, AuthTransport, AuthoritativeBindingResolution,
-    AuthoritativeFederatedResolution, AuthorityAdapterError, BindingVersion,
+    AuthoritativeFederatedResolution, AuthorityAdapterError, AuthorizationReason, BindingVersion,
     CapabilityFinalizationSeal, FederatedAuthorityAdapter, FederatedPolicyStamp,
     FederatedPrincipal, ResolvedFederatedPolicy, VerifiedFederatedAssertion, VerifiedNostrProof,
-    VerifiedOwnerAdmission,
+    VerifiedOwnerAdmission, VersionedBindingRef,
 };
 
 const MAX_OPAQUE_ID_BYTES: usize = 256;
@@ -68,6 +68,11 @@ impl fmt::Debug for AuthorizationCapability {
 pub struct CapabilitySet(Vec<AuthorizationCapability>);
 
 impl CapabilitySet {
+    /// Build the exact one-capability request used by on-demand authorization.
+    pub fn single(capability: AuthorizationCapability) -> Self {
+        Self(vec![capability])
+    }
+
     /// Build a non-empty set, sorting and removing duplicate capabilities.
     pub fn new(
         mut capabilities: Vec<AuthorizationCapability>,
@@ -91,6 +96,11 @@ impl CapabilitySet {
             .iter()
             .all(|capability| self.0.binary_search(capability).is_ok())
     }
+
+    /// Return whether the set includes one exact portable capability.
+    pub fn contains(&self, capability: AuthorizationCapability) -> bool {
+        self.0.binary_search(&capability).is_ok()
+    }
 }
 
 impl fmt::Debug for CapabilitySet {
@@ -124,6 +134,10 @@ impl AuthorizationProfileId {
         }
         Ok(Self(value))
     }
+    #[cfg(test)]
+    pub(crate) fn new(value: impl Into<String>) -> Result<Self, ProviderContractError> {
+        Self::from_server_configuration(value)
+    }
     /// Exact profile identifier for provider routing.
     pub fn as_str(&self) -> &str {
         &self.0
@@ -354,6 +368,90 @@ impl AuthorizationRequest {
         let Some(delegation) = proof.verified_delegation() else {
             return Err(ProviderContractError::DelegationRequired);
         };
+        if !delegation.capability().is_transport_wide() {
+            return Err(ProviderContractError::UnsupportedDelegationScope);
+        }
+        if delegation.owner_pubkey() != owner.bound_pubkey() {
+            return Err(ProviderContractError::DelegatedOwnerMismatch);
+        }
+        if delegation
+            .expires_at()
+            .is_some_and(|bound| bound.is_expired_at(now_unix_seconds))
+        {
+            return Err(ProviderContractError::DelegationExpired);
+        }
+        if owner
+            .expires_at()
+            .is_some_and(|bound| bound.is_expired_at(now_unix_seconds))
+        {
+            return Err(ProviderContractError::BindingExpired);
+        }
+        let evidence_valid_from = federated_policy.stamp().effective_from();
+        let mut evidence_valid_until = federated_policy.stamp().effective_until();
+        if let Some(delegation) = delegation.expires_at() {
+            evidence_valid_until = evidence_valid_until.min(delegation.unix_seconds());
+        }
+        if let Some(binding) = owner.expires_at() {
+            evidence_valid_until = evidence_valid_until.min(binding.unix_seconds());
+        }
+        Ok(Self {
+            authorization_domain: proof.authorization_domain(),
+            transport: proof.authorized_transport(),
+            actor_pubkey: proof.actor_pubkey(),
+            proof_method: proof.proof_method(),
+            authority: AuthorizationAuthority::Delegated {
+                owner_pubkey: owner.bound_pubkey(),
+                binding_id: owner.binding_id(),
+                binding_version: owner.binding_version(),
+            },
+            principal: owner.principal().clone(),
+            key_attested: false,
+            assertion_transport: None,
+            assertion_not_before: None,
+            assertion_expires_at: None,
+            federated_policy: federated_policy.into_stamp(),
+            requested_capabilities,
+            correlation_id,
+            decision_source: DecisionSource::DelegatedOwnerBinding,
+            evidence_valid_from,
+            evidence_valid_until,
+        })
+    }
+
+    /// Build a delegated request from a sealed active binding-store record.
+    ///
+    /// The evidence adapter can create this record only from typed current
+    /// storage output. Enrolled-in-request bindings are rejected so delegated
+    /// authorization retains O3's existing-active requirement.
+    pub fn delegated_from_active_binding(
+        proof: &VerifiedNostrProof,
+        owner: &VersionedBindingRef,
+        federated_policy: ResolvedFederatedPolicy,
+        requested_capabilities: CapabilitySet,
+        correlation_id: Uuid,
+        now_unix_seconds: u64,
+    ) -> Result<Self, ProviderContractError> {
+        if correlation_id.is_nil() {
+            return Err(ProviderContractError::InvalidCorrelationId);
+        }
+        validate_federated_policy(
+            &federated_policy,
+            proof.authorization_domain(),
+            correlation_id,
+            now_unix_seconds,
+        )?;
+        if proof.authorization_domain() != owner.authorization_domain() {
+            return Err(ProviderContractError::AuthorizationDomainMismatch);
+        }
+        if owner.authorization_reason() != AuthorizationReason::ExistingBinding {
+            return Err(ProviderContractError::DelegatedBindingNotExistingActive);
+        }
+        let Some(delegation) = proof.verified_delegation() else {
+            return Err(ProviderContractError::DelegationRequired);
+        };
+        if !delegation.capability().is_transport_wide() {
+            return Err(ProviderContractError::UnsupportedDelegationScope);
+        }
         if delegation.owner_pubkey() != owner.bound_pubkey() {
             return Err(ProviderContractError::DelegatedOwnerMismatch);
         }
@@ -1133,7 +1231,6 @@ impl CapabilitySnapshot {
     pub const fn reason(&self) -> ProviderAllowReason {
         self.reason
     }
-
     /// Consume a direct capability decision and finalize authoritative context.
     ///
     /// The current enrollment policy is reread after provider I/O, then the
@@ -1241,7 +1338,7 @@ impl CapabilitySnapshot {
         {
             return Err(ProviderContractError::CapabilityBindingChanged.into());
         }
-        let admission = VerifiedOwnerAdmission::new(
+        let admission = VerifiedOwnerAdmission::from_capability_snapshot(
             self.authorization_domain,
             self.principal,
             AdmissionExpiry::new(self.effective_until)?,
@@ -1363,6 +1460,38 @@ impl CapabilitySnapshot {
         }
         Ok(())
     }
+
+    /// Derive current delegated-owner admission for one exact active binding.
+    pub fn verified_owner_admission(
+        &self,
+        owner: &VersionedBindingRef,
+    ) -> Result<VerifiedOwnerAdmission, OwnerAdmissionError> {
+        if self.decision_source != DecisionSource::DelegatedOwnerBinding {
+            return Err(OwnerAdmissionError::NotDelegatedSnapshot);
+        }
+        if self.authorization_domain != owner.authorization_domain() {
+            return Err(OwnerAdmissionError::AuthorizationDomainMismatch);
+        }
+        if self.owner_pubkey != Some(owner.bound_pubkey()) {
+            return Err(OwnerAdmissionError::OwnerKeyMismatch);
+        }
+        if self.principal != *owner.principal() {
+            return Err(OwnerAdmissionError::PrincipalMismatch);
+        }
+        if self.binding_id != Some(owner.binding_id()) {
+            return Err(OwnerAdmissionError::BindingIdMismatch);
+        }
+        if self.binding_version != Some(owner.binding_version()) {
+            return Err(OwnerAdmissionError::BindingVersionMismatch);
+        }
+        let fresh_until = AdmissionExpiry::new(self.fresh_until)
+            .map_err(|_| OwnerAdmissionError::InvalidFreshnessBound)?;
+        Ok(VerifiedOwnerAdmission::from_capability_snapshot(
+            self.authorization_domain,
+            self.principal.clone(),
+            fresh_until,
+        ))
+    }
 }
 
 fn finalization_time<E>(
@@ -1435,6 +1564,40 @@ where
     }
 }
 
+/// Failure to derive delegated-owner admission from a capability snapshot.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
+#[non_exhaustive]
+pub enum OwnerAdmissionError {
+    #[error("capability snapshot does not carry delegated owner authority")]
+    NotDelegatedSnapshot,
+    #[error("capability snapshot and owner binding domains do not match")]
+    AuthorizationDomainMismatch,
+    #[error("capability snapshot and owner binding keys do not match")]
+    OwnerKeyMismatch,
+    #[error("capability snapshot and owner binding principals do not match")]
+    PrincipalMismatch,
+    #[error("capability snapshot and owner binding identifiers do not match")]
+    BindingIdMismatch,
+    #[error("capability snapshot and owner binding versions do not match")]
+    BindingVersionMismatch,
+    #[error("capability snapshot has an invalid admission freshness bound")]
+    InvalidFreshnessBound,
+}
+
+impl OwnerAdmissionError {
+    pub const fn code(self) -> &'static str {
+        match self {
+            Self::NotDelegatedSnapshot => "authorization_owner_admission_001",
+            Self::AuthorizationDomainMismatch => "authorization_owner_admission_002",
+            Self::OwnerKeyMismatch => "authorization_owner_admission_003",
+            Self::PrincipalMismatch => "authorization_owner_admission_004",
+            Self::BindingIdMismatch => "authorization_owner_admission_005",
+            Self::BindingVersionMismatch => "authorization_owner_admission_006",
+            Self::InvalidFreshnessBound => "authorization_owner_admission_007",
+        }
+    }
+}
+
 impl fmt::Debug for CapabilitySnapshot {
     fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
         formatter
@@ -1496,7 +1659,7 @@ impl fmt::Debug for AuthorizationOutcome {
 /// completes, an allowed decision is checked against exactly one fresh sample.
 /// Provider freshness and all effective evidence bounds use that same value;
 /// callers must not precompute and pass a decision-start timestamp.
-async fn resolve_authorization(
+pub async fn resolve_authorization(
     provider: &dyn AuthorizationProvider,
     request: &AuthorizationRequest,
     clock: &dyn AuthorizationClock,
@@ -1716,6 +1879,9 @@ pub enum ProviderContractError {
     /// A capability snapshot was presented to a different configured runtime.
     #[error("provider capability snapshot does not belong to this authorization runtime")]
     AuthorizationRuntimeMismatch,
+    /// A narrower operation-bound delegation reached a transport-wide request path.
+    #[error("delegation scope is not valid for transport-wide authorization")]
+    UnsupportedDelegationScope,
 }
 
 impl ProviderContractError {
@@ -1756,6 +1922,7 @@ impl ProviderContractError {
             Self::CapabilityBindingChanged => "authorization_provider_contract_032",
             Self::FederatedPolicyChanged => "authorization_provider_contract_033",
             Self::AuthorizationRuntimeMismatch => "authorization_provider_contract_034",
+            Self::UnsupportedDelegationScope => "authorization_provider_contract_035",
         }
     }
 }

crates/buzz-auth/src/provider/tests.rs

@@ -336,15 +336,13 @@ fn capability_coverage_is_exhaustive(capability: AuthorizationCapability) {
 fn proof_method_for_transport(transport: AuthTransport) -> AuthMethod {
     match transport {
         AuthTransport::RelayWebSocket => AuthMethod::Nip42,
-        AuthTransport::HttpBridge | AuthTransport::Git | AuthTransport::MediaDownload => {
-            AuthMethod::Nip98
-        }
-        AuthTransport::MediaUpload => AuthMethod::Blossom,
+        AuthTransport::HttpBridge | AuthTransport::Git => AuthMethod::Nip98,
+        AuthTransport::MediaUpload | AuthTransport::MediaDownload => AuthMethod::Blossom,
         AuthTransport::Audio => AuthMethod::Nip42,
     }
 }
 
-fn all_contract_errors() -> [ProviderContractError; 34] {
+fn all_contract_errors() -> [ProviderContractError; 35] {
     [
         ProviderContractError::EmptyCapabilitySet,
         ProviderContractError::EmptyProfileId,
@@ -380,6 +378,7 @@ fn all_contract_errors() -> [ProviderContractError; 34] {
         ProviderContractError::CapabilityBindingChanged,
         ProviderContractError::FederatedPolicyChanged,
         ProviderContractError::AuthorizationRuntimeMismatch,
+        ProviderContractError::UnsupportedDelegationScope,
     ]
 }
 
@@ -1837,6 +1836,12 @@ async fn delegated_owner_admission_does_not_require_owner_assertion() {
     assert_eq!(snapshot.binding_id(), Some(Uuid::from_u128(10)));
     assert_eq!(snapshot.binding_version(), Some(BindingVersion::INITIAL));
     assert_eq!(snapshot.transport(), AuthTransport::RelayWebSocket);
+    let admission = snapshot
+        .verified_owner_admission(&existing_binding(&owner))
+        .expect("delegated snapshot matches the exact owner binding");
+    assert_eq!(admission.authorization_domain(), domain(1));
+    assert_eq!(admission.principal(), request.principal());
+    assert_eq!(admission.fresh_until().unix_seconds(), 180);
 }
 
 #[tokio::test]

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

@@ -0,0 +1,406 @@
+//! Relay-authenticated authority carried only across ephemeral Redis fan-out.
+
+use std::{fmt, sync::Arc};
+
+use async_trait::async_trait;
+use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
+use buzz_core::CommunityId;
+use hmac::{Hmac, KeyInit, Mac};
+use sha2::{Digest, Sha256};
+use thiserror::Error;
+
+use super::executor::{revalidate_ephemeral_claim, EphemeralAuthorityClaim};
+use super::transport::ProtectedAuthorization;
+use crate::connection::QueuedOutboundReleaseFence;
+use crate::state::AppState;
+
+const DOMAIN_SEPARATOR: &[u8] = b"buzz-ephemeral-redis-authority-v1";
+type HmacSha256 = Hmac<Sha256>;
+const PRESENCE_PREFIX: &str = "pa1:";
+
+#[derive(Clone, serde::Deserialize, serde::Serialize)]
+pub(crate) struct ProtectedPresenceValue {
+    pub(crate) status: String,
+    pub(crate) context_id: [u8; 32],
+    pub(crate) authority: String,
+}
+
+impl fmt::Debug for ProtectedPresenceValue {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ProtectedPresenceValue")
+            .field("status", &"[redacted]")
+            .field("context_id", &"[redacted]")
+            .field("authority", &"[redacted]")
+            .finish()
+    }
+}
+
+pub(crate) fn encode_presence(
+    status: String,
+    context_id: [u8; 32],
+    authority: String,
+) -> Result<String, EphemeralAuthorityError> {
+    let value = serde_json::to_vec(&ProtectedPresenceValue {
+        status,
+        context_id,
+        authority,
+    })?;
+    Ok(format!(
+        "{PRESENCE_PREFIX}{}",
+        URL_SAFE_NO_PAD.encode(value)
+    ))
+}
+
+pub(crate) fn decode_presence(
+    value: &str,
+) -> Result<Option<ProtectedPresenceValue>, EphemeralAuthorityError> {
+    let Some(value) = value.strip_prefix(PRESENCE_PREFIX) else {
+        return Ok(None);
+    };
+    let value = URL_SAFE_NO_PAD
+        .decode(value)
+        .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?;
+    Ok(Some(serde_json::from_slice(&value)?))
+}
+
+/// Seal one provider-neutral sender authority for trusted relay Redis fan-out.
+pub(crate) fn seal(
+    state: &AppState,
+    authority: &ProtectedAuthorization,
+    event: &nostr::Event,
+) -> Result<String, EphemeralAuthorityError> {
+    seal_context(state, authority, event.id.to_bytes())
+}
+
+/// Seal enforcing authority for a non-event ephemeral effect boundary.
+pub(crate) fn seal_context(
+    state: &AppState,
+    authority: &ProtectedAuthorization,
+    context_id: [u8; 32],
+) -> Result<String, EphemeralAuthorityError> {
+    let claim = authority
+        .seal_ephemeral_delivery(context_id)?
+        .ok_or(EphemeralAuthorityError::AuthorityRequired)?;
+    seal_claim(&signing_key(state), &claim)
+}
+
+fn seal_claim(
+    signing_key: &[u8; 32],
+    claim: &EphemeralAuthorityClaim,
+) -> Result<String, EphemeralAuthorityError> {
+    let payload = serde_json::to_vec(claim)?;
+    let mut mac = <HmacSha256 as KeyInit>::new_from_slice(signing_key)
+        .map_err(|_| EphemeralAuthorityError::InvalidSignature)?;
+    mac.update(DOMAIN_SEPARATOR);
+    mac.update(&(payload.len() as u64).to_be_bytes());
+    mac.update(&payload);
+    let signature = mac.finalize().into_bytes();
+    Ok(format!(
+        "{}.{}",
+        URL_SAFE_NO_PAD.encode(payload),
+        URL_SAFE_NO_PAD.encode(signature)
+    ))
+}
+
+/// Verify and preflight a remote sender authority before any local fan-out.
+pub(crate) async fn verify(
+    state: &AppState,
+    community_id: CommunityId,
+    event: &nostr::Event,
+    token: &str,
+) -> Result<Arc<dyn QueuedOutboundReleaseFence>, EphemeralAuthorityError> {
+    let claim = verify_claim(&signing_key(state), token)?;
+    if claim.community_id != *community_id.as_uuid()
+        || claim.event_id != event.id.to_bytes()
+        || claim.actor_pubkey != event.pubkey.to_bytes()
+        || claim.capability != "community_write"
+    {
+        return Err(EphemeralAuthorityError::ContextMismatch);
+    }
+    let authority = Arc::new(RemoteEphemeralAuthority {
+        db: state.db.clone(),
+        claim,
+    });
+    if !authority.release().await {
+        return Err(EphemeralAuthorityError::ExpiredOrInvalidated);
+    }
+    Ok(authority)
+}
+
+/// Database/signing material retained by an internal cross-node effect owner.
+#[derive(Clone)]
+pub(crate) enum AuthorityTokenVerifier {
+    Database {
+        db: buzz_db::Db,
+        signing_key: [u8; 32],
+    },
+    #[cfg(test)]
+    TestAllow,
+    #[cfg(test)]
+    TestDeny,
+    #[cfg(test)]
+    TestConditional(Arc<std::sync::atomic::AtomicBool>),
+}
+
+/// Authority retained by an ephemeral effect owner through cleanup. The
+/// absolute lease deadline is available to synchronous realtime boundaries;
+/// invalidation and binding state remain asynchronously revalidated.
+#[derive(Clone)]
+pub(crate) struct RetainedEphemeralAuthority {
+    fence: Arc<dyn QueuedOutboundReleaseFence>,
+    expires_at: u64,
+}
+
+impl RetainedEphemeralAuthority {
+    pub(crate) fn expires_at(&self) -> u64 {
+        self.expires_at
+    }
+
+    pub(crate) async fn release(&self) -> bool {
+        self.is_time_valid() && self.fence.release().await
+    }
+
+    pub(crate) fn is_time_valid(&self) -> bool {
+        std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .is_ok_and(|duration| duration.as_secs() < self.expires_at)
+    }
+}
+
+impl AuthorityTokenVerifier {
+    /// Build from the shared relay signing secret without retaining it.
+    pub(crate) fn new(db: buzz_db::Db, relay_secret: &[u8]) -> Self {
+        Self::Database {
+            db,
+            signing_key: signing_key_from_secret(relay_secret),
+        }
+    }
+
+    #[cfg(test)]
+    pub(crate) fn allow_for_test() -> Self {
+        Self::TestAllow
+    }
+
+    #[cfg(test)]
+    pub(crate) fn deny_for_test() -> Self {
+        Self::TestDeny
+    }
+
+    #[cfg(test)]
+    pub(crate) fn conditional_for_test(gate: Arc<std::sync::atomic::AtomicBool>) -> Self {
+        Self::TestConditional(gate)
+    }
+
+    /// Verify exact context and current database authority at the effect owner.
+    pub(crate) async fn verify_context(
+        &self,
+        community_id: CommunityId,
+        context_id: [u8; 32],
+        token: &str,
+    ) -> Result<RetainedEphemeralAuthority, EphemeralAuthorityError> {
+        match self {
+            Self::Database { db, signing_key } => {
+                let claim = verify_claim(signing_key, token)?;
+                if claim.community_id != *community_id.as_uuid()
+                    || claim.event_id != context_id
+                    || claim.capability != "audio_join"
+                {
+                    return Err(EphemeralAuthorityError::ContextMismatch);
+                }
+                let expires_at = claim.expires_at;
+                let authority = RetainedEphemeralAuthority {
+                    fence: Arc::new(RemoteEphemeralAuthority {
+                        db: db.clone(),
+                        claim,
+                    }),
+                    expires_at,
+                };
+                if !authority.release().await {
+                    return Err(EphemeralAuthorityError::ExpiredOrInvalidated);
+                }
+                Ok(authority)
+            }
+            #[cfg(test)]
+            Self::TestAllow => Ok(RetainedEphemeralAuthority {
+                fence: Arc::new(TestConditionalAuthority {
+                    gate: Arc::new(std::sync::atomic::AtomicBool::new(true)),
+                }),
+                expires_at: test_authority_expiry(),
+            }),
+            #[cfg(test)]
+            Self::TestDeny => Err(EphemeralAuthorityError::ExpiredOrInvalidated),
+            #[cfg(test)]
+            Self::TestConditional(gate) => {
+                if !gate.load(std::sync::atomic::Ordering::SeqCst) {
+                    return Err(EphemeralAuthorityError::ExpiredOrInvalidated);
+                }
+                Ok(RetainedEphemeralAuthority {
+                    fence: Arc::new(TestConditionalAuthority {
+                        gate: Arc::clone(gate),
+                    }),
+                    expires_at: test_authority_expiry(),
+                })
+            }
+        }
+    }
+
+    /// Verify exact context, actor, and current database authority.
+    pub(crate) async fn verify_actor_context(
+        &self,
+        community_id: CommunityId,
+        context_id: [u8; 32],
+        actor_pubkey: [u8; 32],
+        token: &str,
+    ) -> Result<(), EphemeralAuthorityError> {
+        match self {
+            Self::Database { db, signing_key } => {
+                let claim = verify_claim(signing_key, token)?;
+                if claim.community_id != *community_id.as_uuid()
+                    || claim.event_id != context_id
+                    || claim.actor_pubkey != actor_pubkey
+                    || claim.capability != "community_write"
+                {
+                    return Err(EphemeralAuthorityError::ContextMismatch);
+                }
+                revalidate_ephemeral_claim(db, &claim)
+                    .await
+                    .map_err(|_| EphemeralAuthorityError::ExpiredOrInvalidated)
+            }
+            #[cfg(test)]
+            Self::TestAllow => Ok(()),
+            #[cfg(test)]
+            Self::TestDeny => Err(EphemeralAuthorityError::ExpiredOrInvalidated),
+            #[cfg(test)]
+            Self::TestConditional(gate) => gate
+                .load(std::sync::atomic::Ordering::SeqCst)
+                .then_some(())
+                .ok_or(EphemeralAuthorityError::ExpiredOrInvalidated),
+        }
+    }
+}
+
+#[cfg(test)]
+fn test_authority_expiry() -> u64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .expect("test wall clock after Unix epoch")
+        .as_secs()
+        + 3_600
+}
+
+#[cfg(test)]
+struct TestConditionalAuthority {
+    gate: Arc<std::sync::atomic::AtomicBool>,
+}
+
+#[cfg(test)]
+#[async_trait]
+impl QueuedOutboundReleaseFence for TestConditionalAuthority {
+    async fn release(&self) -> bool {
+        self.gate.load(std::sync::atomic::Ordering::SeqCst)
+    }
+}
+
+fn verify_claim(
+    signing_key: &[u8; 32],
+    token: &str,
+) -> Result<EphemeralAuthorityClaim, EphemeralAuthorityError> {
+    let (payload, signature) = token
+        .split_once('.')
+        .ok_or(EphemeralAuthorityError::InvalidEnvelope)?;
+    let payload = URL_SAFE_NO_PAD
+        .decode(payload)
+        .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?;
+    let signature = URL_SAFE_NO_PAD
+        .decode(signature)
+        .map_err(|_| EphemeralAuthorityError::InvalidEnvelope)?;
+    let mut mac = <HmacSha256 as KeyInit>::new_from_slice(signing_key)
+        .map_err(|_| EphemeralAuthorityError::InvalidSignature)?;
+    mac.update(DOMAIN_SEPARATOR);
+    mac.update(&(payload.len() as u64).to_be_bytes());
+    mac.update(&payload);
+    mac.verify_slice(&signature)
+        .map_err(|_| EphemeralAuthorityError::InvalidSignature)?;
+    Ok(serde_json::from_slice(&payload)?)
+}
+
+fn signing_key(state: &AppState) -> [u8; 32] {
+    signing_key_from_secret(state.relay_keypair.secret_key().as_secret_bytes())
+}
+
+fn signing_key_from_secret(secret: &[u8]) -> [u8; 32] {
+    let mut digest = Sha256::new();
+    digest.update(DOMAIN_SEPARATOR);
+    digest.update(secret);
+    digest.finalize().into()
+}
+
+struct RemoteEphemeralAuthority {
+    db: buzz_db::Db,
+    claim: EphemeralAuthorityClaim,
+}
+
+#[async_trait]
+impl QueuedOutboundReleaseFence for RemoteEphemeralAuthority {
+    async fn release(&self) -> bool {
+        revalidate_ephemeral_claim(&self.db, &self.claim)
+            .await
+            .is_ok()
+    }
+}
+
+#[derive(Debug, Error)]
+pub(crate) enum EphemeralAuthorityError {
+    #[error("ephemeral sender authority is required")]
+    AuthorityRequired,
+    #[error("ephemeral sender authority envelope is invalid")]
+    InvalidEnvelope,
+    #[error("ephemeral sender authority signature is invalid")]
+    InvalidSignature,
+    #[error("ephemeral sender authority context does not match")]
+    ContextMismatch,
+    #[error("ephemeral sender authority is expired or invalidated")]
+    ExpiredOrInvalidated,
+    #[error("ephemeral sender authority could not be sealed")]
+    Transport(#[from] super::transport::ProtectedTransportError),
+    #[error("ephemeral sender authority serialization failed")]
+    Serialization(#[from] serde_json::Error),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn protected_presence_round_trips_exact_authority_context() {
+        let context_id = [0x5a; 32];
+        let encoded = encode_presence("online".into(), context_id, "sealed-authority".into())
+            .expect("encode presence");
+        let decoded = decode_presence(&encoded)
+            .expect("decode presence")
+            .expect("protected value");
+        assert_eq!(decoded.status, "online");
+        assert_eq!(decoded.context_id, context_id);
+        assert_eq!(decoded.authority, "sealed-authority");
+        let debug = format!("{decoded:?}");
+        assert!(!debug.contains("online"));
+        assert!(!debug.contains("sealed-authority"));
+        assert!(!debug.contains("5a5a"));
+    }
+
+    #[test]
+    fn legacy_presence_is_distinct_from_protected_presence() {
+        assert!(decode_presence("online")
+            .expect("legacy presence is valid")
+            .is_none());
+    }
+
+    #[test]
+    fn malformed_protected_presence_fails_closed() {
+        assert!(matches!(
+            decode_presence("pa1:not-base64***"),
+            Err(EphemeralAuthorityError::InvalidEnvelope)
+        ));
+    }
+}

crates/buzz-relay/src/state.rs

@@ -16,7 +16,7 @@ use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
 use buzz_audit::AuditService;
-use buzz_auth::{AuthService, Nip98ReplayGuard};
+use buzz_auth::{AuthService, Nip98ReplayGuard, VerifiedFederatedAssertion, VerifiedNostrProof};
 use buzz_core::tenant::TenantContext;
 use buzz_core::CommunityId;
 use buzz_db::Db;
@@ -32,6 +32,7 @@ use deadpool_redis;
 use crate::audio::AudioRoomManager;
 use crate::config::Config;
 use crate::connection::ConnectionSubscriptions;
+use crate::connection::OutboundData;
 use crate::corporate_identity::CorporateIdentityService;
 use crate::subscription::SubscriptionRegistry;
 
@@ -41,7 +42,7 @@ type ScopedRateLimiter = DashMap<ScopedPubkeyKey, SlidingWindowCounter>;
 
 /// Per-connection entry in the connection manager.
 struct ConnEntry {
-    tx: mpsc::Sender<WsMessage>,
+    tx: mpsc::Sender<OutboundData>,
     /// Control-frame sender, drained ahead of data and before cancel wins in
     /// the send loop. Used to deliver a ban-disconnect frame that must reach
     /// the client before the socket is closed (see [`ConnectionManager::disconnect_pubkey`]).
@@ -55,9 +56,57 @@ struct ConnEntry {
     backpressure_count: Arc<AtomicU8>,
     subscriptions: ConnectionSubscriptions,
     authenticated_pubkey: Arc<std::sync::RwLock<Option<Vec<u8>>>>,
+    authenticated_owner_pubkey: Arc<std::sync::RwLock<Option<Vec<u8>>>>,
+    /// Sealed NIP-42 proof retained for every protected operation on this
+    /// connection. Legacy test registrations may intentionally leave it empty.
+    verified_nostr_proof: Arc<std::sync::RwLock<Option<Arc<VerifiedNostrProof>>>>,
+    /// Current direct federated evidence sealed during authentication.
+    verified_federated_assertion: Arc<std::sync::RwLock<Option<Arc<VerifiedFederatedAssertion>>>>,
+    /// The enforcing authority and its hard-expiry task share one lock so
+    /// concurrent operations can only tighten, never extend, the session.
+    protected_session: Arc<std::sync::Mutex<ProtectedSessionState>>,
     grace_limit: u8,
 }
 
+struct ProtectedSessionExpiryTask {
+    deadline: u64,
+    cancel: CancellationToken,
+}
+
+#[derive(Default)]
+struct ProtectedSessionState {
+    /// Retaining this value keeps its invalidation observer registered until
+    /// tighter authority replaces it or the connection is removed.
+    authority: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+    expiry: Option<ProtectedSessionExpiryTask>,
+}
+
+fn should_replace_protected_session_deadline(current: Option<u64>, candidate: u64) -> bool {
+    current.is_none_or(|current| candidate < current)
+}
+
+fn protected_session_wake_at_from_samples(
+    deadline: u64,
+    monotonic_anchor: tokio::time::Instant,
+    wall_now: std::time::Duration,
+    coarse_delay: std::time::Duration,
+) -> Option<tokio::time::Instant> {
+    let wall_remaining = std::time::Duration::from_secs(deadline).checked_sub(wall_now)?;
+    let conservative_coarse = coarse_delay.saturating_sub(std::time::Duration::from_secs(1));
+    monotonic_anchor.checked_add(wall_remaining.min(conservative_coarse))
+}
+
+fn protected_session_wake_at(
+    deadline: u64,
+    monotonic_anchor: tokio::time::Instant,
+    coarse_delay: std::time::Duration,
+) -> Option<tokio::time::Instant> {
+    let wall_now = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .ok()?;
+    protected_session_wake_at_from_samples(deadline, monotonic_anchor, wall_now, coarse_delay)
+}
+
 /// Community-scoped lifecycle registry shared by every long-lived socket type.
 ///
 /// A handler registers before durable active-state revalidation. Archival after
@@ -206,7 +255,7 @@ impl ConnectionManager {
     pub fn register(
         &self,
         conn_id: Uuid,
-        tx: mpsc::Sender<WsMessage>,
+        tx: mpsc::Sender<OutboundData>,
         ctrl_tx: mpsc::Sender<WsMessage>,
         cancel: CancellationToken,
         community_id: CommunityId,
@@ -226,6 +275,12 @@ impl ConnectionManager {
                 backpressure_count,
                 subscriptions,
                 authenticated_pubkey: Arc::new(std::sync::RwLock::new(None)),
+                authenticated_owner_pubkey: Arc::new(std::sync::RwLock::new(None)),
+                verified_nostr_proof: Arc::new(std::sync::RwLock::new(None)),
+                verified_federated_assertion: Arc::new(std::sync::RwLock::new(None)),
+                protected_session: Arc::new(
+                    std::sync::Mutex::new(ProtectedSessionState::default()),
+                ),
                 grace_limit,
             },
         );
@@ -241,7 +296,64 @@ impl ConnectionManager {
 
     /// Removes a connection from the registry.
     pub fn deregister(&self, conn_id: Uuid) {
-        self.connections.remove(&conn_id);
+        if let Some((_, entry)) = self.connections.remove(&conn_id) {
+            Self::clear_protected_session(&entry);
+        }
+    }
+
+    fn clear_protected_session(entry: &ConnEntry) {
+        if let Ok(mut session) = entry.protected_session.lock() {
+            if let Some(task) = session.expiry.take() {
+                task.cancel.cancel();
+            }
+            session.authority = None;
+        } else {
+            entry.cancel.cancel();
+        }
+    }
+
+    /// Atomically retain only authority with an earlier hard deadline.
+    ///
+    /// The comparison, task replacement, and authority replacement must stay
+    /// in this critical section: WebSocket handlers for one connection run
+    /// concurrently and a split read/install would permit lease extension.
+    fn retain_earlier_protected_session(
+        entry: &ConnEntry,
+        deadline: u64,
+        wake_at: tokio::time::Instant,
+        authority: Option<Arc<crate::authorization_runtime::transport::ProtectedAuthorization>>,
+        after_read_hook: Option<&dyn Fn()>,
+    ) -> bool {
+        if let Ok(mut session) = entry.protected_session.lock() {
+            let current = session.expiry.as_ref().map(|task| task.deadline);
+            if let Some(hook) = after_read_hook {
+                hook();
+            }
+            if !should_replace_protected_session_deadline(current, deadline) {
+                return false;
+            }
+            if let Some(previous) = session.expiry.take() {
+                previous.cancel.cancel();
+            }
+            let expiry_task = CancellationToken::new();
+            let expiry_cancel = expiry_task.clone();
+            let connection_cancel = entry.cancel.clone();
+            tokio::spawn(async move {
+                tokio::select! {
+                    _ = expiry_cancel.cancelled() => {}
+                    _ = tokio::time::sleep_until(wake_at) => connection_cancel.cancel(),
+                }
+            });
+            session.expiry = Some(ProtectedSessionExpiryTask {
+                deadline,
+                cancel: expiry_task,
+            });
+            session.authority = authority;
+            true
+        } else {
+            entry.cancel.cancel();
+            false
+        }
     }
 
     /// Record the authenticated pubkey for a connection after NIP-42 succeeds.
@@ -250,6 +362,81 @@ impl ConnectionManager {
             if let Ok(mut slot) = entry.authenticated_pubkey.write() {
                 *slot = Some(pubkey_bytes);
             }
+            if let Ok(mut slot) = entry.authenticated_owner_pubkey.write() {
+                *slot = None;
+            }
+            if let Ok(mut slot) = entry.verified_nostr_proof.write() {
+                *slot = None;
+            }
+            if let Ok(mut slot) = entry.verified_federated_assertion.write() {
+                *slot = None;
+            }
+            Self::clear_protected_session(&entry);
+        }
+    }
+
+    /// Record sealed connection evidence and derive actor/owner indexes from it.
+    pub fn set_authenticated_authority(
+        &self,
+        conn_id: Uuid,
+        proof: Arc<VerifiedNostrProof>,
+        assertion: Option<Arc<VerifiedFederatedAssertion>>,
+    ) {
+        if let Some(entry) = self.connections.get(&conn_id) {
+            if let Ok(mut slot) = entry.authenticated_pubkey.write() {
+                *slot = Some(proof.actor_pubkey().to_bytes().to_vec());
+            }
+            if let Ok(mut slot) = entry.authenticated_owner_pubkey.write() {
+                *slot = proof
+                    .verified_delegation()
+                    .map(|delegation| delegation.owner_pubkey().to_bytes().to_vec());
+            }
+            if let Ok(mut slot) = entry.verified_nostr_proof.write() {
+                *slot = Some(proof);
+            }
+            if let Ok(mut slot) = entry.verified_federated_assertion.write() {
+                *slot = assertion;
+            }
+            Self::clear_protected_session(&entry);
+        }
+    }
+
+    /// Retain the latest enforcing authority for an established connection.
+    /// Legacy and observational results clear any older authority without
+    /// creating a new invalidation registration.
+    pub fn retain_protected_session_authority(
+        &self,
+        conn_id: Uuid,
+        authority: &crate::authorization_runtime::transport::ProtectedAuthorization,
+    ) {
+        if let Some(entry) = self.connections.get(&conn_id) {
+            if authority.is_enforcing() {
+                let candidate = authority.expires_at().unwrap_or_default();
+                // Anchor monotonic time before consulting the injected
+                // whole-second authorization clock. Time spent sampling or
+                // installing the task must consume authority, never extend it.
+                let monotonic_anchor = tokio::time::Instant::now();
+                match authority.expiry_delay() {
+                    Ok(Some(delay)) => {
+                        if let Some(wake_at) =
+                            protected_session_wake_at(candidate, monotonic_anchor, delay)
+                        {
+                            Self::retain_earlier_protected_session(
+                                &entry,
+                                candidate,
+                                wake_at,
+                                Some(Arc::new(authority.clone())),
+                                None,
+                            );
+                        } else {
+                            entry.cancel.cancel();
+                        }
+                    }
+                    Ok(None) | Err(_) => entry.cancel.cancel(),
+                }
+            } else {
+                Self::clear_protected_session(&entry);
+            }
         }
     }
 
@@ -290,6 +477,37 @@ impl ConnectionManager {
             .and_then(|entry| entry.authenticated_pubkey.read().ok()?.clone())
     }
 
+    /// Return the sealed verifier evidence recorded for a connection.
+    pub fn authority_for_conn(&self, conn_id: Uuid) -> Option<Arc<VerifiedNostrProof>> {
+        self.connections
+            .get(&conn_id)
+            .and_then(|entry| entry.verified_nostr_proof.read().ok()?.clone())
+    }
+
+    /// Return current direct federated evidence recorded for a connection.
+    pub fn federated_assertion_for_conn(
+        &self,
+        conn_id: Uuid,
+    ) -> Option<Arc<VerifiedFederatedAssertion>> {
+        self.connections
+            .get(&conn_id)
+            .and_then(|entry| entry.verified_federated_assertion.read().ok()?.clone())
+    }
+
+    /// Return the server-owned cancellation token for a live connection.
+    pub fn cancellation_for_conn(&self, conn_id: Uuid) -> Option<CancellationToken> {
+        self.connections
+            .get(&conn_id)
+            .map(|entry| entry.cancel.clone())
+    }
+
+    /// Cancel one connection after protected session authority expires.
+    pub fn cancel_connection(&self, conn_id: Uuid) {
+        if let Some(entry) = self.connections.get(&conn_id) {
+            entry.cancel.cancel();
+        }
+    }
+
     /// Disconnect every live connection authenticated as `pubkey` **in
     /// `community`**, delivering a final `OK false` frame carrying `reason`
     /// before closing.
@@ -435,7 +653,7 @@ impl ConnectionManager {
     /// On sustained backpressure (>grace_limit consecutive full buffers),
     /// cancels the connection. Transient stalls get a warning only.
     pub fn send_to(&self, conn_id: Uuid, msg: String) -> bool {
-        self.try_send_ws_message(conn_id, WsMessage::Text(msg.into()))
+        self.try_send_outbound(conn_id, OutboundData::plain(WsMessage::Text(msg.into())))
     }
 
     /// Sends an already-serialized UTF-8 text payload to the given connection.
@@ -445,10 +663,73 @@ impl ConnectionManager {
     pub fn send_to_text_bytes(&self, conn_id: Uuid, msg: Arc<Bytes>) -> bool {
         let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
             .expect("relay fan-out frames are serialized UTF-8 JSON");
-        self.try_send_ws_message(conn_id, WsMessage::Text(text))
+        self.try_send_outbound(conn_id, OutboundData::plain(WsMessage::Text(text)))
     }
 
-    fn try_send_ws_message(&self, conn_id: Uuid, msg: WsMessage) -> bool {
+    /// Queue protected text and retain its exact authority until socket drain.
+    pub fn send_to_text_bytes_protected(
+        &self,
+        conn_id: Uuid,
+        msg: Arc<Bytes>,
+        authority: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+    ) -> bool {
+        let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
+            .expect("relay fan-out frames are serialized UTF-8 JSON");
+        self.try_send_outbound(
+            conn_id,
+            OutboundData::protected(WsMessage::Text(text), authority),
+        )
+    }
+
+    /// Queue output guarded by both the emitting operation and the recipient's
+    /// current read authority until socket drain.
+    pub fn send_to_text_bytes_protected_pair(
+        &self,
+        conn_id: Uuid,
+        msg: Arc<Bytes>,
+        sender: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+        recipient: Arc<crate::authorization_runtime::transport::ProtectedAuthorization>,
+    ) -> bool {
+        let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
+            .expect("relay fan-out frames are serialized UTF-8 JSON");
+        self.try_send_outbound(
+            conn_id,
+            OutboundData::protected_pair(WsMessage::Text(text), sender, recipient),
+        )
+    }
+
+    /// Queue output behind an arbitrary asynchronous sender fence.
+    pub(crate) fn send_to_text_bytes_guarded(
+        &self,
+        conn_id: Uuid,
+        msg: Arc<Bytes>,
+        authority: Arc<dyn crate::connection::QueuedOutboundReleaseFence>,
+    ) -> bool {
+        let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
+            .expect("relay fan-out frames are serialized UTF-8 JSON");
+        self.try_send_outbound(
+            conn_id,
+            OutboundData::guarded(WsMessage::Text(text), authority),
+        )
+    }
+
+    /// Queue output behind a remote sender fence and local recipient fence.
+    pub(crate) fn send_to_text_bytes_guarded_pair(
+        &self,
+        conn_id: Uuid,
+        msg: Arc<Bytes>,
+        sender: Arc<dyn crate::connection::QueuedOutboundReleaseFence>,
+        recipient: Arc<dyn crate::connection::QueuedOutboundReleaseFence>,
+    ) -> bool {
+        let text = WsUtf8Bytes::try_from(Bytes::clone(msg.as_ref()))
+            .expect("relay fan-out frames are serialized UTF-8 JSON");
+        self.try_send_outbound(
+            conn_id,
+            OutboundData::guarded_pair(WsMessage::Text(text), sender, recipient),
+        )
+    }
+
+    fn try_send_outbound(&self, conn_id: Uuid, msg: OutboundData) -> bool {
         if let Some(entry) = self.connections.get(&conn_id) {
             let conn = entry.value();
             match conn.tx.try_send(msg) {
@@ -628,6 +909,36 @@ pub struct AppState {
     /// byte-identically to a relay without the mesh. Access via
     /// [`AppState::mesh`].
     pub mesh: Arc<std::sync::OnceLock<crate::mesh_boot::MeshHandle>>,
+    /// Optional exact-domain protected-transport runtime.
+    ///
+    /// Unset preserves legacy behavior. Once installed it is immutable for the
+    /// process lifetime so request data cannot switch provider policy.
+    pub protected_transport: Arc<
+        std::sync::OnceLock<
+            Arc<crate::authorization_runtime::transport::ProtectedTransportRuntime>,
+        >,
+    >,
+    /// Deployment-verified ingress provenance for direct identity assertions.
+    ///
+    /// Unset by the stock binary. Header presence alone is never trusted.
+    pub identity_assertion_provenance: Arc<
+        std::sync::OnceLock<
+            Arc<dyn crate::corporate_identity::IdentityAssertionProvenanceVerifier>,
+        >,
+    >,
+    /// Complete-stack-gated NIP-FI discovery. The stock binary leaves this
+    /// unset; only an immutable reviewed conformance input can construct it.
+    pub nip_fi_discovery: Arc<std::sync::OnceLock<crate::nip11::ConformanceReadyNipFiDiscovery>>,
+    /// Optional RFC-gated, dedicated client-status presentation runtime.
+    pub client_status_runtime: Arc<
+        std::sync::OnceLock<
+            Arc<crate::authorization_runtime::status::ProductionClientStatusRuntime>,
+        >,
+    >,
+    /// Independent version witness installed with protected authorization.
+    pub restore_protection: Arc<
+        std::sync::OnceLock<Arc<crate::authorization_runtime::restore::RestoreProtectionRuntime>>,
+    >,
 }
 
 impl AppState {
@@ -804,6 +1115,11 @@ impl AppState {
             // `crates/buzz-test-client` once those land).
             tracer: Arc::new(crate::conformance::NoopTracer),
             mesh: Arc::new(std::sync::OnceLock::new()),
+            protected_transport: Arc::new(std::sync::OnceLock::new()),
+            identity_assertion_provenance: Arc::new(std::sync::OnceLock::new()),
+            nip_fi_discovery: Arc::new(std::sync::OnceLock::new()),
+            client_status_runtime: Arc::new(std::sync::OnceLock::new()),
+            restore_protection: Arc::new(std::sync::OnceLock::new()),
         };
         (
             state,
@@ -820,6 +1136,99 @@ impl AppState {
         self.mesh.get()
     }
 
+    /// Install immutable protected-transport policy exactly once.
+    pub fn install_protected_transport(
+        &self,
+        runtime: Arc<crate::authorization_runtime::transport::ProtectedTransportRuntime>,
+    ) -> Result<(), Arc<crate::authorization_runtime::transport::ProtectedTransportRuntime>> {
+        self.protected_transport.set(runtime)
+    }
+
+    /// Current protected-transport runtime, if configured.
+    pub fn protected_transport(
+        &self,
+    ) -> Option<&Arc<crate::authorization_runtime::transport::ProtectedTransportRuntime>> {
+        self.protected_transport.get()
+    }
+
+    /// Install the immutable deployment-owned ingress provenance adapter.
+    pub fn install_identity_assertion_provenance(
+        &self,
+        verifier: Arc<dyn crate::corporate_identity::IdentityAssertionProvenanceVerifier>,
+    ) -> Result<(), Arc<dyn crate::corporate_identity::IdentityAssertionProvenanceVerifier>> {
+        self.identity_assertion_provenance.set(verifier)
+    }
+
+    /// Return the installed ingress provenance adapter, if any.
+    pub fn identity_assertion_provenance(
+        &self,
+    ) -> Option<&Arc<dyn crate::corporate_identity::IdentityAssertionProvenanceVerifier>> {
+        self.identity_assertion_provenance.get()
+    }
+
+    /// Install exact-revision NIP-FI discovery readiness once.
+    pub fn install_nip_fi_discovery(
+        &self,
+        ready: crate::nip11::ConformanceReadyNipFiDiscovery,
+    ) -> Result<(), crate::nip11::ConformanceReadyNipFiDiscovery> {
+        self.nip_fi_discovery.set(ready)
+    }
+
+    /// Return complete-stack-gated NIP-FI discovery, if installed.
+    pub fn nip_fi_discovery(&self) -> Option<&crate::nip11::ConformanceReadyNipFiDiscovery> {
+        self.nip_fi_discovery.get()
+    }
+
+    /// Install an externally approved dedicated client-status runtime once.
+    pub fn install_client_status_runtime(
+        &self,
+        runtime: Arc<crate::authorization_runtime::status::ProductionClientStatusRuntime>,
+    ) -> Result<(), Arc<crate::authorization_runtime::status::ProductionClientStatusRuntime>> {
+        self.client_status_runtime.set(runtime)
+    }
+
+    /// Return the approved client-status runtime, if installed.
+    pub fn client_status_runtime(
+        &self,
+    ) -> Option<&Arc<crate::authorization_runtime::status::ProductionClientStatusRuntime>> {
+        self.client_status_runtime.get()
+    }
+
+    /// Install the restore-independent version witness exactly once.
+    pub fn install_restore_protection(
+        &self,
+        runtime: Arc<crate::authorization_runtime::restore::RestoreProtectionRuntime>,
+    ) -> Result<(), Arc<crate::authorization_runtime::restore::RestoreProtectionRuntime>> {
+        self.restore_protection.set(runtime)
+    }
+
+    /// Current restore-independent version witness, if protected mode exists.
+    pub fn restore_protection(
+        &self,
+    ) -> Option<&Arc<crate::authorization_runtime::restore::RestoreProtectionRuntime>> {
+        self.restore_protection.get()
+    }
+
+    /// Database UUIDs for exact domains where autonomous effects must not run.
+    pub fn enforcing_protected_domain_ids(&self) -> Vec<uuid::Uuid> {
+        self.protected_transport()
+            .map(|runtime| {
+                runtime
+                    .enforcing_domains()
+                    .into_iter()
+                    .map(|domain| *domain.as_uuid())
+                    .collect()
+            })
+            .unwrap_or_default()
+    }
+
+    /// Whether one exact server-resolved domain is in authoritative Enforce.
+    pub fn is_protected_enforcing(&self, domain: CommunityId) -> bool {
+        self.protected_transport()
+            .and_then(|runtime| runtime.mode_for_domain(domain))
+            == Some(crate::authorization_runtime::finalization::AuthorizationMode::Enforce)
+    }
+
     /// Record an event ID as locally-published for dedup, scoped to the
     /// community it was fanned out in. Called before Redis publish so the
     /// multi-node consumer can skip the echo for *this* community only — a
@@ -1237,7 +1646,7 @@ mod tests {
     ) -> (
         ConnectionManager,
         Uuid,
-        mpsc::Receiver<WsMessage>,
+        mpsc::Receiver<OutboundData>,
         mpsc::Receiver<WsMessage>,
         CancellationToken,
         Arc<AtomicU8>,
@@ -1261,6 +1670,228 @@ mod tests {
         (mgr, conn_id, rx, ctrl_rx, cancel, bp)
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn protected_session_deadline_is_conservative_and_anchor_bound() {
+        let anchor = tokio::time::Instant::now();
+        let wake_at = protected_session_wake_at_from_samples(
+            102,
+            anchor,
+            std::time::Duration::from_millis(100_900),
+            std::time::Duration::from_secs(2),
+        )
+        .expect("future deadline");
+        assert_eq!(
+            wake_at.duration_since(anchor),
+            std::time::Duration::from_secs(1),
+            "whole-second authority is shortened conservatively instead of rounded late"
+        );
+
+        // Simulate work between authority-clock sampling and task install. The
+        // absolute monotonic wake stays tied to the earlier anchor.
+        tokio::time::advance(std::time::Duration::from_millis(750)).await;
+        let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8);
+        let entry = mgr
+            .connections
+            .get(&conn_id)
+            .expect("registered connection");
+        ConnectionManager::retain_earlier_protected_session(&entry, 102, wake_at, None, None);
+        drop(entry);
+
+        tokio::time::advance(std::time::Duration::from_millis(249)).await;
+        tokio::task::yield_now().await;
+        assert!(!cancel.is_cancelled());
+        tokio::time::advance(std::time::Duration::from_millis(1)).await;
+        tokio::task::yield_now().await;
+        assert!(cancel.is_cancelled());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn protected_session_expiry_never_extends_and_disconnect_cleans_task() {
+        let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8);
+        {
+            let entry = mgr
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                10,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(10),
+                None,
+                None,
+            );
+        }
+        tokio::task::yield_now().await;
+        tokio::time::advance(std::time::Duration::from_secs(9)).await;
+        assert!(!cancel.is_cancelled());
+
+        assert!(!should_replace_protected_session_deadline(Some(10), 20));
+        assert!(!should_replace_protected_session_deadline(Some(10), 10));
+        tokio::time::advance(std::time::Duration::from_secs(1)).await;
+        tokio::task::yield_now().await;
+        assert!(
+            cancel.is_cancelled(),
+            "ordinary traffic cannot extend the first hard session deadline"
+        );
+
+        let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8);
+        {
+            let entry = mgr
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                20,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(20),
+                None,
+                None,
+            );
+        }
+        tokio::task::yield_now().await;
+        tokio::time::advance(std::time::Duration::from_secs(5)).await;
+        assert!(should_replace_protected_session_deadline(Some(20), 10));
+        {
+            let entry = mgr
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                10,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(5),
+                None,
+                None,
+            );
+        }
+        tokio::task::yield_now().await;
+        tokio::time::advance(std::time::Duration::from_secs(5)).await;
+        tokio::task::yield_now().await;
+        assert!(
+            cancel.is_cancelled(),
+            "a shorter authority tightens the hard deadline"
+        );
+
+        let (mgr, conn_id, _rx, _ctrl_rx, cancel, _bp) = setup_conn(8);
+        {
+            let entry = mgr
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                5,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(5),
+                None,
+                None,
+            );
+        }
+        tokio::task::yield_now().await;
+        mgr.deregister(conn_id);
+        tokio::time::advance(std::time::Duration::from_secs(5)).await;
+        assert!(
+            !cancel.is_cancelled(),
+            "disconnect removes the obsolete expiry task"
+        );
+    }
+
+    #[tokio::test]
+    async fn concurrent_protected_operations_cannot_replace_an_earlier_deadline() {
+        let (mgr, conn_id, _rx, _ctrl_rx, _cancel, _bp) = setup_conn(8);
+        let mgr = Arc::new(mgr);
+        let entry = mgr
+            .connections
+            .get(&conn_id)
+            .expect("registered connection");
+        let protected_session = Arc::clone(&entry.protected_session);
+        drop(entry);
+
+        let runtime = tokio::runtime::Handle::current();
+        let short_authority =
+            Arc::new(crate::authorization_runtime::transport::ProtectedAuthorization::Legacy);
+        let long_authority =
+            Arc::new(crate::authorization_runtime::transport::ProtectedAuthorization::Legacy);
+        let (short_locked, wait_for_short_lock) = std::sync::mpsc::channel();
+        let (release_short, wait_for_release) = std::sync::mpsc::channel();
+        let short_manager = Arc::clone(&mgr);
+        let retained_short_authority = Arc::clone(&short_authority);
+        let short_runtime = runtime.clone();
+        let short = std::thread::spawn(move || {
+            let _runtime = short_runtime.enter();
+            let entry = short_manager
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            let hook = || {
+                short_locked.send(()).expect("test still waiting");
+                wait_for_release.recv().expect("test releases first lock");
+            };
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                10,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(10),
+                Some(retained_short_authority),
+                Some(&hook),
+            );
+        });
+        wait_for_short_lock
+            .recv_timeout(std::time::Duration::from_secs(1))
+            .expect("short contender holds the comparison/install lock");
+
+        let (long_attempting, wait_for_long_attempt) = std::sync::mpsc::channel();
+        let (long_read, wait_for_long_read) = std::sync::mpsc::channel();
+        let long_manager = Arc::clone(&mgr);
+        let long_runtime = runtime.clone();
+        let long = std::thread::spawn(move || {
+            let _runtime = long_runtime.enter();
+            long_attempting.send(()).expect("test still waiting");
+            let entry = long_manager
+                .connections
+                .get(&conn_id)
+                .expect("registered connection");
+            let hook = || {
+                long_read.send(()).expect("test still waiting");
+            };
+            ConnectionManager::retain_earlier_protected_session(
+                &entry,
+                20,
+                tokio::time::Instant::now() + std::time::Duration::from_secs(20),
+                Some(long_authority),
+                Some(&hook),
+            );
+        });
+        wait_for_long_attempt
+            .recv_timeout(std::time::Duration::from_secs(1))
+            .expect("long contender reached the atomic helper");
+        assert!(
+            matches!(
+                wait_for_long_read.recv_timeout(std::time::Duration::from_millis(50)),
+                Err(std::sync::mpsc::RecvTimeoutError::Timeout)
+            ),
+            "a second contender cannot read the deadline before the first install completes"
+        );
+        release_short.send(()).expect("short contender is waiting");
+        short.join().expect("short retention thread");
+        wait_for_long_read
+            .recv_timeout(std::time::Duration::from_secs(1))
+            .expect("long contender reads only after the short install");
+        long.join().expect("long retention thread");
+
+        let session = protected_session.lock().expect("protected session lock");
+        assert_eq!(
+            session.expiry.as_ref().map(|task| task.deadline),
+            Some(10),
+            "the minimum concurrent deadline is retained regardless of completion order"
+        );
+        assert!(
+            session
+                .authority
+                .as_ref()
+                .is_some_and(|authority| Arc::ptr_eq(authority, &short_authority)),
+            "the retained authority belongs to the minimum-deadline contender"
+        );
+    }
+
     async fn test_state() -> Arc<AppState> {
         let mut config = crate::config::Config::from_env().expect("default config loads");
         config.require_relay_membership = false;
@@ -1363,7 +1994,7 @@ mod tests {
                 "test.local".to_string(),
             ),
             remote_addr: "127.0.0.1:1234".parse().unwrap(),
-            corporate_identity_jwt: None,
+            corporate_identity_assertion: None,
             auth_state: RwLock::new(AuthState::Failed),
             subscriptions: Arc::new(Mutex::new(HashMap::new())),
             send_tx: tx.clone(),