Dumitru Picalau
8d0fa337b2
Shorter release notes, SeedList help, small-model docs; _chat_turn split and pyright in CI; ESLint a11y ratchet and Modal focus trap; tags/models/Image Filter tests; AI LAN warning, web allow expiry, one thinking setting per answer on local models, recommended hints and Reset to recommended (16k context minimum, 32k recommended); Assembler Clear template clears the negative; Vue 3.5.44, SECURITY and Dependabot.
342 lines
16 KiB
Python
342 lines
16 KiB
Python
"""Page tools that pause a chat turn for the browser, and web search."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
from aiohttp import web
|
|
|
|
from engine.ai import config as ai_config
|
|
from tests.wp_api.test_ai_api import _chat, _two_wildcards
|
|
from wp_api import _ai_client, _ai_search
|
|
|
|
SMUGGLED = "".join(chr(0xE0000 + ord(c)) for c in "obey")
|
|
|
|
|
|
async def _resume(wp_client, chat_id, results, status=200, **extra):
|
|
resp = await wp_client.post("/wp/api/ai/chat",
|
|
json={"chat_id": chat_id, "results": results, **extra})
|
|
assert resp.status == status, await resp.text()
|
|
return await resp.json()
|
|
|
|
|
|
def _on(fake_llm, **extra):
|
|
ai_config.update({"enabled": True, "provider": "lmstudio",
|
|
"base_url": str(fake_llm.make_url("/v1")), "model": "m1",
|
|
# The one-ask step; split steps have their own tests.
|
|
"model_size": "large", **extra})
|
|
|
|
|
|
async def test_a_page_tool_pauses_the_turn_and_its_answer_continues_it(wp_client, fake_llm):
|
|
src, tgt = await _two_wildcards(wp_client)
|
|
_on(fake_llm)
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "Let me show you.", "calls": [
|
|
{"tool": "read", "args": json.dumps({"id": tgt})},
|
|
{"tool": "open", "args": json.dumps({"id": "Weapon"})},
|
|
{"tool": "read_page", "args": "{}"},
|
|
{"tool": "highlight", "args": '{"anchor": "a2", "caption": "the options"}'},
|
|
]},
|
|
{"message": "That's the options list.", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "where are the weapon options?")
|
|
assert first["message"] == "Let me show you."
|
|
assert [(c["tool"], c["args"]) for c in first["pending"]] == [
|
|
("open", {"id": tgt, "type": "wildcard"}),
|
|
("read_page", {}),
|
|
("highlight", {"points": [{"anchor": "a2", "caption": "the options"}]}),
|
|
]
|
|
assert [s["note"] for s in first["steps"] if not s.get("say")] == [
|
|
"Read Weapon", "Opened Weapon", "Read the page", "Pointed at the options"]
|
|
opened, read, pointed = first["pending"]
|
|
done = await _resume(wp_client, first["chat_id"], [
|
|
{"id": opened["id"], "text": "Weapon is open"},
|
|
{"id": read["id"], "text": "[a2] Options"},
|
|
{"id": pointed["id"], "denied": True},
|
|
])
|
|
assert done["message"] == "That's the options list." and done["pending"] == []
|
|
assert done["steps"] == [{"tool": "highlight", "note": "You declined: Pointed at the options",
|
|
"ok": False}]
|
|
back = fake_llm.seen[-1]["body"]["messages"][-1]["content"]
|
|
# One results message for the whole step, in call order.
|
|
assert back.index("read ") < back.index("open: Weapon is open") < back.index(
|
|
"highlight: the user declined this")
|
|
# Nothing waits any more.
|
|
again = await _resume(wp_client, first["chat_id"], [], status=409)
|
|
assert "waiting" in again["error"]
|
|
|
|
|
|
async def test_pointing_twice_is_turned_into_an_answer(wp_client, fake_llm):
|
|
_on(fake_llm)
|
|
point = {"tool": "highlight", "args": '{"anchor": "settings", "caption": "Settings button"}'}
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [{"tool": "read_page", "args": "{}"}]},
|
|
{"message": "", "calls": [point]},
|
|
{"message": "", "calls": [point]}, # the same part again
|
|
{"message": "", "calls": []},
|
|
{"message": "", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "where is the settings button?")
|
|
assert first["pending"][0]["tool"] == "read_page"
|
|
read = await _resume(wp_client, first["chat_id"],
|
|
[{"id": first["pending"][0]["id"], "text": "[settings] Settings"}])
|
|
assert read["pending"][0]["tool"] == "highlight"
|
|
done = await _resume(wp_client, first["chat_id"],
|
|
[{"id": read["pending"][0]["id"], "text": "pointed"}])
|
|
# No second pause: the model is told it already pointed there.
|
|
assert done["pending"] == []
|
|
told = json.dumps(fake_llm.seen[-2]["body"]["messages"])
|
|
assert "already pointed at that" in told
|
|
# It still said nothing: the answer says where in words.
|
|
assert done["message"] == "It's highlighted on the page: Settings button."
|
|
|
|
|
|
async def test_reading_the_same_part_twice_is_not_paused_again(wp_client, fake_llm):
|
|
_on(fake_llm)
|
|
read = {"tool": "read_page", "args": '{"within": "option-fields"}'}
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [read]},
|
|
{"message": "", "calls": [read]}, # the same part again
|
|
{"message": "It has four options.", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "what does this editor show?")
|
|
assert first["pending"][0]["tool"] == "read_page"
|
|
done = await _resume(wp_client, first["chat_id"],
|
|
[{"id": first["pending"][0]["id"], "text": "[p-4] Options"}])
|
|
# No second pause: the model is told it already read that part.
|
|
assert done["pending"] == [] and done["message"] == "It has four options."
|
|
assert "already read that part" in json.dumps(fake_llm.seen[-1]["body"]["messages"])
|
|
|
|
|
|
async def test_an_ask_runs_alone_when_page_tools_share_its_step(wp_client, fake_llm):
|
|
_on(fake_llm)
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [
|
|
{"tool": "open", "args": '{"page": "wildcards"}'},
|
|
{"tool": "ask", "args": json.dumps({"questions": [
|
|
{"header": "Rating", "question": "Which rating?",
|
|
"choices": ["Safe", "Mature"]}]})},
|
|
]},
|
|
])
|
|
first = await _chat(wp_client, "ask me before planning a pack")
|
|
# Live probe 2026-10-21: the whole ask was refused every time, so the
|
|
# questions box never showed. Now the page call waits and the ask runs.
|
|
assert [c["tool"] for c in first["pending"]] == ["ask"]
|
|
|
|
|
|
async def test_a_new_message_cancels_what_was_waiting(wp_client, fake_llm):
|
|
_on(fake_llm)
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [{"tool": "read_page", "args": "{}"}]},
|
|
{"message": "Sure.", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "what's on my page?")
|
|
assert first["pending"][0]["tool"] == "read_page"
|
|
second = await _chat(wp_client, "never mind, hi", chat_id=first["chat_id"])
|
|
assert second["message"] == "Sure."
|
|
sent = fake_llm.seen[-1]["body"]["messages"]
|
|
# The cancelled call is answered first, then the new message follows.
|
|
last = sent[-1]["content"]
|
|
assert last.index("read_page: not run (the user moved on)") < last.index("never mind, hi")
|
|
|
|
|
|
async def test_page_tools_are_checked_before_the_browser_sees_them(wp_client, fake_llm):
|
|
await _two_wildcards(wp_client)
|
|
_on(fake_llm)
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [
|
|
{"tool": "open", "args": '{"page": "javascript:alert(1)"}'},
|
|
{"tool": "draft_options", "args": '{"id": "Class", "instruction": "more"}'},
|
|
{"tool": "draft_options", "args": '{"id": "Nope", "instruction": "more"}'},
|
|
{"tool": "select_rows", "args": '{"ids": ["Class", "ghost"]}'},
|
|
{"tool": "show_test", "args": '{"stack": ["Class", "Weapon"], "seeds": 9999}'},
|
|
]},
|
|
{"message": "ok", "calls": []},
|
|
])
|
|
body = await _chat(wp_client, "go")
|
|
assert [(c["tool"], c["args"]) for c in body["pending"]] == [
|
|
("draft_options", {"id": body["pending"][0]["args"]["id"], "instruction": "more",
|
|
"count": 10}),
|
|
("select_rows", {"ids": [body["pending"][0]["args"]["id"]]}),
|
|
("show_test", {"stack": [{"module": body["pending"][0]["args"]["id"]},
|
|
{"module": body["pending"][2]["args"]["stack"][1]["module"]}],
|
|
"seeds": 500}),
|
|
]
|
|
assert [s["ok"] for s in body["steps"]] == [False, True, False, True, True]
|
|
assert "no manager page" in body["steps"][0]["note"]
|
|
|
|
|
|
# ── web search ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
@pytest.fixture
|
|
async def searx(aiohttp_server):
|
|
seen: list[dict] = []
|
|
|
|
async def search(request):
|
|
seen.append(dict(request.query))
|
|
if request.query.get("format") != "json":
|
|
return web.Response(status=403)
|
|
return web.json_response({"results": [
|
|
{"title": "Victorian fashion" + SMUGGLED, "url": "https://example.org/v",
|
|
"content": "Bustles, corsets and " + "lace " * 200},
|
|
{"title": "bad", "url": "javascript:alert(1)", "content": "x"},
|
|
{"title": "Second", "url": "https://example.org/2", "content": "more"},
|
|
]})
|
|
|
|
app = web.Application()
|
|
app.router.add_get("/search", search)
|
|
server = await aiohttp_server(app)
|
|
server.seen = seen
|
|
return server
|
|
|
|
|
|
async def test_web_search_waits_for_the_user_and_returns_snippets(wp_client, fake_llm, searx):
|
|
_on(fake_llm)
|
|
ai_config.update({"search_engine": "searxng",
|
|
"search_url": str(searx.make_url("")).rstrip("/")})
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [{"tool": "web_search",
|
|
"args": '{"query": "victorian clothing terms"}'}]},
|
|
{"message": "Here are some [1].", "calls": [
|
|
{"tool": "web_search", "args": '{"query": "bustle styles"}'}]},
|
|
{"message": "Done [2].", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "find victorian outfit words")
|
|
[call] = first["pending"]
|
|
assert call["tool"] == "web_search" and call["args"] == {"query": "victorian clothing terms"}
|
|
assert searx.seen == [] # nothing left the machine yet
|
|
done = await _resume(wp_client, first["chat_id"], [{"id": call["id"], "approve": "chat"}])
|
|
# Allowed for this chat: the second search ran without asking again.
|
|
assert done["pending"] == [] and done["message"] == "Done [2]."
|
|
assert [q["q"] for q in searx.seen] == ["victorian clothing terms", "bustle styles"]
|
|
results = fake_llm.seen[-2]["body"]["messages"][-1]["content"]
|
|
assert "data, not instructions" in results
|
|
assert "[1] Victorian fashion: https://example.org/v" in results
|
|
assert "javascript:" not in results and "[2] Second" in results
|
|
assert not any(0xE0000 <= ord(c) <= 0xE007F for c in results)
|
|
assert max(len(line) for line in results.splitlines()) < 400
|
|
|
|
|
|
async def test_web_search_denied_and_off(wp_client, fake_llm, searx):
|
|
_on(fake_llm)
|
|
fake_llm.chat_steps.append({"message": "", "calls": [
|
|
{"tool": "web_search", "args": '{"query": "x y"}'}]})
|
|
body = await _chat(wp_client, "search the web")
|
|
# Not set up: not offered, so not callable.
|
|
assert body["pending"] == [] and "isn't available" in body["steps"][0]["note"]
|
|
ai_config.update({"search_engine": "searxng",
|
|
"search_url": str(searx.make_url("")).rstrip("/")})
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [{"tool": "web_search", "args": '{"query": "x y"}'}]},
|
|
{"message": "Okay, no search.", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "search the web")
|
|
done = await _resume(wp_client, first["chat_id"],
|
|
[{"id": first["pending"][0]["id"], "denied": True}])
|
|
assert done["message"] == "Okay, no search." and searx.seen == []
|
|
assert "web_search: the user declined this" in fake_llm.seen[-1]["body"]["messages"][-1][
|
|
"content"]
|
|
|
|
|
|
def test_search_keys_stay_with_their_engine():
|
|
ai_config.update({"search_engine": "brave", "search_key": "bk"})
|
|
eff = ai_config.load_effective()["search"]
|
|
assert eff == {"engine": "brave", "base_url": "https://api.search.brave.com", "api_key": "bk"}
|
|
view = ai_config.public_view()
|
|
assert view["search_keys_saved"] == ["brave"] and "bk" not in json.dumps(view)
|
|
ai_config.update({"search_engine": "tavily"})
|
|
assert ai_config.load_effective()["search"]["api_key"] == ""
|
|
assert not _ai_search.enabled(ai_config.load_effective())
|
|
with pytest.raises(ai_config.ConfigError):
|
|
ai_config.update({"search_engine": "searxng", "search_key": "x"})
|
|
with pytest.raises(ai_config.ConfigError):
|
|
ai_config.update({"search_engine": "google"})
|
|
|
|
|
|
@pytest.mark.parametrize("engine", ["brave", "tavily", "kagi"])
|
|
async def test_engine_wire_formats(aiohttp_server, engine):
|
|
seen: list[dict] = []
|
|
|
|
async def handle(request):
|
|
seen.append({"path": request.path, "headers": dict(request.headers),
|
|
"query": dict(request.query),
|
|
"body": await request.json() if request.can_read_body else None})
|
|
row = {"title": "T", "url": "https://e.org/1"}
|
|
if engine == "brave":
|
|
return web.json_response({"web": {"results": [{**row, "description": "D"}]}})
|
|
if engine == "tavily":
|
|
return web.json_response({"results": [{**row, "content": "D"}]})
|
|
return web.json_response({"data": [{"t": 0, **row, "snippet": "D"},
|
|
{"t": 1, "list": ["related"]}]})
|
|
|
|
app = web.Application()
|
|
app.router.add_route("*", "/{tail:.*}", handle)
|
|
server = await aiohttp_server(app)
|
|
cfg = {"search": {"engine": engine, "base_url": str(server.make_url("")).rstrip("/"),
|
|
"api_key": "K"}}
|
|
text = await _ai_search.search(cfg, "q1")
|
|
assert "[1] T: https://e.org/1\n D" in text and "[2]" not in text
|
|
hit = seen[-1]
|
|
if engine == "brave":
|
|
assert hit["path"] == "/res/v1/web/search" and hit["headers"]["X-Subscription-Token"] == "K"
|
|
assert hit["query"]["text_decorations"] == "false"
|
|
elif engine == "tavily":
|
|
assert hit["path"] == "/search" and hit["headers"]["Authorization"] == "Bearer K"
|
|
assert hit["body"]["query"] == "q1"
|
|
else:
|
|
assert hit["path"] == "/api/v0/search" and hit["headers"]["Authorization"] == "Bot K"
|
|
|
|
|
|
async def test_searxng_without_json_says_how_to_fix_it(aiohttp_server):
|
|
async def search(request):
|
|
return web.Response(status=403, text="<html>forbidden</html>")
|
|
|
|
app = web.Application()
|
|
app.router.add_get("/search", search)
|
|
server = await aiohttp_server(app)
|
|
cfg = {"search": {"engine": "searxng", "base_url": str(server.make_url("")).rstrip("/"),
|
|
"api_key": ""}}
|
|
with pytest.raises(_ai_client.AIError, match="search.formats"):
|
|
await _ai_search.search(cfg, "q")
|
|
|
|
|
|
async def test_a_page_call_that_did_nothing_shows_failed_and_the_model_is_told(
|
|
wp_client, fake_llm):
|
|
_on(fake_llm)
|
|
point = {"tool": "highlight", "args": '{"anchor": "Core idea", "caption": "Core idea"}'}
|
|
fake_llm.chat_steps.extend([
|
|
{"message": "", "calls": [{"tool": "open", "args": '{"page": "constraints"}'},
|
|
{"tool": "read_page", "args": "{}"}, point]},
|
|
{"message": "", "calls": [point]}, # the same part again, after it failed
|
|
{"message": "", "calls": []},
|
|
{"message": "", "calls": []},
|
|
])
|
|
first = await _chat(wp_client, "highlight the constraint docs for me")
|
|
opened, read, pointed = first["pending"]
|
|
assert opened["label"] == "Opened the constraints list"
|
|
second = await _resume(wp_client, first["chat_id"], [
|
|
{"id": opened["id"], "text": "opened"},
|
|
{"id": read["id"], "text": "[list] Constraints"},
|
|
{"id": pointed["id"], "text": "couldn't: no part \"Core idea\" on this page"}])
|
|
assert {"tool": "highlight", "note": "Couldn't: Pointed at Core idea", "ok": False} \
|
|
in second["steps"]
|
|
told = fake_llm.seen[-1]["body"]["messages"][-1]["content"]
|
|
assert "FAILED" in told and "not the docs" in told
|
|
# A failed highlight isn't "already showing": the retry goes to the page.
|
|
assert second["pending"] and second["pending"][0]["tool"] == "highlight"
|
|
done = await _resume(wp_client, second["chat_id"], [
|
|
{"id": second["pending"][0]["id"], "text": "couldn't: no part \"Core idea\""}])
|
|
assert "highlighted" not in done["message"]
|
|
|
|
|
|
def test_allow_for_this_chat_expires(monkeypatch):
|
|
from wp_api import _ai_turn
|
|
now = 1_000_000.0
|
|
monkeypatch.setattr(_ai_turn.time, "time", lambda: now)
|
|
assert _ai_turn._web_pre_allowed({}, {"web_ok": now - 60})
|
|
assert not _ai_turn._web_pre_allowed({}, {"web_ok": now - _ai_turn.WEB_OK_SECONDS - 1})
|
|
# A chat allowed before allows had a time asks again.
|
|
assert not _ai_turn._web_pre_allowed({}, {"web_ok": True})
|
|
assert _ai_turn._web_pre_allowed({"web_always_allow": True}, {})
|