Merge branch 'guard-acting-tools' into aihawk-app

src/aihawk/apply.py

@@ -20,6 +20,7 @@ from contextlib import asynccontextmanager
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Protocol
+from urllib.parse import urlsplit
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
@@ -175,8 +176,17 @@ async def _element(session: Session, found: str) -> dict[str, Any] | None:
     return element if isinstance(element, dict) else None
 
 
-async def refusal(session: Session, name: str, arguments: dict[str, Any], form_touched: bool) -> str | None:
+def _host(url: str) -> str:
+    return (urlsplit(url).hostname or "").lower()
+
+
+async def refusal(session: Session, name: str, arguments: dict[str, Any], form_touched: bool, hosts: frozenset[str] = frozenset()) -> str | None:
     """Why the model may not make this call; None when it may."""
+    if name == "browser_navigate":
+        # A link can send a form (a GET that confirms), so the model goes only where the form and its mailbox are.
+        if _host(str(arguments.get("url", ""))) in hosts:
+            return None
+        return "Refused: that address is not the form's site or a mailbox a plugin uses. Go back to the form's own address."
     if name == "browser_press_key":
         # Every value, not the one argument: a name the server changes tomorrow must not let Enter through.
         key = " ".join(str(value) for value in arguments.values()).lower()
@@ -240,11 +250,42 @@ def _images(result: Any) -> list[dict[str, Any]]:
     ]
 
 
-async def fill_form(session: Session, tools: list[dict[str, Any]], client: AsyncOpenAI, model: str, brief: str, guidance: str = "", page_rules: str = "") -> dict[str, Any]:
+class LeftForm(Exception):
+    """A call of the model's moved the page to another site, or could not be checked: it may have sent the form."""
+
+
+# The calls that can change the page. A navigation is the model's own choice, so it is not checked after: `refusal` judges it.
+ACTING = {"browser_click", "browser_click_at", "browser_type", "browser_select_option", "browser_upload_files", "browser_press_key"}
+PAGE = "({url: location.href, text: document.body ? document.body.innerText : ''})"
+
+
+async def _page(session: Session) -> dict[str, str] | None:
+    """The page's address and text, as the server reads them; None when they cannot be read."""
+    result = await session.call_tool("browser_evaluate", {"expression": PAGE})
+    try:
+        page = None if result.isError else json.loads(tool_text(result))
+    except ValueError:
+        return None
+    return page if isinstance(page, dict) else None
+
+
+async def _after(session: Session, name: str, form_host: str) -> None:
+    """Stops the form when a call moved the page to another site. A same-site send is not detected: a multi-step form
+    advances by address too, so the address alone cannot tell the two apart."""
+    after = await _page(session)
+    if after is None:
+        raise LeftForm(f"After {name} the page could not be read, so it is not known whether the application went out. It is not retried.")
+    if _host(after.get("url", "")) != form_host:
+        raise LeftForm(f"After {name} the page went to {after.get('url')}, not the form. It is not known whether the application went out. It is not retried.")
+
+
+async def fill_form(session: Session, tools: list[dict[str, Any]], client: AsyncOpenAI, model: str, brief: str, guidance: str = "", page_rules: str = "", form_url: str = "", reaches: frozenset[str] = frozenset()) -> dict[str, Any]:
     """The model works on the open page until it calls finish_form; returns what it reported."""
     messages: list[Any] = [{"role": "system", "content": instructions(guidance, page_rules)}, {"role": "user", "content": brief}]
     form_touched = False
     allowed = MODEL_TOOLS | {tool["function"]["name"] for tool in tools}
+    form_host = _host(form_url)
+    hosts = frozenset({form_host, *reaches})
     for _ in range(MAX_STEPS):
         response = await client.chat.completions.create(model=model, messages=messages, tools=[*tools, FINISH_TOOL])
         message = response.choices[0].message
@@ -261,11 +302,14 @@ async def fill_form(session: Session, tools: list[dict[str, Any]], client: Async
                 continue
             if call.function.name == "finish_form":
                 return arguments
-            why = await refusal(session, call.function.name, arguments, form_touched) if call.function.name in allowed else "Refused: no such tool."
+            why = await refusal(session, call.function.name, arguments, form_touched, hosts) if call.function.name in allowed else "Refused: no such tool."
             if why:
                 content = why
             else:
+                before = await _page(session) if call.function.name in ACTING else None
                 result = await session.call_tool(call.function.name, arguments)
+                if before and _host(before.get("url", "")) == form_host:
+                    await _after(session, call.function.name, form_host)
                 seen += _images(result)
                 content = tool_text(result) or "Done: what it shows comes next."
                 if call.function.name in {"browser_type", "browser_select_option", "browser_upload_files"} and not result.isError:
@@ -299,12 +343,15 @@ async def submit_and_verify(session: Session, client: AsyncOpenAI, model: str, s
     return Outcome("unconfirmed", f"{summary}\nClicked, not confirmed by the page: {verdict['evidence']}")
 
 
-async def run_on_page(session: Session, tools: list[dict[str, Any]], client: AsyncOpenAI, model: str, job: Job, cv_text: str, answers: str, cv_file: Path, confirm: Confirm | None, guidance: str = "", page_rules: str = "") -> Outcome:
+async def run_on_page(session: Session, tools: list[dict[str, Any]], client: AsyncOpenAI, model: str, job: Job, cv_text: str, answers: str, cv_file: Path, confirm: Confirm | None, guidance: str = "", page_rules: str = "", reaches: frozenset[str] = frozenset()) -> Outcome:
     brief = (
         f"The form: {job.apply_url}\nThe job: {job.title} at {job.company}\nThe CV file to upload: {cv_file}\n\n"
         f"The CV:\n{cv_text}\n\nThe answers:\n{answers}"
     )
-    report = await fill_form(session, tools, client, model, brief, guidance, page_rules)
+    try:
+        report = await fill_form(session, tools, client, model, brief, guidance, page_rules, job.apply_url, reaches)
+    except LeftForm as error:
+        return Outcome("unconfirmed", str(error))
     status, summary = report.get("status"), str(report.get("summary", ""))
     if status == "needs_input":
         return Outcome("needs_input", summary, [str(question) for question in report.get("questions") or []])
@@ -359,7 +406,7 @@ async def _server(upload_dir: Path, home: Path) -> AsyncIterator[tuple[ClientSes
             yield session, started.instructions or ""
 
 
-async def apply_to_job(client: AsyncOpenAI, model: str, job: Job, cv_text: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Confirm | None, guidance: str = "") -> Outcome:
+async def apply_to_job(client: AsyncOpenAI, model: str, job: Job, cv_text: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Confirm | None, guidance: str = "", reaches: frozenset[str] = frozenset()) -> Outcome:
     """Open the browser on the person's profile, fill the job's form and send it as `confirm` decides; None sends without asking."""
     async with _server(cv_file.parent, data_dir / "browser") as (session, page_rules):
         listed = await session.list_tools()
@@ -373,7 +420,7 @@ async def apply_to_job(client: AsyncOpenAI, model: str, job: Job, cv_text: str,
             navigated = await session.call_tool("browser_navigate", {"url": job.apply_url})
             if navigated.isError:
                 return Outcome("blocked", f"the form did not load: {tool_text(navigated)}")
-            return await run_on_page(session, tools, client, model, job, cv_text, answers, cv_file, confirm, guidance, page_rules)
+            return await run_on_page(session, tools, client, model, job, cv_text, answers, cv_file, confirm, guidance, page_rules, reaches)
         finally:
             await session.call_tool("browser_close", {})
 

src/aihawk/plugins.py

@@ -34,6 +34,7 @@ class Plugin:
     instructions: str
     apply_path: str = ""
     always: bool = False
+    reaches: tuple[str, ...] = ()  # the mailboxes an always-plugin takes the model to; the only other sites it may visit
 
     def matches(self, url: str) -> bool:
         host = (urlsplit(url).hostname or "").lower()
@@ -69,7 +70,10 @@ def parse_plugin(name: str, text: str) -> Plugin:
     if not body.strip():
         raise PluginError(f"plugin {name}: it has no instructions")
     names = tuple(host.lower() for host in hosts or ())
-    return Plugin(str(meta.get("name") or name), names, body.strip(), str(meta.get("apply_path") or ""), always)
+    reaches = meta.get("reaches") or []
+    if not isinstance(reaches, list) or not all(isinstance(host, str) for host in reaches):
+        raise PluginError(f"plugin {name}: `reaches` lists the sites its mailbox is on")
+    return Plugin(str(meta.get("name") or name), names, body.strip(), str(meta.get("apply_path") or ""), always, tuple(reaches))
 
 
 def _files(folder: Traversable | Path) -> dict[str, str]:

src/aihawk/plugins/gmail.md

@@ -1,6 +1,7 @@
 ---
 name: gmail
 when: always
+reaches: [mail.google.com]
 ---
 Use this only when the form, or the site it belongs to, asks for a code or a confirmation link that it sent by email.
 

src/aihawk/plugins/outlook.md

@@ -1,6 +1,7 @@
 ---
 name: outlook
 when: always
+reaches: [outlook.live.com]
 ---
 Use this only when the form, or the site it belongs to, asks for a code or a confirmation link that it sent by email.
 

src/aihawk/session.py

@@ -160,9 +160,11 @@ class Session:
         for _ in range(MAX_ROUNDS):
             answers = self.config.answers.read_text(encoding="utf-8") if self.config.answers.is_file() else ""
             log(f"filling the form on {plugin.name}")
-            guidance = "\n\n".join([plugin.instructions, *(extra.instructions for extra in always_plugins(self.plugins))])
+            extras = always_plugins(self.plugins)
+            guidance = "\n\n".join([plugin.instructions, *(extra.instructions for extra in extras)])
+            reaches = frozenset(host for extra in extras for host in extra.reaches)
             try:
-                outcome = await apply_to_job(self.client, self.config.model, job, cv, answers, cv_file, self.config.data_dir, self.config.browser_profile, confirm, guidance)
+                outcome = await apply_to_job(self.client, self.config.model, job, cv, answers, cv_file, self.config.data_dir, self.config.browser_profile, confirm, guidance, reaches)
             except Exception as error:  # noqa: BLE001  One job's failure is reported and recorded; the next job still runs.
                 outcome = Outcome("failed", str(error) or type(error).__name__)
             if outcome.status != "needs_input":

tests/fakes.py

@@ -38,16 +38,21 @@ class ScriptedModel:
 class ScriptedBrowser:
     """call_tool records every call; browser_evaluate says what the page's element is."""
 
-    def __init__(self, elements: dict[str, dict[str, Any]] | None = None) -> None:
+    def __init__(self, elements: dict[str, dict[str, Any]] | None = None, page: dict[str, str] | None = None) -> None:
         self.elements = elements or {}
+        self.page = page or {"url": "", "text": ""}  # the page's address and text; navigation moves the address
         self.calls: list[tuple[str, dict[str, Any]]] = []
 
     async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> SimpleNamespace:
         arguments = arguments or {}
         self.calls.append((name, arguments))
         text = "ok"
-        if name == "browser_evaluate":
+        if name == "browser_evaluate" and "location.href" in arguments["expression"]:
+            text = json.dumps(self.page)
+        elif name == "browser_evaluate":
             text = json.dumps(next((value for key, value in self.elements.items() if json.dumps(key) in arguments["expression"]), None))
+        elif name == "browser_navigate":
+            self.page = {"url": arguments["url"], "text": self.page["text"]}
         elif name == "browser_read_text":
             text = "Thank you, your application was received"
         return SimpleNamespace(isError=False, content=[SimpleNamespace(text=text)])

tests/test_apply.py

@@ -377,3 +377,90 @@ def test_sign_in_closes_the_browser_even_when_the_wait_fails(tmp_path: Path, mon
     with pytest.raises(RuntimeError, match="window was closed"):
         asyncio.run(sign_in(tmp_path / "browser-profile", tmp_path, broken))
     assert server.calls[-1] == ("browser_close", {})
+
+
+FORM_PAGE = {"url": "https://u/apply", "text": "Apply here"}
+SELECT = [tool_call("browser_select_option", selector="#pick", value="a")]
+
+
+class Moves(ScriptedBrowser):
+    """A form whose select, when changed, puts the page where the test says: a change event that leaves, or confirms."""
+
+    def __init__(self, after: dict[str, str] | None = None, reads_fail: bool = False) -> None:
+        super().__init__({"#send": SUBMIT}, page=FORM_PAGE)
+        self.after = after
+        self.reads_fail = reads_fail
+        self.changed = False
+
+    async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
+        if name == "browser_select_option":
+            self.changed = True
+            self.page = self.after or self.page
+        if name == "browser_evaluate" and self.reads_fail and self.changed:
+            return SimpleNamespace(isError=True, content=[SimpleNamespace(text="the page is gone")])
+        return await super().call_tool(name, arguments)
+
+
+def run_with(model: ScriptedModel, browser: ScriptedBrowser, confirm: Any = None, reaches: frozenset[str] = frozenset()) -> Outcome:
+    return asyncio.run(run_on_page(browser, [], model, "m", JOB, "cv", "answers", Path("cv.pdf"), confirm, reaches=reaches))  # type: ignore[arg-type]
+
+
+def test_the_model_may_go_to_the_forms_site_and_to_a_mailbox_a_plugin_reaches_and_nowhere_else() -> None:
+    browser = ScriptedBrowser()
+    mailbox = frozenset({"u", "mail.google.com"})
+    assert asyncio.run(refusal(browser, "browser_navigate", {"url": "https://u/other"}, False, mailbox)) is None
+    assert asyncio.run(refusal(browser, "browser_navigate", {"url": "https://mail.google.com/mail/u/0/"}, False, mailbox)) is None
+    assert asyncio.run(refusal(browser, "browser_navigate", {"url": "https://evil.example/confirm?action=submit"}, False, mailbox))
+
+
+def test_a_navigation_by_the_model_to_another_site_is_refused_and_never_reaches_the_browser() -> None:
+    browser = ScriptedBrowser({"#send": SUBMIT}, page=FORM_PAGE)
+    model = ScriptedModel([tool_call("browser_navigate", url="https://evil.example/")], ready(), VERIFIED)
+    assert run_with(model, browser).status == "submitted"
+    assert [args["url"] for name, args in browser.calls if name == "browser_navigate"] == []
+
+
+def test_a_change_event_that_moves_the_page_off_the_form_stops_as_unconfirmed_and_nothing_is_retried() -> None:
+    browser = Moves(after={"url": "https://thanks.example/ok", "text": "Thanks"})
+    model = ScriptedModel(SELECT, ready(), VERIFIED)
+    outcome = run_with(model, browser)
+    assert outcome.status == "unconfirmed"
+    assert "thanks.example" in outcome.summary
+    assert browser.clicks() == []
+    assert len(model.requests) == 1
+
+
+def test_a_change_that_stays_on_the_site_goes_on_to_the_form_so_a_same_site_send_is_not_detected() -> None:
+    # The address alone cannot tell a multi-step form from a send that stays on the site: this is the gap, written down.
+    browser = Moves(after={"url": "https://u/apply", "text": "Thank you, your application was received"})
+    model = ScriptedModel(SELECT, ready(), VERIFIED)
+    assert run_with(model, browser).status == "submitted"
+
+
+def test_a_page_that_cannot_be_read_after_a_change_is_unconfirmed_not_failed() -> None:
+    browser = Moves(after={"url": "https://u/apply", "text": "x"}, reads_fail=True)
+    outcome = run_with(ScriptedModel(SELECT), browser)
+    assert outcome.status == "unconfirmed"
+    assert browser.clicks() == []
+
+
+def test_a_mail_trip_to_the_mailbox_and_back_does_not_stop_the_form() -> None:
+    class Mail(ScriptedBrowser):
+        async def call_tool(self, name: str, arguments: dict[str, Any] | None = None) -> Any:
+            if name == "browser_click":
+                self.page = {"url": "https://mail.google.com/mail/u/0/#inbox/1", "text": "Your code is 123456"}
+            return await super().call_tool(name, arguments)
+
+    steps = [
+        [tool_call("browser_navigate", url="https://mail.google.com/mail/u/0/")],
+        [tool_call("browser_click", selector="#msg")],
+        [tool_call("browser_navigate", url="https://u/apply")],
+        ready(),
+        VERIFIED,
+    ]
+    browser = Mail({"#send": SUBMIT, "#msg": {"tag": "a", "type": "", "in_form": False, "text": "Verification code"}}, page=FORM_PAGE)
+    model = ScriptedModel(*steps)
+    outcome = run_with(model, browser, reaches=frozenset({"mail.google.com"}))
+    assert outcome.status == "submitted"
+    # No verdict was asked for the mail click: the five requests are the four steps and the verdict on the submit.
+    assert len(model.requests) == 5

tests/test_real_browser.py

@@ -30,9 +30,38 @@ FORM = b"""<!doctype html><title>Apply</title>
 <button id="send">Submit application</button>
 </form>"""
 
+# A form that sends itself when its select changes: the change is the send, and no button is pressed. It leaves for
+# another host (a second loopback address) by a form submit, which navigates the page.
+SELECT_FORM = b"""<!doctype html><title>Pick</title>
+<form method="post" action="%s"><select id="pick" name="pick" onchange="this.form.submit()">
+<option value="">Choose</option><option value="a">A</option></select></form>"""
+# The same send on the same site, by a script: the page stays where it is and only its text changes.
+SAME_SITE_SELECT = b"""<!doctype html><title>Pick</title>
+<select id="pick" name="pick" onchange="fetch('/sent', {method: 'POST', body: new URLSearchParams({pick: this.value})}).then(() => { document.body.innerHTML = '<h1>Thank you, your application was received</h1>'; })">
+<option value="">Choose</option><option value="a">A</option></select>"""
+
+
+class Elsewhere(BaseHTTPRequestHandler):
+    """Another host of the loopback: a form that leaves the site lands here."""
+
+    posts: ClassVar[list[bytes]] = []
+
+    def do_POST(self) -> None:
+        Elsewhere.posts.append(self.rfile.read(int(self.headers["Content-Length"])))
+        body = b"<title>Thanks</title><h1>Thank you, your application was received</h1>"
+        self.send_response(200)
+        self.send_header("Content-Type", "text/html; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args: Any) -> None:
+        pass
+
 
 class Site(BaseHTTPRequestHandler):
     posts: ClassVar[list[bytes]] = []
+    leave_to: ClassVar[str] = ""  # where the /leave form posts: another host
 
     def _page(self, body: bytes) -> None:
         self.send_response(200)
@@ -42,7 +71,12 @@ class Site(BaseHTTPRequestHandler):
         self.wfile.write(body)
 
     def do_GET(self) -> None:
-        self._page(FORM)
+        if self.path == "/select":
+            self._page(SAME_SITE_SELECT)
+        elif self.path == "/leave":
+            self._page(SELECT_FORM % Site.leave_to.encode())
+        else:
+            self._page(FORM)
 
     def do_POST(self) -> None:
         Site.posts.append(self.rfile.read(int(self.headers["Content-Length"])))
@@ -110,6 +144,50 @@ def test_a_yes_sends_the_form_with_the_name_and_the_cv(site: str, tmp_path: Path
     assert b"harmless-click-reached-the-page" in Site.posts[0]
 
 
+@pytest.fixture
+def elsewhere() -> Any:
+    Elsewhere.posts = []
+    server = ThreadingHTTPServer(("127.0.0.2", 0), Elsewhere)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    yield f"http://127.0.0.2:{server.server_port}/"
+    server.shutdown()
+
+
+def run_select(site: str, path: str, tmp_path: Path, model: ScriptedModel) -> Any:
+    cv = tmp_path / "cvs" / "cv.pdf"
+    write_cv_pdf("# Ada Rossi", cv)
+    job = Job("t:2", "Acme", "Engineer", "Rome", site, site + path, "d")
+    return asyncio.run(apply_to_job(model, "m", job, "cv", "answers", cv, tmp_path, tmp_path / "browser-profile", None))  # type: ignore[arg-type]
+
+
+def test_a_select_whose_form_leaves_for_another_host_is_unconfirmed_and_the_post_is_counted(site: str, elsewhere: str, tmp_path: Path) -> None:
+    Site.leave_to = elsewhere + "sent"
+    # The change submits the form, and the page navigates away. The address cannot be read while it loads, so the
+    # check stops it as unconfirmed on that read failure; the host check itself is not what fires here.
+    model = ScriptedModel([tool_call("browser_select_option", selector="#pick", value="a")])
+    outcome = run_select(site, "leave", tmp_path, model)
+    assert outcome.status == "unconfirmed", outcome.summary
+    assert "not known whether the application went out" in outcome.summary
+    assert len(model.requests) == 1  # asked for the select, and never again
+    assert len(Elsewhere.posts) == 1
+    assert b"pick=a" in Elsewhere.posts[0]
+    assert Site.posts == []
+
+
+def test_a_select_that_sends_on_the_same_site_is_not_detected_and_the_model_goes_on(site: str, tmp_path: Path) -> None:
+    # Known gap, written down: a send that stays on the site is not seen. The script posts and rewrites the page to say it
+    # was received, the address does not change, and the model is asked for its next step as if nothing had been sent.
+    model = ScriptedModel(
+        [tool_call("browser_select_option", selector="#pick", value="a")],
+        [tool_call("finish_form", status="blocked", summary="sent already", submit_selector="", questions=[])],
+    )
+    outcome = run_select(site, "select", tmp_path, model)
+    assert outcome.status == "blocked"
+    assert len(Site.posts) == 1
+    assert b"pick=a" in Site.posts[0]
+    assert len(model.requests) == 2
+
+
 def test_the_newest_server_still_has_what_aihawk_relies_on(tmp_path: Path) -> None:
     async def check() -> None:
         async with stdio_client(server_parameters(tmp_path, tmp_path), errlog=sys.stderr) as (read, write), ClientSession(read, write) as session:

tests/test_tui.py

@@ -189,7 +189,7 @@ def test_no_key_asks_for_one_and_keeps_it_on_this_computer(home: Path, resume: P
 
 
 def fake_apply(app: AIHawk, outcomes: list[Outcome], calls: list[Any]) -> None:
-    async def apply_to_job(client: Any, model: str, job: Any, cv: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Any, guidance: str = "") -> Outcome:
+    async def apply_to_job(client: Any, model: str, job: Any, cv: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Any, guidance: str = "", reaches: frozenset[str] = frozenset()) -> Outcome:
         calls.append({"job": job, "answers": answers, "guidance": guidance, "cv_file": cv_file, "profile": profile})
         outcome = outcomes.pop(0)
         if outcome.status == "submitted" and confirm is not None and not await confirm("the form's summary"):
@@ -490,7 +490,7 @@ def test_the_sign_in_key_is_refused_while_an_application_runs(home: Path, resume
     async def run() -> None:
         release = asyncio.Event()
 
-        async def slow_apply(client: Any, model: str, job: Any, cv: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Any, guidance: str = "") -> Outcome:
+        async def slow_apply(client: Any, model: str, job: Any, cv: str, answers: str, cv_file: Path, data_dir: Path, profile: Path, confirm: Any, guidance: str = "", reaches: frozenset[str] = frozenset()) -> Outcome:
             await release.wait()
             return Outcome("declined", "not sent")