259 lines
13 KiB
Python
259 lines
13 KiB
Python
"""Goal execution preserves pinned Browser Use lifecycle without platform/model calls."""
|
|
|
|
import asyncio
|
|
import math
|
|
from types import MethodType, SimpleNamespace
|
|
from unittest.mock import AsyncMock, Mock
|
|
|
|
import pytest
|
|
from browser_use import ActionResult, Agent
|
|
from browser_use.agent.message_manager.service import MessageManager
|
|
from browser_use.agent.message_manager.views import HistoryItem, MessageManagerState
|
|
from browser_use.agent.views import AgentStepInfo, MessageCompactionSettings
|
|
|
|
from backend.browser import runner, tools
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_runner(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(runner, "task_doc_dir", lambda _task_id: tmp_path)
|
|
monkeypatch.setattr(tools, "task_doc_dir", lambda _task_id: tmp_path)
|
|
monkeypatch.setattr(runner, "append_progress", lambda *args, **kwargs: None)
|
|
task_id = "goal_runner_fixture"
|
|
tools.clear_task_context(task_id)
|
|
runner._stalled_tasks.discard(task_id)
|
|
yield task_id, tmp_path
|
|
runner._live.pop(task_id, None)
|
|
runner._stalled_tasks.discard(task_id)
|
|
tools.clear_task_context(task_id)
|
|
|
|
|
|
async def test_unlimited_context_keeps_real_step_for_pinned_message_compaction():
|
|
goal_class = runner._prospecting_agent_class(Agent)
|
|
agent = object.__new__(goal_class)
|
|
agent._goal_based = True
|
|
agent.state = SimpleNamespace(n_steps=70, last_model_output=None, last_result=[])
|
|
summary = SimpleNamespace(screenshot=None, url="https://www.xiaohongshu.com/explore")
|
|
agent.browser_session = SimpleNamespace(
|
|
id="local-fixture", agent_focus_target_id=None,
|
|
get_browser_state_summary=AsyncMock(return_value=summary),
|
|
)
|
|
manager = object.__new__(MessageManager)
|
|
manager.state = MessageManagerState(agent_history_items=[HistoryItem(memory="existing successful page observations")])
|
|
manager.sensitive_data = None
|
|
manager.prepare_step_state = Mock()
|
|
manager.create_state_messages = Mock()
|
|
manager._add_context_message = Mock()
|
|
agent._message_manager = manager
|
|
agent.llm = SimpleNamespace(ainvoke=AsyncMock(return_value=SimpleNamespace(completion="compressed prior observations")))
|
|
agent.settings = SimpleNamespace(
|
|
message_compaction=MessageCompactionSettings(compact_every_n_steps=1, trigger_char_count=1),
|
|
page_extraction_llm=None, use_vision=False,
|
|
)
|
|
agent.include_recent_events = False
|
|
agent.sensitive_data = None
|
|
agent.available_file_paths = []
|
|
agent.skill_service = None
|
|
agent.tools = SimpleNamespace(registry=SimpleNamespace(get_prompt_description=Mock(return_value="")))
|
|
agent._check_and_update_downloads = AsyncMock()
|
|
agent._log_step_context = Mock()
|
|
agent._check_stop_or_pause = AsyncMock()
|
|
agent._update_action_models_for_page = AsyncMock()
|
|
agent._render_plan_description = Mock(return_value=None)
|
|
agent._inject_replan_nudge = Mock()
|
|
agent._inject_exploration_nudge = Mock()
|
|
agent._update_loop_detector_page_state = Mock()
|
|
agent._inject_loop_detection_nudge = Mock()
|
|
agent._force_done_after_failure = AsyncMock()
|
|
|
|
step = AgentStepInfo(step_number=69, max_steps=math.inf)
|
|
assert await agent._prepare_context(step) is summary
|
|
assert manager.state.compaction_count == 1
|
|
assert manager.state.last_compaction_step == 69
|
|
assert manager.state.compacted_memory == "compressed prior observations"
|
|
# The real pinned preparation path must not expose a finite step horizon
|
|
# or budget instruction while forwarding the step to compaction.
|
|
assert manager.create_state_messages.call_args.kwargs["step_info"] is None
|
|
manager._add_context_message.assert_not_called()
|
|
agent.llm.ainvoke.assert_awaited_once()
|
|
|
|
|
|
def test_successful_completion_does_not_inherit_old_recovered_rate_limit():
|
|
compact = {"is_done": True, "is_successful": True,
|
|
"errors": ["Error code: 429 - retry later"], "final_result": "阶段结果已保存"}
|
|
assert runner.stop_reason_from_history(compact, None) == "agent_done"
|
|
assert runner.verdict_from_history(compact, None) == ("success", "阶段结果已保存")
|
|
|
|
|
|
def test_unsuccessful_done_preserves_failure_in_stop_reason():
|
|
compact = {"is_done": True, "is_successful": False, "errors": [], "final_result": "仅保存部分结果，尚有路线未执行"}
|
|
assert runner.verdict_from_history(compact, None) == ("failed", compact["final_result"])
|
|
assert runner.stop_reason_from_history(compact, None) == "agent_reported_failure"
|
|
|
|
|
|
async def test_people_target_checkpoint_stops_before_another_browser_attempt(isolated_runner):
|
|
task_id, directory = isolated_runner
|
|
class LocalAgent:
|
|
def __init__(self):
|
|
self.state = SimpleNamespace(n_steps=1, stopped=False)
|
|
self.calls = 0
|
|
async def run(self, max_steps, on_step_start, on_step_end):
|
|
while not self.state.stopped:
|
|
await on_step_start(self)
|
|
self.calls += 1
|
|
self.state.n_steps += 1
|
|
await on_step_end(self)
|
|
assert self.calls < 3
|
|
return SimpleNamespace(number_of_steps=lambda: self.calls, is_done=lambda: False,
|
|
is_successful=lambda: None, final_result=lambda: None, urls=lambda: [],
|
|
action_names=lambda: [], errors=lambda: [], total_duration_seconds=lambda: 0)
|
|
checkpoint = AsyncMock(return_value={"code": "target_reached", "current": 1, "target": 1, "detail": "达到1位目标"})
|
|
agent = LocalAgent()
|
|
result = await runner._run_agent(task_id, runner.LiveSession(agent, None), agent, None,
|
|
keep_session=True, on_evidence_checkpoint=checkpoint)
|
|
assert result["verdict"] == "success" and result["stop_reason"] == "target_reached"
|
|
assert agent.calls == result["history"]["steps_this_run"] == 1
|
|
assert runner.read_browser_budget(task_id)["total_steps_used"] == 1
|
|
assert result["history"]["is_done"] is False
|
|
|
|
|
|
async def test_unlimited_runs_and_resumes_preserve_attempts_beyond_legacy_ceiling(isolated_runner):
|
|
task_id, directory = isolated_runner
|
|
tools._write_json_atomic(directory / "browser-budget.json", {"total_steps_used": 61, "last_run_steps": 11})
|
|
|
|
class LocalAgent:
|
|
def __init__(self):
|
|
self.state = SimpleNamespace(n_steps=1)
|
|
self.calls = 0
|
|
|
|
async def run(self, max_steps, on_step_start):
|
|
assert math.isinf(max_steps)
|
|
self.calls += 1
|
|
for _ in range(3 if self.calls == 1 else 2):
|
|
await on_step_start(self)
|
|
self.state.n_steps += 1
|
|
return SimpleNamespace(
|
|
number_of_steps=lambda: self.state.n_steps - 1,
|
|
is_done=lambda: True, is_successful=lambda: True, final_result=lambda: "阶段完成",
|
|
urls=lambda: [], action_names=lambda: [], errors=lambda: [], total_duration_seconds=lambda: 0,
|
|
)
|
|
|
|
agent = LocalAgent()
|
|
session = runner.LiveSession(agent, None)
|
|
first = await runner._run_agent(task_id, session, agent, None, keep_session=True, budget_steps_used=61)
|
|
second = await runner._run_agent(task_id, session, agent, None, keep_session=True)
|
|
assert first["history"]["steps_this_run"] == 3 and first["history"]["total_steps_used"] == 64
|
|
assert second["history"]["steps_this_run"] == 2 and second["history"]["total_steps_used"] == 66
|
|
record = runner.read_browser_budget(task_id)
|
|
assert record["total_steps_used"] == 66 and record["max_total_steps"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("evidence_error", [False, True])
|
|
async def test_stall_stops_in_pinned_step_before_repeating_action_without_phantom_attempt(isolated_runner, monkeypatch, evidence_error):
|
|
task_id, directory = isolated_runner
|
|
goal_class = runner._prospecting_agent_class(Agent)
|
|
agent = object.__new__(goal_class)
|
|
agent.state = SimpleNamespace(n_steps=1, stopped=False, paused=False, consecutive_failures=0,
|
|
last_model_output=None, last_result=None)
|
|
agent._external_pause_event = asyncio.Event()
|
|
agent._demo_mode_enabled = False
|
|
agent.register_should_stop_callback = None
|
|
agent.register_external_agent_status_raise_error_callback = None
|
|
agent.register_done_callback = None
|
|
agent.browser_session = SimpleNamespace(id="fixture", agent_focus_target_id=None,
|
|
wait_if_captcha_solving=AsyncMock(return_value=None))
|
|
agent.settings = SimpleNamespace(step_timeout=2, use_judge=False, save_conversation_path=None)
|
|
agent.register_new_step_callback = runner._make_step_callback(task_id, directory)
|
|
summary = SimpleNamespace(screenshot=None, url="https://www.xiaohongshu.com/explore/abc123" if evidence_error else "https://www.xiaohongshu.com/explore",
|
|
dom_state=SimpleNamespace(llm_representation=lambda: "unchanged visible page"))
|
|
action = SimpleNamespace(model_dump=lambda **_kwargs: {"inspect_source_candidates": {"max_comments": 5}})
|
|
output = SimpleNamespace(action=[action], next_goal="检查新来源", evaluation_previous_goal="无新增")
|
|
history_items = []
|
|
agent.history = SimpleNamespace(
|
|
history=history_items, is_done=lambda: False, number_of_steps=lambda: len(history_items),
|
|
is_successful=lambda: False, final_result=lambda: None,
|
|
urls=lambda: [], action_names=lambda: [], errors=lambda: [], total_duration_seconds=lambda: 0,
|
|
)
|
|
agent._prepare_context = AsyncMock(return_value=summary)
|
|
agent._message_manager = SimpleNamespace(last_state_message_text="local fixture")
|
|
model_calls = 0
|
|
|
|
async def next_action(self, state):
|
|
nonlocal model_calls
|
|
model_calls += 1
|
|
self.state.last_model_output = output
|
|
# Preserve the pinned post-model callback and its subsequent stop
|
|
# check; only the external model response is replaced.
|
|
await self._handle_post_llm_processing(state, [])
|
|
await self._check_stop_or_pause()
|
|
|
|
agent._get_next_action = MethodType(next_action, agent)
|
|
|
|
async def execute_actions():
|
|
agent.state.last_result = [ActionResult(error="Error executing action save_source: 原文不属于当前容器")] if evidence_error else [ActionResult(extracted_content="no new source")]
|
|
|
|
agent._execute_actions = AsyncMock(side_effect=execute_actions)
|
|
agent._post_process = AsyncMock()
|
|
|
|
async def make_history(*_args, **kwargs):
|
|
history_items.append(SimpleNamespace(metadata=kwargs.get("metadata"), result=list(agent.state.last_result or []), state=summary))
|
|
|
|
agent._make_history_item = AsyncMock(side_effect=make_history)
|
|
agent._log_step_completion_summary = Mock(return_value=None)
|
|
agent.save_file_system_state = Mock()
|
|
agent.eventbus = SimpleNamespace(dispatch=Mock())
|
|
monkeypatch.setattr("browser_use.agent.service.CreateAgentStepEvent.from_agent_step", Mock(return_value=object()))
|
|
|
|
async def local_run(self, max_steps, on_step_start, on_step_end):
|
|
while not self.state.stopped:
|
|
assert model_calls < 10, "stall did not terminate the pinned step path"
|
|
await self._execute_step(self.state.n_steps - 1, max_steps,
|
|
AgentStepInfo(self.state.n_steps - 1, max_steps), on_step_start, on_step_end)
|
|
return self.history
|
|
|
|
agent.run = MethodType(local_run, agent)
|
|
tools.set_task_context(task_id)
|
|
runner._live[task_id] = session = runner.LiveSession(agent, None, tool_profile="prospecting")
|
|
result = await runner._run_agent(task_id, session, agent, None)
|
|
assert result["stop_reason"] == ("evidence_stalled" if evidence_error else "no_progress")
|
|
assert model_calls == (3 if evidence_error else 6)
|
|
assert agent._execute_actions.await_count == (3 if evidence_error else 5)
|
|
assert result["history"]["steps_this_run"] == model_calls
|
|
assert runner.read_browser_budget(task_id)["total_steps_used"] == model_calls
|
|
|
|
|
|
async def test_new_session_hydrates_persisted_evidence_and_action_ledger(isolated_runner, monkeypatch):
|
|
import browser_use
|
|
|
|
task_id, _directory = isolated_runner
|
|
saved = {"sources": [{"source_id": "source-a", "content": "existing source"}],
|
|
"profiles": [{"source_id": "source-a", "xiaohongshu_id": "visible-account"}],
|
|
"reply_expansions": [{"status": "expanded"}],
|
|
"action_ledger": [{"action_id": "already-sent", "status": "success"}]}
|
|
|
|
class LocalBrowser:
|
|
def __init__(self, **_kwargs):
|
|
pass
|
|
|
|
class LocalAgent:
|
|
def __init__(self, **_kwargs):
|
|
self.state = SimpleNamespace(n_steps=1)
|
|
|
|
async def inspect_hydration(*_args, **_kwargs):
|
|
assert tools.get_task_artifacts(task_id) == saved
|
|
return {"verdict": "success", "artifacts": tools.get_task_artifacts(task_id)}
|
|
|
|
monkeypatch.setattr(browser_use, "Agent", LocalAgent)
|
|
monkeypatch.setattr(browser_use, "Browser", LocalBrowser)
|
|
monkeypatch.setattr(runner, "embedded_mode", lambda: False)
|
|
monkeypatch.setattr(runner, "_dock_bounds", None)
|
|
monkeypatch.setattr(runner, "get_llm_config", lambda: {"provider": "fixture", "model": "fixture"})
|
|
monkeypatch.setattr(runner, "create_llm", lambda: object())
|
|
monkeypatch.setattr(runner, "use_vision_for_model", lambda *_args: False)
|
|
monkeypatch.setattr(runner, "_run_agent", inspect_hydration)
|
|
await runner.run_browser_step(task_id=task_id, agent_task="只读夹具", profile_path="/fixture/profile",
|
|
allowed_domains=["www.xiaohongshu.com"], tool_profile="prospecting",
|
|
max_steps=None, prospecting_artifacts=saved)
|
|
saved["sources"].clear()
|
|
assert len(tools.get_task_artifacts(task_id)["sources"]) == 1
|