Start warming turns with callback-form run_asyncHEADv0.2.0main

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.

1 files changed, 220 insertions, 0 deletions

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")