import io
import json
import unittest
import os
from dataclasses import replace
from pathlib import Path
from unittest.mock import Mock, patch

from browser_worker import (
    BrowserWorker, BrowserRuntimeWorker, ElementMetadata, WorkerError, BrowserOperationError,
    RealBrowserBackend, MAX_LINE_BYTES, MAX_READ_TEXT, process_request, run_stream,
    validate_cdp_url, main,
)


class StartupDiagnosticsTests(unittest.TestCase):
    def test_missing_approved_tab_reports_origins_without_private_url_components(self):
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend.allowed_origins = {"https://fixture.test"}
        backend._helpers = Mock()
        backend._helpers.list_tabs.return_value = [
            {"url": "https://login.example.test/private-path?token=private-token#private-fragment"},
            {"url": "chrome://welcome"},
            {"url": "https://user:private-password@login.example.test/"},
        ]
        with self.assertRaises(WorkerError) as caught:
            backend._attach_approved_tab()
        message = str(caught.exception)
        self.assertIn("found 0", message)
        self.assertIn('open web origins: ["https://login.example.test"]', message)
        self.assertNotIn("private-", message)
        backend._helpers.switch_tab.assert_not_called()

    def test_startup_errors_use_safe_ipc_and_remain_closable(self):
        for error, expected in [
            (WorkerError("expected exactly one existing approved-origin browser tab; found 0"), "found 0"),
            (RuntimeError("private-startup-value"), "browser worker initialization failed"),
        ]:
            with self.subTest(error=type(error).__name__):
                reader = io.BytesIO(b'{"id":1,"op":"observe"}\n{"id":2,"op":"close"}\n')
                writer = io.BytesIO()
                with patch("browser_worker.RealBrowserBackend", side_effect=error), \
                     patch("browser_worker.signal.signal"), \
                     patch("browser_worker.run_stream", side_effect=lambda worker: run_stream(worker, reader, writer)):
                    self.assertEqual(main(["--cdp-url", "http://127.0.0.1:9222", "--allowed-origins", '["http://fixture.test"]']), 0)
                messages = [json.loads(line) for line in writer.getvalue().splitlines()]
                self.assertEqual(messages[0]["id"], 1)
                self.assertFalse(messages[0]["ok"])
                self.assertIn(expected, messages[0]["error"])
                self.assertTrue(messages[1]["ok"])
                self.assertNotIn(b"private-startup-value", writer.getvalue())


class FakeBackend:
    def __init__(self):
        self.url = "http://fixture.test/form"
        self.target = "tab-1"
        self.frame = "frame-1"
        self.loader = "loader-1"
        self.elements = {
            1: ElementMetadata(1, "input", "text", "textbox", "Email", True, False),
            2: ElementMetadata(2, "input", "password", "textbox", "Password", True, False),
            3: ElementMetadata(3, "button", "", "button", "Submit", True, False),
            4: ElementMetadata(4, "input", "text", "textbox", "Disabled", True, True),
        }
        self.values = {1: ""}
        self.nodes = [
            {"backendDOMNodeId": 1, "role": {"value": "textbox"}, "name": {"value": "Email"}},
            {"backendDOMNodeId": 3, "role": {"value": "button"}, "name": {"value": "Submit"}},
            {"backendDOMNodeId": 4, "role": {"value": "textbox"}, "name": {"value": "Disabled"}},
            # Password controls are supported; their values remain private.
            {"backendDOMNodeId": 2, "role": {"value": "textbox"}, "name": {"value": "Password"}},
        ]
        self.clicks = 0

    def current_tab(self):
        return {"targetId": self.target, "url": self.url, "title": "Fixture"}

    def document_identity(self):
        return self.frame, self.loader

    def accessibility_nodes(self):
        return self.nodes

    def metadata(self, backend_id):
        return self.elements[backend_id]

    def click(self, backend_id):
        self.clicks += 1
        if backend_id == 3:
            self.url = "http://fixture.test/done"

    def fill(self, backend_id, text):
        self.values[backend_id] = text

    def value_matches(self, backend_id, text):
        return self.values.get(backend_id) == text

    def visible_text(self, text):
        return text == "Finished"

    def close(self):
        pass


class PopupFakeBackend(FakeBackend):
    def __init__(self):
        super().__init__()
        self.status = "Scenario fixture ready"

    def click(self, backend_id):
        super().click(backend_id)
        if backend_id == 3:
            self.status = "Popup opened"

    def visible_text(self, text):
        return text == self.status


class RuntimeFakeBackend(FakeBackend):
    def __init__(self):
        super().__init__()
        self.tabs = [
            {"targetId": "tab-a", "url": "http://fixture.test/form", "title": "Fixture"},
            {"targetId": "tab-b", "url": "http://other.test/second", "title": "Second"},
        ]
        self.target = "tab-a"
        self.text_content = "Fixture complete Привет"
        self.events = []
        self._session_id = "fake-session"
        self._session_counter = 0
        self.switches = []
        self.closed_tabs = []

    def list_tabs(self):
        return [dict(tab) for tab in self.tabs]

    def switch_tab(self, target_id):
        if not any(tab["targetId"] == target_id for tab in self.tabs):
            raise RuntimeError("missing target")
        self.target = target_id
        self._session_counter += 1
        self._session_id = f"fake-session-{self._session_counter}"
        self.switches.append(target_id)

    def current_tab(self):
        return next(tab for tab in self.tabs if tab["targetId"] == self.target)

    def visible_text_content(self):
        return self.text_content

    def navigate(self, url):
        self.current_tab()["url"] = url
        self.loader = self.loader + "-nav"

    def reload(self):
        self.loader = self.loader + "-reload"

    def wait_ready(self, timeout):
        return True

    def close_tab(self, target_id):
        self.closed_tabs.append(target_id)
        self.tabs[:] = [tab for tab in self.tabs if tab["targetId"] != target_id]

    def drain_events(self):
        events, self.events = self.events, []
        return events

    def screenshot_bytes(self):
        return b"\\x89PNG" + (b"x" * 32)


class RuntimeTests(unittest.TestCase):
    def make_runtime(self, backend=None, origins="*"):
        backend = backend or RuntimeFakeBackend()
        return BrowserRuntimeWorker(backend, origins, session_id="runtime-test")

    def test_list_tabs_does_not_fail_or_select_ambiguous_pages(self):
        backend = RuntimeFakeBackend()
        runtime = self.make_runtime(backend)
        tabs = runtime.list_tabs()
        self.assertEqual(len(tabs), 2)
        self.assertFalse(any(tab["selected"] for tab in tabs))
        with self.assertRaisesRegex(BrowserOperationError, "select an approved tab"):
            runtime.read({})
        chosen = tabs[0]["tabId"]
        selected = runtime.select_tab({"tabId": chosen})
        self.assertTrue(selected["tab"]["selected"])
        self.assertEqual(backend.target, "tab-a" if tabs[0]["origin"] == "http://fixture.test" else "tab-b")

    def test_read_is_bounded_unicode_and_has_no_form_or_hidden_values(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        backend.text_content = "Visible Привет\n" + ("é" * (MAX_READ_TEXT + 20))
        runtime = self.make_runtime(backend)
        result = runtime.read({})
        self.assertTrue(result["visibleTextTruncated"])
        self.assertEqual(len(result["visibleText"]), MAX_READ_TEXT)
        self.assertIn("Привет", result["visibleText"])
        self.assertNotIn("password-value", result["visibleText"])
        self.assertNotIn("<input", result["visibleText"])

    def test_assertion_polling_is_read_only_and_reports_timeout(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        result = runtime.assert_page({
            "assertion": {"kind": "visibleText", "match": "contains", "value": "missing"},
            "wait": "none",
        })
        self.assertEqual(result["status"], "not_satisfied")
        result = runtime.assert_page({
            "assertion": {"kind": "title", "match": "contains", "value": "missing"},
            "wait": "bounded",
        })
        self.assertEqual(result["status"], "timed_out")
        self.assertEqual(backend.clicks, 0)

    def test_post_dispatch_failures_are_unknown_and_never_retryable(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        observed = runtime.observe()
        def mutate_then_raise(_backend_id):
            backend.clicks += 1
            raise RuntimeError("browser evaluation failed")
        backend.click = mutate_then_raise
        with self.assertRaises(BrowserOperationError) as caught:
            runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
                         "postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
        self.assertEqual(caught.exception.details["mutationState"], "unknown")
        self.assertEqual(caught.exception.details["retry"], "never")

        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        observed = runtime.observe()
        checks = [0]
        def verify_then_raise(_text):
            checks[0] += 1
            if checks[0] == 1:
                return False
            raise RuntimeError("postcondition evaluation failed")
        backend.visible_text = verify_then_raise
        with self.assertRaises(BrowserOperationError) as caught:
            runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
                         "postcondition": {"kind": "text", "value": "done"}})
        self.assertEqual(caught.exception.details["mutationState"], "unknown")
        self.assertEqual(caught.exception.details["retry"], "never")

        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        observed = runtime.observe()
        def cancel_after_dispatch(_backend_id):
            class Cancelled(BaseException):
                pass
            raise Cancelled()
        backend.click = cancel_after_dispatch
        with self.assertRaises(BrowserOperationError) as caught:
            runtime.act({"observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
                         "postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
        self.assertEqual(caught.exception.details["code"], "aborted")
        self.assertEqual(caught.exception.details["mutationState"], "unknown")

        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        def navigate_then_raise(url):
            backend.current_tab()["url"] = url
            raise RuntimeError("navigation acknowledgement failed")
        backend.navigate = navigate_then_raise
        with self.assertRaises(BrowserOperationError) as caught:
            runtime.navigate({"url": "http://fixture.test/next"})
        self.assertEqual(caught.exception.details["mutationState"], "unknown")

        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        tab_id = runtime.list_tabs()[0]["tabId"]
        original_close = backend.close_tab
        def close_then_raise(target):
            original_close(target)
            raise RuntimeError("close acknowledgement failed")
        backend.close_tab = close_then_raise
        with self.assertRaises(BrowserOperationError) as caught:
            runtime.close_tab({"tabId": tab_id})
        self.assertEqual(caught.exception.details["mutationState"], "unknown")

    def test_navigation_reload_change_document_without_retrying_mutation(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend, ["http://fixture.test"])
        first = runtime.read({})["documentId"]
        navigated = runtime.navigate({"url": "http://fixture.test/next"})
        self.assertEqual(navigated["mutationState"], "verified")
        self.assertNotEqual(first, navigated["documentId"])
        reloaded = runtime.reload({})
        self.assertNotEqual(navigated["documentId"], reloaded["documentId"])
        with self.assertRaisesRegex(BrowserOperationError, "outside"):
            runtime.navigate({"url": "https://not-approved.test/"})

    def test_narrow_policy_is_per_dispatch_and_cannot_broaden_base(self):
        backend = RuntimeFakeBackend()
        runtime = self.make_runtime(backend)
        narrowed = runtime.narrow_origin_policy({"allowedOrigins": ["http://fixture.test"]})
        self.assertEqual(narrowed, {"allowedOrigins": ["http://fixture.test"]})
        runtime.apply_context(narrowed)
        self.assertEqual([tab["origin"] for tab in runtime.list_tabs()], ["http://fixture.test"])
        runtime.apply_context(None)
        self.assertEqual(len(runtime.list_tabs()), 2)
        restricted = self.make_runtime(RuntimeFakeBackend(), ["http://fixture.test"])
        with self.assertRaisesRegex(BrowserOperationError, "broaden"):
            restricted.apply_context({"allowedOrigins": ["https://not-in-base.test"]})

    def test_popup_handle_is_owned_but_wait_does_not_switch(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        opener = runtime.list_tabs()[0]["tabId"]
        backend.tabs.append({"targetId": "popup", "url": "http://fixture.test/popup", "title": "Popup", "openerTargetId": "tab-a"})
        popup = runtime.wait_popup({"openerTabId": opener})
        self.assertEqual(popup["status"], "found")
        self.assertFalse(popup["autoSwitched"])
        self.assertEqual(backend.target, "tab-a")
        self.assertEqual(popup["tab"]["kind"], "popup")

    def test_immediate_popup_is_found_through_typed_dispatcher_sequence(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        original_click = backend.click
        def click_and_open(backend_id):
            original_click(backend_id)
            backend.visible_text = lambda text: text == "POPUP_OPEN_REQUESTED"
            backend.text_content = "POPUP_OPEN_REQUESTED"
            backend.tabs.append({"targetId": "immediate-popup", "url": "http://fixture.test/popup",
                                 "title": "Popup", "openerTargetId": "tab-a"})
        backend.click = click_and_open
        runtime = self.make_runtime(backend)
        observed = process_request(runtime, {"id": 1, "operation": {"op": "observe"}})["result"]["result"]
        acted = process_request(runtime, {"id": 2, "operation": {"op": "act", "request": {
            "observationId": observed["observationId"], "candidateId": "c-1", "action": "click",
            "postcondition": {"kind": "text", "value": "POPUP_OPEN_REQUESTED"},
        }}})
        self.assertTrue(acted["result"]["result"]["verified"])
        opener = runtime.list_tabs()[0]["tabId"]
        waited = process_request(runtime, {"id": 3, "operation": {"op": "waitPopup", "request": {
            "openerTabId": opener,
        }}})
        popup = waited["result"]["result"]
        self.assertEqual(popup["status"], "found")
        self.assertFalse(popup["autoSwitched"])
        self.assertEqual(popup["tab"]["kind"], "popup")
        self.assertEqual(backend.target, "tab-a")

    def test_popup_wait_excludes_baseline_unrelated_and_claimed_targets(self):
        from unittest.mock import patch
        with patch("browser_worker.POPUP_WAIT_SECONDS", 0.01):
            backend = RuntimeFakeBackend()
            baseline_popup = {"targetId": "baseline-popup", "url": "http://fixture.test/old-popup",
                              "title": "Old", "openerTargetId": "tab-a"}
            # Deliberately put the popup before its opener in the CDP inventory.
            backend.tabs = [baseline_popup, backend.tabs[0], backend.tabs[1]]
            runtime = self.make_runtime(backend)
            opener = next(tab for tab in runtime.list_tabs() if tab["url"].endswith("/form"))["tabId"]
            self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")

            backend.tabs.append({"targetId": "unrelated-popup", "url": "http://fixture.test/unrelated",
                                 "title": "Unrelated", "openerTargetId": "tab-b"})
            self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")

            backend.tabs.append({"targetId": "claimed-popup", "url": "http://fixture.test/claimed",
                                 "title": "Claimed", "openerTargetId": "tab-a"})
            found = runtime.wait_popup({"openerTabId": opener})
            self.assertEqual(found["status"], "found")
            self.assertEqual(runtime.wait_popup({"openerTabId": opener})["status"], "timed_out")

    def test_closed_tab_handle_is_invalidated(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        tab_id = runtime.list_tabs()[0]["tabId"]
        self.assertEqual(runtime.close_tab({"tabId": tab_id})["mutationState"], "verified")
        with self.assertRaisesRegex(BrowserOperationError, "stale|unavailable"):
            runtime.select_tab({"tabId": tab_id})

    def test_transport_wrapper_context_restores_base_and_uses_exact_result_shape(self):
        backend = RuntimeFakeBackend()
        runtime = self.make_runtime(backend)
        requests = (
            b'{"id":1,"operation":{"op":"narrowOriginPolicy","request":{"allowedOrigins":["http://fixture.test"]}}}\n'
            b'{"id":2,"operation":{"op":"listTabs"},"context":{"allowedOrigins":["http://fixture.test"]}}\n'
            b'{"id":3,"operation":{"op":"listTabs"}}\n'
        )
        output = io.BytesIO()
        run_stream(runtime, io.BytesIO(requests), output)
        rows = [json.loads(line) for line in output.getvalue().splitlines()]
        self.assertEqual(rows[0]["result"]["op"], "narrowOriginPolicy")
        self.assertEqual(rows[0]["result"]["result"], {"allowedOrigins": ["http://fixture.test"]})
        self.assertEqual(len(rows[1]["result"]["result"]), 1)
        self.assertEqual(len(rows[2]["result"]["result"]), 2)

    def test_real_screenshot_temp_is_private_bounded_and_cleaned_on_failure(self):
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        # Use an isolated directory rather than any user temp location.
        import tempfile
        root = Path(tempfile.mkdtemp(prefix="fb-shot-test-"))
        shot_dir = root / "shots"
        shot_dir.mkdir(mode=0o700)
        backend._screenshot_dir = shot_dir
        backend._helpers = Mock()
        observed = []
        def capture(path, **_kwargs):
            candidate = Path(path)
            observed.append((candidate, candidate.stat().st_mode & 0o777))
            candidate.write_bytes(b"png")
        backend._helpers.capture_screenshot.side_effect = capture
        self.assertEqual(backend.screenshot_bytes(), b"png")
        self.assertEqual(observed[0][0].parent, shot_dir)
        self.assertEqual(observed[0][1], 0o600)
        self.assertEqual(list(shot_dir.iterdir()), [])
        backend._helpers.capture_screenshot.side_effect = RuntimeError("capture failed")
        with self.assertRaises(RuntimeError):
            backend.screenshot_bytes()
        self.assertEqual(list(shot_dir.iterdir()), [])
        import shutil
        shutil.rmtree(root, ignore_errors=True)

    def test_screenshot_transport_is_capped_and_marked_truncated(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        backend.screenshot_bytes = lambda: b"x" * (8 * 1024 * 1024 + 1)
        runtime = self.make_runtime(backend)
        evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
        self.assertTrue(evidence["screenshot"]["truncated"])
        self.assertEqual(evidence["screenshot"]["byteLength"], 8 * 1024 * 1024)

    def test_evidence_filters_third_party_network_failures_by_effective_policy(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend, ["http://fixture.test"])
        backend.events = [
            {"method": "Runtime.consoleAPICalled", "session_id": backend._session_id,
             "params": {"type": "error", "args": [{"value": "https://third-party.test/console?token=bad"}]}},
            {"method": "Network.loadingFailed", "session_id": backend._session_id,
             "params": {"requestId": "third", "url": "https://third-party.test/fail?token=bad"}},
            {"method": "Network.loadingFailed", "session_id": backend._session_id,
             "params": {"requestId": "local", "url": "http://fixture.test/fail?token=bad"}},
        ]
        evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
        self.assertEqual([item["origin"] for item in evidence["failedRequests"]], ["http://fixture.test"])
        self.assertNotIn("third-party.test", json.dumps(evidence["console"]))

    def test_evidence_does_not_replay_buffered_history_after_tab_switches(self):
        backend = RuntimeFakeBackend()
        runtime = self.make_runtime(backend)
        tabs = runtime.list_tabs()
        runtime.select_tab({"tabId": tabs[0]["tabId"]})
        session_a = backend._session_id
        backend.events = [{"method": "Network.loadingFailed", "session_id": session_a,
                           "params": {"requestId": "a", "url": "http://fixture.test/a"}}]
        first = runtime.capture_evidence({"runId": "run-a", "reason": "requested"})
        self.assertEqual([item["origin"] for item in first["failedRequests"]], ["http://fixture.test"])

        runtime.select_tab({"tabId": tabs[1]["tabId"]})
        second = runtime.capture_evidence({"runId": "run-b", "reason": "requested"})
        self.assertEqual(second["failedRequests"], [])

        # Returning to the target creates a new CDP session; old A history is
        # still retained privately but must not be invented as current evidence.
        runtime.select_tab({"tabId": tabs[0]["tabId"]})
        third = runtime.capture_evidence({"runId": "run-a2", "reason": "requested"})
        self.assertEqual(third["failedRequests"], [])

    def test_evidence_drops_events_from_prior_tab_sessions(self):
        backend = RuntimeFakeBackend()
        runtime = self.make_runtime(backend)
        tabs = runtime.list_tabs()
        runtime.select_tab({"tabId": tabs[0]["tabId"]})
        prior_session = backend._session_id
        runtime.select_tab({"tabId": tabs[1]["tabId"]})
        current_session = backend._session_id
        backend.events = [
            {"method": "Network.loadingFailed", "session_id": prior_session,
             "params": {"requestId": "old", "url": "http://fixture.test/old?secret=bad"}},
            {"method": "Network.loadingFailed", "session_id": current_session,
             "params": {"requestId": "new", "url": "http://other.test/new?secret=bad"}},
            {"method": "Network.loadingFailed", "params": {"requestId": "unknown", "url": "http://other.test/unknown"}},
        ]
        evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
        self.assertEqual([item["origin"] for item in evidence["failedRequests"]], ["http://other.test"])

    def test_protocol_dispatch_returns_structured_safe_errors(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0], backend.tabs[1]]
        runtime = self.make_runtime(backend)
        output = io.BytesIO()
        run_stream(runtime, io.BytesIO(b'{"id":9,"operation":{"op":"read","request":{}}}\n'), output)
        response = json.loads(output.getvalue())
        self.assertFalse(response["ok"])
        self.assertEqual(response["details"]["code"], "ambiguous_tab")
        self.assertEqual(response["details"]["operation"], "read")
        self.assertEqual(response["details"]["mutationState"], "not_attempted")
        self.assertNotIn("targetId", json.dumps(response))

    def test_evidence_bounds_redacts_and_returns_no_caller_path(self):
        backend = RuntimeFakeBackend()
        backend.tabs = [backend.tabs[0]]
        runtime = self.make_runtime(backend)
        backend.events = [
            {"method": "Runtime.consoleAPICalled", "session_id": backend._session_id, "params": {"type": "error", "args": [{"value": "token=private https://fixture.test/x?secret=bad"}]}},
            {"method": "Network.loadingFailed", "session_id": backend._session_id, "params": {"requestId": "r", "url": "http://fixture.test/fail?token=bad"}},
        ]
        evidence = runtime.capture_evidence({"runId": "run", "reason": "requested"})
        self.assertEqual(set(evidence["screenshot"]), {"kind", "contentType", "encoding", "dataBase64", "byteLength", "truncated"})
        self.assertEqual(set(evidence["diagnostics"]), {"kind", "contentType", "encoding", "dataBase64", "byteLength", "truncated"})
        self.assertEqual(evidence["screenshot"]["encoding"], "base64")
        self.assertEqual(evidence["diagnostics"]["encoding"], "base64")
        self.assertFalse(evidence["screenshot"]["truncated"])
        self.assertNotIn("path", evidence)
        self.assertNotIn("private", json.dumps(evidence))
        self.assertNotIn("?token=bad", json.dumps(evidence))
        self.assertEqual(evidence["failedRequests"][0]["origin"], "http://fixture.test")


class WorkerTests(unittest.TestCase):
    def test_unrestricted_policy_accepts_any_http_origin_but_not_internal_urls(self):
        backend = FakeBackend()
        backend.url = "https://other.example/path"
        worker = BrowserWorker(backend, "*")
        observation = worker.observe()
        self.assertEqual(observation["url"], "https://other.example/path")
        for url in ["file:///tmp/private", "chrome://settings", "about:blank",
                    "https://user:private-password@example.test/", "https://*.example.test/"]:
            with self.subTest(url=url):
                backend.url = url
                # Leaving policy stops a worker permanently; use a fresh one
                # to exercise each unsupported URL rather than its stopped fuse.
                worker = BrowserWorker(backend, "*")
                with self.assertRaisesRegex(WorkerError, "outside"):
                    worker.observe()

    def test_observe_bounds_candidates_and_does_not_return_values(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        self.assertEqual(observation["url"], "http://fixture.test/form")
        self.assertEqual([c["name"] for c in observation["candidates"]], ["Email", "Submit", "Password"])
        self.assertNotIn("values", json.dumps(observation))
        self.assertNotIn("Disabled", json.dumps(observation))

    def test_password_fill_uses_literal_without_exposing_value_in_metadata_or_result(self):
        backend = FakeBackend()
        backend.values[2] = "old-test-secret"
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        password = next(c for c in observation["candidates"] if c["name"] == "Password")
        self.assertEqual(password["actions"], ["fill"])
        self.assertNotIn("old-test-secret", json.dumps(observation))
        result = worker.act({"observationId": observation["observationId"], "candidateId": password["id"],
                             "action": "fill", "text": "new-test-secret"})
        self.assertTrue(result["verified"])
        self.assertEqual(backend.values[2], "new-test-secret")
        self.assertNotIn("new-test-secret", json.dumps(result))
        self.assertNotIn("new-test-secret", json.dumps(worker.observe()))

    def test_fill_verifies_literal_and_consumes_observation(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        result = worker.act({
            "observationId": observation["observationId"],
            "candidateId": "c-0",
            "action": "fill",
            "text": "literal@example.test",
        })
        self.assertEqual(result["verified"], True)
        self.assertEqual(backend.values[1], "literal@example.test")
        with self.assertRaisesRegex(WorkerError, "missing|consumed"):
            worker.act({
                "observationId": observation["observationId"],
                "candidateId": "c-0",
                "action": "fill",
                "text": "retry",
            })

    def test_mutation_invalidates_all_previous_observations(self):
        worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
        old = worker.observe()
        new = worker.observe()
        worker.act({"observationId": new["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada"})
        with self.assertRaisesRegex(WorkerError, "missing|consumed"):
            worker.act({"observationId": old["observationId"], "candidateId": "c-0", "action": "fill", "text": "replay"})

    def test_click_rejects_postcondition_that_was_already_satisfied(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        with self.assertRaisesRegex(WorkerError, "already satisfied"):
            worker.act({
                "observationId": observation["observationId"],
                "candidateId": "c-1",
                "action": "click",
                "postcondition": {"kind": "url", "value": "http://fixture.test/form"},
            })
        self.assertEqual(backend.clicks, 0)

    def test_popup_fixture_postcondition_is_false_before_click_and_true_after(self):
        backend = PopupFakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        self.assertFalse(backend.visible_text("Popup opened"))
        result = worker.act({
            "observationId": observation["observationId"],
            "candidateId": "c-1",
            "action": "click",
            "postcondition": {"kind": "text", "value": "Popup opened"},
        })
        self.assertTrue(result["verified"])
        self.assertTrue(backend.visible_text("Popup opened"))
        self.assertEqual(backend.clicks, 1)

    def test_click_requires_postcondition_and_does_not_retry(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        result = worker.act({
            "observationId": observation["observationId"],
            "candidateId": "c-1",
            "action": "click",
            "postcondition": {"kind": "url", "value": "http://fixture.test/done"},
        })
        self.assertTrue(result["verified"])
        self.assertEqual(backend.clicks, 1)

    def test_stale_document_is_rejected_before_mutation(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        backend.loader = "new-loader"
        with self.assertRaisesRegex(WorkerError, "stale"):
            worker.act({
                "observationId": observation["observationId"],
                "candidateId": "c-1",
                "action": "click",
                "postcondition": {"kind": "url", "value": "http://fixture.test/done"},
            })
        self.assertEqual(backend.clicks, 0)

    def test_outside_redirect_blocks_worker(self):
        backend = FakeBackend()
        original_click = backend.click

        def redirect(backend_id):
            original_click(backend_id)
            backend.url = "https://not-approved.test/"

        backend.click = redirect
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        with self.assertRaisesRegex(WorkerError, "outside"):
            worker.act({
                "observationId": observation["observationId"],
                "candidateId": "c-1",
                "action": "click",
                "postcondition": {"kind": "url", "value": "http://fixture.test/done"},
            })
        with self.assertRaisesRegex(WorkerError, "stopped"):
            worker.observe()

    def test_jsonl_protocol_returns_bounded_errors(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        source = io.BytesIO(b'{"id":1,"op":"observe"}\nnot-json\n')
        output = io.BytesIO()
        run_stream(worker, source, output)
        rows = [json.loads(line) for line in output.getvalue().splitlines()]
        self.assertEqual(rows[0]["ok"], True)
        self.assertEqual(rows[1]["ok"], False)
        self.assertEqual(rows[1]["error"], "browser operation failed; no action was retried")

    def test_process_request_rejects_extra_protocol_fields(self):
        worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
        with self.assertRaisesRegex(WorkerError, "malformed"):
            process_request(worker, {"id": 1, "op": "observe", "extra": True})


class BridgeRegressionTests(unittest.TestCase):
    def test_loopback_endpoint_contract(self):
        for url in ["http://127.0.0.1:9222", "http://localhost:9222", "http://[::1]:9222"]:
            self.assertEqual(validate_cdp_url(url), url)
        for url in ["https://localhost:9222", "http://example.com:9222", "http://localhost:9222/path", "http://localhost:9222?key=x"]:
            with self.assertRaises(ValueError):
                validate_cdp_url(url)

    def test_changed_candidate_role_or_name_fails_before_click(self):
        for changes in [{"role": "link"}, {"name": "Delete"}, {"visible": False}, {"main_frame": False}]:
            backend = FakeBackend()
            worker = BrowserWorker(backend, ["http://fixture.test"])
            obs = worker.observe()
            backend.elements[3] = replace(backend.elements[3], **changes)
            with self.assertRaises(WorkerError):
                worker.act({"observationId": obs["observationId"], "candidateId": "c-1", "action": "click",
                            "postcondition": {"kind": "url", "value": "http://fixture.test/done"}})
            self.assertEqual(backend.clicks, 0)

    def test_fill_optional_condition_and_post_fill_navigation_guard(self):
        backend = FakeBackend()
        worker = BrowserWorker(backend, ["http://fixture.test"])
        obs = worker.observe()
        result = worker.act({"observationId": obs["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada",
                             "postcondition": {"kind": "text", "value": "Finished"}})
        self.assertTrue(result["verified"])
        obs = worker.observe()
        original = backend.fill
        def redirected(node, text):
            original(node, text)
            backend.url = "https://outside.test"
        backend.fill = redirected
        with self.assertRaisesRegex(WorkerError, "outside"):
            worker.act({"observationId": obs["observationId"], "candidateId": "c-0", "action": "fill", "text": "Ada"})

    def test_fill_url_postcondition_reuses_post_action_tab_state(self):
        backend = FakeBackend()
        original_fill = backend.fill
        def fill_and_navigate(node, text):
            original_fill(node, text)
            backend.url = "http://fixture.test/done"
        backend.fill = fill_and_navigate
        current_tab = backend.current_tab
        backend.current_tab = Mock(side_effect=current_tab)
        worker = BrowserWorker(backend, ["http://fixture.test"])
        observation = worker.observe()
        result = worker.act({
            "observationId": observation["observationId"], "candidateId": "c-0",
            "action": "fill", "text": "Ada",
            "postcondition": {"kind": "url", "value": "http://fixture.test/done"},
        })
        self.assertTrue(result["verified"])
        # observe, fresh pre-mutation revalidation, and one post-action read.
        self.assertEqual(backend.current_tab.call_count, 3)

    def test_oversized_input_and_sensitive_error_are_not_echoed(self):
        worker = BrowserWorker(FakeBackend(), ["http://fixture.test"])
        out = io.BytesIO()
        self.assertEqual(run_stream(worker, io.BytesIO(b"x" * (MAX_LINE_BYTES + 2)), out), 1)
        self.assertLess(len(out.getvalue()), 200)
        worker.backend.current_tab = Mock(side_effect=RuntimeError("private-password"))
        out = io.BytesIO()
        run_stream(worker, io.BytesIO(b'{"id":1,"op":"observe"}\n'), out)
        self.assertNotIn(b"private-password", out.getvalue())

    def test_version_guard_precedes_import_or_daemon_spawn(self):
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        with patch("importlib.metadata.version", return_value="0.1.8"), patch("browser_worker.subprocess.Popen") as spawn:
            with self.assertRaisesRegex(WorkerError, "0.1.13 is required"):
                backend._load_harness()
            spawn.assert_not_called()

    def test_real_backend_list_tabs_uses_one_inventory_with_helper_filtering_and_openers(self):
        helper = Mock()
        helper.cdp.return_value = {"targetInfos": [
            {"type": "page", "targetId": "page", "url": "http://fixture.test/form", "title": "Fixture",
             "openerId": "opener"},
            {"type": "page", "targetId": "internal", "url": "chrome://settings", "title": "Settings"},
            {"type": "page", "targetId": "devtools", "url": "devtools://devtools", "title": "DevTools"},
            {"type": "page", "targetId": "placeholder", "url": "about:blank#startup", "title": "Starting agent 1"},
            {"type": "service_worker", "targetId": "worker", "url": "http://fixture.test/sw", "title": "Worker"},
            {"type": "page", "targetId": "plain", "url": "https://other.test/", "title": "Other", "openerId": ""},
            {"type": "page", "targetId": "malformed-opener", "url": "http://fixture.test/other", "title": "Other",
             "openerId": 42},
        ]}
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = helper

        tabs = backend.list_tabs()

        self.assertEqual([tab["targetId"] for tab in tabs], ["page", "plain", "malformed-opener"])
        self.assertEqual(tabs[0]["openerTargetId"], "opener")
        self.assertIsNone(tabs[1]["openerTargetId"])
        self.assertIsNone(tabs[2]["openerTargetId"])
        helper.cdp.assert_called_once_with("Target.getTargets")
        helper.list_tabs.assert_not_called()

    def test_real_backend_list_tabs_propagates_inventory_errors_without_fallback(self):
        helper = Mock()
        helper.cdp.side_effect = RuntimeError("inventory unavailable")
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = helper

        with self.assertRaisesRegex(RuntimeError, "inventory unavailable"):
            backend.list_tabs()
        helper.cdp.assert_called_once_with("Target.getTargets")
        helper.list_tabs.assert_not_called()

    def test_real_backend_switch_tab_does_not_reenable_harness_domains(self):
        helper = Mock()
        helper.switch_tab.return_value = "session-1"
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = helper

        backend.switch_tab("target-1")

        self.assertEqual(backend._session_id, "session-1")
        self.assertEqual(backend._attached_target, "target-1")
        helper.switch_tab.assert_called_once_with("target-1", activate=False)
        helper.cdp.assert_not_called()

    def test_real_backend_caches_select_all_modifier_for_backend_lifetime(self):
        helper = Mock()
        calls = []
        def cdp(method, **_kwargs):
            calls.append(method)
            return {"userAgent": "Macintosh"} if method == "Browser.getVersion" else {}
        helper.cdp.side_effect = cdp
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = helper
        backend._call = Mock(return_value=True)

        backend.fill(1, "first")
        backend.fill(1, "second")

        self.assertEqual(calls.count("Browser.getVersion"), 1)
        keydowns = [call for call in helper.cdp.call_args_list
                    if call.args[0] == "Input.dispatchKeyEvent" and call.kwargs["type"] == "rawKeyDown"]
        self.assertEqual([call.kwargs["modifiers"] for call in keydowns], [4, 4])

    def test_real_backend_unrestricted_policy_still_requires_one_web_tab(self):
        helper = Mock()
        helper.list_tabs.return_value = [
            {"targetId": "blank", "url": "about:blank"},
            {"targetId": "web", "url": "https://other.example/form"},
        ]
        child = Mock()
        child.poll.return_value = None
        child.wait.return_value = 0
        with patch.dict(os.environ, dict(os.environ)), \
             patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
             patch("browser_worker.subprocess.Popen", return_value=child):
            backend = RealBrowserBackend("http://127.0.0.1:9222", "*")
            helper.switch_tab.assert_called_once_with("web", activate=False)
            backend.close()

    def test_real_backend_owns_one_daemon_and_only_attaches_approved_tab(self):
        helper = Mock()
        helper.list_tabs.return_value = [
            {"targetId": "blank-owned-by-harness", "url": "about:blank"},
            {"targetId": "approved", "url": "http://fixture.test/form"},
            {"targetId": "unrelated", "url": "https://other.test/"},
        ]
        child = Mock()
        child.poll.return_value = None
        child.wait.return_value = 0
        with patch.dict(os.environ, dict(os.environ)), \
             patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
             patch("browser_worker.subprocess.Popen", return_value=child) as spawn:
            backend = RealBrowserBackend("http://127.0.0.1:9222", ["http://fixture.test"])
            root = backend._sandbox
            spawn.assert_called_once()
            self.assertTrue(spawn.call_args.kwargs["start_new_session"])
            helper.switch_tab.assert_called_once_with("approved", activate=False)
            self.assertEqual(os.environ["BU_AUTOSPAWN"], "")
            self.assertEqual(os.environ["BROWSER_USE_API_KEY"], "")
            backend.close()
            backend.close()
            self.assertFalse(root.exists())
            helper._send.assert_called_once_with({"meta": "shutdown"}, response_timeout=0.3)
            child.wait.assert_called_once()

    def test_ambiguous_tabs_and_startup_errors_cleanup(self):
        helper = Mock()
        helper.list_tabs.return_value = [
            {"targetId": str(n), "url": "http://fixture.test/form"} for n in range(2)
        ]
        child = Mock()
        child.poll.return_value = None
        child.wait.return_value = 0
        with patch.dict(os.environ, dict(os.environ)), \
             patch.object(RealBrowserBackend, "_load_harness", lambda self: setattr(self, "_helpers", helper)), \
             patch("browser_worker.subprocess.Popen", return_value=child):
            backend = RealBrowserBackend("http://127.0.0.1:9222", ["http://fixture.test"])
            root = Path(os.environ["BH_HOME"]).parent
            self.assertTrue(root.exists())
            helper.switch_tab.assert_not_called()
            backend.close()
            self.assertFalse(root.exists())
            helper._send.assert_called_once()

    def test_hit_test_blocks_covered_click_and_bulk_fill_uses_insert_text(self):
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = Mock()
        backend._helpers.cdp.return_value = {"model": {"content": [0, 0, 20, 0, 20, 20, 0, 20]}}
        backend._call = Mock(return_value=False)
        with self.assertRaisesRegex(WorkerError, "covered"):
            backend.click(1)
        self.assertEqual(backend._helpers.cdp.call_count, 1)
        backend._call.return_value = True
        backend._helpers.cdp.reset_mock()
        backend._helpers.cdp.return_value = {"userAgent": "Macintosh"}
        backend.fill(1, "hello\nworld")
        calls = backend._helpers.cdp.call_args_list
        insert = [c for c in calls if c.args[0] == "Input.insertText"]
        self.assertEqual(len(insert), 1)
        self.assertEqual(insert[0].kwargs["text"], "hello\nworld")
        keydowns = [c for c in calls if c.args[0] == "Input.dispatchKeyEvent" and c.kwargs["type"] == "rawKeyDown"]
        self.assertEqual(keydowns[0].kwargs["modifiers"], 4)
        backend._helpers.press_key.assert_called_once_with("Backspace")

    def test_call_releases_remote_object_on_browser_exception(self):
        backend = RealBrowserBackend.__new__(RealBrowserBackend)
        backend._helpers = Mock()
        backend._helpers.cdp.side_effect = [{"object": {"objectId": "obj"}}, {"exceptionDetails": {"text": "private"}}, {}]
        with self.assertRaisesRegex(WorkerError, "evaluation failed"):
            backend._call(1, "function () {}")
        self.assertEqual(backend._helpers.cdp.call_args.args[0], "Runtime.releaseObject")


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