WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-10 03:37:44 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] browser: refuse zero-LLM replay of send/submit skills, the live agent confirms those instead

Browse Source
This commit is contained in:
ciregenz
2026-06-05 00:54:31 -07:00
parent 79492416aa
commit 262d48ff6c
4 changed files with 91 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/agents/browser/browser_agent.py
+7
View File
@@ -499,6 +499,13 @@ async def run_browser_agent(
return None
if not (sk_obj and steps):
return None
# Audit finding: replay bypasses the per-tool gate and act-and-confirm,
# so a recorded Send/Submit must never re-fire silently. Those flows
# always run the live agent, which confirms before anything outward.
safe, why = browser_skills.replay_safety(steps)
if not safe:
logger.info(f"[browser-skills] skill on {host} not replayed: {why}; running the full agent so the send is confirmed")
return None
replay_attempted = True
logger.info(f"[browser-skills] REPLAY attempt: {len(steps)} steps on {host} (after {turns_spent} LLM turn(s))")
rlog: list[dict] = []
backend/apps/agents/browser/browser_skills.py
+20
View File
@@ -287,6 +287,26 @@ def distill_steps(action_log: list[dict]) -> list[dict]:
return steps
def replay_safety(steps: list[dict]) -> tuple[bool, str]:
"""A skill with an outward-facing step (click Send/Submit/Pay, type into a
composer) must never auto-replay with zero LLM and zero confirmation; only
the live agent path confirms sends. Reuses the batch replayer's wordlist so
there is exactly one definition of "irreversible"."""
from backend.apps.agents.browser import browser_batch_replay
for i, s in enumerate(steps):
tool = s.get("tool", "")
p = s.get("params", {}) or {}
probe = None
if tool in ("BrowserClickByName", "BrowserClick"):
probe = {"action": "click", "name": p.get("name") or p.get("selector") or ""}
elif tool == "BrowserType":
probe = {"action": "type", "selector": p.get("selector") or ""}
if probe and browser_batch_replay.is_send_step(probe):
what = probe.get("name") or probe.get("selector")
return False, f"step {i+1} looks irreversible/outward-facing ({what!r})"
return True, ""
def steps_are_persistable(steps: list[dict]) -> bool:
"""True only if NO step touches sensitive text / a password-shaped field /
a tokenized URL. Sensitive skills stay in-memory; they never hit disk."""
backend/tests/test_browser_agent_loop.py
+47 -22
View File
@@ -98,10 +98,12 @@ def _install(monkeypatch, primary, aux):
found = "const spec=" in expr and 'const spec=""' not in expr
return {"text": json.dumps({"ready": True, "quiet": 9999, "elems": 100, "found": found}), "url": DOC_URL}
if action == "list_interactives":
return {"text": '1 interactive elements:\n[1]<button "Submit">', "url": DOC_URL}
# a non-irreversible label on purpose: Send/Submit-named steps are
# refused by the replay send-gate, which has its own test below
return {"text": '1 interactive elements:\n[1]<button "Search">', "url": DOC_URL}
if action == "click_index":
# frontend surfaces the clicked element's role/name for skill recording
return {"text": "Clicked index 1", "url": DOC_URL, "clickedRole": "button", "clickedName": "Submit"}
return {"text": "Clicked index 1", "url": DOC_URL, "clickedRole": "button", "clickedName": "Search"}
if action == "click_by_name":
return {"text": f'Clicked button "{params.get("name")}"', "url": DOC_URL}
if action == "click":
@@ -129,7 +131,7 @@ def _install(monkeypatch, primary, aux):
def test_full_loop_goal_stagnation_adjudication_and_hint_write(monkeypatch):
BH._browser_history.clear(); BH._domain_notes.clear()
primary = FakeLLM([
Resp([_rp("click the Submit button"), _tu("BrowserListInteractives")]),
Resp([_rp("click the Search button"), _tu("BrowserListInteractives")]),
Resp([_rp("click submit"), _tu("BrowserClick", selector=".s1")]),
Resp([_rp("retry"), _tu("BrowserClick", selector=".s2")]),
Resp([_rp("retry"), _tu("BrowserClick", selector=".s3")]),
@@ -148,7 +150,7 @@ def test_full_loop_goal_stagnation_adjudication_and_hint_write(monkeypatch):
# 1) goal threaded into the loop's list_interactives call (a no-goal perception
# front-load may precede it now, so assert SOME call carries the goal)
list_calls = [c for c in sent if c["action"] == "list_interactives"]
assert any(c["params"].get("goal") == "click the Submit button" for c in list_calls)
assert any(c["params"].get("goal") == "click the Search button" for c in list_calls)
# 2) stagnation nudge injected into a tool_result (seen by a later LLM turn)
all_msgs = json.dumps([c["messages"] for c in primary.calls])
@@ -244,24 +246,24 @@ def test_skill_is_recorded_then_replayed_with_zero_llm_calls(monkeypatch):
primary = FakeLLM([
Resp([_rp("click submit"), _tu("BrowserListInteractives")]),
Resp([_rp("click it"), _tu("BrowserClickIndex", index=1)]),
Resp([Blk("text", "Done, clicked Submit.")], stop_reason="end_turn"),
Resp([Blk("text", "Done, clicked Search.")], stop_reason="end_turn"),
])
aux = FakeAux()
sent = _install(monkeypatch, primary, aux)
# Run 1 (learns). initial_url gives the host for record+replay keying.
r1 = asyncio.run(BA.run_browser_agent(
task="click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
task="click the Search button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
))
assert not r1.get("replayed")
assert SK.find_skill("docs.google.com", "click the Submit button") is not None
assert SK.find_skill("docs.google.com", "click the Search button") is not None
calls_after_run1 = len(primary.calls)
assert calls_after_run1 > 0 # run 1 used the LLM
# Run 2 (replays). Must NOT call the LLM at all, and must use click_by_name.
sent.clear()
r2 = asyncio.run(BA.run_browser_agent(
task="Please click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
task="Please click the Search button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
))
assert r2.get("replayed") is True
assert len(primary.calls) == calls_after_run1, "run 2 must make ZERO LLM calls"
@@ -307,9 +309,9 @@ def test_deferred_replay_fires_after_navigating_to_the_right_host(monkeypatch):
import backend.apps.agents.browser.browser_skills as SK
SK.clear()
BH._browser_history.clear()
SK.record_skill("docs.google.com", "click the Submit button", [
SK.record_skill("docs.google.com", "click the Search button", [
{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"},
"clicked_role": "button", "clicked_name": "Search"},
])
# turn 0 navigates to the doc; the re-check should preempt everything after.
primary = FakeLLM([
@@ -331,13 +333,13 @@ def test_deferred_replay_fires_after_navigating_to_the_right_host(monkeypatch):
# NO initial_url -> dispatch perceives google -> dispatch replay misses.
r = asyncio.run(BA.run_browser_agent(
task="Please click the Submit button", browser_id="b1", model="sonnet",
task="Please click the Search button", browser_id="b1", model="sonnet",
))
assert r.get("replayed") is True, "deferred re-check must replay after the navigation"
assert any(c["action"] == "click_by_name" for c in sent), "replay re-resolved by name"
assert len(primary.calls) == 1, "only the navigate turn ran; the re-check preempted the rest"
# and the deferred replay still promotes the skill through the trust gate
assert SK.find_skill("docs.google.com", "click the Submit button")["state"] == SK._TRUSTED
assert SK.find_skill("docs.google.com", "click the Search button")["state"] == SK._TRUSTED
def test_deferred_replay_does_not_fire_after_the_page_was_dirtied(monkeypatch):
@@ -347,9 +349,9 @@ def test_deferred_replay_does_not_fire_after_the_page_was_dirtied(monkeypatch):
import backend.apps.agents.browser.browser_skills as SK
SK.clear()
BH._browser_history.clear()
SK.record_skill("docs.google.com", "click the Submit button", [
SK.record_skill("docs.google.com", "click the Search button", [
{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"},
"clicked_role": "button", "clicked_name": "Search"},
])
# turn 0 TYPES (dirties the page), THEN turn 1 navigates to the doc host.
primary = FakeLLM([
@@ -368,7 +370,7 @@ def test_deferred_replay_does_not_fire_after_the_page_was_dirtied(monkeypatch):
monkeypatch.setattr(BA.ws_manager, "send_browser_command", _cmd, raising=False)
r = asyncio.run(BA.run_browser_agent(
task="Please click the Submit button", browser_id="b1", model="sonnet",
task="Please click the Search button", browser_id="b1", model="sonnet",
))
# a dirtied page must NOT trigger the deferred replay; the LLM ran to the end
assert not r.get("replayed"), "must not replay from a dirtied page state"
@@ -386,16 +388,16 @@ def test_replay_resolves_host_from_live_page_when_no_initial_url(monkeypatch):
SK.clear()
BH._browser_history.clear()
# a skill exists for the host the live page will report (DOC_URL -> docs.google.com)
SK.record_skill("docs.google.com", "click the Submit button", [
SK.record_skill("docs.google.com", "click the Search button", [
{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Submit"},
"clicked_role": "button", "clicked_name": "Search"},
])
primary = FakeLLM([Resp([Blk("text", "should not be needed")], stop_reason="end_turn")])
aux = FakeAux()
sent = _install(monkeypatch, primary, aux)
# NOTE: no initial_url passed; the fake browser reports url=DOC_URL via perception
r = asyncio.run(BA.run_browser_agent(
task="Please click the Submit button", browser_id="b1", model="sonnet",
task="Please click the Search button", browser_id="b1", model="sonnet",
))
assert r.get("replayed") is True, "must replay via host learned from the live page"
assert len(primary.calls) == 0, "replay must make ZERO LLM calls"
@@ -491,14 +493,37 @@ def test_replay_success_promotes_skill_to_trusted_through_the_loop(monkeypatch):
aux = FakeAux()
_install(monkeypatch, primary, aux)
asyncio.run(BA.run_browser_agent(
task="click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
task="click the Search button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
))
assert SK.find_skill("docs.google.com", "click the Submit button")["state"] == SK._PROBATION
assert SK.find_skill("docs.google.com", "click the Search button")["state"] == SK._PROBATION
r2 = asyncio.run(BA.run_browser_agent(
task="click the Submit button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
task="click the Search button", browser_id="b1", model="sonnet", initial_url=DOC_URL,
))
assert r2.get("replayed") is True
assert SK.find_skill("docs.google.com", "click the Submit button")["state"] == SK._TRUSTED
assert SK.find_skill("docs.google.com", "click the Search button")["state"] == SK._TRUSTED
def test_skill_with_send_step_never_replays_silently(monkeypatch):
# The audit finding: replay bypasses act-and-confirm and the per-tool gate,
# so a recorded Send/Submit must NOT auto-replay; the live agent (which
# confirms before anything outward) runs instead, and trust is untouched.
import backend.apps.agents.browser.browser_skills as SK
SK.clear()
BH._browser_history.clear()
SK.record_skill("docs.google.com", "message tyler saying hi", [
{"tool": "BrowserClickIndex", "input": {}, "ok": True,
"clicked_role": "button", "clicked_name": "Send"},
])
primary = FakeLLM([Resp([Blk("text", "handled live with confirmation")], stop_reason="end_turn")])
sent = _install(monkeypatch, primary, FakeAux())
r = asyncio.run(BA.run_browser_agent(
task="message tyler saying hi", browser_id="b1", model="sonnet", initial_url=DOC_URL,
))
assert not r.get("replayed"), "a send-step skill must never auto-replay"
assert not any(c["action"] == "click_by_name" for c in sent), "the recorded Send was not re-fired"
assert len(primary.calls) > 0, "the live agent ran instead"
assert SK.find_skill("docs.google.com", "message tyler saying hi")["state"] == SK._PROBATION, \
"skipping replay is not a replay failure; trust stays untouched"
def test_unproven_skill_that_fails_is_quarantined_and_never_retried(monkeypatch):
backend/tests/test_browser_skills.py
+17
View File
@@ -458,3 +458,20 @@ def test_list_skills_surfaces_state_rev_and_builds_on(_isolated_skills):
assert foundation["state"] == sk._TRUSTED and foundation["builds_on"] == []
assert composed["builds_on"] == [sk._sig("search shoes now")]
assert "rev" in composed and "steps" in composed
def test_replay_safety_refuses_send_steps_and_passes_reads():
safe, _ = sk.replay_safety([
{"tool": "BrowserNavigate", "params": {"url": "https://a.com"}},
{"tool": "BrowserClickByName", "params": {"role": "button", "name": "Search"}},
{"tool": "BrowserPressKey", "params": {"key": "Enter"}},
])
assert safe is True
for bad in (
[{"tool": "BrowserClickByName", "params": {"role": "button", "name": "Send"}}],
[{"tool": "BrowserClickByName", "params": {"role": "button", "name": "Place order"}}],
[{"tool": "BrowserClick", "params": {"selector": "button[aria-label='Submit form']"}}],
[{"tool": "BrowserType", "params": {"selector": "#message-body", "text": "hi"}}],
):
ok, why = sk.replay_safety(bad)
assert ok is False and "irreversible" in why