diff options
Diffstat (limited to 'src/lua_runtime.zig')
| -rw-r--r-- | src/lua_runtime.zig | 522 |
1 files changed, 486 insertions, 36 deletions
diff --git a/src/lua_runtime.zig b/src/lua_runtime.zig index fb00951..71e4ae7 100644 --- a/src/lua_runtime.zig +++ b/src/lua_runtime.zig @@ -58,6 +58,23 @@ pub const LuaRuntime = struct { /// the Lua registry (`luaL_ref` index). handlers: std.StringHashMap(c_int), + /// Registry ref to the wrapper closure that runs a user handler + /// inside a `pcall` and reports the result back to Zig via + /// `panto._record_result`. Allocated once at `create`; reused for + /// every call. `0` until `installScheduler` runs. + wrapper_ref: c_int = 0, + /// Registry ref to `require("luv").run`, the function we call to + /// tick libuv between coroutine resumes. `0` until + /// `installScheduler` runs. + uv_run_ref: c_int = 0, + + /// Pointer to the in-flight batch, valid only for the duration of + /// one `invoke_batch` call. The `panto._record_result` C function + /// writes through this. `null` between batches; not concurrently + /// accessible (libpanto's source-grouped dispatch guarantees one + /// thread per source per turn). + current_batch: ?*BatchState = null, + /// Create a new runtime. The `lua_State` is opened, standard libs /// loaded, and the `panto.register_tool` bridge installed. pub fn create(allocator: Allocator) !*LuaRuntime { @@ -79,6 +96,22 @@ pub const LuaRuntime = struct { return self; } + /// Install the libuv-driven coroutine scheduler: + /// - Register `panto._record_result` (C function with `self` as + /// light-userdata upvalue) so the wrapper closure can hand + /// results back to Zig. + /// - Create the wrapper closure that runs a user handler in + /// `pcall` and reports the result. + /// - Cache `require("luv").run` for fast per-tick access. + /// + /// Must be called after luarocks bootstrap has installed luv, + /// otherwise the `require("luv")` step will fail. + pub fn installScheduler(self: *LuaRuntime) !void { + try installRecordResult(self); + try installWrapperClosure(self); + try cacheUvRun(self); + } + /// Tear down the runtime: free every owned string, unref every /// handler, close the Lua state. pub fn deinit(self: *LuaRuntime) void { @@ -90,6 +123,12 @@ pub const LuaRuntime = struct { c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, entry.value_ptr.*); } self.handlers.deinit(); + if (self.wrapper_ref != 0) { + c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.wrapper_ref); + } + if (self.uv_run_ref != 0) { + c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, self.uv_run_ref); + } c.lua_close(self.L); @@ -320,6 +359,52 @@ fn deinitSrc(_: *anyopaque, _: Allocator) void { // libpanto's source.deinit here is a no-op. } +// =========================================================================== +// Scheduler: libuv-driven cooperative coroutine dispatch +// =========================================================================== +// +// libpanto's `invoke_batch` delivers all of a turn's tool-call requests +// at once, on a single thread. We answer the contract by running each +// call as a Lua coroutine inside our long-lived `lua_State`, then +// driving `uv.run("once")` to wake any of those coroutines that are +// blocked on libuv-aware I/O. This is the entire scheduler — luv's +// libuv binding does the actual event-loop work; we just resume +// coroutines and call `run` between resumes. +// +// Capturing return values requires a wrapper. When a coroutine is +// resumed by a luv callback after yielding, the eventual return value +// of the coroutine flows back to *that callback*, not to us. So we +// install a Lua wrapper closure that does +// +// pcall(handler, input) → panto._record_result(idx, ok, val) +// +// before the handler returns. `_record_result` is a C function that +// stores into a per-runtime `BatchState`, accessed via a light-userdata +// upvalue carrying the runtime pointer. + +/// One coroutine's outcome, recorded by `_record_result` and read by +/// `invokeBatch` once the coroutine has terminated. +const Slot = struct { + /// Set true the moment `_record_result` writes a result for this + /// index. Used to detect coroutines that terminated without + /// calling the wrapper (a bug / API misuse). + recorded: bool = false, + /// `true` if the handler returned cleanly, `false` if it raised + /// via the `pcall` wrapping. + ok: bool = false, + /// Result payload as owned bytes. Allocated from `allocator`. + /// Caller frees. + value: ?[]u8 = null, + /// On `ok = false`, an owned copy of the error message. + err_msg: ?[]u8 = null, +}; + +/// State shared between Zig and the in-flight Lua wrapper closure. +const BatchState = struct { + allocator: Allocator, + slots: []Slot, +}; + fn invokeBatch( ctx: *anyopaque, calls: []const panto.ToolCall, @@ -327,21 +412,205 @@ fn invokeBatch( allocator: Allocator, ) anyerror!void { const self: *LuaRuntime = @ptrCast(@alignCast(ctx)); + return runBatch(self, calls, results, allocator); +} - // Step 2 of LUA_MAKEOVER.md: no batteries yet — each call is run - // as a coroutine, but the scheduler doesn't drive an event loop. - // A handler that yields with no batteries available has nothing - // to wake it; we surface that as `LuaHandlerYielded`. - // - // Once `luv` and the `coro-*` wrappers are installed, this loop - // becomes "drive uv.run() until every coroutine is dead/erroring, - // then collect results". +fn runBatch( + self: *LuaRuntime, + calls: []const panto.ToolCall, + results: []panto.ToolCallResult, + allocator: Allocator, +) !void { + if (self.wrapper_ref == 0 or self.uv_run_ref == 0) { + // Scheduler not installed. We can still run synchronous + // handlers — use the legacy path that drives one coroutine at + // a time without an event loop. + for (calls, 0..) |call, i| { + results[i] = runLegacySync(self, call, allocator); + } + return; + } + + var slots = try allocator.alloc(Slot, calls.len); + defer allocator.free(slots); + for (slots) |*s| s.* = .{}; + + var batch_state: BatchState = .{ .allocator = allocator, .slots = slots }; + self.current_batch = &batch_state; + defer self.current_batch = null; + + // Track each call's coroutine reference in the parent stack (we + // hold them in registry refs so they survive across `uv.run` + // ticks). `0` after a coroutine has been reaped. + var thread_refs = try allocator.alloc(c_int, calls.len); + defer allocator.free(thread_refs); + @memset(thread_refs, 0); + defer for (thread_refs) |r| { + if (r != 0) c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, r); + }; + + var pending: usize = 0; for (calls, 0..) |call, i| { - results[i] = runOneCall(self, call, allocator); + const handler_ref = self.handlers.get(call.tool_name) orelse { + // Synthesize a recorded "err" result; don't even bother + // spawning a coroutine. + slots[i] = .{ + .recorded = true, + .ok = false, + .err_msg = try allocator.dupe(u8, "panto: unknown tool name"), + }; + continue; + }; + const t = try startCoroutine(self, i, handler_ref, call.input, allocator); + thread_refs[i] = t.thread_ref; + if (t.still_pending) pending += 1; + } + + while (pending > 0) { + const active = try driveUvOnce(self); + // Reap any coroutines that terminated during the tick. + var reaped: usize = 0; + for (thread_refs, 0..) |tref, i| { + if (tref == 0) continue; + // Push the thread, check status, pop. + _ = c.lua_rawgeti(self.L, lua_bridge.LUA_REGISTRYINDEX, tref); + const co: *c.lua_State = @ptrCast(c.lua_tothread(self.L, -1).?); + const status = c.lua_status(co); + c.lua_settop(self.L, c.lua_gettop(self.L) - 1); + if (status != c.LUA_YIELD) { + // Terminated (LUA_OK or error). The wrapper should + // have called `_record_result` already; if not, synthesize. + if (!slots[i].recorded) { + slots[i] = .{ + .recorded = true, + .ok = false, + .err_msg = try allocator.dupe( + u8, + "panto: handler terminated without recording a result", + ), + }; + } + c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, tref); + thread_refs[i] = 0; + reaped += 1; + } + } + if (reaped > 0) { + if (reaped > pending) { + // Defensive: keep the counter sane. + pending = 0; + } else { + pending -= reaped; + } + continue; + } + if (active == 0) { + // No libuv handles are pending, but we still have alive + // coroutines. They yielded without arranging to be woken. + // Mark them as failed and break. + for (thread_refs, 0..) |tref, i| { + if (tref == 0) continue; + slots[i] = .{ + .recorded = true, + .ok = false, + .err_msg = try allocator.dupe( + u8, + "panto: handler yielded but no libuv handle is pending; " ++ + "did you forget to await with luv?", + ), + }; + c.luaL_unref(self.L, lua_bridge.LUA_REGISTRYINDEX, tref); + thread_refs[i] = 0; + } + break; + } + } + + // Translate slots into the libpanto-shaped results. + for (slots, 0..) |slot, i| { + if (!slot.recorded) { + results[i] = .{ .err = RuntimeError.BadHandlerReturn }; + continue; + } + if (slot.ok) { + results[i] = .{ .ok = slot.value orelse try allocator.dupe(u8, "") }; + // Free the err_msg if both ended up set somehow. + if (slot.err_msg) |m| allocator.free(m); + } else { + if (slot.value) |v| allocator.free(v); + // The error message was logged for the user; we still + // surface a typed error to libpanto so it can route the + // failure to the agent's error path. + std.log.warn( + "panto-lua: tool '{s}' failed: {s}", + .{ + calls[i].tool_name, + slot.err_msg orelse "(no message)", + }, + ); + if (slot.err_msg) |m| allocator.free(m); + results[i] = .{ .err = RuntimeError.LuaHandlerCrashed }; + } } } -fn runOneCall( +/// Start one coroutine: create a thread under the runtime's lua_State, +/// push the wrapper closure + (idx, handler, input), `lua_resume` once. +/// +/// If the coroutine returns immediately (sync handler), the wrapper +/// has already recorded its result via `panto._record_result` — +/// `still_pending` will be `false`. +fn startCoroutine( + self: *LuaRuntime, + idx: usize, + handler_ref: c_int, + input: []const u8, + allocator: Allocator, +) !struct { thread_ref: c_int, still_pending: bool } { + const L = self.L; + + const co = c.lua_newthread(L) orelse return RuntimeError.LuaInitFailed; + // luaL_ref pops the topmost value (the thread) and returns a + // registry ref to it. We keep the ref alive for the lifetime of + // the call so GC doesn't collect the thread mid-yield. + const thread_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX); + + // Push the wrapper onto the coroutine's stack. + _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.wrapper_ref)); + // Push (idx, handler, input) as the resume args. + c.lua_pushinteger(co, @intCast(idx)); + _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref)); + var arena_state = std.heap.ArenaAllocator.init(allocator); + defer arena_state.deinit(); + try lua_bridge.pushJsonAsLua(co, arena_state.allocator(), input); + + var nres: c_int = 0; + const status = c.lua_resume(co, L, 3, &nres); + return .{ + .thread_ref = thread_ref, + .still_pending = status == c.LUA_YIELD, + }; +} + +/// Call `uv.run("once")`. Returns the number of active handles luv +/// reports remain pending (0 means the loop is drained). +fn driveUvOnce(self: *LuaRuntime) !c_int { + const L = self.L; + _ = c.lua_rawgeti(L, lua_bridge.LUA_REGISTRYINDEX, @intCast(self.uv_run_ref)); + _ = c.lua_pushlstring(L, "once", 4); + if (c.lua_pcallk(L, 1, 1, 0, 0, null) != 0) { + logTopAsError(L, "panto-lua: uv.run failed"); + c.lua_settop(L, c.lua_gettop(L) - 1); + return error.UvRunFailed; + } + const n = c.lua_tointegerx(L, -1, null); + c.lua_settop(L, c.lua_gettop(L) - 1); + return @intCast(n); +} + +/// Pre-scheduler fallback (used in unit tests and during early +/// startup before `installScheduler` has run). +fn runLegacySync( self: *LuaRuntime, call: panto.ToolCall, allocator: Allocator, @@ -349,62 +618,40 @@ fn runOneCall( const handler_ref = self.handlers.get(call.tool_name) orelse { return .{ .err = RuntimeError.LuaHandlerNotFound }; }; - - const out_bytes = invokeCoroutine(self.L, handler_ref, call.input, allocator) catch |e| { + const out_bytes = invokeCoroutineSync(self.L, handler_ref, call.input, allocator) catch |e| { return .{ .err = e }; }; return .{ .ok = out_bytes }; } -/// Create a fresh coroutine, push the handler + JSON-decoded input as -/// the resume args, then resume. Returns the handler's return value as -/// owned JSON bytes (the slot `ok` in CallResult). -/// -/// Resume outcomes: -/// - LUA_OK: coroutine returned. Read its top value as the result. -/// - LUA_YIELD: coroutine yielded. With no event loop installed, we -/// treat this as an error so the user sees that their handler is -/// trying to do async I/O that isn't yet supported. -/// - other (errors): error message is on the coroutine's stack; -/// copy it to a log line and return LuaHandlerCrashed. -fn invokeCoroutine( +fn invokeCoroutineSync( L: *c.lua_State, handler_ref: c_int, input: []const u8, allocator: Allocator, ) ![]u8 { - // Create the coroutine thread. After this, `co` is the child - // thread; the parent stack also gains a thread value at the top. const co = c.lua_newthread(L) orelse return RuntimeError.LuaInitFailed; - defer c.lua_settop(L, c.lua_gettop(L) - 1); // pop the thread when done + defer c.lua_settop(L, c.lua_gettop(L) - 1); - // Push handler from the registry onto the coroutine's stack. _ = c.lua_rawgeti(co, lua_bridge.LUA_REGISTRYINDEX, @intCast(handler_ref)); if (c.lua_type(co, -1) != lua_bridge.T_FUNCTION) { return RuntimeError.LuaHandlerNotFound; } - // Push the parsed JSON input as the resume arg. var arena_state = std.heap.ArenaAllocator.init(allocator); defer arena_state.deinit(); try lua_bridge.pushJsonAsLua(co, arena_state.allocator(), input); - // Resume with 1 arg. var nresults: c_int = 0; const status = c.lua_resume(co, L, 1, &nresults); switch (status) { c.LUA_OK => { - // Coroutine returned. We expect exactly one string return - // value (the tool result). If there are zero or extra - // values we still try to read top-of-stack. if (nresults < 1) return RuntimeError.BadHandlerReturn; return try lua_bridge.readHandlerResult(co, -1, allocator); }, c.LUA_YIELD => { - // Nothing to wake this coroutine without an event loop. - // Surface the situation so the user knows what's wrong. - const msg = "lua: tool handler yielded with no event loop installed (step 4+ of LUA_MAKEOVER.md not yet implemented)"; + const msg = "lua: tool handler yielded with no event loop installed; call installScheduler() before dispatching"; if (@import("builtin").is_test) { std.log.warn("{s}", .{msg}); } else { @@ -420,6 +667,124 @@ fn invokeCoroutine( } // --------------------------------------------------------------------------- +// Scheduler setup (called once at startup, after luarocks bootstrap) +// --------------------------------------------------------------------------- + +/// Register `panto._record_result(idx, ok, value)` on the `panto` +/// global. The C function carries the runtime pointer as an upvalue, +/// reaches the in-flight `BatchState` through `current_batch`, and +/// stores into the matching slot. +fn installRecordResult(self: *LuaRuntime) !void { + const L = self.L; + _ = c.lua_getglobal(L, "panto"); + if (c.lua_type(L, -1) != lua_bridge.T_TABLE) { + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + c.lua_pushlightuserdata(L, @ptrCast(self)); + c.lua_pushcclosure(L, recordResultC, 1); + c.lua_setfield(L, -2, "_record_result"); + c.lua_settop(L, c.lua_gettop(L) - 1); // pop `panto` +} + +fn recordResultC(L: ?*c.lua_State) callconv(.c) c_int { + const Lst = L.?; + const self_ptr = c.lua_touserdata(Lst, c.lua_upvalueindex(1)); + if (self_ptr == null) return 0; + const self: *LuaRuntime = @ptrCast(@alignCast(self_ptr.?)); + const batch = self.current_batch orelse return 0; + + const idx_i64 = c.lua_tointegerx(Lst, 1, null); + const ok = c.lua_toboolean(Lst, 2) != 0; + const idx: usize = @intCast(idx_i64); + if (idx >= batch.slots.len) return 0; + + if (ok) { + // Result value at index 3. The handler's return type is + // free-form; we serialize via the existing `readHandlerResult` + // helper which already knows how to JSON-encode any Lua value. + const value = lua_bridge.readHandlerResult(Lst, 3, batch.allocator) catch |e| { + // Allocation failure mid-callback is unrecoverable from + // Lua's POV; record a synthetic error and bail. + const msg = std.fmt.allocPrint( + batch.allocator, + "panto: failed to serialize handler result: {s}", + .{@errorName(e)}, + ) catch null; + batch.slots[idx] = .{ .recorded = true, .ok = false, .err_msg = msg }; + return 0; + }; + batch.slots[idx] = .{ .recorded = true, .ok = true, .value = value }; + } else { + // Error message at index 3. May be any Lua value; coerce to + // string via `tostring`-equivalent semantics. + var len: usize = 0; + const ptr = c.luaL_tolstring(Lst, 3, &len); + const owned = if (ptr != null) + batch.allocator.dupe(u8, ptr[0..len]) catch null + else + null; + c.lua_settop(Lst, c.lua_gettop(Lst) - 1); // pop tolstring's pushed string + batch.slots[idx] = .{ .recorded = true, .ok = false, .err_msg = owned }; + } + return 0; +} + +/// Create the per-call wrapper closure: +/// +/// local function wrapper(idx, handler, input) +/// local ok, val = pcall(handler, input) +/// panto._record_result(idx, ok, val) +/// end +/// +/// Stored in the Lua registry under `self.wrapper_ref`. +fn installWrapperClosure(self: *LuaRuntime) !void { + const L = self.L; + const snippet = + \\return function(idx, handler, input) + \\ local ok, val = pcall(handler, input) + \\ panto._record_result(idx, ok, val) + \\end + ; + if (c.luaL_loadstring(L, snippet) != 0) { + logTopAsError(L, "panto-lua: wrapper closure failed to compile"); + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) { + logTopAsError(L, "panto-lua: wrapper closure failed to evaluate"); + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + // Top of stack: the wrapper function. luaL_ref pops it. + self.wrapper_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX); +} + +/// Cache `require("luv").run` in the registry so the scheduler can +/// invoke it cheaply per tick. +fn cacheUvRun(self: *LuaRuntime) !void { + const L = self.L; + const snippet = + \\return require("luv").run + ; + if (c.luaL_loadstring(L, snippet) != 0) { + logTopAsError(L, "panto-lua: failed to compile luv lookup"); + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + if (c.lua_pcallk(L, 0, 1, 0, 0, null) != 0) { + logTopAsError(L, "panto-lua: require('luv') failed (was the bootstrap successful?)"); + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + if (c.lua_type(L, -1) != lua_bridge.T_FUNCTION) { + c.lua_settop(L, c.lua_gettop(L) - 1); + return RuntimeError.LuaInitFailed; + } + self.uv_run_ref = c.luaL_ref(L, lua_bridge.LUA_REGISTRYINDEX); +} + +// --------------------------------------------------------------------------- // Internals // --------------------------------------------------------------------------- @@ -715,6 +1080,91 @@ test "yielding handler with no event loop surfaces LuaHandlerYielded" { try testing.expectEqual(@as(anyerror, RuntimeError.LuaHandlerYielded), results[0].err); } +// Integration test: requires a `$PANTO_HOME` with luv already +// installed. Skipped if luv isn't on disk — unit tests stay offline. +test "scheduler: yielding handler is resumed by libuv" { + const home_z = std.c.getenv("PANTO_HOME") orelse return error.SkipZigTest; + const panto_home_env = std.mem.sliceTo(home_z, 0); + // Check for `<home>/rocks/lua-<version>/lib/lua/5.4/luv.so`. + const manifest = @import("manifest.zig"); + var path_buf: [std.fs.max_path_bytes]u8 = undefined; + const so_path = try std.fmt.bufPrint( + &path_buf, + "{s}/rocks/lua-{s}/lib/lua/{s}/luv.so", + .{ panto_home_env, manifest.lua_version, manifest.lua_short_version }, + ); + std.Io.Dir.cwd().access(testing.io, so_path, .{}) catch return error.SkipZigTest; + + var tmp = testing.tmpDir(.{}); + defer tmp.cleanup(); + + const source = + \\local uv = require("luv") + \\panto.register_tool { + \\ name = "timer_say", description = "sleep then return", + \\ schema = { type = "object" }, + \\ handler = function(input) + \\ local co = coroutine.running() + \\ local timer = uv.new_timer() + \\ uv.timer_start(timer, 5, 0, function() + \\ uv.timer_stop(timer) + \\ uv.close(timer) + \\ coroutine.resume(co) + \\ end) + \\ coroutine.yield() + \\ return "awake" + \\ end, + \\} + ; + const path = try writeTempScript(tmp.dir, "timer.lua", source); + defer testing.allocator.free(path); + + var rt = try LuaRuntime.create(testing.allocator); + defer rt.deinit(); + + // Bootstrap luarocks (so `require("luv")` works), then install + // the scheduler. We use the real environment so the test picks + // up the same PANTO_HOME the developer's machine has. + var env: std.process.Environ.Map = .init(testing.allocator); + defer env.deinit(); + try env.put("PANTO_HOME", panto_home_env); + + const luarocks_runtime = @import("luarocks_runtime.zig"); + // The bootstrap needs a panto executable path for the wrapper + // script; tests don't actually invoke it, so a placeholder is + // fine (the wrapper is only consulted when luarocks itself + // shells out, which the test never triggers). + const luarocks_rt = try luarocks_runtime.bootstrap( + testing.allocator, + testing.io, + &env, + rt.L, + "/usr/bin/true", + ); + defer luarocks_rt.deinit(); + + try rt.installScheduler(); + try rt.loadExtension(path, null); + + var src = rt.toolSource(); + const calls = [_]panto.ToolCall{ + .{ .tool_name = "timer_say", .input = "{}" }, + .{ .tool_name = "timer_say", .input = "{}" }, + }; + var results: [2]panto.ToolCallResult = .{ + .{ .err = error.SourceDroppedCall }, + .{ .err = error.SourceDroppedCall }, + }; + try src.vtable.invoke_batch(src.ctx, &calls, &results, testing.allocator); + defer for (results) |r| switch (r) { + .ok => |b| testing.allocator.free(b), + .err => {}, + }; + + try testing.expectEqualStrings("awake", results[0].ok); + try testing.expectEqualStrings("awake", results[1].ok); +} + test "loadExtension: duplicate tool name from a second extension errors" { var tmp = testing.tmpDir(.{}); defer tmp.cleanup(); |