diff options
| author | t <t@tjp.lol> | 2026-08-04 12:56:25 -0600 |
|---|---|---|
| committer | t <t@tjp.lol> | 2026-08-04 13:42:28 -0600 |
| commit | ab6c0d1a9605be2e72de5095ff73bc4c005206aa (patch) | |
| tree | f77cb14102240ecfbacdde43dd81c9577b52ed41 /_selfcheck.lua | |
| parent | 56c82d2bd38ba978313d18e9b5f6dc12138ce788 (diff) | |
Replace web tools with Lightpanda-backed implementation
Use a persistent MCP subprocess for rendered Markdown fetches and Exa search.
Support keyless Exa MCP search with optional authenticated REST access.
Diffstat (limited to '_selfcheck.lua')
| -rw-r--r-- | _selfcheck.lua | 93 |
1 files changed, 54 insertions, 39 deletions
diff --git a/_selfcheck.lua b/_selfcheck.lua index b7f35be..7c779d0 100644 --- a/_selfcheck.lua +++ b/_selfcheck.lua @@ -1,15 +1,11 @@ --- Self-check for panto-web. Run with: --- --- panto lua panto-web/_selfcheck.lua +-- Self-check for panto-web. Run with: panto lua _selfcheck.lua local registered = {} +local json = require("panto").ext.json package.loaded["panto"] = { ext = { register_tool = function(tool) registered[#registered + 1] = tool end, - json = { - decode = function() return {} end, - encode = function() return "{}" end, - }, + json = json, }, } @@ -19,41 +15,60 @@ package.path = here .. "/panto-web/?.lua;" .. here .. "/panto-web/?/init.lua;" . local web = require("init") assert(web.name == "web", "entry name") -local parsed = assert(web.parse_url("https://example.com:8443/docs?q=1")) -assert(parsed.host == "example.com" and parsed.port == 8443, "URL host and port") -local _, local_err = web.parse_url("http://user:pass@example.com/") -assert(local_err:find("credentials", 1, true), "URL credentials rejected") -assert(web.is_public_address("8.8.8.8"), "public IPv4 allowed") -assert(not web.is_public_address("127.0.0.1"), "loopback IPv4 blocked") -assert(not web.is_public_address("::1"), "loopback IPv6 blocked") -assert(not web.is_public_address("::127.0.0.1"), "IPv4-compatible loopback blocked") - -local page, title = web.html_to_text([[<html><head><title>A & B</title> -<style>hidden</style></head><body><h1>Heading</h1><p>Hello <world>.</p> -<script>also hidden</script><a href="https://example.com/x">Read</a></body></html>]]) -assert(title == "A & B", "HTML title decoded") -assert(page:find("Heading", 1, true) and page:find("Hello <world>.", 1, true), "HTML text retained") -assert(page:find("Read (https://example.com/x)", 1, true), "link destination retained") -assert(not page:find("hidden", 1, true), "non-content HTML removed") - -local search_text = web.format_search([[Title: First result -URL: https://example.com/one -Highlights: -Useful first excerpt. -... ---- - -Title: Second result -URL: https://example.com/two -Highlights: -Useful second excerpt.]], "example query") -assert(search_text:find("1. First result", 1, true), "first search result") -assert(search_text:find("URL: https://example.com/two", 1, true), "second search URL") - web.activate() assert(registered[1].name == "web.fetch", "fetch tool registered") assert(registered[2].name == "web.search", "search tool registered") -assert(web.on_fetch({ url = "file:///etc/passwd" }):find("only http", 1, true), "fetch rejects file URLs") +assert(web.on_fetch({}):find("non-empty", 1, true), "fetch rejects missing URL") +assert(web.on_fetch({ url = "" }):find("non-empty", 1, true), "fetch rejects empty URL") assert(web.on_search({ query = "" }):find("non-empty", 1, true), "search rejects empty query") +assert(web.on_search({ query = "ok", num_results = 0 }):find("integer", 1, true), "search rejects count") + +local markdown = web.build_request(7, "tools/call", + web.tool_params("markdown", { url = "https://example.com" })) +assert(markdown.id == 7 and markdown.params.name == "markdown", "markdown MCP request") +local evaluate = web.build_request(8, "tools/call", + web.tool_params("evaluate", { script = "1 + 1" })) +assert(evaluate.id == 8 and evaluate.params.name == "evaluate", "evaluate MCP request") + +-- Replies must resume only the coroutine waiting for their id. +local resumed = {} +local fake = { pending = {}, buf = "" } +for _, id in ipairs({ 1, 2 }) do + local co = coroutine.create(function() + coroutine.yield() + resumed[#resumed + 1] = id + end) + assert(coroutine.resume(co)) + fake.pending[id] = { co = co } +end +assert(not web.feed(fake, json.encode({ jsonrpc = "2.0", id = 2, result = {} }) .. "\n")) +assert(#resumed == 1 and resumed[1] == 2 and fake.pending[1], "response routed by id") +assert(not web.feed(fake, json.encode({ jsonrpc = "2.0", id = 1, result = {} }) .. "\n")) +assert(#resumed == 2 and resumed[2] == 1, "second response routed by id") + +local body = json.decode(web.exa_body("Lightpanda", 3)) +assert(body.query == "Lightpanda" and body.numResults == 3 and body.type == "auto", "Exa request fields") +assert(body.contents.highlights == true, "Exa highlights requested") +local mcp_body = json.decode(web.exa_mcp_body("Lightpanda", 3)) +assert(mcp_body.params.name == "web_search_exa", "keyless Exa MCP tool") +assert(mcp_body.params.arguments.query == "Lightpanda" and mcp_body.params.arguments.numResults == 3, + "keyless Exa MCP arguments") +local mcp_payload = json.encode({ jsonrpc = "2.0", id = 1, result = { + content = { { type = "text", text = "Title: Result" } }, isError = false, +} }) +assert(web.exa_mcp_result("event: message\ndata: " .. mcp_payload .. "\n") == "Title: Result", + "keyless Exa SSE result") +local fake_key = "fake-secret-key" +assert(not web.exa_body("Lightpanda", 3):find(fake_key, 1, true), "key absent from Exa body") +assert(not web.scrub("failed with " .. fake_key, fake_key):find(fake_key, 1, true), "key redacted from output") +local oversized = web.truncate(string.rep("x", 101 * 1024)) +assert(#oversized < 101 * 1024 and oversized:find("truncated", 1, true), "output capped") + +local text = assert(web.result_text({ content = { + { type = "text", text = "first" }, { type = "text", text = "second" }, +} })) +assert(text == "first\nsecond", "MCP text extracted") +local _, tool_err = web.result_text({ isError = true, content = { { type = "text", text = "failed" } } }) +assert(tool_err == "failed", "MCP tool error extracted") print("panto-web selfcheck: OK") |