fix(adapters): dead-connection fast-fail + bounded from_wss sweep exit (CON-18)

The review-001 CON-02 monitor swept the pending map once EOF held but
never exited: every fire-and-forget import whose peer died left a
spawned task + pending map + watch receiver alive for the process
lifetime, and once the read pump ended the watch sender dropped so the
old select's 'changed()' error branch spun.

- replaced the 1s never-exiting sweep with a 50ms post-EOF drain
  (fast-fail) and a bounded grace window: the monitor ends after
  PENDING_SWEEP_MAX_POST_EOF (8) consecutive empty drains (~400ms)
- post-EOF registrations now resolve in ~50ms with retryable
  CONNECTION_CLOSED instead of waiting up to 1s for the next sweep
- removed the silent busy-spin: the loop now only wakes on the sweep
  tick or the session-close signal (close path still fails all + exits)
- retained the monitor JoinHandle on WssDropMonitor (test builds) and
  added a lifecycle test asserting the monitor joins after EOF+grace
- tightened call_registered_after_eof test: 500ms bound + CONNECTION_
  CLOSED code assertion (was a pre-sleep + 5s timeout)

Verification:
- cargo test --features wss: 358 passed, 0 failed
- cargo test --all-features: 454 passed, 0 failed
- cargo clippy --all-targets -- -D warnings (default, wss, all): clean
- cargo fmt --check: clean
- scripts/verify.sh + verify.sh --all-features: VERIFY OK

src/adapters/from_wss.rs

@@ -36,14 +36,20 @@
 //! signal and fails all pending calls with retryable `CONNECTION_CLOSED`.
 //! The EOF signal is lossless (a retained watch value, WS-02): EOF is
 //! 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.
+//! session was forgotten (fire-and-forget import). Once EOF has been
+//! observed, registrations that land in the pending map *after* the
+//! initial `fail_all` (still possible on the forgotten-session path,
+//! where imports may race the drop) are failed fast: the monitor
+//! watches the pending map at a short interval during a bounded
+//! post-EOF window and drains any entry that lands — no wait for the
+//! next one-second sweep tick (CON-18). No hang: pendings resolve
+//! retryable regardless of registration-vs-EOF ordering (CON-02), and
+//! when the map stays drained past the grace window the monitor task
+//! ends: no per-dead-session permanent task remains (the WS-02
+//! losslessness invariant holds — fast-fail is an additional
+//! resolution path, never a replacement for the retained watch
+//! signal). Subsequent handler calls fail on write; reconnect policy
+//! is the assembly layer's job.
 //!
 //! Session teardown (explicit limitation, review-001 CON-09): `import()`
 //! detaches the session fire-and-forget — there is no close/shutdown
@@ -69,10 +75,15 @@ 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);
+/// How often the drop monitor drains the pending map once EOF has been
+/// observed, failing `fail_all`-after registrations fast (CON-18).
+const PENDING_SWEEP_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
+
+/// The bounded post-EOF grace window, as a number of consecutive idle
+/// drains: when this many drains in a row find the pending map empty,
+/// the monitor ends — no per-dead-session task remains for the process
+/// lifetime (CON-18).
+const PENDING_SWEEP_MAX_POST_EOF: u32 = 8;
 
 fn connection_closed_error() -> CallError {
     CallError::new("CONNECTION_CLOSED", "from_wss connection dropped", true)
@@ -151,10 +162,31 @@ pub struct WssSession {
     _monitor: WssDropMonitor,
 }
 
+impl WssSession {
+    /// The drop monitor's task handle, for observability: tests assert
+    /// the monitor ends after EOF + the bounded grace window (CON-18).
+    /// The handle lives on the session; dropping the session detaches
+    /// the handle (tokio semantics) — the close signal already ends
+    /// the task on explicit drop, and the bounded sweep ends it on EOF.
+    #[cfg(test)]
+    fn monitor_handle(&mut self) -> &mut tokio::task::JoinHandle<()> {
+        &mut self._monitor.monitor_task
+    }
+}
+
 /// The drop monitor: fires `fail_all` on WS read EOF or on explicit
 /// session close (the session's `Drop` impl sends the close signal).
+/// When the session drops, the close signal resolves `fail_all` and
+/// the task ends; on read-EOF the bounded post-EOF sweep ends the task
+/// (CON-18). The task handle is retained on the session in test builds
+/// so the monitor's lifecycle stays observable; in non-test builds it
+/// is detached (tokio semantics) — the close signal already ends the
+/// task on explicit drop, and the bounded sweep ends it on EOF.
+#[cfg_attr(not(test), allow(dead_code))]
 struct WssDropMonitor {
     close_tx: Option<tokio::sync::oneshot::Sender<()>>,
+    #[cfg(test)]
+    monitor_task: tokio::task::JoinHandle<()>,
 }
 
 impl Drop for WssDropMonitor {
@@ -243,29 +275,36 @@ impl WssSession {
 
         let (close_tx, mut close_rx) = tokio::sync::oneshot::channel();
         let pending = Arc::clone(call_connection.pending());
-        tokio::spawn(async move {
+        #[cfg_attr(not(test), allow(unused_variables))]
+        let monitor_task = tokio::spawn(async move {
             let mut eof_rx = pumps.read_eof();
             let mut sweep = tokio::time::interval(PENDING_SWEEP_INTERVAL);
             sweep.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-            let mut eof_observed = *eof_rx.borrow_and_update();
+            let eof_observed = *eof_rx.borrow_and_update();
             if eof_observed {
                 pending.lock().fail_all(connection_closed_error());
             }
+            let mut idle_sweeps_post_eof: u32 = 0;
             loop {
                 tokio::select! {
-                    changed = eof_rx.changed() => {
-                        eof_observed = changed.is_err() || *eof_rx.borrow_and_update();
-                        if eof_observed {
-                            pending.lock().fail_all(connection_closed_error());
-                        }
-                    }
                     _ = &mut close_rx => {
                         pending.lock().fail_all(connection_closed_error());
                         return;
                     }
                     _ = sweep.tick() => {
                         if *eof_rx.borrow() {
-                            pending.lock().fail_all(connection_closed_error());
+                            if pending
+                                .lock()
+                                .fail_all(connection_closed_error())
+                                .is_empty()
+                            {
+                                idle_sweeps_post_eof += 1;
+                            } else {
+                                idle_sweeps_post_eof = 0;
+                            }
+                            if idle_sweeps_post_eof >= PENDING_SWEEP_MAX_POST_EOF {
+                                return;
+                            }
                         }
                     }
                 }
@@ -277,6 +316,8 @@ impl WssSession {
             call_connection,
             _monitor: WssDropMonitor {
                 close_tx: Some(close_tx),
+                #[cfg(test)]
+                monitor_task,
             },
         })
     }
@@ -962,7 +1003,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn call_registered_after_eof_resolves_via_pending_sweep() {
+    async fn call_registered_after_eof_fails_fast_retryable() {
         let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
         let session = WssSession::connect(&endpoint, Some("tok-1"), true)
             .await
@@ -983,20 +1024,41 @@ mod tests {
 
         std::mem::forget(session);
         abort_pumps_and_wait(&pumps_slot).await;
-        // Headroom past the monitor's 1 s sweep tick: the call
-        // registered below must be caught by the next sweep, not hang.
-        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
 
         let ctx = noop_context("req-late");
         let call_task = tokio::spawn(async move { handler(serde_json::json!({}), ctx).await });
 
-        let response = tokio::time::timeout(std::time::Duration::from_secs(5), call_task)
+        let response = tokio::time::timeout(std::time::Duration::from_millis(500), call_task)
             .await
-            .expect("post-EOF registered call resolves via sweep (no hang)")
+            .expect("post-EOF registered call fails fast (no 1s sweep wait, no hang)")
             .expect("join");
         match response.result {
-            Err(e) => assert!(e.retryable, "drop error must be retryable, got {e:?}"),
+            Err(e) => {
+                assert_eq!(e.code, "CONNECTION_CLOSED", "fast-fail code, got {e:?}");
+                assert!(e.retryable, "drop error must be retryable, got {e:?}");
+            }
             Ok(_) => panic!("expected Err after connection drop"),
         }
     }
+
+    #[tokio::test]
+    async fn drop_monitor_ends_after_eof_plus_grace_window() {
+        let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
+        let mut session = WssSession::connect(&endpoint, Some("tok-1"), true)
+            .await
+            .expect("connect");
+
+        let config = FromCallConfig::new();
+        let bundles = import_from_call(&session.call_connection, config)
+            .await
+            .expect("import");
+        assert!(!bundles.is_empty());
+
+        abort_pumps_and_wait(&pumps_slot).await;
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), session.monitor_handle())
+            .await
+            .expect("monitor ends after EOF + bounded grace window (no permanent task)")
+            .expect("monitor task was not aborted, no panic payload");
+    }
 }