diff --git a/.agents/skills/verify-diff-lifecycle/covered.lock b/.agents/skills/verify-diff-lifecycle/covered.lock index 88e0fe98a..e72c2397f 100644 --- a/.agents/skills/verify-diff-lifecycle/covered.lock +++ b/.agents/skills/verify-diff-lifecycle/covered.lock @@ -1,5 +1,5 @@ == lua/claudecode/config.lua -file lua/claudecode/config.lua b4eaec92b6d04b40 +file lua/claudecode/config.lua 787fa0432d1d77d7 == lua/claudecode/cwd.lua file lua/claudecode/cwd.lua 55d26ea362173afa == lua/claudecode/diff.lua @@ -125,7 +125,7 @@ site lua/claudecode/diff_inline.lua: resolution_callback = resolution_callback, == lua/claudecode/health.lua file lua/claudecode/health.lua 5ab73bf388f3ac97 == lua/claudecode/init.lua -file lua/claudecode/init.lua f7c7fff32489cedd +file lua/claudecode/init.lua fc28de832b4056df block user_close_all 10e024fd8ada1c16 site lua/claudecode/init.lua: local pending = diff.close_pending_diffs("user command") @ function M._create_commands() site lua/claudecode/init.lua: local rejected = diff.close_rejected_diffs("user command") @ function M._create_commands() @@ -139,7 +139,7 @@ file lua/claudecode/logger.lua 17b404dda41c3591 == lua/claudecode/selection.lua file lua/claudecode/selection.lua 4d6678af786fa69f == lua/claudecode/server/client.lua -file lua/claudecode/server/client.lua d630d84ea8ce44a7 +file lua/claudecode/server/client.lua b8165df16f194471 block create_client 24fa2a266b76e6ad == lua/claudecode/server/frame.lua file lua/claudecode/server/frame.lua 3c50766875878d56 @@ -211,7 +211,7 @@ file lua/claudecode/tools/open_file.lua 6c7154d5427af9df == lua/claudecode/tools/save_document.lua file lua/claudecode/tools/save_document.lua df68d71e47300988 == lua/claudecode/types.lua -file lua/claudecode/types.lua 8028e6f291700f28 +file lua/claudecode/types.lua dbd0dcb8b70c8107 == lua/claudecode/utils.lua file lua/claudecode/utils.lua c4a21fe74858afe2 == lua/claudecode/visual_commands.lua diff --git a/dev-config.lua b/dev-config.lua index 18b1ad4f7..2427d123d 100644 --- a/dev-config.lua +++ b/dev-config.lua @@ -76,7 +76,7 @@ return { -- Connection Management -- connection_wait_delay = 600, -- Wait time after Claude's handshake before sending queued @ mentions (ms) - -- connection_timeout = 10000, -- Max time queued @ mentions wait for Claude Code to connect (ms) + -- connection_timeout = 10000, -- Time Claude Code has to connect before queued @ mentions time out (ms) -- queue_timeout = 5000, -- Unused, still accepted: queued @ mentions wait up to connection_timeout -- Diff Integration diff --git a/fixtures/issue-70/README.md b/fixtures/issue-70/README.md index 2b8463126..77ae7d74b 100644 --- a/fixtures/issue-70/README.md +++ b/fixtures/issue-70/README.md @@ -7,10 +7,11 @@ ## The one fact behind every report `:ClaudeCodeSend` (and the file/buffer/tree senders) call `send_at_mention`. When -Claude is not connected, the mention is **queued** and a `connection_timeout` -timer (default **10s**, `lua/claudecode/init.lua`) is armed. If Claude has not -opened a WebSocket back to the plugin's server by then, the queue is cleared with -the error above (`start_connection_timeout_if_needed`). +Claude is not connected, the mention is **queued** with a deadline, +`connection_timeout + connection_wait_delay` after the send (default **10.6s**, +`lua/claudecode/init.lua`). If no Claude has opened a WebSocket back to the +plugin's server and taken the mention by then, the mention is cleared with the +error above (`expire_mentions`). So the bug is never really "send is broken" — it is always **"the Claude CLI that the plugin launched never connected back to the plugin's WebSocket server."** The diff --git a/lua/claudecode/README.md b/lua/claudecode/README.md index a2e7f9ac0..91fd226b7 100644 --- a/lua/claudecode/README.md +++ b/lua/claudecode/README.md @@ -45,7 +45,7 @@ Module map (direct files): - `M.state`: `config`, `server`, `port`, `auth_token`, `initialized`, `mention_queue`, `mention_timer`, and `connection_timer` [init.lua:M.state] - `M.is_claude_connected()`: false when the server is not running. When `get_status().clients` is non-empty, true only when a client is in state `connected` with `handshake_complete`, so a `closing` client does not count. When it is empty, true when `client_count > 0`, with no handshake check [init.lua:M.is_claude_connected] - `M.send_at_mention(file_path, start_line?, end_line?, context?) -> ok, err?`: line numbers are 0-indexed [init.lua:M.send_at_mention] -- `M.process_mention_queue(from_new_connection?)`: the server's `on_open` callback calls it with `true` once a client completed its WebSocket handshake. Without the argument it sends the oldest queued mention now [init.lua:M.process_mention_queue] +- `M.process_mention_queue(from_new_connection?)`: the server's `on_open` callback calls it with `true` once a client completed its WebSocket handshake. Without the argument it flushes the queue now. A flush sends the oldest mention only once some connected client has been connected for `connection_wait_delay` ms, and otherwise waits for that moment [init.lua:M.process_mention_queue] - `M.open_with_model(args)`, `M.get_version()`, and `M.version`, read from the repository's `VERSION` file at load [init.lua:M.version] - Internal API: `M._create_commands`, `M._broadcast_at_mention`, `M._add_paths_to_claude` (default `max_files` 100, `batch_size` 10), `M._format_path_for_at_mention`, `M._ensure_terminal_visible_if_connected`, and `M._maybe_warn_unfocusable_provider` [init.lua:M._add_paths_to_claude] @@ -111,7 +111,7 @@ Module map (direct files): - Line numbers sent to Claude are 0-indexed. `:ClaudeCodeAdd` converts the 1-indexed input [init.lua:M._create_commands] - A mention sent while disconnected goes to the queue with the same absolute path as a connected send, and the terminal opens. A connected send goes to the queue too, unless broadcast debouncing is off and the queue is empty [init.lua:M._broadcast_at_mention] -- The queue sends each mention once, oldest first: 50 ms after a send, `connection_wait_delay` ms after a client completed its handshake, and 25 ms apart. A mention leaves the queue only when a connected client took it. When the write of that frame then fails on every client, the error "Claude's connection failed while the @ mention was sent" appears, and the mention is not sent again. When Claude is not connected, the mentions wait up to `connection_timeout` ms for a handshake, then the queue is cleared with the error "Connection timeout - clearing N queued @ mentions". A handshake stops that timeout, and a flush that finds no connected client starts it again. `queue_timeout` is accepted but unused [init.lua:schedule_mention_flush] +- The queue sends each mention once, oldest first: 50 ms after a send, `connection_wait_delay` ms after a client completed its handshake, and 25 ms apart. A mention goes out only once some connected client completed its handshake `connection_wait_delay` ms ago, so another client's handshake does not postpone it [init.lua:ms_until_a_client_is_ready]. The broadcast then reaches every connected client, also one that connected later and is still in its own wait. A mention leaves the queue only when a connected client took it. When the write of that frame then fails on every client, the error "Claude's connection failed while the @ mention was sent" appears, and the mention is not sent again. Each mention has a deadline, `connection_timeout + connection_wait_delay` ms after its send, and no handshake or disconnect moves it. At its deadline, a mention still in the queue is cleared with the error "Connection timeout - clearing N queued @ mentions", or with "Send timeout - clearing N queued @ mentions that the connected Claude did not take in time" while a client that is not ready yet is connected [init.lua:expire_mentions]. While a ready client is connected, no deadline clears a mention: the queue drains to it, 25 ms apart. When that client leaves, the next flush clears the mentions whose deadline passed meanwhile [init.lua:flush_mention_queue]. A client that completes its handshake less than `connection_timeout` ms after the send, and stays connected, is ready before the deadline and gets every queued mention. `queue_timeout` is accepted but unused [init.lua:schedule_mention_flush] - `M.stop` clears the queue and stops its timers. A queue timer's callback does nothing once its timer was stopped or replaced, so nothing queued before a stop reaches Claude [init.lua:start_timer] - Selection tracking is a no-op unless it is enabled. Updates are debounced by 100 ms, except that leaving visual mode, editing or consuming the held text and moving the cursor away from a held selection report at once [selection.lua:M.flush_visual_selection] [selection.lua:reconcile_held_selection]. The held text can also change while you are in the Claude terminal (a reload, a plugin's edit), where `TextChanged` does not fire for its buffer: a listener from `nvim_buf_attach` on that buffer sends the file's cursor then [selection.lua:watch_selection_buffer]. Entering another file waits for the debounce. Claude terminal buffers are skipped [selection.lua:M.update_selection] - The automatic paths send `selection_changed` only when the selection differs from the last one sent [selection.lua:M.update_selection] [selection.lua:M.flush_visual_selection]. `send_current_selection` sends with no change check [selection.lua:M.send_current_selection] diff --git a/lua/claudecode/config.lua b/lua/claudecode/config.lua index af114120e..cf2b2d8b6 100644 --- a/lua/claudecode/config.lua +++ b/lua/claudecode/config.lua @@ -18,7 +18,9 @@ M.defaults = { focus_after_send = false, visual_demotion_delay_ms = 50, -- Milliseconds to wait before demoting a visual selection connection_wait_delay = 600, -- Milliseconds to wait after connection before sending queued @ mentions - connection_timeout = 10000, -- Maximum time queued @ mentions wait for Claude Code to connect (milliseconds) + -- Milliseconds Claude Code has to connect for queued @ mentions. Each one leaves with an error connection_timeout + + -- connection_wait_delay ms after its send, unless a ready Claude takes the queue first. + connection_timeout = 10000, queue_timeout = 5000, -- Unused, kept so that configs that set it stay valid (connection_timeout bounds the wait) diff_opts = { layout = "vertical", diff --git a/lua/claudecode/init.lua b/lua/claudecode/init.lua index 42bf0885a..11c96c9c4 100644 --- a/lua/claudecode/init.lua +++ b/lua/claudecode/init.lua @@ -80,13 +80,31 @@ function M.is_claude_connected() end end +---@return number|nil wait Milliseconds until some connected client completed its handshake connection_wait_delay ms +--- ago: 0 when one did, nil when no connected client reports when it connected +local function ms_until_a_client_is_ready() + local wait + for _, info in ipairs(require("claudecode.server.init").get_status().clients or {}) do + if info.state == "connected" and info.handshake_complete == true and info.connected_at then + local left = math.max(0, info.connected_at + M.state.config.connection_wait_delay - vim.loop.now()) + wait = wait and math.min(wait, left) or left + end + end + return wait +end + -- The @ mention queue. A send that does not go out at once waits in M.state.mention_queue, oldest first. A mention --- leaves the queue only when a connected Claude took it, when its wait timed out with an error, or when the --- integration stops. While the queue holds a mention, one of two timers runs: --- - M.state.mention_timer is the next flush: 50 ms after a send, connection_wait_delay ms after Claude's handshake, --- or 25 ms after the previous mention went out. A flush sends only the oldest mention, so the order holds. --- - M.state.connection_timer runs while the mentions wait for Claude to connect. After connection_timeout ms it clears --- them with an error. Claude's handshake stops it, and a flush that finds no connected Claude starts it. +-- leaves the queue only when a connected Claude took it, when its deadline passed while no client was ready, or when +-- the integration stops. Two timers move it on: +-- - M.state.mention_timer is the next flush, while Claude is connected: 50 ms after a send, connection_wait_delay ms +-- after Claude's handshake, or 25 ms after the previous mention went out. A flush sends only the oldest mention, so +-- the order holds, and only once some connected client is ready: it completed its handshake connection_wait_delay +-- ms ago. +-- - M.state.connection_timer clears the mentions whose deadline passed, connection_timeout + connection_wait_delay ms +-- after their send, with an error. No handshake or disconnect moves a deadline. A Claude that completes its +-- handshake less than connection_timeout ms after the send, and stays connected, is ready before the deadline. +-- While a ready client is connected, no deadline clears a mention: the queue drains to it, 25 ms apart. A flush that +-- finds no ready client clears the mentions whose deadline passed meanwhile, and starts the timer again. -- A timer's callback runs on the main loop and does nothing once its timer was stopped or replaced, so a stop cancels -- every pending flush and timeout. local debounce_delay = 50 -- ms from a send to its flush, so a burst of sends goes out as one batch @@ -141,21 +159,42 @@ end local schedule_mention_flush ----Start the timeout of the mentions that wait for Claude to connect, unless it runs already. -local function start_connection_timeout_if_needed() - if M.state.connection_timer then +---Clear the queued mentions whose deadline passed, with an error, then wait for the deadline of the oldest one left. +---The connection_timer callback: the timer may fire for a mention that went out since, and then clears nothing. +---While a ready client is connected, it clears nothing and stops: the queue drains to that client, and a flush that +---finds no ready client calls it again. +local function expire_mentions() + if ms_until_a_client_is_ready() == 0 then + -- A flush must be due, so that the drain goes on, or finds the client gone and calls this again. The handshake's + -- own flush can come later than this: Neovim runs the scheduled callbacks in order, late when its loop was busy. + schedule_mention_flush(inter_message_delay) return end - start_timer("connection_timer", M.state.config.connection_timeout, function() - local queue = mention_queue() - if #queue > 0 and not M.is_claude_connected() then - logger.error("queue", "Connection timeout - clearing " .. #queue .. " queued @ mentions") - clear_mention_queue() + local queue = mention_queue() + local now = vim.loop.now() + local expired = 0 + -- Deadlines grow with the queue order, so the mentions whose deadline passed are the oldest ones. + while expired < #queue and queue[expired + 1].deadline <= now do + expired = expired + 1 + end + if expired > 0 then + if M.is_claude_connected() then + logger.error( + "queue", + "Send timeout - clearing " .. expired .. " queued @ mentions that the connected Claude did not take in time" + ) else - -- Claude completed its handshake, and the flush that the handshake schedules has not run yet. - schedule_mention_flush(M.state.config.connection_wait_delay) + logger.error("queue", "Connection timeout - clearing " .. expired .. " queued @ mentions") end - end) + for _ = 1, expired do + table.remove(queue, 1) + end + end + if #queue == 0 then + clear_mention_queue() + else + start_timer("connection_timer", queue[1].deadline - now, expire_mentions) + end end ---A server.broadcast callback that shows an error when no write of the mention succeeded: the connection failed after @@ -180,14 +219,25 @@ local function flush_mention_queue() end -- M.stop clears the queue when it drops the server, and the stop cancels this flush too. local server = assert(M.state.server, "the @ mention queue holds mentions while the integration is stopped") + -- A mention goes out only once some connected client has been connected for connection_wait_delay ms. The broadcast + -- then reaches every connected client, also one that connected later. Another client's handshake cannot postpone it. + local wait = ms_until_a_client_is_ready() + if wait and wait > 0 then + -- A ready client that took the queue may have left: the mentions whose deadline passed meanwhile leave now. + expire_mentions() + if #mention_queue() > 0 then + start_timer("mention_timer", wait, flush_mention_queue) + end + return + end local mention = queue[1] local params = { filePath = mention.file_path, lineStart = mention.start_line, lineEnd = mention.end_line } if not server.broadcast("at_mentioned", params, report_unwritten("queue", mention.file_path)) then -- No connected client took the frame, as Claude went away since this flush was scheduled. The mention waits for - -- Claude's next handshake, or times out with an error. + -- Claude's next handshake, or for its deadline. One that passed while a ready client took the queue leaves now. logger.debug("queue", "No connected client took queued @ mention: " .. mention.file_path) - start_connection_timeout_if_needed() + expire_mentions() return end table.remove(queue, 1) @@ -195,20 +245,15 @@ local function flush_mention_queue() schedule_mention_flush(inter_message_delay) end ----Keep the queue moving. While Claude is connected, a flush is due: in delay ms, unless one is due already. Otherwise ----the connection timeout runs. An empty queue stops both timers. +---Keep the queue moving. While Claude is connected, a flush is due: in delay ms, unless one is due already. An empty +---queue stops both timers. ---@param delay number Milliseconds to the flush when Claude is connected and no flush is due yet schedule_mention_flush = function(delay) if #mention_queue() == 0 then stop_timer("mention_timer") stop_timer("connection_timer") - elseif M.is_claude_connected() then - stop_timer("connection_timer") - if not M.state.mention_timer then - start_timer("mention_timer", delay, flush_mention_queue) - end - else - start_connection_timeout_if_needed() + elseif M.is_claude_connected() and not M.state.mention_timer then + start_timer("mention_timer", delay, flush_mention_queue) end end @@ -218,17 +263,25 @@ end ---@param end_line number|nil Optional end line local function queue_mention(file_path, start_line, end_line) local queue = mention_queue() - table.insert(queue, { file_path = file_path, start_line = start_line, end_line = end_line }) + -- Claude has connection_timeout ms to complete its handshake, and connection_wait_delay ms after it to get ready. + local timeout = M.state.config.connection_timeout + M.state.config.connection_wait_delay + local deadline = vim.loop.now() + timeout + table.insert(queue, { file_path = file_path, start_line = start_line, end_line = end_line, deadline = deadline }) logger.debug("queue", "Queued @ mention: " .. file_path .. " (queue size: " .. #queue .. ")") + if not M.state.connection_timer then + start_timer("connection_timer", timeout, expire_mentions) + end schedule_mention_flush(debounce_delay) end ---Process the mention queue. Claude completed its WebSocket handshake when from_new_connection is true (the server's ----on_open callback): the queue then flushes connection_wait_delay ms later, so that Claude is ready for the mentions. ----@param from_new_connection boolean|nil true after a handshake; otherwise the oldest mention goes out now +---on_open callback): a flush is due in connection_wait_delay ms unless one is due already. Without the argument, the +---flush runs now. +---Either way, a flush sends the oldest mention only once some connected client has been connected for +---connection_wait_delay ms, and otherwise waits for that moment (see flush_mention_queue). +---@param from_new_connection boolean|nil true after a handshake; otherwise the queue flushes now function M.process_mention_queue(from_new_connection) if from_new_connection then - stop_timer("mention_timer") schedule_mention_flush(M.state.config.connection_wait_delay) else flush_mention_queue() @@ -1343,7 +1396,7 @@ function M._broadcast_at_mention(file_path, start_line, end_line) queue_mention(formatted_path, start_line, end_line) -- Always return success since we're queuing the message. The queue sends it to a connected Claude, or shows an - -- error when Claude does not connect within connection_timeout. + -- error when no ready Claude took it by its deadline, connection_timeout + connection_wait_delay ms after the send. return true, nil, sent end diff --git a/lua/claudecode/server/client.lua b/lua/claudecode/server/client.lua index 02080ceba..c17db769c 100644 --- a/lua/claudecode/server/client.lua +++ b/lua/claudecode/server/client.lua @@ -30,6 +30,7 @@ local CLOSE_REASONS = { ---@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 connected_at number|nil When the WebSocket handshake completed (vim.loop.now()) ---@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 @@ -71,20 +72,29 @@ 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. +---callback of the last piece says that the whole frame is written, or that a piece failed. A piece that libuv refuses at +---once ends the frame there. ---@param client WebSocketClient ---@param data string ---@param callback function? Optional callback: function(err) +---@return any queued, string|nil err What `tcp_handle:write` returned for the request that carries the callback 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 + return client.tcp_handle:write(data, callback) end + local queued, err 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) + queued, err = + client.tcp_handle:write(data:sub(first, first + WRITE_PIECE_BYTES - 1), last_piece and callback or nil) + if not queued and err then + -- The frame is cut short, and the callback rides on the last piece: no later piece may follow, and the caller + -- hears the error here. + return nil, err + end end + return queued, err end ---Stop the deadline of the Close frame. Safe to call when none runs. @@ -277,6 +287,9 @@ function M.process_data(client, data, on_message, on_close, on_error, auth_token end client.handshake_complete = true client.state = "connected" + -- init.lua sends queued @ mentions only once some connected client has been connected for + -- connection_wait_delay ms since this point. The broadcast then also reaches clients that connected later. + client.connected_at = vim.loop.now() logger.debug("client", "WebSocket connection established for client:", client.id) if on_open then on_open(client) @@ -379,7 +392,7 @@ end ---Send a text message to a client ---@param client WebSocketClient The client object ---@param message string The message to send ----@param callback function? Optional callback: function(err) +---@param callback function? Optional callback: function(err), run once when the write finished or failed function M.send_message(client, message, callback) if client.state ~= "connected" then if callback then @@ -391,7 +404,12 @@ 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)) - write(client, text_frame, callback) + -- libuv runs the callback only for a write it queued. A write it refuses at once returns nil and the error, and + -- never runs the callback, so the caller hears about that failure here. + local queued, err = write(client, text_frame, callback) + if not queued and err and callback then + callback(err) + end end ---Send a ping to a client @@ -509,6 +527,7 @@ function M.get_client_info(client) buffer_size = #client.buffer + (client.pending and client.pending.size or 0), last_ping = client.last_ping, last_pong = client.last_pong, + connected_at = client.connected_at, } end diff --git a/lua/claudecode/types.lua b/lua/claudecode/types.lua index 02f4118b7..5684a9fe9 100644 --- a/lua/claudecode/types.lua +++ b/lua/claudecode/types.lua @@ -59,6 +59,7 @@ ---@field file_path string The absolute file path to mention ---@field start_line number? Optional start line (0-indexed for Claude compatibility) ---@field end_line number? Optional end line (0-indexed for Claude compatibility) +---@field deadline number When the queue clears the mention with an error (vim.loop.now() time) -- Payload delivered as `data` to a `User ClaudeCodeSendComplete` autocmd. Fired -- once per file, synchronously, when a send is accepted while Claude is connected. @@ -152,7 +153,7 @@ ---@field initialized boolean ---@field mention_queue ClaudeCodeMention[] ---@field mention_timer uv.uv_timer_t? -- the next flush of the mention queue (compatible with vim.loop timer) ----@field connection_timer uv.uv_timer_t? -- the timeout of the mentions that wait for Claude to connect +---@field connection_timer uv.uv_timer_t? -- the deadline of the oldest queued mention -- This module only defines types, no runtime functionality return {} diff --git a/tests/unit/at_mention_edge_cases_spec.lua b/tests/unit/at_mention_edge_cases_spec.lua index 1684822e4..c9c328f89 100644 --- a/tests/unit/at_mention_edge_cases_spec.lua +++ b/tests/unit/at_mention_edge_cases_spec.lua @@ -204,7 +204,7 @@ describe("At Mention Edge Cases", function() end, } init_module.state.server = { broadcast = function() end } - init_module.state.config = { connection_timeout = 10000 } + init_module.state.config = { connection_timeout = 10000, connection_wait_delay = 600 } init_module.state.mention_queue = {} init_module.is_claude_connected = function() return false diff --git a/tests/unit/mention_queue_spec.lua b/tests/unit/mention_queue_spec.lua index 3a560f6e9..1b1a81384 100644 --- a/tests/unit/mention_queue_spec.lua +++ b/tests/unit/mention_queue_spec.lua @@ -25,7 +25,7 @@ describe("The @ mention queue", function() "claudecode.server.handshake", } local saved = {} - local loop, errors, claudecode, frame, terminal_opens + local loop, errors, error_times, claudecode, frame, terminal_opens -- A fake libuv loop. Like libuv, it fires every timer that is due in one phase, in the order they were started, -- and then the main loop runs every scheduled callback before time moves on. A callback that raises is recorded, @@ -65,6 +65,7 @@ describe("The @ mention queue", function() ---Let `ms` milliseconds pass. local function run(ms) local stop_at = loop.now + ms + local phases_now = 0 while true do while #loop.scheduled > 0 do protect(table.remove(loop.scheduled, 1)) @@ -78,6 +79,8 @@ describe("The @ mention queue", function() if not due then break end + phases_now = due == loop.now and phases_now + 1 or 0 + assert(phases_now < 1000, "the timers keep firing without letting time pass") loop.now = due local firing = {} for _, timer in ipairs(loop.timers) do @@ -108,6 +111,7 @@ describe("The @ mention queue", function() before_each(function() loop = { now = 0, timers = {}, scheduled = {}, lua_errors = {} } errors = {} + error_times = {} saved = { new_timer = vim.loop.new_timer, now = vim.loop.now, @@ -155,6 +159,7 @@ describe("The @ mention queue", function() parts[i] = tostring(select(i, ...)) end table.insert(errors, table.concat(parts, " ")) + table.insert(error_times, loop.now) end package.loaded["claudecode.logger"] = { setup = function() end, @@ -232,6 +237,9 @@ describe("The @ mention queue", function() assert(not claude.closed, "write after close") if data:sub(1, 12) == "HTTP/1.1 101" then claude.reply_written = callback + elseif claude.write_refused then + -- libuv refuses the write at once: it returns nil and the error, and never runs the callback. + return nil, claude.write_refused elseif claude.write_error then -- The socket fails during the write: libuv reports it only to the write callback, in a later loop turn. if callback then @@ -274,6 +282,10 @@ describe("The @ mention queue", function() handle._read_cb(nil, frame.create_frame(frame.OPCODE.CLOSE, "\3\232", true, true)) assert.equals("closing", client.state) end + -- Claude answers a ping: the server sets last_pong. + function claude.pong() + handle._read_cb(nil, frame.create_frame(frame.OPCODE.PONG, "", true, true)) + end ---@return string[] paths The filePath of each at_mentioned notification Claude received, in order. function claude.received() local paths = {} @@ -307,6 +319,28 @@ describe("The @ mention queue", function() assert.same({}, loop.lua_errors) end + ---From now until `till` ms from now, a client connects every `period` ms and completes its handshake. Each one + ---leaves (EOF) `life` ms after it connected. + ---@return table[] clients The clients that connected so far + local function churn(period, life, till) + local clients = {} + for start = 0, till - 1, period do + at(start, function() + local claude = claude_connects() + claude.handshake() + table.insert(clients, claude) + at(life, claude.drop) + end) + end + return clients + end + + local function assert_nothing_received(clients) + for _, claude in ipairs(clients) do + assert.same({}, claude.received()) + end + end + it("control: mentions queued while Claude starts reach Claude once it connects, in order, each once", function() send("/project/a.lua") send("/project/b.lua") @@ -335,7 +369,7 @@ describe("The @ mention queue", function() assert.same({}, errors) end) - it("A: Claude's handshake just before connection_timeout stops the timeout", function() + it("A: a Claude whose handshake comes 9.9 s after the send is ready at 10.5 s and gets the mention", function() send("/project/a.lua") run(9900) local claude = claude_connects() @@ -344,8 +378,29 @@ describe("The @ mention queue", function() assert.same({ "a.lua" }, names(claude.received())) assert.same({}, errors) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.connection_timer) end) + it( + "A, F4: a Claude whose handshake comes 10.3 s after the send is not ready by the deadline: you see an error", + function() + send("/project/a.lua") + run(10300) + local claude = claude_connects() + claude.handshake() + run(1500) + + assert.same({}, claude.received()) + assert.same({ 10600 }, error_times) + assert.matches( + "Send timeout %- clearing 1 queued @ mentions that the connected Claude did not take in time", + errors[1] + ) + assert.same({}, claudecode.state.mention_queue) + end + ) + it("H: Claude drops during the wait after its handshake: the next Claude gets the mention", function() send("/project/a.lua") local first = claude_connects() @@ -378,22 +433,19 @@ describe("The @ mention queue", function() assert.same({}, claudecode.state.mention_queue) end) - it("H: after a drop in the wait, the mention gets a whole connection_timeout for Claude to come back", function() + it("H: a handshake and a drop just before the deadline do not move it", function() send("/project/a.lua") run(9900) - local first = claude_connects() - first.handshake() + local claude = claude_connects() + claude.handshake() run(50) - first.drop() + claude.drop() run(2000) - assert.same({}, errors, "the timeout that ran before the handshake must not fire after it") - - local second = claude_connects() - second.handshake() - run(1500) - assert.same({ "a.lua" }, names(second.received())) - assert.same({}, errors) + assert.same({}, claude.received()) + assert.same({ 10600 }, error_times) + assert.matches("Connection timeout %- clearing 1 queued @ mentions", errors[1]) + assert.same({}, claudecode.state.mention_queue) end) it("control: a send in the tick after the handshake waits connection_wait_delay with the queued mentions", function() @@ -423,6 +475,20 @@ describe("The @ mention queue", function() assert.matches("connection failed while the @ mention /project/a%.lua was sent", errors[1]) end) + it("H: a write that libuv refuses at once shows an error too", function() + local claude = claude_connects() + claude.handshake() + run(1000) + claude.write_refused = "EBADF: bad file descriptor" + send("/project/a.lua") + run(1000) + + assert.same({}, claude.received()) + assert.same({}, claudecode.state.mention_queue) + assert.equals(1, #errors) + assert.matches("connection failed while the @ mention /project/a%.lua was sent", errors[1]) + end) + it("H: with disable_broadcast_debouncing, a send whose write fails shows an error with the absolute path", function() claudecode.state.config.disable_broadcast_debouncing = true local claude = claude_connects() @@ -440,6 +506,169 @@ describe("The @ mention queue", function() assert.matches("connection failed while the @ mention /project/src/a%.lua was sent", errors[1]) end) + it("D: handshakes of other clients do not postpone a flush that is due for a connected Claude", function() + local claude = claude_connects() + claude.handshake() + run(1000) + send("/project/a.lua") + for _ = 1, 10 do + claude_connects().handshake() + run(40) + end + + assert.same({ "a.lua" }, names(claude.received())) + assert.same({}, errors) + end) + + it( + "D: when the Claude a flush was kept for leaves, a newly connected Claude still gets connection_wait_delay", + function() + local first = claude_connects() + first.handshake() + run(1000) + send("/project/a.lua") + run(10) + local second = claude_connects() + second.handshake() + run(10) + first.drop() + run(280) + assert.same({}, second.received(), "nothing goes out before the new Claude's connection_wait_delay ends") + + run(400) + assert.same({ "a.lua" }, names(second.received())) + assert.same({}, errors) + end + ) + + it("D: a Claude whose handshake came 700 ms after its TCP connect still gets connection_wait_delay", function() + local claude = claude_connects() + run(700) + claude.handshake() + send("/project/a.lua") + run(300) + assert.same({}, claude.received(), "the wait starts at the handshake, not at the TCP connect") + + run(400) + assert.same({ "a.lua" }, names(claude.received())) + end) + + it("D: a Claude whose CLOSE frame arrives as a flush is due does not count as ready", function() + local first = claude_connects() + first.handshake() + run(1000) + local second = claude_connects() + second.handshake() + send("/project/a.lua") + -- In the timer phase of the due flush, after it: the ready Claude sends its CLOSE frame and is "closing". + at(50, first.close) + run(300) + assert.same({}, second.received(), "only a connected Claude counts as ready") + + run(400) + assert.same({ "a.lua" }, names(second.received())) + assert.same({}, first.received()) + end) + + it("D: when the Claude a flush waits for leaves during the wait, the next Claude gets its own wait", function() + local first = claude_connects() + first.handshake() + send("/project/a.lua") + run(300) + first.drop() + run(200) + local second = claude_connects() + second.handshake() + run(300) + assert.same({}, second.received(), "the flush checks again after its wait") + + run(400) + assert.same({ "a.lua" }, names(second.received())) + end) + + it("D: a Claude that answered a ping just now has still been connected since its handshake", function() + local claude = claude_connects() + claude.handshake() + run(1000) + claude.pong() + send("/project/a.lua") + run(100) + + assert.same({ "a.lua" }, names(claude.received())) + end) + + it("S: clients that overlap but each leave before they are ready do not hold a mention past its deadline", function() + -- A client connects every 200 ms and stays 300 ms, for 20 s: one is always connected, and none is connected for + -- connection_wait_delay (600 ms). The mention is sent while the first client is connected. + local clients = churn(200, 300, 20000) + run(1) + send("/project/a.lua") + run(20400) + + assert.same({ 10601 }, error_times) + assert.matches("Send timeout %- clearing 1 queued", errors[1]) + assert.same({}, claudecode.state.mention_queue) + assert_nothing_received(clients) + assert_no_lua_error() + end) + + -- A client connects every `period` ms and stays `life` ms, for 30 s: never two at once, and none for + -- connection_wait_delay (600 ms). The mention is sent while the first client is connected, after its handshake. + for _, case in ipairs({ { 1000, 300 }, { 700, 300 }, { 650, 500 }, { 900, 100 } }) do + local period, life = case[1], case[2] + it( + ("S: clients that connect every %d ms and stay %d ms do not hold a mention past its deadline"):format( + period, + life + ), + function() + local clients = churn(period, life, 30000) + run(1) + send("/project/a.lua") + run(30000) + + assert.same({ 10601 }, error_times) + assert.matches("clearing 1 queued", errors[1]) + assert.same({}, claudecode.state.mention_queue) + assert_nothing_received(clients) + assert_no_lua_error() + end + ) + end + + it("control: a mention sent 12 s after the last one went out gets its own deadline", function() + local first = claude_connects() + first.handshake() + send("/project/a.lua") + run(12000) + assert.same({ "a.lua" }, names(first.received())) + first.drop() + local second = claude_connects() + second.handshake() + send("/project/b.lua") + run(1000) + + assert.same({ "b.lua" }, names(second.received())) + assert.same({}, errors) + end) + + it("C, control: a mention sent 12 s after a stop and a start gets its own deadline", function() + local first = claude_connects() + first.handshake() + send("/project/a.lua") + run(100) + claudecode.stop() + assert(claudecode.start(false)) + run(12000) + local second = claude_connects() + second.handshake() + send("/project/b.lua") + run(1000) + + assert.same({ "b.lua" }, names(second.received())) + assert.same({}, errors) + end) + it("E: a mention sent during the wait after a connect arrives after the one queued before it", function() send("/project/first.lua") local claude = claude_connects() @@ -473,7 +702,9 @@ describe("The @ mention queue", function() claude.handshake() run(100) claudecode.stop() - run(1500) + assert.is_nil(claudecode.state.mention_timer) + assert.is_nil(claudecode.state.connection_timer) + run(11000) assert_no_lua_error() assert.same({}, claude.received()) @@ -495,57 +726,103 @@ describe("The @ mention queue", function() assert.same({}, second.received()) end) - it("C: a stop cancels a connection timeout whose callback already waits for the main loop", function() + it("C: a stop cancels a deadline whose callback already waits for the main loop", function() send("/project/old.lua") - -- In the timer phase in which the timeout fires, before its callback runs: stop, start, and send again. - at(10000, function() + -- In the timer phase in which the deadline passes, before its callback runs: stop, start, and send again. + at(10600, function() claudecode.stop() assert(claudecode.start(false)) send("/project/new.lua") end) - run(10000) - assert.same({}, errors, "the old timeout must not clear the mention sent after the stop") + run(10600) + assert.same({}, errors, "the old deadline must not clear the mention sent after the stop") - run(10100) - assert.equals(1, #errors) + run(10700) + assert.same({ 21200 }, error_times) assert.matches("clearing 1 queued", errors[1]) assert_no_lua_error() end) - it("B: :ClaudeCodeStop resets the connection timeout", function() - send("/project/old.lua") + it( + "B, C: :ClaudeCodeStop drops the deadline of the old mention: a mention sent after a restart gets its own", + function() + send("/project/old.lua") + run(5000) + claudecode.stop() + assert(claudecode.start(false)) + send("/project/new.lua") + run(10599) + assert.same({}, errors, "the old deadline at 10.6 s must not clear the mention sent after the restart") + + run(1) + assert.same({ 15600 }, error_times) + assert.matches("Connection timeout %- clearing 1 queued", errors[1]) + assert_no_lua_error() + end + ) + + it("B: mentions sent 5 s apart with no Claude each time out at their own deadline", function() + send("/project/first.lua") run(5000) - claudecode.stop() - assert(claudecode.start(false)) - send("/project/new.lua") - run(9000) - assert.same({}, errors, "a mention sent after a restart gets the whole connection_timeout") + send("/project/second.lua") + run(5000) + send("/project/third.lua") + run(599) + assert.same({}, errors) - run(1100) - assert.equals(1, #errors) - assert.matches("Connection timeout", errors[1]) + run(1) + assert.matches("Connection timeout %- clearing 1 queued", errors[1]) + assert.equals(2, #claudecode.state.mention_queue) + run(10000) + + assert.same({ 10600, 15600, 20600 }, error_times) + assert.matches("Connection timeout %- clearing 1 queued", errors[2]) + assert.matches("Connection timeout %- clearing 1 queued", errors[3]) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.connection_timer) end) - it("B, control: the connection timeout runs from the first queued mention, not from the latest", function() + it("B: of two mentions sent 5 s apart, a Claude ready between their deadlines gets the second", function() send("/project/first.lua") - run(8000) + run(5000) send("/project/second.lua") - run(1900) - assert.same({}, errors) + run(6000) + local claude = claude_connects() + claude.handshake() + run(10000) - run(200) - assert.equals(1, #errors) - assert.matches("clearing 2 queued", errors[1]) + assert.same({ "second.lua" }, names(claude.received())) + assert.same({ 10600 }, error_times) assert.same({}, claudecode.state.mention_queue) end) + it("B, control: the deadline of a mention that went out clears no later mention", function() + send("/project/first.lua") + run(5000) + send("/project/second.lua") + run(1000) + local first = claude_connects() + first.handshake() + -- The first mention goes out once Claude is ready, at 6.6 s. Claude leaves before the second one goes out. + run(610) + first.drop() + run(5390) + local second = claude_connects() + second.handshake() + run(1000) + + assert.same({ "first.lua" }, names(first.received())) + assert.same({ "second.lua" }, names(second.received())) + assert.same({}, errors) + end) + it("B: after a connection timeout, a later mention times out with an error too", function() send("/project/first.lua") - run(10100) + run(10700) assert.equals(1, #errors) send("/project/second.lua") - run(10100) + run(10700) assert.equals(2, #errors) assert.matches("Connection timeout", errors[2]) @@ -617,17 +894,157 @@ describe("The @ mention queue", function() assert.same({}, errors) end) - it("D: a connection timeout that fires as Claude completes its handshake leaves the queue to that Claude", function() + it("D: a Claude that gets ready just at the deadline gets the mention", function() send("/project/a.lua") run(9000) local claude = claude_connects() claude.upgrade() - -- The 101 write completes in the timer phase in which connection_timeout fires, before the main loop runs. + -- The handshake completes at 10 s, so Claude is ready at the deadline, 10.6 s, and the flush is due then too. at(1000, claude.handshake_written) run(2000) assert.same({ "a.lua" }, names(claude.received())) assert.same({}, errors) + assert_no_lua_error() + end) + + ---Send `count` mentions at once, as a tree add of that many files does, and return their names in order. + local function send_many(count) + local sent = {} + for index = 1, count do + send("/project/f" .. index .. ".lua") + sent[index] = "f" .. index .. ".lua" + end + return sent + end + + for _, handshake_at in ipairs({ 8000, 9800 }) do + it( + ("F1: 100 mentions sent at once all reach a Claude whose handshake comes %d ms later"):format(handshake_at), + function() + local sent = send_many(100) + run(handshake_at) + local claude = claude_connects() + claude.handshake() + -- Claude is ready connection_wait_delay later, and the queue drains to it, 25 ms apart, past the deadline. + run(5000) + + assert.same(sent, names(claude.received())) + assert.same({}, errors) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.mention_timer) + assert.is_nil(claudecode.state.connection_timer) + assert_no_lua_error() + end + ) + end + + ---The count in the error that cleared the queue. + local function cleared_count(message) + return tonumber((assert(message:match("clearing (%d+) queued"), message))) + end + + it("F1: when a ready Claude leaves in the middle of the queue, the rest leave at their deadline", function() + send_many(100) + run(8000) + local claude = claude_connects() + claude.handshake() + run(1000) + claude.drop() + local got = #claude.received() + assert.is_true(got > 0 and got < 100, "Claude got " .. got) + run(5000) + + assert.same({ 10600 }, error_times) + assert.matches("Connection timeout", errors[1]) + assert.equals(100 - got, cleared_count(errors[1])) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.connection_timer) + end) + + it("F1: when a ready Claude leaves after the deadline, the rest leave at the next flush", function() + send_many(100) + run(9800) + local claude = claude_connects() + claude.handshake() + run(1200) + claude.drop() + local got = #claude.received() + assert.is_true(got > 0 and got < 100, "Claude got " .. got) + run(5000) + + assert.equals(1, #error_times) + assert.is_true(error_times[1] > 11000 and error_times[1] <= 11025, "the error came at " .. error_times[1]) + assert.matches("Connection timeout", errors[1]) + assert.equals(100 - got, cleared_count(errors[1])) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.connection_timer) + end) + + it("F1: when a ready Claude leaves after the deadline while another waits, the rest leave at once", function() + send_many(100) + run(9800) + local first = claude_connects() + first.handshake() + run(1100) + -- A second Claude connects at 10.9 s and is ready only at 11.5 s. + local second = claude_connects() + second.handshake() + run(100) + first.drop() + local got = #first.received() + assert.is_true(got > 0 and got < 100, "the first Claude got " .. got) + run(5000) + + assert.equals(1, #error_times) + assert.is_true(error_times[1] > 11000 and error_times[1] <= 11025, "the error came at " .. error_times[1]) + assert.matches("Send timeout", errors[1]) + assert.equals(100 - got, cleared_count(errors[1])) + assert.same({}, claudecode.state.mention_queue) + assert_no_lua_error() + end) + + it("F1: a ready Claude that leaves before its handshake callback ran does not strand the queue", function() + send("/project/a.lua") + run(10000) + -- Neovim runs the scheduled callbacks in order, late when its loop was busy: the handshake's on_open callback runs + -- only after the deadline, at which Claude is ready, and after Claude left. + local held = {} + local schedule = vim.schedule + vim.schedule = function(fn) + table.insert(held, fn) + end + local claude = claude_connects() + claude.handshake() + vim.schedule = schedule + run(610) + claude.drop() + for _, fn in ipairs(held) do + fn() + end + run(1000) + + assert.same({}, claude.received()) + assert.equals(1, #errors) + assert.matches("Connection timeout %- clearing 1 queued", errors[1]) + assert.same({}, claudecode.state.mention_queue) + assert.is_nil(claudecode.state.connection_timer) + assert_no_lua_error() + end) + + it("E: a send while a flush is due does not postpone that flush", function() + local claude = claude_connects() + claude.handshake() + run(1000) + send("/project/first.lua") + run(40) + send("/project/second.lua") + run(15) + assert.same({ "first.lua" }, names(claude.received()), "the flush was due 50 ms after the first send") + + run(100) + assert.same({ "first.lua", "second.lua" }, names(claude.received())) + assert.same({}, errors) end) it("G: a bad path sent while Claude is not connected returns an error and queues nothing", function() diff --git a/tests/unit/server/connection_lifecycle_spec.lua b/tests/unit/server/connection_lifecycle_spec.lua index 67f2c0c0c..9a06b55f7 100644 --- a/tests/unit/server/connection_lifecycle_spec.lua +++ b/tests/unit/server/connection_lifecycle_spec.lua @@ -1683,6 +1683,25 @@ describe("The lifecycle of a WebSocket connection", function() end) end + it("stops at a piece that libuv refuses at once, and passes its error to the callback once", function() + local client, writes, handle = connect_and_upgrade() + local queue_write, refused = handle.write, false + function handle:write(data, callback) + if not refused then -- the first piece: libuv refuses it, and never runs its callback + refused = true + return nil, "EPIPE: broken pipe" + end + return queue_write(self, data, callback) + end + local callback = spy.new(function() end) + + tcp.send_to_client(server, client.id, string.rep("x", 3 * piece), callback) + + assert.equals(1, #writes, "a piece followed the refused one") + assert.spy(callback).was_called(1) + assert.spy(callback).was_called_with("EPIPE: broken pipe") + end) + it("queues a message of one piece or less as one write", function() local client, writes = connect_and_upgrade() diff --git a/tests/unit/server_spec.lua b/tests/unit/server_spec.lua index e61307490..7cca67691 100644 --- a/tests/unit/server_spec.lua +++ b/tests/unit/server_spec.lua @@ -286,6 +286,42 @@ describe("WebSocket Server", function() assert.same({ 2 }, reports) end) + it("counts a write that libuv refuses at once as failed, and still reports once", function() + server.start({ port_range = { min = 10000, max = 65535 } }) + local function add_client(id, write) + server.state.server.clients[id] = { + id = id, + state = "connected", + queued_bytes = 0, + tcp_handle = { + write = write, + is_closing = function() + return false + end, + close = function() end, + }, + } + end + -- libuv returns nil and the error for a write it refuses, and never runs that write's callback. + add_client("refused", function() + return nil, "EBADF: bad file descriptor" + end) + add_client("written", function(_, _, callback) + callback() + return true + end) + local reports = {} + local took = server.broadcast("m", {}, function(count) + table.insert(reports, count) + end) + + server.state.server.clients = {} + server.stop() + + assert.is_true(took) + assert.same({ 1 }, reports) + end) + describe("invalid JSON-RPC messages", function() local tcp = require("claudecode.server.tcp") local original = {}