tests/test_browser_runtime.py

import json
import os
import sys
import tempfile
import threading
import types
import unittest
import asyncio
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

from app.agents.browser_use_bridge import (
    BROWSER_USE_VERSION,
    CAMOUFOX_SETUP_COMMAND,
    CAMOUFOX_VERSION,
    CDP_PROXY_COMMAND,
    REMOTE_CDP_PROXY_PATH,
    REMOTE_STEALTH_SERVER_PATH,
    _create_e2b_desktop,
    _start_local_worker,
    _wait_for_external_cdp,
    _worker_environment,
    build_chrome_command,
    build_e2b_sandbox_kwargs,
    build_public_result,
    build_worker_config,
    cdp_proxy_script_source,
    run_browser_use_agent,
    translate_worker_event,
    worker_script_source,
)
from app.agents.browser_use_worker import (
    _browser_search_urls,
    _normalize_browser_links,
    _normalize_stealth_links,
    _playwright_cookies,
)
import app.agents.cdp_proxy as cdp_proxy
from app.agents.cdp_proxy import _authorized_and_sanitized


MODEL = "nvidia/nemotron-3-ultra-550b-a55b:free"


class _FakeFiles:
    def __init__(self):
        self.data = {}

    def write(self, path, data):
        self.data[path] = data
        return SimpleNamespace(path=path)


class _FakeRemoteHandle:
    def __init__(self):
        self.killed = False

    def kill(self):
        self.killed = True
        return True


class _FakeCommands:
    def __init__(self):
        self.calls = []
        self.handles = []

    def run(self, command, **kwargs):
        self.calls.append((command, kwargs))
        handle = _FakeRemoteHandle()
        self.handles.append(handle)
        return handle


class _FakeDesktop:
    def __init__(self):
        self.files = _FakeFiles()
        self.commands = _FakeCommands()
        self.stream = _FakeStream()
        self.killed = False

    def get_host(self, port):
        return f"{port}-sandbox.e2b.app"

    def kill(self):
        self.killed = True


class _FakeStream:
    def __init__(self):
        self.started = False
        self.stopped = False

    def start(self, **kwargs):
        self.started = True

    def get_auth_key(self):
        return "visual-auth-key"

    def get_url(self, **kwargs):
        return "https://visual.example/session"

    def stop(self):
        self.stopped = True


class _FakeSandbox:
    desktop = None
    create_kwargs = None

    @classmethod
    def create(cls, **kwargs):
        cls.create_kwargs = kwargs
        cls.desktop = _FakeDesktop()
        return cls.desktop


class _FakeLocalProcess:
    def __init__(self, exit_code=0):
        self.returncode = exit_code
        self.terminated = False
        self.killed = False

    async def wait(self):
        return self.returncode

    def terminate(self):
        self.terminated = True
        self.returncode = -15

    def kill(self):
        self.killed = True
        self.returncode = -9


def _successful_events_and_result():
    events = [
        {
            "type": "step",
            "step": 1,
            "url": "https://duckduckgo.com/?q=lulinha",
            "title": "Search",
            "actions": [{"navigate": {"url": "https://source-one.example/profile"}}],
            "next_goal": "Open the first direct source",
        },
        {
            "type": "evidence",
            "record": {
                "claim": "Lulinha is Fábio Luís",
                "exact_quote": "Fábio Luís, conhecido como Lulinha",
                "url": "https://source-one.example/profile",
                "title": "Profile",
                "hostname": "source-one.example",
            },
        },
        {
            "type": "evidence",
            "record": {
                "claim": "A second source identifies the same person",
                "exact_quote": "Fábio Luís Lula da Silva, o Lulinha",
                "url": "https://source-two.example/story",
                "title": "Story",
                "hostname": "source-two.example",
            },
        },
    ]
    result = {
        "ok": True,
        "answer": "Lulinha is Fábio Luís. Sources are linked below.",
        "agent_success": True,
        "is_done": True,
        "urls": [
            "https://duckduckgo.com/?q=lulinha",
            "https://source-one.example/profile",
            "https://source-two.example/story",
        ],
        "steps": 3,
        "duration_seconds": 9.2,
        "evidence": [item["record"] for item in events if item["type"] == "evidence"],
        "model": MODEL,
    }
    return events, result


class _FakeResponse:
    status_code = 200

    def json(self):
        return {
            "webSocketDebuggerUrl": (
                "ws://127.0.0.1:9222/devtools/browser/internal-browser-id"
            )
        }


class _FakeHttpClient:
    calls = []

    def __init__(self, *args, **kwargs):
        pass

    async def __aenter__(self):
        return self

    async def __aexit__(self, *args):
        return False

    async def get(self, url, headers):
        self.calls.append((url, headers))
        return _FakeResponse()


class BrowserUseBridgeTests(unittest.IsolatedAsyncioTestCase):
    def test_host_worker_is_standalone_remote_cdp_code(self):
        source = worker_script_source()
        compile(source, "<browser-use-worker>", "exec")
        self.assertIn("from browser_use import", source)
        self.assertIn("record_evidence", source)
        self.assertIn("final_response_after_failure=True", source)
        self.assertNotIn('            "done",', source)
        self.assertNotIn("finish_research", source)
        self.assertIn("LANCER_CDP_URL", source)
        self.assertIn("x-lancer-cdp-token", source)
        self.assertIn("BrowserProfile", source)
        self.assertIn("block_ip_addresses=True", source)
        self.assertIn("MAX_PAGE_TEXT_CHARS", source)
        self.assertIn(".slice(0, limit)", source)
        self.assertIn("browser_search", source)
        self.assertIn("open_source", source)
        self.assertIn("_inspect_search_page", source)
        self.assertIn('"navigate",', source)
        self.assertIn('"search",', source)
        self.assertIn("enable_planning=False", source)
        self.assertIn("captcha_solver=True", source)
        self.assertIn('llm_kwargs["reasoning_effort"] = "low"', source)
        self.assertIn('llm_kwargs["reasoning_models"] = [worker_model]', source)
        self.assertIn('"max_retries": 3', source)
        self.assertIn("use_thinking=False", source)
        self.assertNotIn("SearXNG", source)
        self.assertNotIn("searxng", source.casefold())
        self.assertIn("raw_page_text = await page.evaluate", source)
        self.assertIn("quote_normalized in page_text_normalized", source)
        self.assertNotIn("Fuzzy fallback", source)
        self.assertNotIn("match_ratio", source)
        self.assertNotIn("executable_path=", source)
        self.assertIn("Camoufox", source)
        self.assertIn("stealth_gateway_url", source)
        self.assertIn("captcha_solved", source)

    def test_camoufox_snapshot_links_and_cookies_are_sanitized(self):
        links = _normalize_stealth_links(
            {
                "elements": [
                    {
                        "text": "Search link",
                        "href": "https://search.brave.com/help",
                    },
                    {
                        "text": "Direct source",
                        "href": "https://news.example.org/article",
                    },
                ]
            }
        )
        self.assertEqual(
            links,
            [{"title": "Direct source", "url": "https://news.example.org/article"}],
        )
        cookies = _playwright_cookies(
            [
                {
                    "name": "clearance",
                    "value": "ok",
                    "domain": ".example.org",
                    "path": "/",
                    "sameSite": "lax",
                    "unsupported": "discard-me",
                }
            ]
        )
        self.assertEqual(cookies[0]["sameSite"], "Lax")
        self.assertNotIn("unsupported", cookies[0])

    def test_live_browser_search_filters_serp_links_and_unwraps_targets(self):
        engines = dict(_browser_search_urls("quem é lulinha?"))
        self.assertIn("search.brave.com", engines["Brave Search"])
        self.assertNotIn("google.com", " ".join(engines.values()))

        links = _normalize_browser_links(
            [
                {"title": "Search navigation", "url": "https://search.brave.com/help"},
                {
                    "title": "Direct article",
                    "url": "https://noticias.example.org/reportagem",
                },
                {
                    "title": "Wrapped article",
                    "url": (
                        "https://duckduckgo.com/l/?uddg="
                        "https%3A%2F%2Fsecond.example.com%2Fprofile"
                    ),
                },
                {
                    "title": "Yahoo wrapped article",
                    "url": (
                        "https://r.search.yahoo.com/_ylt=x/RV=2/RU="
                        "https%3A%2F%2Fthird.example.net%2Fbio/RK=2/RS=x"
                    ),
                },
            ]
        )
        self.assertEqual(
            [item["url"] for item in links],
            [
                "https://noticias.example.org/reportagem",
                "https://second.example.com/profile",
                "https://third.example.net/bio",
            ],
        )

    def test_proxy_is_dependency_free_authenticated_and_rewrites_host(self):
        source = cdp_proxy_script_source()
        compile(source, "<cdp-proxy>", "exec")
        request = (
            b"GET /json/version HTTP/1.1\r\n"
            b"Host: 9223-public.e2b.app\r\n"
            b"X-Lancer-CDP-Token: correct-token\r\n"
            b"Connection: Upgrade\r\n\r\n"
        )
        sanitized = _authorized_and_sanitized(request, "correct-token")
        self.assertIsNotNone(sanitized)
        self.assertIn(b"Host: 127.0.0.1:9222", sanitized)
        self.assertNotIn(b"correct-token", sanitized)
        self.assertIsNone(_authorized_and_sanitized(request, "wrong-token"))

    async def test_proxy_tunnels_websocket_bytes_after_authenticated_upgrade(self):
        captured = {}
        upstream_done = asyncio.Event()
        proxy_done = asyncio.Event()

        async def fake_chrome(reader, writer):
            try:
                captured["headers"] = await reader.readuntil(b"\r\n\r\n")
                writer.write(
                    b"HTTP/1.1 101 Switching Protocols\r\n"
                    b"Upgrade: websocket\r\n"
                    b"Connection: Upgrade\r\n\r\n"
                )
                await writer.drain()
                payload = await reader.readexactly(4)
                writer.write(payload.upper())
                await writer.drain()
            finally:
                writer.close()
                await writer.wait_closed()
                upstream_done.set()

        async def proxy_connection(reader, writer):
            try:
                await cdp_proxy._handle_client(reader, writer)
            finally:
                proxy_done.set()

        upstream = await asyncio.start_server(fake_chrome, "127.0.0.1", 0)
        upstream_port = upstream.sockets[0].getsockname()[1]
        with (
            patch.object(cdp_proxy, "TOKEN", "correct-token"),
            patch.object(cdp_proxy, "UPSTREAM_PORT", upstream_port),
        ):
            proxy = await asyncio.start_server(
                proxy_connection,
                "127.0.0.1",
                0,
            )
            proxy_port = proxy.sockets[0].getsockname()[1]
            reader, writer = await asyncio.open_connection("127.0.0.1", proxy_port)
            writer.write(
                b"GET /devtools/browser/id HTTP/1.1\r\n"
                b"Host: public.example\r\n"
                b"Upgrade: websocket\r\n"
                b"Connection: Upgrade\r\n"
                b"X-Lancer-CDP-Token: correct-token\r\n\r\n"
            )
            await writer.drain()
            response = await reader.readuntil(b"\r\n\r\n")
            self.assertIn(b"101 Switching Protocols", response)
            writer.write(b"ping")
            await writer.drain()
            self.assertEqual(await reader.readexactly(4), b"PING")
            writer.close()
            await writer.wait_closed()
            await asyncio.wait_for(upstream_done.wait(), timeout=2)
            await asyncio.wait_for(proxy_done.wait(), timeout=2)
            proxy.close()
            await proxy.wait_closed()
        upstream.close()
        await upstream.wait_closed()
        self.assertIn(f"Host: 127.0.0.1:{upstream_port}".encode(), captured["headers"])
        self.assertNotIn(b"correct-token", captured["headers"])

    def test_chrome_launch_is_remote_debuggable_and_headless_is_explicit(self):
        visual = build_chrome_command(visual=True)
        headless = build_chrome_command(visual=False)
        self.assertIn("--remote-debugging-port=9222", visual)
        self.assertIn("--remote-allow-origins=*", visual)
        self.assertNotIn("--headless=new", visual)
        self.assertIn("--headless=new", headless)

    def test_config_is_non_secret_and_uses_current_normal_depth_preset(self):
        config = build_worker_config(
            task="quem é lulinha?",
            url=None,
            visual=False,
            model=MODEL,
        )
        serialized = json.dumps(config)
        self.assertEqual(config["minimum_sources"], 2)
        self.assertEqual(config["min_evidence_to_finish"], 2)
        self.assertTrue(config["headless"])
        self.assertEqual(config["max_steps"], 18)
        self.assertEqual(config["llm_timeout"], 120)
        self.assertEqual(config["step_timeout"], 150)
        self.assertEqual(config["worker_timeout"], 600)
        self.assertNotIn("api_key", serialized.casefold())
        self.assertNotIn("secret", serialized.casefold())

    def test_prebuilt_e2b_template_is_optional_and_explicit(self):
        default_settings = SimpleNamespace(e2b_api_key="key")
        template_settings = SimpleNamespace(
            e2b_api_key="key",
            e2b_browser_template="lancer-browser-v1",
        )

        self.assertNotIn("template", build_e2b_sandbox_kwargs(default_settings))
        self.assertEqual(
            build_e2b_sandbox_kwargs(template_settings)["template"],
            "lancer-browser-v1",
        )

    def test_worker_environment_does_not_copy_unrelated_host_secrets(self):
        with patch.dict(
            os.environ,
            {
                "E2B_API_KEY": "must-not-cross",
                "AWS_SECRET_ACCESS_KEY": "must-not-cross-either",
            },
        ):
            environment = _worker_environment(
                openrouter_api_key="openrouter-key",
                cdp_url="wss://9223-sandbox.e2b.app/devtools/browser/id",
                cdp_token="cdp-token",
                runtime_dir=Path("/tmp/test-runtime"),
            )
        self.assertNotIn("E2B_API_KEY", environment)
        self.assertNotIn("AWS_SECRET_ACCESS_KEY", environment)
        self.assertEqual(environment["OPENROUTER_API_KEY"], "openrouter-key")
        self.assertEqual(environment["LANCER_CDP_TOKEN"], "cdp-token")
        self.assertEqual(environment["PYTHON_DOTENV_DISABLED"], "1")
        self.assertEqual(environment["PYTHONNOUSERSITE"], "1")
        self.assertEqual(environment["HOME"], str(Path("/tmp/test-runtime") / "home"))
        self.assertNotIn("PYTHONPATH", environment)

    async def test_local_worker_starts_in_its_private_runtime_directory(self):
        fake_process = _FakeLocalProcess()
        with tempfile.TemporaryDirectory() as temp_dir:
            runtime = Path(temp_dir)
            config_path = runtime / "config.json"
            config_path.write_text("{}", encoding="utf-8")
            with patch(
                "app.agents.browser_use_bridge.asyncio.create_subprocess_exec",
                new=AsyncMock(return_value=fake_process),
            ) as spawn:
                await _start_local_worker(
                    config_path=config_path,
                    events_path=runtime / "events.jsonl",
                    result_path=runtime / "result.json",
                    log_path=runtime / "worker.log",
                    environment={"PATH": os.environ.get("PATH", "")},
                )
        self.assertEqual(spawn.await_args.kwargs["cwd"], str(runtime))

    async def test_cancelled_creation_kills_late_e2b_sandbox(self):
        started = threading.Event()
        release = threading.Event()
        created = _FakeDesktop()

        class SlowSandbox:
            @classmethod
            def create(cls, **kwargs):
                started.set()
                release.wait(timeout=2)
                return created

        task = asyncio.create_task(_create_e2b_desktop(SlowSandbox, timeout=720))
        self.assertTrue(await asyncio.to_thread(started.wait, 1))
        task.cancel()
        release.set()
        with self.assertRaises(asyncio.CancelledError):
            await task
        self.assertTrue(created.killed)

    async def test_external_cdp_probe_converts_internal_ws_to_public_wss(self):
        _FakeHttpClient.calls = []
        desktop = _FakeDesktop()
        with patch(
            "app.agents.browser_use_bridge.httpx.AsyncClient",
            _FakeHttpClient,
        ):
            url = await _wait_for_external_cdp(desktop, "per-run-token")
        self.assertEqual(
            url,
            "wss://9223-sandbox.e2b.app/devtools/browser/internal-browser-id",
        )
        request_url, headers = _FakeHttpClient.calls[0]
        self.assertEqual(request_url, "https://9223-sandbox.e2b.app/json/version")
        self.assertEqual(headers["x-lancer-cdp-token"], "per-run-token")

    def test_translation_exposes_only_accepted_evidence_as_facts(self):
        progress = {"evidence": [], "next_goals": [], "queries": []}
        events = translate_worker_event(
            {
                "type": "step",
                "step": 2,
                "url": "https://example.com",
                "title": "Example",
                "actions": [{"browser_search": {"query": "focused query"}}],
                "memory": "Unverified model memory must not appear as evidence",
                "next_goal": "Open the original source",
            },
            progress,
        )
        public_progress = next(item for item in events if item["type"] == "progress")
        self.assertEqual(public_progress["known_facts"], [])
        self.assertEqual(public_progress["last_queries"], ["focused query"])
        self.assertEqual(public_progress["missing_points"], ["Open the original source"])

    def test_public_result_never_promotes_search_or_visited_urls(self):
        config = build_worker_config(
            task="quem é alguém?",
            url=None,
            visual=False,
            model="test-model",
        )
        public = build_public_result(
            {
                "ok": True,
                "answer": "Useful partial answer",
                "agent_success": True,
                "urls": [
                    "https://duckduckgo.com/?q=someone",
                    "https://unverified.example/page",
                    "https://verified.example/page",
                ],
                "evidence": [
                    {
                        "claim": "Verified claim",
                        "exact_quote": "A sufficiently long exact quotation",
                        "url": "https://verified.example/page",
                    },
                    {
                        "claim": "Snippet claim",
                        "exact_quote": "A search snippet must never count",
                        "url": "https://duckduckgo.com/?q=someone",
                    },
                ],
            },
            config,
        )
        self.assertEqual(public["links"], ["https://verified.example/page"])
        self.assertFalse(public["success"])
        self.assertIn("1/2", public["content"])
        self.assertIn("https://unverified.example/page", public["visited_links"])
        self.assertEqual(public["verification_status"], "partial")

    def test_public_result_without_recorded_evidence_is_never_verified(self):
        config = build_worker_config(
            task="research a person",
            url=None,
            visual=False,
            model="test-model",
        )
        public = build_public_result(
            {
                "ok": True,
                "answer": "A useful evaluator draft",
                "agent_success": True,
                "urls": ["https://visited.example/profile"],
                "evidence": [],
            },
            config,
        )

        self.assertFalse(public["success"])
        self.assertEqual(public["links"], [])
        self.assertEqual(public["evidence"], [])
        self.assertEqual(public["verification_status"], "partial")
        self.assertIn("0/1", public["content"])
        self.assertNotIn("Citação validada pelo Evaluator", worker_script_source())

    def test_subdomains_of_one_publisher_are_not_independent_sources(self):
        config = build_worker_config(
            task="quem é alguém?",
            url=None,
            visual=False,
            model="test-model",
        )
        public = build_public_result(
            {
                "ok": True,
                "answer": "Answer",
                "agent_success": True,
                "evidence": [
                    {
                        "claim": "Claim one",
                        "exact_quote": "A sufficiently long literal quote one",
                        "url": "https://pt.wikipedia.org/wiki/Person",
                    },
                    {
                        "claim": "Claim two",
                        "exact_quote": "A sufficiently long literal quote two",
                        "url": "https://en.wikipedia.org/wiki/Person",
                    },
                ],
            },
            config,
        )
        self.assertEqual(public["verified_sources"], 1)
        self.assertFalse(public["success"])

    async def test_full_bridge_runs_host_browser_use_with_camoufox_in_e2b(self):
        fake_module = types.ModuleType("e2b_desktop")
        fake_module.Sandbox = _FakeSandbox
        settings = SimpleNamespace(
            e2b_api_key="e2b-test-key",
            openrouter_api_key="openrouter-test-key",
            llm_model=MODEL,
        )
        captured = {}

        async def fake_start_local_worker(**kwargs):
            captured["environment"] = dict(kwargs["environment"])
            captured["config"] = kwargs["config_path"].read_text(encoding="utf-8")
            events, result = _successful_events_and_result()
            kwargs["events_path"].write_text(
                "\n".join(json.dumps(item) for item in events) + "\n",
                encoding="utf-8",
            )
            kwargs["result_path"].write_text(json.dumps(result), encoding="utf-8")
            return _FakeLocalProcess()

        _FakeSandbox.desktop = None
        with (
            patch.dict(sys.modules, {"e2b_desktop": fake_module}),
            patch("app.agents.browser_use_bridge.get_settings", return_value=settings),
            patch(
                "app.agents.browser_use_bridge.installed_browser_use_version",
                return_value=BROWSER_USE_VERSION,
            ),
            patch(
                "app.agents.browser_use_bridge._wait_for_external_cdp",
                new=AsyncMock(
                    return_value=(
                        "wss://9223-sandbox.e2b.app/devtools/browser/test-browser"
                    )
                ),
            ),
            patch(
                "app.agents.browser_use_bridge._wait_for_external_stealth_browser",
                new=AsyncMock(return_value="https://8766-sandbox.e2b.app"),
            ),
            patch(
                "app.agents.browser_use_bridge._start_local_worker",
                side_effect=fake_start_local_worker,
            ),
        ):
            events = [
                event
                async for event in run_browser_use_agent(
                    "quem é lulinha?",
                    visual=False,
                )
            ]

        desktop = _FakeSandbox.desktop
        commands = [command for command, _ in desktop.commands.calls]
        self.assertTrue(any("google-chrome" in command for command in commands))
        self.assertIn(CDP_PROXY_COMMAND, commands)
        self.assertIn(CAMOUFOX_SETUP_COMMAND, commands)
        self.assertTrue(any(f"camoufox=={CAMOUFOX_VERSION}" in command for command in commands))
        self.assertFalse(any("browser-use==" in command for command in commands))
        self.assertNotIn("openrouter-test-key", " ".join(commands))
        self.assertIn(REMOTE_CDP_PROXY_PATH, desktop.files.data)
        self.assertIn(REMOTE_STEALTH_SERVER_PATH, desktop.files.data)
        self.assertTrue(desktop.killed)
        background_handles = [
            handle
            for (_, kwargs), handle in zip(
                desktop.commands.calls,
                desktop.commands.handles,
            )
            if kwargs.get("background")
        ]
        self.assertTrue(background_handles)
        self.assertTrue(all(handle.killed for handle in background_handles))
        # Regression: E2B rejects several mathematically valid reserved CIDRs
        # (including 0.0.0.0/8) with HTTP 400. Network policy must not prevent
        # the browser sandbox from being created.
        self.assertNotIn("network", _FakeSandbox.create_kwargs)
        self.assertNotIn("template", _FakeSandbox.create_kwargs)

        config_text = captured["config"]
        worker_environment = captured["environment"]
        self.assertNotIn("openrouter-test-key", config_text)
        self.assertNotIn("e2b-test-key", config_text)
        self.assertIn("https://8766-sandbox.e2b.app", config_text)
        self.assertEqual(worker_environment["OPENROUTER_API_KEY"], "openrouter-test-key")
        self.assertNotIn("E2B_API_KEY", worker_environment)
        result = next(item for item in events if item["type"] == "result")
        self.assertTrue(result["success"])
        self.assertEqual(result["verified_sources"], 2)
        self.assertNotIn("https://duckduckgo.com/?q=lulinha", result["links"])

    async def test_missing_pinned_host_runtime_fails_before_creating_e2b(self):
        settings = SimpleNamespace(
            e2b_api_key="e2b-test-key",
            openrouter_api_key="openrouter-test-key",
            llm_model=MODEL,
        )
        _FakeSandbox.desktop = None
        with (
            patch("app.agents.browser_use_bridge.get_settings", return_value=settings),
            patch(
                "app.agents.browser_use_bridge.installed_browser_use_version",
                return_value="",
            ),
        ):
            events = [
                event
                async for event in run_browser_use_agent("research", visual=False)
            ]
        self.assertEqual(len(events), 1)
        self.assertEqual(events[0]["type"], "error")
        self.assertIn(f"browser-use=={BROWSER_USE_VERSION}", events[0]["message"])
        self.assertIsNone(_FakeSandbox.desktop)

    async def test_runtime_failure_redacts_every_request_credential(self):
        fake_module = types.ModuleType("e2b_desktop")
        fake_module.Sandbox = _FakeSandbox
        settings = SimpleNamespace(
            e2b_api_key="e2b-test-key",
            openrouter_api_key="openrouter-test-key",
            llm_model=MODEL,
        )
        failure = RuntimeError(
            "gateway rejected cdp-secret-token openrouter-test-key e2b-test-key"
        )

        _FakeSandbox.desktop = None
        with (
            patch.dict(sys.modules, {"e2b_desktop": fake_module}),
            patch("app.agents.browser_use_bridge.get_settings", return_value=settings),
            patch(
                "app.agents.browser_use_bridge.installed_browser_use_version",
                return_value=BROWSER_USE_VERSION,
            ),
            patch(
                "app.agents.browser_use_bridge.secrets.token_urlsafe",
                return_value="cdp-secret-token",
            ),
            patch(
                "app.agents.browser_use_bridge._wait_for_external_cdp",
                new=AsyncMock(side_effect=failure),
            ),
            self.assertLogs("app.agents.browser_use_bridge", level="ERROR") as logs,
        ):
            events = [
                event
                async for event in run_browser_use_agent("research", visual=False)
            ]

        combined = json.dumps(events, ensure_ascii=False) + "\n" + "\n".join(logs.output)
        self.assertIn("[REDACTED]", combined)
        self.assertNotIn("cdp-secret-token", combined)
        self.assertNotIn("openrouter-test-key", combined)
        self.assertNotIn("e2b-test-key", combined)
        self.assertTrue(_FakeSandbox.desktop.killed)
        self.assertTrue(
            all(handle.killed for handle in _FakeSandbox.desktop.commands.handles)
        )

    async def test_visual_failure_ends_stream_before_reporting_error(self):
        fake_module = types.ModuleType("e2b_desktop")
        fake_module.Sandbox = _FakeSandbox
        settings = SimpleNamespace(
            e2b_api_key="e2b-test-key",
            openrouter_api_key="openrouter-test-key",
            llm_model=MODEL,
        )

        _FakeSandbox.desktop = None
        with (
            patch.dict(sys.modules, {"e2b_desktop": fake_module}),
            patch("app.agents.browser_use_bridge.get_settings", return_value=settings),
            patch(
                "app.agents.browser_use_bridge.installed_browser_use_version",
                return_value=BROWSER_USE_VERSION,
            ),
            patch(
                "app.agents.browser_use_bridge._wait_for_external_cdp",
                new=AsyncMock(
                    return_value="wss://9223-sandbox.e2b.app/devtools/browser/id"
                ),
            ),
            patch(
                "app.agents.browser_use_bridge._start_local_worker",
                new=AsyncMock(side_effect=RuntimeError("worker startup failed")),
            ),
        ):
            events = [
                event
                async for event in run_browser_use_agent("research", visual=True)
            ]

        event_types = [event["type"] for event in events]
        self.assertIn("stream", event_types)
        self.assertLess(event_types.index("stream_end"), event_types.index("error"))
        self.assertTrue(_FakeSandbox.desktop.stream.stopped)


if __name__ == "__main__":
    unittest.main()