From 90c0d65e1b255cd61ba3079cff1d1c87836564d4 Mon Sep 17 00:00:00 2001
From: Alexey <247128645+axkurcom@users.noreply.github.com>
Date: Sun, 23 Aug 2026 09:04:53 +0300
Subject: [PATCH] WEB Carrier: https-lanes

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>
---
 src/config/load/runtime_web.rs                |   1 +
 src/config/load/strict_keys.rs                |   2 +-
 src/config/load/validate_web.rs               |   5 +
 .../tests/load_basic_tests/web_tests.rs       |  20 +-
 src/config/types.rs                           |   2 +-
 src/config/types/web.rs                       |  26 +
 src/web/bridge.rs                             | 115 +++-
 src/web/frame.rs                              |   2 +
 src/web/http.rs                               |  63 ++-
 src/web/http/tests.rs                         |  89 ++-
 src/web/manager.rs                            |  24 +-
 src/web/manager/state.rs                      |   1 +
 src/web/session.rs                            |  64 ++-
 src/web/session/backend.rs                    |  12 +-
 src/web/session/downlink.rs                   |  29 +
 src/web/session/lanes.rs                      | 525 ++++++++++++++++++
 src/web/session/lanes/tests.rs                | 129 +++++
 src/web/session/uplink.rs                     |  28 +-
 18 files changed, 1063 insertions(+), 74 deletions(-)
 create mode 100644 src/web/session/lanes.rs
 create mode 100644 src/web/session/lanes/tests.rs

diff --git a/src/config/load/runtime_web.rs b/src/config/load/runtime_web.rs
index 24be591..a85175c 100644
--- a/src/config/load/runtime_web.rs
+++ b/src/config/load/runtime_web.rs
@@ -62,6 +62,7 @@ pub(super) fn rebuild(config: &mut ProxyConfig) -> Result<()> {
                 public_addr: vhost.public_addr,
                 user: profile.user.clone(),
                 secret_mode: profile.secret_mode,
+                carrier: config.web.carrier,
                 capability,
                 max_sessions: profile
                     .max_sessions
diff --git a/src/config/load/strict_keys.rs b/src/config/load/strict_keys.rs
index 807de7f..d047920 100644
--- a/src/config/load/strict_keys.rs
+++ b/src/config/load/strict_keys.rs
@@ -259,7 +259,7 @@ const LISTENER_CONFIG_KEYS: &[&str] = &[
     "web_trusted_proxy_cidrs",
 ];
 
-const WEB_CONFIG_KEYS: &[&str] = &["enabled", "limits", "timeouts", "vhosts"];
+const WEB_CONFIG_KEYS: &[&str] = &["enabled", "carrier", "limits", "timeouts", "vhosts"];
 
 const WEB_LIMITS_CONFIG_KEYS: &[&str] = &[
     "max_header_bytes",
diff --git a/src/config/load/validate_web.rs b/src/config/load/validate_web.rs
index 1f64a99..ba202f6 100644
--- a/src/config/load/validate_web.rs
+++ b/src/config/load/validate_web.rs
@@ -59,6 +59,11 @@ pub(super) fn validate(config: &mut ProxyConfig) -> Result<()> {
     }
 
     validate_limits(&config.web.limits)?;
+    if config.web.carrier == WebCarrier::HttpsLanes
+        && config.web.limits.max_http_handlers < 2
+    {
+        return config_error("web.carrier=https-lanes requires web.limits.max_http_handlers >= 2");
+    }
     validate_timeouts(&config.web.timeouts)?;
     validate_vhosts(config)?;
     Ok(())
diff --git a/src/config/tests/load_basic_tests/web_tests.rs b/src/config/tests/load_basic_tests/web_tests.rs
index b781601..e803cda 100644
--- a/src/config/tests/load_basic_tests/web_tests.rs
+++ b/src/config/tests/load_basic_tests/web_tests.rs
@@ -14,6 +14,7 @@ web_trusted_proxy_cidrs = ["127.0.0.1/32"]
 
 [web]
 enabled = true
+carrier = "https-lanes"
 
 [[web.vhosts]]
 host = "Proxy.Example.COM"
@@ -42,11 +43,22 @@ fn web_config_builds_canonical_runtime_snapshot() {
     assert_eq!(vhost.profiles.len(), 1);
     assert_eq!(vhost.profiles[0].user, "alice");
     assert_eq!(vhost.profiles[0].secret_mode, WebSecretMode::Dd);
+    assert_eq!(vhost.profiles[0].carrier, WebCarrier::HttpsLanes);
     assert_eq!(vhost.profiles[0].max_sessions, 4);
     assert_eq!(vhost.profiles[0].max_streams, 64);
     assert_eq!(vhost.profiles[0].max_streams_per_session, 16);
 }
 
+#[test]
+fn https_lanes_requires_separate_poll_and_control_handler_capacity() {
+    let invalid = WEB_CONFIG.replace(
+        "carrier = \"https-lanes\"",
+        "carrier = \"https-lanes\"\n\n[web.limits]\nmax_http_handlers = 1\nmax_body_readers = 1",
+    );
+    let error = load_config_error_from_temp_toml(&invalid);
+    assert!(error.contains("web.carrier=https-lanes requires"));
+}
+
 #[test]
 fn web_listener_requires_an_explicit_trusted_proxy() {
     let invalid = WEB_CONFIG.replace(
@@ -60,8 +72,8 @@ fn web_listener_requires_an_explicit_trusted_proxy() {
 #[test]
 fn web_queue_limits_preserve_control_and_uplink_progress() {
     let invalid = WEB_CONFIG.replace(
-        "[web]\nenabled = true",
-        "[web]\nenabled = true\n\n[web.limits]\ncontrol_bytes_per_session = 1",
+        "carrier = \"https-lanes\"",
+        "carrier = \"https-lanes\"\n\n[web.limits]\ncontrol_bytes_per_session = 1",
     );
     let error = load_config_error_from_temp_toml(&invalid);
     assert!(error.contains("control reserves must cover bounded control frames"));
@@ -70,9 +82,9 @@ fn web_queue_limits_preserve_control_and_uplink_progress() {
 #[test]
 fn web_semaphore_limits_are_rejected_before_runtime_construction() {
     let invalid = WEB_CONFIG.replace(
-        "[web]\nenabled = true",
+        "carrier = \"https-lanes\"",
         &format!(
-            "[web]\nenabled = true\n\n[web.limits]\nmax_http_connections = {}",
+            "carrier = \"https-lanes\"\n\n[web.limits]\nmax_http_connections = {}",
             tokio::sync::Semaphore::MAX_PERMITS + 1,
         ),
     );
diff --git a/src/config/types.rs b/src/config/types.rs
index ecdea93..89f9be7 100644
--- a/src/config/types.rs
+++ b/src/config/types.rs
@@ -49,7 +49,7 @@ pub use server::{
 };
 #[allow(unused_imports)]
 pub use web::{
-    WebConfig, WebDecoyConfig, WebLimitsConfig, WebProfileConfig, WebSecretMode,
+    WebCarrier, WebConfig, WebDecoyConfig, WebLimitsConfig, WebProfileConfig, WebSecretMode,
     WebTimeoutsConfig, WebVhostConfig,
 };
 pub(crate) use web::{
diff --git a/src/config/types/web.rs b/src/config/types/web.rs
index c246b58..d8396a5 100644
--- a/src/config/types/web.rs
+++ b/src/config/types/web.rs
@@ -16,6 +16,27 @@ pub enum WebSecretMode {
     Dd,
 }
 
+/// HTTP carrier selected for newly issued WEB bridge sessions.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum WebCarrier {
+    /// Serialize all logical streams through one uplink and one downlink sequence.
+    #[default]
+    Https,
+    /// Give every logical stream independent HTTPS sequencing and polling state.
+    HttpsLanes,
+}
+
+impl WebCarrier {
+    /// Returns the exact carrier token advertised to the browser bridge.
+    pub(crate) const fn as_str(self) -> &'static str {
+        match self {
+            Self::Https => "https",
+            Self::HttpsLanes => "https-lanes",
+        }
+    }
+}
+
 /// One access user explicitly exposed through a WEB virtual host.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct WebProfileConfig {
@@ -273,6 +294,9 @@ pub struct WebConfig {
     /// Enables issuance of new WEB bridge and session credentials.
     #[serde(default)]
     pub enabled: bool,
+    /// Carrier selected for newly issued WEB bridge sessions.
+    #[serde(default)]
+    pub carrier: WebCarrier,
     /// Hard process and protocol limits.
     #[serde(default)]
     pub limits: WebLimitsConfig,
@@ -320,6 +344,8 @@ pub(crate) struct WebRuntimeProfile {
     pub(crate) user: String,
     /// Client secret representation and inner protocol policy.
     pub(crate) secret_mode: WebSecretMode,
+    /// Carrier frozen into bridge and session state at issuance time.
+    pub(crate) carrier: WebCarrier,
     /// HMAC-derived bridge capability.
     pub(crate) capability: [u8; 32],
     /// Per-profile live session ceiling.
diff --git a/src/web/bridge.rs b/src/web/bridge.rs
index 575715b..e640ef4 100644
--- a/src/web/bridge.rs
+++ b/src/web/bridge.rs
@@ -1,5 +1,6 @@
 use base64::Engine as _;
 
+use crate::config::WebCarrier;
 use crate::crypto::SecureRandom;
 
 /// Browser security policy for the transient Telegram Desktop bridge page.
@@ -13,13 +14,14 @@ pub(crate) struct BridgePage {
     pub(crate) content_security_policy: String,
 }
 
-/// Renders the HTTPS-only WEB carrier bridge with a fresh CSP nonce.
+/// Renders the selected HTTPS WEB carrier bridge with a fresh CSP nonce.
 pub(crate) fn render(
     host: &str,
     bootstrap: &str,
     batch_limit: usize,
     queue_limit: usize,
     queue_items: usize,
+    carrier: WebCarrier,
     rng: &SecureRandom,
 ) -> BridgePage {
     let mut nonce = [0u8; 18];
@@ -31,7 +33,8 @@ pub(crate) fn render(
         .replace("__BOOTSTRAP__", bootstrap)
         .replace("__BATCH_LIMIT__", &batch_limit.to_string())
         .replace("__QUEUE_LIMIT__", &queue_limit.to_string())
-        .replace("__QUEUE_ITEMS__", &queue_items.to_string());
+        .replace("__QUEUE_ITEMS__", &queue_items.to_string())
+        .replace("__CARRIER__", carrier.as_str());
     BridgePage {
         body,
         content_security_policy: format!(
@@ -51,24 +54,26 @@ const DOCUMENT: &str = r##"<!doctype html>
 <script nonce="__NONCE__">
 (()=>{
 'use strict';
-const relayOrigin='https://__HOST__',bootstrap='__BOOTSTRAP__';
+const relayOrigin='https://__HOST__',bootstrap='__BOOTSTRAP__',carrier='__CARRIER__';
 const batchLimit=__BATCH_LIMIT__,queueLimit=__QUEUE_LIMIT__,queueItemLimit=__QUEUE_ITEMS__;
+const laneQueueLimit=Math.min(queueLimit,8388608),laneItemLimit=Math.min(queueItemLimit,1024),closedLaneLimit=4096;
 const fragment=location.hash,androidNonce=/^#android=([A-Za-z0-9_-]{43})$/.exec(fragment)?.[1]||'';
 history.replaceState(null,'',location.pathname);
 let initialized=false,closed=false,port=null,sessionToken='',createStarted=false;
 let queuedBytes=0,queuedItems=0,upSequence=1,downCursor='0',upRunning=false,pollController=null;
-const pending=[],upPending=[];
+const pending=[],upPending=[],lanes=new Map(),closedLanes=new Set(),closedLaneOrder=[];
 const status=state=>{if(port&&!closed)port.postMessage({t:'status',state})};
 const pause=milliseconds=>new Promise(resolve=>setTimeout(resolve,milliseconds));
 const options=(method,token,body,headers,signal,keepalive)=>({
  method,body,signal,keepalive:!!keepalive,mode:'same-origin',credentials:'omit',cache:'no-store',redirect:'error',referrerPolicy:'no-referrer',
  headers:Object.assign(token?{Authorization:'Bearer '+token}:{},body?{'Content-Type':'application/octet-stream'}:{},headers||{})
 });
-function reserve(data){
+function reserve(data,lane){
  if(!data.byteLength||data.byteLength>queueLimit-queuedBytes||queuedItems>=queueItemLimit)return false;
- queuedBytes+=data.byteLength;queuedItems++;return true;
+ if(lane&&(data.byteLength>laneQueueLimit-lane.bytes||lane.items>=laneItemLimit))return false;
+ queuedBytes+=data.byteLength;queuedItems++;if(lane){lane.bytes+=data.byteLength;lane.items++}return true;
 }
-function release(bytes,items){queuedBytes-=bytes;queuedItems-=items}
+function release(bytes,items,lane){queuedBytes-=bytes;queuedItems-=items;if(lane){lane.bytes-=bytes;lane.items-=items}}
 function frameBound(value,maxFrames,maxBytes){
  const view=new DataView(value);let offset=0,frames=0;
  while(offset<value.byteLength){
@@ -85,19 +90,20 @@ function splitFrames(value){
  const view=new DataView(value),result=[];let offset=0;
  while(offset<value.byteLength){
   if(value.byteLength-offset<8||result.length>=4096)throw new Error('invalid frame batch');
+  const type=view.getUint8(offset),id=(view.getUint8(offset+1)<<16)|(view.getUint8(offset+2)<<8)|view.getUint8(offset+3);
   const size=view.getUint32(offset+4),end=offset+8+size;
-  if(size>1048576||end>value.byteLength)throw new Error('invalid frame');
-  result.push(offset===0&&end===value.byteLength?value:value.slice(offset,end));offset=end;
+  if((type===2&&!size)||size>1048576||end>value.byteLength)throw new Error('invalid frame');
+  result.push({type,id,data:offset===0&&end===value.byteLength?value:value.slice(offset,end)});offset=end;
  }
  if(!result.length)throw new Error('empty frame batch');return result;
 }
-function joinPending(values){
+function joinPending(values,lane){
  let total=0,count=0,frames=0;
  while(count<values.length){
   const bound=frameBound(values[count],4096,batchLimit),whole=bound.bytes===values[count].byteLength;
   if(count===0&&!whole){
    const head=new Uint8Array(values[0],0,bound.bytes).slice();
-   values[0]=values[0].slice(bound.bytes);queuedItems++;
+   values[0]=values[0].slice(bound.bytes);queuedItems++;if(lane)lane.items++;
    return {body:head.buffer,total:bound.bytes,count:1};
   }
   if(count&&(total+values[count].byteLength>batchLimit||frames+bound.frames>4096))break;
@@ -136,7 +142,7 @@ async function createSession(first){
  try{
   status('connecting');
   const response=await request('/api/v1/session',()=>options('POST',bootstrap,first));
-  if(response.status!==200||response.headers.get('X-Carrier-Mode')!=='https')throw new Error('session rejected');
+  if(response.status!==200||response.headers.get('X-Carrier-Mode')!==carrier)throw new Error('session rejected');
   sessionToken=response.headers.get('X-Session-Token')||'';downCursor=response.headers.get('X-Down-Cursor')||'0';
   if(!/^[A-Za-z0-9_-]{43}$/.test(sessionToken)||downCursor!=='0')throw new Error('invalid session metadata');
   if(closed){deleteSession();return}
@@ -144,19 +150,23 @@ async function createSession(first){
   const welcomeBytes=new Uint8Array(welcome);
   if(welcomeBytes.length!==8||welcomeBytes[0]!==17||welcomeBytes.slice(1).some(value=>value!==0))throw new Error('invalid welcome');
   port.postMessage(welcome,[welcome]);status('connected');
-  for(const data of pending.splice(0)){release(data.byteLength,1);queueUp(data)}
-  poll();
+  if(carrier==='https-lanes')ensureLane(0);
+  for(const data of pending.splice(0)){release(data.byteLength,1,null);queueCarrier(data)}
+  if(carrier==='https')poll();else pollLane(lanes.get(0));
  }catch(error){fail()}
 }
-function queueUp(data){if(!reserve(data)){fail();return}upPending.push(data);runUp()}
+function queueCarrier(data){
+ try{if(carrier==='https')queueUp(data);else for(const value of splitFrames(data))queueLane(value)}catch(error){fail()}
+}
+function queueUp(data){if(!reserve(data,null)){fail();return}upPending.push(data);runUp()}
 async function runUp(){
  if(upRunning)return;upRunning=true;
  try{
   while(!closed&&sessionToken&&upPending.length){
-   const batch=joinPending(upPending),sequence=String(upSequence);
+   const batch=joinPending(upPending,null),sequence=String(upSequence);
    const response=await request('/api/v1/up',()=>options('POST',sessionToken,batch.body,{'X-Up-Seq':sequence}));
    if(response.status!==204||response.headers.get('X-Up-Ack')!==sequence)throw new Error('uplink rejected');
-   release(batch.total,batch.count);port.postMessage({t:'traffic',up:batch.total,down:0});upSequence++;
+   release(batch.total,batch.count,null);port.postMessage({t:'traffic',up:batch.total,down:0});upSequence++;
   }
  }catch(error){fail()}
  finally{upRunning=false;if(!closed&&sessionToken&&upPending.length)runUp()}
@@ -175,20 +185,78 @@ async function poll(){
   }catch(error){if(!closed)fail();return}
  }
 }
+function ensureLane(id){
+ let lane=lanes.get(id);
+ if(!lane){lane={id,sequence:1,cursor:'0',pending:[],bytes:0,items:0,running:false,polling:false,controller:null};lanes.set(id,lane)}
+ return lane;
+}
+function rememberLaneClosed(id){
+ if(!id||closedLanes.has(id))return;
+ if(closedLaneOrder.length===closedLaneLimit)closedLanes.delete(closedLaneOrder.shift());
+ closedLanes.add(id);closedLaneOrder.push(id);
+}
+function finishLane(lane){
+ if(lanes.get(lane.id)!==lane)return;
+ if(lane.bytes||lane.items)release(lane.bytes,lane.items,lane);
+ lane.pending.length=0;lanes.delete(lane.id);rememberLaneClosed(lane.id);
+}
+function queueLane(value){
+ let lane=lanes.get(value.id);
+ if(!lane&&(value.type===2||value.type===3||value.type===4))return;
+ if(!lane&&closedLanes.has(value.id))throw new Error('closed lane was reused');
+ if(!lane&&value.type!==1)throw new Error('lane did not begin with OPEN');
+ lane=lane||ensureLane(value.id);
+ if(!reserve(value.data,lane)){fail();return}
+ lane.pending.push(value.data);runLaneUp(lane);
+}
+async function runLaneUp(lane){
+ if(lane.running)return;lane.running=true;
+ try{
+  while(!closed&&sessionToken&&lane.pending.length){
+   const batch=joinPending(lane.pending,lane),sequence=String(lane.sequence),laneID=String(lane.id);
+   const response=await request('/api/v1/up',()=>options('POST',sessionToken,batch.body,{'X-Up-Seq':sequence,'X-Lane-ID':laneID}));
+   if(response.status!==204||response.headers.get('X-Up-Ack')!==sequence)throw new Error('lane uplink rejected');
+   release(batch.total,batch.count,lane);port.postMessage({t:'traffic',up:batch.total,down:0});lane.sequence++;
+   if(!lane.polling)pollLane(lane);
+  }
+ }catch(error){fail()}
+ finally{lane.running=false;if(!closed&&sessionToken&&lane.pending.length)runLaneUp(lane)}
+}
+async function pollLane(lane){
+ if(!lane||lane.polling)return;lane.polling=true;
+ try{
+  while(!closed&&sessionToken&&lanes.get(lane.id)===lane){
+   const controller=new AbortController(),laneID=String(lane.id);lane.controller=controller;
+   const response=await request('/api/v1/down',()=>options('POST',sessionToken,null,{'X-Down-Cursor':lane.cursor,'X-Lane-ID':laneID},controller.signal));
+   if(response.status===204){
+    if(response.headers.get('X-Lane-Closed')==='1'){finishLane(lane);return}
+    status('connected');continue;
+   }
+   if(response.status!==200)throw new Error('lane downlink rejected');
+   const next=response.headers.get('X-Down-Cursor')||'',data=await response.arrayBuffer();
+   if(!next||!data.byteLength||splitFrames(data).some(value=>value.id!==lane.id))throw new Error('invalid lane downlink response');
+   if(closed)return;
+   port.postMessage({t:'traffic',up:0,down:data.byteLength});port.postMessage(data,[data]);lane.cursor=next;status('connected');
+  }
+ }catch(error){if(!closed)fail()}
+ finally{lane.polling=false;lane.controller=null}
+}
 function deleteSession(){
  if(sessionToken)fetch(relayOrigin+'/api/v1/session',options('DELETE',sessionToken,null,null,undefined,true)).catch(()=>{});
 }
 function close(notifyServer){
- if(closed)return;closed=true;if(pollController)pollController.abort();if(notifyServer)deleteSession();
- pending.length=0;upPending.length=0;queuedBytes=0;queuedItems=0;if(port)port.close();
+ if(closed)return;closed=true;if(pollController)pollController.abort();
+ for(const lane of lanes.values())if(lane.controller)lane.controller.abort();
+ if(notifyServer)deleteSession();pending.length=0;upPending.length=0;
+ for(const lane of lanes.values())lane.pending.length=0;lanes.clear();queuedBytes=0;queuedItems=0;if(port)port.close();
 }
 function activatePort(nextPort){
  initialized=true;port=nextPort;
  port.onmessage=message=>{
   if(message.data instanceof ArrayBuffer){
    if(!createStarted){createStarted=true;createSession(message.data)}
-   else if(!sessionToken){if(!reserve(message.data)){fail();return}pending.push(message.data)}
-   else queueUp(message.data);
+   else if(!sessionToken){if(!reserve(message.data,null)){fail();return}pending.push(message.data)}
+   else queueCarrier(message.data);
   }else if(message.data&&message.data.t==='close')close(true);
  };
  port.start();status('connecting');
@@ -204,7 +272,7 @@ addEventListener('message',event=>{
 const androidBridge=globalThis.TelegramWebProxy;
 if(!initialized&&androidNonce&&androidBridge&&typeof androidBridge.postMessage==='function'){
  const androidPort={onmessage:null,start(){},close(){androidBridge.onmessage=null},postMessage(value){
-  if(value instanceof ArrayBuffer){for(const item of splitFrames(value))androidBridge.postMessage(item)}else androidBridge.postMessage(JSON.stringify(value));
+  if(value instanceof ArrayBuffer){for(const item of splitFrames(value))androidBridge.postMessage(item.data)}else androidBridge.postMessage(JSON.stringify(value));
  }};
  androidBridge.onmessage=event=>{let data=event.data;if(typeof data==='string'){try{data=JSON.parse(data)}catch(error){return}}if(androidPort.onmessage)androidPort.onmessage({data})};
  activatePort(androidPort);androidBridge.postMessage(JSON.stringify({t:'tproxy-android-init',v:1,nonce:androidNonce}));
@@ -228,11 +296,14 @@ mod tests {
             2 * 1024 * 1024,
             32 * 1024 * 1024,
             16 * 1024,
+            WebCarrier::HttpsLanes,
             &SecureRandom::new(),
         );
         assert!(!page.body.contains("__"));
         assert!(!page.body.contains("bridge="));
         assert!(page.body.contains("X-Up-Seq"));
+        assert!(page.body.contains("carrier='https-lanes'"));
+        assert!(page.body.contains("X-Lane-ID"));
         assert!(page
             .content_security_policy
             .contains("frame-ancestors http://127.0.0.1:*"));
diff --git a/src/web/frame.rs b/src/web/frame.rs
index 0d1de8f..f02f757 100644
--- a/src/web/frame.rs
+++ b/src/web/frame.rs
@@ -4,6 +4,8 @@ use crate::config::WebLimitsConfig;
 
 /// Fixed WEB frame header size.
 pub(crate) const HEADER_BYTES: usize = 8;
+/// Largest stream identifier representable by the WEB frame header.
+pub(crate) const MAX_STREAM_ID: u32 = 0x00ff_ffff;
 /// Initial bidirectional stream credit.
 pub(crate) const INITIAL_STREAM_WINDOW: u32 = 4 * 1024 * 1024;
 /// Maximum data chunk emitted by the server.
diff --git a/src/web/http.rs b/src/web/http.rs
index 8b16df2..3af4865 100644
--- a/src/web/http.rs
+++ b/src/web/http.rs
@@ -18,7 +18,7 @@ use parking_lot::Mutex;
 use tokio::net::TcpStream;
 use tokio_util::sync::CancellationToken;
 
-use crate::config::{WebClientIpSource, WebRuntimeVhost};
+use crate::config::{WebCarrier, WebClientIpSource, WebRuntimeVhost};
 use crate::web::bridge;
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::{ManagerError, WebProcessRuntime};
@@ -183,18 +183,19 @@ async fn handle_root(
         strip_query(&mut request);
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    let carrier = profile.carrier;
     let Ok(bootstrap) = runtime.issue_bootstrap(profile, client_ip) else {
         strip_query(&mut request);
         return serve_decoy(request, vhost, true, &runtime).await;
     };
     let generation = runtime.active_generation();
-    let config = generation.config();
     let page = bridge::render(
         &vhost.host,
         &bootstrap,
-        config.web.limits.carrier_batch_bytes,
-        config.web.limits.pending_bytes_per_session,
-        config.web.limits.pending_items_per_session,
+        generation.config().web.limits.carrier_batch_bytes,
+        generation.config().web.limits.pending_bytes_per_session,
+        generation.config().web.limits.pending_items_per_session,
+        carrier,
         &generation.rng,
     );
     let mut response = full_response(StatusCode::OK, Bytes::from(page.body));
@@ -238,10 +239,7 @@ async fn handle_api(
     runtime: Arc<WebProcessRuntime>,
     vhost: Arc<WebRuntimeVhost>,
 ) -> HttpResponse {
-    if request.uri().query().is_some()
-        || request.headers().contains_key(header::COOKIE)
-        || request.headers().contains_key("x-lane-id")
-    {
+    if request.uri().query().is_some() || request.headers().contains_key(header::COOKIE) {
         return serve_decoy(request, vhost, true, &runtime).await;
     }
     let Some(client_ip) = client_ip(
@@ -272,6 +270,9 @@ async fn handle_session(
     token_hash: crate::web::manager::TokenHash,
     client_ip: IpAddr,
 ) -> HttpResponse {
+    if request.headers().contains_key("x-lane-id") {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    }
     if request.method() == Method::DELETE {
         if request.headers().contains_key(header::CONTENT_TYPE) {
             return serve_decoy(request, vhost, true, &runtime).await;
@@ -321,7 +322,7 @@ async fn handle_session(
             );
             response.headers_mut().insert(
                 HeaderName::from_static("x-carrier-mode"),
-                HeaderValue::from_static("https"),
+                HeaderValue::from_static(result.carrier.as_str()),
             );
             response.headers_mut().insert(
                 HeaderName::from_static("x-down-cursor"),
@@ -352,6 +353,9 @@ async fn handle_up(
     let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    };
     let limit = runtime.active_generation().config().web.limits.max_body_bytes;
     let CollectedBody {
         request,
@@ -364,7 +368,11 @@ async fn handle_up(
             return serve_decoy(request, vhost, true, &runtime).await;
         }
     };
-    match session.process_up(sequence, &body) {
+    let result = match lane_id {
+        Some(lane_id) => session.process_up_lane(lane_id, sequence, &body),
+        None => session.process_up(sequence, &body),
+    };
+    match result {
         Ok(ack) => {
             let mut response = carrier_empty(StatusCode::NO_CONTENT);
             insert_header(
@@ -396,6 +404,9 @@ async fn handle_down(
     let Ok(session) = runtime.get_session(token_hash, &vhost.host) else {
         return serve_decoy(request, vhost, true, &runtime).await;
     };
+    let Some(lane_id) = carrier_lane(&request, session.carrier()) else {
+        return serve_decoy(request, vhost, true, &runtime).await;
+    };
     let CollectedBody {
         request,
         body,
@@ -410,7 +421,19 @@ async fn handle_down(
     if !body.is_empty() {
         return serve_decoy(request, vhost, true, &runtime).await;
     }
-    match session.poll_down(cursor).await {
+    let _lane_poll = if lane_id.is_some() {
+        let Some(permit) = runtime.try_lane_poll() else {
+            return service_unavailable();
+        };
+        Some(permit)
+    } else {
+        None
+    };
+    let result = match lane_id {
+        Some(lane_id) => session.poll_down_lane(lane_id, cursor).await,
+        None => session.poll_down(cursor).await,
+    };
+    match result {
         Ok(result) if result.body.is_empty() => {
             let mut response = carrier_empty(StatusCode::NO_CONTENT);
             insert_header(
@@ -418,6 +441,12 @@ async fn handle_down(
                 HeaderName::from_static("x-down-cursor"),
                 &result.next_cursor.to_string(),
             );
+            if result.lane_closed {
+                response.headers_mut().insert(
+                    HeaderName::from_static("x-lane-closed"),
+                    HeaderValue::from_static("1"),
+                );
+            }
             response
         }
         Ok(result) => {
@@ -437,6 +466,16 @@ async fn handle_down(
     }
 }
 
+fn carrier_lane<B>(request: &Request<B>, carrier: WebCarrier) -> Option<Option<u32>> {
+    match carrier {
+        WebCarrier::Https => (!request.headers().contains_key("x-lane-id")).then_some(None),
+        WebCarrier::HttpsLanes => canonical_u64_header(request, "x-lane-id")
+            .and_then(|value| u32::try_from(value).ok())
+            .filter(|value| *value <= frame::MAX_STREAM_ID)
+            .map(Some),
+    }
+}
+
 fn carrier_headers(response: &mut HttpResponse) {
     response.headers_mut().insert(
         header::CONTENT_TYPE,
diff --git a/src/web/http/tests.rs b/src/web/http/tests.rs
index 86f81e8..1b093fd 100644
--- a/src/web/http/tests.rs
+++ b/src/web/http/tests.rs
@@ -10,19 +10,20 @@ use tokio_util::sync::CancellationToken;
 
 use super::serve_connection;
 use crate::config::{
-    ProxyConfig, WebClientIpSource, WebRuntimeConfig, WebRuntimeDecoy,
+    ProxyConfig, WebCarrier, WebClientIpSource, WebRuntimeConfig, WebRuntimeDecoy,
     WebRuntimeProfile, WebRuntimeVhost, WebSecretMode, WebStaticAsset, WebStaticSite,
 };
 use crate::maestro::generation::test_runtime_generation;
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::WebProcessRuntime;
 
-fn runtime_config(capability: [u8; 32]) -> ProxyConfig {
+fn runtime_config(capability: [u8; 32], carrier: WebCarrier) -> ProxyConfig {
     let profile = Arc::new(WebRuntimeProfile {
         host: "proxy.example.com".to_string(),
         public_addr: "203.0.113.10:443".parse().unwrap(),
         user: "alice".to_string(),
         secret_mode: WebSecretMode::Plain,
+        carrier,
         capability,
         max_sessions: 4,
         max_streams: 16,
@@ -60,6 +61,7 @@ fn runtime_config(capability: [u8; 32]) -> ProxyConfig {
     );
     let mut config = ProxyConfig::default();
     config.web.enabled = true;
+    config.web.carrier = carrier;
     config.web.limits.max_bootstraps_per_ip = 1;
     config.web.timeouts.shutdown_secs = 1;
     config.web.runtime = Some(Arc::new(WebRuntimeConfig {
@@ -115,7 +117,7 @@ fn response_header<'a>(headers: &'a [u8], name: &str) -> &'a str {
 #[tokio::test]
 async fn https_carrier_bootstraps_and_closes_one_session() {
     let capability = [7u8; 32];
-    let generation = test_runtime_generation(1, runtime_config(capability));
+    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(Arc::clone(&active_runtime));
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -171,13 +173,17 @@ async fn https_carrier_bootstraps_and_closes_one_session() {
     let session = response_header(create_headers, "x-session-token");
     assert_eq!(session.len(), 43);
 
-    let replacement = test_runtime_generation(2, runtime_config(capability));
+    let replacement = test_runtime_generation(
+        2,
+        runtime_config(capability, WebCarrier::HttpsLanes),
+    );
     active_runtime.store(Arc::clone(&replacement));
     tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
     let retry_response = request(&listener, &runtime, create_retry).await;
     let (retry_headers, retry_body) = split_response(&retry_response);
     assert!(retry_headers.starts_with(b"HTTP/1.1 200"));
     assert_eq!(response_header(retry_headers, "x-session-token"), session);
+    assert_eq!(response_header(retry_headers, "x-carrier-mode"), "https");
     assert_eq!(retry_body, frame::encode(FrameType::Welcome, 0, &[]));
 
     let next_root = format!(
@@ -187,6 +193,9 @@ async fn https_carrier_bootstraps_and_closes_one_session() {
     let next_root_response = request(&listener, &runtime, next_root).await;
     let (_, next_root_body) = split_response(&next_root_response);
     assert!(next_root_body.windows(11).any(|value| value == b"bootstrap='"));
+    assert!(next_root_body
+        .windows(21)
+        .any(|value| value == b"carrier='https-lanes'"));
 
     let close = format!(
         "DELETE /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n"
@@ -204,3 +213,75 @@ async fn https_carrier_bootstraps_and_closes_one_session() {
     replacement.stop_sessions().await;
     replacement.stop_background_tasks().await;
 }
+
+#[tokio::test]
+async fn https_lanes_is_advertised_and_requires_canonical_lane_headers() {
+    let capability = [9u8; 32];
+    let generation = test_runtime_generation(
+        1,
+        runtime_config(capability, WebCarrier::HttpsLanes),
+    );
+    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 root = format!(
+        "GET /?bridge={encoded} HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nConnection: close\r\n\r\n"
+    )
+    .into_bytes();
+    let root_response = request(&listener, &runtime, root).await;
+    let (_, root_body) = split_response(&root_response);
+    let root_body = std::str::from_utf8(root_body).unwrap();
+    assert!(root_body.contains("carrier='https-lanes'"));
+    let bootstrap = root_body
+        .split_once("bootstrap='")
+        .and_then(|(_, suffix)| suffix.split_once('\''))
+        .map(|(token, _)| token)
+        .unwrap();
+
+    let hello = frame::encode(FrameType::Hello, 0, &[1]);
+    let mut create = format!(
+        "POST /api/v1/session HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {bootstrap}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        hello.len()
+    )
+    .into_bytes();
+    create.extend_from_slice(&hello);
+    let create_response = request(&listener, &runtime, create).await;
+    let (create_headers, _) = split_response(&create_response);
+    assert_eq!(response_header(create_headers, "x-carrier-mode"), "https-lanes");
+    let session = response_header(create_headers, "x-session-token").to_string();
+
+    let pong = frame::encode(FrameType::Pong, 0, &[]);
+    let mut uplink = format!(
+        "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 1\r\nX-Lane-ID: 0\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        pong.len()
+    )
+    .into_bytes();
+    uplink.extend_from_slice(&pong);
+    let uplink_response = request(&listener, &runtime, uplink).await;
+    let (uplink_headers, _) = split_response(&uplink_response);
+    assert!(uplink_headers.starts_with(b"HTTP/1.1 204"));
+    assert_eq!(response_header(uplink_headers, "x-up-ack"), "1");
+
+    let mut missing_lane = format!(
+        "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 2\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        pong.len()
+    )
+    .into_bytes();
+    missing_lane.extend_from_slice(&pong);
+    let missing_lane_response = request(&listener, &runtime, missing_lane).await;
+    assert!(!missing_lane_response.starts_with(b"HTTP/1.1 204"));
+
+    let mut aliased_lane = format!(
+        "POST /api/v1/up HTTP/1.1\r\nHost: proxy.example.com\r\nX-Forwarded-For: 192.0.2.10\r\nAuthorization: Bearer {session}\r\nContent-Type: application/octet-stream\r\nX-Up-Seq: 2\r\nX-Lane-ID: 00\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+        pong.len()
+    )
+    .into_bytes();
+    aliased_lane.extend_from_slice(&pong);
+    let aliased_lane_response = request(&listener, &runtime, aliased_lane).await;
+    assert!(!aliased_lane_response.starts_with(b"HTTP/1.1 204"));
+
+    runtime.shutdown().await;
+    generation.stop_sessions().await;
+    generation.stop_background_tasks().await;
+}
diff --git a/src/web/manager.rs b/src/web/manager.rs
index 896d20a..51f114f 100644
--- a/src/web/manager.rs
+++ b/src/web/manager.rs
@@ -13,7 +13,7 @@ use tokio_util::sync::CancellationToken;
 use tokio_util::task::TaskTracker;
 use zeroize::Zeroizing;
 
-use crate::config::{WebLimitsConfig, WebRuntimeProfile};
+use crate::config::{WebCarrier, WebLimitsConfig, WebRuntimeProfile};
 use crate::maestro::generation::RuntimeGeneration;
 use crate::web::frame;
 use crate::web::session::WebSession;
@@ -59,6 +59,8 @@ pub(crate) enum ManagerError {
 pub(crate) struct CreateResult {
     /// Opaque bearer token for the created or replayed session.
     pub(crate) token: String,
+    /// Carrier frozen into the created or replayed session.
+    pub(crate) carrier: WebCarrier,
 }
 
 /// Process-owned bounded WEB credential, session, and memory coordinator.
@@ -68,6 +70,7 @@ pub(crate) struct WebProcessRuntime {
     state: Mutex<ManagerState>,
     http_connections: Arc<Semaphore>,
     http_handlers: Arc<Semaphore>,
+    lane_polls: Arc<Semaphore>,
     body_readers: Arc<Semaphore>,
     body_bytes: Arc<Semaphore>,
     stream_handshakes: Arc<Semaphore>,
@@ -94,6 +97,7 @@ impl WebProcessRuntime {
             active_runtime,
             http_connections: Arc::new(Semaphore::new(limits.max_http_connections)),
             http_handlers: Arc::new(Semaphore::new(limits.max_http_handlers)),
+            lane_polls: Arc::new(Semaphore::new((limits.max_http_handlers / 2).max(1))),
             body_readers: Arc::new(Semaphore::new(limits.max_body_readers)),
             body_bytes: Arc::new(Semaphore::new(limits.max_body_bytes_global)),
             stream_handshakes: Arc::new(Semaphore::new(limits.max_stream_handshakes)),
@@ -154,6 +158,15 @@ impl WebProcessRuntime {
         permit
     }
 
+    /// Reserves one parked lane poll without exhausting all HTTP handlers.
+    pub(crate) fn try_lane_poll(&self) -> Option<OwnedSemaphorePermit> {
+        let permit = Arc::clone(&self.lane_polls).try_acquire_owned().ok();
+        if permit.is_none() {
+            self.record_limit_hit();
+        }
+        permit
+    }
+
     /// Reserves one logical stream in the inner MTProxy handshake phase.
     pub(crate) fn try_stream_handshake(&self) -> Option<OwnedSemaphorePermit> {
         let permit = Arc::clone(&self.stream_handshakes)
@@ -307,11 +320,13 @@ impl WebProcessRuntime {
             if !digest_matches {
                 return Err(ManagerError::Authentication);
             }
-            if entry.session.is_none() {
-                return Err(ManagerError::Authentication);
-            }
+            let session = entry
+                .session
+                .as_ref()
+                .ok_or(ManagerError::Authentication)?;
             return Ok(CreateResult {
                 token: entry.session_token.as_str().to_owned(),
+                carrier: session.carrier(),
             });
         }
         if entry.generation_id != generation.id {
@@ -379,6 +394,7 @@ impl WebProcessRuntime {
         self.sessions_created.fetch_add(1, Ordering::Relaxed);
         Ok(CreateResult {
             token: session_token,
+            carrier: session.carrier(),
         })
     }
 
diff --git a/src/web/manager/state.rs b/src/web/manager/state.rs
index 811d3da..28a47a5 100644
--- a/src/web/manager/state.rs
+++ b/src/web/manager/state.rs
@@ -130,6 +130,7 @@ pub(super) fn matching_profile(
                 && profile.public_addr == expected.public_addr
                 && profile.user == expected.user
                 && profile.secret_mode == expected.secret_mode
+                && profile.carrier == expected.carrier
                 && profile.capability == expected.capability
         })
         .cloned()
diff --git a/src/web/session.rs b/src/web/session.rs
index 0646fe2..45795f0 100644
--- a/src/web/session.rs
+++ b/src/web/session.rs
@@ -12,7 +12,7 @@ use tokio::io::ReadBuf;
 use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 
-use crate::config::{WebLimitsConfig, WebRuntimeProfile, WebTimeoutsConfig};
+use crate::config::{WebCarrier, WebLimitsConfig, WebRuntimeProfile, WebTimeoutsConfig};
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::{ProfileKey, TokenHash, WebProcessRuntime};
 
@@ -20,6 +20,8 @@ use crate::web::manager::{ProfileKey, TokenHash, WebProcessRuntime};
 mod backend;
 // Downlink queues own cursor replay, flow control, and memory reservations.
 mod downlink;
+// Lane carrier state isolates request sequencing and downlink replay per logical stream.
+mod lanes;
 // Uplink batches own exactly-once sequencing and client-frame validation.
 mod uplink;
 
@@ -64,6 +66,34 @@ struct DownBatch {
     control_items: usize,
 }
 
+struct CarrierLane {
+    pending_frames: VecDeque<QueuedFrame>,
+    pending_windows: HashMap<u32, usize>,
+    unacked: Option<DownBatch>,
+    down_cursor: u64,
+    down_epoch: u64,
+    last_up_sequence: u64,
+    last_up_digest: TokenHash,
+    up_active: bool,
+    notify: Arc<Notify>,
+}
+
+impl CarrierLane {
+    fn new() -> Self {
+        Self {
+            pending_frames: VecDeque::new(),
+            pending_windows: HashMap::new(),
+            unacked: None,
+            down_cursor: 0,
+            down_epoch: 0,
+            last_up_sequence: 0,
+            last_up_digest: [0; 32],
+            up_active: false,
+            notify: Arc::new(Notify::new()),
+        }
+    }
+}
+
 struct SessionState {
     streams: HashMap<u32, StreamState>,
     active_peer_ports: HashSet<u16>,
@@ -76,6 +106,7 @@ struct SessionState {
     down_epoch: u64,
     last_up_sequence: u64,
     last_up_digest: TokenHash,
+    carrier_lanes: HashMap<u32, CarrierLane>,
     pending_bytes: usize,
     pending_items: usize,
     pending_control_bytes: usize,
@@ -108,6 +139,8 @@ pub(crate) struct PollResult {
     pub(crate) body: Bytes,
     /// Cursor the client must present on its next downlink request.
     pub(crate) next_cursor: u64,
+    /// Indicates that a drained non-zero lane no longer needs polling.
+    pub(crate) lane_closed: bool,
 }
 
 impl WebSession {
@@ -122,6 +155,10 @@ impl WebSession {
         limits: WebLimitsConfig,
         timeouts: WebTimeoutsConfig,
     ) -> Arc<Self> {
+        let mut carrier_lanes = HashMap::new();
+        if profile.carrier == WebCarrier::HttpsLanes {
+            carrier_lanes.insert(0, CarrierLane::new());
+        }
         Arc::new(Self {
             manager,
             token_hash,
@@ -142,6 +179,7 @@ impl WebSession {
                 down_epoch: 0,
                 last_up_sequence: 0,
                 last_up_digest: [0; 32],
+                carrier_lanes,
                 pending_bytes: 0,
                 pending_items: 0,
                 pending_control_bytes: 0,
@@ -168,6 +206,11 @@ impl WebSession {
         self.profile.host == host
     }
 
+    /// Returns the immutable carrier selected when this session was created.
+    pub(crate) fn carrier(&self) -> WebCarrier {
+        self.profile.carrier
+    }
+
     /// Closes carrier state while relay tasks retain their admission until exit.
     pub(crate) fn close(&self) {
         let (data_bytes, data_items, control_bytes, control_items) = {
@@ -188,6 +231,10 @@ impl WebSession {
             state.pending_frames.clear();
             state.pending_windows.clear();
             state.unacked = None;
+            for lane in state.carrier_lanes.values() {
+                lane.notify.notify_waiters();
+            }
+            state.carrier_lanes.clear();
             let control_bytes = state.pending_control_bytes;
             let control_items = state.pending_control_items;
             let data_bytes = state.pending_bytes.saturating_sub(control_bytes);
@@ -199,7 +246,9 @@ impl WebSession {
             (data_bytes, data_items, control_bytes, control_items)
         };
         self.cancel.cancel();
-        self.down_notify.notify_waiters();
+        if self.carrier() == WebCarrier::Https {
+            self.down_notify.notify_waiters();
+        }
         if let Some(manager) = self.manager.upgrade() {
             manager.release_pending(data_bytes, data_items, false);
             manager.release_pending(control_bytes, control_items, true);
@@ -311,7 +360,9 @@ impl WebSession {
         stream.send_credit -= count as u64;
         state.last_activity = Instant::now();
         drop(state);
-        self.down_notify.notify_waiters();
+        if self.carrier() == WebCarrier::Https {
+            self.down_notify.notify_waiters();
+        }
         Poll::Ready(Ok(count))
     }
 
@@ -342,14 +393,17 @@ fn inbound_queue_cost(queue: &VecDeque<InboundChunk>) -> (usize, usize) {
     (bytes, queue.len())
 }
 
-fn remember_closed(state: &mut SessionState, stream_id: u32, limit: usize) {
+fn remember_closed(state: &mut SessionState, stream_id: u32, limit: usize) -> Option<u32> {
     if !state.closed_streams.insert(stream_id) {
-        return;
+        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
 }
diff --git a/src/web/session/backend.rs b/src/web/session/backend.rs
index 50f6374..083f39d 100644
--- a/src/web/session/backend.rs
+++ b/src/web/session/backend.rs
@@ -7,7 +7,7 @@ use crate::web::frame::FrameType;
 use crate::web::stream::WebLogicalStream;
 use crate::proxy::shared_state::ConntrackClosePolicy;
 
-use super::{WebSession, inbound_queue_cost, remember_closed};
+use super::{WebSession, inbound_queue_cost};
 
 impl WebSession {
     /// Starts one owned inner handshake and relay task for an admitted stream.
@@ -65,11 +65,7 @@ impl WebSession {
             let queued = state.streams.remove(&stream_id).map(|stream| {
                 let (bytes, items) = inbound_queue_cost(&stream.inbound);
                 self.release_locked(&mut state, bytes, items, false);
-                remember_closed(
-                    &mut state,
-                    stream_id,
-                    self.limits.max_tombstones_per_session,
-                );
+                self.remember_closed_locked(&mut state, stream_id);
                 self.queue_control_locked(&mut state, FrameType::Close, stream_id, &[])
             });
             (queued, reserved)
@@ -88,7 +84,9 @@ impl WebSession {
             if !queued {
                 self.close();
             }
-            self.down_notify.notify_waiters();
+            if self.carrier() == crate::config::WebCarrier::Https {
+                self.down_notify.notify_waiters();
+            }
         }
     }
 }
diff --git a/src/web/session/downlink.rs b/src/web/session/downlink.rs
index 40ef15d..c40f86f 100644
--- a/src/web/session/downlink.rs
+++ b/src/web/session/downlink.rs
@@ -6,12 +6,16 @@ use super::{
     DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame, SessionState,
     WebSession,
 };
+use crate::config::WebCarrier;
 use crate::web::frame::{self, FrameType};
 use crate::web::manager::ManagerError;
 
 impl WebSession {
     /// Polls pending downlink frames with cursor replay and newest-poll-wins semantics.
     pub(crate) async fn poll_down(&self, cursor: u64) -> Result<PollResult, ManagerError> {
+        if self.carrier() != WebCarrier::Https {
+            return Err(ManagerError::Protocol);
+        }
         let epoch = {
             let mut state = self.state.lock();
             if state.closed {
@@ -23,6 +27,7 @@ impl WebSession {
                     return Ok(PollResult {
                         body: unacked.body.clone(),
                         next_cursor: unacked.next_cursor,
+                        lane_closed: false,
                     });
                 }
                 if cursor != unacked.next_cursor {
@@ -51,6 +56,7 @@ impl WebSession {
                         return Ok(PollResult {
                             body: Bytes::new(),
                             next_cursor: cursor,
+                            lane_closed: false,
                         });
                     }
                     if !state.pending_frames.is_empty() {
@@ -65,6 +71,7 @@ impl WebSession {
                         let result = PollResult {
                             body: batch.body.clone(),
                             next_cursor: batch.next_cursor,
+                            lane_closed: false,
                         };
                         if let Some(manager) = self.manager.upgrade() {
                             manager.record_down(result.body.len());
@@ -89,6 +96,7 @@ impl WebSession {
                 Ok(PollResult {
                     body: Bytes::new(),
                     next_cursor: cursor,
+                    lane_closed: false,
                 })
             }
         }
@@ -210,6 +218,14 @@ impl WebSession {
         if amount == 0 {
             return true;
         }
+        if self.carrier() == WebCarrier::HttpsLanes {
+            return self.queue_control_locked(
+                state,
+                FrameType::Window,
+                stream_id,
+                &frame::window_payload(amount),
+            );
+        }
         if let Some(index) = state.pending_windows.get(&stream_id).copied()
             && let Some(queued) = state.pending_frames.get_mut(index)
         {
@@ -251,6 +267,9 @@ impl WebSession {
         stream_id: u32,
         payload: &[u8],
     ) -> bool {
+        if self.carrier() == WebCarrier::HttpsLanes {
+            return self.queue_frame_locked(state, FrameType::Data, stream_id, payload, false);
+        }
         let can_coalesce = state.pending_frames.back().is_some_and(|last| {
             last.frame_type == FrameType::Data
                 && last.stream_id == stream_id
@@ -281,6 +300,15 @@ impl WebSession {
         payload: &[u8],
         control: bool,
     ) -> bool {
+        if self.carrier() == WebCarrier::HttpsLanes {
+            return self.queue_lane_frame_locked(
+                state,
+                frame_type,
+                stream_id,
+                payload,
+                control,
+            );
+        }
         let cost = frame::HEADER_BYTES + payload.len() + QUEUE_ITEM_COST;
         let class = if control {
             PendingClass::Control
@@ -408,6 +436,7 @@ mod tests {
             public_addr: SocketAddr::from(([203, 0, 113, 10], 443)),
             user: "alice".to_string(),
             secret_mode: WebSecretMode::Plain,
+            carrier: WebCarrier::Https,
             capability: [0; 32],
             max_sessions: 1,
             max_streams: 1,
diff --git a/src/web/session/lanes.rs b/src/web/session/lanes.rs
new file mode 100644
index 0000000..cf85c02
--- /dev/null
+++ b/src/web/session/lanes.rs
@@ -0,0 +1,525 @@
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use bytes::{BufMut, Bytes, BytesMut};
+use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
+
+use super::uplink::{inbound_reservation, validate_batch};
+use super::{
+    CarrierLane, DownBatch, PendingClass, PollResult, QUEUE_ITEM_COST, QueuedFrame,
+    SessionState, WebSession, remember_closed,
+};
+use crate::config::{WebCarrier, WebLimitsConfig};
+use crate::web::frame::{self, Frame, FrameType};
+use crate::web::manager::{ManagerError, TokenHash};
+
+impl WebSession {
+    /// Applies one exactly-once uplink batch to an independent HTTPS lane.
+    pub(crate) fn process_up_lane(
+        self: &Arc<Self>,
+        lane_id: u32,
+        sequence: u64,
+        body: &[u8],
+    ) -> Result<u64, ManagerError> {
+        if self.carrier() != WebCarrier::HttpsLanes || lane_id > frame::MAX_STREAM_ID {
+            return Err(ManagerError::Protocol);
+        }
+        let frames = match frame::parse_all(body, &self.limits) {
+            Ok(frames) => frames,
+            Err(_) => {
+                self.close();
+                return Err(ManagerError::Protocol);
+            }
+        };
+        if frames.iter().copied().any(|value| {
+            value.stream_id != lane_id || frame::validate_client_shape(value).is_err()
+        }) {
+            self.close();
+            return Err(ManagerError::Protocol);
+        }
+        let digest: TokenHash = Sha256::digest(body).into();
+        let mut opened = Vec::new();
+        let result = {
+            let mut state = self.state.lock();
+            if state.closed {
+                return Err(ManagerError::Closed);
+            }
+            state.last_activity = Instant::now();
+            if !state.carrier_lanes.contains_key(&lane_id) {
+                if lane_id != 0
+                    && frames.first().is_some_and(|value| value.frame_type != FrameType::Open)
+                    && only_late_frames(&frames)
+                {
+                    return Ok(sequence);
+                }
+                if lane_id == 0
+                    || frames.first().is_none_or(|value| value.frame_type != FrameType::Open)
+                {
+                    drop(state);
+                    self.close();
+                    return Err(ManagerError::Protocol);
+                }
+                state.carrier_lanes.insert(lane_id, CarrierLane::new());
+            }
+            let lane = state
+                .carrier_lanes
+                .get_mut(&lane_id)
+                .ok_or(ManagerError::Protocol)?;
+            if sequence == lane.last_up_sequence && sequence != 0 {
+                return if bool::from(lane.last_up_digest.ct_eq(&digest)) {
+                    Ok(sequence)
+                } else {
+                    drop(state);
+                    self.close();
+                    Err(ManagerError::Protocol)
+                };
+            }
+            if sequence == 0 || sequence != lane.last_up_sequence.saturating_add(1) {
+                drop(state);
+                self.close();
+                return Err(ManagerError::Protocol);
+            }
+            if lane.up_active {
+                return Err(ManagerError::Concurrent);
+            }
+            lane.up_active = true;
+            if !validate_batch(&state, &frames) {
+                drop(state);
+                self.close();
+                return Err(ManagerError::Protocol);
+            }
+            let (reserve_bytes, reserve_items) = inbound_reservation(&state, &frames);
+            if !self.reserve_locked(
+                &mut state,
+                reserve_bytes,
+                reserve_items,
+                PendingClass::Uplink,
+            ) {
+                if let Some(lane) = state.carrier_lanes.get_mut(&lane_id) {
+                    lane.up_active = false;
+                }
+                return Err(ManagerError::Backpressure);
+            }
+            let mut unused_bytes = reserve_bytes;
+            let mut unused_items = reserve_items;
+            let applied = self.apply_batch_locked(
+                &mut state,
+                &frames,
+                &mut opened,
+                &mut unused_bytes,
+                &mut unused_items,
+            );
+            self.release_locked(&mut state, unused_bytes, unused_items, false);
+            if let Some(lane) = state.carrier_lanes.get_mut(&lane_id) {
+                lane.up_active = false;
+                if applied {
+                    lane.last_up_sequence = sequence;
+                    lane.last_up_digest = digest;
+                }
+            }
+            applied.then_some(sequence).ok_or(ManagerError::Closed)
+        };
+        if matches!(result, Err(ManagerError::Backpressure)) {
+            return result;
+        }
+        if result.is_err() {
+            self.close();
+            for (_, peer_port) in opened {
+                self.release_stream_reservation(peer_port);
+            }
+            return result;
+        }
+        for (stream_id, peer_port) in opened {
+            self.spawn_stream(stream_id, peer_port);
+        }
+        if let Some(manager) = self.manager.upgrade() {
+            manager.record_up(body.len());
+        }
+        result
+    }
+
+    /// Polls one lane with independent cursor replay and newest-poll-wins semantics.
+    pub(crate) async fn poll_down_lane(
+        &self,
+        lane_id: u32,
+        cursor: u64,
+    ) -> Result<PollResult, ManagerError> {
+        if self.carrier() != WebCarrier::HttpsLanes || lane_id > frame::MAX_STREAM_ID {
+            return Err(ManagerError::Protocol);
+        }
+        let (epoch, notify) = {
+            let mut state = self.state.lock();
+            if state.closed {
+                return Err(ManagerError::Closed);
+            }
+            let acknowledged = {
+                let Some(lane) = state.carrier_lanes.get_mut(&lane_id) else {
+                    return Err(ManagerError::Protocol);
+                };
+                if let Some(unacked) = &lane.unacked {
+                    if cursor == unacked.base_cursor {
+                        return Ok(PollResult {
+                            body: unacked.body.clone(),
+                            next_cursor: unacked.next_cursor,
+                            lane_closed: false,
+                        });
+                    }
+                    if cursor != unacked.next_cursor {
+                        drop(state);
+                        self.close();
+                        return Err(ManagerError::Protocol);
+                    }
+                    lane.unacked.take()
+                } else {
+                    if cursor != lane.down_cursor {
+                        drop(state);
+                        self.close();
+                        return Err(ManagerError::Protocol);
+                    }
+                    None
+                }
+            };
+            if let Some(batch) = acknowledged {
+                self.release_locked(&mut state, batch.data_bytes, batch.data_items, false);
+                self.release_locked(
+                    &mut state,
+                    batch.control_bytes,
+                    batch.control_items,
+                    true,
+                );
+                if let Some(stream) = state.streams.get_mut(&lane_id)
+                    && let Some(waker) = stream.write_waker.take()
+                {
+                    waker.wake();
+                }
+            }
+            state.last_activity = Instant::now();
+            let lane = state
+                .carrier_lanes
+                .get_mut(&lane_id)
+                .ok_or(ManagerError::Protocol)?;
+            lane.down_epoch = lane.down_epoch.wrapping_add(1).max(1);
+            (lane.down_epoch, Arc::clone(&lane.notify))
+        };
+        notify.notify_waiters();
+
+        let deadline = Duration::from_secs(self.timeouts.long_poll_secs);
+        let poll = async {
+            loop {
+                let notified = notify.notified();
+                {
+                    let mut state = self.state.lock();
+                    if state.closed {
+                        return Err(ManagerError::Closed);
+                    }
+                    let Some(lane) = state.carrier_lanes.get_mut(&lane_id) else {
+                        return Ok(PollResult {
+                            body: Bytes::new(),
+                            next_cursor: cursor,
+                            lane_closed: true,
+                        });
+                    };
+                    if lane.down_epoch != epoch {
+                        return Ok(PollResult {
+                            body: Bytes::new(),
+                            next_cursor: cursor,
+                            lane_closed: false,
+                        });
+                    }
+                    if !lane.pending_frames.is_empty() {
+                        let batch = match take_lane_down_batch(&self.limits, lane, cursor) {
+                            Ok(batch) => batch,
+                            Err(error) => {
+                                drop(state);
+                                self.close();
+                                return Err(error);
+                            }
+                        };
+                        let result = PollResult {
+                            body: batch.body.clone(),
+                            next_cursor: batch.next_cursor,
+                            lane_closed: false,
+                        };
+                        lane.unacked = Some(batch);
+                        drop(state);
+                        if let Some(manager) = self.manager.upgrade() {
+                            manager.record_down(result.body.len());
+                        }
+                        return Ok(result);
+                    }
+                    if lane_id != 0
+                        && !state.streams.contains_key(&lane_id)
+                        && state.closed_streams.contains(&lane_id)
+                    {
+                        return Ok(PollResult {
+                            body: Bytes::new(),
+                            next_cursor: cursor,
+                            lane_closed: true,
+                        });
+                    }
+                }
+                notified.await;
+            }
+        };
+        match tokio::time::timeout(deadline, poll).await {
+            Ok(result) => result,
+            Err(_) => {
+                let mut state = self.state.lock();
+                if state.closed {
+                    return Err(ManagerError::Closed);
+                }
+                if !state.carrier_lanes.contains_key(&lane_id) {
+                    return Ok(PollResult {
+                        body: Bytes::new(),
+                        next_cursor: cursor,
+                        lane_closed: true,
+                    });
+                }
+                if lane_id != 0
+                    && !state.streams.contains_key(&lane_id)
+                    && state.closed_streams.contains(&lane_id)
+                {
+                    return Ok(PollResult {
+                        body: Bytes::new(),
+                        next_cursor: cursor,
+                        lane_closed: true,
+                    });
+                }
+                if state
+                    .carrier_lanes
+                    .get(&lane_id)
+                    .is_some_and(|lane| lane.down_epoch == epoch)
+                {
+                    state.last_activity = Instant::now();
+                }
+                Ok(PollResult {
+                    body: Bytes::new(),
+                    next_cursor: cursor,
+                    lane_closed: false,
+                })
+            }
+        }
+    }
+
+    pub(super) fn queue_lane_frame_locked(
+        &self,
+        state: &mut SessionState,
+        frame_type: FrameType,
+        stream_id: u32,
+        payload: &[u8],
+        control: bool,
+    ) -> bool {
+        if !state.carrier_lanes.contains_key(&stream_id) {
+            return false;
+        }
+        if frame_type == FrameType::Window {
+            let coalesced = state.carrier_lanes.get(&stream_id).and_then(|lane| {
+                let index = lane.pending_windows.get(&stream_id).copied()?;
+                let queued = lane.pending_frames.get(index)?;
+                let previous = u32::from_be_bytes(
+                    queued.encoded[frame::HEADER_BYTES..frame::HEADER_BYTES + 4]
+                        .try_into()
+                        .unwrap_or([0; 4]),
+                );
+                previous
+                    .checked_add(frame::window_amount(payload).unwrap_or(0))
+                    .map(|total| (index, total))
+            });
+            if let Some((index, total)) = coalesced
+                && let Some(lane) = state.carrier_lanes.get_mut(&stream_id)
+                && let Some(queued) = lane.pending_frames.get_mut(index)
+            {
+                queued.encoded[frame::HEADER_BYTES..frame::HEADER_BYTES + 4]
+                    .copy_from_slice(&total.to_be_bytes());
+                lane.notify.notify_waiters();
+                return true;
+            }
+        }
+        let can_coalesce = frame_type == FrameType::Data
+            && state
+                .carrier_lanes
+                .get(&stream_id)
+                .and_then(|lane| lane.pending_frames.back())
+                .is_some_and(|last| {
+                    last.frame_type == FrameType::Data
+                        && last.stream_id == stream_id
+                        && last.encoded.len() - frame::HEADER_BYTES + payload.len()
+                            <= self.limits.max_frame_payload_bytes
+                });
+        if can_coalesce {
+            if !self.reserve_locked(state, payload.len(), 0, PendingClass::Downlink) {
+                return false;
+            }
+            let Some(lane) = state.carrier_lanes.get_mut(&stream_id) else {
+                self.release_locked(state, payload.len(), 0, false);
+                return false;
+            };
+            let Some(last) = lane.pending_frames.back_mut() else {
+                self.release_locked(state, payload.len(), 0, false);
+                return false;
+            };
+            last.encoded.extend_from_slice(payload);
+            last.cost += payload.len();
+            let payload_len = (last.encoded.len() - frame::HEADER_BYTES) as u32;
+            last.encoded[4..8].copy_from_slice(&payload_len.to_be_bytes());
+            lane.notify.notify_waiters();
+            return true;
+        }
+        let cost = frame::HEADER_BYTES + payload.len() + QUEUE_ITEM_COST;
+        let class = if control {
+            PendingClass::Control
+        } else {
+            PendingClass::Downlink
+        };
+        if !self.reserve_locked(state, cost, 1, class) {
+            return false;
+        }
+        let mut encoded = BytesMut::with_capacity(frame::HEADER_BYTES + payload.len());
+        encoded.put_u8(frame_type as u8);
+        encoded.put_u8((stream_id >> 16) as u8);
+        encoded.put_u8((stream_id >> 8) as u8);
+        encoded.put_u8(stream_id as u8);
+        encoded.put_u32(payload.len() as u32);
+        encoded.extend_from_slice(payload);
+        let Some(lane) = state.carrier_lanes.get_mut(&stream_id) else {
+            self.release_locked(state, cost, 1, control);
+            return false;
+        };
+        let index = lane.pending_frames.len();
+        lane.pending_frames.push_back(QueuedFrame {
+            encoded,
+            frame_type,
+            stream_id,
+            control,
+            cost,
+        });
+        if frame_type == FrameType::Window {
+            lane.pending_windows.insert(stream_id, index);
+        }
+        lane.notify.notify_waiters();
+        true
+    }
+
+    pub(super) fn remember_closed_locked(&self, state: &mut SessionState, stream_id: u32) {
+        let evicted = remember_closed(
+            state,
+            stream_id,
+            self.limits.max_tombstones_per_session,
+        );
+        if self.carrier() != WebCarrier::HttpsLanes {
+            return;
+        }
+        if let Some(evicted) = evicted {
+            self.release_lane_locked(state, evicted);
+        }
+        if let Some(lane) = state.carrier_lanes.get(&stream_id) {
+            lane.notify.notify_waiters();
+        }
+    }
+
+    fn release_lane_locked(&self, state: &mut SessionState, lane_id: u32) {
+        let Some(mut lane) = state.carrier_lanes.remove(&lane_id) else {
+            return;
+        };
+        lane.notify.notify_waiters();
+        let mut data_bytes = 0usize;
+        let mut data_items = 0usize;
+        let mut control_bytes = 0usize;
+        let mut control_items = 0usize;
+        for queued in lane.pending_frames.drain(..) {
+            if queued.control {
+                control_bytes = control_bytes.saturating_add(queued.cost);
+                control_items = control_items.saturating_add(1);
+            } else {
+                data_bytes = data_bytes.saturating_add(queued.cost);
+                data_items = data_items.saturating_add(1);
+            }
+        }
+        if let Some(batch) = lane.unacked.take() {
+            data_bytes = data_bytes.saturating_add(batch.data_bytes);
+            data_items = data_items.saturating_add(batch.data_items);
+            control_bytes = control_bytes.saturating_add(batch.control_bytes);
+            control_items = control_items.saturating_add(batch.control_items);
+        }
+        self.release_locked(state, data_bytes, data_items, false);
+        self.release_locked(state, control_bytes, control_items, true);
+    }
+}
+
+fn only_late_frames(frames: &[Frame<'_>]) -> bool {
+    frames.iter().all(|value| {
+        matches!(
+            value.frame_type,
+            FrameType::Data | FrameType::Window | FrameType::Close
+        )
+    })
+}
+
+fn take_lane_down_batch(
+    limits: &WebLimitsConfig,
+    lane: &mut CarrierLane,
+    cursor: u64,
+) -> Result<DownBatch, ManagerError> {
+    let next_cursor = lane
+        .down_cursor
+        .checked_add(1)
+        .ok_or(ManagerError::Protocol)?;
+    let mut count = 0usize;
+    let mut body_len = 0usize;
+    for queued in &lane.pending_frames {
+        if count >= limits.max_frames_per_body
+            || (count != 0
+                && body_len.saturating_add(queued.encoded.len()) > limits.carrier_batch_bytes)
+        {
+            break;
+        }
+        body_len += queued.encoded.len();
+        count += 1;
+    }
+    let mut body = BytesMut::with_capacity(body_len);
+    let mut data_bytes = 0usize;
+    let mut data_items = 0usize;
+    let mut control_bytes = 0usize;
+    let mut control_items = 0usize;
+    for index in 0..count {
+        let Some(queued) = lane.pending_frames.get(index) else {
+            break;
+        };
+        if queued.frame_type == FrameType::Window
+            && lane.pending_windows.get(&queued.stream_id) == Some(&index)
+        {
+            lane.pending_windows.remove(&queued.stream_id);
+        }
+    }
+    for _ in 0..count {
+        let Some(queued) = lane.pending_frames.pop_front() else {
+            break;
+        };
+        body.extend_from_slice(&queued.encoded);
+        if queued.control {
+            control_bytes += queued.cost;
+            control_items += 1;
+        } else {
+            data_bytes += queued.cost;
+            data_items += 1;
+        }
+    }
+    for index in lane.pending_windows.values_mut() {
+        *index = index.saturating_sub(count);
+    }
+    lane.down_cursor = next_cursor;
+    Ok(DownBatch {
+        body: body.freeze(),
+        base_cursor: cursor,
+        next_cursor,
+        data_bytes,
+        data_items,
+        control_bytes,
+        control_items,
+    })
+}
+
+// Lane-specific protocol, replay, and lifecycle tests.
+#[cfg(test)]
+mod tests;
diff --git a/src/web/session/lanes/tests.rs b/src/web/session/lanes/tests.rs
new file mode 100644
index 0000000..6281800
--- /dev/null
+++ b/src/web/session/lanes/tests.rs
@@ -0,0 +1,129 @@
+use std::net::SocketAddr;
+use std::sync::Arc;
+
+use bytes::BytesMut;
+
+use super::*;
+use crate::config::{
+    WebRuntimeProfile, WebSecretMode, WebTimeoutsConfig,
+};
+use crate::web::manager::WebProcessRuntime;
+
+fn session_with_limits(limits: WebLimitsConfig) -> 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: WebCarrier::HttpsLanes,
+        capability: [0; 32],
+        max_sessions: 1,
+        max_streams: 2,
+        max_streams_per_session: 2,
+    });
+    WebSession::new(
+        std::sync::Weak::<WebProcessRuntime>::new(),
+        [1; 32],
+        "192.0.2.10".parse().unwrap(),
+        profile,
+        [2; 32],
+        limits,
+        WebTimeoutsConfig::default(),
+    )
+}
+
+fn session() -> Arc<WebSession> {
+    session_with_limits(WebLimitsConfig::default())
+}
+
+#[test]
+fn lane_uplink_sequences_are_independent_and_exactly_once() {
+    let session = session();
+    {
+        let mut state = session.state.lock();
+        for lane_id in [51, 52] {
+            state.carrier_lanes.insert(lane_id, CarrierLane::new());
+            state.closed_streams.insert(lane_id);
+        }
+    }
+    let first = frame::encode(FrameType::Data, 51, b"first");
+    let second = frame::encode(FrameType::Data, 52, b"second");
+    assert_eq!(session.process_up_lane(51, 1, &first), Ok(1));
+    assert_eq!(session.process_up_lane(52, 1, &second), Ok(1));
+    assert_eq!(session.process_up_lane(51, 1, &first), Ok(1));
+    assert_eq!(session.state.lock().carrier_lanes[&52].last_up_sequence, 1);
+}
+
+#[test]
+fn cross_lane_frame_is_fatal_to_https_lane_session() {
+    let session = session();
+    let body = frame::encode(FrameType::Data, 52, b"wrong lane");
+    assert_eq!(
+        session.process_up_lane(51, 1, &body),
+        Err(ManagerError::Protocol)
+    );
+    assert!(session.state.lock().closed);
+}
+
+#[tokio::test]
+async fn drained_closed_lane_replays_then_signals_completion() {
+    let session = session();
+    {
+        let mut state = session.state.lock();
+        state.carrier_lanes.insert(7, CarrierLane::new());
+        state.closed_streams.insert(7);
+        let lane = state.carrier_lanes.get_mut(&7).unwrap();
+        let encoded = frame::encode(FrameType::Close, 7, &[]);
+        lane.pending_frames.push_back(QueuedFrame {
+            encoded: BytesMut::from(encoded.as_ref()),
+            frame_type: FrameType::Close,
+            stream_id: 7,
+            control: true,
+            cost: frame::HEADER_BYTES + QUEUE_ITEM_COST,
+        });
+    }
+    let first = session.poll_down_lane(7, 0).await.unwrap();
+    let replay = session.poll_down_lane(7, 0).await.unwrap();
+    assert_eq!(first.body, replay.body);
+    assert!(!replay.lane_closed);
+    let finished = session.poll_down_lane(7, 1).await.unwrap();
+    assert!(finished.body.is_empty());
+    assert!(finished.lane_closed);
+}
+
+#[test]
+fn tombstone_eviction_releases_lane_budget_and_accepts_late_frames() {
+    let limits = WebLimitsConfig {
+        max_tombstones_per_session: 1,
+        ..WebLimitsConfig::default()
+    };
+    let session = session_with_limits(limits);
+    {
+        let mut state = session.state.lock();
+        state.carrier_lanes.insert(7, CarrierLane::new());
+        let encoded = frame::encode(FrameType::Close, 7, &[]);
+        let cost = encoded.len() + QUEUE_ITEM_COST;
+        state.carrier_lanes.get_mut(&7).unwrap().pending_frames.push_back(
+            QueuedFrame {
+                encoded: BytesMut::from(encoded.as_ref()),
+                frame_type: FrameType::Close,
+                stream_id: 7,
+                control: true,
+                cost,
+            },
+        );
+        state.pending_bytes = cost;
+        state.pending_items = 1;
+        state.pending_control_bytes = cost;
+        state.pending_control_items = 1;
+        session.remember_closed_locked(&mut state, 7);
+        state.carrier_lanes.insert(8, CarrierLane::new());
+        session.remember_closed_locked(&mut state, 8);
+        assert!(!state.carrier_lanes.contains_key(&7));
+        assert_eq!(state.pending_bytes, 0);
+        assert_eq!(state.pending_items, 0);
+    }
+    let late = frame::encode(FrameType::Data, 7, b"late");
+    assert_eq!(session.process_up_lane(7, 7, &late), Ok(7));
+    assert!(!session.state.lock().closed);
+}
diff --git a/src/web/session/uplink.rs b/src/web/session/uplink.rs
index d2285d1..649c7f0 100644
--- a/src/web/session/uplink.rs
+++ b/src/web/session/uplink.rs
@@ -9,8 +9,9 @@ use subtle::ConstantTimeEq;
 
 use super::{
     InboundChunk, PendingClass, QUEUE_ITEM_COST, SessionState, StreamState, WebSession,
-    inbound_queue_cost, remember_closed,
+    inbound_queue_cost,
 };
+use crate::config::WebCarrier;
 use crate::web::frame::{self, Frame, FrameType};
 use crate::web::manager::{ManagerError, TokenHash};
 
@@ -21,6 +22,9 @@ impl WebSession {
         sequence: u64,
         body: &[u8],
     ) -> Result<u64, ManagerError> {
+        if self.carrier() != WebCarrier::Https {
+            return Err(ManagerError::Protocol);
+        }
         if self
             .up_active
             .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
@@ -117,7 +121,7 @@ impl WebSession {
         result
     }
 
-    fn apply_batch_locked(
+    pub(super) fn apply_batch_locked(
         &self,
         state: &mut SessionState,
         frames: &[Frame<'_>],
@@ -133,11 +137,7 @@ impl WebSession {
             match value.frame_type {
                 FrameType::Open => {
                     let Some(peer_port) = self.reserve_stream_locked(state) else {
-                        remember_closed(
-                            state,
-                            value.stream_id,
-                            self.limits.max_tombstones_per_session,
-                        );
+                        self.remember_closed_locked(state, value.stream_id);
                         if !self.queue_control_locked(
                             state,
                             FrameType::Close,
@@ -195,11 +195,7 @@ impl WebSession {
                     };
                     let (bytes, items) = inbound_queue_cost(&stream.inbound);
                     self.release_locked(state, bytes, items, false);
-                    remember_closed(
-                        state,
-                        value.stream_id,
-                        self.limits.max_tombstones_per_session,
-                    );
+                    self.remember_closed_locked(state, value.stream_id);
                     if let Some(waker) = stream.read_waker {
                         waker.wake();
                     }
@@ -246,7 +242,7 @@ impl Drop for UplinkGuard<'_> {
     }
 }
 
-fn validate_batch(state: &SessionState, frames: &[Frame<'_>]) -> bool {
+pub(super) fn validate_batch(state: &SessionState, frames: &[Frame<'_>]) -> bool {
     let mut live = state
         .streams
         .iter()
@@ -312,7 +308,10 @@ fn validate_batch(state: &SessionState, frames: &[Frame<'_>]) -> bool {
     true
 }
 
-fn inbound_reservation(state: &SessionState, frames: &[Frame<'_>]) -> (usize, usize) {
+pub(super) fn inbound_reservation(
+    state: &SessionState,
+    frames: &[Frame<'_>],
+) -> (usize, usize) {
     let mut live = state.streams.keys().copied().collect::<HashSet<_>>();
     let mut bytes = 0usize;
     let mut items = 0usize;
@@ -350,6 +349,7 @@ mod tests {
             public_addr: SocketAddr::from(([203, 0, 113, 10], 443)),
             user: "alice".to_string(),
             secret_mode: WebSecretMode::Plain,
+            carrier: WebCarrier::Https,
             capability: [0; 32],
             max_sessions: 1,
             max_streams: 1,