299 lines
16 KiB
Python
299 lines
16 KiB
Python
"""Actual runner branches and pinned failure handoff, without browser/model I/O."""
|
|
from copy import deepcopy
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
from pydantic import BaseModel, ValidationError
|
|
from browser_use import Agent
|
|
|
|
from backend.browser import diagnostics, runner, tools
|
|
from backend.browser.progress_guard import ProgressGuard
|
|
|
|
|
|
def state(index=1, profile="a" * 24, *, path="/explore/" + "b" * 24, dom=""):
|
|
node = SimpleNamespace(attributes={"href": "/user/profile/" + profile}, node_name="A", node_value="fixture author", parent_node=None)
|
|
return SimpleNamespace(url="https://www.xiaohongshu.com" + path + "?xsec_token=PRIVATE_TOKEN",
|
|
dom_state=SimpleNamespace(selector_map={index: node}, llm_representation=lambda: dom))
|
|
|
|
|
|
def output(name="click", **args):
|
|
return SimpleNamespace(action=[SimpleNamespace(model_dump=lambda **_kwargs: {name: args})])
|
|
|
|
|
|
def test_same_author_hover_loop_ignores_temporary_indices_dom_and_signed_url():
|
|
guard = ProgressGuard()
|
|
artifacts = {"sources": [{"source_id": "one", "author": "fixture author"}]}
|
|
for step in range(5):
|
|
page = state(step + 1, dom=f"[{step + 1}] hover style changed {step}")
|
|
page.url += str(step)
|
|
assert not guard.observe(page, output(index=step + 1), artifacts)
|
|
assert guard.observe(state(900, dom="entirely new transient hover DOM"), output(index=900), artifacts)
|
|
|
|
|
|
@pytest.mark.parametrize("progress", ["source", "author_link", "profile", "navigation", "page", "scroll"])
|
|
def test_real_progress_or_navigation_resets_same_operation_stagnation(progress):
|
|
guard = ProgressGuard(repeats=2)
|
|
page = state()
|
|
artifacts = {"sources": [{"source_id": "one"}]}
|
|
assert not guard.observe(page, output(index=1), artifacts)
|
|
assert not guard.observe(page, output(index=1), artifacts)
|
|
changed = deepcopy(artifacts)
|
|
if progress == "source":
|
|
changed["sources"].append({"source_id": "two"})
|
|
elif progress == "author_link":
|
|
changed["sources"][0]["author_profile_url"] = "https://www.xiaohongshu.com/user/profile/" + "a" * 24
|
|
elif progress == "profile":
|
|
changed["profiles"] = [{"source_id": "one", "xiaohongshu_id": "observed public account"}]
|
|
elif progress == "navigation":
|
|
changed["navigation_attempts"] = [{"capability": "advance_comments", "object_key": "note", "status": "comments_visible", "post_fingerprint": "new view"}]
|
|
elif progress == "page":
|
|
page = state(path="/user/profile/" + "a" * 24)
|
|
else:
|
|
assert not guard.observe(page, output("scroll", down=True, num_pages=1), changed)
|
|
page.page_info = SimpleNamespace(scroll_x=0, scroll_y=100)
|
|
assert not guard.observe(page, output(index=1), changed)
|
|
|
|
|
|
def test_hover_only_navigation_records_are_not_new_progress_and_other_author_is_different():
|
|
guard = ProgressGuard(repeats=2)
|
|
artifacts = {}
|
|
assert not guard.observe(state(), output(index=1), artifacts)
|
|
artifacts["navigation_attempts"] = [{"capability": "open_source_profile", "status": "hover_only", "post_fingerprint": "popup1"}]
|
|
assert not guard.observe(state(10), output(index=10), artifacts)
|
|
artifacts["navigation_attempts"].append({"capability": "open_source_profile", "status": "hover_only", "post_fingerprint": "popup2"})
|
|
assert guard.observe(state(20), output(index=20), artifacts)
|
|
assert not guard.observe(state(profile="c" * 24), output(index=1), artifacts)
|
|
|
|
|
|
def test_search_keyword_change_is_real_navigation_but_new_signatures_are_not():
|
|
guard = ProgressGuard(repeats=1)
|
|
first, second = state(path="/search_result_ai"), state(path="/search_result_ai")
|
|
first.url += "&keyword=one"
|
|
second.url += "&keyword=two"
|
|
assert not guard.observe(first, output("inspect_source_candidates", max_comments=5), {})
|
|
assert not guard.observe(second, output("inspect_source_candidates", max_comments=5), {})
|
|
|
|
|
|
@pytest.mark.parametrize("between", ["wait", "inspect_source_candidates"])
|
|
def test_wait_or_observe_between_same_author_attempts_does_not_erase_loop(between):
|
|
guard = ProgressGuard(repeats=2)
|
|
assert not guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(), output(between), {})
|
|
assert not guard.observe(state(22), output(index=22), {})
|
|
assert not guard.observe(state(), output(between), {})
|
|
assert guard.observe(state(99), output(index=99), {})
|
|
assert not guard.observe(state(), output("scroll", down=True), {})
|
|
moved = state()
|
|
moved.page_info = SimpleNamespace(scroll_x=0, scroll_y=100)
|
|
assert not guard.observe(moved, output(index=1), {})
|
|
|
|
|
|
def test_requested_but_unobserved_scroll_does_not_erase_same_author_loop():
|
|
guard = ProgressGuard(repeats=2)
|
|
assert not guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(), output("scroll", down=True), {})
|
|
assert not guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(), output("scroll", down=True), {})
|
|
assert guard.observe(state(), output(index=1), {})
|
|
|
|
|
|
@pytest.mark.parametrize("between,args", [("go_back", {}), ("navigate", {"url": "https://www.xiaohongshu.com/explore/" + "b" * 24})])
|
|
def test_unmoved_navigation_request_does_not_erase_same_author_loop(between, args):
|
|
guard = ProgressGuard(repeats=2)
|
|
assert not guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(), output(between, **args), {})
|
|
assert not guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(), output(between, **args), {})
|
|
assert guard.observe(state(), output(index=1), {})
|
|
assert not guard.observe(state(path="/explore/" + "d" * 24), output(index=1), {})
|
|
|
|
|
|
@pytest.mark.parametrize("real_expansion", [False, True])
|
|
def test_only_program_confirmed_reply_expansion_resets_progress(real_expansion):
|
|
guard = ProgressGuard(repeats=2)
|
|
artifacts = {}
|
|
assert not guard.observe(state(), output(index=1), artifacts)
|
|
artifacts["reply_expansions"] = [{"status": "no_new_content", "new_visible_replies": 0}]
|
|
assert not guard.observe(state(), output(index=1), artifacts)
|
|
artifacts["reply_expansions"].append({"status": "state_changed", "parent_scope_checked": True, "new_visible_replies": 2})
|
|
artifacts["operation_evidence"] = [{"id": "real_delta", "capability": "expand_replies", "outcome": "success" if real_expansion else "unknown"}]
|
|
assert guard.observe(state(), output(index=1), artifacts) is not real_expansion
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated(tmp_path, monkeypatch):
|
|
tid = "runtime_d38_fixture"
|
|
for module in (runner, tools, diagnostics):
|
|
monkeypatch.setattr(module, "task_doc_dir", lambda _tid: tmp_path)
|
|
monkeypatch.setattr(runner, "append_progress", lambda *_args, **_kwargs: None)
|
|
monkeypatch.setattr(runner, "get_llm_config", lambda: {"provider": "fixture", "model": "fixture"})
|
|
tools.clear_task_context(tid)
|
|
yield tid, tmp_path
|
|
tools.clear_task_context(tid)
|
|
runner._live.pop(tid, None)
|
|
runner._stalled_tasks.discard(tid)
|
|
|
|
|
|
async def test_new_embedded_session_reaches_actual_run_context_with_original_goal_version(isolated, monkeypatch):
|
|
import browser_use
|
|
tid, directory = isolated
|
|
class LocalBrowser:
|
|
def __init__(self, **kwargs):
|
|
self.options = kwargs
|
|
self.agent_focus_target_id = "fixture_target"
|
|
self.cdp_client = SimpleNamespace(send=SimpleNamespace(Target=SimpleNamespace(getTargetInfo=AsyncMock(
|
|
return_value={"targetInfo": {"targetId": "fixture_target", "type": "page", "url": "about:blank"}}))))
|
|
async def start(self):
|
|
pass
|
|
async def get_or_create_cdp_session(self, target_id, focus):
|
|
assert target_id == "fixture_target" and focus
|
|
return SimpleNamespace(session_id="fixture", cdp_client=SimpleNamespace(send=SimpleNamespace(
|
|
Emulation=SimpleNamespace(clearDeviceMetricsOverride=AsyncMock()))))
|
|
class LocalAgent:
|
|
def __init__(self, **kwargs):
|
|
self.state = SimpleNamespace(n_steps=1)
|
|
self.history = SimpleNamespace(history=[], number_of_steps=lambda: 1, total_duration_seconds=lambda: 0,
|
|
is_done=lambda: True, is_successful=lambda: True, urls=lambda: [], action_names=lambda: [],
|
|
errors=lambda: [], final_result=lambda: "local complete")
|
|
async def run(self, max_steps, on_step_start, on_step_end):
|
|
await on_step_start(self)
|
|
tools._operation_evidence(tid, "save_source", "note", "fixture_note", "pre", "post", verifier="fixture")
|
|
self.state.n_steps += 1
|
|
await on_step_end(self)
|
|
return self.history
|
|
monkeypatch.setattr(browser_use, "Browser", LocalBrowser)
|
|
monkeypatch.setattr(browser_use, "Agent", LocalAgent)
|
|
monkeypatch.setattr(runner, "embedded_mode", lambda: True)
|
|
monkeypatch.setattr(runner, "_ctrl", AsyncMock(side_effect=lambda method, path, *args:
|
|
{"views": [{"taskId": tid, "targetId": "fixture_target"}]} if path == "/health" else {"target_id": "fixture_target"}))
|
|
monkeypatch.setattr(runner, "create_llm", lambda: object())
|
|
monkeypatch.setattr(runner, "use_vision_for_model", lambda *_args: False)
|
|
monkeypatch.setattr(runner, "build_prospecting_tools", lambda *_args, **_kwargs: object())
|
|
assert tid not in runner._live
|
|
result = await runner.run_browser_step(task_id=tid, agent_task="local fixture", profile_path="fixture", allowed_domains=[],
|
|
max_steps=None, keep_session=True, tool_profile="prospecting", prospecting_artifacts={"_knowledge_goal_version": 2})
|
|
fact = result["artifacts"]["operation_evidence"][0]
|
|
run_id = json.loads((directory / "browser-budget.json").read_text())["last_run_id"]
|
|
assert fact["details"]["trace"] == {"goal_version": 2, "run_id": run_id, "action_id": fact["id"]}
|
|
assert result["history"]["model_diagnostics"]["events"] == []
|
|
|
|
|
|
class ActionSchema(BaseModel):
|
|
index: int
|
|
|
|
|
|
def schema_failure():
|
|
try:
|
|
ActionSchema.model_validate({"index": "private text"})
|
|
except ValidationError as error:
|
|
return error
|
|
|
|
|
|
async def test_pinned_failure_handoff_records_true_consecutive_schema_reason_before_done_resets(isolated):
|
|
tid, _directory = isolated
|
|
goal_class = runner._prospecting_agent_class(Agent)
|
|
agent = object.__new__(goal_class)
|
|
agent.state = SimpleNamespace(consecutive_failures=5)
|
|
agent.settings = SimpleNamespace(max_failures=5, final_response_after_failure=True)
|
|
agent._message_manager = SimpleNamespace(_add_context_message=Mock())
|
|
agent.DoneAgentOutput = object()
|
|
agent._logger = Mock()
|
|
with diagnostics.run_scope(tid, "fixture_run"):
|
|
for _ in range(3):
|
|
diagnostics.schema_error(None, schema_failure(), ActionSchema)
|
|
diagnostics.validated_response() # Successful attempt 18 resets the first group.
|
|
for _ in range(5):
|
|
diagnostics.schema_error(None, schema_failure(), ActionSchema)
|
|
await agent._force_done_after_failure()
|
|
assert agent.AgentOutput is agent.DoneAgentOutput # Real pinned behavior retained.
|
|
assert agent._program_failure_limit == {"reason": "model_schema_error", "consecutive_failures": 5, "max_failures": 5}
|
|
diagnostics.validated_response() # The final done response is valid.
|
|
agent.state.consecutive_failures = 0
|
|
compact = {"is_done": True, "is_successful": False, "program_failure_limit": agent._program_failure_limit}
|
|
assert runner.stop_reason_from_history(compact, None) == "model_schema_error"
|
|
|
|
|
|
async def test_nonconsecutive_schema_errors_never_relabel_generic_failure_limit(isolated):
|
|
tid, _directory = isolated
|
|
class Base:
|
|
async def _force_done_after_failure(self):
|
|
pass
|
|
agent = runner._prospecting_agent_class(Base)()
|
|
agent.state = SimpleNamespace(consecutive_failures=5)
|
|
agent.settings = SimpleNamespace(max_failures=5, final_response_after_failure=True)
|
|
with diagnostics.run_scope(tid, "fixture_run"):
|
|
for _ in range(8):
|
|
diagnostics.schema_error(None, schema_failure(), ActionSchema)
|
|
diagnostics.validated_response()
|
|
await agent._force_done_after_failure()
|
|
assert agent._program_failure_limit["reason"] == "model_failure_limit"
|
|
|
|
|
|
def test_historical_cumulative_error_strings_do_not_infer_failure_threshold():
|
|
assert runner.stop_reason_from_history({"is_done": True, "is_successful": False,
|
|
"errors": ["DeepSeek 返回的动作不符合任务结构，未执行。"] * 8}, None) == "agent_reported_failure"
|
|
assert runner.stop_reason_from_history({"is_done": True, "is_successful": True,
|
|
"program_failure_limit": {"reason": "model_schema_error"}}, None) == "model_schema_error"
|
|
|
|
|
|
@pytest.mark.parametrize("code,reason", [(401, "model_auth"), (402, "model_balance"), (429, "model_rate_limit")])
|
|
def test_specific_current_provider_error_survives_generic_failure_handoff(code, reason):
|
|
compact = {"is_done": True, "is_successful": True, "program_failure_limit": {"reason": "model_failure_limit"},
|
|
"errors_this_run": [f"Error code: {code}"]}
|
|
assert runner.stop_reason_from_history(compact, None) == reason
|
|
compact["program_failure_limit"] = {"reason": "model_schema_error"}
|
|
assert runner.stop_reason_from_history(compact, None) == "model_schema_error"
|
|
|
|
|
|
def test_recovered_http_does_not_override_final_unknown_or_schema_handoff():
|
|
compact = {"is_done": True, "is_successful": False, "program_failure_limit": {"reason": "model_failure_limit"},
|
|
"errors_this_run": ["Error code: 402", "unknown action failure"]}
|
|
assert runner.stop_reason_from_history(compact, None) == "model_failure_limit"
|
|
compact["program_failure_limit"]["cause"] = "model_rate_limit"
|
|
assert runner.stop_reason_from_history(compact, None) == "model_rate_limit"
|
|
|
|
|
|
def test_progress_url_discards_userinfo_signed_query_and_fragment():
|
|
assert runner._progress_url("https://PRIVATE:SECRET@www.xiaohongshu.com/explore/abc?xsec_token=PRIVATE#PRIVATE") == "https://www.xiaohongshu.com/explore/abc"
|
|
assert runner._progress_url("javascript:PRIVATE") == "页面地址未记录"
|
|
|
|
|
|
async def test_run_summary_preserves_failure_handoff_after_final_valid_done(isolated):
|
|
from types import MethodType
|
|
tid, _directory = isolated
|
|
cls = runner._prospecting_agent_class(Agent)
|
|
agent = object.__new__(cls)
|
|
agent.state = SimpleNamespace(n_steps=1, consecutive_failures=0, last_result=[])
|
|
agent.settings = SimpleNamespace(max_failures=5, final_response_after_failure=True)
|
|
agent._logger = Mock()
|
|
agent._message_manager = SimpleNamespace(_add_context_message=Mock())
|
|
agent.DoneAgentOutput = object()
|
|
items = []
|
|
agent.history = SimpleNamespace(history=items, number_of_steps=lambda: len(items), total_duration_seconds=lambda: 0,
|
|
is_done=lambda: True, is_successful=lambda: True, urls=lambda: [], action_names=lambda: ["done"],
|
|
errors=lambda: [], final_result=lambda: "model claimed completion")
|
|
async def local_run(self, max_steps, on_step_start, on_step_end):
|
|
for outcome in ["schema"] * 3 + ["valid"] + ["schema"] * 5 + ["done"]:
|
|
await on_step_start(self)
|
|
await self._force_done_after_failure()
|
|
if outcome == "schema":
|
|
diagnostics.schema_error(None, schema_failure(), ActionSchema)
|
|
self.state.consecutive_failures += 1
|
|
else:
|
|
diagnostics.validated_response()
|
|
self.state.consecutive_failures = 0
|
|
self.state.n_steps += 1
|
|
items.append(SimpleNamespace(result=[], state=SimpleNamespace(url="")))
|
|
await on_step_end(self)
|
|
return self.history
|
|
agent.run = MethodType(local_run, agent)
|
|
session = runner.LiveSession(agent, None, tool_profile="prospecting")
|
|
result = await runner._run_agent(tid, session, agent, None)
|
|
assert result["verdict"] == "failed" and result["stop_reason"] == "model_schema_error"
|
|
assert result["history"]["is_successful"] is True # Raw claim is still available.
|
|
assert result["history"]["steps_this_run"] == 10
|
|
assert result["history"]["program_failure_limit"]["consecutive_failures"] == 5
|
|
assert len(result["history"]["model_diagnostics"]["events"]) == 8
|