fix: address PR#60 review — refCount, shutdown path, dead code, SHA-256 keys

Must-fix #1: Remove dead ensureHealthy() + reconnectCount
  - ensureHealthy() was defined but never called (acquire() used direct
    socket.readyState check instead). Removed entirely along with the
    reconnectCount field from PooledConnection.

Must-fix #2: Replace isActive boolean with refCount for concurrent safety
  - When maxConcurrentTasks > 1, multiple dispatches share the same WS
    connection. The old isActive boolean meant release() from dispatch A
    would mark the connection idle and start eviction timer, killing the
    connection for dispatch B still in progress.
  - Now: acquire() increments refCount, release() decrements. Heartbeat
    and eviction only start when refCount drops to 0.
  - Added concurrent acquire test that verifies: 2 acquires → 1 release
    → connection stays alive through eviction timeout → final release
    → connection properly evicted.

Must-fix #3: Add OpenClawAgentExecutor.close() + wire into index.ts stop()
  - The WS pool's destroy() was never called during plugin shutdown,
    leaking connections and timers on hot-reload/test scenarios.
  - Added close() method to OpenClawAgentExecutor, called from the
    plugin's stop() lifecycle hook.

Nice-to-have #4: Replace conn["socket"]?.readyState with conn.isOpen getter
  - Added public isOpen getter to GatewayRpcConnection for type-safe
    socket state checking. Pool code now uses conn.isOpen instead of
    bypassing TypeScript's private access control.

Nice-to-have #5: SHA-256 buildKey instead of plaintext in Map keys
  - gatewayPassword was stored as plaintext in Map keys. Now uses
    crypto.createHash('sha256') to produce a one-way hash, reducing
    credential surface in heap dumps.

Author: zycaskevin

index.ts

@@ -330,8 +330,9 @@ const plugin = {
     const pushStore = new PushNotificationStore();
     const client = new A2AClient();
     const taskStore = new FileTaskStore(config.storage.tasksDir);
+    const agentExecutor = new OpenClawAgentExecutor(api, config);
     const executor = new QueueingAgentExecutor(
-      new OpenClawAgentExecutor(api, config),
+      agentExecutor,
       telemetry,
       config.limits,
       config.routing.defaultAgentId,
@@ -981,6 +982,7 @@ const plugin = {
         healthManager?.stop();
         auditLogger.close();
         client.destroy();
+        agentExecutor.close();
 
         // Stop task cleanup timer
         if (cleanupTimer) {

src/executor.ts

@@ -760,6 +760,11 @@ export class GatewayRpcConnection {
     this.rejectAllPending(error);
   }
 
+  /** Whether the underlying WebSocket socket is currently open (readyState === OPEN). */
+  get isOpen(): boolean {
+    return this.socket?.readyState === 1;
+  }
+
   private awaitConnectChallenge(): Promise<void> {
     if (this.connectChallengeRejecter) {
       this.connectChallengeRejecter(new Error("gateway connect challenge wait superseded"));
@@ -990,10 +995,11 @@ interface PooledConnection {
   conn: GatewayRpcConnection;
   key: string;
   lastUsedAt: number;
-  isActive: boolean;
+  /** Reference count: how many concurrent acquire() calls are using this connection.
+   *  When refCount > 0 the connection is active; when refCount === 0 it is idle. */
+  refCount: number;
   heartbeatTimer: ReturnType<typeof setTimeout> | null;
   evictionTimer: ReturnType<typeof setTimeout> | null;
-  reconnectCount: number;
 }
 
 /**
@@ -1048,11 +1054,12 @@ export class GatewayRpcConnectionPool {
     const existing = this.pool.get(key);
 
     // Hot path: reuse healthy connection
-    if (existing && existing.conn && existing.conn["socket"]?.readyState === 1) {
-      existing.isActive = true;
+    if (existing && existing.conn && existing.conn.isOpen) {
+      existing.refCount++;
       existing.lastUsedAt = Date.now();
       this.stopHeartbeat(existing);
-      this.resetEviction(existing);
+      // Eviction timer is already stopped while refCount > 0 (set in release),
+      // but we reset it here for housekeeping so the clock starts fresh on next release.
       this.totalReuses++;
       return existing.conn;
     }
@@ -1067,19 +1074,18 @@ export class GatewayRpcConnectionPool {
       conn,
       key,
       lastUsedAt: Date.now(),
-      isActive: true,
+      refCount: 1,
       heartbeatTimer: null,
       evictionTimer: null,
-      reconnectCount: 0,
     };
     this.pool.set(key, entry);
-    this.resetEviction(entry);
     return conn;
   }
 
   /**
-   * Return a connection to the pool. After release, the connection may be
-   * reused by the next acquire() call for the same key.
+   * Return a connection to the pool. Decrements the reference count; only
+   * when all acquire() calls have been released (refCount === 0) does the
+   * connection transition to idle state with heartbeat + eviction timer.
    */
   release(config: GatewayRuntimeConfig): void {
     if (this.destroyed) return;
@@ -1088,17 +1094,21 @@ export class GatewayRpcConnectionPool {
     const entry = this.pool.get(key);
     if (!entry) return;
 
-    entry.isActive = false;
+    entry.refCount = Math.max(0, entry.refCount - 1);
     entry.lastUsedAt = Date.now();
-    this.startHeartbeat(entry);
-    this.resetEviction(entry);
+
+    if (entry.refCount === 0) {
+      // Connection is now idle — start keep-alive and eviction timer
+      this.startHeartbeat(entry);
+      this.resetEviction(entry);
+    }
   }
 
   getStats(): WsPoolStats {
     let idle = 0;
     let active = 0;
     for (const entry of this.pool.values()) {
-      if (entry.isActive) active++;
+      if (entry.refCount > 0) active++;
       else idle++;
     }
     return {
@@ -1131,7 +1141,9 @@ export class GatewayRpcConnectionPool {
   // -- Internal helpers -------------------------------------------------------
 
   private buildKey(config: GatewayRuntimeConfig): string {
-    return `${config.wsUrl}::${config.gatewayToken}::${config.gatewayPassword}`;
+    const h = crypto.createHash("sha256");
+    h.update(config.wsUrl + "\0" + config.gatewayToken + "\0" + config.gatewayPassword);
+    return h.digest("hex");
   }
 
   private async createAndConnect(config: GatewayRuntimeConfig, attempt: number): Promise<GatewayRpcConnection> {
@@ -1151,40 +1163,6 @@ export class GatewayRpcConnectionPool {
     }
   }
 
-  /**
-   * Check connection health. If the socket is dead, attempt a transparent reconnect.
-   * Returns true if the connection is healthy (or was successfully reconnected).
-   */
-  private async ensureHealthy(entry: PooledConnection, config: GatewayRuntimeConfig): Promise<boolean> {
-    if (entry.conn["socket"]?.readyState === 1) {
-      return true;
-    }
-
-    // Connection is dead — reconnect
-    if (entry.reconnectCount >= this.maxReconnectAttempts) {
-      return false;
-    }
-
-    try {
-      entry.conn.close();
-    } catch {
-      // ignore close errors on dead socket
-    }
-
-    entry.reconnectCount++;
-    this.totalReconnects++;
-
-    try {
-      const conn = await this.createAndConnect(config, 0);
-      entry.conn = conn;
-      entry.lastUsedAt = Date.now();
-      entry.reconnectCount = 0; // reset on successful reconnect
-      return true;
-    } catch {
-      return false;
-    }
-  }
-
   private evictConnection(entry: PooledConnection): void {
     this.stopHeartbeat(entry);
     this.stopEviction(entry);
@@ -1202,7 +1180,7 @@ export class GatewayRpcConnectionPool {
       // Send a lightweight ping to keep the connection alive.
       // We use the `request` method with a minimal method — if the socket
       // is dead, the pending request will timeout and we'll learn about it.
-      if (entry.conn["socket"]?.readyState !== 1) {
+      if (!entry.conn.isOpen) {
         // Socket is not open — stop heartbeat, let eviction or next acquire handle it
         this.stopHeartbeat(entry);
         return;
@@ -1231,7 +1209,7 @@ export class GatewayRpcConnectionPool {
   private resetEviction(entry: PooledConnection): void {
     this.stopEviction(entry);
     entry.evictionTimer = setTimeout(() => {
-      if (!entry.isActive) {
+      if (entry.refCount === 0) {
         this.evictConnection(entry);
       }
     }, this.idleTimeoutMs);
@@ -1275,6 +1253,11 @@ export class OpenClawAgentExecutor implements AgentExecutor {
     this.wsPool = new GatewayRpcConnectionPool();
   }
 
+  /** Clean up the WS connection pool. Call this during plugin shutdown. */
+  close(): void {
+    this.wsPool.destroy();
+  }
+
   async execute(requestContext: RequestContext, eventBus: ExecutionEventBus): Promise<void> {
     const agentId = pickAgentId(requestContext, this.defaultAgentId);
     const taskId = requestContext.taskId;

tests/ws-pool.test.ts

@@ -460,13 +460,13 @@ describe("GatewayRpcConnectionPool", () => {
     const pool = new GatewayRpcConnectionPool({ idleTimeoutMs: 5_000 });
     const config = makeConfig();
 
-    // Acquire — active
+    // Acquire — active (refCount = 1)
     await pool.acquire(config);
     let stats = pool.getStats();
     assert.equal(stats.activeConnections, 1);
     assert.equal(stats.idleConnections, 0);
 
-    // Release — idle
+    // Release — idle (refCount = 0)
     pool.release(config);
     stats = pool.getStats();
     assert.equal(stats.activeConnections, 0);
@@ -475,6 +475,49 @@ describe("GatewayRpcConnectionPool", () => {
     pool.destroy();
   });
 
+  it("concurrent acquire: connection is not evicted while still in use", async () => {
+    const pool = new GatewayRpcConnectionPool({
+      idleTimeoutMs: 200,  // short timeout for test
+      heartbeatIntervalMs: 999_999,
+    });
+    const config = makeConfig();
+
+    // Two concurrent acquires — both get the same connection
+    const conn1 = await pool.acquire(config);
+    const conn2 = await pool.acquire(config);
+
+    // Same underlying connection object
+    assert.strictEqual(conn1, conn2, "Concurrent acquires should return the same connection");
+
+    let stats = pool.getStats();
+    assert.equal(stats.activeConnections, 1, "One active connection with refCount=2");
+    assert.equal(stats.idleConnections, 0);
+
+    // Release one — connection should still be active (refCount = 1)
+    pool.release(config);
+    stats = pool.getStats();
+    assert.equal(stats.activeConnections, 1, "Still active after partial release");
+    assert.equal(stats.idleConnections, 0);
+
+    // Wait past the idle timeout — should NOT be evicted because refCount > 0
+    await new Promise((r) => setTimeout(r, 350));
+    stats = pool.getStats();
+    assert.equal(stats.connections, 1, "Connection not evicted while still in use");
+
+    // Release the remaining one — now idle (refCount = 0)
+    pool.release(config);
+    stats = pool.getStats();
+    assert.equal(stats.activeConnections, 0);
+    assert.equal(stats.idleConnections, 1, "Now idle after final release");
+
+    // Wait for eviction
+    await new Promise((r) => setTimeout(r, 350));
+    stats = pool.getStats();
+    assert.equal(stats.connections, 0, "Connection evicted after becoming idle");
+
+    pool.destroy();
+  });
+
   it("reuses connection for many sequential dispatches", async () => {
     const pool = new GatewayRpcConnectionPool({ idleTimeoutMs: 5_000 });
     const config = makeConfig();