v0.2.7-pre.007

crates/ksp-core-lib/tests/workspace_dependencies.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-core-lib/tests/workspace_dependencies.rs
-// version: 2
+// version: 3
 
 //! Workspace-level dependency policy canaries owned by the foundational KSP test surface.
 
@@ -63,7 +63,7 @@ fn transport_manifest_preserves_ksp_dependency_firewall() {
     assert!(manifest.contains("ksp-logging-lib"));
     assert!(manifest.contains("futures-util = { workspace = true, features = [\"sink\", \"std\"] }"));
     assert!(manifest.contains("reqwest = { workspace = true, features = [\"rustls\"] }"));
-    assert!(manifest.contains("tokio = { workspace = true, features = [\"macros\", \"rt\", \"sync\", \"time\"] }"));
+    assert!(manifest.contains("tokio = { workspace = true, features = [\"macros\", \"net\", \"rt\", \"sync\", \"time\"] }"));
     assert!(manifest.contains("tokio-tungstenite = { workspace = true, features = [\"connect\", \"rustls-tls-webpki-roots\"] }"));
     assert!(manifest.contains("[dev-dependencies]"));
     assert!(manifest.contains("tokio = { workspace = true, features = [\"net\", \"rt\"] }"));

crates/ksp-onchain-transport-lib/Cargo.toml

@@ -1,5 +1,5 @@
 # file: crates/ksp-onchain-transport-lib/Cargo.toml
-# version: 4
+# version: 5
 
 [package]
 name = "ksp-onchain-transport-lib"
@@ -14,7 +14,7 @@ futures-util = { workspace = true, features = ["sink", "std"] }
 reqwest = { workspace = true, features = ["rustls"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
-tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
+tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] }
 tokio-tungstenite = { workspace = true, features = ["connect", "rustls-tls-webpki-roots"] }
 
 [dev-dependencies]

crates/ksp-onchain-transport-lib/README.md

@@ -1,5 +1,5 @@
 <!-- file: crates/ksp-onchain-transport-lib/README.md -->
-<!-- version: 12 -->
+<!-- version: 13 -->
 
 # `ksp-onchain-transport-lib`
 
@@ -101,7 +101,7 @@ La première session physique WebSocket est matérialisée sans introduire de po
 - ne projette jamais l'URL dans `Debug`, snapshot, erreurs KSP ou logs ;
 - répond aux `Ping` reçus et tolère les `Pong`; le lifecycle complet `Close`/shutdown a été durci ensuite en `pre.005`.
 
-Le chemin JSON-RPC générique reste `pub(crate)`. Il sert de primitive au moteur typed de subscriptions et **ne constitue pas une API publique raw provider-extension**. Le registry subscriptions et le mapping remote/local sont matérialisés en `pre.006`; reconnect/resubscribe puis le durcissement backpressure restent dans les tranches suivantes.
+Le chemin JSON-RPC générique reste `pub(crate)`. Il sert de primitive au moteur typed de subscriptions et **ne constitue pas une API publique raw provider-extension**. Le registry subscriptions et le mapping remote/local sont matérialisés en `pre.006`; reconnect/resubscribe est actif depuis `pre.007`, tandis que le durcissement backpressure per-sub reste dans la tranche suivante.
 
 Les tests déterministes utilisent un serveur WebSocket local et prouvent le handshake, le round-trip JSON-RPC, le dispatch de réponses hors ordre, l'isolation des erreurs RPC applicatives, deux sessions physiques distinctes sur la même URL et la redaction des erreurs de connexion.
 
@@ -111,7 +111,7 @@ La session physique dispose désormais de `WsSession::close().await`. Le signal
 
 Les limites `max_message_size`, `max_frame_size`, `max_write_buffer_size` et `max_pending_requests` sont couvertes par des fixtures adversariales locales. Les requests outbound qui dépassent les bornes message/frame sont rejetées avant écriture ; les frames/messages inbound surdimensionnés sont rejetés par Tungstenite avant parse JSON. Les timeouts pending libèrent leur capacité sans faire tomber une session encore saine.
 
-Ping/Pong/Close sont traités comme control frames : le Pong automatique Tungstenite est flushé, un Close distant propre mène à `Closed`, tandis qu'une erreur I/O ou une violation de protocole mène à `Failed`. Aucun heartbeat applicatif périodique n'est ajouté.
+Ping/Pong/Close sont traités comme control frames : le Pong automatique Tungstenite est flushé et aucun heartbeat applicatif périodique n'est ajouté. Depuis `pre.007`, un Close distant inattendu, EOF, erreur I/O/TLS/WebSocket ou violation protocolaire structurelle entre dans le reconnect borné ; `Closed` reste réservé au shutdown local explicite ou à la disparition des handles.
 
 ### Registry de subscriptions `0.2.7-pre.006`
 
@@ -126,7 +126,17 @@ La création générique typed reste `pub(crate)` jusqu'aux wrappers standard de
 
 L'ACK de subscribe est traité atomiquement dans l'actor : le remote ID est lié au local ID avant que la notification suivante puisse être dispatchée. Les notifications inconnues/stale sont ignorées avec un diagnostic sûr. Un mismatch de méthode de notification ou un échec de décodage typed termine uniquement la subscription concernée ; la session physique reste `Active`.
 
-Le reconnect/resubscribe et les races associées restent hors scope jusqu'à `pre.007`. Le durcissement adversarial final de backpressure/leaks reste `pre.008`.
+### Reconnect et resubscribe `0.2.7-pre.007`
+
+Une perte de connexion physique invalide immédiatement les remote subscription IDs et incrémente `continuity_gap_count`. Le runtime utilise `WsReconnectSettings` pour appliquer un nombre fini de tentatives avec backoff exponentiel borné et sans jitter. Le shutdown surveille les phases de backoff et de handshake et interrompt la reprise sans reconnecter uniquement pour nettoyer des subscriptions.
+
+Avec `WsResubscribePolicy::ActiveSubscriptions`, les subscriptions encore désirées passent en `Resubscribing` et sont restaurées dans l'ordre croissant de leur `WsSubscriptionId`. Les paramètres de subscribe conservés par l'actor sont rejoués, puis chaque nouvel ACK remappe un remote ID sans changer l'identité locale. Le retour à `Active` ne se produit qu'après la fin de cette restauration, ce qui réinitialise alors le budget de reconnect.
+
+Avec `WsResubscribePolicy::Never`, la session physique peut se reconnecter mais les subscriptions précédentes deviennent terminales. Une cancellation locale reçue pendant reconnect gagne toujours : elle retire la subscription de la restauration ; si un ACK distant arrive après cette cancellation, l'actor envoie un unsubscribe best-effort du nouvel ID sans réactiver le handle local.
+
+Le compteur de continuity gaps est un signal d'observabilité, pas une garantie de livraison. Transport n'ajoute aucun backfill HTTP et ne promet aucune continuité lossless pendant l'intervalle de déconnexion.
+
+Le durcissement adversarial final de backpressure/leaks reste `pre.008`.
 
 ## Résilience
 

crates/ksp-onchain-transport-lib/USAGE.md

@@ -1,5 +1,5 @@
 <!-- file: crates/ksp-onchain-transport-lib/USAGE.md -->
-<!-- version: 12 -->
+<!-- version: 13 -->
 
 # Utilisation de `ksp-onchain-transport-lib`
 
@@ -98,6 +98,14 @@ Le socket brut et la primitive JSON-RPC générique ne sont pas publics. À part
 
 Le snapshot de session expose les subscriptions actuellement enregistrées via `WsSubscriptionSnapshot`, avec `remote_bound: bool` seulement. Le remote ID réel n'est jamais projeté.
 
+### Reconnect automatique borné
+
+Depuis `0.2.7-pre.007`, les settings de session contrôlent réellement le reconnect physique. Une perte de socket publie `Reconnecting { attempt }`, invalide les remote IDs et incrémente `continuity_gap_count`. Avec la policy par défaut `ActiveSubscriptions`, les handles logiques gardent leur `WsSubscriptionId` et passent temporairement en `Resubscribing`; l'actor recrée leurs subscriptions dans l'ordre local avant de republier `Active`.
+
+`WsResubscribePolicy::Never` reconnecte uniquement la session physique : les subscriptions existantes deviennent terminales et doivent être recréées explicitement par le consumer. Dans les deux modes, les requests applicatives qui étaient en vol lors de la coupure échouent et ne sont pas rejouées implicitement.
+
+`unsubscribe().await` peut être appelé pendant `Reconnecting` ou `Resubscribing`. La cancellation locale gagne et le handle ne redevient jamais `Active`. Un ACK distant tardif est nettoyé best-effort par l'actor. Aucun backfill HTTP n'est déclenché automatiquement ; le consumer doit traiter `continuity_gap_count` comme un signal de réconciliation éventuelle.
+
 ### Fermeture explicite
 
 À partir de `0.2.7-pre.005`, fermer explicitement la session est la voie normale de shutdown :

crates/ksp-onchain-transport-lib/src/ws_subscription.rs

@@ -1,10 +1,10 @@
 // file: crates/ksp-onchain-transport-lib/src/ws_subscription.rs
-// version: 1
+// version: 2
 
 /// Typed handle for one logical Solana WebSocket subscription.
 ///
 /// The handle owns the bounded typed notification receiver while the physical session actor owns the remote subscription identity and socket. The remote
-/// numeric subscription identifier is intentionally never exposed because it is transient and will be remapped by reconnect support in later tranches.
+/// numeric subscription identifier is intentionally never exposed because it is transient and is remapped by the session actor after reconnect.
 pub struct WsSubscription<T> {
     session_id: crate::WsSessionId,
     id: crate::WsSubscriptionId,
@@ -63,8 +63,9 @@ impl<T> WsSubscription<T> {
 
     /// Cancels this logical subscription and sends the matching Solana unsubscribe request when a remote binding still exists.
     ///
-    /// The returned boolean preserves the standard Solana unsubscribe result. Local cancellation is terminal for this handle even when the remote endpoint
-    /// returns an application error; reconnect race hardening is completed in `0.2.7-pre.007`.
+    /// The returned boolean preserves the standard Solana unsubscribe result when the current remote binding is reachable. Local cancellation is terminal
+    /// for this handle; during reconnect it wins before resubscribe selection, and a late remote acknowledgement is cleaned up best-effort without
+    /// reactivation.
     pub async fn unsubscribe(&mut self) -> ksp_core_lib::Result<bool> {
         match self.state() {
             crate::WsSubscriptionState::Closed => return std::result::Result::Ok(false),
@@ -171,6 +172,8 @@ pub(crate) struct WsSubscriptionRuntime {
     pub(crate) kind: crate::WsSubscriptionKind,
     /// Current logical lifecycle state.
     pub(crate) state: crate::WsSubscriptionState,
+    /// Original standard subscribe parameters retained internally for deterministic resubscribe.
+    pub(crate) params: std::vec::Vec<serde_json::Value>,
     /// Current transient remote subscription identity when bound.
     pub(crate) remote_id: std::option::Option<u64>,
     /// Lifecycle publisher observed by the public typed handle.

crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs

@@ -1,5 +1,5 @@
 // file: crates/ksp-onchain-transport-lib/unit_tests/ws_session.rs
-// version: 4
+// version: 5
 
 use futures_util::SinkExt; // rust-rules: trait-import
 use futures_util::StreamExt; // rust-rules: trait-import
@@ -44,6 +44,23 @@ fn session_settings(
     );
 }
 
+fn reconnect_session_settings(max_retries: u32, backoff: std::time::Duration, resubscribe: crate::WsResubscribePolicy) -> crate::WsSessionSettings {
+    let defaults = crate::WsSessionSettings::default();
+    return crate::WsSessionSettings::new(
+        std::time::Duration::from_millis(250),
+        std::time::Duration::from_millis(200),
+        crate::WsReconnectSettings::new(max_retries, backoff, backoff),
+        resubscribe,
+        defaults.command_queue_capacity(),
+        defaults.notification_queue_capacity(),
+        defaults.max_active_subscriptions(),
+        defaults.max_pending_requests(),
+        defaults.max_message_size_bytes(),
+        defaults.max_frame_size_bytes(),
+        defaults.max_write_buffer_size_bytes(),
+    );
+}
+
 async fn bind_local_listener() -> (tokio::net::TcpListener, std::string::String) {
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("local listener must bind");
     let address = listener.local_addr().expect("local listener must expose address");
@@ -84,6 +101,17 @@ async fn wait_for_pending_count(session: &crate::WsSession, expected: usize) {
     }
 }
 
+async fn wait_for_gap_count(session: &crate::WsSession, expected: u64) {
+    let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
+    loop {
+        if session.snapshot().continuity_gap_count() == expected && session.state() == crate::WsSessionState::Active {
+            return;
+        }
+        assert!(tokio::time::Instant::now() < deadline, "session did not recover with continuity gap count {expected}");
+        tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+    }
+}
+
 #[tokio::test(flavor = "current_thread")]
 async fn websocket_session_connects_and_round_trips_internal_json_rpc() {
     let (listener, url) = bind_local_listener().await;
@@ -314,17 +342,17 @@ async fn websocket_ping_flushes_automatic_pong_and_keeps_session_active() {
 }
 
 #[tokio::test(flavor = "current_thread")]
-async fn websocket_remote_close_transitions_to_closed_instead_of_failed() {
+async fn websocket_remote_close_consumes_bounded_reconnect_budget_before_terminal_failure() {
     let (listener, url) = bind_local_listener().await;
     let server = tokio::spawn(async move {
         let (stream, _) = listener.accept().await.expect("local server must accept client");
         let mut websocket = tokio_tungstenite::accept_async(stream).await.expect("local WebSocket handshake must succeed");
         websocket.send(tokio_tungstenite::tungstenite::Message::Close(std::option::Option::None)).await.expect("fixture Close must send");
-        let _ = tokio::time::timeout(std::time::Duration::from_millis(200), websocket.next()).await;
     });
-    let session = crate::WsSession::connect(local_endpoint(url.as_str())).await.expect("client handshake must succeed");
-    wait_for_state(&session, crate::WsSessionState::Closed).await;
-    assert!(session.close().await.is_ok());
+    let settings = reconnect_session_settings(1, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    wait_for_state(&session, crate::WsSessionState::Failed).await;
+    assert_eq!(session.snapshot().continuity_gap_count(), 1);
     server.await.expect("local server task must complete");
 }
 
@@ -381,6 +409,319 @@ async fn websocket_repeated_connect_close_cycles_are_bounded() {
         .expect("repeated close server task must join");
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_reconnect_resubscribes_in_local_id_order_and_remaps_remote_ids() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let account = read_request(&mut first).await;
+        assert_eq!(account.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("accountSubscribe"));
+        send_result(&mut first, &account, serde_json::json!(101)).await;
+        let logs = read_request(&mut first).await;
+        assert_eq!(logs.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("logsSubscribe"));
+        send_result(&mut first, &logs, serde_json::json!(202)).await;
+        drop(first);
+        let (second_stream, _) = listener.accept().await.expect("replacement client must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("replacement handshake must succeed");
+        let restored_account = read_request(&mut second).await;
+        assert_eq!(restored_account.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("accountSubscribe"));
+        assert_eq!(restored_account.get("params"), account.get("params"));
+        send_result(&mut second, &restored_account, serde_json::json!(301)).await;
+        let restored_logs = read_request(&mut second).await;
+        assert_eq!(restored_logs.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("logsSubscribe"));
+        assert_eq!(restored_logs.get("params"), logs.get("params"));
+        send_result(&mut second, &restored_logs, serde_json::json!(302)).await;
+        send_notification(&mut second, "logsNotification", 302, serde_json::json!({"generation": 2, "family": "logs"})).await;
+        send_notification(&mut second, "accountNotification", 301, serde_json::json!({"generation": 2, "family": "account"})).await;
+        let request = read_request(&mut second).await;
+        send_result(&mut second, &request, serde_json::json!(true)).await;
+    });
+    let settings = reconnect_session_settings(2, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let mut account = session
+        .subscribe_typed(crate::WsSubscriptionKind::Account, std::vec![serde_json::json!("account-canary")], |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("account subscribe must succeed");
+    let mut logs = session
+        .subscribe_typed(crate::WsSubscriptionKind::Logs, std::vec![serde_json::json!({"mentions": ["log-canary"]})], |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("logs subscribe must succeed");
+    let account_id = account.id();
+    let logs_id = logs.id();
+    let logs_value = tokio::time::timeout(std::time::Duration::from_secs(2), logs.recv())
+        .await
+        .expect("logs resubscribe notification must remain bounded")
+        .expect("logs channel must remain open")
+        .expect("logs notification must decode");
+    let account_value = tokio::time::timeout(std::time::Duration::from_secs(2), account.recv())
+        .await
+        .expect("account resubscribe notification must remain bounded")
+        .expect("account channel must remain open")
+        .expect("account notification must decode");
+    assert_eq!(logs_value, serde_json::json!({"generation": 2, "family": "logs"}));
+    assert_eq!(account_value, serde_json::json!({"generation": 2, "family": "account"}));
+    assert_eq!(account.id(), account_id);
+    assert_eq!(logs.id(), logs_id);
+    assert_eq!(account.state(), crate::WsSubscriptionState::Active);
+    assert_eq!(logs.state(), crate::WsSubscriptionState::Active);
+    assert_eq!(session.snapshot().continuity_gap_count(), 1);
+    assert!(session.snapshot().subscriptions().iter().all(|snapshot| return snapshot.remote_bound()));
+    let result = session.execute_json_rpc("afterResubscribe", std::vec::Vec::new()).await.expect("reconnected session must remain usable");
+    assert_eq!(result, serde_json::json!(true));
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_unsubscribe_during_reconnect_prevents_resubscribe() {
+    let (listener, url) = bind_local_listener().await;
+    let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let subscribe = read_request(&mut first).await;
+        send_result(&mut first, &subscribe, serde_json::json!(41)).await;
+        drop(first);
+        let _ = dropped_tx.send(());
+        let (second_stream, _) = listener.accept().await.expect("replacement client must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("replacement handshake must succeed");
+        let first_request = read_request(&mut second).await;
+        assert_eq!(first_request.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("afterReconnectCancel"));
+        send_result(&mut second, &first_request, serde_json::json!(true)).await;
+    });
+    let settings = reconnect_session_settings(2, std::time::Duration::from_millis(120), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let mut subscription = session
+        .subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("root subscribe must succeed");
+    dropped_rx.await.expect("fixture must drop initial socket");
+    wait_for_state(&session, crate::WsSessionState::Reconnecting { attempt: 1 }).await;
+    assert!(!subscription.unsubscribe().await.expect("local cancellation during reconnect must succeed"));
+    assert_eq!(subscription.state(), crate::WsSubscriptionState::Closed);
+    wait_for_gap_count(&session, 1).await;
+    assert_eq!(session.snapshot().subscription_count(), 0);
+    let result = session
+        .execute_json_rpc("afterReconnectCancel", std::vec::Vec::new())
+        .await
+        .expect("session must remain usable after reconnect cancellation");
+    assert_eq!(result, serde_json::json!(true));
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_late_resubscribe_ack_after_unsubscribe_is_cleaned_up_without_reactivation() {
+    let (listener, url) = bind_local_listener().await;
+    let (resubscribe_seen_tx, resubscribe_seen_rx) = tokio::sync::oneshot::channel();
+    let (release_ack_tx, release_ack_rx) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let subscribe = read_request(&mut first).await;
+        send_result(&mut first, &subscribe, serde_json::json!(41)).await;
+        drop(first);
+        let (second_stream, _) = listener.accept().await.expect("replacement client must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("replacement handshake must succeed");
+        let resubscribe = read_request(&mut second).await;
+        assert_eq!(resubscribe.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("rootSubscribe"));
+        let _ = resubscribe_seen_tx.send(());
+        release_ack_rx.await.expect("test must release stale acknowledgement");
+        send_result(&mut second, &resubscribe, serde_json::json!(99)).await;
+        let cleanup = read_request(&mut second).await;
+        assert_eq!(cleanup.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("rootUnsubscribe"));
+        assert_eq!(cleanup.get("params"), std::option::Option::Some(&serde_json::json!([99])));
+        send_result(&mut second, &cleanup, serde_json::json!(true)).await;
+        let request = read_request(&mut second).await;
+        assert_eq!(request.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("afterStaleAck"));
+        send_result(&mut second, &request, serde_json::json!(true)).await;
+    });
+    let settings = reconnect_session_settings(2, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let mut subscription = session
+        .subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("root subscribe must succeed");
+    resubscribe_seen_rx.await.expect("fixture must observe replacement subscribe");
+    assert!(!subscription.unsubscribe().await.expect("local cancellation must win while resubscribe acknowledgement is pending"));
+    assert_eq!(subscription.state(), crate::WsSubscriptionState::Closed);
+    release_ack_tx.send(()).expect("fixture stale acknowledgement must be released");
+    wait_for_gap_count(&session, 1).await;
+    assert_eq!(subscription.state(), crate::WsSubscriptionState::Closed);
+    assert_eq!(session.snapshot().subscription_count(), 0);
+    let result = session.execute_json_rpc("afterStaleAck", std::vec::Vec::new()).await.expect("session must remain usable after stale ack cleanup");
+    assert_eq!(result, serde_json::json!(true));
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_reconnect_budget_resets_after_full_active_recovery() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let first_request = read_request(&mut first).await;
+        send_result(&mut first, &first_request, serde_json::json!(0)).await;
+        drop(first);
+        let (second_stream, _) = listener.accept().await.expect("first replacement must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("first replacement handshake must succeed");
+        let second_request = read_request(&mut second).await;
+        send_result(&mut second, &second_request, serde_json::json!(1)).await;
+        drop(second);
+        let (third_stream, _) = listener.accept().await.expect("second replacement must connect");
+        let mut third = tokio_tungstenite::accept_async(third_stream).await.expect("second replacement handshake must succeed");
+        let third_request = read_request(&mut third).await;
+        send_result(&mut third, &third_request, serde_json::json!(2)).await;
+    });
+    let settings = reconnect_session_settings(1, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let first = session.execute_json_rpc("cycleZero", std::vec::Vec::new()).await.expect("initial request must succeed");
+    assert_eq!(first, serde_json::json!(0));
+    wait_for_gap_count(&session, 1).await;
+    let second = session.execute_json_rpc("cycleOne", std::vec::Vec::new()).await.expect("first recovered request must succeed");
+    assert_eq!(second, serde_json::json!(1));
+    wait_for_gap_count(&session, 2).await;
+    let third = session.execute_json_rpc("cycleTwo", std::vec::Vec::new()).await.expect("second recovered request must succeed");
+    assert_eq!(third, serde_json::json!(2));
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_resubscribe_never_keeps_session_but_fails_logical_subscription() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let subscribe = read_request(&mut first).await;
+        send_result(&mut first, &subscribe, serde_json::json!(71)).await;
+        drop(first);
+        let (second_stream, _) = listener.accept().await.expect("replacement client must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("replacement handshake must succeed");
+        let request = read_request(&mut second).await;
+        assert_eq!(request.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("afterNeverPolicy"));
+        send_result(&mut second, &request, serde_json::json!(true)).await;
+    });
+    let settings = reconnect_session_settings(2, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::Never);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let subscription = session
+        .subscribe_typed(crate::WsSubscriptionKind::Root, std::vec::Vec::new(), |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("root subscribe must succeed");
+    wait_for_gap_count(&session, 1).await;
+    assert_eq!(subscription.state(), crate::WsSubscriptionState::Failed);
+    assert_eq!(session.snapshot().subscription_count(), 0);
+    let result = session.execute_json_rpc("afterNeverPolicy", std::vec::Vec::new()).await.expect("physical session must reconnect without resubscribe");
+    assert_eq!(result, serde_json::json!(true));
+    server.await.expect("local server task must complete");
+}
+
+#[test]
+fn websocket_reconnect_backoff_is_exponential_and_bounded() {
+    let settings = crate::WsReconnectSettings::new(5, std::time::Duration::from_millis(10), std::time::Duration::from_millis(40));
+    assert_eq!(super::reconnect_backoff(&settings, 1), std::time::Duration::from_millis(10));
+    assert_eq!(super::reconnect_backoff(&settings, 2), std::time::Duration::from_millis(20));
+    assert_eq!(super::reconnect_backoff(&settings, 3), std::time::Duration::from_millis(40));
+    assert_eq!(super::reconnect_backoff(&settings, 4), std::time::Duration::from_millis(40));
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_shutdown_interrupts_reconnect_backoff_without_new_connection() {
+    let (listener, url) = bind_local_listener().await;
+    let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.expect("initial client must connect");
+        let websocket = tokio_tungstenite::accept_async(stream).await.expect("initial handshake must succeed");
+        drop(websocket);
+        let _ = dropped_tx.send(());
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    });
+    let defaults = crate::WsSessionSettings::default();
+    let settings = crate::WsSessionSettings::new(
+        std::time::Duration::from_millis(250),
+        std::time::Duration::from_millis(150),
+        crate::WsReconnectSettings::new(5, std::time::Duration::from_secs(2), std::time::Duration::from_secs(2)),
+        crate::WsResubscribePolicy::ActiveSubscriptions,
+        defaults.command_queue_capacity(),
+        defaults.notification_queue_capacity(),
+        defaults.max_active_subscriptions(),
+        defaults.max_pending_requests(),
+        defaults.max_message_size_bytes(),
+        defaults.max_frame_size_bytes(),
+        defaults.max_write_buffer_size_bytes(),
+    );
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    dropped_rx.await.expect("fixture must drop physical connection");
+    wait_for_state(&session, crate::WsSessionState::Reconnecting { attempt: 1 }).await;
+    tokio::time::timeout(std::time::Duration::from_millis(300), session.close())
+        .await
+        .expect("shutdown must interrupt reconnect backoff")
+        .expect("shutdown during reconnect must finish cleanly");
+    assert_eq!(session.state(), crate::WsSessionState::Closed);
+    server.await.expect("local server task must complete");
+}
+
+#[tokio::test(flavor = "current_thread")]
+async fn websocket_resubscribe_application_error_fails_only_one_subscription() {
+    let (listener, url) = bind_local_listener().await;
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.expect("initial client must connect");
+        let mut first = tokio_tungstenite::accept_async(first_stream).await.expect("initial handshake must succeed");
+        let account = read_request(&mut first).await;
+        send_result(&mut first, &account, serde_json::json!(11)).await;
+        let logs = read_request(&mut first).await;
+        send_result(&mut first, &logs, serde_json::json!(12)).await;
+        drop(first);
+        let (second_stream, _) = listener.accept().await.expect("replacement client must connect");
+        let mut second = tokio_tungstenite::accept_async(second_stream).await.expect("replacement handshake must succeed");
+        let failed = read_request(&mut second).await;
+        let failed_id = failed.get("id").and_then(serde_json::Value::as_u64).expect("resubscribe id must be numeric");
+        let error = serde_json::json!({"jsonrpc":"2.0","id":failed_id,"error":{"code":-32602,"message":"fixture resubscribe rejection"}});
+        second.send(tokio_tungstenite::tungstenite::Message::Text(error.to_string().into())).await.expect("resubscribe application error must send");
+        let restored = read_request(&mut second).await;
+        assert_eq!(restored.get("method").and_then(serde_json::Value::as_str), std::option::Option::Some("logsSubscribe"));
+        send_result(&mut second, &restored, serde_json::json!(88)).await;
+        send_notification(&mut second, "logsNotification", 88, serde_json::json!({"restored": true})).await;
+        let request = read_request(&mut second).await;
+        send_result(&mut second, &request, serde_json::json!(true)).await;
+    });
+    let settings = reconnect_session_settings(2, std::time::Duration::from_millis(20), crate::WsResubscribePolicy::ActiveSubscriptions);
+    let session = crate::WsSession::connect(local_endpoint_with_session(url.as_str(), settings)).await.expect("client handshake must succeed");
+    let account = session
+        .subscribe_typed(crate::WsSubscriptionKind::Account, std::vec![serde_json::json!("account")], |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("account subscribe must succeed");
+    let mut logs = session
+        .subscribe_typed(crate::WsSubscriptionKind::Logs, std::vec![serde_json::json!("all")], |value| {
+            return std::result::Result::Ok(value);
+        })
+        .await
+        .expect("logs subscribe must succeed");
+    let notification = tokio::time::timeout(std::time::Duration::from_secs(2), logs.recv())
+        .await
+        .expect("restored logs notification must remain bounded")
+        .expect("logs channel must remain open")
+        .expect("logs notification must decode");
+    assert_eq!(notification, serde_json::json!({"restored": true}));
+    assert_eq!(account.state(), crate::WsSubscriptionState::Failed);
+    assert_eq!(logs.state(), crate::WsSubscriptionState::Active);
+    assert_eq!(session.state(), crate::WsSessionState::Active);
+    assert_eq!(session.snapshot().continuity_gap_count(), 1);
+    let result = session.execute_json_rpc("afterPartialResubscribe", std::vec::Vec::new()).await.expect("session must remain usable");
+    assert_eq!(result, serde_json::json!(true));
+    server.await.expect("local server task must complete");
+}
+
 async fn send_notification(
     websocket: &mut tokio_tungstenite::WebSocketStream<tokio::net::TcpStream>,
     method: &str,