feat: 会话感知收发 ISessionAwareStreamConnection

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 上限保护的缓冲快照。纯增量、默认行为不变。
 
 ### 变更

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
 

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` 是同一通道的两个竞争消费视图，同时枚举会互相分流——请二选一使用。
 
 ## 诊断与调试
 

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)
                     {