rework lua tool scheduler

1 files changed, 94 insertions, 10 deletions

diff --git a/agent/tools/read.lua b/agent/tools/read.lua
index d345f99..f232890 100644
--- a/agent/tools/read.lua
+++ b/agent/tools/read.lua
@@ -19,18 +19,98 @@ local MAX_BYTES = 50 * 1024   -- 50 KB
 local MAX_LINES = 2000
 local READ_CHUNK_SIZE = MAX_BYTES
 
+-- ---------------------------------------------------------------------------
+-- Coroutine-synchronous libuv wrappers.
+--
+-- panto runs every tool handler inside its own Lua coroutine and drives
+-- a single `uv.run()` to completion. The contract for an async handler
+-- is: the ONLY way it may block is to yield on a pending libuv request
+-- whose callback resumes it exactly once. These helpers encapsulate
+-- that pattern — each fires one async libuv op, yields, and returns the
+-- callback's results once resumed. They look synchronous to the caller
+-- but cooperate with sibling tool calls sharing the event loop.
+--
+-- INVARIANTS each wrapper upholds (and every async tool must too):
+--   * exactly one libuv op armed per yield;
+--   * the callback resumes the coroutine exactly once;
+--   * `coroutine.resume` errors are re-raised so a crash in the resumed
+--     handler isn't silently swallowed inside the uv callback.
+-- ---------------------------------------------------------------------------
+
+-- Run a single callback-style libuv fs operation and block the
+-- coroutine until its callback fires. `arm` receives a `resolve`
+-- function and must start exactly one libuv op that calls it.
+local function await(arm)
+    local co = assert(coroutine.running(),
+        "read: await must run inside a tool handler coroutine")
+    local res, fired = nil, false
+    arm(function(...)
+        assert(not fired, "read: libuv callback fired twice")
+        fired = true
+        res = table.pack(...)
+        local ok, err = coroutine.resume(co)
+        if not ok then error(err, 0) end
+    end)
+    coroutine.yield()
+    return table.unpack(res, 1, res.n)
+end
+
+local function fs_stat(path)
+    return await(function(resolve) uv.fs_stat(path, resolve) end)
+end
+
+local function fs_open(path, flags, mode)
+    return await(function(resolve) uv.fs_open(path, flags, mode, resolve) end)
+end
+
+local function fs_read(fd, size, offset)
+    return await(function(resolve) uv.fs_read(fd, size, offset, resolve) end)
+end
+
+local function fs_close(fd)
+    return await(function(resolve) uv.fs_close(fd, resolve) end)
+end
+
+-- `uv.fs_opendir` takes the entries-per-readdir batch hint as its
+-- SECOND positional arg, before the callback. We pass it explicitly so
+-- each `fs_readdir` returns up to `n` entries per round-trip.
+local READDIR_BATCH = 256
+local function fs_opendir(path)
+    return await(function(resolve) uv.fs_opendir(path, resolve, READDIR_BATCH) end)
+end
+local function fs_readdir(dir)
+    return await(function(resolve) uv.fs_readdir(dir, resolve) end)
+end
+local function fs_closedir(dir)
+    return await(function(resolve) uv.fs_closedir(dir, resolve) end)
+end
+
 local function scandir_sorted(path)
-    local req, err = uv.fs_scandir(path)
-    if not req then
+    -- Use the streaming opendir/readdir/closedir API rather than
+    -- `fs_scandir` + `fs_scandir_next`: the latter's iterator advances
+    -- with synchronous `readdir(3)` syscalls, which would block the
+    -- shared event loop on large directories. opendir/readdir route
+    -- every batch through libuv's thread pool, keeping us cooperative.
+    local err, dir = fs_opendir(path)
+    if not dir then
         return nil, err
     end
 
     local entries = {}
     while true do
-        local name, typ = uv.fs_scandir_next(req)
-        if not name then break end
-        entries[#entries + 1] = { name = name, typ = typ }
+        local rerr, batch = fs_readdir(dir)
+        if rerr then
+            fs_closedir(dir)
+            return nil, rerr
+        end
+        -- A nil/empty batch signals end of the directory stream.
+        if batch == nil or #batch == 0 then break end
+        for _, e in ipairs(batch) do
+            -- async readdir yields { name = ..., type = ... }.
+            entries[#entries + 1] = { name = e.name, typ = e.type }
+        end
     end
+    fs_closedir(dir)
 
     table.sort(entries, function(a, b)
         return a.name < b.name
@@ -136,7 +216,7 @@ return {
             return string.format("Error: limit (%d) must be >= 1.", limit)
         end
 
-        local stat, stat_err = uv.fs_stat(path)
+        local stat_err, stat = fs_stat(path)
         if not stat then
             return "Error: " .. (stat_err or ("could not stat " .. path))
         end
@@ -144,7 +224,7 @@ return {
             return render_directory_listing(path, offset, limit)
         end
 
-        local fd, open_err = uv.fs_open(path, "r", 0)
+        local open_err, fd = fs_open(path, "r", 0)
         if not fd then
             return "Error: " .. (open_err or ("could not open " .. path))
         end
@@ -190,12 +270,16 @@ return {
         end
 
         while not done do
-            local data, err = uv.fs_read(fd, READ_CHUNK_SIZE, file_offset)
+            local err, data = fs_read(fd, READ_CHUNK_SIZE, file_offset)
             if err then
                 read_err = err
                 break
             end
-            if data == nil then
+            -- luv's `fs_read` signals EOF by returning the empty
+            -- string, NOT nil (nil only accompanies an error). Treat
+            -- both nil and "" as EOF; otherwise the loop re-reads at
+            -- the same offset forever, pinning a core at 100%.
+            if data == nil or #data == 0 then
                 if #carry > 0 then
                     process_line(carry)
                     carry = ""
@@ -226,7 +310,7 @@ return {
             end
         end
 
-        uv.fs_close(fd)
+        fs_close(fd)
 
         if read_err then
             return "Error: read failed: " .. tostring(read_err)