[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

@@ -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}"}
                                 ]

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

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
 

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`, "

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

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

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):

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")

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