fix(server): drop a client that answers pings but never reads

.agents/skills/verify-diff-lifecycle/covered.lock

@@ -139,8 +139,8 @@ file lua/claudecode/logger.lua 17b404dda41c3591
 == lua/claudecode/selection.lua
 file lua/claudecode/selection.lua 05663913bc425259
 == lua/claudecode/server/client.lua
-file lua/claudecode/server/client.lua 5789c6ff18045f9f
-block create_client          647b7523dea26af6
+file lua/claudecode/server/client.lua 4304e6496b5c4f32
+block create_client          24fa2a266b76e6ad
 == lua/claudecode/server/frame.lua
 file lua/claudecode/server/frame.lua 3c50766875878d56
 == lua/claudecode/server/handshake.lua
@@ -161,7 +161,7 @@ site lua/claudecode/server/init.lua: if not _G.claude_deferred_responses then @
 == lua/claudecode/server/mock.lua
 file lua/claudecode/server/mock.lua def1248d2793d559
 == lua/claudecode/server/tcp.lua
-file lua/claudecode/server/tcp.lua f1f9833b4322b1ae
+file lua/claudecode/server/tcp.lua 63cbb8922056279a
 == lua/claudecode/server/utils.lua
 file lua/claudecode/server/utils.lua 2915ae87b48f04ca
 == lua/claudecode/terminal.lua

lua/claudecode/server/client.lua

@@ -27,15 +27,22 @@ local CLOSE_REASONS = {
 ---@field pending table|nil The reads of an incomplete frame, with their total length in `size` and the frame's in `needed`
 ---@field fragments table|nil Joined pieces of the text message whose last fragment has not arrived, and their total `size`
 ---@field handshake_complete boolean Whether WebSocket handshake is complete
----@field close_pending boolean|nil True while a Close frame waits to be written: its write callback closes the socket
+---@field close_pending boolean|nil True while a Close frame waits to be written: its write callback closes the socket, or its deadline does
+---@field close_timer table|nil The deadline of the Close frame: a repeating timer that runs while `close_pending` is set
+---@field keepalive_timeout number|nil The window in milliseconds for a pong and for write progress, from the server's ping interval
 ---@field last_ping number Timestamp of last ping sent
 ---@field last_pong number Timestamp of last pong received
+---@field queued_bytes integer The bytes passed to the socket's write queue since the connection began
+---@field drained_bytes integer The bytes libuv had written to the socket when `sample_write_progress` last saw progress
+---@field last_drain number Timestamp of the last `sample_write_progress` check that found the write queue empty or smaller
+---@field write_stalled boolean|nil True after a ping tick found no write progress. The next such tick asks for a look after the next poll, and only that look drops the client
 ---@field pong_overdue boolean|nil True after a ping timer tick found no pong for two intervals. The next such tick drops the client
 
 ---Create a new WebSocket client
 ---@param tcp_handle table The vim.loop TCP handle
+---@param keepalive_timeout number|nil The window in milliseconds that the server's ping timer allows for a pong and for write progress. Without it the client has no deadline for its Close frame
 ---@return WebSocketClient client The client object
-function M.create_client(tcp_handle)
+function M.create_client(tcp_handle, keepalive_timeout)
   _G.claudecode_clients_created = (_G.claudecode_clients_created or 0) + 1
   local client_id = "client_" .. _G.claudecode_clients_created
 
@@ -47,11 +54,88 @@ function M.create_client(tcp_handle)
     handshake_complete = false,
     last_ping = 0,
     last_pong = vim.loop.now(),
+    queued_bytes = 0,
+    drained_bytes = 0,
+    last_drain = vim.loop.now(),
+    keepalive_timeout = keepalive_timeout,
   }
 
   return client
 end
 
+-- The size of one write request for a large frame. libuv lowers the size of a socket's write queue by the bytes it wrote
+-- on Linux and macOS, but on Windows only by a whole request when it is done. Queued in pieces, a large frame shows its
+-- progress there too, and `is_client_reading` does not take a Claude that reads it for a stalled one.
+local WRITE_PIECE_BYTES = 1024 * 1024
+
+---Write to the socket of a client. Every write goes through here, so `queued_bytes` counts them all: the bytes still in
+---libuv's write queue are `queued_bytes - drained_bytes`, and `is_client_reading` tells a stalled queue from a slow one.
+---A frame of more than `WRITE_PIECE_BYTES` is queued in pieces. libuv finishes the requests of a socket in order, so the
+---callback of the last piece says that the whole frame is written, or that a piece failed.
+---@param client WebSocketClient
+---@param data string
+---@param callback function? Optional callback: function(err)
+local function write(client, data, callback)
+  client.queued_bytes = client.queued_bytes + #data
+  if #data <= WRITE_PIECE_BYTES then
+    client.tcp_handle:write(data, callback)
+    return
+  end
+  for first = 1, #data, WRITE_PIECE_BYTES do
+    local last_piece = first + WRITE_PIECE_BYTES > #data
+    client.tcp_handle:write(data:sub(first, first + WRITE_PIECE_BYTES - 1), last_piece and callback or nil)
+  end
+end
+
+---Stop the deadline of the Close frame. Safe to call when none runs.
+---@param client WebSocketClient
+local function stop_close_timer(client)
+  local timer = client.close_timer
+  if timer then
+    client.close_timer = nil
+    timer:stop()
+    timer:close()
+  end
+end
+
+---Start the deadline of a Close frame that was just queued. The ping timer sees only the clients in the server's list,
+---and a client that broke the protocol, or that the server stopped, has left it with its frame still queued. A socket
+---that wrote no byte for a keepalive window is closed, and the queue with it, as for the ping timer. The window starts
+---with the frame, and a Claude that reads keeps the socket open. One window without progress is enough here: the client
+---is leaving already, so a close that comes too early after a blocked loop costs only its Close frame.
+---@param client WebSocketClient
+local function start_close_timer(client)
+  local timeout = client.keepalive_timeout
+  -- Not when the write callback already ran: the frame is written and the socket is closed.
+  if not timeout or not client.close_pending then
+    return
+  end
+  local timer = vim.loop.new_timer()
+  if not timer then
+    return
+  end
+  stop_close_timer(client)
+  client.close_timer = timer
+  client.drained_bytes = client.queued_bytes - client.tcp_handle:get_write_queue_size()
+  client.last_drain = vim.loop.now()
+  timer:start(timeout, timeout, function()
+    -- The write callback, a drop by the ping timer, or a close of the handle ended the wait: stop, and touch nothing.
+    -- Whichever of this callback and the write callback runs first closes the socket, and the other finds it closed.
+    if client.close_timer ~= timer then
+      return
+    end
+    local handle = client.tcp_handle
+    if not client.close_pending or handle:is_closing() then
+      stop_close_timer(client)
+    elseif not M.sample_write_progress(client) then
+      stop_close_timer(client)
+      client.close_pending = nil
+      client.state = "closed"
+      handle:close()
+    end
+  end)
+end
+
 ---Adds a data frame to the message the client is sending. A text message arrives as one TEXT frame with FIN set, or
 ---as fragments: a TEXT frame with FIN clear, then CONTINUATION frames up to one with FIN set. A fragment may end
 ---inside a code point, so only the whole text is checked for UTF-8.
@@ -174,7 +258,7 @@ function M.process_data(client, data, on_message, on_close, on_error, auth_token
         client.state = "closing"
       end
 
-      client.tcp_handle:write(response_from_handshake, function(err)
+      write(client, response_from_handshake, function(err)
         if err or not success then
           -- on_error removes the client from the server and closes its socket. A server that stopped while the reply
           -- was written has already dropped the client, so the socket is closed here too.
@@ -268,7 +352,7 @@ function M.process_data(client, data, on_message, on_close, on_error, auth_token
 
       if client.state == "connected" then
         local close_frame = frame.create_close_frame(code, reason)
-        client.tcp_handle:write(close_frame)
+        write(client, close_frame)
         client.state = "closing"
       end
 
@@ -283,7 +367,7 @@ function M.process_data(client, data, on_message, on_close, on_error, auth_token
       break
     elseif opcode == frame.OPCODE.PING then
       local pong_frame = frame.create_pong_frame(parsed_frame.payload)
-      client.tcp_handle:write(pong_frame)
+      write(client, pong_frame)
     elseif opcode == frame.OPCODE.PONG then
       client.last_pong = vim.loop.now()
     end
@@ -310,7 +394,7 @@ function M.send_message(client, message, callback)
   -- Every outbound message passes here. A buffer name or path in it can hold any bytes, and the client drops the
   -- connection on a text frame that is not valid UTF-8.
   local text_frame = frame.create_text_frame(utils.valid_utf8(message))
-  client.tcp_handle:write(text_frame, callback)
+  write(client, text_frame, callback)
 end
 
 ---Send a ping to a client
@@ -322,7 +406,7 @@ function M.send_ping(client, data)
   end
 
   local ping_frame = frame.create_ping_frame(data or "")
-  client.tcp_handle:write(ping_frame)
+  write(client, ping_frame)
   client.last_ping = vim.loop.now()
 end
 
@@ -353,16 +437,19 @@ function M.close_client(client, code, reason)
   -- An upgrading client gets the Close frame too: libuv writes it after the 101 reply that is still queued.
   if client.handshake_complete or client.state == "upgrading" then
     local close_frame = frame.create_close_frame(code, reason)
-    -- The write callback closes the socket, so nothing may close it first: that would cancel the frame. libuv always
-    -- invokes the callback (on success or error), so the socket is never left open.
+    -- The write callback closes the socket, so nothing may close it first: that would cancel the frame. libuv invokes
+    -- the callback on success, on error and when the socket closes, so the socket is never left open. A Claude that
+    -- does not read can keep the frame in the queue for good: the deadline closes the socket then.
     client.close_pending = true
-    client.tcp_handle:write(close_frame, function()
+    write(client, close_frame, function()
       client.close_pending = nil
+      stop_close_timer(client)
       client.state = "closed"
       if not client.tcp_handle:is_closing() then
         client.tcp_handle:close()
       end
     end)
+    start_close_timer(client)
     -- Mark as "closing" while the Close frame write/teardown is in flight. The
     -- write callback transitions to "closed". Do not clobber the "closed" state
     -- set synchronously on the not-handshake-complete branch below.
@@ -373,6 +460,13 @@ function M.close_client(client, code, reason)
   end
 end
 
+---Stop waiting for the write of a Close frame, because the caller closes the socket at once.
+---@param client WebSocketClient The client object
+function M.abandon_close_frame(client)
+  client.close_pending = nil
+  stop_close_timer(client)
+end
+
 ---Check if a client connection is alive
 ---@param client WebSocketClient The client object
 ---@param timeout number Timeout in milliseconds (default: 30000)
@@ -388,6 +482,40 @@ function M.is_client_alive(client, timeout)
   return (now - client.last_pong) < timeout
 end
 
+---Check whether libuv wrote a byte of the write queue of a client since the last check that saw progress, or the queue
+---is empty. A client that reads slowly makes progress, and so does one that is sent a single large message, however big.
+---@param client WebSocketClient The client object
+---@return boolean progress False when the queue is not empty and nothing of it was written since the last progress
+function M.sample_write_progress(client)
+  local queue_size = client.tcp_handle:get_write_queue_size()
+  local drained = client.queued_bytes - queue_size
+  if queue_size == 0 or drained > client.drained_bytes then
+    client.drained_bytes, client.last_drain = drained, vim.loop.now()
+    return true
+  end
+  return false
+end
+
+---Check that a client still reads what the server writes to it. The keepalive cannot tell: a Claude whose reader is
+---stuck can still answer each ping, while replies and pings pile up in libuv's write queue, in Neovim's memory.
+---Call it once per ping tick. A tick can run before libuv reported the bytes that Claude read during a main-loop block,
+---so a tick never drops a client. A tick that finds progress, or an empty queue, clears the count. The first that finds
+---none sets it, and the second returns false. `tcp.lua` then looks once more right after the next poll, and only that
+---look drops the client.
+---@param client WebSocketClient The client object
+---@return boolean reading False when this tick, and the one before it, found no write progress
+function M.is_client_reading(client)
+  if M.sample_write_progress(client) then
+    client.write_stalled = nil
+    return true
+  end
+  if client.write_stalled then
+    return false
+  end
+  client.write_stalled = true
+  return true
+end
+
 ---Get client info for debugging
 ---@param client WebSocketClient The client object
 ---@return table info Client information