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