diff --git a/CHANGELOG.md b/CHANGELOG.md
index 7434cb5..d89a9a9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,7 @@
 ## [Unreleased]
 
 ### 新增
+- **会话感知收发（`ISessionAwareStreamConnection<TMessage>` 可选能力接口，#39）**：为有严格会话边界的协议（HSMS 的 Select/T3/T6、禁止重放）提供——`CurrentSessionId`（每次 TCP 会话建立时分配、单调递增不复用，无会话时为 0；分配/归零先于对应状态对外发布，回调与 `WaitForConnectedAsync` 完成时读取必得一致值）、`SendInSessionAsync`（**整帧写入本机 socket 后才完成**；会话在写出前终止以 `SessionExpiredException` 失败且绝不转移到新会话重放；会话拆除时立即 fault 挂起条目，不空等重连；调用方取消的提交点 = worker 认领，认领后取消不撕裂帧）、`GetSessionMessages`（消息带所属会话编号的接收视图，与 `GetMessages` 为二选一的竞争消费视图）。发送队列内部信封化，`SendAsync`/`GetMessages` 现有语义一字不改。
 - **未完成帧超时（`IncompleteFrameTimeoutMs`，默认 0 = 关闭）**（#38）：帧已开头、缓冲里留着半帧字节，但连续这么久未收到后续字节时判定会话失效并断线重连——与 `ReceiveIdleTimeoutMs`（完全静默也计时，要求周期流量）互补，适合允许长时间空闲、但半帧卡死必须判死的长度前缀协议（如 HSMS T8）。计时只在半帧进行中生效：缓冲为空不计时、收到新字节即重置、整帧切尽后归零；超时经 `FrameError` 上报新类别 `IncompleteFrameTimeout` 并携带受 8KB 上限保护的缓冲快照。纯增量、默认行为不变。
 
 ### 变更
diff --git a/README.en.md b/README.en.md
index 84b9848..27bffdf 100644
--- a/README.en.md
+++ b/README.en.md
@@ -99,6 +99,35 @@ public interface ICodec<TMessage>
 - **Liveness detection (optional)**: TCP KeepAlive and receive idle timeout to catch half-open connections (peer power loss / unplugged cable)
 - **Events**: `ConnectionChanged` state changes, `FrameError` frame-level diagnostics, `RawBytesReceived/Sent` raw bytes (HEX debugging)
 - **Backpressure**: bounded send queue — `SendAsync` waits when full; the receive side buffers without limit by default, or set `ReceiveQueueCapacity` — decoding pauses when the consumer lags, propagating TCP backpressure to the peer and preventing unbounded memory growth
+- **Session-aware messaging (optional, advanced)**: `CurrentSessionId` / `SendInSessionAsync` (completes only after the whole frame is written to the socket; fails on session loss and never replays across sessions) / `GetSessionMessages` (messages carry their session id) — for protocols with strict session boundaries such as HSMS; see the dedicated section below
+
+## Session-aware messaging (advanced)
+
+For typical business traffic that tolerates replay, `SendAsync` (completes on enqueue; queued messages continue on the next session) + `GetMessages` (a stable stream across reconnects) is enough. Some protocols have strict session boundaries — re-handshake after reconnect, no replay of old-session messages, protocol timers starting when the frame is actually written (HSMS Select/T3/T6/T8). For those there is an optional capability interface `ISessionAwareStreamConnection<TMessage>` (implemented by `StreamConnection<TMessage>`; upper layers depending on the interface abstraction detect it with `is`):
+
+```csharp
+if (connection is ISessionAwareStreamConnection<MyMessage> sessionAware)
+{
+    long id = sessionAware.CurrentSessionId;   // allocated per established TCP session, monotonic, never reused; 0 when no session
+
+    // Completes only after the whole frame is handed to the local socket;
+    // session ends before that → SessionExpiredException, the message is never replayed
+    await sessionAware.SendInSessionAsync(id, message, ct);
+
+    // Receive view: every message carries the session it arrived on
+    // (late deliveries from an old session's decoder still carry the old id)
+    await foreach (var m in sessionAware.GetSessionMessages(ct))
+        Handle(m.SessionId, m.Message);
+}
+```
+
+Semantics at a glance:
+
+- **Id linearization**: reading `CurrentSessionId` from a `ConnectionChanged` callback or after `WaitForConnectedAsync` always yields a valid id (allocation happens before Connected becomes visible); it is already zero when a non-Connected state becomes visible.
+- **"Written" means**: the whole frame has been handed to the local socket (kernel buffer), peer ACK not included — the best signal available at the application layer. When the task fails, **treat the remote outcome as unknown** (the peer may have received part or all of the bytes); reconcile via your protocol's transaction correlation, idempotency or recovery flow.
+- **Cancellation commit point**: cancelling before the send worker claims the entry cancels the task and the message is never sent; cancelling after the claim (frame write in progress) has no effect — cancelling one message neither tears frames nor kills the connection.
+- **Relation to `SendAsync`**: both share one FIFO (serialized in enqueue order); the plain `SendAsync` cross-session replay behavior is unchanged. Session-bound sends never transfer to a new session on any failure path.
+- **The two receive APIs are not a broadcast**: `GetMessages` and `GetSessionMessages` are competing consumer views of the same channel — enumerating both splits messages between them; pick one.
 
 ## Diagnostics & debugging
 
diff --git a/README.md b/README.md
index fd6b4dd..538b431 100644
--- a/README.md
+++ b/README.md
@@ -99,6 +99,33 @@ public interface ICodec<TMessage>
 - **活性探测（可选）**：TCP KeepAlive 与接收空闲超时，兜底半开连接（对端断电/拔线）
 - **事件**：`ConnectionChanged` 状态变化、`FrameError` 帧层诊断、`RawBytesReceived/Sent` 原始字节（HEX 调试）
 - **发送背压**：有界发送队列，队列满时 `SendAsync` 自动等待；接收侧默认无上限缓冲，可用 `ReceiveQueueCapacity` 设上限——消费慢时解码暂停、TCP 背压自然传导到对端，防内存无限增长
+- **会话感知收发（可选高级）**：`CurrentSessionId` / `SendInSessionAsync`（整帧写入 socket 才完成、会话失效即失败、绝不跨会话重放）/ `GetSessionMessages`（消息带会话编号）——为有严格会话边界的协议（如 HSMS）准备，详见下文
+
+## 会话感知收发（高级）
+
+对允许消息重放的一般业务，`SendAsync`（入队即完成、断线后由新会话续发）+ `GetMessages`（跨重连稳定流）就够了。部分协议有严格的会话边界——重连后必须重新握手、旧会话消息禁止重放、协议计时器要从"整帧实际写出"起算（HSMS 的 Select/T3/T6/T8 即是）。为此提供可选能力接口 `ISessionAwareStreamConnection<TMessage>`（`StreamConnection<TMessage>` 已实现；依赖接口抽象的上层用 `is` 探测）：
+
+```csharp
+if (connection is ISessionAwareStreamConnection<MyMessage> sessionAware)
+{
+    long id = sessionAware.CurrentSessionId;          // 每次 TCP 会话建立时分配，单调递增不复用；无会话时为 0
+
+    // 整帧全部写入本机 socket 后才完成；会话在写出前终止 → SessionExpiredException，消息绝不重放
+    await sessionAware.SendInSessionAsync(id, message, ct);
+
+    // 接收视图：每条消息携带它所属的会话编号（旧会话解码任务迟到投递的消息带旧编号）
+    await foreach (var m in sessionAware.GetSessionMessages(ct))
+        Handle(m.SessionId, m.Message);
+}
+```
+
+语义要点：
+
+- **编号的线性化**：`ConnectionChanged` 回调与 `WaitForConnectedAsync` 完成时读 `CurrentSessionId` 必得有效值（分配先于 Connected 对外发布）；状态离开 Connected（Retry/Disconnected）可见时已归零。
+- **"写完"的定义**：整帧字节已交给本机 socket（内核缓冲），不含对端 ACK——应用层可得的最好信号。任务失败时**远端处理结果视为未知**（可能已收到部分/全部字节），由上层协议的事务关联、幂等或恢复流程兜底。
+- **调用方取消的提交点**：发送 worker 认领条目之前取消 → 任务取消且消息不再发送；认领之后（帧已开始写出）取消对结果无副作用——取消单条消息不会撕裂帧、不会杀死连接。
+- **与 `SendAsync` 的关系**：两类发送共享同一条 FIFO（按入队顺序串行化）；普通 `SendAsync` 的跨会话续发行为不变。会话绑定发送在任何失败路径下都不转移到新会话。
+- **两个接收 API 不是广播**：`GetMessages` 与 `GetSessionMessages` 是同一通道的两个竞争消费视图，同时枚举会互相分流——请二选一使用。
 
 ## 诊断与调试
 
diff --git a/src/StreamFrame/Connection/FrameDecoder.cs b/src/StreamFrame/Connection/FrameDecoder.cs
index 7adb4da..9a62978 100644
--- a/src/StreamFrame/Connection/FrameDecoder.cs
+++ b/src/StreamFrame/Connection/FrameDecoder.cs
@@ -26,7 +26,8 @@ internal sealed class FrameDecoder<TMessage>
     private readonly PipeReader _reader;
     private readonly IFramer _framing;
     private readonly ICodec<TMessage> _codec;
-    private readonly Channel<TMessage> _relay;
+    private readonly Channel<SessionMessage<TMessage>> _relay;
+    private readonly long _sessionId;
     private readonly int _maxIncompleteFrameBytes;
     private readonly int _incompleteFrameTimeoutMs;
     private readonly DecodeErrorPolicy _decodeErrorPolicy;
@@ -36,7 +37,8 @@ public FrameDecoder(
         PipeReader reader,
         IFramer framing,
         ICodec<TMessage> codec,
-        Channel<TMessage> relay,
+        Channel<SessionMessage<TMessage>> relay,
+        long sessionId,
         int maxIncompleteFrameBytes,
         int incompleteFrameTimeoutMs,
         DecodeErrorPolicy decodeErrorPolicy,
@@ -46,6 +48,7 @@ public FrameDecoder(
         _framing = framing;
         _codec = codec;
         _relay = relay;
+        _sessionId = sessionId;
         _maxIncompleteFrameBytes = maxIncompleteFrameBytes;
         _incompleteFrameTimeoutMs = incompleteFrameTimeoutMs;
         _decodeErrorPolicy = decodeErrorPolicy;
@@ -104,7 +107,8 @@ public async Task RunAsync(CancellationToken ct)
 
                     try
                     {
-                        await _relay.Writer.WriteAsync(message, CancellationToken.None).ConfigureAwait(false);
+                        await _relay.Writer.WriteAsync(
+                            new SessionMessage<TMessage>(_sessionId, message), CancellationToken.None).ConfigureAwait(false);
                     }
                     catch (ChannelClosedException)
                     {
diff --git a/src/StreamFrame/Connection/StreamConnection.cs b/src/StreamFrame/Connection/StreamConnection.cs
index f9b854a..1c043aa 100644
--- a/src/StreamFrame/Connection/StreamConnection.cs
+++ b/src/StreamFrame/Connection/StreamConnection.cs
@@ -1,4 +1,5 @@
 using System.Buffers;
+using System.Collections.Concurrent;
 using System.IO.Pipelines;
 using System.Net;
 using System.Net.Sockets;
@@ -16,15 +17,17 @@ namespace StreamFrame;
 /// <code>
 /// socket --ReceiveAsync--> pipe.Writer --[Pipe]--> 解码循环
 ///   (producer task)                                  (consumer task)
-///                                                      │ while(TryDecodeFrame) → codec.Decode → Channel&lt;TMessage&gt;
-/// 发送：业务 SendAsync 入队 --有界Channel--> 发送worker → codec.Encode → framing.EncodeFrame → _sendLock → socket
+///                                                      │ while(TryDecodeFrame) → codec.Decode → Channel&lt;SessionMessage&gt;
+/// 发送：业务 SendAsync/SendInSessionAsync 入队 --有界Channel--> 发送worker → codec.Encode → framing.EncodeFrame → _sendLock → socket
 /// </code>
 ///
-/// 会话模型：一次 TCP 连接 = 一个会话（Pipe + 三个会话任务）；断线或会话故障时整个会话
-/// 作废重建，而 <c>_messageRelay</c> 消息通道是连接级的、跨会话存活——业务侧的
+/// 会话模型：一次 TCP 连接 = 一个会话（Pipe + 三个会话任务 + 公共会话编号）；断线或会话故障时
+/// 整个会话作废重建，而 <c>_messageRelay</c> 消息通道是连接级的、跨会话存活——业务侧的
 /// <see cref="GetMessages"/> 枚举在重连前后是同一条稳定流，仅在 Dispose 时正常结束。
+/// 会话感知收发（<see cref="ISessionAwareStreamConnection{TMessage}"/>）在此之上提供
+/// 绑定会话的发送与带会话编号的接收视图。
 /// </summary>
-public sealed class StreamConnection<TMessage> : IStreamConnection<TMessage>
+public sealed class StreamConnection<TMessage> : ISessionAwareStreamConnection<TMessage>
 {
     /// <inheritdoc />
     public event EventHandler<ConnectionState>? ConnectionChanged;
@@ -81,8 +84,20 @@ public string? RemoteIpAddress
     private Task? _sendWorkerTask;
     private int _sessionEpoch;
 
-    private readonly Channel<TMessage> _sendQueue;
-    private readonly Channel<TMessage> _messageRelay;
+    /// <inheritdoc cref="ISessionAwareStreamConnection{TMessage}.CurrentSessionId" />
+    public long CurrentSessionId => Volatile.Read(ref _currentSessionId);
+
+    /// <summary>公共会话编号的分配计数器：仅在真实 TCP 会话建立时递增（有间隔、单调、不复用）；checked 防静默回绕。</summary>
+    private long _sessionCounter;
+
+    /// <summary>当前会话编号：0 = 无会话。分配/归零都必须先于对应状态对外发布（线性化点，见 CommunicationStateChanging）。</summary>
+    private long _currentSessionId;
+
+    private readonly Channel<SessionSendEntry> _sendQueue;
+    private readonly Channel<SessionMessage<TMessage>> _messageRelay;
+
+    /// <summary>已注册、尚未终结的会话绑定发送条目：会话拆除时立即 fault，避免调用方空等重连全程。</summary>
+    private readonly ConcurrentDictionary<SessionSendEntry, byte> _pendingSessionSends = new();
     private readonly SemaphoreSlim _sendLock = new(initialCount: 1);
     private readonly SemaphoreSlim _acceptLock = new(1, 1);
     private readonly ILogger _logger;
@@ -125,7 +140,7 @@ public StreamConnection(
         _options.Validate();
         _logger = logger ?? NullLogger.Instance;
 
-        _sendQueue = Channel.CreateBounded<TMessage>(new BoundedChannelOptions(_options.SendQueueCapacity)
+        _sendQueue = Channel.CreateBounded<SessionSendEntry>(new BoundedChannelOptions(_options.SendQueueCapacity)
         {
             SingleReader = true,
             SingleWriter = false,
@@ -135,7 +150,7 @@ public StreamConnection(
         // 消息通道归连接所有、跨会话复用：会话重建偶发新旧解码循环并存，不能声明 SingleWriter。
         if (_options.ReceiveQueueCapacity > 0)
         {
-            _messageRelay = Channel.CreateBounded<TMessage>(new BoundedChannelOptions(_options.ReceiveQueueCapacity)
+            _messageRelay = Channel.CreateBounded<SessionMessage<TMessage>>(new BoundedChannelOptions(_options.ReceiveQueueCapacity)
             {
                 SingleReader = false,
                 SingleWriter = false,
@@ -145,7 +160,7 @@ public StreamConnection(
         }
         else
         {
-            _messageRelay = Channel.CreateUnbounded<TMessage>(new UnboundedChannelOptions
+            _messageRelay = Channel.CreateUnbounded<SessionMessage<TMessage>>(new UnboundedChannelOptions
             {
                 SingleReader = false,
                 SingleWriter = false,
@@ -392,6 +407,19 @@ private void ScheduleReconnect(int epoch)
 
     private void CommunicationStateChanging(ConnectionState newState, int? observedEpoch = null)
     {
+        // 公共会话编号的线性化点：Connected 对外发布（事件回调/等待器完成）之前完成分配；
+        // 离开 Connected 的状态（Retry/Disconnected）对外发布之前归零——状态可见时旧会话
+        // 已不可用于新的会话绑定发送（SendInSessionAsync 以编号不匹配快速失败）。
+        if (newState == ConnectionState.Connected)
+        {
+            var id = checked(Interlocked.Increment(ref _sessionCounter));
+            Volatile.Write(ref _currentSessionId, id);
+        }
+        else if (newState is ConnectionState.Retry or ConnectionState.Disconnected)
+        {
+            Volatile.Write(ref _currentSessionId, 0);
+        }
+
         State = newState;
 
         try
@@ -413,7 +441,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
         switch (newState)
         {
             case ConnectionState.Connected:
-                StartSession();
+                StartSession(Volatile.Read(ref _currentSessionId));
                 break;
 
             case ConnectionState.Retry:
@@ -431,6 +459,7 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
                             return; // 故障会话已被替换，避免误杀新会话
 
                         _sessionEpoch++; // 同会话的其它迟到故障立即过期，防止二次拆建
+                        FaultPendingSessionSends(); // 拆除时立即失败挂起的会话绑定发送（先于任务拆除）
                         StopSessionCore();
                     }
 
@@ -445,12 +474,20 @@ private void CommunicationStateChanging(ConnectionState newState, int? observedE
         }
     }
 
-    private void StartSession()
+    private void StartSession(long sessionId)
     {
         lock (_sessionGate)
         {
             StopSessionCore();
 
+            // 病态重入兜底：Connected 回调里同步触发 Reconnect() 会让编号在 StartSession
+            // 执行前被归零——此时重新分配一个，保证会话任务拿到的编号恒非零。
+            if (sessionId == 0)
+            {
+                sessionId = checked(Interlocked.Increment(ref _sessionCounter));
+                Volatile.Write(ref _currentSessionId, sessionId);
+            }
+
             var epoch = ++_sessionEpoch;
             var cts = new CancellationTokenSource();
             var token = cts.Token;
@@ -464,7 +501,7 @@ private void StartSession()
                 : _framing.MaxPayloadBytes + 4096;
 
             var decoder = new FrameDecoder<TMessage>(
-                pipe.Reader, _framing, _codec, _messageRelay,
+                pipe.Reader, _framing, _codec, _messageRelay, sessionId,
                 maxIncompleteFrameBytes, _options.IncompleteFrameTimeoutMs,
                 _options.DecodeErrorPolicy, RaiseFrameError);
 
@@ -495,6 +532,20 @@ private async Task WatchSessionFaultsAsync(Task sessionTask, CancellationTokenSo
         }
     }
 
+    /// <summary>
+    /// 会话拆除（Retry/停机）时立即 fault 所有挂起的会话绑定发送：排队中未写出的条目以
+    /// <see cref="SessionExpiredException"/> 结束，调用方不必空等重连；已认领（正在写出）的条目
+    /// 由写出路径给出结果。普通条目不受影响（按既有语义留给新会话续发）。
+    /// 注意：只在真正的拆除点调用——<see cref="StartSession"/> 开头对旧会话的清理不能调用，
+    /// 否则会误杀"Connected 已发布、新会话任务尚未就绪"窗口内注册的新条目。
+    /// </summary>
+    private void FaultPendingSessionSends()
+    {
+        foreach (var entry in _pendingSessionSends.Keys)
+            entry.TryExpire(new SessionExpiredException(
+                entry.SessionId, $"会话 {entry.SessionId} 已终止（断线重连/停机），消息未写出。"));
+    }
+
     private void StopSession()
     {
         lock (_sessionGate)
@@ -621,12 +672,40 @@ private async Task<int> ReceiveWithIdleTimeoutAsync(Socket socket, Memory<byte>
     private async Task SendWorkerAsync(CancellationToken ct)
     {
         // 发送失败（socket 故障/会话失效）不在此吞掉：上抛给会话守护 → 断线重连，
-        // 未发送的消息留在连接级队列，由重连后的新 worker 继续发送。
+        // 未发送的普通消息留在连接级队列，由重连后的新 worker 继续发送。
+        // 会话绑定条目已在拆除时被 fault（或调用方已取消）：只跳过、不发送、不重放。
         var reader = _sendQueue.Reader;
         while (await reader.WaitToReadAsync(ct).ConfigureAwait(false))
         {
-            while (reader.TryRead(out var message))
-                await SendFramedAsync(message, ct).ConfigureAwait(false);
+            while (reader.TryRead(out var entry))
+            {
+                if (!entry.IsSessionBound)
+                {
+                    await SendFramedAsync(entry.Message, ct).ConfigureAwait(false);
+                    continue;
+                }
+
+                // 认领 = 调用方取消的提交点：认领成功后写入只受会话令牌控制
+                if (!entry.TryClaimForSend())
+                    continue;
+
+                try
+                {
+                    await SendFramedAsync(entry.Message, ct).ConfigureAwait(false);
+                    entry.Complete(); // 整帧已写入本机 socket，任务成功完成
+                }
+                catch (OperationCanceledException ex) when (ct.IsCancellationRequested)
+                {
+                    entry.Fault(new SessionExpiredException(
+                        entry.SessionId, $"会话 {entry.SessionId} 在整帧写出前终止。", ex));
+                    throw;
+                }
+                catch (Exception ex)
+                {
+                    entry.Fault(ex);
+                    throw;
+                }
+            }
         }
     }
 
@@ -714,7 +793,126 @@ private void RaiseFrameError(FrameErrorEventArgs args)
 
     /// <inheritdoc />
     public Task SendAsync(TMessage message, CancellationToken ct = default)
-        => _sendQueue.Writer.WriteAsync(message, ct).AsTask();
+        => _sendQueue.Writer.WriteAsync(new SessionSendEntry(message), ct).AsTask();
+
+    /// <inheritdoc cref="ISessionAwareStreamConnection{TMessage}.SendInSessionAsync" />
+    public async Task SendInSessionAsync(long sessionId, TMessage message, CancellationToken ct = default)
+    {
+        ThrowIfDisposed();
+
+        // 快速失败（非权威判定；权威在 worker 出队认领与会话拆除 fault）
+        if (Volatile.Read(ref _currentSessionId) != sessionId)
+            throw NewSessionExpired(sessionId);
+
+        var entry = new SessionSendEntry(sessionId, message);
+        _pendingSessionSends.TryAdd(entry, 0);
+        try
+        {
+            // 双检：注册与"拆除归零编号"的竞态——先注册、再复核，失效则立即 fault 本条目
+            if (Volatile.Read(ref _currentSessionId) != sessionId)
+            {
+                entry.TryExpire(NewSessionExpired(sessionId));
+                throw NewSessionExpired(sessionId);
+            }
+
+            // 调用方取消：提交点（worker 认领）之前使条目退出；认领之后无副作用
+            using var ctReg = ct.Register(
+                static state => ((SessionSendEntry)state!).TryCancelByCaller(), entry);
+
+            // 入队（队列满时等待）。期间允许两类及时结束：调用方取消 → WriteAsync 取消；
+            // 会话失效 → 条目被拆除 fault（CompletionTask 先完成，迟到的入队由 worker 跳过）
+            var enqueueTask = _sendQueue.Writer.WriteAsync(entry, ct).AsTask();
+            var first = await Task.WhenAny(enqueueTask, entry.CompletionTask).ConfigureAwait(false);
+
+            if (first == enqueueTask && enqueueTask.Status == TaskStatus.RanToCompletion)
+            {
+                // 正常入队：等待写出结果（整帧写完成功 / 会话失效 / 提交点前取消）
+                await entry.CompletionTask.ConfigureAwait(false);
+                return;
+            }
+
+            if (entry.CompletionTask.IsCompleted)
+            {
+                // 会话拆除已 fault（或提交点前已取消）：以条目结果为准（统一失败类型）；
+                // 迟到的入队可能仍会成功，旧信封由 worker 跳过——观察其异常避免未观察任务异常
+                ObserveLater(enqueueTask);
+                await entry.CompletionTask.ConfigureAwait(false);
+                return;
+            }
+
+            // 入队失败且与会话无关（调用方取消 / 通道已关闭）：传播入队异常
+            await enqueueTask.ConfigureAwait(false);
+        }
+        finally
+        {
+            _pendingSessionSends.TryRemove(entry, out _);
+        }
+    }
+
+    private static SessionExpiredException NewSessionExpired(long sessionId)
+        => new(sessionId, $"会话 {sessionId} 已失效（当前无此会话）。");
+
+    /// <summary>后台观察一个可能迟到完成的入队任务的异常，避免未观察任务异常。</summary>
+    private static void ObserveLater(Task task)
+        => _ = task.ContinueWith(static completed => _ = completed.Exception, TaskScheduler.Default);
+
+    /// <summary>
+    /// 发送队列条目。普通条目（<see cref="SendAsync"/>）无完成源：入队即完成调用方任务，
+    /// 未发送的消息跨会话续发（既有语义）。会话绑定条目（<see cref="SendInSessionAsync"/>）
+    /// 携带完成源：整帧写入 socket 后成功完成；会话终止时失败、绝不重放。
+    ///
+    /// 状态字三向互斥（提交点/取消/失效的线性化）：排队中 → 已认领（worker 正在写出，
+    /// 只有写出路径能给结果）/ 已取消（调用方在提交点前取消）/ 已失效（会话拆除 fault）。
+    /// </summary>
+    private sealed class SessionSendEntry
+    {
+        private const int Queued = 0;
+        private const int Claimed = 1;
+        private const int CancelledByCaller = 2;
+        private const int Expired = 3;
+
+        private readonly TaskCompletionSource<bool>? _completion;
+        private int _state;
+
+        public SessionSendEntry(TMessage message)
+            => Message = message;
+
+        public SessionSendEntry(long sessionId, TMessage message)
+        {
+            SessionId = sessionId;
+            Message = message;
+            _completion = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+        }
+
+        public TMessage Message { get; }
+        public long SessionId { get; }
+        public bool IsSessionBound => _completion is not null;
+        public Task CompletionTask => _completion?.Task ?? Task.CompletedTask;
+
+        /// <summary>worker 认领（发送提交点）：仅排队中的条目可发送；已取消/已失效一律跳过。</summary>
+        public bool TryClaimForSend()
+            => Interlocked.CompareExchange(ref _state, Claimed, Queued) == Queued;
+
+        /// <summary>调用方在提交点前取消：条目退出（worker 跳过），完成源以取消收尾。</summary>
+        public void TryCancelByCaller()
+        {
+            if (Interlocked.CompareExchange(ref _state, CancelledByCaller, Queued) == Queued)
+                _completion?.TrySetCanceled(CancellationToken.None);
+        }
+
+        /// <summary>会话拆除 fault：仅对排队中的条目生效；已认领的由写出路径给出结果。</summary>
+        public void TryExpire(Exception reason)
+        {
+            if (Interlocked.CompareExchange(ref _state, Expired, Queued) == Queued)
+                _completion?.TrySetException(reason);
+        }
+
+        public void Complete()
+            => _completion?.TrySetResult(true);
+
+        public void Fault(Exception exception)
+            => _completion?.TrySetException(exception);
+    }
 
     /// <inheritdoc />
     public Task WaitForConnectedAsync(CancellationToken ct = default)
@@ -750,8 +948,20 @@ public async IAsyncEnumerable<TMessage> GetMessages(
         var reader = _messageRelay.Reader;
         while (await reader.WaitToReadAsync(ct).ConfigureAwait(false))
         {
-            while (reader.TryRead(out var message))
-                yield return message;
+            while (reader.TryRead(out var envelope))
+                yield return envelope.Message;
+        }
+    }
+
+    /// <inheritdoc cref="ISessionAwareStreamConnection{TMessage}.GetSessionMessages" />
+    public async IAsyncEnumerable<SessionMessage<TMessage>> GetSessionMessages(
+        [EnumeratorCancellation] CancellationToken ct = default)
+    {
+        var reader = _messageRelay.Reader;
+        while (await reader.WaitToReadAsync(ct).ConfigureAwait(false))
+        {
+            while (reader.TryRead(out var envelope))
+                yield return envelope;
         }
     }
 
@@ -789,6 +999,7 @@ private void Shutdown()
         _lifetimeRegistration.Dispose();
         _lifetimeCts?.Cancel(); // 停止连接/监听重试循环
         _sendQueue.Writer.TryComplete();
+        FaultPendingSessionSends(); // 停机：挂起的会话绑定发送全部以会话失效收尾
 
         StopSession();
 
diff --git a/src/StreamFrame/ISessionAwareStreamConnection.cs b/src/StreamFrame/ISessionAwareStreamConnection.cs
new file mode 100644
index 0000000..39d56f0
--- /dev/null
+++ b/src/StreamFrame/ISessionAwareStreamConnection.cs
@@ -0,0 +1,54 @@
+namespace StreamFrame;
+
+/// <summary>
+/// 会话感知的可选能力：为有严格会话边界的协议（如 HSMS——重连后必须重新握手、
+/// 旧会话消息禁止重放、协议计时器从整帧实际写出起算）提供绑定会话的收发。
+/// 由 <see cref="StreamConnection{TMessage}"/> 实现；依赖
+/// <see cref="IStreamConnection{TMessage}"/> 抽象的上层可通过
+/// <c>connection is ISessionAwareStreamConnection&lt;TMessage&gt; sessionAware</c> 探测后使用。
+///
+/// 现有 <see cref="IStreamConnection{TMessage}.SendAsync"/>（入队即完成、断线后由新会话
+/// 续发）与 <see cref="IStreamConnection{TMessage}.GetMessages"/>（跨重连稳定流）语义不变，
+/// 两套发送按入队顺序共享同一条 FIFO。
+/// </summary>
+/// <typeparam name="TMessage">业务消息类型。</typeparam>
+public interface ISessionAwareStreamConnection<TMessage> : IStreamConnection<TMessage>
+{
+    /// <summary>
+    /// 当前 TCP 会话的编号：每次成功建立 TCP 会话时分配，单调递增、跨重连不复用
+    /// （编号允许有间隔）；无会话（未连接 / 会话拆除 / 停机）时为 0。
+    /// 在 <see cref="IStreamConnection{TMessage}.ConnectionChanged"/> 回调与
+    /// <see cref="IStreamConnection{TMessage}.WaitForConnectedAsync"/> 完成时读取必定得到
+    /// 有效编号（分配先于 Connected 对外发布）；状态离开 Connected 对外可见时已归零。
+    ///
+    /// 仅作发送绑定的"目标声明"；权威校验在发送 worker 出队时进行，读取时刻与使用
+    /// 时刻之间会话仍可能失效——以 <see cref="SendInSessionAsync"/> 的结果为准。
+    /// </summary>
+    long CurrentSessionId { get; }
+
+    /// <summary>
+    /// 发送一条绑定当前会话的消息：任务在<b>整帧全部写入本机 socket</b>（内核接收缓冲，
+    /// 不含对端 ACK）后才成功完成。
+    ///
+    /// 失败语义：会话在整帧写出之前终止（断线重连、对端关闭、Dispose），任务以
+    /// <see cref="SessionExpiredException"/> 失败，该消息<b>不会</b>自动转移到新会话重放，
+    /// 远端处理结果视为未知（见异常说明）；发送中途 socket 故障，任务以底层 socket 异常
+    /// 失败。调用方令牌的提交点：发送 worker 认领本条之前取消会使任务以取消结束且消息
+    /// 不再发送；认领之后（帧已开始写出）取消对结果无副作用，写入由会话令牌控制——
+    /// 取消单条消息不会撕裂帧、不会杀死连接。
+    /// </summary>
+    /// <param name="sessionId">目标会话编号（取自 <see cref="CurrentSessionId"/>）。会话已失效时任务以 <see cref="SessionExpiredException"/> 失败。</param>
+    /// <param name="message">业务消息。</param>
+    /// <param name="ct">调用方取消令牌（仅在提交点之前有效，见备注）。</param>
+    Task SendInSessionAsync(long sessionId, TMessage message, CancellationToken ct = default);
+
+    /// <summary>
+    /// 以异步流方式消费"消息 + 所属会话编号"：<see cref="SessionMessage{TMessage}.SessionId"/>
+    /// 标记每条消息来自哪次 TCP 会话，旧会话解码任务迟到投递的消息带旧编号。
+    ///
+    /// <b>与 <see cref="IStreamConnection{TMessage}.GetMessages"/> 是同一通道的两个竞争
+    /// 消费视图，不是广播</b>：两者同时枚举会互相竞争消费，每条消息只会到达其中一个——
+    /// 请二选一使用。生命周期与 GetMessages 相同：跨重连不中断，仅在停机时结束。
+    /// </summary>
+    IAsyncEnumerable<SessionMessage<TMessage>> GetSessionMessages(CancellationToken ct = default);
+}
diff --git a/src/StreamFrame/SessionExpiredException.cs b/src/StreamFrame/SessionExpiredException.cs
new file mode 100644
index 0000000..2d5d4f5
--- /dev/null
+++ b/src/StreamFrame/SessionExpiredException.cs
@@ -0,0 +1,26 @@
+namespace StreamFrame;
+
+/// <summary>
+/// 会话绑定发送无法完成：消息绑定的会话在整帧写入 socket 之前终止（断线重连、
+/// 对端关闭、Dispose 等）。该消息<b>不会</b>被自动转移到新会话重放。
+///
+/// 收到此异常时的公共保证：本地未取得"整帧已交给本机 socket"的完成确认；本次发送
+/// 以失败结束。<b>远端是否已收到（部分）字节视为未知</b>——TCP 发送失败与取消存在
+/// 竞态边界，调用方应把远端处理结果视为未知，由上层协议的事务关联、幂等或恢复流程
+/// 兜底，而不是假设远端一定未收到。
+/// </summary>
+public sealed class SessionExpiredException : Exception
+{
+    /// <summary>创建会话失效异常。</summary>
+    /// <param name="sessionId">已失效的会话编号。</param>
+    /// <param name="message">异常消息。</param>
+    /// <param name="innerException">导致会话终止的底层异常（可选）。</param>
+    public SessionExpiredException(long sessionId, string message, Exception? innerException = null)
+        : base(message, innerException)
+    {
+        SessionId = sessionId;
+    }
+
+    /// <summary>已失效的会话编号。</summary>
+    public long SessionId { get; }
+}
diff --git a/src/StreamFrame/SessionMessage.cs b/src/StreamFrame/SessionMessage.cs
new file mode 100644
index 0000000..37da15d
--- /dev/null
+++ b/src/StreamFrame/SessionMessage.cs
@@ -0,0 +1,24 @@
+namespace StreamFrame;
+
+/// <summary>
+/// 带会话归属的业务消息：<see cref="ISessionAwareStreamConnection{TMessage}.GetSessionMessages"/>
+/// 的产出元素。旧会话的解码任务在会话拆除后迟到投递的消息，仍携带旧会话的编号。
+/// </summary>
+/// <typeparam name="TMessage">业务消息类型。</typeparam>
+public readonly record struct SessionMessage<TMessage>
+{
+    /// <summary>创建带会话归属的消息。</summary>
+    /// <param name="sessionId">消息所属的 TCP 会话编号（会话建立时分配，单调递增、不复用）。</param>
+    /// <param name="message">业务消息。</param>
+    public SessionMessage(long sessionId, TMessage message)
+    {
+        SessionId = sessionId;
+        Message = message;
+    }
+
+    /// <summary>消息所属的 TCP 会话编号；0 不会出现（0 保留表示"无会话"）。</summary>
+    public long SessionId { get; }
+
+    /// <summary>业务消息。</summary>
+    public TMessage Message { get; }
+}
diff --git a/test/StreamFrame.Tests/FrameDecoderTests.cs b/test/StreamFrame.Tests/FrameDecoderTests.cs
index c78e17a..463a1c8 100644
--- a/test/StreamFrame.Tests/FrameDecoderTests.cs
+++ b/test/StreamFrame.Tests/FrameDecoderTests.cs
@@ -24,7 +24,7 @@ private static IEnumerable<byte[]> Chunk(byte[] bytes, int chunkSize)
 
     private sealed record DecoderHarness(
         FrameDecoder<string> Decoder,
-        Channel<string> Relay,
+        Channel<SessionMessage<string>> Relay,
         Pipe Pipe,
         List<FrameErrorEventArgs> Errors,
         List<string> Messages,
@@ -41,7 +41,7 @@ private static DecoderHarness CreateDecoder(
         int incompleteFrameTimeoutMs = 0)
     {
         var errors = new List<FrameErrorEventArgs>();
-        var relay = Channel.CreateUnbounded<string>(new UnboundedChannelOptions
+        var relay = Channel.CreateUnbounded<SessionMessage<string>>(new UnboundedChannelOptions
         {
             SingleReader = false,
             SingleWriter = true,
@@ -49,6 +49,7 @@ private static DecoderHarness CreateDecoder(
         var pipe = new Pipe();
         var decoder = new FrameDecoder<string>(
             pipe.Reader, framing, codec ?? StringCodec.Instance, relay,
+            sessionId: 1,
             maxIncompleteFrameBytes ?? framing.MaxPayloadBytes + 4096,
             incompleteFrameTimeoutMs,
             policy,
@@ -62,8 +63,8 @@ private static void StartDrain(DecoderHarness h)
         {
             try
             {
-                await foreach (var message in h.Relay.Reader.ReadAllAsync(h.Done.Token))
-                    h.Messages.Add(message);
+                await foreach (var envelope in h.Relay.Reader.ReadAllAsync(h.Done.Token))
+                    h.Messages.Add(envelope.Message);
             }
             catch (OperationCanceledException)
             {
@@ -181,7 +182,7 @@ public async Task Decoder_DoesNotCompleteRelay_OnStreamEnd()
         h.Pipe.Writer.Complete();
         await decodeTask;
 
-        Assert.True(h.Relay.Writer.TryWrite("still-alive"));
+        Assert.True(h.Relay.Writer.TryWrite(new SessionMessage<string>(1, "still-alive")));
         await WaitForMessagesAsync(h, 2);
         lock (h.Messages)
             Assert.Contains("still-alive", h.Messages);
diff --git a/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs b/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs
index 2500609..b0e58b9 100644
--- a/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs
+++ b/test/StreamFrame.Tests/IncompleteFrameTimeoutTests.cs
@@ -299,7 +299,7 @@ public async Task RepeatedTimeoutCycles_RemainIdempotent()
         {
             using var client = new TcpClient();
             await ConnectWithRetryAsync(client, port);
-            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000);
             await client.GetStream().WriteAsync(new byte[] { 0x00, 0x0F });
 
             var expected = cycle + 1;
diff --git a/test/StreamFrame.Tests/SessionAwareConnectionTests.cs b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs
new file mode 100644
index 0000000..5ba77fb
--- /dev/null
+++ b/test/StreamFrame.Tests/SessionAwareConnectionTests.cs
@@ -0,0 +1,722 @@
+using System.Buffers;
+using System.Buffers.Binary;
+using System.Net;
+using System.Net.Sockets;
+using System.Text;
+using StreamFrame;
+
+namespace StreamFrame.Tests;
+
+/// <summary>
+/// 会话感知收发（ISessionAwareStreamConnection）的连接级端到端测试（真实 TCP 回环）：
+/// 会话编号的线性化与生命周期、绑定会话发送的完成/失败语义（整帧写完才完成、不跨会话重放、
+/// 拆除立即失败、调用方取消的提交点）、接收视图带会话编号、与普通 SendAsync 的行为对照。
+///
+/// 时序技巧：Windows 回环对背压几乎免疫（对端不读、RCVBUF 调小也能全量灌入），
+/// "写阻塞"用 <see cref="DelayedCodec"/> 在 worker 的编码阶段制造确定性停顿。
+/// </summary>
+public class SessionAwareConnectionTests
+{
+    /// <summary>编码停顿的测试 codec：让发送 worker 在写出前可预测地停住。</summary>
+    private sealed class DelayedCodec : ICodec<string>
+    {
+        private readonly int _encodeDelayMs;
+
+        public DelayedCodec(int encodeDelayMs) => _encodeDelayMs = encodeDelayMs;
+
+        public string Decode(in ReadOnlySequence<byte> frame, CancellationToken ct = default)
+            => StringCodec.Instance.Decode(frame, ct);
+
+        public void Encode(string message, IBufferWriter<byte> writer, CancellationToken ct = default)
+        {
+            Thread.Sleep(_encodeDelayMs);
+            StringCodec.Instance.Encode(message, writer, ct);
+        }
+    }
+
+    private static int GetFreePort()
+    {
+        var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+        listener.Stop();
+        return port;
+    }
+
+    private static byte[] Frame(string payload)
+    {
+        var bytes = Encoding.UTF8.GetBytes(payload);
+        var frame = new byte[4 + bytes.Length];
+        BinaryPrimitives.WriteInt32BigEndian(frame.AsSpan(0, 4), bytes.Length);
+        bytes.CopyTo(frame.AsSpan(4));
+        return frame;
+    }
+
+    private static async Task WaitForStateAsync(
+        StreamConnection<string> connection,
+        Func<ConnectionState, bool> predicate,
+        int timeoutMs = 5000)
+    {
+        var deadline = TestClock.TickCount64 + timeoutMs;
+        while (TestClock.TickCount64 < deadline)
+        {
+            if (predicate(connection.State))
+                return;
+            await Task.Delay(20);
+        }
+
+        Assert.True(predicate(connection.State), $"等待连接状态超时（{timeoutMs}ms），当前 {connection.State}。");
+    }
+
+    private static async Task WaitForAsync(Func<bool> condition, int timeoutMs = 5000, string? what = null)
+    {
+        var deadline = TestClock.TickCount64 + timeoutMs;
+        while (TestClock.TickCount64 < deadline)
+        {
+            if (condition())
+                return;
+            await Task.Delay(20);
+        }
+
+        Assert.True(condition(), $"等待条件超时（{timeoutMs}ms）：{what ?? "未描述"}。");
+    }
+
+    private static async Task ConnectWithRetryAsync(TcpClient client, int port)
+    {
+        for (var attempt = 0; ; attempt++)
+        {
+            try
+            {
+                await client.ConnectAsync(IPAddress.Loopback, port);
+                return;
+            }
+            catch (SocketException) when (attempt < 10)
+            {
+                await Task.Delay(200);
+            }
+        }
+    }
+
+    private static StreamConnection<string> CreateServer(
+        int port, StreamConnectionOptions? options = null, ICodec<string>? codec = null)
+        => new(
+            new LengthPrefixFramer(),
+            codec ?? StringCodec.Instance,
+            IPAddress.Loopback,
+            port,
+            isActive: false,
+            options ?? new StreamConnectionOptions());
+
+    /// <summary>从流中读一帧并返回其负载（长度前缀 + 正文）。</summary>
+    private static async Task<string> ReadFrameAsync(NetworkStream stream, int timeoutMs = 10_000)
+    {
+        var header = await ReadExactlyAsync(stream, 4, timeoutMs);
+        var length = BinaryPrimitives.ReadInt32BigEndian(header);
+        Assert.True(length is > 0 and < 1024 * 1024, $"帧长度异常：{length}。");
+        var body = await ReadExactlyAsync(stream, length, timeoutMs);
+        return Encoding.UTF8.GetString(body);
+    }
+
+    /// <summary>精确读 n 字节（超时即失败）。net48 无 Task.WaitAsync，统一用 WhenAny 模式。</summary>
+    private static async Task<byte[]> ReadExactlyAsync(NetworkStream stream, int n, int timeoutMs = 10_000)
+    {
+        var buffer = new byte[n];
+        var read = 0;
+        while (read < n)
+        {
+            var readTask = stream.ReadAsync(buffer, read, n - read);
+            var winner = await Task.WhenAny(readTask, Task.Delay(timeoutMs));
+            Assert.True(ReferenceEquals(readTask, winner), $"读字节超时（{timeoutMs}ms）。");
+            var count = await readTask;
+            Assert.True(count > 0, "对端提前关闭。");
+            read += count;
+        }
+
+        return buffer;
+    }
+
+    /// <summary>断言窗口期内没有任何字节到达。</summary>
+    private static async Task AssertNoBytesAsync(NetworkStream stream, int windowMs = 400)
+    {
+        var buffer = new byte[256];
+        var readTask = stream.ReadAsync(buffer, 0, buffer.Length);
+        var winner = await Task.WhenAny(readTask, Task.Delay(windowMs));
+        if (ReferenceEquals(winner, readTask))
+            Assert.Fail($"不应有字节到达，实际收到 {await readTask} 字节。");
+    }
+
+    /// <summary>等待任务在时限内成功完成，超时直接失败（net48 无 Task.WaitAsync）。</summary>
+    private static async Task AwaitDoneAsync(Task task, int timeoutMs = 8000, string? what = null)
+    {
+        var winner = await Task.WhenAny(task, Task.Delay(timeoutMs));
+        Assert.True(ReferenceEquals(task, winner), $"任务超时（{timeoutMs}ms）：{what ?? "未描述"}。");
+        await task;
+    }
+
+    /// <summary>等待任务在时限内以异常结束并返回该异常，超时/成功完成直接失败。</summary>
+    private static async Task<Exception> AwaitFailureAsync(Task task, int timeoutMs = 8000, string? what = null)
+    {
+        var winner = await Task.WhenAny(task, Task.Delay(timeoutMs));
+        Assert.True(ReferenceEquals(task, winner), $"任务超时（{timeoutMs}ms）：{what ?? "未描述"}。");
+        return await Assert.ThrowsAnyAsync<Exception>(() => task);
+    }
+
+    [Fact]
+    public void Implements_SessionAware_Interface()
+        => Assert.IsAssignableFrom<ISessionAwareStreamConnection<string>>(
+            CreateServer(GetFreePort()));
+
+    [Fact]
+    public async Task SessionId_ValidAtLinearizationPoints_AndNeverReused()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        Assert.Equal(0, server.CurrentSessionId);
+
+        var inCallback = new List<(ConnectionState State, long Id)>();
+        server.ConnectionChanged += (_, s) =>
+        {
+            lock (inCallback)
+                inCallback.Add((s, server.CurrentSessionId));
+        };
+        server.Start(CancellationToken.None);
+
+        long id1;
+        using (var client1 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client1, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+
+            id1 = server.CurrentSessionId;
+            Assert.NotEqual(0, id1);
+            lock (inCallback)
+            {
+                // Connected 回调可见时编号必须已分配（线性化点）
+                var connected = inCallback.First(x => x.State == ConnectionState.Connected);
+                Assert.Equal(id1, connected.Id);
+            }
+
+            client1.Dispose(); // 触发重连
+        }
+
+        await WaitForStateAsync(server, s => s != ConnectionState.Connected);
+        Assert.Equal(0, server.CurrentSessionId); // 离开 Connected 可见时已归零
+
+        long id2;
+        using (var client2 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client2, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000);
+            id2 = server.CurrentSessionId;
+            Assert.True(id2 > id1, $"会话编号必须单调递增：{id1} → {id2}");
+        }
+
+        await server.DisposeAsync();
+        Assert.Equal(0, server.CurrentSessionId); // 停机后归零
+    }
+
+    [Fact]
+    public async Task SendInSession_CompletesOnlyAfterFullFrameWritten()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port, codec: new DelayedCodec(500));
+        server.Start(CancellationToken.None);
+
+        using var client = new TcpClient();
+        await ConnectWithRetryAsync(client, port);
+        await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+        var stream = client.GetStream();
+
+        var sendTask = server.SendInSessionAsync(server.CurrentSessionId, "delayed");
+        await Task.Delay(200); // worker 仍停在编码阶段：字节未写，任务不可能完成
+        Assert.False(sendTask.IsCompleted, "整帧尚未写出，任务不应完成（不能入队即完成）。");
+
+        var message = await ReadFrameAsync(stream); // 对端收完整帧
+        await AwaitDoneAsync(sendTask, 5000, "整帧写完");
+
+        Assert.Equal("delayed", message);
+        Assert.Equal(TaskStatus.RanToCompletion, sendTask.Status);
+    }
+
+    [Fact]
+    public async Task SessionTeardown_PendingBoundSends_FailPromptly()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(
+            port, new StreamConnectionOptions { SendQueueCapacity = 1 }, new DelayedCodec(400));
+        server.Start(CancellationToken.None);
+
+        using var client = new TcpClient();
+        await ConnectWithRetryAsync(client, port);
+        await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+        var id = server.CurrentSessionId;
+
+        var writingTask = server.SendInSessionAsync(id, "writing"); // worker 认领后停在编码阶段
+        var tasks = new List<Task>
+        {
+            server.SendInSessionAsync(id, "queued"),   // 占满队列（容量 1）
+            server.SendInSessionAsync(id, "waiting"),  // 队列满，WriteAsync 等待空位
+        };
+        await Task.Delay(100); // 让第 1 条进入编码
+
+        server.Reconnect(); // 会话拆除：无需等重连完成，挂起发送必须立即失败
+
+        foreach (var task in tasks)
+        {
+            var ex = await AwaitFailureAsync(task, 3000, "挂起的会话绑定发送及时失败");
+            Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。");
+        }
+
+        // 正在写出的条目是拆除竞争的合法双结局：整帧恰好写完 → 成功；否则会话失效失败。
+        // 两者都符合契约，但必须在时限内终结（不悬挂）
+        var writingDone = await Task.WhenAny(writingTask, Task.Delay(3000));
+        Assert.True(ReferenceEquals(writingDone, writingTask), "写出中条目应在时限内终结。");
+        if (writingTask.Status != TaskStatus.RanToCompletion)
+        {
+            var ex = await Assert.ThrowsAnyAsync<Exception>(() => writingTask);
+            Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。");
+        }
+    }
+
+    [Fact]
+    public async Task SessionSwitch_BoundSendFails_NoReplay_PlainSendStillReplays()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port, codec: new DelayedCodec(400));
+        server.Start(CancellationToken.None);
+
+        using (var client1 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client1, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            var id1 = server.CurrentSessionId;
+
+            var writingTask = server.SendInSessionAsync(id1, "writing");
+            var bound = new List<Task> { server.SendInSessionAsync(id1, "bound-drops") };
+            _ = server.SendAsync("plain-replays"); // 普通发送：跨会话续发（既有语义）
+
+            await Task.Delay(100);
+            server.Reconnect(); // 触发会话切换
+
+            foreach (var task in bound)
+            {
+                var ex = await AwaitFailureAsync(task, 3000, "会话切换后旧绑定发送失败");
+                Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                    $"失败类型意外：{ex.GetType().Name}。");
+            }
+
+            // 正在写出的条目：拆除竞争的合法双结局（整帧恰好写完→成功；否则失效失败），不悬挂即可
+            var writingDone = await Task.WhenAny(writingTask, Task.Delay(3000));
+            Assert.True(ReferenceEquals(writingDone, writingTask), "写出中条目应在时限内终结。");
+            if (writingTask.Status != TaskStatus.RanToCompletion)
+            {
+                var ex = await Assert.ThrowsAnyAsync<Exception>(() => writingTask);
+                Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                    $"失败类型意外：{ex.GetType().Name}。");
+            }
+        }
+
+        // 新会话：绑定旧编号的发送直接失败；普通消息由新 worker 续发；绑定消息绝不重放
+        using (var client2 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client2, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000);
+            var id2 = server.CurrentSessionId;
+            Assert.NotEqual(0, id2);
+
+            await Assert.ThrowsAsync<SessionExpiredException>(
+                () => server.SendInSessionAsync(id2 + 100, "stale"));
+
+            await server.SendInSessionAsync(id2, "new-session-ok");
+
+            var stream = client2.GetStream();
+            // 队列顺序：bound-drops（跳过）→ plain-replays → new-session-ok
+            Assert.Equal("plain-replays", await ReadFrameAsync(stream));
+            Assert.Equal("new-session-ok", await ReadFrameAsync(stream));
+            await AssertNoBytesAsync(stream); // 旧会话的绑定消息没有转移到新会话
+        }
+    }
+
+    [Fact]
+    public async Task Dispose_PendingBoundSends_AllFail_Promptly()
+    {
+        var port = GetFreePort();
+        var server = CreateServer(port, codec: new DelayedCodec(400));
+        server.Start(CancellationToken.None);
+
+        using (var client = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            var id = server.CurrentSessionId;
+
+            var writingTask = server.SendInSessionAsync(id, "writing");
+            var tasks = new List<Task>();
+            for (var i = 1; i < 4; i++)
+                tasks.Add(server.SendInSessionAsync(id, $"pending-{i}"));
+            await Task.Delay(100);
+
+            await server.DisposeAsync(); // 停机：挂起的会话绑定发送必须全部异常收尾
+
+            foreach (var task in tasks)
+            {
+                var ex = await AwaitFailureAsync(task, 3000, "Dispose 后挂起发送异常收尾");
+                Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                    $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。");
+            }
+
+            // 正在写出的条目：拆除竞争的合法双结局（写完→成功；否则失效失败），不悬挂即可
+            var writingDone = await Task.WhenAny(writingTask, Task.Delay(3000));
+            Assert.True(ReferenceEquals(writingDone, writingTask), "写出中条目应在时限内终结。");
+            if (writingTask.Status != TaskStatus.RanToCompletion)
+            {
+                var ex = await Assert.ThrowsAnyAsync<Exception>(() => writingTask);
+                Assert.True(ex is SessionExpiredException or SocketException or OperationCanceledException,
+                    $"失败类型应为会话失效/socket 故障，实际 {ex.GetType().Name}。");
+            }
+        }
+
+        Assert.True(server.IsDisposed);
+    }
+
+    [Fact]
+    public async Task CallerCancel_BeforeCommit_CancelsAndNotSent()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port, codec: new DelayedCodec(500));
+        server.Start(CancellationToken.None);
+
+        using var client = new TcpClient();
+        await ConnectWithRetryAsync(client, port);
+        await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+        var stream = client.GetStream();
+
+        var firstTask = server.SendInSessionAsync(server.CurrentSessionId, "first");
+        await Task.Delay(100); // worker 认领第 1 条并停在编码
+
+        using var cts = new CancellationTokenSource();
+        var cancelTask = server.SendInSessionAsync(server.CurrentSessionId, "cancel-me", cts.Token);
+        cts.Cancel(); // 提交点之前：任务及时以取消结束，不等待第 1 条写完
+
+        await Assert.ThrowsAnyAsync<OperationCanceledException>(
+            () => AwaitDoneAsync(cancelTask, 2000, "提交点前取消"));
+
+        // 对端读走第 1 帧后，第 2 条（已取消）绝不出现
+        Assert.Equal("first", await ReadFrameAsync(stream));
+        await AwaitDoneAsync(firstTask, 5000, "第 1 条整帧写完");
+        await AssertNoBytesAsync(stream);
+    }
+
+    [Fact]
+    public async Task CallerCancel_AfterCommit_HasNoEffect()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        server.Start(CancellationToken.None);
+
+        using var client = new TcpClient();
+        await ConnectWithRetryAsync(client, port);
+        await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+
+        using var cts = new CancellationTokenSource();
+        var task = server.SendInSessionAsync(server.CurrentSessionId, "committed", cts.Token);
+        await AwaitDoneAsync(task, 5000, "提交后发送完成"); // 已成功完成（已过提交点）
+
+        cts.Cancel(); // 迟到的取消不得翻转结果
+        await Task.Delay(200);
+        Assert.Equal(TaskStatus.RanToCompletion, task.Status);
+    }
+
+    [Fact]
+    public async Task GetSessionMessages_CarrySessionId_AcrossReconnects()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        var received = new List<SessionMessage<string>>();
+        var drainCts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var message in server.GetSessionMessages(drainCts.Token))
+                    lock (received)
+                        received.Add(message);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+        });
+        server.Start(CancellationToken.None);
+
+        long id1;
+        using (var client1 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client1, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            id1 = server.CurrentSessionId;
+
+            var stream = client1.GetStream();
+            await stream.WriteAsync(Frame("s1a"));
+            await stream.WriteAsync(Frame("s1b"));
+            await WaitForAsync(() => { lock (received) return received.Count == 2; }, what: "会话 1 消息送达");
+        }
+
+        using (var client2 = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client2, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000);
+            var id2 = server.CurrentSessionId;
+            Assert.True(id2 > id1);
+
+            await client2.GetStream().WriteAsync(Frame("s2a"));
+            await WaitForAsync(() => { lock (received) return received.Count == 3; }, what: "会话 2 消息送达");
+        }
+
+        lock (received)
+        {
+            Assert.Equal(new[] { ("s1a", id1), ("s1b", id1), ("s2a", received[2].SessionId) },
+                received.Select(m => (m.Message, m.SessionId)).ToArray());
+            Assert.True(received[2].SessionId > id1, "新会话消息必须带新编号。");
+        }
+    }
+
+    [Fact]
+    public async Task ConsumerViews_Compete_NotBroadcast()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(10));
+        var fromPlain = new List<string>();
+        var fromSession = new List<SessionMessage<string>>();
+
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var m in server.GetMessages(cts.Token))
+                    lock (fromPlain)
+                        fromPlain.Add(m);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+        });
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var m in server.GetSessionMessages(cts.Token))
+                    lock (fromSession)
+                        fromSession.Add(m);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+        });
+        server.Start(CancellationToken.None);
+
+        using (var client = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            await client.GetStream().WriteAsync(Frame("only-one"));
+        }
+
+        await WaitForAsync(() =>
+        {
+            lock (fromPlain)
+            lock (fromSession)
+                return fromPlain.Count + fromSession.Count == 1;
+        }, what: "消息恰好被两个视图之一消费");
+
+        lock (fromPlain)
+        lock (fromSession)
+            Assert.Equal(1, fromPlain.Count + fromSession.Count);
+    }
+
+    [Fact]
+    public async Task GetMessages_And_GetSessionMessages_SameSequence_AcrossRuns()
+    {
+        // 单消费者视图的等价性在独立运行中比较（同一连接上同时枚举会竞争消费）
+        var plain = await CollectViaGetMessagesAsync();
+        var sessionView = await CollectViaGetSessionMessagesAsync();
+
+        Assert.Equal(plain, sessionView.Messages);
+        Assert.All(sessionView.Ids, id => Assert.True(id > 0));
+        Assert.Single(sessionView.Ids.Distinct()); // 单次运行内全部消息来自同一会话
+    }
+
+    private async Task<List<string>> CollectViaGetMessagesAsync()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        var received = new List<string>();
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var m in server.GetMessages(cts.Token))
+                    lock (received)
+                        received.Add(m);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+        });
+        await SendThreeFramesAsync(server, port, () => { lock (received) return received.Count == 3; });
+        return received;
+    }
+
+    private async Task<(List<string> Messages, List<long> Ids)> CollectViaGetSessionMessagesAsync()
+    {
+        var port = GetFreePort();
+        await using var server = CreateServer(port);
+        var received = new List<SessionMessage<string>>();
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await foreach (var m in server.GetSessionMessages(cts.Token))
+                    lock (received)
+                        received.Add(m);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+        });
+        await SendThreeFramesAsync(server, port, () => { lock (received) return received.Count == 3; });
+
+        lock (received)
+            return (received.Select(m => m.Message).ToList(), received.Select(m => m.SessionId).ToList());
+    }
+
+    private async Task SendThreeFramesAsync(
+        StreamConnection<string> server, int port, Func<bool> receivedAll)
+    {
+        server.Start(CancellationToken.None);
+        using var client = new TcpClient();
+        await ConnectWithRetryAsync(client, port);
+        await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+        var stream = client.GetStream();
+        await stream.WriteAsync(Frame("m1"));
+        await stream.WriteAsync(Frame("m2"));
+        await stream.WriteAsync(Frame("m3"));
+        await WaitForAsync(receivedAll, what: "三个帧送达");
+    }
+
+    [Fact]
+    public async Task TeardownVersusRegister_Stress_AllTasksEnd_NoCrossSessionSend()
+    {
+        const int Rounds = 15;
+        const int SendsPerRound = 4;
+
+        var port = GetFreePort();
+        await using var server = CreateServer(port, codec: new DelayedCodec(30));
+        server.Start(CancellationToken.None);
+
+        for (var round = 0; round < Rounds; round++)
+        {
+            var peerBytes = new MemoryStream();
+            var successes = 0;
+            using var client = new TcpClient();
+            await ConnectWithRetryAsync(client, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected);
+            var stream = client.GetStream();
+
+            // 后台持续读（模拟对端消费）
+            var readCts = new CancellationTokenSource(TimeSpan.FromSeconds(15));
+            var readTask = Task.Run(async () =>
+            {
+                var buffer = new byte[16384];
+                try
+                {
+                    while (true)
+                    {
+#if NET48
+                        var n = await stream.ReadAsync(buffer, 0, buffer.Length, readCts.Token);
+#else
+                        var n = await stream.ReadAsync(buffer, readCts.Token);
+#endif
+                        if (n == 0)
+                            break;
+                        lock (peerBytes)
+                            peerBytes.Write(buffer, 0, n);
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                }
+            });
+
+            var sends = Enumerable.Range(0, SendsPerRound)
+                .Select(_ => server.SendInSessionAsync(server.CurrentSessionId, $"r{round}"))
+                .ToArray();
+
+            // 与拆除并发：不等发送结果，直接切换会话（按轮次交替两种拆除路径）
+            if (round % 2 == 0)
+                server.Reconnect();
+            else
+                client.Dispose(); // 对端关闭路径
+
+            foreach (var send in sends)
+            {
+                var done = await Task.WhenAny(send, Task.Delay(8000));
+                Assert.True(ReferenceEquals(send, done), "发送任务在时限内未结束（悬挂）。");
+                if (send.Status == TaskStatus.RanToCompletion)
+                    successes++;
+                else
+                    Assert.True(send.Exception?.InnerExceptions[0] is SessionExpiredException
+                        or SocketException or OperationCanceledException,
+                        $"失败类型意外：{send.Exception?.InnerExceptions[0].GetType().Name}");
+            }
+
+            // 停读前留出排空窗口：成功的帧可能还在对端接收缓冲里未被读出
+            await Task.Delay(300);
+            readCts.Cancel();
+            _ = await Task.WhenAny(readTask, Task.Delay(2000)); // 尽力收尾，超时不再等待
+
+            // 成功核对（仅 Reconnect 轮次：对端全程在读，成功帧必然可达且完整）：
+            // 客户端主动 Dispose 轮次不对账——本地关闭时未读数据随 RST 丢弃属 TCP 语义，
+            // "写入本机 socket"之外的投递结果不在本 API 的公共保证内
+            if (round % 2 == 0)
+            {
+                lock (peerBytes)
+                {
+                    var bytes = peerBytes.ToArray();
+                    var offset = 0;
+                    var frames = 0;
+                    while (offset + 4 <= bytes.Length)
+                    {
+                        var length = BinaryPrimitives.ReadInt32BigEndian(bytes.AsSpan(offset, 4));
+                        Assert.True(length is > 0 and < 1024, $"帧长度异常：{length}。");
+                        if (offset + 4 + length > bytes.Length)
+                            break; // 拆除时刻正在写的半帧（其任务必然已失败）
+                        offset += 4 + length;
+                        frames++;
+                    }
+
+                    Assert.True(successes <= frames,
+                        $"成功 {successes} 次发送但对端仅收到 {frames} 个完整帧——存在未完整写出的\"成功\"。");
+                }
+            }
+
+            // 下一轮前的清理：等待离开 Connected（对端关闭路径由 FIN 触发）
+            if (round % 2 == 1)
+                await WaitForStateAsync(server, s => s != ConnectionState.Connected, timeoutMs: 8000);
+        }
+
+        // 压测后连接仍健康：新会话正常收发
+        using (var client = new TcpClient())
+        {
+            await ConnectWithRetryAsync(client, port);
+            await WaitForStateAsync(server, s => s == ConnectionState.Connected, timeoutMs: 8000);
+            var task = server.SendInSessionAsync(server.CurrentSessionId, "healthy");
+            Assert.Equal("healthy", await ReadFrameAsync(client.GetStream()));
+            await AwaitDoneAsync(task, 5000, "压测后发送");
+        }
+    }
+}
diff --git a/test/StreamFrame.Tests/TestHostSetup.cs b/test/StreamFrame.Tests/TestHostSetup.cs
new file mode 100644
index 0000000..c81a56f
--- /dev/null
+++ b/test/StreamFrame.Tests/TestHostSetup.cs
@@ -0,0 +1,18 @@
+using System.Runtime.CompilerServices;
+
+namespace StreamFrame.Tests;
+
+/// <summary>
+/// 测试宿主初始化：CI 跑机（2 核）上并行测试里的 Thread.Sleep（编码停顿的测试 codec）
+/// 会阻塞线程池线程，net48 的线程注入速率慢（约 1 个/500ms），曾把其他测试的秒级等待
+/// 拖到超时。抬高最小线程数根治这类环境性抖动（仅影响测试宿主进程）。
+/// </summary>
+internal static class TestHostSetup
+{
+    [ModuleInitializer]
+    internal static void ConfigureThreadPool()
+    {
+        var workers = Math.Max(8, Environment.ProcessorCount * 4);
+        ThreadPool.SetMinThreads(workers, workers);
+    }
+}