Remote Control Qutebrowser

early = await share()
missing = await early.call("readonly_read_dom")
assert missing.isError and "qutebrowser is not running" in missing.content[0].text, (
    f"the agent was not told qutebrowser is not running: {missing.content[0].text}"
)

open_page("payment.html")

agent = await share()
claimed = await agent.call("claim_shown_tab")
assert not claimed.isError, f"the agent could not take the tab I show: {claimed.content[0].text}"

dom = (await agent.call("readonly_read_dom")).content[0].text
assert "Total: 42.00 €" in dom, f"the payment page the agent reads lacks its total: {dom}"

with pytest.raises(Refused):
    await agent.call("click", {"selector": "button", "why": "pay"})
assert not pages.paid.is_set(), "« Pay » was clicked"

open_page("aws.html")
keep_clock(runtime)

claimed = await agent.call("claim_shown_tab")
assert not claimed.isError, f"the agent could not take the AWS tab: {claimed.content[0].text}"

agent.grant_all()

field = (await agent.call("readonly_read_dom", {"selector": "[name=region]"})).content[0].text
assert "Save" not in field, f"the agent read more than the field it named: {field}"
region = re.search(r'value="([^"]*)"', field)
assert region and region[1] == "us-east-1", f"the region the agent reads is not us-east-1: {field}"

blind = await agent.call("readonly_screenshot", {"why": " "})
assert blind.isError, "the agent took a screenshot without saying why"
shot = (await agent.call("readonly_screenshot", {"why": "see the page's orange band"})).content[0]
image = PIL.Image.open(io.BytesIO(base64.b64decode(shot.data))).convert("RGB")
assert AWS_ORANGE in image.getdata(), "the screenshot the agent sees lacks the page's orange band"

blind = await agent.call("fill", {"selector": "[name=region]", "text": "eu-west-3", "why": " "})
assert blind.isError, "the agent filled unasked without saying why"
_, seen = await watch(agent, "fill", {"selector": "[name=region]", "text": "eu-west-3", "why": "you said fill it unasked"})

assert any(FILL_CYAN in shot.getdata() for _, shot in seen), "the field did not stand out in the fill's colour"

assert lead(seen, FILL_CYAN) >= STAND_OUT_LEAD, "the field did not stand out long enough before it was filled"

assert FILL_CYAN in (await screen(agent)).getdata(), "the field no longer stood out once filled"

logo = tmp_path / "logo.png"
logo.write_bytes(b"\x89PNG\r\n\x1a\n")
uploaded, seen = await watch(agent, "upload", {"selector": "#upload-logo", "path": str(logo), "why": "you said upload it unasked"})
done = landed_at(seen, UPLOAD_YELLOW)
assert not uploaded.isError, f"the agent was told the upload failed: {uploaded.content[0].text}"
dom = (await agent.call("readonly_read_dom")).content[0].text
chosen = re.search(r'<output id="logo">([^<]*)</output>', dom)
assert chosen and chosen[1] == "logo.png", f"the AWS page got {chosen and chosen[1]!r} as logo, not my logo.png"

assert lead(seen, UPLOAD_YELLOW) >= STAND_OUT_LEAD, "« Upload » did not stand out long enough before the upload"

assert UPLOAD_YELLOW in (await screen(agent)).getdata(), "« Upload » no longer stood out once uploaded"

assert FILL_CYAN not in (await screen(agent)).getdata(), "the region field still stood out after the upload"

assert UPLOAD_YELLOW in (await screen_at(agent, runtime, done, STILL_SEEN)).getdata(), "« Upload » no longer stood out 8 seconds on"

assert UPLOAD_YELLOW not in (await screen_at(agent, runtime, done, GONE_BY)).getdata(), "« Upload » still stood out 12 seconds on"

uploaded = await agent.call("upload", {"selector": '[aria-label="Choose file"]', "path": str(logo), "why": "you said upload it unasked"})
assert not uploaded.isError, f"the agent was told the upload failed: {uploaded.content[0].text}"
dom = (await agent.call("readonly_read_dom")).content[0].text
chosen = re.search(r'<output id="product-logo">([^<]*)</output>', dom)
assert chosen and chosen[1] == "logo.png", f"the AWS page got {chosen and chosen[1]!r} as product logo, not my logo.png"

await agent.call("click", {"selector": "#categories", "why": "opening a list changes nothing"})
categories = (await agent.call("readonly_read_dom", {"selector": "#categories"})).content[0].text
assert 'aria-expanded="true"' in categories, f"the list of categories stayed shut: {categories}"

assert UPLOAD_YELLOW not in (await screen(agent)).getdata(), "« Choose file » still stood out once the categories opened"

terminal = subprocess.Popen(["xterm", "-title", "my-work"])
try:
    mine = subprocess.run(["xdotool", "search", "--sync", "--name", "^my-work$"],
                          capture_output=True, text=True, check=True).stdout.split()[0]
    subprocess.run(["xdotool", "windowactivate", "--sync", mine], check=True)
    assert focused_window() == mine, "I could not get into my terminal"
    await agent.call("click", {"selector": "#owner", "why": "opening a list changes nothing"})
    owner = (await agent.call("readonly_read_dom", {"selector": "#owner"})).content[0].text
    assert 'aria-expanded="true"' in owner, f"the list of owners stayed shut while I worked elsewhere: {owner}"
finally:
    terminal.terminate()

await agent.call("click", {"selector": '[data-value="keyban"]', "why": "you said pick keyban unasked"})
picked = (await agent.call("readonly_read_dom", {"selector": "#owner-picked", "text": True})).content[0].text
assert picked == "Keyban", f"the owner the page took is {picked!r}, not Keyban"

await agent.call("click", {"selector": "#close-notice", "why": "closing a notice changes nothing"})
assert HIGHLIGHT_PURPLE not in (await screen(agent)).getdata(), "« Close » still stood out once the notice was gone"

names = (await agent.call("readonly_read_dom", {"selector": "#category-list [role=option]", "text": True})).content[0].text
assert names == "AI Agents & Tools\nFinance & Accounting", f"the agent read more, or less, than the names: {names}"

headers = (await agent.call("readonly_read_dom", {"selector": "h2", "text": True, "shown": True})).content[0].text
assert headers == "Region settings", f"the agent read other headers than those I see: {headers}"

await agent.call("click", {"selector": "#tier", "why": "opening a list changes nothing"})

selected, seen = await watch(agent, "select", {"selector": "#zone", "option": "b", "why": "you said pick it unasked"})
done = landed_at(seen, SELECT_GREEN)
assert not selected.isError, f"the agent could not pick b: {selected.content[0].text}"
shown = (await agent.call("readonly_read_dom", {"selector": "#zone-picked", "text": True})).content[0].text
assert shown == "b", f"the page took {shown!r}, not b"

assert HIGHLIGHT_PURPLE not in (await screen(agent)).getdata(), "« Choose a tier » still stood out once the zone was picked"

assert lead(seen, SELECT_GREEN) >= STAND_OUT_LEAD, "the zone did not stand out long enough before its pick"

assert SELECT_GREEN in (await screen(agent)).getdata(), "the zone no longer stood out once picked"

assert SELECT_GREEN in (await screen_at(agent, runtime, done, STILL_SEEN)).getdata(), "the zone no longer stood out 8 seconds on"

assert SELECT_GREEN not in (await screen_at(agent, runtime, done, GONE_BY)).getdata(), "the zone still stood out 12 seconds on"

watched, after = {}, {}
for list_, option, picked, name in [
    ("#tier", "premium", "#tier-picked", "Premium"),
    ("#request-changes", "remove a version", "#requested", "Remove a version"),
]:
    selected, watched[list_] = await watch(agent, "select", {"selector": list_, "option": option, "why": "you said pick them unasked"})
    after[list_] = await screen(agent)
    assert not selected.isError, f"the agent could not pick {option}: {selected.content[0].text}"
    shown = (await agent.call("readonly_read_dom", {"selector": picked, "text": True})).content[0].text
    assert shown == name, f"the page took {shown!r}, not {name}"

for list_, seen in watched.items():
    assert lead(seen, SELECT_GREEN) >= STAND_OUT_LEAD, f"{list_} did not stand out long enough before its pick"

for list_, look in after.items():
    assert SELECT_GREEN in look.getdata(), f"{list_} no longer stood out once picked"

hidden = await agent.call("click", {"selector": "[name=logo]", "why": "you said click it unasked"})
assert hidden.isError and "not shown" in hidden.content[0].text, (
    f"the agent was not told the element is not shown: {hidden.content[0].text}"
)

collapsed = await agent.call("click", {"selector": "#iam", "why": "you said click it unasked"})
opener = '[aria-controls~="auth-body"]'
assert collapsed.isError and opener in collapsed.content[0].text, (
    f"the agent was not told what opens the section: {collapsed.content[0].text}"
)
await agent.call("click", {"selector": opener, "why": "unfolding a part changes nothing"})
opened = await agent.call("click", {"selector": "#iam", "why": "you said click it unasked"})
assert not opened.isError, f"the agent could not pick IAM once the section opened: {opened.content[0].text}"

assert SELECT_GREEN not in (await screen(agent)).getdata(), "« Request changes » still stood out once IAM was picked"

ticked = await agent.call("select", {"selector": "#networks", "option": "BaseSepolia", "why": "you said tick it unasked"})
assert not ticked.isError, f"the agent could not tick BaseSepolia: {ticked.content[0].text}"
item = (await agent.call("readonly_read_dom", {"selector": "#base-sepolia"})).content[0].text
assert 'aria-checked="true"' in item, f"BaseSepolia is not ticked: {item}"

followed = await agent.call("click", {"selector": 'a[target="_blank" href="#/console?account_id=42&role_name=AdministratorAccess"]', "why": "following a link changes nothing"})
assert not followed.isError, f"the agent could not follow the link: {followed.content[0].text}"
route = (await agent.call("readonly_read_dom", {"selector": "#route", "text": True})).content[0].text
assert route == "#/console?account_id=42&role_name=AdministratorAccess", f"the link led elsewhere: {route!r}"

wrong = await agent.call("click", {"selector": 'button:has-text("Save")', "why": "you said save unasked"})
assert wrong.isError and "not a valid selector" in wrong.content[0].text, (
    f"the agent was not told its selector is wrong: {wrong.content[0].text}"
)

_, seen = await watch(agent, "click", {"selector": "button", "why": "you said click it unasked"})
done = landed_at(seen, HIGHLIGHT_PURPLE)
dom = (await agent.call("readonly_read_dom")).content[0].text
saved = re.search(r'<output id="saved">([^<]*)</output>', dom)
assert saved and saved[1] == "eu-west-3", f"the AWS page saved {saved and saved[1]!r}, not what the agent filled"

assert any(HIGHLIGHT_PURPLE in shot.getdata() for _, shot in seen), "« Save » did not stand out as it was clicked"

assert lead(seen, HIGHLIGHT_PURPLE) >= STAND_OUT_LEAD, "« Save » did not stand out long enough before the click"

assert HIGHLIGHT_PURPLE in (await screen(agent)).getdata(), "« Save » no longer stood out once clicked"

assert HIGHLIGHT_PURPLE in (await screen_at(agent, runtime, done, STILL_SEEN)).getdata(), "« Save » no longer stood out 8 seconds on"

assert HIGHLIGHT_PURPLE not in (await screen_at(agent, runtime, done, GONE_BY)).getdata(), "« Save » still stood out 12 seconds on"

agent.granted = (agent.granted - {"fill"}) | {"confirm_fill"}
with pytest.raises(Refused):
    await agent.call("fill", {"selector": "[name=region]", "text": "eu-west-1", "why": "you said fill it unasked"})

async with anyio.create_task_group() as withdrawn:
    withdrawn.start_soon(agent.call, "confirm_fill", {"selector": "[name=region]", "text": "eu-west-2"})
    await question_seen(agent)
    withdrawn.cancel_scope.cancel()

questions = []

async def count_questions():
    while not any("eu-west-1" in question for question in questions):
        read = (await agent.call("readonly_read_dom", {"selector": "[data-qutebrowser-mcp-question]", "text": True})).content[0].text
        questions[:] = [line for line in read.splitlines() if line]

refused = await asked(agent, runtime, "confirm_fill", {"selector": "[name=region]", "text": "eu-west-1"}, "no", count_questions)
assert refused.isError and "refused" in refused.content[0].text, f"the agent was not told I refused: {refused.content[0].text}"
field = (await agent.call("readonly_read_dom", {"selector": "[name=region]"})).content[0].text
assert 'value="eu-west-3"' in field, f"the field changed though I refused: {field}"

assert len(questions) == 1, f"the page asked me {len(questions)} questions at once: {questions}"

accepted = await asked(agent, runtime, "confirm_fill", {"selector": "[name=region]", "text": "eu-west-1"}, "yes")
done = time.monotonic()
assert not accepted.isError, f"the fill failed though I accepted: {accepted.content[0].text}"
field = (await agent.call("readonly_read_dom", {"selector": "[name=region]"})).content[0].text
assert 'value="eu-west-1"' in field, f"the field did not take what I accepted: {field}"

assert FILL_CYAN in (await screen_at(agent, runtime, done, STILL_SEEN)).getdata(), "the field no longer stood out 8 seconds on"

assert FILL_CYAN not in (await screen_at(agent, runtime, done, GONE_BY)).getdata(), "the field still stood out 12 seconds on"

shown, screens = await watch(agent, "readonly_show", {"selector": "#scrolled"})
assert not shown.isError, f"the agent could not show me the counter: {shown.content[0].text}"
assert any(SHOW_BLUE in shot.getdata() for _, shot in screens), "I never saw the counter framed in blue"
scrolled = (await agent.call("readonly_read_dom", {"selector": "#scrolled", "text": True})).content[0].text
assert scrolled != "0", "the page did not scroll to the counter"

titled = []

async def look_at_title():
    page = (await agent.session.call_tool("readonly_read_dom", {"selector": "title", "text": True})).content[0].text
    titled.append((page.startswith("🤖"), window_named("🤖")))

await asked(agent, runtime, "confirm_fill", {"selector": "[name=region]", "text": "eu-west-1"}, "yes", look_at_title)
assert titled == [(True, True)], f"the page's and window's titles did not tell me the agent was acting: {titled}"
assert window_named("🤖"), "the window's title no longer says the tab is the agent's, once it is done"

async def scroll_away():
    tell_qutebrowser(runtime, ":scroll-to-perc 100")
    await anyio.sleep(ANSWER_POLL)

back = await asked(agent, runtime, "confirm_fill", {"selector": "[name=region]", "text": "eu-west-2"}, "sy", scroll_away)
assert not back.isError, f"the fill failed though I accepted: {back.content[0].text}"
scrolled = (await agent.call("readonly_read_dom", {"selector": "#scrolled", "text": True})).content[0].text
assert scrolled == "0", f"s did not bring the region field back into view: scrolled by {scrolled}"

await agent.call("click", {"selector": "#request-changes", "why": "you said click it unasked"})
picked = await asked(agent, runtime, "confirm_click", {"selector": "#add-version"}, "yes")
assert not picked.isError, f"the pick failed though I accepted: {picked.content[0].text}"
requested = (await agent.call("readonly_read_dom", {"selector": "#requested", "text": True})).content[0].text
assert requested == "Add new version", f"the menu did not take the pick: {requested!r}"

trusting = await asked(agent, runtime, "confirm_fill", {"selector": "[name=variables]", "text": "REGION=eu-west-1"}, "t+y")
assert not trusting.isError, f"the fill I trusted failed: {trusting.content[0].text}"

shot = (await agent.call("readonly_screenshot", {"why": "see whether the trusted part stays framed"})).content[0]
image = PIL.Image.open(io.BytesIO(base64.b64decode(shot.data))).convert("RGB")
assert TRUSTED_TEAL in image.getdata(), "the part I trusted no longer shows as trusted"

with anyio.fail_after(UNASKED_WAIT):
    unasked = await agent.call("confirm_fill", {"selector": "[name=region]", "text": "eu-south-1"})
assert not unasked.isError, f"the fill within the trusted part failed: {unasked.content[0].text}"

outside = await asked(agent, runtime, "confirm_click", {"selector": "#request-changes"}, "no")
assert outside.isError and "refused" in outside.content[0].text, f"a click outside went unasked: {outside.content[0].text}"

tell_qutebrowser(runtime, ":click-element --select-first css '[data-qutebrowser-mcp-trust=edit]'")
await anyio.sleep(ANSWER_POLL)
for key in "-y":
    tell_qutebrowser(runtime, f":fake-key {key}")
    await anyio.sleep(ANSWER_POLL)
await seen_trusted(agent, "textarea")
narrowed = await asked(agent, runtime, "confirm_fill", {"selector": "[name=region]", "text": "eu-north-1"}, "no")
assert narrowed.isError and "refused" in narrowed.content[0].text, f"the region went unasked: {narrowed.content[0].text}"

tell_qutebrowser(runtime, ":click-element --select-first css '[data-qutebrowser-mcp-trust=edit]'")
await anyio.sleep(ANSWER_POLL)
for key in "<<<y":
    tell_qutebrowser(runtime, f":fake-key {key}")
    await anyio.sleep(ANSWER_POLL)
await seen_trusted(agent, "label")
kept = (await agent.call("readonly_read_dom", {"selector": "[data-qutebrowser-mcp-trust=edit]"})).content[0].text
assert kept, "y did not keep the trust I stretched"
with anyio.fail_after(UNASKED_WAIT):
    stretched = await agent.call("confirm_fill", {"selector": "[name=region]", "text": "eu-west-3"})
assert not stretched.isError, f"the fill in the stretched trust failed: {stretched.content[0].text}"
after = await asked(agent, runtime, "confirm_click", {"selector": "#upload-logo"}, "no")
assert after.isError and "refused" in after.content[0].text, f"the Upload button went unasked: {after.content[0].text}"

tell_qutebrowser(runtime, ":click-element --select-first css '[data-qutebrowser-mcp-trust=edit]'")
await anyio.sleep(ANSWER_POLL)
tell_qutebrowser(runtime, ":click-element --select-first css '[name=region]'")
await seen_trusted(agent, "input")
wander()
await anyio.sleep(ANSWER_POLL)
trusted = (await agent.call("readonly_read_dom", {"selector": "[data-qutebrowser-mcp-trusted]"})).content[0].text
assert trusted.startswith("<input"), f"the trust followed my mouse: {trusted}"
tell_qutebrowser(runtime, ":click-element --select-first css '[data-qutebrowser-mcp-trust=keep]'")
await anyio.sleep(ANSWER_POLL)
with anyio.fail_after(UNASKED_WAIT):
    moved = await agent.call("confirm_fill", {"selector": "[name=region]", "text": "eu-south-2"})
assert not moved.isError, f"the fill in the part I moved the trust to failed: {moved.content[0].text}"

tell_qutebrowser(runtime, ":click-element --select-first css '[data-qutebrowser-mcp-trust=drop]'")
await seen_trusted(agent, "")
dropped = await asked(agent, runtime, "confirm_fill", {"selector": "[name=variables]", "text": "REGION=eu-north-1"}, "no")
assert dropped.isError and "refused" in dropped.content[0].text, f"the variables went unasked: {dropped.content[0].text}"

site = f"http://127.0.0.1:{pages.server_port}"
opened = (await agent.call("open_url", {"url": f"{site}/docs.html?from=console&account=42#runtime"})).content[0].text
title = (await agent.call("readonly_read_dom", {"selector": "h1", "text": True})).content[0].text
assert title == "AgentCore documentation", f"the agent did not take me to the documentation: {title}"

button = (await agent.call("readonly_read_dom", {"selector": "#feedback", "text": True})).content[0].text
assert button == "Send feedback", f"the agent read no button right after opening the page: {button!r}"

assert "not settled" not in opened, f"the agent was told the page never settled: {opened}"

await agent.call("open_url", {"url": f"{site}/sign-in.html"})
waited = await agent.call("readonly_wait_for", {"selector": "#feedback"})
assert not waited.isError, f"the agent lost the button across the move: {waited.content[0].text}"

await agent.call("open_url_in_new_tab", {"url": f"{site}/payment.html"})
total = (await agent.call("readonly_read_dom", {"selector": "#total", "text": True})).content[0].text
assert total == "Total: 42.00 €", f"the new tab does not show the payment page: {total}"
tell_qutebrowser(runtime, ":tab-prev")
total = (await agent.call("readonly_read_dom", {"selector": "#total", "text": True})).content[0].text
assert total == "Total: 42.00 €", f"the agent left its tab for mine: {total}"
title = (await agent.call("readonly_read_dom", {"selector": "title", "text": True})).content[0].text
assert title.startswith("🤖"), f"the agent's tab does not tell me it is the agent's: {title}"
terminal = subprocess.Popen(["xterm", "-title", "my-work"])
try:
    mine = subprocess.run(["xdotool", "search", "--sync", "--name", "^my-work$"],
                          capture_output=True, text=True, check=True).stdout.split()[0]
    subprocess.run(["xdotool", "mousemove", "--window", mine, "10", "10"], check=True)
    subprocess.run(["xdotool", "windowactivate", "--sync", mine], check=True)
    await anyio.sleep(ANSWER_POLL)
    assert focused_window() == mine, "I could not get into my terminal"
    await agent.call("readonly_read_dom", {"selector": "#total", "text": True})
    assert focused_window() == mine, "the agent's call took me out of the window I was in"
finally:
    terminal.terminate()

await agent.call("open_url", {"url": f"{site}/visits.html"})
before = (await agent.call("readonly_read_dom", {"selector": "h1", "text": True})).content[0].text
reloaded = await agent.call("reload", {"bypass_cache": True})
assert not reloaded.isError, f"the reload failed: {reloaded.content[0].text}"
after = (await agent.call("readonly_read_dom", {"selector": "h1", "text": True})).content[0].text
assert after != before, f"the page read after the reload is the one before: {after}"

opened = await agent.call("open_url_in_new_tab", {"url": f"{site}/docs.html"})
assert not opened.isError, f"the agent could not open the documentation in a new tab: {opened.content[0].text}"
title = (await agent.call("readonly_read_dom", {"selector": "title", "text": True})).content[0].text
assert title.startswith("🤖"), f"the agent's new tab does not tell me it is the agent's: {title}"
tell_qutebrowser(runtime, ":tab-prev")
former = await shown_title(runtime, tmp_path)
assert not former.startswith("🤖"), f"the agent's former tab still tells me it is the agent's: {former}"

tell_qutebrowser(runtime, f":open --tab {site}/configure")

await agent.call("open_url", {"url": f"{site}/aws.html"})
await agent.call("click", {"selector": "#launch", "why": "you said click it unasked"})
tabs = json.loads((await agent.call("readonly_list_tabs")).content[0].text)
agents = next(index for index, tab in enumerate(tabs) if tab["agent's"])
after = tabs[agents + 1]["name"]
assert after.startswith("Configure software ") and after.endswith("/2"), f"the tabs read: {tabs}"
followed = await agent.call("use_tab", {"name": after})
assert not followed.isError, f"the agent could not take {after}: {followed.content[0].text}"
heading = (await agent.call("readonly_read_dom", {"selector": "h1", "text": True})).content[0].text
assert heading == "Configuration 2", f"the agent did not follow the configuration it opened: {heading!r}"

def runtime():
    return pathlib.Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}")) / "qutebrowser"


AGENT_TAB = {}


OPEN_LINGER = 0.5  # seconds


def send(command, linger=0, anywhere=False, answer=None):
    if AGENT_TAB and not anywhere:
        told = f"--answer {answer} " if answer else ""
        command = f":agent-run {told}{command.removeprefix(':')}"
    name = hashlib.md5(getpass.getuser().encode()).hexdigest()
    path = runtime() / f"ipc-{name}"
    with socket.socket(socket.AF_UNIX) as ipc:
        try:
            ipc.connect(str(path))
        except (FileNotFoundError, ConnectionRefusedError):
            raise RuntimeError(f"qutebrowser is not running: nothing listens on {path}") from None
        ipc.sendall((json.dumps({"args": [command], "target_arg": None, "protocol_version": 1}) + "\n").encode())
        time.sleep(linger)

import json
import pathlib

from qutebrowser.api import cmdutils, interceptor
from qutebrowser.commands import runners
from qutebrowser.completion.models import miscmodels
from qutebrowser.misc import objects
from qutebrowser.qt import sip
from qutebrowser.qt.core import QUrl
from qutebrowser.utils import objreg, usertypes

for name in ["agent-claim", "agent-former-eval", "agent-run", "agent-eval", "agent-requests", "agent-tabs", "agent-take"]:
    objects.commands.pop(name, None)

AGENT_TAB = {}


@cmdutils.register()
@cmdutils.argument("win_id", value=cmdutils.Value.win_id)
def agent_claim(win_id: int) -> None:
    """Make the tab this window shows the agent's, keeping the former one to take its 🤖 off."""
    tabbed = objreg.get("tabbed-browser", scope="window", window=win_id)
    AGENT_TAB.update(former=AGENT_TAB.get("tab"), win_id=win_id, tab=tabbed.widget.currentWidget())


@cmdutils.register()
def agent_former_eval(script: str, answer: str) -> None:
    """Run the javascript in SCRIPT in the agent's former tab, where it is still there and not its tab now.

    Write what it gives back to ANSWER, as JSON, null when there is no such tab."""
    def answered(value):
        part = pathlib.Path(f"{answer}.part")
        part.write_text(json.dumps(value))
        part.rename(answer)

    former = AGENT_TAB.get("former")
    if former is None or sip.isdeleted(former) or former is AGENT_TAB.get("tab"):
        answered(None)
        return
    former.run_js_async(pathlib.Path(script).read_text(), answered, world=usertypes.JsWorld.jseval)


def agent_tab():
    """The window and tab of the agent: the one pointed at, which bears the 🤖."""
    if not AGENT_TAB or AGENT_TAB["win_id"] not in objreg.window_registry or sip.isdeleted(AGENT_TAB["tab"]):
        raise cmdutils.CommandError("the agent has no tab of its own: open one with open_url_in_new_tab")
    tabbed = objreg.get("tabbed-browser", scope="window", window=AGENT_TAB["win_id"])
    if tabbed.widget.indexOf(AGENT_TAB["tab"]) < 0:
        raise cmdutils.CommandError("the agent's tab is gone: open one with open_url_in_new_tab")
    return tabbed, AGENT_TAB["win_id"], AGENT_TAB["tab"]


@cmdutils.register()
@cmdutils.argument("win_id", value=cmdutils.Value.win_id)
def agent_eval(script: str, answer: str, win_id: int) -> None:
    """Run the javascript in SCRIPT in the tab this window shows, writing what it gives back to ANSWER, as JSON."""
    tab = objreg.get("tabbed-browser", scope="window", window=win_id).widget.currentWidget()

    def answered(value):
        part = pathlib.Path(f"{answer}.part")
        part.write_text(json.dumps(value))
        part.rename(answer)

    tab.run_js_async(pathlib.Path(script).read_text(), answered, world=usertypes.JsWorld.jseval)


def page_of(url):
    return url.adjusted(QUrl.UrlFormattingOption.RemoveFragment).toString(QUrl.ComponentFormattingOption.FullyEncoded)


REQUESTED = interceptor.__dict__.setdefault("agent_requested", {})  # page → what it asked for, kept across re-sourcing


AWAITED = {interceptor.ResourceType.stylesheet, interceptor.ResourceType.script, interceptor.ResourceType.xhr}


def counted(request):
    page = page_of(request.first_party_url)
    if request.resource_type == interceptor.ResourceType.main_frame:
        REQUESTED[page] = []
    elif request.resource_type in AWAITED and not request.is_blocked:
        REQUESTED.setdefault(page, []).append(request.request_url.toString(QUrl.ComponentFormattingOption.FullyEncoded))


if not interceptor.__dict__.setdefault("agent_counting", False):
    interceptor.register(lambda request: counted(request))
    interceptor.agent_counting = True


@cmdutils.register()
@cmdutils.argument("win_id", value=cmdutils.Value.win_id)
def agent_requests(answer: str, win_id: int) -> None:
    """Write to ANSWER, as JSON, what the page this window shows asked for since it began loading."""
    tab = objreg.get("tabbed-browser", scope="window", window=win_id).widget.currentWidget()
    part = pathlib.Path(f"{answer}.part")
    part.write_text(json.dumps(REQUESTED.get(page_of(tab.url()), [])))
    part.rename(answer)


@cmdutils.register()
def agent_tabs(answer: str) -> None:
    """Write to ANSWER, as JSON, every tab: as WINDOW/INDEX, its title, its address, and whether it is the agent's."""
    tabs = []
    for win_id in list(objreg.window_registry):
        tabbed = objreg.get("tabbed-browser", scope="window", window=win_id)
        for index in range(tabbed.widget.count()):
            tab = tabbed.widget.widget(index)
            tabs.append({"tab": f"{win_id}/{index + 1}", "title": tab.title(), "url": tab.url().toDisplayString(),
                         "agent's": tab is AGENT_TAB.get("tab")})
    part = pathlib.Path(f"{answer}.part")
    part.write_text(json.dumps(tabs, ensure_ascii=False))
    part.rename(answer)


@cmdutils.register()
def agent_take(tab: str) -> None:
    """Make the tab TAB, as WINDOW/INDEX, the agent's."""
    win_id, index = (int(part) for part in tab.split("/"))
    if win_id not in objreg.window_registry:
        raise cmdutils.CommandError(f"no window {win_id}")
    widget = objreg.get("tabbed-browser", scope="window", window=win_id).widget.widget(index - 1)
    if widget is None:
        raise cmdutils.CommandError(f"no tab {tab}")
    AGENT_TAB.update(win_id=win_id, tab=widget)


@cmdutils.register(maxsplit=0)
def agent_run(command: str, *, answer: str = None) -> None:
    """Run COMMAND in the agent's tab, made current in its window, raising no window.

    When the agent has no tab, write why to ANSWER, as JSON, for whoever waits on it."""
    try:
        tabbed, win_id, tab = agent_tab()
    except cmdutils.CommandError as error:
        if answer:
            part = pathlib.Path(f"{answer}.part")
            part.write_text(json.dumps({"agent_run_failed": str(error)}))
            part.rename(answer)
        raise
    tabbed.widget.setCurrentWidget(tab)
    runners.CommandRunner(win_id).run_safely(command)

HAND_OVER_WAIT = 10  # seconds
HAND_OVER_POLL = 0.1  # seconds


def wait_for(path, whole=lambda data: True, wait=HAND_OVER_WAIT):
    deadline = time.monotonic() + wait
    while not (path.exists() and whole(path.read_bytes())):
        if time.monotonic() > deadline:
            raise TimeoutError(f"qutebrowser never handed {path.name} over")
        time.sleep(HAND_OVER_POLL)


def answered(path):
    """What qutebrowser wrote to PATH, or why it could not reach the agent's tab, raised at once."""
    value = json.loads(path.read_text())
    if isinstance(value, dict) and "agent_run_failed" in value:
        raise RuntimeError(value["agent_run_failed"])
    return value


def page_read(wait=HAND_OVER_WAIT):
    """The page's HTML, and the calls it ended, each with why it failed, empty when it did not."""
    deadline = time.monotonic() + wait
    with tempfile.TemporaryDirectory() as tmp:
        dom, script = pathlib.Path(tmp) / "dom.json", pathlib.Path(tmp) / "dom.js"
        script.write_text("[document.documentElement.outerHTML, (window.qutebrowserMcpState || {}).done || {}]")
        while True:
            dom.unlink(missing_ok=True)
            send(f":agent-eval {script} {dom}", answer=dom)
            wait_for(dom, wait=max(deadline - time.monotonic(), HAND_OVER_POLL))
            read = answered(dom)
            if read is not None:
                return read
            if time.monotonic() > deadline:
                raise TimeoutError("the page never let a script read it")
            time.sleep(HAND_OVER_POLL)


AGENT_MARK = "🤖"


def shown_title(wait=HAND_OVER_WAIT):
    deadline = time.monotonic() + wait
    while True:
        with tempfile.TemporaryDirectory() as tmp:
            shown, script = pathlib.Path(tmp) / "title", pathlib.Path(tmp) / "hand-over"
            script.write_text('#!/bin/sh\necho "$QUTE_TITLE" > "$1.part" && mv "$1.part" "$1"\n')
            script.chmod(0o755)
            send(f":spawn --userscript {script} {shown}")
            try:
                wait_for(shown, wait=CHECK_WAIT)
                return shown.read_text().strip()
            except TimeoutError:
                if time.monotonic() > deadline:
                    raise


ACTING = {"calls": 0, "lock": threading.Lock()}


def title_acting(acting):
    try:
        act(call("titleActing", acting), CHECK_WAIT)
    except (RuntimeError, TimeoutError, OSError):
        pass


@contextlib.contextmanager
def titled():
    with ACTING["lock"]:
        ACTING["calls"] += 1
        if ACTING["calls"] == 1:
            title_acting(True)
    try:
        yield
    finally:
        with ACTING["lock"]:
            ACTING["calls"] -= 1
            if not ACTING["calls"]:
                title_acting(False)


def acting(tool):
    @functools.wraps(tool)
    def acted(*arguments, **keywords):
        return on_agent_tab(tool, *arguments, **keywords)
    return acted


def on_agent_tab(tool, *arguments, **keywords):
    if not AGENT_TAB:
        send(":nop", anywhere=True)
        raise RuntimeError("the agent has no tab of its own: point at the tab shown with claim_shown_tab, "
                           "or open one with open_url_in_new_tab")
    began = time.monotonic()
    try:
        shown_title(CHECK_WAIT)
    except TimeoutError:
        raise RuntimeError("the agent's tab is gone: open one again with open_url_in_new_tab") from None
    checked = time.monotonic()
    result = tool(*arguments, **keywords)
    acted = time.monotonic()
    title_acting(True)
    print(f"timing {tool.__name__}: check {checked - began:.2f} s, tool {acted - checked:.2f} s,"
          f" mark {time.monotonic() - acted:.2f} s", file=sys.stderr, flush=True)
    return result


@qutebrowserserver.tool(annotations=ToolAnnotations(readOnlyHint=True))
@acting
def readonly_read_dom(selector: str = "", text: bool = False, shown: bool = False) -> str:
    """The DOM of the page qutebrowser shows, as HTML, its fields holding their live values, or only the elements SELECTOR picks.

    With TEXT, only the text of each element, one to a line, without its markup.
    With SHOWN, only what the page shows, leaving out what it keeps hidden.
    Prefer it to the screenshot: it is text."""
    page = bs4.BeautifulSoup(act(call("liveValues") + (call("markHidden") if shown else "")), "html.parser")
    if shown:
        for hidden in page.select("[data-qutebrowser-mcp-hidden]"):
            hidden.decompose()
    elements = page.select(selector) if selector else [page]
    if text:
        return "\n".join(element.get_text(" ", strip=True) for element in elements)
    return "\n".join(str(element) for element in elements)

PNG_END = b"IEND\xaeB`\x82"


def telling(why, doing, rather):
    """Show me WHY the agent is DOING, refusing a blank one, RATHER saying what it skips."""
    if not why.strip():
        raise RuntimeError(f"say why you need this rather than {rather}")
    send(f":message-info {shlex.quote(f'the agent {doing}: {why}')}")


@qutebrowserserver.tool(annotations=ToolAnnotations(readOnlyHint=True))
@acting
def readonly_screenshot(why: str) -> Image:
    """What qutebrowser shows of the page, as a PNG, telling me WHY the agent looks.

    Only for what the DOM cannot say: layout, colours, images, whether something really shows.
    Otherwise use readonly_read_dom."""
    telling(why, "looks", "readonly_read_dom")
    with tempfile.TemporaryDirectory() as tmp:
        shot = pathlib.Path(tmp) / "shot.png"
        send(f":screenshot {shot}")
        wait_for(shot, lambda data: data.endswith(PNG_END))
        return Image(data=shot.read_bytes(), format="png")