WEB Manager Session + Generation Fence tests

src/api/config_store/atomic.rs

@@ -11,6 +11,8 @@ use nix::fcntl::{Flock, FlockArg, OFlag, openat, renameat};
 use nix::sys::stat::Mode;
 #[cfg(unix)]
 use nix::unistd::{UnlinkatFlags, fsync, unlinkat};
+#[cfg(unix)]
+use tracing::warn;
 
 use super::compute_source_revision;
 use crate::api::model::ApiFailure;
@@ -75,7 +77,7 @@ impl ConfigWriteLock {
     }
 }
 
-/// Replaces one config source through a durable same-directory rename.
+/// Replaces one config source through a same-directory rename after syncing file data.
 pub(in crate::api) async fn write_atomic(
     path: PathBuf,
     contents: String,
@@ -362,7 +364,15 @@ fn write_and_publish(
         anchored.name(),
     )
     .map_err(errno_to_io)?;
-    fsync(anchored.parent()).map_err(errno_to_io)?;
+    // Rename is the commit boundary. A later directory-sync error cannot be reported as an
+    // uncommitted mutation because mandatory in-process publication must still run.
+    if let Err(error) = fsync(anchored.parent()).map_err(errno_to_io) {
+        warn!(
+            path = %path.display(),
+            %error,
+            "Config rename committed but directory durability sync failed"
+        );
+    }
     Ok(committed_revision)
 }
 

src/api/config_store/tests.rs

@@ -346,13 +346,13 @@ async fn config_sidecar_lock_serializes_competing_revision_writers() {
     let second = second.await.unwrap();
 
     assert_ne!(first.is_ok(), second.is_ok());
-    let conflict = if let Err(error) = first {
-        error
-    } else {
-        second.unwrap_err()
+    let (winner_revision, conflict) = match (first, second) {
+        (Ok(revision), Err(error)) | (Err(error), Ok(revision)) => (revision, error),
+        _ => unreachable!("exactly one cooperative writer must commit"),
     };
     assert_eq!(conflict.code, "revision_conflict");
-    let persisted = tokio::fs::read_to_string(path).await.unwrap();
+    assert_eq!(winner_revision, current_revision(&path).await.unwrap());
+    let persisted = tokio::fs::read_to_string(&path).await.unwrap();
     assert!(persisted.contains("first.example") || persisted.contains("second.example"));
 }
 

src/proxy/traffic_limiter/tests.rs

@@ -281,3 +281,47 @@ fn active_lease_observes_policy_removal() {
 
     assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 1);
 }
+
+#[test]
+fn active_lease_observes_policy_addition() {
+    let limiter = TrafficLimiter::new();
+    let lease = limiter
+        .acquire_lease("alice", "203.0.113.7".parse().unwrap())
+        .unwrap();
+    assert_eq!(lease.try_consume(RateDirection::Up, 2).granted, 2);
+
+    let mut user_limits = HashMap::new();
+    user_limits.insert("alice".to_string(), rate(1, 0));
+    limiter.apply_policy(user_limits, HashMap::new());
+
+    assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 1);
+    assert_eq!(lease.try_consume(RateDirection::Up, 1).granted, 0);
+}
+
+#[test]
+fn reservation_refund_stays_with_retired_binding() {
+    let limiter = TrafficLimiter::new();
+    let mut user_limits = HashMap::new();
+    user_limits.insert("alice".to_string(), rate(400_000, 400_000));
+    limiter.apply_policy(user_limits, HashMap::new());
+    let lease = limiter
+        .acquire_lease("alice", "203.0.113.7".parse().unwrap())
+        .unwrap();
+
+    let reservation = lease.try_reserve(RateDirection::Down, 800);
+    let old_bucket = Arc::clone(
+        reservation
+            ._binding
+            .user_bucket
+            .as_ref()
+            .expect("the original policy must bind a user bucket"),
+    );
+    let epoch = reservation.user.as_ref().unwrap().epoch;
+    limiter.apply_policy(HashMap::new(), HashMap::new());
+    assert_eq!(lease.try_consume(RateDirection::Down, 1).granted, 1);
+    assert!(lease.binding.load().user_bucket.is_none());
+
+    drop(reservation);
+
+    assert_eq!(old_bucket.down.used_at(epoch), Some(0));
+}

src/proxy/user_admission.rs

@@ -1,8 +1,8 @@
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicU8, Ordering};
+use std::sync::atomic::AtomicU8;
 
-use parking_lot::{Mutex, MutexGuard};
+use parking_lot::Mutex;
 use tokio_util::sync::CancellationToken;
 
 use crate::crypto::sha256;
@@ -12,6 +12,10 @@ const REGISTRATION_PENDING: u8 = 0;
 const REGISTRATION_ACTIVE: u8 = 1;
 const REGISTRATION_DROPPED: u8 = 2;
 
+// Authenticated-owner publication and RAII deregistration.
+mod registration;
+pub(crate) use registration::{UserAdmissionPublication, UserSessionRegistration};
+
 /// Stable secret identity used to fence authentication across runtime generations.
 pub(crate) type UserCredentialId = [u8; 16];
 
@@ -501,96 +505,6 @@ impl UserAdmissionAuthority {
     }
 }
 
-/// Authority lock retained until the caller publishes its owned object.
-pub(crate) struct UserAdmissionPublication<'a> {
-    state: MutexGuard<'a, UserAdmissionState>,
-    authority: Arc<UserAdmissionAuthority>,
-    user: String,
-    registration_id: u64,
-    incarnation: UserIncarnation,
-    token: CancellationToken,
-    active: Arc<AtomicU8>,
-    registration_taken: bool,
-}
-
-impl UserAdmissionPublication<'_> {
-    /// Moves the registered owner out while retaining the authority lock.
-    pub(crate) fn take_registration(&mut self) -> Option<UserSessionRegistration> {
-        if self.registration_taken {
-            return None;
-        }
-        self.registration_taken = true;
-        Some(UserSessionRegistration {
-            authority: Arc::clone(&self.authority),
-            user: self.user.clone(),
-            registration_id: self.registration_id,
-            incarnation: self.incarnation,
-            token: self.token.clone(),
-            active: Arc::clone(&self.active),
-        })
-    }
-
-    /// Commits the owner record after the caller publishes its lifecycle object.
-    pub(crate) fn commit(mut self) {
-        if !self.registration_taken {
-            return;
-        }
-        if self.active.compare_exchange(
-            REGISTRATION_PENDING, REGISTRATION_ACTIVE, Ordering::AcqRel, Ordering::Acquire,
-        ).is_err() {
-            return;
-        }
-        self.state
-            .owners_by_user
-            .entry(self.user.clone())
-            .or_default()
-            .insert(
-                self.registration_id,
-                RegisteredOwner {
-                    token: self.token.clone(),
-                    incarnation: self.incarnation,
-                },
-            );
-    }
-}
-
-/// RAII ownership registered against one user incarnation.
-#[must_use = "registered user ownership must be retained until lifecycle completion"]
-pub(crate) struct UserSessionRegistration {
-    authority: Arc<UserAdmissionAuthority>,
-    user: String,
-    registration_id: u64,
-    incarnation: UserIncarnation,
-    token: CancellationToken,
-    active: Arc<AtomicU8>,
-}
-
-impl UserSessionRegistration {
-    /// Returns the cancellation signal for revocation or credential replacement.
-    pub(crate) fn token(&self) -> CancellationToken {
-        self.token.clone()
-    }
-
-    /// Returns the immutable user incarnation owned by this registration.
-    pub(crate) fn incarnation(&self) -> UserIncarnation {
-        self.incarnation
-    }
-
-    /// Returns whether revocation has cancelled this ownership.
-    pub(crate) fn is_cancelled(&self) -> bool {
-        self.token.is_cancelled()
-    }
-}
-
-impl Drop for UserSessionRegistration {
-    fn drop(&mut self) {
-        if self.active.swap(REGISTRATION_DROPPED, Ordering::AcqRel) == REGISTRATION_ACTIVE {
-            self.authority
-                .unregister(&self.user, self.registration_id, self.incarnation);
-        }
-    }
-}
-
 /// Derives the stable credential identity from one decoded MTProxy secret.
 pub(crate) fn credential_id(secret: &[u8; 16]) -> UserCredentialId {
     let digest = sha256(secret);

src/proxy/user_admission/registration.rs

@@ -0,0 +1,112 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU8, Ordering};
+
+use parking_lot::MutexGuard;
+use tokio_util::sync::CancellationToken;
+
+use super::*;
+
+/// Authority lock retained until the caller publishes its owned object.
+pub(crate) struct UserAdmissionPublication<'a> {
+    /// Locked authority state that linearizes publication with policy mutation.
+    pub(super) state: MutexGuard<'a, UserAdmissionState>,
+    /// Process authority used by the eventual registration guard.
+    pub(super) authority: Arc<UserAdmissionAuthority>,
+    /// Username owning the published lifecycle object.
+    pub(super) user: String,
+    /// Unique registration identity within the process authority.
+    pub(super) registration_id: u64,
+    /// User incarnation authenticated by this publication.
+    pub(super) incarnation: UserIncarnation,
+    /// Revocation signal shared with the lifecycle owner.
+    pub(super) token: CancellationToken,
+    /// Atomic publication state shared with the registration guard.
+    pub(super) active: Arc<AtomicU8>,
+    /// Whether ownership has already moved into a registration guard.
+    pub(super) registration_taken: bool,
+}
+
+impl UserAdmissionPublication<'_> {
+    /// Moves the registered owner out while retaining the authority lock.
+    pub(crate) fn take_registration(&mut self) -> Option<UserSessionRegistration> {
+        if self.registration_taken {
+            return None;
+        }
+        self.registration_taken = true;
+        Some(UserSessionRegistration {
+            authority: Arc::clone(&self.authority),
+            user: self.user.clone(),
+            registration_id: self.registration_id,
+            incarnation: self.incarnation,
+            token: self.token.clone(),
+            active: Arc::clone(&self.active),
+        })
+    }
+
+    /// Commits the owner record after the caller publishes its lifecycle object.
+    pub(crate) fn commit(mut self) {
+        if !self.registration_taken {
+            return;
+        }
+        if self
+            .active
+            .compare_exchange(
+                REGISTRATION_PENDING,
+                REGISTRATION_ACTIVE,
+                Ordering::AcqRel,
+                Ordering::Acquire,
+            )
+            .is_err()
+        {
+            return;
+        }
+        self.state
+            .owners_by_user
+            .entry(self.user.clone())
+            .or_default()
+            .insert(
+                self.registration_id,
+                RegisteredOwner {
+                    token: self.token.clone(),
+                    incarnation: self.incarnation,
+                },
+            );
+    }
+}
+
+/// RAII ownership registered against one user incarnation.
+#[must_use = "registered user ownership must be retained until lifecycle completion"]
+pub(crate) struct UserSessionRegistration {
+    authority: Arc<UserAdmissionAuthority>,
+    user: String,
+    registration_id: u64,
+    incarnation: UserIncarnation,
+    token: CancellationToken,
+    active: Arc<AtomicU8>,
+}
+
+impl UserSessionRegistration {
+    /// Returns the cancellation signal for revocation or credential replacement.
+    pub(crate) fn token(&self) -> CancellationToken {
+        self.token.clone()
+    }
+
+    /// Returns the immutable user incarnation owned by this registration.
+    pub(crate) fn incarnation(&self) -> UserIncarnation {
+        self.incarnation
+    }
+
+    /// Returns whether revocation has cancelled this ownership.
+    pub(crate) fn is_cancelled(&self) -> bool {
+        self.token.is_cancelled()
+    }
+}
+
+impl Drop for UserSessionRegistration {
+    fn drop(&mut self) {
+        if self.active.swap(REGISTRATION_DROPPED, Ordering::AcqRel) == REGISTRATION_ACTIVE {
+            self.authority
+                .unregister(&self.user, self.registration_id, self.incarnation);
+        }
+    }
+}

src/proxy/user_admission/tests.rs

@@ -60,6 +60,19 @@ fn stale_candidate_cannot_overwrite_newer_mutation() {
             .is_none()
     );
     assert!(!authority.is_user_enabled("alice"));
+
+    let disabled = HashMap::from([("alice".to_string(), false)]);
+    assert!(
+        authority
+            .apply_config_from_source(1, &users(secret), &disabled)
+            .is_none()
+    );
+    assert!(
+        authority
+            .apply_config_from_source(2, &users(secret), &disabled)
+            .is_some()
+    );
+    assert!(!authority.is_user_enabled("alice"));
 }
 
 #[test]

src/stats/replay.rs

@@ -90,6 +90,13 @@ struct ReplayShard {
     capacity: usize,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum ReplayClaimResult {
+    Claimed,
+    Duplicate,
+    Capacity,
+}
+
 impl ReplayShard {
     fn new(cap: NonZeroUsize) -> Self {
         Self {
@@ -169,22 +176,22 @@ impl ReplayShard {
         now: Instant,
         window: Duration,
         token: u64,
-    ) -> bool {
+    ) -> ReplayClaimResult {
         if window.is_zero() {
-            return true;
+            return ReplayClaimResult::Claimed;
         }
         self.cleanup(now, window);
         if self.cache.peek(key.as_slice()).is_some() || self.pending.contains_key(key.as_slice()) {
-            return false;
+            return ReplayClaimResult::Duplicate;
         }
         while self.cache.len().saturating_add(self.pending.len()) >= self.capacity {
             if self.queue.is_empty() {
-                return false;
+                return ReplayClaimResult::Capacity;
             }
             self.evict_queue_front();
         }
         self.pending.insert(key, token);
-        true
+        ReplayClaimResult::Claimed
     }
 
     fn remove_pending(&mut self, key: &[u8], token: u64) -> bool {
@@ -377,10 +384,17 @@ impl ReplayChecker {
         }
         let token = self.reserve_claim_token()?;
         let mut shard = self.tls_shards[shard_idx].lock();
-        if !shard.claim_owned(key.clone(), Instant::now(), self.tls_window, token) {
+        match shard.claim_owned(key.clone(), Instant::now(), self.tls_window, token) {
+            ReplayClaimResult::Claimed => {}
+            ReplayClaimResult::Duplicate => {
                 self.hits.fetch_add(1, Ordering::Relaxed);
                 return None;
             }
+            ReplayClaimResult::Capacity => {
+                self.capacity_rejections.fetch_add(1, Ordering::Relaxed);
+                return None;
+            }
+        }
         drop(shard);
         Some(TlsReplayClaim {
             checker: self,
@@ -478,6 +492,7 @@ pub struct ReplayStats {
     pub total_checks: u64,
     pub total_hits: u64,
     pub total_additions: u64,
+    /// Claims rejected because committed and pending entries exhausted a shard.
     pub total_capacity_rejections: u64,
     pub total_cleanups: u64,
     pub num_shards: usize,
@@ -503,22 +518,5 @@ impl ReplayStats {
 }
 
 #[cfg(test)]
-mod capacity_tests {
-    use super::*;
-
-    #[test]
-    fn committed_and_pending_entries_share_one_shard_capacity() {
-        let capacity = NonZeroUsize::new(2).unwrap();
-        let mut shard = ReplayShard::new(capacity);
-        let now = Instant::now();
-        let window = Duration::from_secs(60);
-        assert!(shard.claim_owned(ReplayKey::from_slice(b"pending-a"), now, window, 1));
-        assert!(shard.claim_owned(ReplayKey::from_slice(b"pending-b"), now, window, 2));
-
-        shard.add_owned(ReplayKey::from_slice(b"committed"), now, window);
-
-        assert!(shard.len() <= capacity.get());
-        assert!(shard.pending.contains_key(b"pending-a".as_slice()));
-        assert!(shard.pending.contains_key(b"pending-b".as_slice()));
-    }
-}
+#[path = "replay/tests.rs"]
+mod capacity_tests;

src/stats/replay/tests.rs

@@ -0,0 +1,47 @@
+use super::*;
+
+#[test]
+fn committed_and_pending_entries_share_one_shard_capacity() {
+    let capacity = NonZeroUsize::new(2).unwrap();
+    let mut shard = ReplayShard::new(capacity);
+    let now = Instant::now();
+    let window = Duration::from_secs(60);
+    assert_eq!(
+        shard.claim_owned(ReplayKey::from_slice(b"pending-a"), now, window, 1),
+        ReplayClaimResult::Claimed
+    );
+    assert_eq!(
+        shard.claim_owned(ReplayKey::from_slice(b"pending-b"), now, window, 2),
+        ReplayClaimResult::Claimed
+    );
+
+    assert!(!shard.add_owned(ReplayKey::from_slice(b"committed"), now, window));
+
+    assert_eq!(shard.len(), capacity.get());
+    assert!(shard.pending.contains_key(b"pending-a".as_slice()));
+    assert!(shard.pending.contains_key(b"pending-b".as_slice()));
+}
+
+#[test]
+fn pending_capacity_rejection_is_not_reported_as_a_replay_hit() {
+    let checker = ReplayChecker::new(64, Duration::from_secs(60));
+    let first = checker
+        .claim_tls_digest(b"same-shard-key")
+        .expect("first claim must reserve the shard");
+    let shard_idx = checker.get_shard_idx(b"same-shard-key");
+    let rejected = (0..10_000u64)
+        .map(u64::to_le_bytes)
+        .find(|candidate| checker.get_shard_idx(candidate) == shard_idx)
+        .expect("bounded search must find a second key in the selected shard");
+    let before = checker.stats();
+
+    assert!(checker.claim_tls_digest(&rejected).is_none());
+
+    let after = checker.stats();
+    assert_eq!(after.total_hits, before.total_hits);
+    assert_eq!(
+        after.total_capacity_rejections,
+        before.total_capacity_rejections + 1
+    );
+    drop(first);
+}

src/transport/middle_proxy/pool_reinit.rs

@@ -62,15 +62,22 @@ struct ReinitCommitOutcome {
 }
 
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
-struct DcFamilyGroup {
-    dc: i32,
-    family: IpFamily,
+/// One independently enforced hardswap writer-floor group.
+pub(in crate::transport::middle_proxy) struct DcFamilyGroup {
+    /// Telegram DC owning the group.
+    pub(in crate::transport::middle_proxy) dc: i32,
+    /// Address family whose floor is evaluated independently.
+    pub(in crate::transport::middle_proxy) family: IpFamily,
 }
 
-struct HardswapCoverage {
-    ratio: f32,
-    missing_groups: Vec<DcFamilyGroup>,
-    writer_deficit: usize,
+/// Complete floor-coverage result for one candidate hardswap generation.
+pub(in crate::transport::middle_proxy) struct HardswapCoverage {
+    /// Fraction of configured DC-family groups that reached their full floor.
+    pub(in crate::transport::middle_proxy) ratio: f32,
+    /// Stable list of DC-family groups that remain below their floor.
+    pub(in crate::transport::middle_proxy) missing_groups: Vec<DcFamilyGroup>,
+    /// Total writer count still required across all missing groups.
+    pub(in crate::transport::middle_proxy) writer_deficit: usize,
 }
 
 #[derive(Debug)]

src/transport/middle_proxy/pool_reinit/coordination.rs

@@ -287,7 +287,7 @@ impl MePool {
     }
 
     /// Evaluates full writer-floor coverage independently for every DC and address family.
-    pub(super) fn hardswap_coverage(
+    pub(in crate::transport::middle_proxy) fn hardswap_coverage(
         &self,
         desired_by_dc: &HashMap<i32, HashSet<SocketAddr>>,
         writer_addrs: &[(i32, SocketAddr)],

src/transport/middle_proxy/pool_reinit/tests.rs

@@ -537,7 +537,12 @@ async fn generation_role_reconciliation_promotes_active_warm_and_drains_orphans(
     assert!(orphan_warm.draining.load(Ordering::Acquire));
     assert!(!orphan_warm.allow_drain_fallback.load(Ordering::Acquire));
     let snapshot = pool.api_hardswap_snapshot().await;
+    let desired = pool.desired_dc_endpoints().await;
+    let coverage = pool.hardswap_coverage(&desired, &[(1, endpoint)]);
     assert_eq!(snapshot.orphan_warm_writers_current, 0);
-    assert_eq!(snapshot.pending_writer_deficit, 2);
-    assert_eq!(snapshot.pending_missing_dc_groups, 1);
+    assert_eq!(snapshot.pending_writer_deficit, coverage.writer_deficit);
+    assert_eq!(
+        snapshot.pending_missing_dc_groups,
+        coverage.missing_groups.len()
+    );
 }

src/transport/middle_proxy/pool_status.rs

@@ -6,7 +6,6 @@ use std::time::Instant;
 
 use super::pool::{MePool, ReinitStatusSnapshot, WriterContour};
 use crate::config::{MeBindStaleMode, MeFloorMode, MeSocksKdfPolicy};
-use crate::network::IpFamily;
 use crate::transport::upstream::IpPreference;
 
 // ME writer and DC coverage snapshots.

src/transport/middle_proxy/pool_status/hardswap_snapshot.rs

@@ -41,7 +41,7 @@ impl MePool {
         let pending_generation = reinit.pending_hardswap_generation;
         let pending = pending_generation != 0;
         let mut pending_writers_current = 0usize;
-        let mut pending_by_group = HashMap::<(i32, IpFamily), usize>::new();
+        let mut pending_writer_addrs = Vec::<(i32, SocketAddr)>::new();
         let mut orphan_warm_writers_current = 0usize;
 
         for writer in writers.iter() {
@@ -60,40 +60,18 @@ impl MePool {
                     .is_some_and(|endpoints| endpoints.contains(&writer.addr))
             {
                 pending_writers_current = pending_writers_current.saturating_add(1);
-                let family = if writer.addr.is_ipv4() {
-                    IpFamily::V4
-                } else {
-                    IpFamily::V6
-                };
-                *pending_by_group
-                    .entry((writer.writer_dc, family))
-                    .or_insert(0) += 1;
+                pending_writer_addrs.push((writer.writer_dc, writer.addr));
             }
         }
 
-        let mut pending_writer_deficit = 0usize;
-        let mut pending_missing_dc_groups = 0usize;
-        if pending {
-            for (dc, endpoints) in &desired_by_dc {
-                for family in [IpFamily::V4, IpFamily::V6] {
-                    let endpoint_count = endpoints
-                        .iter()
-                        .filter(|endpoint| endpoint.is_ipv4() == (family == IpFamily::V4))
-                        .count();
-                    if endpoint_count == 0 {
-                        continue;
-                    }
-                    let alive = pending_by_group.get(&(*dc, family)).copied().unwrap_or(0);
-                    let required = self.required_writers_for_dc(endpoint_count);
-                    let deficit = required.saturating_sub(alive);
-                    pending_writer_deficit = pending_writer_deficit.saturating_add(deficit);
-                    if deficit > 0 {
-                        pending_missing_dc_groups =
-                            pending_missing_dc_groups.saturating_add(1);
-                    }
-                }
-            }
-        }
+        let pending_coverage = pending
+            .then(|| self.hardswap_coverage(&desired_by_dc, &pending_writer_addrs));
+        let pending_writer_deficit = pending_coverage
+            .as_ref()
+            .map_or(0, |coverage| coverage.writer_deficit);
+        let pending_missing_dc_groups = pending_coverage
+            .as_ref()
+            .map_or(0, |coverage| coverage.missing_groups.len());
         let (replacement_preparing_current, replacement_retiring_current) =
             self.registry.writer_replacement_counts();
         let pending_age_secs = pending.then(|| {

src/web/http/generation_fence_tests.rs

@@ -0,0 +1,51 @@
+use super::*;
+use crate::web::manager::ManagerError;
+
+#[tokio::test]
+async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() {
+    let initial = test_runtime_generation(1, runtime_config([21; 32], WebCarrier::Https));
+    let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial)));
+    let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime));
+    let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone();
+    let replacement = test_runtime_generation(2, runtime_config([22; 32], WebCarrier::HttpsLanes));
+    active_runtime.store(Arc::clone(&replacement));
+
+    let result =
+        runtime.issue_bootstrap_for_generation(&initial, profile, "192.0.2.10".parse().unwrap());
+
+    assert!(result.is_ok());
+    runtime.shutdown().await;
+    initial.stop_sessions().await;
+    initial.stop_background_tasks().await;
+    replacement.stop_sessions().await;
+    replacement.stop_background_tasks().await;
+}
+
+#[tokio::test]
+async fn stale_generation_cannot_publish_bootstrap_after_disabled_cutover() {
+    let capability = [23u8; 32];
+    let initial = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https));
+    let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial)));
+    let runtime = WebProcessRuntime::start(active_runtime);
+    let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone();
+    let mut disabled_config = runtime_config(capability, WebCarrier::Https);
+    disabled_config.web.enabled = false;
+    let disabled = test_runtime_generation(2, disabled_config);
+
+    runtime.activate_generation(Arc::clone(&disabled));
+    let result = runtime.issue_bootstrap_for_generation(
+        &initial,
+        profile,
+        "192.0.2.10".parse().unwrap(),
+    );
+
+    assert!(matches!(result, Err(ManagerError::Closed)));
+    let status = serde_json::to_value(runtime.try_status()).unwrap();
+    assert_eq!(status["manager"]["bootstraps"], 0);
+
+    runtime.shutdown().await;
+    initial.stop_sessions().await;
+    initial.stop_background_tasks().await;
+    disabled.stop_sessions().await;
+    disabled.stop_background_tasks().await;
+}

src/web/http/tests.rs

@@ -17,8 +17,8 @@ use crate::config::{
 use crate::maestro::generation::test_runtime_generation;
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::{
-    CloseOperationSelector, ControlError, ManagerError, SessionDetail, SessionFilter,
-    SessionListRequest, SessionRefError, WebProcessRuntime,
+    CloseOperationSelector, ControlError, SessionDetail, SessionFilter, SessionListRequest,
+    SessionRefError, WebProcessRuntime,
 };
 
 #[path = "legacy_tests.rs"]
@@ -34,6 +34,9 @@ mod diagnostic_tests;
 // Reload-stability tests for session-owned timeout policy.
 #[path = "session_policy_tests.rs"]
 mod session_policy_tests;
+// Runtime-generation authority fences for WEB bootstrap publication.
+#[path = "generation_fence_tests.rs"]
+mod generation_fence_tests;
 // Runtime control integration stays separate from carrier protocol scenarios.
 #[path = "control_tests.rs"]
 mod control_tests;
@@ -413,55 +416,6 @@ async fn unused_bootstrap_survives_equivalent_runtime_generation_swap() {
     replacement.stop_background_tasks().await;
 }
 
-#[tokio::test]
-async fn bridge_bootstrap_uses_the_generation_that_selected_its_profile() {
-    let initial = test_runtime_generation(1, runtime_config([21; 32], WebCarrier::Https));
-    let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial)));
-    let runtime = WebProcessRuntime::start(Arc::clone(&active_runtime));
-    let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone();
-    let replacement = test_runtime_generation(2, runtime_config([22; 32], WebCarrier::HttpsLanes));
-    active_runtime.store(Arc::clone(&replacement));
-
-    let result =
-        runtime.issue_bootstrap_for_generation(&initial, profile, "192.0.2.10".parse().unwrap());
-
-    assert!(result.is_ok());
-    runtime.shutdown().await;
-    initial.stop_sessions().await;
-    initial.stop_background_tasks().await;
-    replacement.stop_sessions().await;
-    replacement.stop_background_tasks().await;
-}
-
-#[tokio::test]
-async fn stale_generation_cannot_publish_bootstrap_after_disabled_cutover() {
-    let capability = [23u8; 32];
-    let initial = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https));
-    let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&initial)));
-    let runtime = WebProcessRuntime::start(active_runtime);
-    let profile = initial.config().web.runtime.as_ref().unwrap().profiles[0].clone();
-    let mut disabled_config = runtime_config(capability, WebCarrier::Https);
-    disabled_config.web.enabled = false;
-    let disabled = test_runtime_generation(2, disabled_config);
-
-    runtime.activate_generation(Arc::clone(&disabled));
-    let result = runtime.issue_bootstrap_for_generation(
-        &initial,
-        profile,
-        "192.0.2.10".parse().unwrap(),
-    );
-
-    assert!(matches!(result, Err(ManagerError::Closed)));
-    let status = serde_json::to_value(runtime.try_status()).unwrap();
-    assert_eq!(status["manager"]["bootstraps"], 0);
-
-    runtime.shutdown().await;
-    initial.stop_sessions().await;
-    initial.stop_background_tasks().await;
-    disabled.stop_sessions().await;
-    disabled.stop_background_tasks().await;
-}
-
 #[tokio::test]
 async fn unused_bootstrap_is_rejected_after_profile_identity_change() {
     let capability = [11u8; 32];

src/web/session/lanes.rs

@@ -2,7 +2,6 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use bytes::{BufMut, Bytes, BytesMut};
-use tokio::sync::OwnedSemaphorePermit;
 
 use super::lane_downlink::take_lane_down_batch;
 use super::{
@@ -13,6 +12,9 @@ use crate::web::frame::{self, FrameType};
 use crate::web::manager::ManagerError;
 use crate::web::telemetry::WebSessionLifecycleObservation;
 
+// Bounded lane-open admission and cancellation-safe wait lifecycle.
+mod lane_open_wait;
+
 impl WebSession {
     /// Polls one lane with independent cursor replay and newest-poll-wins semantics.
     pub(crate) async fn poll_down_lane(
@@ -311,97 +313,6 @@ impl WebSession {
         }
     }
 
-    async fn wait_for_lane_open(&self, lane_id: u32, cursor: u64) -> Result<bool, ManagerError> {
-        if self.close_if_cancelled() {
-            return Err(ManagerError::Closed);
-        }
-        let wait = {
-            let mut state = self.state.lock();
-            if state.closed || self.cancel.is_cancelled() {
-                drop(state);
-                self.close_if_cancelled();
-                return Err(ManagerError::Closed);
-            }
-            if state.carrier_lanes.contains_key(&lane_id) {
-                return Ok(true);
-            }
-            if cursor != 0 || lane_id == 0 {
-                drop(state);
-                self.close(SessionCloseReason::Protocol);
-                return Err(ManagerError::Protocol);
-            }
-            if state.closed_streams.contains(&lane_id)
-                || state.closing_streams.contains_key(&lane_id)
-            {
-                return Ok(true);
-            }
-            if state.lane_open_waits >= self.limits.max_lane_open_waits_per_session {
-                return Err(ManagerError::Limit);
-            }
-            let Some(manager) = self.manager.upgrade() else {
-                return Err(ManagerError::Closed);
-            };
-            let Some(auxiliary) = manager.try_lane_poll(true) else {
-                return Err(ManagerError::Limit);
-            };
-            state.lane_open_waits += 1;
-            let observation = WebSessionLifecycleObservation::HttpActivityAfterGap;
-            self.touch_peer_locked(&mut state, Instant::now(), observation);
-            LaneOpenWaitGuard {
-                session: self,
-                _auxiliary: auxiliary,
-            }
-        };
-        let deadline = Duration::from_secs(self.timeouts.lane_open_wait_secs);
-        let opened = tokio::time::timeout(deadline, async {
-            loop {
-                let notified = self.lane_open_notify.notified();
-                tokio::pin!(notified);
-                notified.as_mut().enable();
-                {
-                    let state = self.state.lock();
-                    if state.closed || self.cancel.is_cancelled() {
-                        drop(state);
-                        self.close_if_cancelled();
-                        return Err(ManagerError::Closed);
-                    }
-                    if state.carrier_lanes.contains_key(&lane_id)
-                        || state.closed_streams.contains(&lane_id)
-                        || state.closing_streams.contains_key(&lane_id)
-                    {
-                        return Ok(true);
-                    }
-                }
-                notified.await;
-            }
-        });
-        let opened = tokio::select! {
-            biased;
-            _ = self.cancel.cancelled() => {
-                drop(wait);
-                self.close_if_cancelled();
-                return Err(ManagerError::Closed);
-            }
-            opened = opened => opened,
-        };
-        drop(wait);
-        match opened {
-            Ok(result) => result,
-            Err(_) => {
-                let state = self.state.lock();
-                if state.closed || self.cancel.is_cancelled() {
-                    drop(state);
-                    self.close_if_cancelled();
-                    Err(ManagerError::Closed)
-                } else {
-                    Ok(state.carrier_lanes.contains_key(&lane_id)
-                        || state.closed_streams.contains(&lane_id)
-                        || state.closing_streams.contains_key(&lane_id))
-                }
-            }
-        }
-    }
-
     pub(super) fn queue_lane_frame_locked(
         &self,
         state: &mut SessionState,
@@ -571,18 +482,6 @@ impl WebSession {
     }
 }
 
-struct LaneOpenWaitGuard<'a> {
-    session: &'a WebSession,
-    _auxiliary: OwnedSemaphorePermit,
-}
-
-impl Drop for LaneOpenWaitGuard<'_> {
-    fn drop(&mut self) {
-        let mut state = self.session.state.lock();
-        state.lane_open_waits = state.lane_open_waits.saturating_sub(1);
-    }
-}
-
 // Lane-specific protocol, replay, and lifecycle tests.
 #[cfg(test)]
 mod tests;

src/web/session/lanes/lane_open_wait.rs

@@ -0,0 +1,118 @@
+use std::time::{Duration, Instant};
+
+use tokio::sync::OwnedSemaphorePermit;
+
+use super::WebSession;
+use crate::web::manager::ManagerError;
+use crate::web::session::SessionCloseReason;
+use crate::web::telemetry::WebSessionLifecycleObservation;
+
+impl WebSession {
+    /// Waits for a non-control lane to become observable within the bounded admission budget.
+    pub(super) async fn wait_for_lane_open(
+        &self,
+        lane_id: u32,
+        cursor: u64,
+    ) -> Result<bool, ManagerError> {
+        if self.close_if_cancelled() {
+            return Err(ManagerError::Closed);
+        }
+        let wait = {
+            let mut state = self.state.lock();
+            if state.closed || self.cancel.is_cancelled() {
+                drop(state);
+                self.close_if_cancelled();
+                return Err(ManagerError::Closed);
+            }
+            if state.carrier_lanes.contains_key(&lane_id) {
+                return Ok(true);
+            }
+            if cursor != 0 || lane_id == 0 {
+                drop(state);
+                self.close(SessionCloseReason::Protocol);
+                return Err(ManagerError::Protocol);
+            }
+            if state.closed_streams.contains(&lane_id)
+                || state.closing_streams.contains_key(&lane_id)
+            {
+                return Ok(true);
+            }
+            if state.lane_open_waits >= self.limits.max_lane_open_waits_per_session {
+                return Err(ManagerError::Limit);
+            }
+            let Some(manager) = self.manager.upgrade() else {
+                return Err(ManagerError::Closed);
+            };
+            let Some(auxiliary) = manager.try_lane_poll(true) else {
+                return Err(ManagerError::Limit);
+            };
+            state.lane_open_waits += 1;
+            let observation = WebSessionLifecycleObservation::HttpActivityAfterGap;
+            self.touch_peer_locked(&mut state, Instant::now(), observation);
+            LaneOpenWaitGuard {
+                session: self,
+                _auxiliary: auxiliary,
+            }
+        };
+        let deadline = Duration::from_secs(self.timeouts.lane_open_wait_secs);
+        let opened = tokio::time::timeout(deadline, async {
+            loop {
+                let notified = self.lane_open_notify.notified();
+                tokio::pin!(notified);
+                notified.as_mut().enable();
+                {
+                    let state = self.state.lock();
+                    if state.closed || self.cancel.is_cancelled() {
+                        drop(state);
+                        self.close_if_cancelled();
+                        return Err(ManagerError::Closed);
+                    }
+                    if state.carrier_lanes.contains_key(&lane_id)
+                        || state.closed_streams.contains(&lane_id)
+                        || state.closing_streams.contains_key(&lane_id)
+                    {
+                        return Ok(true);
+                    }
+                }
+                notified.await;
+            }
+        });
+        let opened = tokio::select! {
+            biased;
+            _ = self.cancel.cancelled() => {
+                drop(wait);
+                self.close_if_cancelled();
+                return Err(ManagerError::Closed);
+            }
+            opened = opened => opened,
+        };
+        drop(wait);
+        match opened {
+            Ok(result) => result,
+            Err(_) => {
+                let state = self.state.lock();
+                if state.closed || self.cancel.is_cancelled() {
+                    drop(state);
+                    self.close_if_cancelled();
+                    Err(ManagerError::Closed)
+                } else {
+                    Ok(state.carrier_lanes.contains_key(&lane_id)
+                        || state.closed_streams.contains(&lane_id)
+                        || state.closing_streams.contains_key(&lane_id))
+                }
+            }
+        }
+    }
+}
+
+struct LaneOpenWaitGuard<'a> {
+    session: &'a WebSession,
+    _auxiliary: OwnedSemaphorePermit,
+}
+
+impl Drop for LaneOpenWaitGuard<'_> {
+    fn drop(&mut self) {
+        let mut state = self.session.state.lock();
+        state.lane_open_waits = state.lane_open_waits.saturating_sub(1);
+    }
+}

src/web/session/negotiation.rs

@@ -299,258 +299,6 @@ impl WebSession {
     }
 }
 
+// Carrier health publication and negotiation race tests.
 #[cfg(test)]
-mod tests {
-    use std::net::SocketAddr;
-    use std::sync::{Arc, Barrier};
-
-    use super::*;
-    use crate::config::{
-        WebCarrier, WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig,
-    };
-    use crate::web::manager::{CarrierClientClass, WebProcessRuntime};
-    use crate::web::session::{SessionCloseOutcome, SessionCloseReason};
-
-    fn session(carrier: WebCarrier, deadline: Instant) -> Arc<WebSession> {
-        let profile = Arc::new(WebRuntimeProfile {
-            host: "proxy.example.com".to_string(),
-            public_addr: SocketAddr::from(([203, 0, 113, 10], 443)),
-            user: "alice".to_string(),
-            secret_mode: WebSecretMode::Plain,
-            carrier,
-            carrier_negotiation_enabled: true,
-            carrier_learning: false,
-            carriers: Arc::from([carrier]),
-            carrier_negotiation_deadlines_secs: [3, 5, 8, 12],
-            capability: [0; 32],
-            credential_id: [0; 16],
-            key_fingerprint: "0000000000000000".to_string(),
-            max_sessions: 1,
-            max_streams: 1,
-            max_streams_per_session: 1,
-        });
-        WebSession::new(
-            std::sync::Weak::<WebProcessRuntime>::new(),
-            [1; 32],
-            "192.0.2.10".parse().unwrap(),
-            1,
-            profile,
-            [2; 32],
-            carrier,
-            1,
-            [3; 32],
-            Some(deadline),
-            CarrierClientClass::Bridge,
-            None,
-            true,
-            false,
-            WebLimitsConfig::default(),
-            WebTimeoutsConfig::default(),
-            None,
-        )
-    }
-
-    fn arm_http_health(session: &WebSession, now: Instant) {
-        let mut state = session.state.lock();
-        state.negotiation_phase = SessionNegotiationPhase::Committed;
-        state.carrier_commit_published = true;
-        state.carrier_health_due_at = Some(now - Duration::from_secs(1));
-        state.carrier_health_uplink = true;
-        state.carrier_health_downlink = true;
-        state.carrier_health_activity_at = Some(now);
-    }
-
-    #[test]
-    fn final_deadline_refuses_uncommitted_progress() {
-        let session = session(WebCarrier::Https, Instant::now() - Duration::from_secs(1));
-        let state = session.state.lock();
-        assert_eq!(
-            session.ensure_carrier_active_locked(&state),
-            Err(crate::web::manager::ManagerError::Closed)
-        );
-        assert!(matches!(
-            state.negotiation_phase,
-            SessionNegotiationPhase::Uncommitted
-        ));
-    }
-
-    #[test]
-    fn http_health_requires_authenticated_activity_after_the_window() {
-        let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-        let now = Instant::now();
-        let mut state = session.state.lock();
-        state.negotiation_phase = SessionNegotiationPhase::Committed;
-        state.carrier_commit_published = true;
-        state.carrier_health_due_at = Some(now - Duration::from_secs(1));
-        state.carrier_health_uplink = true;
-        state.carrier_health_downlink = true;
-        state.carrier_health_activity_at = Some(now - Duration::from_secs(2));
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_none()
-        );
-        state.carrier_health_activity_at = Some(now);
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_some()
-        );
-    }
-
-    #[test]
-    fn websocket_health_requires_the_exact_live_probe_owner() {
-        let session = session(
-            WebCarrier::Websocket,
-            Instant::now() + Duration::from_secs(60),
-        );
-        let now = Instant::now();
-        let mut state = session.state.lock();
-        state.negotiation_phase = SessionNegotiationPhase::Committed;
-        state.carrier_commit_published = true;
-        state.carrier_health_due_at = Some(now - Duration::from_secs(1));
-        state.websocket_carrier_active = true;
-        state.websocket_commit_ack_owner = Some(7);
-        state.websocket_commit_ack_written = true;
-        state.carrier_health_uplink = true;
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_none()
-        );
-        state.websocket_probe_claimed = true;
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_some()
-        );
-    }
-
-    #[test]
-    fn health_waits_for_manager_commit_publication() {
-        let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-        let now = Instant::now();
-        let mut state = session.state.lock();
-        state.negotiation_phase = SessionNegotiationPhase::Committed;
-        state.carrier_health_due_at = Some(now - Duration::from_secs(1));
-        state.carrier_health_uplink = true;
-        state.carrier_health_downlink = true;
-        state.carrier_health_activity_at = Some(now);
-
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_none()
-        );
-        assert_eq!(
-            session.carrier_health_publication_state(),
-            CarrierHealthPublicationState::Awaiting
-        );
-    }
-
-    #[test]
-    fn health_publication_claim_is_single_shot() {
-        let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-        let now = Instant::now();
-        arm_http_health(&session, now);
-        let mut state = session.state.lock();
-
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_some()
-        );
-        assert!(
-            session
-                .carrier_health_ready_locked(&mut state, now)
-                .is_none()
-        );
-        drop(state);
-        assert_eq!(
-            session.carrier_health_publication_state(),
-            CarrierHealthPublicationState::Publishing
-        );
-        assert!(session.publish_carrier_health());
-        assert!(!session.publish_carrier_health());
-        assert_eq!(
-            session.carrier_health_publication_state(),
-            CarrierHealthPublicationState::Published
-        );
-    }
-
-    #[test]
-    fn concurrent_health_and_close_always_reach_one_terminal_state() {
-        for _ in 0..512 {
-            let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-            let now = Instant::now();
-            arm_http_health(&session, now);
-            let barrier = Arc::new(Barrier::new(3));
-            let health_session = Arc::clone(&session);
-            let health_barrier = Arc::clone(&barrier);
-            let health = std::thread::spawn(move || {
-                health_barrier.wait();
-                std::thread::yield_now();
-                let claim = {
-                    let mut state = health_session.state.lock();
-                    health_session.carrier_health_ready_locked(&mut state, now)
-                };
-                if claim.is_some() {
-                    health_session.publish_carrier_health();
-                }
-            });
-            let close_session = Arc::clone(&session);
-            let close_barrier = Arc::clone(&barrier);
-            let close = std::thread::spawn(move || {
-                close_barrier.wait();
-                std::thread::yield_now();
-                close_session.close(SessionCloseReason::ApiClose);
-            });
-            barrier.wait();
-            health.join().unwrap();
-            close.join().unwrap();
-
-            assert!(matches!(
-                session.carrier_health_publication_state(),
-                CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected
-            ));
-            assert!(!session.publish_carrier_health());
-            assert_eq!(
-                session.close(SessionCloseReason::ApiClose),
-                SessionCloseOutcome::AlreadyClosing
-            );
-        }
-    }
-
-    #[test]
-    fn commit_and_supersede_have_one_session_lock_winner() {
-        let committed = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-        {
-            let mut state = committed.state.lock();
-            assert!(
-                committed
-                    .record_uplink_progress_locked(
-                        &mut state,
-                        AppliedProgress {
-                            accepted_open: true,
-                            accepted_data: true,
-                        },
-                    )
-                    .0
-            );
-        }
-        assert!(!committed.begin_carrier_supersede());
-
-        let replacing = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
-        assert!(replacing.begin_carrier_supersede());
-        assert_eq!(
-            replacing.ensure_carrier_active_locked(&replacing.state.lock()),
-            Err(crate::web::manager::ManagerError::Closed)
-        );
-        replacing.cancel_carrier_supersede();
-        assert!(
-            replacing
-                .ensure_carrier_active_locked(&replacing.state.lock())
-                .is_ok()
-        );
-    }
-}
+mod tests;

src/web/session/negotiation/tests.rs

@@ -0,0 +1,252 @@
+use std::net::SocketAddr;
+use std::sync::{Arc, Barrier};
+
+use super::*;
+use crate::config::{
+    WebCarrier, WebLimitsConfig, WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig,
+};
+use crate::web::manager::{CarrierClientClass, WebProcessRuntime};
+use crate::web::session::{SessionCloseOutcome, SessionCloseReason};
+
+fn session(carrier: WebCarrier, deadline: Instant) -> Arc<WebSession> {
+    let profile = Arc::new(WebRuntimeProfile {
+        host: "proxy.example.com".to_string(),
+        public_addr: SocketAddr::from(([203, 0, 113, 10], 443)),
+        user: "alice".to_string(),
+        secret_mode: WebSecretMode::Plain,
+        carrier,
+        carrier_negotiation_enabled: true,
+        carrier_learning: false,
+        carriers: Arc::from([carrier]),
+        carrier_negotiation_deadlines_secs: [3, 5, 8, 12],
+        capability: [0; 32],
+        credential_id: [0; 16],
+        key_fingerprint: "0000000000000000".to_string(),
+        max_sessions: 1,
+        max_streams: 1,
+        max_streams_per_session: 1,
+    });
+    WebSession::new(
+        std::sync::Weak::<WebProcessRuntime>::new(),
+        [1; 32],
+        "192.0.2.10".parse().unwrap(),
+        1,
+        profile,
+        [2; 32],
+        carrier,
+        1,
+        [3; 32],
+        Some(deadline),
+        CarrierClientClass::Bridge,
+        None,
+        true,
+        false,
+        WebLimitsConfig::default(),
+        WebTimeoutsConfig::default(),
+        None,
+    )
+}
+
+fn arm_http_health(session: &WebSession, now: Instant) {
+    let mut state = session.state.lock();
+    state.negotiation_phase = SessionNegotiationPhase::Committed;
+    state.carrier_commit_published = true;
+    state.carrier_health_due_at = Some(now - Duration::from_secs(1));
+    state.carrier_health_uplink = true;
+    state.carrier_health_downlink = true;
+    state.carrier_health_activity_at = Some(now);
+}
+
+#[test]
+fn final_deadline_refuses_uncommitted_progress() {
+    let session = session(WebCarrier::Https, Instant::now() - Duration::from_secs(1));
+    let state = session.state.lock();
+    assert_eq!(
+        session.ensure_carrier_active_locked(&state),
+        Err(crate::web::manager::ManagerError::Closed)
+    );
+    assert!(matches!(
+        state.negotiation_phase,
+        SessionNegotiationPhase::Uncommitted
+    ));
+}
+
+#[test]
+fn http_health_requires_authenticated_activity_after_the_window() {
+    let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+    let now = Instant::now();
+    let mut state = session.state.lock();
+    state.negotiation_phase = SessionNegotiationPhase::Committed;
+    state.carrier_commit_published = true;
+    state.carrier_health_due_at = Some(now - Duration::from_secs(1));
+    state.carrier_health_uplink = true;
+    state.carrier_health_downlink = true;
+    state.carrier_health_activity_at = Some(now - Duration::from_secs(2));
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_none()
+    );
+    state.carrier_health_activity_at = Some(now);
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_some()
+    );
+}
+
+#[test]
+fn websocket_health_requires_the_exact_live_probe_owner() {
+    let session = session(
+        WebCarrier::Websocket,
+        Instant::now() + Duration::from_secs(60),
+    );
+    let now = Instant::now();
+    let mut state = session.state.lock();
+    state.negotiation_phase = SessionNegotiationPhase::Committed;
+    state.carrier_commit_published = true;
+    state.carrier_health_due_at = Some(now - Duration::from_secs(1));
+    state.websocket_carrier_active = true;
+    state.websocket_commit_ack_owner = Some(7);
+    state.websocket_commit_ack_written = true;
+    state.carrier_health_uplink = true;
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_none()
+    );
+    state.websocket_probe_claimed = true;
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_some()
+    );
+}
+
+#[test]
+fn health_waits_for_manager_commit_publication() {
+    let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+    let now = Instant::now();
+    let mut state = session.state.lock();
+    state.negotiation_phase = SessionNegotiationPhase::Committed;
+    state.carrier_health_due_at = Some(now - Duration::from_secs(1));
+    state.carrier_health_uplink = true;
+    state.carrier_health_downlink = true;
+    state.carrier_health_activity_at = Some(now);
+
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_none()
+    );
+    assert_eq!(
+        session.carrier_health_publication_state(),
+        CarrierHealthPublicationState::Awaiting
+    );
+}
+
+#[test]
+fn health_publication_claim_is_single_shot() {
+    let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+    let now = Instant::now();
+    arm_http_health(&session, now);
+    let mut state = session.state.lock();
+
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_some()
+    );
+    assert!(
+        session
+            .carrier_health_ready_locked(&mut state, now)
+            .is_none()
+    );
+    drop(state);
+    assert_eq!(
+        session.carrier_health_publication_state(),
+        CarrierHealthPublicationState::Publishing
+    );
+    assert!(session.publish_carrier_health());
+    assert!(!session.publish_carrier_health());
+    assert_eq!(
+        session.carrier_health_publication_state(),
+        CarrierHealthPublicationState::Published
+    );
+}
+
+#[test]
+fn concurrent_health_and_close_always_reach_one_terminal_state() {
+    for _ in 0..512 {
+        let session = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+        let now = Instant::now();
+        arm_http_health(&session, now);
+        let barrier = Arc::new(Barrier::new(3));
+        let health_session = Arc::clone(&session);
+        let health_barrier = Arc::clone(&barrier);
+        let health = std::thread::spawn(move || {
+            health_barrier.wait();
+            std::thread::yield_now();
+            let claim = {
+                let mut state = health_session.state.lock();
+                health_session.carrier_health_ready_locked(&mut state, now)
+            };
+            if claim.is_some() {
+                health_session.publish_carrier_health();
+            }
+        });
+        let close_session = Arc::clone(&session);
+        let close_barrier = Arc::clone(&barrier);
+        let close = std::thread::spawn(move || {
+            close_barrier.wait();
+            std::thread::yield_now();
+            close_session.close(SessionCloseReason::ApiClose);
+        });
+        barrier.wait();
+        health.join().unwrap();
+        close.join().unwrap();
+
+        assert!(matches!(
+            session.carrier_health_publication_state(),
+            CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected
+        ));
+        assert!(!session.publish_carrier_health());
+        assert_eq!(
+            session.close(SessionCloseReason::ApiClose),
+            SessionCloseOutcome::AlreadyClosing
+        );
+    }
+}
+
+#[test]
+fn commit_and_supersede_have_one_session_lock_winner() {
+    let committed = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+    {
+        let mut state = committed.state.lock();
+        assert!(
+            committed
+                .record_uplink_progress_locked(
+                    &mut state,
+                    AppliedProgress {
+                        accepted_open: true,
+                        accepted_data: true,
+                    },
+                )
+                .0
+        );
+    }
+    assert!(!committed.begin_carrier_supersede());
+
+    let replacing = session(WebCarrier::Https, Instant::now() + Duration::from_secs(60));
+    assert!(replacing.begin_carrier_supersede());
+    assert_eq!(
+        replacing.ensure_carrier_active_locked(&replacing.state.lock()),
+        Err(crate::web::manager::ManagerError::Closed)
+    );
+    replacing.cancel_carrier_supersede();
+    assert!(
+        replacing
+            .ensure_carrier_active_locked(&replacing.state.lock())
+            .is_ok()
+    );
+}