diff --git a/coworker/connectors/browser_automation.py b/coworker/connectors/browser_automation.py index 93236eba..1c921505 100644 --- a/coworker/connectors/browser_automation.py +++ b/coworker/connectors/browser_automation.py @@ -68,6 +68,7 @@ class _BrowserController: self._browser = None self._context = None self._page = None + self._tracked_page_ids: set[int] = set() self._error: Optional[str] = None self._executor = ThreadPoolExecutor( max_workers=1, thread_name_prefix="coworker-browser" @@ -85,6 +86,30 @@ class _BrowserController: "controls": [], } + def _open_pages(self) -> list[Any]: + if self._context is None: + return [] + return [page for page in self._context.pages if not page.is_closed()] + + def _activate_page(self, page: Any) -> None: + """Make a newly opened tab the target for subsequent browser tools.""" + with self._lock: + if page.is_closed(): + return + self._page = page + page_id = id(page) + if page_id not in self._tracked_page_ids: + self._tracked_page_ids.add(page_id) + page.on("close", lambda _closed_page: self._page_closed(page)) + + def _page_closed(self, page: Any) -> None: + with self._lock: + self._tracked_page_ids.discard(id(page)) + if self._page is not page: + return + pages = self._open_pages() + self._page = pages[-1] if pages else None + def _touch(self, **changes: Any) -> None: self._state.update(changes) self._state["updated_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) @@ -128,7 +153,11 @@ class _BrowserController: self._context = self._browser.new_context( viewport={"width": 1280, "height": 900} ) - self._page = self._context.new_page() + # Playwright keeps every tab in one BrowserContext. A page opened by + # window.open() is the user's new working surface, so make it the active + # target for screenshot/snapshot/click and the other browser tools. + self._context.on("page", self._activate_page) + self._activate_page(self._context.new_page()) self._touch( open=True, status="open", last_action="open browser", last_error="" ) @@ -159,6 +188,7 @@ class _BrowserController: self._browser = None self._context = None self._page = None + self._tracked_page_ids.clear() self._touch(open=False, status="closed", url="", title="", controls=[]) return {"ok": True} @@ -170,6 +200,56 @@ class _BrowserController: self._refresh_page_state() return dict(self._state) + def tabs(self) -> dict[str, Any]: + return self._submit(self._tabs_locked) + + def _tabs_locked(self) -> dict[str, Any]: + with self._lock: + tabs = [] + for index, page in enumerate(self._open_pages()): + try: + title = page.title() + except Exception: + title = "" + tabs.append( + { + "index": index, + "active": page is self._page, + "url": page.url, + "title": title, + } + ) + return {"tabs": tabs} + + def switch_tab(self, index: int) -> dict[str, Any]: + return self._submit(lambda: self._switch_tab_locked(index)) + + def _switch_tab_locked(self, index: int) -> dict[str, Any]: + with self._lock: + pages = self._open_pages() + if not pages: + return {"error": "no browser tabs are open"} + if isinstance(index, bool) or not isinstance(index, int): + return {"error": "tab index must be an integer"} + if index < 0 or index >= len(pages): + return { + "error": f"tab index {index} is out of range", + "tab_count": len(pages), + } + page = pages[index] + try: + page.bring_to_front() + self._page = page + self._refresh_page_state() + except Exception as exc: + return {"error": str(exc)} + return { + "ok": True, + "index": index, + "url": page.url, + "title": self._state.get("title", ""), + } + def screenshot(self) -> dict[str, Any]: return self._submit(self._screenshot_locked) @@ -425,7 +505,7 @@ def make_browser_automation_tools( browser_read_page, _schema( "browser_read_page", - "Read the current page: its text plus visible controls and selector " + "Read the active browser tab: its text plus visible controls and selector " "hints (for browser_click/browser_type). Not an image — use " "browser_screenshot for pixels.", {"max_chars": {"type": "integer"}}, @@ -566,6 +646,40 @@ def make_browser_automation_tools( ) ) + def browser_list_tabs() -> dict[str, Any]: + return _BROWSER.tabs() + + browser_list_tabs.__name__ = "browser_list_tabs" + tools.append( + _attach( + browser_list_tabs, + _schema( + "browser_list_tabs", + "List open browser tabs, including index, title, URL, and which tab is active.", + {}, + [], + ), + approval=True, + ) + ) + + def browser_switch_tab(index: int) -> dict[str, Any]: + return _BROWSER.switch_tab(index) + + browser_switch_tab.__name__ = "browser_switch_tab" + tools.append( + _attach( + browser_switch_tab, + _schema( + "browser_switch_tab", + "Switch subsequent browser actions to the tab at an index returned by browser_list_tabs.", + {"index": {"type": "integer"}}, + ["index"], + ), + approval=True, + ) + ) + def browser_screenshot(path: str = "") -> dict[str, Any]: if path: _target, target_err = _writable_target(path) @@ -592,7 +706,7 @@ def make_browser_automation_tools( browser_screenshot, _schema( "browser_screenshot", - "Save a full-page screenshot of the current browser page and return the local path.", + "Save a full-page screenshot of the active browser tab and return the local path.", {"path": {"type": "string"}}, [], ), diff --git a/coworker/connectors/tool_defs.py b/coworker/connectors/tool_defs.py index d0a7a860..47559302 100644 --- a/coworker/connectors/tool_defs.py +++ b/coworker/connectors/tool_defs.py @@ -72,6 +72,20 @@ TOOL_DEFS: tuple[ConnectorToolDef, ...] = ( ConnectorToolDef( "browser", "browser_wait", "Wait", "read", "Wait for time or an element." ), + ConnectorToolDef( + "browser", + "browser_list_tabs", + "List tabs", + "read", + "List open browser tabs and identify the active tab.", + ), + ConnectorToolDef( + "browser", + "browser_switch_tab", + "Switch tab", + "read", + "Switch subsequent browser actions to another tab.", + ), ConnectorToolDef( "browser", # Writes an image file to a resolved path (creating parents) — a local write, diff --git a/tests/test_browser_tabs.py b/tests/test_browser_tabs.py new file mode 100644 index 00000000..36bb5dbb --- /dev/null +++ b/tests/test_browser_tabs.py @@ -0,0 +1,131 @@ +from __future__ import annotations + +from coworker.connectors.browser_automation import ( + _BrowserController, + make_browser_automation_tools, +) + + +class _FakePage: + def __init__(self, url: str, title: str) -> None: + self.url = url + self._title = title + self._closed = False + self._handlers: dict[str, list] = {} + self.brought_to_front = False + + def on(self, event: str, callback) -> None: + self._handlers.setdefault(event, []).append(callback) + + def is_closed(self) -> bool: + return self._closed + + def title(self) -> str: + return self._title + + def bring_to_front(self) -> None: + self.brought_to_front = True + + def evaluate(self, _script: str) -> dict: + return { + "title": self._title, + "url": self.url, + "text": self._title, + "controls": [], + } + + def close(self) -> None: + self._closed = True + for callback in self._handlers.get("close", []): + callback(self) + + +class _FakeContext: + def __init__(self, *pages: _FakePage) -> None: + self.pages = list(pages) + + +def _controller_with_pages(*pages: _FakePage) -> _BrowserController: + controller = _BrowserController() + controller._context = _FakeContext(*pages) + for page in pages: + controller._activate_page(page) + return controller + + +def test_new_popup_becomes_active_for_subsequent_browser_calls(): + first = _FakePage("https://example.com/start", "Start") + popup = _FakePage("https://example.com/result", "Result") + controller = _controller_with_pages(first) + + controller._context.pages.append(popup) + controller._activate_page(popup) + + assert controller.call("probe", lambda page: {"url": page.url}) == { + "url": "https://example.com/result" + } + assert controller.tabs() == { + "tabs": [ + { + "index": 0, + "active": False, + "url": "https://example.com/start", + "title": "Start", + }, + { + "index": 1, + "active": True, + "url": "https://example.com/result", + "title": "Result", + }, + ] + } + + +def test_switch_tab_changes_target_and_brings_it_to_front(): + first = _FakePage("https://example.com/one", "One") + second = _FakePage("https://example.com/two", "Two") + controller = _controller_with_pages(first, second) + + assert controller.switch_tab(0) == { + "ok": True, + "index": 0, + "url": "https://example.com/one", + "title": "One", + } + assert first.brought_to_front is True + assert controller.call("probe", lambda page: {"url": page.url})["url"].endswith( + "/one" + ) + + +def test_closing_active_tab_falls_back_to_most_recent_open_tab(): + first = _FakePage("https://example.com/one", "One") + second = _FakePage("https://example.com/two", "Two") + controller = _controller_with_pages(first, second) + + second.close() + + assert controller.call("probe", lambda page: {"url": page.url}) == { + "url": "https://example.com/one" + } + + +def test_switch_tab_validates_index(): + controller = _controller_with_pages(_FakePage("https://example.com", "Example")) + + assert "integer" in controller.switch_tab(True)["error"] + assert controller.switch_tab(2) == { + "error": "tab index 2 is out of range", + "tab_count": 1, + } + + +def test_browser_tab_tools_are_exposed_with_selection_schema(): + tools = {tool.__name__: tool for tool in make_browser_automation_tools()} + + assert "browser_list_tabs" in tools + assert "browser_switch_tab" in tools + schema = tools["browser_switch_tab"].__coworker_schema__["function"]["parameters"] + assert schema["properties"]["index"]["type"] == "integer" + assert schema["required"] == ["index"]