fix(ssh): close zombie local port forwarder on SSH keep-alive ping failures

lib/core/security/ssh_tunnel_manager.dart

@@ -392,18 +392,34 @@ class SshTunnelManager {
 
     // 4. Setup keep-alive ping timer
     Timer? keepAliveTimer;
+    _PooledTunnelSession? session;
     if (config.keepAliveIntervalSeconds > 0) {
+      var failedPings = 0;
       keepAliveTimer = Timer.periodic(
         Duration(seconds: config.keepAliveIntervalSeconds),
-        (_) {
-          if (!client.isClosed) {
-            client.ping().catchError((_) {});
+        (_) async {
+          if (client.isClosed) {
+            if (session != null) {
+              await _dropZombieSession(session);
+            }
+            return;
+          }
+          try {
+            await client.ping();
+            failedPings = 0;
+          } catch (_) {
+            failedPings++;
+            if (failedPings >= _maxKeepAlivePingFailures) {
+              if (session != null) {
+                await _dropZombieSession(session);
+              }
+            }
           }
         },
       );
     }
 
-    final session = _PooledTunnelSession(
+    session = _PooledTunnelSession(
       poolKey: poolKey,
       client: client,
       serverSocket: serverSocket,
@@ -418,10 +434,23 @@ class SshTunnelManager {
       localPort: localPort,
       remoteHost: remoteHost,
       remotePort: remotePort,
-      onRelease: () => _releaseSession(session),
+      onRelease: () => _releaseSession(session!),
     );
   }
 
+  /// Number of consecutive failed keep-alive pings before considering the
+  /// SSH session dead and proactively closing the local forwarder and client.
+  static const int _maxKeepAlivePingFailures = 2;
+
+  /// Proactively drops a zombie session whose SSH connection or keep-alive pings
+  /// failed, closing the local listener and SSH client and evicting it from the pool.
+  Future<void> _dropZombieSession(_PooledTunnelSession session) async {
+    if (identical(_sessions[session.poolKey], session)) {
+      _sessions.remove(session.poolKey);
+    }
+    await session.close();
+  }
+
   /// Drops one reference to [session]; the last one closes it. Handles keep a
   /// reference to their own session, so a handle from a replaced (dropped)
   /// session cannot release the session that took its place.

test/core/security/ssh_tunnel_manager_test.dart

@@ -739,5 +739,55 @@ void main() {
 
       expect(server.clients.single.pingCount, 0);
     });
+
+    test(
+        'consecutive ping failures proactively close zombie session and listener',
+        () async {
+      final first = await open(config: _config(keepAlive: 1));
+      final client = server.clients.single;
+
+      expect(manager.activeSessionCount, 1);
+      expect(await _accepts(first.localPort), isTrue);
+
+      // Trigger ping failures
+      client.failPing = true;
+
+      // Wait for 2 consecutive ping intervals (2 * 1s) plus buffer
+      await Future<void>.delayed(const Duration(milliseconds: 2500));
+
+      // Session should be proactively evicted from manager and closed
+      expect(manager.activeSessionCount, 0);
+      expect(client.isClosed, isTrue);
+      expect(
+        await _accepts(first.localPort),
+        isFalse,
+        reason: 'local listener socket must be closed to prevent zombie accept',
+      );
+
+      // Reconnecting with the same config should open a fresh tunnel session
+      final fresh = await open(config: _config(keepAlive: 1));
+      expect(manager.activeSessionCount, 1);
+      expect(server.clients, hasLength(2));
+      expect(fresh.localPort, isNot(first.localPort));
+      expect(await _accepts(fresh.localPort), isTrue);
+    });
+
+    test('a single transient ping failure followed by success keeps session open',
+        () async {
+      final handle = await open(config: _config(keepAlive: 1));
+      final client = server.clients.single;
+
+      // First interval: fail ping once
+      client.failPing = true;
+      await Future<void>.delayed(const Duration(milliseconds: 1200));
+
+      // Recover before second failure
+      client.failPing = false;
+      await Future<void>.delayed(const Duration(milliseconds: 1200));
+
+      expect(manager.activeSessionCount, 1);
+      expect(client.isClosed, isFalse);
+      expect(await _accepts(handle.localPort), isTrue);
+    });
   });
 }

test/support/fake_ssh.dart

@@ -193,6 +193,9 @@ class FakeSshClient implements SSHClient {
   @override
   Future<void> get authenticated => _authenticated.future;
 
+  /// If true, calls to [ping] will throw.
+  var failPing = false;
+
   @override
   bool get isClosed => _closed;
 
@@ -202,7 +205,10 @@ class FakeSshClient implements SSHClient {
   }
 
   @override
-  Future<void> ping() async => pingCount++;
+  Future<void> ping() async {
+    pingCount++;
+    if (failPing) throw StateError('ping failed');
+  }
 
   @override
   Future<SSHForwardChannel> forwardLocal(