172 lines
8.3 KiB
Python
172 lines
8.3 KiB
Python
"""Exercise the pinned multi_act loop, including its inter-action await."""
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from browser_use import ActionResult, Agent
|
|
|
|
from backend.browser import runner
|
|
|
|
|
|
NOTE = 'https://www.xiaohongshu.com/explore/'+'a'*24
|
|
|
|
|
|
def action(name, **values):
|
|
return SimpleNamespace(model_dump=lambda **kwargs: {name: values})
|
|
|
|
|
|
def make_agent(*, first_effect=None, terminates=False, wait=.02):
|
|
browser = SimpleNamespace(id='local-fixture', agent_focus_target_id='managed', url=NOTE,
|
|
_cached_browser_state_summary=None, browser_profile=SimpleNamespace(wait_between_actions=wait))
|
|
async def info(params):
|
|
assert params == {'targetId': 'managed'}
|
|
return {'targetInfo': {'targetId': 'managed', 'type': 'page', 'url': browser.url}}
|
|
async def focus(target_id, focus):
|
|
assert (target_id, focus) == ('managed', True)
|
|
browser.agent_focus_target_id = target_id
|
|
return object()
|
|
async def dispatch(event):
|
|
browser.agent_focus_target_id = event.target_id
|
|
async def current_url():
|
|
return browser.url if browser.agent_focus_target_id == 'managed' else 'app://ui/index.html'
|
|
browser.cdp_client = SimpleNamespace(send=SimpleNamespace(Target=SimpleNamespace(getTargetInfo=info)))
|
|
browser.get_or_create_cdp_session = focus
|
|
browser.event_bus = SimpleNamespace(dispatch=dispatch)
|
|
browser.get_current_page_url = current_url
|
|
session = runner.LiveSession(None, browser, embedded=True, target_id='managed',
|
|
allowed_domains=('www.xiaohongshu.com',), last_verified_url=NOTE)
|
|
agent = object.__new__(runner._prospecting_agent_class(Agent))
|
|
agent.browser_session = browser
|
|
agent._managed_session = session
|
|
agent._managed_observed_url = NOTE
|
|
agent.state = SimpleNamespace(n_steps=1, stopped=False)
|
|
agent.file_system = None
|
|
agent.settings = SimpleNamespace(page_extraction_llm=None)
|
|
agent.sensitive_data = None
|
|
agent.available_file_paths = []
|
|
agent.extraction_schema = None
|
|
agent._check_stop_or_pause = AsyncMock()
|
|
agent._log_action = AsyncMock()
|
|
agent._demo_mode_log = AsyncMock()
|
|
calls = []
|
|
first_finished = asyncio.Event()
|
|
async def act(**kwargs):
|
|
calls.append((kwargs['action'].model_dump(), browser.agent_focus_target_id))
|
|
if len(calls) == 1:
|
|
if first_effect:
|
|
await first_effect(browser)
|
|
first_finished.set()
|
|
return ActionResult(extracted_content='local observation only')
|
|
original = SimpleNamespace(act=act, registry=SimpleNamespace(registry=SimpleNamespace(actions={
|
|
'inspect_source_candidates': SimpleNamespace(terminates_sequence=terminates),
|
|
'send_keys': SimpleNamespace(terminates_sequence=False), 'scroll': SimpleNamespace(terminates_sequence=False)})))
|
|
agent.tools = original
|
|
checks = []
|
|
async def guard():
|
|
checks.append(browser.agent_focus_target_id)
|
|
await runner._focus_managed_target('local-fixture', session)
|
|
agent._managed_target_guard = guard
|
|
return agent, browser, original, calls, first_finished, checks
|
|
|
|
|
|
@pytest.mark.parametrize('second', [action('send_keys', keys='End'), action('scroll', down=True, num_pages=1)])
|
|
async def test_focus_drift_during_upstream_inter_action_sleep_never_executes_index_free_second_action(second):
|
|
agent, browser, original, calls, finished, checks = make_agent()
|
|
async def drift():
|
|
await finished.wait()
|
|
await asyncio.sleep(.005)
|
|
browser.agent_focus_target_id = 'ui-target'
|
|
drifting = asyncio.create_task(drift())
|
|
results = await agent.multi_act([action('inspect_source_candidates', max_comments=1), second,
|
|
action('send_keys', keys='Home')])
|
|
await drifting
|
|
assert len(calls) == 1 and calls[0][1] == 'managed'
|
|
assert len(results) == 2 and results[0].error is None
|
|
assert '旧动作未执行' in results[1].error
|
|
assert 'ui-target' in checks # Rechecked after upstream's await, before act.
|
|
assert browser.agent_focus_target_id == 'managed'
|
|
assert agent.tools is original
|
|
|
|
|
|
async def test_url_drift_during_batch_sleep_also_discards_remaining_old_actions():
|
|
agent, browser, original, calls, finished, _ = make_agent()
|
|
async def drift():
|
|
await finished.wait()
|
|
await asyncio.sleep(.005)
|
|
browser.url = NOTE.replace('a'*24, 'b'*24)
|
|
drifting = asyncio.create_task(drift())
|
|
results = await agent.multi_act([action('inspect_source_candidates', max_comments=1), action('scroll', down=True)])
|
|
await drifting
|
|
assert len(calls) == 1 and len(results) == 2 and results[-1].error
|
|
assert agent.tools is original
|
|
|
|
|
|
async def test_unchanged_batch_retains_the_original_upstream_loop_and_all_results():
|
|
agent, _, original, calls, _, checks = make_agent(wait=0)
|
|
results = await agent.multi_act([action('inspect_source_candidates', max_comments=1), action('scroll', down=True),
|
|
action('send_keys', keys='Home')])
|
|
assert len(calls) == len(results) == 3
|
|
assert len(checks) == 4 # Batch entry plus every actual action boundary.
|
|
assert all(result.error is None for result in results)
|
|
assert agent.tools is original
|
|
|
|
|
|
async def test_upstream_terminates_sequence_metadata_is_still_obeyed():
|
|
agent, _, original, calls, _, checks = make_agent(terminates=True)
|
|
results = await agent.multi_act([action('inspect_source_candidates', max_comments=1), action('send_keys', keys='End')])
|
|
assert len(calls) == len(results) == 1 and len(checks) == 2
|
|
assert agent.tools is original
|
|
|
|
|
|
async def test_action_ledger_counts_only_invoked_tools_not_remaining_model_proposals(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(runner.diagnostics, 'task_doc_dir', lambda _: tmp_path)
|
|
agent, _, _, calls, _, _ = make_agent(terminates=True)
|
|
with runner.diagnostics.run_scope('baseline-fixture', 'baseline-run'):
|
|
await agent.multi_act([action('inspect_source_candidates', max_comments=1),
|
|
action('send_keys', keys='private-parameter')])
|
|
rows = [json.loads(line) for line in (tmp_path/'browser-actions.jsonl').read_text().splitlines()]
|
|
assert len(calls) == 1
|
|
assert [row['status'] for row in rows] == ['invoked', 'returned']
|
|
assert {row['action_name'] for row in rows} == {'inspect_source_candidates'}
|
|
assert rows[0]['action_id'] == rows[1]['action_id']
|
|
assert 'private-parameter' not in (tmp_path/'browser-actions.jsonl').read_text()
|
|
|
|
|
|
async def test_action_ledger_distinguishes_rejected_before_invocation(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(runner.diagnostics, 'task_doc_dir', lambda _: tmp_path)
|
|
delegate = SimpleNamespace(act=AsyncMock())
|
|
async def reject(_):
|
|
return ActionResult(error='fixture rejected')
|
|
boundary = runner._ManagedActionBoundary(delegate, reject)
|
|
with runner.diagnostics.run_scope('baseline-fixture', 'baseline-run'):
|
|
result = await boundary.act(action=action('scroll', secret='private-parameter'))
|
|
rows = [json.loads(line) for line in (tmp_path/'browser-actions.jsonl').read_text().splitlines()]
|
|
assert result.error and [row['status'] for row in rows] == ['blocked']
|
|
delegate.act.assert_not_awaited()
|
|
assert 'private-parameter' not in (tmp_path/'browser-actions.jsonl').read_text()
|
|
|
|
|
|
async def test_upstream_runtime_page_change_guard_is_still_obeyed():
|
|
async def change(browser):
|
|
browser.url = NOTE.replace('a'*24, 'b'*24)
|
|
agent, _, original, calls, _, checks = make_agent(first_effect=change)
|
|
results = await agent.multi_act([action('inspect_source_candidates', max_comments=1), action('send_keys', keys='End')])
|
|
assert len(calls) == len(results) == 1 and len(checks) == 2
|
|
assert agent.tools is original
|
|
|
|
|
|
async def test_cancellation_restores_tools_without_running_a_later_action():
|
|
entered = asyncio.Event()
|
|
async def blocked(_browser):
|
|
entered.set()
|
|
await asyncio.Event().wait()
|
|
agent, _, original, calls, _, _ = make_agent(first_effect=blocked)
|
|
job = asyncio.create_task(agent.multi_act([action('inspect_source_candidates', max_comments=1), action('send_keys', keys='End')]))
|
|
await entered.wait()
|
|
job.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await job
|
|
assert len(calls) == 1 and agent.tools is original
|