fix(websocket): lossless EOF signal + pending sweep (WS-02, CON-02)

Replace the axum/tungstenite pump paths' Notify-based read-EOF signal
with a retained tokio watch channel so a late subscriber observes EOF
regardless of when it fired. Extend the from_wss drop monitor to sweep
the pending map (1 s interval) once EOF is observed, so calls
registered after the initial fail_all also resolve retryable instead
of hanging.

cargo test; cargo clippy --all-targets -- -D warnings (default +
all-features); cargo fmt --check

src/adapters/from_wss.rs

@@ -23,10 +23,18 @@
 //! Reconnect policy (OQ-03 disposition): v1 = none. A connection drop
 //! fails in-flight calls retryable: alkcall's client-side read pump only
 //! routes envelopes and does not observe EOF, so the adapter owns drop
-//! semantics — the [`WssSession`] monitor awaits the WS read pump and
+//! semantics — the [`WssSession`] monitor awaits the WS read pump's EOF
-//! fails all pending calls with retryable `CONNECTION_CLOSED`. Subsequent
+//! signal and fails all pending calls with retryable `CONNECTION_CLOSED`.
-//! handler calls fail on write; reconnect policy is the assembly layer's
+//! The EOF signal is lossless (a retained watch value, WS-02): EOF is
-//! job.
+//! observed even if it fires before the monitor starts, or after the
+//! session was forgotten (fire-and-forget import). The monitor also
+//! sweeps the pending map once a second while the session lives: calls
+//! registered *after* the initial `fail_all` (still possible on the
+//! forgotten-session path, where imports may race the drop) are failed
+//! on the next sweep tick once EOF has been observed. No hang: pendings
+//! resolve retryable regardless of registration-vs-EOF ordering
+//! (CON-02). Subsequent handler calls fail on write; reconnect policy is
+//! the assembly layer's job.
 //!
 //! [ADR-070]: https://docs.rs/alkhttp (docs/architecture/decisions)
 
@@ -43,6 +51,11 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 
 use crate::websocket::split_tungstenite_to_bytes;
 
+/// How often the drop monitor sweeps the pending map for calls
+/// registered after the last `fail_all` (only effective once EOF has
+/// been observed — see the module doc).
+const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
+
 pub struct FromWss {
     endpoint: String,
     auth_token: Option<String>,
@@ -151,18 +164,38 @@ impl WssSession {
                 })?;
         let call_connection = Arc::new(call_connection);
 
-        let (close_tx, close_rx) = tokio::sync::oneshot::channel();
+        let (close_tx, mut close_rx) = tokio::sync::oneshot::channel();
         let pending = Arc::clone(call_connection.pending());
+        let mut eof_rx = pumps.read_eof();
         tokio::spawn(async move {
+            let mut sweep = tokio::time::interval(PENDING_SWEEP_INTERVAL);
+            sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+            loop {
                 tokio::select! {
-                _ = pumps.read_eof() => {},
+                    _ = crate::websocket::wait_for_eof(&mut eof_rx) => break,
-                _ = close_rx => {},
+                    _ = &mut close_rx => break,
-            }
+                    _ = sweep.tick() => {
-            pending.lock().fail_all(CallError::new(
+                        if *eof_rx.borrow() {
+                            pending.lock().fail_all(
+                                CallError::new(
                                     "CONNECTION_CLOSED",
                                     "from_wss connection dropped",
                                     true,
-            ));
+                                ),
+                            );
+                        }
+                    }
+                }
+            }
+            let failure = CallError::new("CONNECTION_CLOSED", "from_wss connection dropped", true);
+            let mut guard = pending.lock();
+            guard.fail_all(failure.clone());
+            if *eof_rx.borrow() {
+                let swept = guard.fail_all(failure);
+                if !swept.is_empty() {
+                    tracing::debug!("from_wss: sweep failed {} late pendings", swept.len());
+                }
+            }
         });
 
         Ok(Self {

src/websocket/byte_adapter.rs

@@ -36,9 +36,6 @@ use std::{
     task::{Context, Poll},
 };
 
-#[cfg(any(test, feature = "wss"))]
-use std::sync::Arc;
-
 use axum::extract::ws::{CloseFrame, Message as AxumMessage, WebSocket};
 use futures::channel::mpsc as futures_mpsc;
 use futures::{SinkExt, StreamExt};
@@ -84,7 +81,7 @@ pub struct WsPumps {
     read_task: tokio::task::JoinHandle<()>,
     write_task: tokio::task::JoinHandle<()>,
     #[cfg(feature = "wss")]
-    read_eof: Arc<tokio::sync::Notify>,
+    read_eof: tokio::sync::watch::Sender<bool>,
 }
 
 impl WsPumps {
@@ -93,12 +90,35 @@ impl WsPumps {
         self.write_task.abort();
     }
 
-    /// Fires when the WS read side reaches EOF (socket close from either
+    /// A lossless receiver for the WS read-EOF signal (socket close from
-    /// side) — used by `from_wss`'s connection-drop monitor to await
+    /// either side): the watch channel retains the latest value, so an
-    /// socket EOF (ADR-070).
+    /// EOF signaled at any point — including before the receiver is
+    /// taken or the observer starts awaiting — is still observed, and
+    /// may be observed repeatedly. Used by `from_wss`'s
+    /// connection-drop monitor (ADR-070); await it with
+    /// [`wait_for_eof`].
     #[cfg(feature = "wss")]
-    pub(crate) async fn read_eof(&self) {
+    pub(crate) fn read_eof(&self) -> tokio::sync::watch::Receiver<bool> {
-        self.read_eof.notified().await;
+        self.read_eof.subscribe()
+    }
+}
+
+/// Resolves once the given WS read-EOF watch receiver observes `true`.
+///
+/// Lossless by construction: a `watch::Receiver` retains the latest
+/// value, so an EOF signaled before this call (or before the receiver
+/// existed) is still observed — the property a one-shot `Notify` lacked
+/// (WS-02). Returns early (treated as EOF) if the sender half is
+/// dropped, e.g. the read pump was aborted.
+#[cfg(feature = "wss")]
+pub(crate) async fn wait_for_eof(rx: &mut tokio::sync::watch::Receiver<bool>) {
+    loop {
+        if *rx.borrow_and_update() {
+            return;
+        }
+        if rx.changed().await.is_err() {
+            return;
+        }
     }
 }
 
@@ -111,11 +131,11 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
     let (write_tx, mut write_rx) = futures_mpsc::channel::<WriteMsg>(WRITE_SLOTS);
 
     #[cfg(feature = "wss")]
-    let read_eof = Arc::new(tokio::sync::Notify::new());
+    let read_eof = tokio::sync::watch::channel(false).0;
 
     let write_tx_for_read = write_tx.clone();
     #[cfg(feature = "wss")]
-    let read_eof_for_task = Arc::clone(&read_eof);
+    let read_eof_for_task = read_eof.clone();
     let read_task = tokio::spawn(async move {
         while let Some(msg) = ws_stream.next().await {
             match msg {
@@ -136,7 +156,9 @@ pub fn split_ws_to_bytes(socket: WebSocket) -> (WsByteStream, WsPumps) {
             }
         }
         #[cfg(feature = "wss")]
-        read_eof_for_task.notify_waiters();
+        {
+            let _ = read_eof_for_task.send(true);
+        }
     });
 
     let write_task = tokio::spawn(async move {
@@ -293,10 +315,10 @@ where
     let (read_tx, read_rx) = mpsc::channel::<Vec<u8>>(READ_SLOTS);
     let (write_tx, mut write_rx) = futures_mpsc::channel::<WriteMsg>(WRITE_SLOTS);
 
-    let read_eof = Arc::new(tokio::sync::Notify::new());
+    let read_eof = tokio::sync::watch::channel(false).0;
 
     let write_tx_for_read = write_tx.clone();
-    let read_eof_for_task = Arc::clone(&read_eof);
+    let read_eof_for_task = read_eof.clone();
     let read_task = tokio::spawn(async move {
         while let Some(msg) = ws_stream.next().await {
             match msg {
@@ -316,7 +338,7 @@ where
                 Ok(_) => {}
             }
         }
-        read_eof_for_task.notify_waiters();
+        let _ = read_eof_for_task.send(true);
     });
 
     let write_task = tokio::spawn(async move {

src/websocket/mod.rs

@@ -18,6 +18,8 @@ pub use byte_adapter::{
 
 #[cfg(any(test, feature = "wss"))]
 pub use byte_adapter::split_tungstenite_to_bytes;
+#[cfg(feature = "wss")]
+pub(crate) use byte_adapter::wait_for_eof;
 pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler};
 
 #[cfg(any(test, feature = "test-support"))]