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