From f25fe95c37b5e81dd000676f2e01e71be9f5ac51 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Sat, 8 Aug 2026 00:37:38 -0700 Subject: [PATCH] [eric] browser: RequestHumanIntervention is never auto-injected, the toggle really removes it, and workflow runs never see it --- backend/apps/agents/browser/browser_agent.py | 70 ++++++++++----- .../agents/browser/browser_login_handoff.py | 12 +-- backend/apps/agents/browser/browser_loop.py | 13 +-- backend/apps/agents/browser/browser_schema.py | 16 ++-- .../browser/human_intervention_allowed.py | 31 +++++++ .../apps/agents/browser/intervention_copy.py | 61 +++++++++++++ backend/tests/test_browser_agent_loop.py | 89 +++++++++++++++---- backend/tests/test_browser_login_handoff.py | 8 -- backend/tests/test_human_intervention_gate.py | 82 +++++++++++++++++ 9 files changed, 313 insertions(+), 69 deletions(-) create mode 100644 backend/apps/agents/browser/human_intervention_allowed.py create mode 100644 backend/apps/agents/browser/intervention_copy.py create mode 100644 backend/tests/test_human_intervention_gate.py diff --git a/backend/apps/agents/browser/browser_agent.py b/backend/apps/agents/browser/browser_agent.py index c63e5f02..c28a743f 100644 --- a/backend/apps/agents/browser/browser_agent.py +++ b/backend/apps/agents/browser/browser_agent.py @@ -53,7 +53,9 @@ from backend.apps.agents.browser.browser_loop import ( stagnation_exhausted, ) from backend.apps.agents.browser.browser_validator import adjudicate_stuck +from backend.apps.agents.browser.human_intervention_allowed import human_intervention_allowed from backend.apps.agents.browser.humanize_element_rows import humanize_element_rows +from backend.apps.agents.browser.intervention_copy import strip_intervention_copy from backend.apps.agents.browser.strip_lone_surrogates import strip_lone_surrogates # Single actions the model could have folded into one BrowserBatch turn; reads, waits, and the batch tools themselves don't count toward the streak. @@ -1052,6 +1054,10 @@ async def run_browser_agent( from backend.apps.agents.agent_manager import agent_manager p_browser_perms = load_builtin_permissions() + # One decision for the whole run: may this agent even OFFER to ask a human? Code never fires + # RequestHumanIntervention itself; this only gates the model's menu (Eric's call, 2026-08-08). + p_hitl_allowed = human_intervention_allowed( + p_browser_perms, parent_session_id, agent_manager.sessions) session_id = uuid4().hex cancel_event = asyncio.Event() @@ -1349,7 +1355,7 @@ async def run_browser_agent( loop_trigger_count = 0 card_gone_streak = 0 # consecutive "card is gone" results -> fail fast, don't spin route_hinted_hosts: set[str] = set() # surface the fast network tier once per host - p_login_prompted: set[str] = set() # login-once handoff: at most one sign-in pause per domain per run + p_login_attempted: set[str] = set() # login-once handoff: at most one silent borrow attempt per domain per run # Stagnation state: busy-but-stuck detection (no URL change + failures across a run of actions), distinct from the exact-repeat loop above. stagnation_streak = 0 @@ -1450,6 +1456,10 @@ async def run_browser_agent( pass # Prompt-caching shapes built once: system as a single cached text block, and the last tool carrying the cache_control marker (Anthropic keys on the trailing marker, so one marker covers the whole tool array + system). + if not p_hitl_allowed: + # No human exists for this run: strip the tool's prompt copy everywhere (including seeded + # and learned playbook lines), so "don't ask" is physics, not a plea the model can ignore. + run_system_prompt = strip_intervention_copy(run_system_prompt) p_cached_system = [{ "type": "text", "text": run_system_prompt, "cache_control": {"type": "ephemeral"}, @@ -1458,6 +1468,8 @@ async def run_browser_agent( p_cached_tools = [dict(t) for t in (APP_VISIBLE_TOOLS if app_mode else browser_schema.MODEL_VISIBLE_TOOLS)] if not browser_delete_script.delete_tool_enabled(): p_cached_tools = [t for t in p_cached_tools if t["name"] != "BrowserDeleteItem"] + if not p_hitl_allowed: + p_cached_tools = [t for t in p_cached_tools if t["name"] != "RequestHumanIntervention"] if p_cached_tools: p_cached_tools[-1] = {**p_cached_tools[-1], "cache_control": {"type": "ephemeral"}} @@ -1901,28 +1913,18 @@ async def run_browser_agent( if done_called or cancel_event.is_set(): break - # Login-once handoff: if we've landed on a login wall, pause so the user can sign in ONCE - # in this card (the persistent partition keeps the session, so future runs won't ask - # again), then continue. At most one pause per domain per run; the model's own - # RequestHumanIntervention stays as the fallback for walls this detector misses. + # Login-once handoff: on a login wall, silently borrow the sign-in the user already has + # in their everyday browser (the persistent partition keeps it for future runs). At + # most one attempt per domain per run. NOTHING here may fire RequestHumanIntervention; + # asking for help is the model's choice alone, and only when the tool is offered at all + # (Eric's call, 2026-08-08: auto-injection made workflows impossible to run unattended). # Soft signed-out (composer withheld behind a "Sign in") only counts once the agent has - # actually tried and is still stuck, so a first-turn glance can't raise a false prompt. + # actually tried and is still stuck, so a first-turn glance can't raise a false attempt. p_wall_dom = browser_login_handoff.login_wall_domain( last_seen_url, "\n".join(attached_state_seen), allow_soft=(turn >= 2)) - if p_wall_dom and p_wall_dom not in p_login_prompted: - p_login_prompted.add(p_wall_dom) - # Borrow the sign-in the user already has in their everyday browser first: when it - # lands nobody is interrupted at all. Anything less falls through to the pause. - p_signed_in = await try_borrow_signin(p_wall_dom, browser_id, tab_id, last_seen_url) - if not p_signed_in: - p_login_problem, p_login_instruction = browser_login_handoff.prompt_copy(p_wall_dom) - p_login_decision = await p_request_browser_approval( - session, "RequestHumanIntervention", - {"problem": p_login_problem, "instruction": p_login_instruction}) - if cancel_event.is_set(): - break - p_signed_in = p_login_decision.get("behavior") != "deny" - if p_signed_in: + if p_wall_dom and p_wall_dom not in p_login_attempted: + p_login_attempted.add(p_wall_dom) + if await try_borrow_signin(p_wall_dom, browser_id, tab_id, last_seen_url): browser_login_handoff.record_login(p_wall_dom) p_signed_note = (f"You are now signed in to {p_wall_dom}. The page has changed; " "look at it fresh and continue the task.") @@ -2450,8 +2452,28 @@ async def run_browser_agent( }) break - # Handle RequestHumanIntervention; pause and wait for user + # Handle RequestHumanIntervention; pause and wait for user. Only the model can + # reach here, and only when the tool was actually offered this run. if tu.name == "RequestHumanIntervention": + if not p_hitl_allowed: + # Non-Anthropic lanes don't enforce the tool schema, so a hallucinated + # call must land as a wall, not a pause. + p_no_hitl_text = ("Human intervention is not available in this run. " + "Adapt, or call Done with success=false naming the blocker.") + tool_results.append({ + "type": "tool_result", "tool_use_id": tu.id, + "content": [{"type": "text", "text": p_no_hitl_text}], + }) + result_msg = Message( + role="tool_result", + content={"text": p_no_hitl_text, "tool_name": tu.name, "elapsed_ms": 0}, + ) + session.messages.append(result_msg) + await ws_manager.send_to_session(session_id, "agent:message", { + "session_id": session_id, + "message": result_msg.model_dump(mode="json"), + }) + continue problem = tu.input.get("problem", "") instruction = tu.input.get("instruction", "") decision = await p_request_browser_approval( @@ -2889,6 +2911,8 @@ async def run_browser_agent( loop_trigger_count += 1 repeat_count = sum(1 for c in recent_tool_calls if c == call_key) warning = LOOP_WARNING_TEXT.format(count=repeat_count) + if not p_hitl_allowed: + warning = strip_intervention_copy(warning) logger.warning( f"[browser-agent {session_id}] loop detected on {tu.name} " f"(trigger #{loop_trigger_count}): {warning}" @@ -2903,6 +2927,8 @@ async def run_browser_agent( ) # Skip the nudge when the loud loop warning already fired this turn (avoid double-messaging), but the aux adjudication below is NOT gated on is_loop: repeated identical failures trip BOTH detectors, and that's exactly when the escape hatch is needed. if stag_nudge and not is_loop: + if not p_hitl_allowed: + stag_nudge = strip_intervention_copy(stag_nudge) logger.warning( f"[browser-agent {session_id}] stagnation streak " f"{stagnation_streak} on {tu.name}" @@ -2957,6 +2983,8 @@ async def run_browser_agent( aux_client, aux_model, current_next_goal, recent, page_text, )) if guidance: + if not p_hitl_allowed: + guidance = strip_intervention_copy(guidance) content_blocks = content_blocks + [ {"type": "text", "text": f"\n\nšŸ’” Suggested next step: {guidance}"} ] diff --git a/backend/apps/agents/browser/browser_login_handoff.py b/backend/apps/agents/browser/browser_login_handoff.py index a79ba4e7..e584839d 100644 --- a/backend/apps/agents/browser/browser_login_handoff.py +++ b/backend/apps/agents/browser/browser_login_handoff.py @@ -11,7 +11,7 @@ the send-script's decline can never disagree about what a login wall is. import datetime import os -from typing import Dict, List, Optional, Tuple +from typing import Dict, List, Optional from urllib.parse import urlparse from typeguard import typechecked @@ -90,13 +90,3 @@ def login_wall_domain(current_url: str, state_text: str, allow_soft: bool = Fals return None -@typechecked -def prompt_copy(domain: str) -> Tuple[str, str]: - """(problem, instruction) for the pause overlay, worded by whether the user has signed into - this site before (re-auth) or it's a first sign-in.""" - if is_authenticated(domain): - problem = f"Your {domain} sign-in looks signed out, it may have expired or be a different account." - else: - problem = f"{domain} needs you to sign in before I can keep going." - instruction = "Log in to the site in the browser above, then click Done and I'll pick up right where I left off." - return problem, instruction diff --git a/backend/apps/agents/browser/browser_loop.py b/backend/apps/agents/browser/browser_loop.py index 33d8e063..add4c23c 100644 --- a/backend/apps/agents/browser/browser_loop.py +++ b/backend/apps/agents/browser/browser_loop.py @@ -11,6 +11,10 @@ import json import re from backend.apps.agents.browser import browser_send_parse +from backend.apps.agents.browser.intervention_copy import ( + LOOP_WARNING_INTERVENTION_FIX, + STAGNATION_INTERVENTION_TAIL, +) # Tools that are read-only / idempotent and should NOT count toward loop detection. Repeating these is normal (scrolling through a feed, taking successive screenshots, polling for an element to appear). LOOP_DETECTION_EXCLUDED_TOOLS = { @@ -106,8 +110,8 @@ LOOP_WARNING_TEXT = ( "hidden, or covered by an overlay; use BrowserGetText or BrowserScreenshot to check " "whether the page is really a login wall, captcha, or error page. THEN fix that exact " "cause: a wrong selector means switch to BrowserListInteractives + BrowserClickIndex " - "or BrowserPressKey; a blocked element means clear the blocker first; a login, " - "captcha, or error page means call RequestHumanIntervention. Don't just try another " + "or BrowserPressKey; a blocked element means clear the blocker first; " + + LOOP_WARNING_INTERVENTION_FIX + ". Don't just try another " "selector if the problem isn't a selector." ) @@ -177,9 +181,8 @@ def stagnation_nudge(streak: int) -> str: base += ( " Switching selectors hasn't worked, so the PLAN itself is likely " "wrong: step back and revise your overall approach (a different page, " - "route, or entry point), not just the selector. If even a fresh plan " - "can't make progress, call RequestHumanIntervention instead of " - "continuing to fail." + "route, or entry point), not just the selector. " + + STAGNATION_INTERVENTION_TAIL ) return base diff --git a/backend/apps/agents/browser/browser_schema.py b/backend/apps/agents/browser/browser_schema.py index fc18f8d9..e04123a3 100644 --- a/backend/apps/agents/browser/browser_schema.py +++ b/backend/apps/agents/browser/browser_schema.py @@ -6,6 +6,11 @@ prompt, and the turn/report invariants. Exceeds the 300-LOC soft ceiling on purpose because it is one cohesive data blob, not multiple responsibilities. """ +from backend.apps.agents.browser.intervention_copy import ( + INTERVENTION_SECTION, + LOOP_AWARENESS_INTERVENTION_PHRASE, +) + # Two prompt levers that A/B-proved out and now ship unconditionally. THINK_SHORTER (no prose beside action tools; ReportProgress IS the thinking) cut per-turn output ~28% and roughly halved narration turns. MERGE_VERIFY (a confirmed `expect` is the proof, skip the re-check) drops a wasted round-trip at the end. P_THINK_SHORTER = ( "Do NOT write a free-text sentence next to your action tools: your ReportProgress " @@ -975,8 +980,8 @@ SYSTEM_PROMPT = ( "have called the same tool with the same parameters and gotten the same " "result multiple times in a row. STOP. Do NOT retry the same approach. " "Switch strategy entirely: try a different tool, a different selector, " - "keyboard shortcuts, or call RequestHumanIntervention if you genuinely " - "cannot proceed. The loop detector will force-exit the agent if you " + "keyboard shortcuts" + LOOP_AWARENESS_INTERVENTION_PHRASE + ". " + "The loop detector will force-exit the agent if you " "ignore it more than 5 times.\n\n" "## Use prior context\n" @@ -1102,12 +1107,7 @@ SYSTEM_PROMPT = ( "whole thing to a file and hands you the path. Then Done, telling the user that path. That " "is one step instead of a dozen.\n\n" - "## When you genuinely cannot proceed\n" - "Use RequestHumanIntervention for:\n" - "- Login walls (the user thinks they're logged in but the session expired)\n" - "- Captchas, 2FA prompts, age verification gates\n" - "- Anything genuinely ambiguous about user intent\n" - "Don't use it for normal tool failures; try a different approach first.\n\n" + + INTERVENTION_SECTION + "Complete the task autonomously. When you're finished, end the run by calling the Done " "tool, never by typing a sentence. Put your reply to the user in Done's `message`, " diff --git a/backend/apps/agents/browser/human_intervention_allowed.py b/backend/apps/agents/browser/human_intervention_allowed.py new file mode 100644 index 00000000..3bcbff3d --- /dev/null +++ b/backend/apps/agents/browser/human_intervention_allowed.py @@ -0,0 +1,31 @@ +"""The one decision for whether a browser run may ASK a human for help. Code never fires +RequestHumanIntervention itself; this gate only controls whether the model is offered the tool.""" + +from typing import Dict, Optional, Set + +from typeguard import typechecked + +from backend.apps.agents.core.models import AgentSession + + +@typechecked +def human_intervention_allowed( + builtin_perms: Dict[str, str], + parent_session_id: Optional[str], + sessions: Dict[str, AgentSession], +) -> bool: + if builtin_perms.get("RequestHumanIntervention") == "deny": + return False + # A workflow run anywhere up the parent chain means nobody is watching the screen. The visited + # set means a corrupt chain degrades to "allowed" instead of hanging the spawn. + seen: Set[str] = set() + sid = parent_session_id + while sid and sid not in seen: + seen.add(sid) + parent = sessions.get(sid) + if parent is None: + return True + if parent.workflow_run_id: + return False + sid = parent.parent_session_id + return True diff --git a/backend/apps/agents/browser/intervention_copy.py b/backend/apps/agents/browser/intervention_copy.py new file mode 100644 index 00000000..ebc9093a --- /dev/null +++ b/backend/apps/agents/browser/intervention_copy.py @@ -0,0 +1,61 @@ +"""Every piece of prompt copy that advertises RequestHumanIntervention, plus the scrubber that +removes all of it for runs where no human exists (workflow runs, or the tool switched off). + +The named constants are spliced into SYSTEM_PROMPT / the loop nudges at their definition sites, so +strip_intervention_copy's exact replaces cannot drift apart from the live copy; a test asserts the +scrubbed output never names the tool, which also catches copy born anywhere else (seed playbooks, +learned playbooks, aux adjudication advice) via the catch-all. +""" + +from typeguard import typechecked + +INTERVENTION_SECTION = ( + "## When you genuinely cannot proceed\n" + "Use RequestHumanIntervention for:\n" + "- Login walls (the user thinks they're logged in but the session expired)\n" + "- Captchas, 2FA prompts, age verification gates\n" + "- Anything genuinely ambiguous about user intent\n" + "Don't use it for normal tool failures; try a different approach first.\n\n" +) + +NO_INTERVENTION_SECTION = ( + "## When you genuinely cannot proceed\n" + "No human is available in this run. If a login wall, captcha, or 2FA gate blocks every " + "route, stop and call Done with success=false, naming exactly what blocked you.\n\n" +) + +LOOP_AWARENESS_INTERVENTION_PHRASE = ( + ", or call RequestHumanIntervention if you genuinely cannot proceed" +) + +LOOP_WARNING_INTERVENTION_FIX = ( + "a login, captcha, or error page means call RequestHumanIntervention" +) +LOOP_WARNING_NO_INTERVENTION_FIX = ( + "a login, captcha, or error page you cannot route around means the task is blocked, " + "so call Done with success=false and name the wall" +) + +STAGNATION_INTERVENTION_TAIL = ( + "If even a fresh plan can't make progress, call RequestHumanIntervention instead of " + "continuing to fail." +) +STAGNATION_NO_INTERVENTION_TAIL = ( + "If even a fresh plan can't make progress, call Done with success=false and report " + "exactly what is blocking you." +) + + +@typechecked +def strip_intervention_copy(text: str) -> str: + out = text.replace(INTERVENTION_SECTION, NO_INTERVENTION_SECTION) + out = out.replace(LOOP_AWARENESS_INTERVENTION_PHRASE, "") + out = out.replace(LOOP_WARNING_INTERVENTION_FIX, LOOP_WARNING_NO_INTERVENTION_FIX) + out = out.replace(STAGNATION_INTERVENTION_TAIL, STAGNATION_NO_INTERVENTION_TAIL) + out = out.replace( + "use RequestHumanIntervention", + "stop and call Done with success=false naming the wall", + ) + # Catch-all: no string reaching the model may name a tool it does not have. + out = out.replace("RequestHumanIntervention", "Done with success=false") + return out diff --git a/backend/tests/test_browser_agent_loop.py b/backend/tests/test_browser_agent_loop.py index 2ac3bf5c..d60b200f 100644 --- a/backend/tests/test_browser_agent_loop.py +++ b/backend/tests/test_browser_agent_loop.py @@ -1871,9 +1871,9 @@ def test_autosend_finishes_the_send_after_the_model_fills(monkeypatch): assert len(primary.calls) == 1, f"model called {len(primary.calls)}x; the send should cost zero model turns" -def test_login_wall_pauses_and_remembers_the_site(monkeypatch, tmp_path): - """Landing on a login wall auto-fires the RequestHumanIntervention pause with sign-in wording, - and once the user resolves it (Done), the domain is remembered so future runs skip re-prompting.""" +def test_login_wall_never_auto_fires_intervention(monkeypatch, tmp_path): + """A login wall must NEVER auto-fire RequestHumanIntervention (ENG-198): the silent cookie + borrow is the only automatic assist, and when it misses the model just keeps its turn.""" from backend.apps.agents.browser import browser_login_handoff as H monkeypatch.setattr(H, "P_STORE_PATH", str(tmp_path / "auth.json")) @@ -1884,6 +1884,42 @@ def test_login_wall_pauses_and_remembers_the_site(monkeypatch, tmp_path): return {"behavior": "allow"} monkeypatch.setattr(BA, "p_request_browser_approval", p_fake_approval) + borrows = [] + + async def p_no_borrow(domain, browser_id, tab_id, url): + borrows.append(domain) + return False + monkeypatch.setattr(BA, "try_borrow_signin", p_no_borrow) + + primary = FakeLLM([ + Resp([p_rp("open the login page"), p_tu("BrowserNavigate", url="https://acme.example/login")]), + Resp([p_tu("Done", message="blocked by the acme login wall", success=False)]), + ]) + p_install(monkeypatch, primary, FakeAux()) + p_run_settled(task="log into acme and open my dashboard", browser_id="b1", model="sonnet") + + assert borrows == ["acme.example"], "the silent borrow is the only automatic assist" + assert approvals == [], "nothing may auto-fire RequestHumanIntervention" + assert not H.is_authenticated("acme.example") + + +def test_login_wall_borrow_success_remembers_the_site(monkeypatch, tmp_path): + """When the cookie borrow lands, the domain is remembered so future runs skip the wall, with + zero human interruptions along the way.""" + from backend.apps.agents.browser import browser_login_handoff as H + monkeypatch.setattr(H, "P_STORE_PATH", str(tmp_path / "auth.json")) + + approvals = [] + + async def p_fake_approval(session, tool_name, tool_input): + approvals.append((tool_name, tool_input)) + return {"behavior": "allow"} + monkeypatch.setattr(BA, "p_request_browser_approval", p_fake_approval) + + async def p_borrow_ok(domain, browser_id, tab_id, url): + return True + monkeypatch.setattr(BA, "try_borrow_signin", p_borrow_ok) + primary = FakeLLM([ Resp([p_rp("open the login page"), p_tu("BrowserNavigate", url="https://acme.example/login")]), Resp([p_tu("Done", message="all set")]), @@ -1891,28 +1927,49 @@ def test_login_wall_pauses_and_remembers_the_site(monkeypatch, tmp_path): p_install(monkeypatch, primary, FakeAux()) p_run_settled(task="log into acme and open my dashboard", browser_id="b1", model="sonnet") - assert any(t == "RequestHumanIntervention" and "sign in" in ti["problem"].lower() - for t, ti in approvals), approvals + assert approvals == [] assert H.is_authenticated("acme.example") -def test_login_wall_skip_does_not_remember(monkeypatch, tmp_path): - """Skipping the sign-in (deny) leaves the site UNremembered and lets the run continue.""" - from backend.apps.agents.browser import browser_login_handoff as H - monkeypatch.setattr(H, "P_STORE_PATH", str(tmp_path / "auth.json")) +def test_workflow_child_never_gets_the_intervention_tool(monkeypatch): + """A browser run whose parent chain contains a workflow run must not even be OFFERED + RequestHumanIntervention: the tool is absent from the wire schema and the system prompt stops + advertising it (ENG-198).""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.core.models import AgentSession - async def p_deny(session, tool_name, tool_input): - return {"behavior": "deny", "message": "Skipped by user"} - monkeypatch.setattr(BA, "p_request_browser_approval", p_deny) + parent_id = "wfparent-" + uuid.uuid4().hex[:8] + agent_manager.sessions[parent_id] = AgentSession( + id=parent_id, name="wf", model="sonnet", mode="chat", workflow_run_id="wf-run-1") primary = FakeLLM([ - Resp([p_rp("open the login page"), p_tu("BrowserNavigate", url="https://acme.example/login")]), - Resp([p_tu("Done", message="ok")]), + Resp([p_rp("look around"), p_tu("BrowserListInteractives")]), + Resp([p_tu("Done", message="done")]), ]) p_install(monkeypatch, primary, FakeAux()) - p_run_settled(task="log into acme", browser_id="b1", model="sonnet") + try: + p_run_settled(task="check the docs page", browser_id="b1", model="sonnet", + parent_session_id=parent_id) + finally: + agent_manager.sessions.pop(parent_id, None) - assert not H.is_authenticated("acme.example") + wire_tools = [t["name"] for t in primary.calls[0]["tools"]] + assert "RequestHumanIntervention" not in wire_tools + system_text = "".join(b.get("text", "") for b in primary.calls[0]["system"]) + assert "RequestHumanIntervention" not in system_text + + +def test_chat_child_still_gets_the_intervention_tool(monkeypatch): + """Interactive runs keep the tool on the menu; removing it is only for workflows/toggle-off.""" + primary = FakeLLM([ + Resp([p_rp("look around"), p_tu("BrowserListInteractives")]), + Resp([p_tu("Done", message="done")]), + ]) + p_install(monkeypatch, primary, FakeAux()) + p_run_settled(task="check the docs page", browser_id="b1", model="sonnet") + + wire_tools = [t["name"] for t in primary.calls[0]["tools"]] + assert "RequestHumanIntervention" in wire_tools def test_an_unconfirmed_post_teaches_us_nothing(monkeypatch): diff --git a/backend/tests/test_browser_login_handoff.py b/backend/tests/test_browser_login_handoff.py index c531e8dd..81a34c1e 100644 --- a/backend/tests/test_browser_login_handoff.py +++ b/backend/tests/test_browser_login_handoff.py @@ -47,14 +47,6 @@ def test_first_seen_preserved_last_login_advances(): assert second["last_login"] >= first["last_login"] -def test_prompt_copy_wording_flips_on_history(): - assert "needs you to sign in" in h.prompt_copy("reddit.com")[0] - h.record_login("reddit.com") - assert "expired or be a different account" in h.prompt_copy("reddit.com")[0] - # instruction is always the same actionable line - assert "click Done" in h.prompt_copy("reddit.com")[1] - - def test_record_login_is_fail_open(monkeypatch): # an unwritable path must not raise; the run just treats it as a fresh sign-in next time monkeypatch.setattr(h, "P_STORE_PATH", "/nonexistent-dir-xyz/authenticated_domains.json") diff --git a/backend/tests/test_human_intervention_gate.py b/backend/tests/test_human_intervention_gate.py new file mode 100644 index 00000000..8fc3a337 --- /dev/null +++ b/backend/tests/test_human_intervention_gate.py @@ -0,0 +1,82 @@ +"""RequestHumanIntervention is opt-in for the model, never injected by code: the gate removes the +tool for workflow runs and when the settings toggle says deny, and the scrubber guarantees no prompt +copy anywhere still advertises a tool the run does not have (ENG-198).""" + +from backend.apps.agents.browser import browser_schema +from backend.apps.agents.browser.browser_loop import ( + LOOP_WARNING_TEXT, + STAGNATION_MAX, + stagnation_nudge, +) +from backend.apps.agents.browser.human_intervention_allowed import human_intervention_allowed +from backend.apps.agents.browser.intervention_copy import ( + INTERVENTION_SECTION, + strip_intervention_copy, +) +from backend.apps.agents.browser.seed_playbooks import SEED_PLAYBOOKS +from backend.apps.agents.core.models import AgentSession + + +def p_session(sid: str, parent: str | None = None, workflow: str | None = None) -> AgentSession: + return AgentSession(id=sid, name="t", model="m", mode="browser-agent", + parent_session_id=parent, workflow_run_id=workflow) + + +def test_allowed_by_default(): + assert human_intervention_allowed({}, None, {}) + + +def test_toggle_deny_removes_it(): + assert not human_intervention_allowed({"RequestHumanIntervention": "deny"}, None, {}) + + +def test_ask_policy_keeps_it(): + assert human_intervention_allowed({"RequestHumanIntervention": "ask"}, None, {}) + + +def test_workflow_parent_removes_it(): + sessions = {"p": p_session("p", workflow="wf1")} + assert not human_intervention_allowed({}, "p", sessions) + + +def test_workflow_grandparent_removes_it_through_the_chain(): + sessions = {"p": p_session("p", parent="gp"), "gp": p_session("gp", workflow="wf1")} + assert not human_intervention_allowed({}, "p", sessions) + + +def test_chat_parent_keeps_it(): + sessions = {"p": p_session("p")} + assert human_intervention_allowed({}, "p", sessions) + + +def test_unknown_parent_degrades_to_allowed(): + assert human_intervention_allowed({}, "ghost", {}) + + +def test_a_parent_chain_cycle_cannot_hang(): + sessions = {"a": p_session("a", parent="b"), "b": p_session("b", parent="a")} + assert human_intervention_allowed({}, "a", sessions) + + +def test_intervention_section_is_really_in_the_system_prompt(): + # strip_intervention_copy works by exact replace; if the prompt copy drifts away from the + # constant, this is the test that fails before the scrub silently stops matching. + assert INTERVENTION_SECTION in browser_schema.SYSTEM_PROMPT + + +def test_scrubbed_copy_never_names_the_tool(): + seeds = [line for lines in SEED_PLAYBOOKS.values() for line in lines] + for text in ( + browser_schema.SYSTEM_PROMPT, + browser_schema.APP_SYSTEM_PROMPT, + LOOP_WARNING_TEXT.format(count=3), + stagnation_nudge(STAGNATION_MAX), + "šŸ’” Suggested next step: call RequestHumanIntervention now.", + *seeds, + ): + assert "RequestHumanIntervention" not in strip_intervention_copy(text) + + +def test_interactive_tool_list_still_offers_it(): + names = [t["name"] for t in browser_schema.MODEL_VISIBLE_TOOLS] + assert "RequestHumanIntervention" in names