diff --git a/src/api/web_runtime/observability.rs b/src/api/web_runtime/observability.rs
index a1d4fc7..971fc8f 100644
--- a/src/api/web_runtime/observability.rs
+++ b/src/api/web_runtime/observability.rs
@@ -3,11 +3,11 @@ use serde::Serialize;
 use crate::config::{ProxyConfig, WebHttpConnectionCapacityAction};
 use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication};
 use crate::web::manager::{WebCapacityResourceStatus, WebCapacitySnapshot, WebProcessRuntime};
-use crate::web::telemetry::{WebOutcomeCounter, WebRejectionCounter};
 use crate::web::telemetry::{
     WebBridgeRecoveryCounter, WebCarrierFailureCounter, WebCarrierLearningCounter,
     WebCarrierSelectionCounter, WebSessionCloseCounter, WebSessionLifecycleObservationCounter,
 };
+use crate::web::telemetry::{WebOutcomeCounter, WebRejectionCounter};
 
 /// Private WEB ingress state owned by this Telemt process.
 #[derive(Serialize)]
diff --git a/src/config/types.rs b/src/config/types.rs
index 72813f7..12bd690 100644
--- a/src/config/types.rs
+++ b/src/config/types.rs
@@ -61,9 +61,9 @@ pub(crate) use web::{
     WebRuntimeConfig, WebRuntimeDecoy, WebRuntimeProfile, WebRuntimeVhost, WebStaticAsset,
     WebStaticSite,
 };
+pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES;
 #[allow(unused_imports)]
 pub use web_carrier::{WebCarrier, WebCarriers};
-pub(crate) use web_carrier::WEB_CARRIER_LEARNING_MIN_ENTRIES;
 pub(crate) use web_debug::web_debug_fits_limits;
 pub use web_debug::{WebDebugBodyCapture, WebDebugConfig};
 
diff --git a/src/maestro/runtime_build.rs b/src/maestro/runtime_build.rs
index f43b19b..d08cb4d 100644
--- a/src/maestro/runtime_build.rs
+++ b/src/maestro/runtime_build.rs
@@ -429,8 +429,7 @@ pub(crate) fn resolve_reload_config(
     }
     if effective.web.carrier_negotiation_enabled()
         && effective.web.carrier_learning
-        && effective.web.limits.max_carrier_learning_entries
-            < WEB_CARRIER_LEARNING_MIN_ENTRIES
+        && effective.web.limits.max_carrier_learning_entries < WEB_CARRIER_LEARNING_MIN_ENTRIES
     {
         if old.web.carrier_learning != desired.web.carrier_learning {
             fields.push("web.carrier_learning".to_string());
diff --git a/src/metrics/web.rs b/src/metrics/web.rs
index 3588941..11d4cf8 100644
--- a/src/metrics/web.rs
+++ b/src/metrics/web.rs
@@ -1,8 +1,7 @@
 use std::fmt::Write;
 
 use crate::config::{
-    ProxyConfig, WebCarrier, WebCarrierNegotiationAggressiveness,
-    WebHttpConnectionCapacityAction,
+    ProxyConfig, WebCarrier, WebCarrierNegotiationAggressiveness, WebHttpConnectionCapacityAction,
 };
 use crate::web::control::{WebRuntimeLifecycle, WebRuntimePublication};
 use crate::web::manager::{CarrierFailure, OperatorLifecycleState};
@@ -290,7 +289,10 @@ fn render_carrier_negotiation(
         "partial"
     } else if !policy_matches {
         "pending"
-    } else if learning.as_ref().is_some_and(|status| status.epoch.is_none()) {
+    } else if learning
+        .as_ref()
+        .is_some_and(|status| status.epoch.is_none())
+    {
         "exhausted"
     } else if learning.as_ref().is_some_and(|status| status.enabled) {
         "enabled"
@@ -317,7 +319,10 @@ fn render_carrier_negotiation(
     let _ = writeln!(out, "# TYPE telemt_web_carrier_learning_entries gauge");
     for (kind, value) in [
         ("used", learning.as_ref().map_or(0, |status| status.entries)),
-        ("limit", learning.as_ref().map_or(0, |status| status.capacity)),
+        (
+            "limit",
+            learning.as_ref().map_or(0, |status| status.capacity),
+        ),
     ] {
         let _ = writeln!(
             out,
@@ -478,7 +483,9 @@ mod tests {
         );
         assert!(output.contains("telemt_web_ingress_lifecycle_state{state=\"starting\"} 1"));
         assert_eq!(
-            output.matches("telemt_web_carrier_selections_total{").count(),
+            output
+                .matches("telemt_web_carrier_selections_total{")
+                .count(),
             crate::config::WebCarrier::ALL.len()
                 * crate::web::telemetry::WebCarrierSelectionDisposition::ALL.len()
         );
@@ -497,9 +504,7 @@ mod tests {
             crate::config::WebCarrier::ALL.len()
                 * crate::web::telemetry::WebCarrierLearningOutcome::ALL.len()
         );
-        assert!(output.contains(
-            "telemt_web_carrier_learning_state{state=\"unavailable\"} 1"
-        ));
+        assert!(output.contains("telemt_web_carrier_learning_state{state=\"unavailable\"} 1"));
         assert_eq!(
             output.matches("telemt_web_session_closures_total{").count(),
             crate::config::WebCarrier::ALL.len()
diff --git a/src/metrics/web/lifecycle.rs b/src/metrics/web/lifecycle.rs
index cff1753..bca4c25 100644
--- a/src/metrics/web/lifecycle.rs
+++ b/src/metrics/web/lifecycle.rs
@@ -5,11 +5,7 @@ use crate::web::control::WebRuntimePublication;
 use crate::web::session::SessionCloseReason;
 use crate::web::telemetry::{WebBridgeRecoveryEvent, WebSessionLifecycleObservation};
 
-pub(super) fn render(
-    out: &mut String,
-    publication: &WebRuntimePublication,
-    config: &ProxyConfig,
-) {
+pub(super) fn render(out: &mut String, publication: &WebRuntimePublication, config: &ProxyConfig) {
     let _ = writeln!(
         out,
         "# HELP telemt_web_session_closures_total Closed WEB session incarnations by carrier and terminal reason"
diff --git a/src/web/bridge.rs b/src/web/bridge.rs
index beed502..68f1b74 100644
--- a/src/web/bridge.rs
+++ b/src/web/bridge.rs
@@ -63,7 +63,10 @@ pub(crate) fn render(
             &bridge_recovery_secs.to_string(),
         )
         .replace("__WEBSOCKET_OPEN_SECS__", &websocket_open_secs.to_string())
-        .replace("__RECONNECT_GRACE_SECS__", &reconnect_grace_secs.to_string())
+        .replace(
+            "__RECONNECT_GRACE_SECS__",
+            &reconnect_grace_secs.to_string(),
+        )
         .replace(
             "__CARRIER_PROBE_COALESCE_MS__",
             &carrier_probe_coalesce_ms.to_string(),
diff --git a/src/web/bridge/recovery.js b/src/web/bridge/recovery.js
index b433bf2..9343189 100644
--- a/src/web/bridge/recovery.js
+++ b/src/web/bridge/recovery.js
@@ -29,7 +29,7 @@ function create(settings){
  function heartbeat(owner){
   if(current!==owner||owner.controller.signal.aborted)return;
   const left=remaining(owner);settings.status(left);
-  if(left>0)owner.heartbeat=setTimeout(()=>heartbeat(owner),Math.min(heartbeatMs,left));
+  if(left<=0){owner.controller.abort();return}owner.heartbeat=setTimeout(()=>heartbeat(owner),Math.min(heartbeatMs,left));
  }
  async function load(owner){
   const left=remaining(owner);if(left<=0)throw new Error('recovery deadline');
diff --git a/src/web/bridge/request.js b/src/web/bridge/request.js
index b613f9b..26e52e9 100644
--- a/src/web/bridge/request.js
+++ b/src/web/bridge/request.js
@@ -21,9 +21,10 @@ function create(settings){
  }
  function retryableStatus(status){return status===408||status===429||status===502||status===503||status===504}
  function responsePolicy(path,status){
-  if(path==='/api/v1/session'&&status===200)return {limit:8,exact:true};
-  if(path==='/api/v1/down'&&status===200)return {limit:settings.batchLimit(),exact:false};
-  return {limit:0,exact:true};
+  if(path==='/api/v1/session'&&status===200)return {limit:8,exact:true,reason:'protocol'};
+  if(path==='/api/v1/down'&&status===200)return {limit:settings.batchLimit(),exact:false,reason:'protocol'};
+  if(status===204&&(path==='/api/v1/up'||path==='/api/v1/down'))return {limit:0,exact:true,reason:'protocol'};
+  return {limit:0,exact:true,reason:'http'};
  }
  async function send(path,frozenOptions,remainingBudget,maxAttempts){
   let delay=250,attempt=0,lastReason='network';maxAttempts=maxAttempts||9;
@@ -49,7 +50,7 @@ function create(settings){
       controller.abort();
       if(external&&external.aborted)throw error;
       if(timedOut)throw settings.failure('timeout','response deadline exceeded');
-      throw settings.failure('protocol',error&&error.message);
+      throw settings.failure(policy.reason,error&&error.message);
      }
      response={status:fetched.status,headers:fetched.headers,body};return response;
     }
diff --git a/src/web/bridge/runtime.js b/src/web/bridge/runtime.js
index dbd1152..b9d2610 100644
--- a/src/web/bridge/runtime.js
+++ b/src/web/bridge/runtime.js
@@ -18,7 +18,7 @@ let initialized=false,closed=false,port=null,sessionToken='',cleanupToken='',cre
 let upSequence=1,downCursor='0',upRunning=false,upLease=null,pollController=null;
 let helloFrame=null,helloTimer=null,welcomeSent=false,carrierAttempt=1,carrierFailure='',carrierCommitted=false,terminalFailure='';
 let negotiationStartedAt=0,carrierTimer=null,probeTimer=null,attemptController=null,attemptEpoch=1,candidateRunning=false,switching=false,currentAttempt=null;
-let recoveryController=null,recoveryCommit=null,recoveryReplaced=false,lastSchedulerWall=Date.now(),lastSchedulerMonotonic=performance.now();
+let recoveryController=null,recoveryCommit=null,recoveryReplaced=false,lastSchedulerWall=Date.now(),lastSchedulerMonotonic=performance.now(),schedulerGapPending=0,schedulerTimer=null;
 const pending=[],upPending=[],recoveryPending=[],lanes=new Map(),closedLanes=new Set(),closedLaneOrder=[];
 const canonicalFailures=['timeout','network','upgrade','http','protocol'];
 const failure=(reason,message)=>Object.assign(new Error(message||reason),{telemtReason:reason});
@@ -48,7 +48,7 @@ function settleBatch(lease){if(!buffers.settleBatch(lease))return false;detachLe
 function cancelBatch(lease){if(!lease||lease.settled)return;buffers.cancelBatch(lease);detachLease(lease)}
 const attemptHeaders=(attempt,failure)=>negotiationEnabled?Object.assign({'X-Carrier-Capabilities':carrierCapabilities,'X-Carrier-Attempt':String(attempt)},failure?{'X-Carrier-Failure':failure}:{}):{};
 function finishOldRecovery(){
- recoveryReplaced=false;status('connected','committed','',0);
+ recoveryReplaced=false;resetScheduler();status('connected','committed','',0);
  while(recoveryPending.length&&!closed){const data=recoveryPending.shift();release(data.byteLength,1,null);queueCarrier(data)}
 }
 function rejectRecoveryCommit(error){
@@ -74,6 +74,7 @@ function retireCarrier(policy){
  longPollMs=policy.timeouts.long_poll_secs*1000;bridgeRequestMs=policy.timeouts.bridge_request_secs*1000;
  bridgeRetryMs=policy.timeouts.bridge_retry_secs*1000;bridgeRecoveryMs=policy.timeouts.bridge_recovery_secs*1000;
  websocketOpenMs=policy.timeouts.websocket_open_secs*1000;reconnectGraceMs=policy.timeouts.reconnect_grace_secs*1000;
+ armScheduler();
  negotiationEnabled=policy.negotiation.enabled;candidateCount=policy.negotiation.candidate_count;
  candidateDeadlines=policy.negotiation.deadlines_secs;probeCoalesceMs=policy.negotiation.carrier_probe_coalesce_ms;
  negotiatedCandidateCount=candidateCount;negotiatedFinalDeadline=candidateDeadlines[3];negotiatedFrozen=false;
@@ -100,8 +101,15 @@ function schedulerGap(){
  const gap=Math.max(0,wall-lastSchedulerWall,monotonic-lastSchedulerMonotonic);
  lastSchedulerWall=wall;lastSchedulerMonotonic=monotonic;return gap;
 }
+function resetScheduler(){schedulerGapPending=0;schedulerGap()}
+function armScheduler(){
+ if(schedulerTimer)clearTimeout(schedulerTimer);schedulerTimer=closed?null:setTimeout(sampleScheduler,Math.max(250,Math.min(30000,Math.floor(reconnectGraceMs/4))));
+}
+function sampleScheduler(){
+ schedulerTimer=null;const gap=schedulerGap();if(carrierCommitted&&!recoveryController.active())schedulerGapPending=Math.max(schedulerGapPending,gap);armScheduler();
+}
 function observeResumeTrigger(){
- const gap=schedulerGap();if(!carrierCommitted||closed)return;
+ const gap=Math.max(schedulerGapPending,schedulerGap());schedulerGapPending=0;if(!carrierCommitted||closed)return;
  if(gap>=2*longPollMs)status('reconnecting','retrying','',bridgeRecoveryMs);
  if(gap>=reconnectGraceMs&&!recoveryController.active())recoveryController.recover('timeout',null);
 }
@@ -235,6 +243,7 @@ function commitCarrier(probe,epoch){
  if(switching){fail('protocol');return}
  clearProbeTimer();try{consumeProbe(probe)}catch(error){fail('protocol');return}
  carrierCommitted=true;candidateRunning=false;if(carrierTimer)clearTimeout(carrierTimer);carrierTimer=null;
+ resetScheduler();
  attemptController=null;currentAttempt=null;
  status('connected','committed','',0);
  if(carrier==='https')poll();
@@ -475,7 +484,7 @@ function deleteSession(){
  if(token)fetch(relayOrigin+'/api/v1/session',options('DELETE',token,null,headers,undefined,true)).catch(()=>{});
 }
 function close(notifyServer){
- if(closed)return;closed=true;if(recoveryController)recoveryController.cancel();rejectRecoveryCommit(failure('network','bridge closed'));if(helloTimer)clearTimeout(helloTimer);helloTimer=null;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(attemptController)attemptController.abort();if(pollController)pollController.abort();
+ if(closed)return;closed=true;if(recoveryController)recoveryController.cancel();rejectRecoveryCommit(failure('network','bridge closed'));if(helloTimer)clearTimeout(helloTimer);helloTimer=null;if(carrierTimer)clearTimeout(carrierTimer);clearProbeTimer();if(schedulerTimer)clearTimeout(schedulerTimer);schedulerTimer=null;if(attemptController)attemptController.abort();if(pollController)pollController.abort();
  if(socket)socket.close();cancelBatch(upLease);releasePending(upPending,null);
  for(const lane of lanes.values()){
   if(lane.controller)lane.controller.abort();cancelBatch(lane.upLease);releasePending(lane.pending,lane);if(lane.socket)lane.socket.close();
@@ -505,6 +514,7 @@ recoveryController=recoverySupport.create({
  restored:finishOldRecovery,replace:replaceCarrier,replaceable:error=>failureReason(error,'network')!=='protocol',
  reason:(error,fallback)=>failureReason(error,fallback),terminal:reason=>fail(recoveryController.remaining()<=0?'timeout':reason)
 });
+armScheduler();
 addEventListener('message',event=>{
  if(event.source!==parent)return;if(initialized){if(event.ports&&event.ports.length===1)event.ports[0].close();return}
  if(event.data===null||typeof event.data!=='object')return;
diff --git a/src/web/bridge/tests.rs b/src/web/bridge/tests.rs
index 4efb271..c2b53b5 100644
--- a/src/web/bridge/tests.rs
+++ b/src/web/bridge/tests.rs
@@ -87,9 +87,10 @@ fn rendered_page_embeds_the_configured_bridge_timing_policy() {
     assert!(page.body.contains("websocketOpenMs=11*1000"));
     assert!(page.body.contains("reconnectGraceMs=119*1000"));
     assert!(page.body.contains("let probeCoalesceMs=4"));
-    assert!(page.body.contains(
-        "helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs)"
-    ));
+    assert!(
+        page.body
+            .contains("helloTimer=setTimeout(()=>fail('timeout'),bridgeRequestMs)")
+    );
 }
 
 #[test]
@@ -143,10 +144,9 @@ fn retry_and_attempt_state_are_frozen_before_fetch() {
             .contains("async function send(path,frozenOptions,remainingBudget,maxAttempts)")
     );
     assert!(!page.body.contains("makeOptions"));
-    assert!(
-        page.body
-            .contains("if(settings.closed()||(external&&external.aborted))throw new Error('request aborted')")
-    );
+    assert!(page.body.contains(
+        "if(settings.closed()||(external&&external.aborted))throw new Error('request aborted')"
+    ));
     assert!(page.body.contains(
         "const frozen=options('POST',bootstrap,snapshot.hello,attemptHeaders(snapshot.attempt,snapshot.failure),controller.signal)"
     ));
@@ -169,5 +169,8 @@ fn ambiguous_commit_is_resolved_before_carrier_advance() {
     );
     assert!(page.body.contains("const token=cleanupToken||sessionToken"));
     assert!(page.body.contains("'X-Carrier-Failure':terminalFailure"));
-    assert!(page.body.contains("addEventListener('pagehide',()=>fail('navigation')"));
+    assert!(
+        page.body
+            .contains("addEventListener('pagehide',()=>fail('navigation')")
+    );
 }
diff --git a/src/web/http.rs b/src/web/http.rs
index ccbe5a1..fcd5c27 100644
--- a/src/web/http.rs
+++ b/src/web/http.rs
@@ -264,16 +264,15 @@ async fn handle_root(
             client_ip,
             user_agent,
         ),
-        recovery::RootRepresentation::Recovery(_) => runtime
-            .issue_recovery_bootstrap_for_request(
-                &generation,
-                Arc::clone(&profile),
-                client_ip,
-                user_agent,
-                recovery_session
-                    .as_ref()
-                    .map(|session| session.trace_session_id()),
-            ),
+        recovery::RootRepresentation::Recovery(_) => runtime.issue_recovery_bootstrap_for_request(
+            &generation,
+            Arc::clone(&profile),
+            client_ip,
+            user_agent,
+            recovery_session
+                .as_ref()
+                .map(|session| session.trace_session_id()),
+        ),
         recovery::RootRepresentation::Invalid => {
             strip_query(&mut request);
             return serve_decoy(request, vhost, true, &runtime).await;
diff --git a/src/web/http/recovery.rs b/src/web/http/recovery.rs
index 1d0f00a..d0cbe4f 100644
--- a/src/web/http/recovery.rs
+++ b/src/web/http/recovery.rs
@@ -52,16 +52,20 @@ pub(super) fn classify(request: &Request<RequestBody>) -> RootRepresentation {
 
 /// Detects a recovery media token even when its Accept syntax is noncanonical.
 pub(super) fn has_media_type(request: &Request<RequestBody>) -> bool {
-    request.headers().get_all(header::ACCEPT).iter().any(|value| {
-        value.to_str().ok().is_some_and(|value| {
-            value.split(',').any(|entry| {
-                entry
-                    .split(';')
-                    .next()
-                    .is_some_and(|media| media.trim().eq_ignore_ascii_case(MEDIA_TYPE))
+    request
+        .headers()
+        .get_all(header::ACCEPT)
+        .iter()
+        .any(|value| {
+            value.to_str().ok().is_some_and(|value| {
+                value.split(',').any(|entry| {
+                    entry
+                        .split(';')
+                        .next()
+                        .is_some_and(|media| media.trim().eq_ignore_ascii_case(MEDIA_TYPE))
+                })
             })
         })
-    })
 }
 
 /// Builds the bounded no-store recovery representation.
diff --git a/src/web/http/recovery_tests.rs b/src/web/http/recovery_tests.rs
index 7027d36..68fb2b3 100644
--- a/src/web/http/recovery_tests.rs
+++ b/src/web/http/recovery_tests.rs
@@ -40,6 +40,21 @@ async fn create_session(
     request(listener, runtime, create).await
 }
 
+async fn send_open(
+    listener: &TcpListener,
+    runtime: &Arc<WebProcessRuntime>,
+    bearer: &str,
+) -> Vec<u8> {
+    let open = frame::encode(FrameType::Open, 1, &[]);
+    let mut request_bytes = format!(
+        "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.40\r\nAuthorization: Bearer {bearer}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        open.len()
+    )
+    .into_bytes();
+    request_bytes.extend_from_slice(&open);
+    request(listener, runtime, request_bytes).await
+}
+
 async fn recover(
     listener: &TcpListener,
     runtime: &Arc<WebProcessRuntime>,
@@ -83,8 +98,14 @@ async fn recovery_retires_current_bearer_before_single_slot_recreation() {
     .await;
     let (recovery_headers, recovery_body) = split_response(&recovery_response);
     assert!(recovery_headers.starts_with(b"HTTP/1.1 200"));
-    assert_eq!(response_header(recovery_headers, "content-type"), RECOVERY_TYPE);
-    assert_eq!(response_header(recovery_headers, "cache-control"), "no-store");
+    assert_eq!(
+        response_header(recovery_headers, "content-type"),
+        RECOVERY_TYPE
+    );
+    assert_eq!(
+        response_header(recovery_headers, "cache-control"),
+        "no-store"
+    );
     assert!(recovery_body.len() <= 1024);
     let document: serde_json::Value = serde_json::from_slice(recovery_body).unwrap();
     assert_eq!(document["v"], 1);
@@ -94,10 +115,9 @@ async fn recovery_retires_current_bearer_before_single_slot_recreation() {
     let recreated = create_session(&listener, &runtime, recovery_bootstrap).await;
     assert!(recreated.starts_with(b"HTTP/1.1 200"));
     assert_eq!(
-        runtime.telemetry().session_close_total(
-            WebCarrier::Https,
-            SessionCloseReason::BridgeRecovery,
-        ),
+        runtime
+            .telemetry()
+            .session_close_total(WebCarrier::Https, SessionCloseReason::BridgeRecovery,),
         1
     );
     assert_eq!(
@@ -136,7 +156,10 @@ async fn recovery_retires_current_bearer_before_single_slot_recreation() {
     .await;
     let (retired_headers, retired_body) = split_response(&retired_recovery);
     assert!(retired_headers.starts_with(b"HTTP/1.1 200"));
-    assert_eq!(response_header(retired_headers, "content-type"), RECOVERY_TYPE);
+    assert_eq!(
+        response_header(retired_headers, "content-type"),
+        RECOVERY_TYPE
+    );
     assert!(serde_json::from_slice::<serde_json::Value>(retired_body).is_ok());
     assert_eq!(
         runtime
@@ -151,6 +174,12 @@ async fn recovery_retires_current_bearer_before_single_slot_recreation() {
     );
 
     runtime.shutdown().await;
+    assert_eq!(
+        runtime
+            .telemetry()
+            .bridge_recovery_total(WebBridgeRecoveryEvent::ClosedBeforeCommit),
+        1
+    );
     generation.stop_sessions().await;
     generation.stop_background_tasks().await;
 }
@@ -167,7 +196,10 @@ async fn malformed_or_over_capacity_recovery_is_indistinguishable_from_decoy() {
     let malformed = recover(&listener, &runtime, &encoded, "Bearer malformed").await;
     let (malformed_headers, malformed_body) = split_response(&malformed);
     assert!(malformed_headers.starts_with(b"HTTP/1.1 200"));
-    assert_eq!(response_header(malformed_headers, "cache-control"), "no-store");
+    assert_eq!(
+        response_header(malformed_headers, "cache-control"),
+        "no-store"
+    );
     assert_eq!(malformed_body, b"<!doctype html><title>decoy</title>");
 
     let invalid_capability = recover(
@@ -200,7 +232,10 @@ async fn malformed_or_over_capacity_recovery_is_indistinguishable_from_decoy() {
     .await;
     let (capacity_headers, capacity_body) = split_response(&over_capacity);
     assert!(capacity_headers.starts_with(b"HTTP/1.1 200"));
-    assert_eq!(response_header(capacity_headers, "cache-control"), "no-store");
+    assert_eq!(
+        response_header(capacity_headers, "cache-control"),
+        "no-store"
+    );
     assert_eq!(capacity_body, b"<!doctype html><title>decoy</title>");
 
     runtime.shutdown().await;
@@ -208,6 +243,52 @@ async fn malformed_or_over_capacity_recovery_is_indistinguishable_from_decoy() {
     generation.stop_background_tasks().await;
 }
 
+#[tokio::test]
+async fn fixed_carrier_recovery_commits_on_real_uplink_progress() {
+    let capability = [43u8; 32];
+    let generation = test_runtime_generation(1, runtime_config(capability, WebCarrier::Https));
+    let active_runtime = Arc::new(ArcSwap::from(Arc::clone(&generation)));
+    let runtime = WebProcessRuntime::start(active_runtime);
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let encoded = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(capability);
+    let bootstrap = bridge_bootstrap(&listener, &runtime, &encoded).await;
+    let created = create_session(&listener, &runtime, &bootstrap).await;
+    let (created_headers, _) = split_response(&created);
+    let old_bearer = response_header(created_headers, "x-session-token");
+    let recovered = recover(
+        &listener,
+        &runtime,
+        &encoded,
+        &format!("Bearer {old_bearer}"),
+    )
+    .await;
+    let (_, recovery_body) = split_response(&recovered);
+    let document: serde_json::Value = serde_json::from_slice(recovery_body).unwrap();
+    let recovery_bootstrap = document["bootstrap"].as_str().unwrap();
+    let recreated = create_session(&listener, &runtime, recovery_bootstrap).await;
+    let (recreated_headers, _) = split_response(&recreated);
+    let bearer = response_header(recreated_headers, "x-session-token");
+
+    let uplink = send_open(&listener, &runtime, bearer).await;
+
+    assert!(uplink.starts_with(b"HTTP/1.1 204"));
+    assert_eq!(
+        runtime
+            .telemetry()
+            .bridge_recovery_total(WebBridgeRecoveryEvent::Committed),
+        1
+    );
+    runtime.shutdown().await;
+    assert_eq!(
+        runtime
+            .telemetry()
+            .bridge_recovery_total(WebBridgeRecoveryEvent::ClosedBeforeCommit),
+        0
+    );
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
+
 #[tokio::test]
 async fn deferred_close_preserves_the_first_reason_across_replacement_cancel() {
     let capability = [42u8; 32];
@@ -224,9 +305,7 @@ async fn deferred_close_preserves_the_first_reason_across_replacement_cancel() {
         .decode(bearer)
         .unwrap();
     let hash: crate::web::manager::TokenHash = Sha256::digest(raw).into();
-    let session = runtime
-        .get_session(hash, "proxy.example.com")
-        .unwrap();
+    let session = runtime.get_session(hash, "proxy.example.com").unwrap();
     let trace_session_id = session.trace_session_id();
 
     assert!(session.begin_carrier_supersede());
@@ -251,10 +330,9 @@ async fn deferred_close_preserves_the_first_reason_across_replacement_cancel() {
         1
     );
     assert_eq!(
-        runtime.telemetry().session_close_total(
-            WebCarrier::Https,
-            SessionCloseReason::CarrierSuperseded,
-        ),
+        runtime
+            .telemetry()
+            .session_close_total(WebCarrier::Https, SessionCloseReason::CarrierSuperseded,),
         0
     );
 
diff --git a/src/web/manager/carrier_learning.rs b/src/web/manager/carrier_learning.rs
index ec63b79..0579e7f 100644
--- a/src/web/manager/carrier_learning.rs
+++ b/src/web/manager/carrier_learning.rs
@@ -265,10 +265,8 @@ impl CarrierLearning {
         {
             return CarrierLearningEpoch::Pending;
         }
-        self.epoch.map_or(
-            CarrierLearningEpoch::Exhausted,
-            CarrierLearningEpoch::Ready,
-        )
+        self.epoch
+            .map_or(CarrierLearningEpoch::Exhausted, CarrierLearningEpoch::Ready)
     }
 
     /// Ranks supported configured candidates without scanning the evidence store.
@@ -482,7 +480,6 @@ impl CarrierLearning {
             entry.update(slot, deltas, cohort);
         }
     }
-
 }
 
 fn supported(configured: &[WebCarrier], request: super::CarrierRequest) -> Vec<WebCarrier> {
diff --git a/src/web/manager/carrier_learning/tests.rs b/src/web/manager/carrier_learning/tests.rs
index 375ec2d..3c328b1 100644
--- a/src/web/manager/carrier_learning/tests.rs
+++ b/src/web/manager/carrier_learning/tests.rs
@@ -352,7 +352,10 @@ fn health_window_change_starts_a_new_empty_epoch() {
 
     assert_ne!(changed.epoch, Some(epoch));
     assert_eq!(
-        changed.detached.as_ref().map(|detached| detached.entries.len()),
+        changed
+            .detached
+            .as_ref()
+            .map(|detached| detached.entries.len()),
         Some(3)
     );
     assert!(learning.entries.is_empty());
@@ -404,8 +407,7 @@ async fn runtime_activation_publishes_matching_policy_and_generation() {
     let mut config = crate::config::ProxyConfig::default();
     config.web.carriers = crate::config::WebCarriers::Enabled(vec![WebCarrier::Websocket]);
     config.web.carrier_learning = true;
-    config.web.carrier_negotiation_aggressiveness =
-        WebCarrierNegotiationAggressiveness::Aggressive;
+    config.web.carrier_negotiation_aggressiveness = WebCarrierNegotiationAggressiveness::Aggressive;
     config.web.timeouts.carrier_learning_secs = 10;
     config.web.timeouts.carrier_health_secs = 3;
     let first = crate::maestro::generation::test_runtime_generation(1, config.clone());
diff --git a/src/web/manager/carrier_outcome.rs b/src/web/manager/carrier_outcome.rs
index d6f237d..7c78ec3 100644
--- a/src/web/manager/carrier_outcome.rs
+++ b/src/web/manager/carrier_outcome.rs
@@ -108,20 +108,15 @@ impl WebProcessRuntime {
                 && entry.carrier_phase == CarrierChainPhase::Provisional
             {
                 entry.carrier_phase = CarrierChainPhase::CommittedPendingHealth;
-                Some((entry.carrier_scores, entry.recovery))
+                Some(entry.carrier_scores)
             } else {
                 None
             }
         });
         drop(state);
-        let Some((scores, recovery)) = outcome else {
+        let Some(scores) = outcome else {
             return false;
         };
-        if recovery {
-            self.telemetry.record_bridge_recovery(
-                crate::web::telemetry::WebBridgeRecoveryEvent::Committed,
-            );
-        }
         self.trace.record_carrier_lifecycle(
             client_ip,
             identity.clone(),
@@ -150,9 +145,7 @@ impl WebProcessRuntime {
         websocket_owner: Option<u64>,
     ) -> CarrierHealthPublicationOutcome {
         if carrier.uses_websocket()
-            && websocket_owner.is_none_or(|owner| {
-                !self.claim_websocket_health(owner, session_hash)
-            })
+            && websocket_owner.is_none_or(|owner| !self.claim_websocket_health(owner, session_hash))
         {
             let outcome = WebCarrierLearningOutcome::OwnerNotLive;
             self.telemetry.record_carrier_learning(carrier, outcome);
@@ -187,13 +180,10 @@ impl WebProcessRuntime {
         };
         let learning_outcome = if let Some(context) = learning_context {
             let now = Instant::now();
-            let outcome = self.learning.lock().record_chain(
-                now,
-                context.epoch,
-                context,
-                &[],
-                carrier,
-            );
+            let outcome =
+                self.learning
+                    .lock()
+                    .record_chain(now, context.epoch, context, &[], carrier);
             match outcome {
                 super::learning::CarrierLearningRecordOutcome::Recorded => {
                     WebCarrierLearningOutcome::Recorded
diff --git a/src/web/manager/credentials.rs b/src/web/manager/credentials.rs
index e66ed62..d5fbe7c 100644
--- a/src/web/manager/credentials.rs
+++ b/src/web/manager/credentials.rs
@@ -319,15 +319,17 @@ impl WebProcessRuntime {
                     .is_some_and(|current| current.token_hash() == hash)
                 {
                     bootstrap.close_requested = true;
-                    failure_phase = Some(if matches!(
-                        bootstrap.carrier_phase,
-                        CarrierChainPhase::CommittedPendingHealth | CarrierChainPhase::Healthy
-                    ) || session.is_carrier_committed()
-                    {
-                        crate::web::telemetry::WebCarrierFailurePhase::Committed
-                    } else {
-                        crate::web::telemetry::WebCarrierFailurePhase::Provisional
-                    });
+                    failure_phase = Some(
+                        if matches!(
+                            bootstrap.carrier_phase,
+                            CarrierChainPhase::CommittedPendingHealth | CarrierChainPhase::Healthy
+                        ) || session.is_carrier_committed()
+                        {
+                            crate::web::telemetry::WebCarrierFailurePhase::Committed
+                        } else {
+                            crate::web::telemetry::WebCarrierFailurePhase::Provisional
+                        },
+                    );
                     break;
                 }
             }
diff --git a/src/web/manager/lifecycle.rs b/src/web/manager/lifecycle.rs
index 9a4aa76..6ce297c 100644
--- a/src/web/manager/lifecycle.rs
+++ b/src/web/manager/lifecycle.rs
@@ -46,7 +46,10 @@ impl WebProcessRuntime {
                 Duration::from_secs(config.web.timeouts.carrier_learning_secs),
                 Duration::from_secs(config.web.timeouts.carrier_health_secs),
             );
-            debug_assert!(outcome.applied, "runtime generations must increase monotonically");
+            debug_assert!(
+                outcome.applied,
+                "runtime generations must increase monotonically"
+            );
             let replaced = self.active_runtime.swap(generation);
             (replaced, outcome.detached)
         };
@@ -68,14 +71,6 @@ impl WebProcessRuntime {
         let Some(session) = state.sessions.remove(&hash) else {
             return;
         };
-        let recovery_closed_before_commit = state.bootstraps.values().any(|bootstrap| {
-            bootstrap.recovery
-                && bootstrap
-                    .session
-                    .as_ref()
-                    .is_some_and(|current| current.token_hash() == hash)
-                && !session.is_carrier_committed()
-        });
         decrement_map(&mut state.sessions_per_ip, &client_ip);
         decrement_map(&mut state.sessions_per_profile, &profile_key);
         remember_closed_token_locked(
@@ -119,11 +114,6 @@ impl WebProcessRuntime {
         }
         self.telemetry
             .record_session_closed(session.carrier(), reason);
-        if recovery_closed_before_commit {
-            self.telemetry.record_bridge_recovery(
-                crate::web::telemetry::WebBridgeRecoveryEvent::ClosedBeforeCommit,
-            );
-        }
         drop(state);
         self.notify_operator_work_changed();
     }
diff --git a/src/web/manager/session_creation.rs b/src/web/manager/session_creation.rs
index 7e473d6..817dbde 100644
--- a/src/web/manager/session_creation.rs
+++ b/src/web/manager/session_creation.rs
@@ -35,6 +35,7 @@ struct Replacement {
     learning_disposition: WebCarrierSelectionDisposition,
     ip_learning_eligible: bool,
     carrier_deadline_at: Instant,
+    recovery: bool,
 }
 
 impl WebProcessRuntime {
@@ -194,6 +195,7 @@ impl WebProcessRuntime {
                 learning_disposition: entry.carrier_learning_disposition,
                 ip_learning_eligible,
                 carrier_deadline_at: entry.carrier_deadline_at.ok_or(ManagerError::Protocol)?,
+                recovery: entry.recovery,
             };
             let _operator_admission = self.try_operator_admission()?;
             state
@@ -213,6 +215,7 @@ impl WebProcessRuntime {
         let trace_session_id = entry.trace_session_id;
         let issued_profile = Arc::clone(&entry.profile);
         let issued_timeouts = entry.timeouts.clone();
+        let recovery = entry.recovery;
         let profile = config
             .web
             .runtime
@@ -232,69 +235,68 @@ impl WebProcessRuntime {
             Duration::from_secs(config.web.timeouts.carrier_learning_secs),
             Duration::from_secs(config.web.timeouts.carrier_health_secs),
         );
-        let (candidates, scores, learning_epoch, learning_disposition) = if capability_selection
-            && profile.carrier_learning
-            && learning_policy.0
-        {
-            let learning = self.learning.lock();
-            match learning.epoch_for_policy(
+        let (candidates, scores, learning_epoch, learning_disposition) =
+            if capability_selection && profile.carrier_learning && learning_policy.0 {
+                let learning = self.learning.lock();
+                match learning.epoch_for_policy(
                     generation.id,
                     learning_policy.0,
                     learning_policy.1,
                     learning_policy.2,
                     learning_policy.3,
                 ) {
-                super::learning::CarrierLearningEpoch::Ready(epoch) => {
-                    let (candidates, scores) = learning.rank(
-                        now,
-                        &profile.carriers,
-                        carrier_request,
-                        profile_key,
-                        client_ip,
-                        ip_learning_eligible,
-                    );
-                    let disposition = if scores.iter().any(|score| *score != 0) {
-                        WebCarrierSelectionDisposition::Applied
-                    } else {
-                        WebCarrierSelectionDisposition::Cold
-                    };
-                    (candidates, scores, Some(epoch), disposition)
+                    super::learning::CarrierLearningEpoch::Ready(epoch) => {
+                        let (candidates, scores) = learning.rank(
+                            now,
+                            &profile.carriers,
+                            carrier_request,
+                            profile_key,
+                            client_ip,
+                            ip_learning_eligible,
+                        );
+                        let disposition = if scores.iter().any(|score| *score != 0) {
+                            WebCarrierSelectionDisposition::Applied
+                        } else {
+                            WebCarrierSelectionDisposition::Cold
+                        };
+                        (candidates, scores, Some(epoch), disposition)
+                    }
+                    super::learning::CarrierLearningEpoch::Pending => (
+                        supported_candidates(&profile.carriers, carrier_request),
+                        [0; 4],
+                        None,
+                        WebCarrierSelectionDisposition::PolicyPending,
+                    ),
+                    super::learning::CarrierLearningEpoch::Exhausted => (
+                        supported_candidates(&profile.carriers, carrier_request),
+                        [0; 4],
+                        None,
+                        WebCarrierSelectionDisposition::EpochExhausted,
+                    ),
                 }
-                super::learning::CarrierLearningEpoch::Pending => (
+            } else if capability_selection {
+                (
                     supported_candidates(&profile.carriers, carrier_request),
                     [0; 4],
                     None,
-                    WebCarrierSelectionDisposition::PolicyPending,
-                ),
-                super::learning::CarrierLearningEpoch::Exhausted => (
-                    supported_candidates(&profile.carriers, carrier_request),
+                    if learning_policy.0 {
+                        WebCarrierSelectionDisposition::ProfileDisabled
+                    } else {
+                        WebCarrierSelectionDisposition::PolicyDisabled
+                    },
+                )
+            } else if carrier_request.uses_capabilities()
+                && !carrier_request.supports(profile.carrier)
+            {
+                return Err(ManagerError::Protocol);
+            } else {
+                (
+                    vec![profile.carrier],
                     [0; 4],
                     None,
-                    WebCarrierSelectionDisposition::EpochExhausted,
-                ),
-            }
-        } else if capability_selection {
-            (
-                supported_candidates(&profile.carriers, carrier_request),
-                [0; 4],
-                None,
-                if learning_policy.0 {
-                    WebCarrierSelectionDisposition::ProfileDisabled
-                } else {
-                    WebCarrierSelectionDisposition::PolicyDisabled
-                },
-            )
-        } else if carrier_request.uses_capabilities() && !carrier_request.supports(profile.carrier)
-        {
-            return Err(ManagerError::Protocol);
-        } else {
-            (
-                vec![profile.carrier],
-                [0; 4],
-                None,
-                WebCarrierSelectionDisposition::ProfileDisabled,
-            )
-        };
+                    WebCarrierSelectionDisposition::ProfileDisabled,
+                )
+            };
         let Some(carrier) = candidates.first().copied() else {
             return Err(ManagerError::Protocol);
         };
@@ -333,20 +335,14 @@ impl WebProcessRuntime {
             carrier_request.class(),
             learning_context,
             carrier_request.is_automatic(),
+            recovery,
             self.limits.clone(),
             issued_timeouts.clone(),
         );
         state.sessions.insert(session_hash, Arc::clone(&session));
         *state.sessions_per_ip.entry(client_ip).or_insert(0) += 1;
         *state.sessions_per_profile.entry(profile_key).or_insert(0) += 1;
-        let (
-            issuance_ip,
-            candidate_count,
-            user_agent,
-            user_agent_id,
-            recovery,
-            predecessor_session_id,
-        ) = {
+        let (issuance_ip, candidate_count, user_agent, user_agent_id, predecessor_session_id) = {
             let entry = state
                 .bootstraps
                 .get_mut(&bootstrap_hash)
@@ -373,7 +369,6 @@ impl WebProcessRuntime {
                 u8::try_from(entry.carrier_candidates.len()).unwrap_or(4),
                 entry.user_agent.clone(),
                 entry.user_agent_id,
-                entry.recovery,
                 entry.predecessor_session_id,
             )
         };
@@ -446,10 +441,7 @@ impl WebProcessRuntime {
     }
 }
 
-fn supported_candidates(
-    configured: &[WebCarrier],
-    request: CarrierRequest,
-) -> Vec<WebCarrier> {
+fn supported_candidates(configured: &[WebCarrier], request: CarrierRequest) -> Vec<WebCarrier> {
     configured
         .iter()
         .copied()
diff --git a/src/web/manager/session_creation/replacement.rs b/src/web/manager/session_creation/replacement.rs
index cdc3581..1af2cef 100644
--- a/src/web/manager/session_creation/replacement.rs
+++ b/src/web/manager/session_creation/replacement.rs
@@ -83,6 +83,7 @@ impl WebProcessRuntime {
             replacement.request.class(),
             learning_context,
             true,
+            replacement.recovery,
             self.limits.clone(),
             replacement.old_session.timeouts().clone(),
         );
@@ -124,9 +125,6 @@ impl WebProcessRuntime {
             self.telemetry.record_bridge_recovery(
                 crate::web::telemetry::WebBridgeRecoveryEvent::SessionCreated,
             );
-            self.telemetry.record_bridge_recovery(
-                crate::web::telemetry::WebBridgeRecoveryEvent::ClosedBeforeCommit,
-            );
         }
         self.telemetry.record_session_closed(
             replacement.old_session.carrier(),
@@ -152,10 +150,8 @@ impl WebProcessRuntime {
         let old_identity = replacement.old_session.trace_identity();
         drop(state);
         supersede.finish();
-        self.telemetry.record_carrier_selection(
-            replacement.carrier,
-            replacement.learning_disposition,
-        );
+        self.telemetry
+            .record_carrier_selection(replacement.carrier, replacement.learning_disposition);
         if let Some(failure) = replacement.request.failure() {
             self.telemetry.record_carrier_failure(
                 replacement.old_session.carrier(),
diff --git a/src/web/manager/status.rs b/src/web/manager/status.rs
index 49622b6..3b6dc64 100644
--- a/src/web/manager/status.rs
+++ b/src/web/manager/status.rs
@@ -510,15 +510,15 @@ impl WebProcessRuntime {
                 .unwrap_or(SessionDetail::Busy);
         }
         let now = Instant::now();
-        state.closed_sessions.get(&trace_session_id).map_or(
-            SessionDetail::NotFound,
-            |closed| SessionDetail::Gone {
+        state
+            .closed_sessions
+            .get(&trace_session_id)
+            .map_or(SessionDetail::NotFound, |closed| SessionDetail::Gone {
                 attempt: closed.attempt,
                 carrier: closed.carrier,
                 reason: closed.reason.as_str(),
                 closed_age_ms: millis(now.saturating_duration_since(closed.closed_at)),
-            },
-        )
+            })
     }
 
     fn row(&self, candidate: Candidate, status: WebSessionStatus) -> SessionRow {
diff --git a/src/web/session.rs b/src/web/session.rs
index 27ae1a5..dd93eb2 100644
--- a/src/web/session.rs
+++ b/src/web/session.rs
@@ -23,6 +23,9 @@ mod backend;
 // Activity clocks separate authenticated peer leases from diagnostic progress.
 mod activity;
 use activity::SessionActivity;
+// Shared state helpers own queue accounting and bounded tombstone updates.
+mod state;
+use state::{inbound_queue_cost, insert_carrier_lane, remember_closed};
 // Downlink queues own cursor replay, flow control, and memory reservations.
 mod downlink;
 // Response ownership keeps detached batches charged until the last body clone drops.
@@ -45,8 +48,8 @@ mod negotiation;
 use negotiation::CarrierHealthPublicationState;
 // Session closure and carrier-attempt transitions share one cancellation boundary.
 mod lifecycle;
-pub(crate) use lifecycle::{SessionCloseOutcome, SessionCloseReason};
 use lifecycle::SessionNegotiationPhase;
+pub(crate) use lifecycle::{SessionCloseOutcome, SessionCloseReason};
 // Uplink batches own exactly-once sequencing and client-frame validation.
 mod uplink;
 
@@ -181,6 +184,7 @@ struct SessionState {
     carrier_health_uplink: bool,
     carrier_health_downlink: bool,
     carrier_commit_published: bool,
+    recovery_committed: bool,
     websocket_carrier_active: bool,
     websocket_commit_ack_pending: bool,
     websocket_commit_ack_owner: Option<u64>,
@@ -205,6 +209,7 @@ pub(crate) struct WebSession {
     carrier_class: CarrierClientClass,
     learning_context: Option<CarrierLearningContext>,
     automatic_carrier: bool,
+    recovery: bool,
     created_at: Instant,
     limits: WebLimitsConfig,
     timeouts: WebTimeoutsConfig,
@@ -249,6 +254,7 @@ impl WebSession {
         carrier_class: CarrierClientClass,
         learning_context: Option<CarrierLearningContext>,
         automatic_carrier: bool,
+        recovery: bool,
         limits: WebLimitsConfig,
         timeouts: WebTimeoutsConfig,
     ) -> Arc<Self> {
@@ -273,6 +279,7 @@ impl WebSession {
             carrier_class,
             learning_context,
             automatic_carrier,
+            recovery,
             created_at,
             limits,
             timeouts,
@@ -305,6 +312,7 @@ impl WebSession {
                 carrier_health_uplink: false,
                 carrier_health_downlink: false,
                 carrier_commit_published: false,
+                recovery_committed: false,
                 websocket_carrier_active: false,
                 websocket_commit_ack_pending: false,
                 websocket_commit_ack_owner: None,
@@ -514,37 +522,3 @@ impl WebSession {
             .map(|manager| manager.budget_notify())
     }
 }
-
-fn inbound_queue_cost(queue: &VecDeque<InboundChunk>) -> (usize, usize) {
-    let bytes = queue.iter().fold(0usize, |total, chunk| {
-        total.saturating_add(chunk.bytes.len().saturating_sub(chunk.offset) + QUEUE_ITEM_COST)
-    });
-    (bytes, queue.len())
-}
-
-fn remember_closed(state: &mut SessionState, stream_id: u32, limit: usize) -> Option<u32> {
-    if !state.closed_streams.insert(stream_id) {
-        return None;
-    }
-    state.closed_order.push_back(stream_id);
-    let mut evicted = None;
-    while state.closed_order.len() > limit {
-        if let Some(oldest) = state.closed_order.pop_front() {
-            state.closed_streams.remove(&oldest);
-            evicted = Some(oldest);
-        }
-    }
-    evicted
-}
-
-fn insert_carrier_lane(state: &mut SessionState, lane_id: u32) -> Option<CarrierLaneIdentity> {
-    if state.carrier_lanes.contains_key(&lane_id) {
-        return None;
-    }
-    let instance = state.next_lane_instance;
-    state.next_lane_instance = instance.checked_add(1)?;
-    state
-        .carrier_lanes
-        .insert(lane_id, CarrierLane::new(instance));
-    Some(CarrierLaneIdentity { lane_id, instance })
-}
diff --git a/src/web/session/backend_tests.rs b/src/web/session/backend_tests.rs
index 9a07c9c..6704c7f 100644
--- a/src/web/session/backend_tests.rs
+++ b/src/web/session/backend_tests.rs
@@ -131,6 +131,7 @@ fn test_runtime_with_dc(
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         limits,
         timeouts,
     );
diff --git a/src/web/session/downlink_tests.rs b/src/web/session/downlink_tests.rs
index 35a0e8f..29bdb58 100644
--- a/src/web/session/downlink_tests.rs
+++ b/src/web/session/downlink_tests.rs
@@ -47,6 +47,7 @@ fn session() -> (Arc<WebSession>, Arc<WebProcessRuntime>) {
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         WebLimitsConfig::default(),
         timeouts,
     );
diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs
index bae0e15..5cfa72e 100644
--- a/src/web/session/lanes.rs
+++ b/src/web/session/lanes.rs
@@ -6,8 +6,8 @@ use tokio::sync::OwnedSemaphorePermit;
 
 use super::lane_downlink::take_lane_down_batch;
 use super::{
-    CarrierLaneIdentity, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionCloseReason,
-    SessionState, WebSession, remember_closed,
+    CarrierLaneIdentity, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame,
+    SessionCloseReason, SessionState, WebSession, remember_closed,
 };
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::ManagerError;
diff --git a/src/web/session/lanes/tests.rs b/src/web/session/lanes/tests.rs
index c1a41a0..a836009 100644
--- a/src/web/session/lanes/tests.rs
+++ b/src/web/session/lanes/tests.rs
@@ -62,6 +62,7 @@ fn new_session_with_automatic(
         },
         None,
         automatic,
+        false,
         limits,
         WebTimeoutsConfig::default(),
     )
diff --git a/src/web/session/lifecycle.rs b/src/web/session/lifecycle.rs
index aaa2cd6..f3f21ca 100644
--- a/src/web/session/lifecycle.rs
+++ b/src/web/session/lifecycle.rs
@@ -97,6 +97,7 @@ struct ReleasedQueues {
     control_bytes: usize,
     control_items: usize,
     closed_before_health: bool,
+    recovery_closed_before_commit: bool,
     reason: SessionCloseReason,
     peer_gap: Duration,
 }
@@ -253,6 +254,7 @@ impl WebSession {
         let closed_before_health = self.automatic_carrier
             && state.negotiation_phase == SessionNegotiationPhase::Committed
             && self.reject_carrier_health_on_close();
+        let recovery_closed_before_commit = self.recovery && !state.recovery_committed;
         state.close_requested = Some(reason);
         state.closed = true;
         if reason == SessionCloseReason::CarrierSuperseded {
@@ -305,6 +307,7 @@ impl WebSession {
             control_bytes,
             control_items,
             closed_before_health,
+            recovery_closed_before_commit,
             reason,
             peer_gap,
         }
@@ -326,6 +329,11 @@ impl WebSession {
                     crate::web::telemetry::WebCarrierLearningOutcome::ClosedBeforeHealth,
                 );
             }
+            if released.recovery_closed_before_commit {
+                manager.telemetry().record_bridge_recovery(
+                    crate::web::telemetry::WebBridgeRecoveryEvent::ClosedBeforeCommit,
+                );
+            }
             manager.release_pending(
                 self.profile_key,
                 released.data_bytes,
@@ -350,10 +358,7 @@ impl WebSession {
                     Some(released.reason.as_str()),
                     crate::web::trace::TraceLifecycleContext {
                         peer_gap_ms: Some(
-                            released
-                                .peer_gap
-                                .as_millis()
-                                .min(u128::from(u64::MAX)) as u64,
+                            released.peer_gap.as_millis().min(u128::from(u64::MAX)) as u64
                         ),
                         predecessor_session_id: None,
                     },
diff --git a/src/web/session/negotiation.rs b/src/web/session/negotiation.rs
index 8ca9a15..dd2a96d 100644
--- a/src/web/session/negotiation.rs
+++ b/src/web/session/negotiation.rs
@@ -107,6 +107,14 @@ impl WebSession {
         state: &mut SessionState,
         progress: AppliedProgress,
     ) -> (bool, Option<CarrierHealthClaim>) {
+        if self.recovery && progress.any() && !state.recovery_committed {
+            state.recovery_committed = true;
+            if let Some(manager) = self.manager.upgrade() {
+                manager.telemetry().record_bridge_recovery(
+                    crate::web::telemetry::WebBridgeRecoveryEvent::Committed,
+                );
+            }
+        }
         if !self.automatic_carrier || !progress.any() {
             return (false, None);
         }
@@ -331,6 +339,7 @@ mod tests {
             CarrierClientClass::Bridge,
             None,
             true,
+            false,
             WebLimitsConfig::default(),
             WebTimeoutsConfig::default(),
         )
@@ -371,9 +380,17 @@ mod tests {
         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());
+        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());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_some()
+        );
     }
 
     #[test]
@@ -391,9 +408,17 @@ mod tests {
         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());
+        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());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_some()
+        );
     }
 
     #[test]
@@ -407,7 +432,11 @@ mod tests {
         state.carrier_health_downlink = true;
         state.carrier_health_activity_at = Some(now);
 
-        assert!(session.carrier_health_ready_locked(&mut state, now).is_none());
+        assert!(
+            session
+                .carrier_health_ready_locked(&mut state, now)
+                .is_none()
+        );
         assert_eq!(
             session.carrier_health_publication_state(),
             CarrierHealthPublicationState::Awaiting
@@ -421,8 +450,16 @@ mod tests {
         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());
+        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(),
@@ -469,8 +506,7 @@ mod tests {
 
             assert!(matches!(
                 session.carrier_health_publication_state(),
-                CarrierHealthPublicationState::Published
-                    | CarrierHealthPublicationState::Rejected
+                CarrierHealthPublicationState::Published | CarrierHealthPublicationState::Rejected
             ));
             assert!(!session.publish_carrier_health());
             assert_eq!(
diff --git a/src/web/session/state.rs b/src/web/session/state.rs
new file mode 100644
index 0000000..0707e56
--- /dev/null
+++ b/src/web/session/state.rs
@@ -0,0 +1,44 @@
+use std::collections::VecDeque;
+
+use super::{CarrierLane, CarrierLaneIdentity, InboundChunk, QUEUE_ITEM_COST, SessionState};
+
+pub(super) fn inbound_queue_cost(queue: &VecDeque<InboundChunk>) -> (usize, usize) {
+    let bytes = queue.iter().fold(0usize, |total, chunk| {
+        total.saturating_add(chunk.bytes.len().saturating_sub(chunk.offset) + QUEUE_ITEM_COST)
+    });
+    (bytes, queue.len())
+}
+
+pub(super) fn remember_closed(
+    state: &mut SessionState,
+    stream_id: u32,
+    limit: usize,
+) -> Option<u32> {
+    if !state.closed_streams.insert(stream_id) {
+        return None;
+    }
+    state.closed_order.push_back(stream_id);
+    let mut evicted = None;
+    while state.closed_order.len() > limit {
+        if let Some(oldest) = state.closed_order.pop_front() {
+            state.closed_streams.remove(&oldest);
+            evicted = Some(oldest);
+        }
+    }
+    evicted
+}
+
+pub(super) fn insert_carrier_lane(
+    state: &mut SessionState,
+    lane_id: u32,
+) -> Option<CarrierLaneIdentity> {
+    if state.carrier_lanes.contains_key(&lane_id) {
+        return None;
+    }
+    let instance = state.next_lane_instance;
+    state.next_lane_instance = instance.checked_add(1)?;
+    state
+        .carrier_lanes
+        .insert(lane_id, CarrierLane::new(instance));
+    Some(CarrierLaneIdentity { lane_id, instance })
+}
diff --git a/src/web/session/uplink_tests.rs b/src/web/session/uplink_tests.rs
index 0ecad83..ed09655 100644
--- a/src/web/session/uplink_tests.rs
+++ b/src/web/session/uplink_tests.rs
@@ -45,6 +45,7 @@ fn session_with_automatic(automatic: bool) -> Arc<WebSession> {
         },
         None,
         automatic,
+        false,
         WebLimitsConfig::default(),
         WebTimeoutsConfig::default(),
     )
diff --git a/src/web/session/websocket/tests.rs b/src/web/session/websocket/tests.rs
index af05cfe..f35c043 100644
--- a/src/web/session/websocket/tests.rs
+++ b/src/web/session/websocket/tests.rs
@@ -18,7 +18,8 @@ struct TestRuntime {
 
 impl TestRuntime {
     async fn shutdown(self) {
-        self.session.close(super::super::SessionCloseReason::ApiClose);
+        self.session
+            .close(super::super::SessionCloseReason::ApiClose);
         self.session.wait().await;
         self.manager.shutdown().await;
         self.generation.stop_sessions().await;
@@ -71,6 +72,7 @@ fn runtime(admission: bool) -> TestRuntime {
         crate::web::manager::CarrierClientClass::Legacy,
         None,
         false,
+        false,
         limits,
         timeouts,
     );
diff --git a/src/web/telemetry.rs b/src/web/telemetry.rs
index fbce47d..4310630 100644
--- a/src/web/telemetry.rs
+++ b/src/web/telemetry.rs
@@ -5,17 +5,17 @@ use std::time::Instant;
 use serde::Serialize;
 
 mod carrier;
+use carrier::{CARRIER_FAILURE_SLOTS, CARRIER_LEARNING_SLOTS, CARRIER_SELECTION_SLOTS};
 pub(crate) use carrier::{
     WebCarrierFailureCounter, WebCarrierFailurePhase, WebCarrierLearningCounter,
     WebCarrierLearningOutcome, WebCarrierSelectionCounter, WebCarrierSelectionDisposition,
 };
-use carrier::{CARRIER_FAILURE_SLOTS, CARRIER_LEARNING_SLOTS, CARRIER_SELECTION_SLOTS};
 mod lifecycle;
+use lifecycle::{SESSION_CLOSE_SLOTS, SESSION_OBSERVATION_SLOTS};
 pub(crate) use lifecycle::{
     WebBridgeRecoveryCounter, WebBridgeRecoveryEvent, WebSessionCloseCounter,
     WebSessionLifecycleObservation, WebSessionLifecycleObservationCounter,
 };
-use lifecycle::{SESSION_CLOSE_SLOTS, SESSION_OBSERVATION_SLOTS};
 
 const LAST_DECOY_OUTCOME_BITS: u32 = 4;
 const LAST_DECOY_OUTCOME_MASK: u64 = (1 << LAST_DECOY_OUTCOME_BITS) - 1;
diff --git a/src/web/telemetry/carrier.rs b/src/web/telemetry/carrier.rs
index ccffc2f..8422232 100644
--- a/src/web/telemetry/carrier.rs
+++ b/src/web/telemetry/carrier.rs
@@ -135,9 +135,8 @@ impl WebCarrierLearningOutcome {
 
 pub(super) const CARRIER_SELECTION_SLOTS: usize =
     WebCarrier::ALL.len() * WebCarrierSelectionDisposition::ALL.len();
-pub(super) const CARRIER_FAILURE_SLOTS: usize = WebCarrier::ALL.len()
-    * WebCarrierFailurePhase::ALL.len()
-    * CarrierFailure::ALL.len();
+pub(super) const CARRIER_FAILURE_SLOTS: usize =
+    WebCarrier::ALL.len() * WebCarrierFailurePhase::ALL.len() * CarrierFailure::ALL.len();
 pub(super) const CARRIER_LEARNING_SLOTS: usize =
     WebCarrier::ALL.len() * WebCarrierLearningOutcome::ALL.len();
 
@@ -238,16 +237,18 @@ impl WebTelemetry {
         WebCarrier::ALL
             .into_iter()
             .flat_map(|carrier| {
-                WebCarrierFailurePhase::ALL.into_iter().flat_map(move |phase| {
-                    CarrierFailure::ALL
-                        .into_iter()
-                        .map(move |reason| WebCarrierFailureCounter {
-                            carrier: carrier.as_str(),
-                            phase: phase.as_str(),
-                            reason: reason.as_str(),
-                            total: self.carrier_failure_total(carrier, phase, reason),
+                WebCarrierFailurePhase::ALL
+                    .into_iter()
+                    .flat_map(move |phase| {
+                        CarrierFailure::ALL.into_iter().map(move |reason| {
+                            WebCarrierFailureCounter {
+                                carrier: carrier.as_str(),
+                                phase: phase.as_str(),
+                                reason: reason.as_str(),
+                                total: self.carrier_failure_total(carrier, phase, reason),
+                            }
                         })
-                })
+                    })
             })
             .collect()
     }
diff --git a/src/web/telemetry/lifecycle.rs b/src/web/telemetry/lifecycle.rs
index 331e0be..c053f2a 100644
--- a/src/web/telemetry/lifecycle.rs
+++ b/src/web/telemetry/lifecycle.rs
@@ -7,8 +7,7 @@ use crate::web::session::SessionCloseReason;
 
 use super::WebTelemetry;
 
-pub(super) const SESSION_CLOSE_SLOTS: usize =
-    WebCarrier::ALL.len() * SessionCloseReason::ALL.len();
+pub(super) const SESSION_CLOSE_SLOTS: usize = WebCarrier::ALL.len() * SessionCloseReason::ALL.len();
 pub(super) const SESSION_OBSERVATION_SLOTS: usize =
     WebCarrier::ALL.len() * WebSessionLifecycleObservation::ALL.len();
 
@@ -111,10 +110,7 @@ pub(crate) struct WebBridgeRecoveryCounter {
     pub(crate) total: u64,
 }
 
-pub(super) const fn session_close_slot(
-    carrier: WebCarrier,
-    reason: SessionCloseReason,
-) -> usize {
+pub(super) const fn session_close_slot(carrier: WebCarrier, reason: SessionCloseReason) -> usize {
     carrier.index() * SessionCloseReason::ALL.len() + reason as usize
 }
 
@@ -131,14 +127,9 @@ pub(super) fn load(counter: &AtomicU64) -> u64 {
 
 impl WebTelemetry {
     /// Records one closed session incarnation and its exact terminal cause.
-    pub(crate) fn record_session_closed(
-        &self,
-        carrier: WebCarrier,
-        reason: SessionCloseReason,
-    ) {
+    pub(crate) fn record_session_closed(&self, carrier: WebCarrier, reason: SessionCloseReason) {
         self.sessions_closed.fetch_add(1, Ordering::Relaxed);
-        self.session_closures[session_close_slot(carrier, reason)]
-            .fetch_add(1, Ordering::Relaxed);
+        self.session_closures[session_close_slot(carrier, reason)].fetch_add(1, Ordering::Relaxed);
     }
 
     /// Returns one fixed session-close counter.
@@ -155,13 +146,13 @@ impl WebTelemetry {
         WebCarrier::ALL
             .into_iter()
             .flat_map(|carrier| {
-                SessionCloseReason::ALL.into_iter().map(move |reason| {
-                    WebSessionCloseCounter {
+                SessionCloseReason::ALL
+                    .into_iter()
+                    .map(move |reason| WebSessionCloseCounter {
                         carrier: carrier.as_str(),
                         reason: reason.as_str(),
                         total: self.session_close_total(carrier, reason),
-                    }
-                })
+                    })
             })
             .collect()
     }
@@ -209,11 +200,7 @@ impl WebTelemetry {
     }
 
     /// Adds a bounded batch of identical recovery milestones.
-    pub(crate) fn record_bridge_recovery_count(
-        &self,
-        event: WebBridgeRecoveryEvent,
-        count: usize,
-    ) {
+    pub(crate) fn record_bridge_recovery_count(&self, event: WebBridgeRecoveryEvent, count: usize) {
         self.bridge_recovery_events[event as usize]
             .fetch_add(u64::try_from(count).unwrap_or(u64::MAX), Ordering::Relaxed);
     }
diff --git a/src/web/telemetry/tests.rs b/src/web/telemetry/tests.rs
index eb031a9..30f7587 100644
--- a/src/web/telemetry/tests.rs
+++ b/src/web/telemetry/tests.rs
@@ -11,26 +11,23 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() {
     telemetry.record_rejection(WebRejectionReason::HttpConnectionCapacity);
     telemetry.record_overload(WebHttpConnectionOverloadOutcome::Dropped);
     telemetry.record_decoy(WebDecoyUpstreamOutcome::ConnectRefused);
-    telemetry.record_carrier_selection(
-        WebCarrier::Https,
-        WebCarrierSelectionDisposition::Cold,
-    );
+    telemetry.record_carrier_selection(WebCarrier::Https, WebCarrierSelectionDisposition::Cold);
     telemetry.record_carrier_failure(
         WebCarrier::Https,
         WebCarrierFailurePhase::Provisional,
         CarrierFailure::Network,
     );
-    telemetry.record_carrier_learning(
-        WebCarrier::Https,
-        WebCarrierLearningOutcome::Recorded,
-    );
+    telemetry.record_carrier_learning(WebCarrier::Https, WebCarrierLearningOutcome::Recorded);
     telemetry.record_session_closed(WebCarrier::Https, SessionCloseReason::ApiClose);
     telemetry.record_session_observation(
         WebCarrier::Https,
         WebSessionLifecycleObservation::RequestAfterClose,
     );
     telemetry.record_bridge_recovery(WebBridgeRecoveryEvent::BootstrapIssued);
-    assert_eq!(telemetry.rejection_counters().len(), WebRejectionReason::ALL.len());
+    assert_eq!(
+        telemetry.rejection_counters().len(),
+        WebRejectionReason::ALL.len()
+    );
     assert_eq!(
         telemetry.overload_counters().len(),
         WebHttpConnectionOverloadOutcome::ALL.len()
@@ -45,9 +42,7 @@ fn fixed_counter_sets_and_acceptor_guard_are_exact() {
     );
     assert_eq!(
         telemetry.carrier_failure_counters().len(),
-        WebCarrier::ALL.len()
-            * WebCarrierFailurePhase::ALL.len()
-            * CarrierFailure::ALL.len()
+        WebCarrier::ALL.len() * WebCarrierFailurePhase::ALL.len() * CarrierFailure::ALL.len()
     );
     assert_eq!(
         telemetry.carrier_learning_counters().len(),