From 61863eb46d88a69e95c3f748a1e9df2d4b547f2a Mon Sep 17 00:00:00 2001 From: t Date: Fri, 30 Oct 2026 01:58:33 -0600 Subject: Start warming turns with callback-form run_async Drive the warm turn from the luv timer with callback run_async, caching only its termination, and abandon with a nonjoining close. Removes native job polling. Adds binding-check.lua, a real-binding check against local HTTP fixtures, to mise run check. Requires a Pantograph host with callback-form run_async. --- README.md | 26 +++-- _selfcheck.lua | 167 +++++++++++++++++++++++++++---- binding-check.lua | 220 +++++++++++++++++++++++++++++++++++++++++ init.lua | 10 +- mise.toml | 4 +- panto-warming-0.2.0-1.rockspec | 10 +- warming/warm.lua | 72 +++++++------- 7 files changed, 436 insertions(+), 73 deletions(-) create mode 100644 binding-check.lua diff --git a/README.md b/README.md index 68876c4..d83def9 100644 --- a/README.md +++ b/README.md @@ -23,6 +23,12 @@ For a local checkout, use `paths = ["/path/to/panto-warming"]` instead. A in panto's own rocks tree first (`mise run deps`, or the `luarocks` line it contains). +## Host compatibility + +This source requires Pantograph's callback form +`agent:run_async(request, callback)`; hosts exposing only the older no-callback +job API are unsupported. + ## Configuration Enable with defaults in any layered `config.toml` (the built-in base layer, @@ -88,10 +94,10 @@ sequenceDiagram participant Provider Timer->>Store: replay transcript (same as /resume) Timer->>Warm: panto.agent { conversation, model, tools, null_store } - Timer->>Warm: run_async { transient prompt, dispatch_tools = false } + Timer->>Warm: run_async { transient prompt, dispatch_tools = false }, callback Warm->>Provider: [session prefix] + [transient user message] Provider-->>Warm: one response - Warm--)Timer: settled; response discarded + Warm--)Timer: callback caches termination; response discarded ``` The warm request shares the session's prompt prefix and differs only by the @@ -106,10 +112,11 @@ prefix again and keeps it hot. Then everything is dropped: - **Model** — `panto.ext.resolve_model` for the session's current model/reasoning label, inheriting the live credentialed transport. - **Tools** — the session agent's declarations, copied onto the warm agent. -- **Turn** — one `run_async` on its own pump thread (the UI never blocks), - with `dispatch_tools = false`: a misbehaving model cannot run real tools - from a keep-alive prompt, and tool calls are handed back unresolved and - discarded with the rest. +- **Turn** — one callback-form `run_async` on its own pump thread (the UI + never blocks), with `dispatch_tools = false`: a misbehaving model cannot run + real tools from a keep-alive prompt, and tool calls are handed back + unresolved and discarded with the rest. Progress deliveries are discarded; + only the terminal status/error is cached for the next timer tick. The warm turn's user message and response live only in the throwaway agent's conversation and a `null_store` — the live session, its transcript, @@ -151,10 +158,11 @@ one uninterrupted 30-minute window sends at most seven warming requests. ```sh mise trust # once mise run deps # install toml2lua into panto's rocks tree (for `paths` use) -mise run check # panto lua _selfcheck.lua +mise run check # self-check plus real-binding HTTP integration ``` `warming/policy.lua` and `warming/config.lua` are pure (durations, layer merging, the timing state machine) and fully covered by the self-check; -`warming/warm.lua` is the binding glue. Release plumbing: `mise run version -`, commit, `mise run release`. +`warming/warm.lua` is the binding glue. `mise run check` also runs +`binding-check.lua` against the real callback binding and local HTTP fixtures. +Release plumbing: `mise run version `, commit, `mise run release`. diff --git a/_selfcheck.lua b/_selfcheck.lua index 6cc3e4a..3231646 100644 --- a/_selfcheck.lua +++ b/_selfcheck.lua @@ -184,37 +184,168 @@ do local previous_warm = package.loaded["warming.warm"] package.loaded["warming.config"] = { default_io = function() return {} end, - read = function() return { enabled = true, interval_ms = MIN, duration_ms = MIN, prompt = "OK" } end, + read = function() return { enabled = true, interval_ms = 1, duration_ms = 100, prompt = "OK" } end, } - package.loaded["warming.policy"] = { - new = function() - return { - activity = function(self, now) self.activity_at = now end, - turn_started = function() end, turn_ended = function() end, - expired = function() return false end, due = function() return false end, - attempted = function() end, + package.loaded["warming.policy"] = policy + local real_warm = require("warming.warm") + local handles = {} + package.loaded["warming.warm"] = { + start = function(...) + local handle, err = real_warm.start(...) + if handle then handles[#handles + 1] = handle end + return handle, err + end, + poll = real_warm.poll, + abandon = real_warm.abandon, + } + local starts, callback_queue, forbidden_calls = {}, {}, 0 + local transcript, no_store = {}, {} + local declarations = { { name = "tool", parameters = {} } } + local session_store = { + load = function(_, id) + assert(id == "session-1", "warm request loads the active transcript") + return transcript + end, + } + local warm_agent = { + set_tools = function(_, tools) + assert(tools == declarations, "session tool declarations are copied") + end, + run_async = function(_, request, on_delivery) + assert(type(on_delivery) == "function", "callback form is used") + assert(request.prompt == "OK" and request.dispatch_tools == false, + "the transient request preserves tool isolation") + local stream = { + close = function(self) self.close_calls = (self.close_calls or 0) + 1 end, } + for _, name in ipairs({ "next_event", "result", "request_cancel", "join" }) do + stream[name] = function() + forbidden_calls = forbidden_calls + 1 + error("legacy native-job method used: " .. name) + end + end + starts[#starts + 1] = { request = request, callback = on_delivery, stream = stream } + return stream end, } - package.loaded["warming.warm"] = { start = function() end, poll = function() end, abandon = function() end } - package.loaded.panto = { agent = function() end, ext = { - on = function(name, handler) handlers[name] = handler end, - resolve_model = function() end, - style = function() return { dim = "", reset = "" } end, - } } + local current_time, timer + package.loaded.panto = { + file_system_jsonl_store = function(options) + assert(options.dir == "/sessions", "the active session store is reused") + return session_store + end, + null_store = function() return no_store end, + agent = function(options) + assert(options.conversation == transcript and options.store == no_store, + "the warm agent is isolated over the replayed transcript") + return warm_agent + end, + ext = { + on = function(name, handler) handlers[name] = handler end, + session_info = function() + return { session_dir = "/sessions", session_id = "session-1", model = "model", reasoning = "low" } + end, + resolve_model = function(options) + assert(options.model == "model" and options.reasoning == "low", "session model is retained") + return { resolved = true } + end, + agent = { tools = function() return declarations end }, + style = function() return { dim = "", reset = "" } end, + }, + } package.loaded.luv = { - now = function() return 123 end, + now = function() return current_time end, new_timer = function() - return { start = function(self) self.active = true end, is_active = function(self) return self.active end, - is_closing = function() return false end, stop = function(self) self.active = false end, - close = function() end } + timer = { + start = function(self, _, _, callback) + self.callback = callback + self.active = true + end, + is_active = function(self) return self.active end, + is_closing = function(self) return self.closing or false end, + stop = function(self) self.active = false end, + close = function(self) self.closing = true end, + } + return timer end, } local active_entry = assert(loadfile(here .. "/init.lua"))() active_entry.activate() assert(handlers.render_header and handlers.session_ready, "new event subscriptions are installed") assert(handlers.session_start == nil, "old session_start subscription is absent") + current_time = 0 handlers.session_ready() + assert(timer and timer.active, "session readiness starts the timer") + + local function tick(at) + current_time = at + timer.callback() + end + local function queue_delivery(index, delivery) + callback_queue[#callback_queue + 1] = { index = index, delivery = delivery } + end + local function advance_delivery() + local queued = table.remove(callback_queue, 1) + assert(queued ~= nil, "callback queue has a delivery to advance") + starts[queued.index].callback(queued.delivery) + end + + tick(0) + assert(#starts == 0, "the interval prevents an immediate turn") + tick(1) + assert(#starts == 1 and #callback_queue == 0 and handles[1].result == nil, + "installation returns before any callback is queued or delivered") + assert(handles[1].agent == warm_agent and handles[1].store == session_store + and handles[1].conv == transcript, "owners stay rooted while the turn is live") + queue_delivery(1, { kind = "result", result = { type = "message_start" } }) + assert(handles[1].result == nil, "queued values do not settle before delivery") + advance_delivery() + tick(2) + assert(#starts == 1 and handles[1].result == nil, "result events are discarded, not settled") + queue_delivery(1, { kind = "termination", status = "completed" }) + advance_delivery() + assert(handles[1].result.status == "completed", "completed termination is cached") + tick(3) + assert(#starts == 1, "the completion tick only settles the previous attempt") + + tick(4) + assert(#starts == 2, "the next timer turn starts after the interval") + queue_delivery(1, { kind = "termination", status = "cancelled" }) + advance_delivery() + assert(handles[1].result.status == "completed", "a stale callback cannot rewrite a prior terminal") + tick(4.5) + assert(#starts == 2 and handles[2].result == nil, + "a stale callback cannot settle or replace the current handle") + queue_delivery(2, { kind = "termination", status = "failed", error = { message = "provider failure" } }) + advance_delivery() + tick(5) + assert(handles[2].result.status == "failed" and handles[2].result.err == "provider failure", + "failure termination and its diagnostic are cached") + + tick(6) + assert(#starts == 3, "a failed attempt retries after the interval") + queue_delivery(3, { kind = "termination", status = "cancelled" }) + advance_delivery() + tick(7) + assert(handles[3].result.status == "cancelled", "cancel termination settles the attempt") + + tick(8) + assert(#starts == 4, "a cancelled attempt also follows the retry policy") + local abandoned = handles[4] + handlers.session_end() + assert(timer.closing and not timer.active, "session teardown stops and closes its timer") + assert(abandoned.stream.close_calls == 1 and abandoned.abandoned, + "teardown closes the live subscription without joining") + assert(abandoned.agent == warm_agent and abandoned.store == session_store + and abandoned.conv == transcript, "teardown retains owners through state retirement") + queue_delivery(4, { kind = "termination", status = "completed" }) + advance_delivery() + assert(real_warm.poll(abandoned) == nil, "a stale post-close delivery cannot settle an abandoned turn") + tick(9) + handlers.session_ready() + assert(#starts == 4, "stale timer/session callbacks cannot start work after teardown") + assert(forbidden_calls == 0, "no next_event, native result, cancel-join, or join API was used") + local inner = { render = function() return { "header" } end } local event = { get_component = function() return inner end } event.set_component = function(_, component) event.component = component end diff --git a/binding-check.lua b/binding-check.lua new file mode 100644 index 0000000..8f35f39 --- /dev/null +++ b/binding-check.lua @@ -0,0 +1,220 @@ +-- Real libpanto-lua callback integration check for panto-warming. +-- Run under the callback-capable host with: panto lua binding-check.lua +-- Local TCP fixtures exercise refusal, a scripted SSE response, and a held +-- request; synchronization uses loop events, never sleeps. +local here = (arg and arg[0] or "binding-check.lua"):match("^(.*)/") or "." +package.path = here .. "/?.lua;" .. here .. "/?/init.lua;" .. package.path + +local panto = require("panto") +local uv = require("luv") +local warm = require("warming.warm") +local declarations = { + { name = "warming_probe", description = "integration fixture", parameters = { + type = "object", properties = {}, + } }, +} +local resources, failure = {}, nil +local attached = false +local shutdown_done = false + +local function track(handle) + resources[#resources + 1] = handle + return handle +end + +local function guarded(fn) + return function(...) + local ok, err = pcall(fn, ...) + if not ok then failure = tostring(err) end + end +end + +local function run_until(done, what) + while not done() do + assert(not failure, failure) + local alive = uv.run("once") + assert(not failure, failure) + assert(alive or done(), what .. ": loop became idle before completion") + end + assert(not failure, failure) +end + +local function close(handle) + if handle and not handle:is_closing() then handle:close() end +end + +local function close_and_wait(handle) + local done = false + handle:close(function() done = true end) + run_until(function() return done end, "waiting for local TCP close") +end + +local function close_resources() + for _, handle in ipairs(resources) do close(handle) end +end + +local function configure(port) + panto.ext = { + session_info = function() + return { session_dir = "/tmp/panto-warming-check", session_id = "check", model = "fixture" } + end, + resolve_model = function() + return { + api_style = "openai_chat", api_key = "local-only", + base_url = "http://127.0.0.1:" .. port, model = "fixture", + } + end, + agent = { tools = function() return declarations end }, + } +end + +local function timer_start(port, prompt) + configure(port) + local handle, start_error + local timer = track(uv.new_timer()) + timer:start(0, 0, guarded(function() + close(timer) + handle, start_error = warm.start(prompt) + if handle then + assert(warm.poll(handle) == nil, "run_async delivered inline from timer startup") + end + end)) + run_until(function() return handle ~= nil or start_error ~= nil end, "starting warm turn") + assert(handle, "warm.start failed: " .. tostring(start_error)) + return handle +end + +local function wait_terminal(handle, what) + run_until(function() return warm.poll(handle) ~= nil end, what) + return warm.poll(handle) +end + +local function closed_unused_port() + local socket = track(uv.new_tcp()) + assert(socket:bind("127.0.0.1", 0)) + local port = socket:getsockname().port + close_and_wait(socket) + return port +end + +local function listener(on_client) + local server = track(uv.new_tcp()) + assert(server:bind("127.0.0.1", 0)) + server:listen(8, guarded(function(err) + assert(not err, tostring(err)) + local client = track(uv.new_tcp()) + server:accept(client) + on_client(client) + end)) + return server, server:getsockname().port +end + +local function serve_sse() + local request, client + local body = table.concat({ + 'data: {"id":"warm","object":"chat.completion.chunk","created":1,"model":"fixture","choices":[{"index":0,"delta":{"role":"assistant","content":"OK"},"finish_reason":null}]}\n\n', + 'data: {"id":"warm","object":"chat.completion.chunk","created":1,"model":"fixture","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\n', + "data: [DONE]\n\n", + }) + local response = "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nContent-Length: " + .. #body .. "\r\nConnection: close\r\n\r\n" .. body + local server, port = listener(function(accepted) + client = accepted + local received = "" + accepted:read_start(guarded(function(err, data) + assert(not err, tostring(err)) + if not data then return end + received = received .. data + local boundary = received:find("\r\n\r\n", 1, true) + if not boundary then return end + local header = received:sub(1, boundary - 1):lower() + local length = tonumber(header:match("\r\ncontent%-length:%s*(%d+)")) + assert(length, "request omitted Content-Length") + local first_body = boundary + 4 + if #received < first_body - 1 + length then return end + request = received:sub(first_body, first_body + length - 1) + accepted:read_stop() + accepted:write(response, guarded(function(write_err) + assert(not write_err, tostring(write_err)) + accepted:shutdown(guarded(function(shutdown_err) + assert(not shutdown_err, tostring(shutdown_err)) + close(accepted) + end)) + end)) + end)) + end) + return server, port, function() return request end, function() return client end +end + +local function serve_and_hold() + local accepted, client + local server, port = listener(function(socket) + client = socket + accepted = true + socket:read_start(guarded(function(err) + assert(not err, tostring(err)) + end)) + end) + return server, port, function() return accepted end +end + +local function main() + assert(panto.async and type(panto.async.attach_luv) == "function", + "callback-capable panto.async API is required") + assert(panto.async.attach_luv()) + attached = true + panto.file_system_jsonl_store = function() + return { load = function() return panto.conversation() end } + end + + -- A closed ephemeral listener gives a deterministic local refusal. + local refused = timer_start(closed_unused_port(), "refused-probe") + local refused_result = wait_terminal(refused, "waiting for refused request") + assert(refused_result.status == "failed" and type(refused_result.err) == "string", + "connection refusal must arrive as failed termination with a diagnostic") + + -- A real OpenAI-compatible SSE response completes a callback turn. Verify + -- that copied declarations and the transient prompt reach the provider. + local sse_server, sse_port, get_request, get_sse_client = serve_sse() + local completed = timer_start(sse_port, "sse-probe") + local complete_result = wait_terminal(completed, "waiting for scripted SSE") + assert(complete_result.status == "completed", "scripted SSE must complete the warm turn") + local request = assert(get_request(), "SSE fixture did not receive the HTTP request") + assert(request:find('"stream":true', 1, true), "warm request must stream") + assert(request:find('"tools"', 1, true) and request:find('"warming_probe"', 1, true), + "session tool declarations must be copied to the warm request") + assert(request:find("sse-probe", 1, true), "transient prompt must reach the provider") + close(get_sse_client()) + close(sse_server) + + -- Keep a request pending, then close its subscription. Native shutdown + -- joins it without delivering a late result to the extension. + local hold_server, hold_port, was_accepted = serve_and_hold() + local abandoned = timer_start(hold_port, "abandon-probe") + run_until(function() return was_accepted() or warm.poll(abandoned) ~= nil end, + "waiting for held request acceptance") + assert(was_accepted() and warm.poll(abandoned) == nil, "held request should still be live") + warm.abandon(abandoned) + assert(abandoned.abandoned and abandoned.result == nil, "abandon clears cached state") + + assert(panto.async.shutdown("wait"), "standalone binding shutdown must finish") + shutdown_done = true + assert(warm.poll(abandoned) == nil and abandoned.result == nil, + "closed subscription must suppress terminal delivery") + close(hold_server) +end + +local ok, err = xpcall(main, debug.traceback) +if not ok then + if attached and not shutdown_done then pcall(panto.async.shutdown, "wait") end + close_resources() + pcall(uv.run, "nowait") + io.stderr:write("panto-warming real-binding check failed: " .. tostring(err) .. "\n") + error(err, 0) +end + +-- Close all local fixtures after orderly binding shutdown, then let luv +-- deliver their close callbacks before this standalone state exits. +close_resources() +uv.run() +io.stderr:write("panto-warming real-binding check: OK\n") diff --git a/init.lua b/init.lua index 5043e60..e552fe5 100644 --- a/init.lua +++ b/init.lua @@ -19,8 +19,8 @@ local config = require("warming.config") local policy_mod = require("warming.policy") local warm = require("warming.warm") --- A tick is only bookkeeping (poll a job, compare timestamps), so a coarse --- period is plenty; it is clamped to [1s, 30s]. +-- A tick only inspects extension-owned terminal state and starts a callback +-- turn when due; it never polls or advances a native operation. local TICK_MAX_MS = 30000 local function host() @@ -126,7 +126,8 @@ local function activate() tick = function() local now = uv.now() if current then - local result = warm.poll(current) + local handle = current + local result = warm.poll(handle) if result then current = nil policy:attempted(now) @@ -173,8 +174,9 @@ local function activate() timer = nil end if current then + -- Keep the wrapper references through Lua-state teardown; close is + -- nonjoining and the binding manager owns native retirement. warm.abandon(current) - current = nil end end) diff --git a/mise.toml b/mise.toml index d773f80..c2cc2d4 100644 --- a/mise.toml +++ b/mise.toml @@ -18,9 +18,9 @@ luarocks --lua-version 5.4 --tree "$TREE" install toml2lua """ [tasks.check] -description = "Run the self-check inside panto's interpreter" +description = "Run the pure and real-binding checks inside panto's interpreter" dir = "{{config_root}}" -run = "panto lua _selfcheck.lua" +run = "panto lua _selfcheck.lua && panto lua binding-check.lua" [tasks.version] description = "Rewrite the rockspec for a new version (then commit it)" diff --git a/panto-warming-0.2.0-1.rockspec b/panto-warming-0.2.0-1.rockspec index 7340e08..198be3e 100644 --- a/panto-warming-0.2.0-1.rockspec +++ b/panto-warming-0.2.0-1.rockspec @@ -14,6 +14,10 @@ -- and put a system package between a user and a working extension. `luv` -- owns the warming timer and ships with panto as a pinned battery, so it is -- normally already present in the tree this rock installs into. +-- +-- Host compatibility. Pantograph is not a LuaRocks dependency: the host pins +-- this extension. This source uses callback-form agent:run_async(request, +-- callback); older no-callback hosts are unsupported. rockspec_format = "3.0" package = "panto-warming" @@ -30,8 +34,10 @@ description = { Periodically sends a minimal transient prompt on top of the session's current context so provider-side prompt caches stay warm between turns, then discards the prompt and the response -- nothing enters the context - window or the session transcript. Configured through a [warming] - section in config.toml, mirroring opencode's warming feature. + window or the session transcript. It requires a Pantograph host exposing + callback-form agent:run_async(request, callback); older no-callback hosts + are unsupported. Configured through a [warming] section in config.toml, + mirroring opencode's warming feature. ]], homepage = "https://code.tjp.lol/public/panto-warming", license = "MIT", diff --git a/warming/warm.lua b/warming/warm.lua index e5e41fc..c8b3af3 100644 --- a/warming/warm.lua +++ b/warming/warm.lua @@ -18,11 +18,8 @@ -- store `panto.null_store()` -- the warm turn's user message and -- response are appended to the throwaway conversation and -- persisted nowhere. --- turn `agent:run_async { prompt, dispatch_tools = false }` on its --- own pump thread, so the UI thread never blocks. Tools are --- never dispatched: a misbehaving model cannot run real --- tools from a keep-alive prompt. The response is drained --- and dropped. +-- turn callback-form `agent:run_async(request, callback)`, started +-- by the timer. Progress is ignored; only termination is cached. -- -- Nothing here raises: the binding reports failures by raising, and every -- call goes through `try` so callers get one `nil, message` shape. @@ -40,7 +37,8 @@ end -- start(prompt) -> handle | nil, err -- -- Everything that can fail before the turn starts fails here: a caller --- either gets a handle to poll or a message. +-- either gets a handle to inspect or a message. The callback retains the +-- throwaway Lua owners until termination, and never retains event payloads. function M.start(prompt) local panto = require("panto") local ext = panto.ext @@ -111,50 +109,48 @@ function M.start(prompt) end end - local started, job = try(agent.run_async, agent, { + local handle = { agent = agent, store = store, conv = conv } + local function on_delivery(delivery) + if handle.abandoned or handle.result or type(delivery) ~= "table" or delivery.kind ~= "termination" then + return + end + local status = delivery.status + local err = type(delivery.error) == "table" and delivery.error.message or nil + handle.result = { status = status, err = err } + end + + local started, stream = try(agent.run_async, agent, { prompt = prompt, dispatch_tools = false, - }) + }, on_delivery) if not started then - return nil, "warm turn: " .. tostring(job) + return nil, "warm turn: " .. tostring(stream) end - if job == nil then + if stream == nil then return nil, "the warm turn could not be started" end - - return { agent = agent, job = job, store = store, conv = conv } + handle.stream = stream + return handle end --- poll(handle) -> nil while the pump runs, then a settled --- { status = "completed"|"failed"|"cancelled", err = ... }. Drains buffered --- events as it goes so a chatty response cannot grow the queue. The result --- content is deliberately discarded; `close` joins the (settled) pump. +-- poll(handle) returns only the cached terminal state. It never calls into +-- the native stream; callbacks own terminal observation and discard events. function M.poll(handle) - local job = handle.job - while true do - local ok, event = pcall(job.next_event, job) - if not ok or event == nil then - break - end - end - local ok, result = pcall(job.result, job) - if not ok then - pcall(job.close, job) - return { status = "failed", err = tostring(result) } - end - if type(result) ~= "table" then - return nil - end - pcall(job.close, job) - return { status = result.status or "failed", err = result.err } + return handle.result end --- abandon(handle): cancel and join the pump, on session teardown. The --- cancel half shuts the response socket down, so the join returns promptly --- even mid-request. +-- Abandon without joining. close() suppresses future callbacks and requests +-- cancellation; the binding manager retains native dependencies until worker +-- retirement. The caller keeps this handle through state teardown. function M.abandon(handle) - pcall(handle.job.request_cancel, handle.job) - pcall(handle.job.close, handle.job) + if handle.abandoned then + return + end + handle.abandoned = true + handle.result = nil + if handle.stream then + pcall(handle.stream.close, handle.stream) + end end return M -- cgit v1.3