Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 98 additions & 6 deletions src/browser_harness/daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -360,13 +360,57 @@ def __init__(self):
self.cdp = None
Comment thread
MagMueller marked this conversation as resolved.
self.session = None
self.target_id = None
self.owned_target_id = None # target created by this named daemon
self._attach_lock = asyncio.Lock()
self.events = deque(maxlen=BUF)
self.dialog = None
self.stop = None # asyncio.Event, set inside start()

async def attach_first_page(self):
"""Attach to a real page (or any page). Sets self.session. Returns attached target or None."""
async with self._attach_lock:
Comment thread
MagMueller marked this conversation as resolved.
Outdated
return await self._attach_first_page_unlocked()

async def _attach_first_page_unlocked(self):
"""Attach while the caller holds _attach_lock."""
targets = (await self.cdp.send_raw("Target.getTargets"))["targetInfos"]
# Named daemons (BU_NAME != "default") share one browser with other
# daemons — attaching to the first page makes parallel daemons fight
# over a single tab (navigations clobber each other). Give each named
# daemon its own dedicated tab instead. REMOTE_ID (cloud) browsers are
# already exclusive to this daemon, so first-page attach stays.
if NAME != "default" and not REMOTE_ID:
# The permission recovery flow can leave chrome://inspect open.
# Clean it up before returning from this early path as well.
if BROWSER_KIND == "local":
await self._close_inspect_tabs(targets)
if self.owned_target_id and not await self._close_owned_target():
raise RuntimeError(
f"cannot close previously owned tab {self.owned_target_id}; refusing to create another"
)

tid = None
try:
tid = (await self.cdp.send_raw("Target.createTarget", {"url": "about:blank"}))["targetId"]
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Outdated
# Record ownership immediately: attach or domain setup can fail
# after Target.createTarget has already made a real tab.
self.owned_target_id = tid
log(f"named daemon {NAME}: created dedicated tab ({tid})")
page = {"targetId": tid, "url": "about:blank", "type": "page"}
self.session = (await self.cdp.send_raw(
"Target.attachToTarget", {"targetId": tid, "flatten": True}
))["sessionId"]
self.target_id = tid
log(f"attached {tid} (about:blank) session={self.session}")
await self._enable_default_domains(self.session)
return page
except Exception:
await self._close_owned_target()
self.session = None
if self.target_id == tid:
self.target_id = None
raise

pages = [t for t in targets if is_real_page(t)]
if not pages:
# Fresh browser (ex: BU cloud) starts w about:blank; reuse it
Expand Down Expand Up @@ -420,6 +464,35 @@ async def _close_inspect_tabs(self, targets):
except OSError:
pass

async def _close_owned_target(self, attempts=1):
"""Best-effort close of the tab created by this named daemon.

`target_id` follows the user's current tab after switch_tab/new_tab,
so it cannot be used for ownership cleanup. Keep the owned target ID
separately. Failed closes retain the ID so this method can retry
without ever targeting the user's currently selected tab.
"""
tid = self.owned_target_id
if not tid or not self.cdp:
return True
attempts = max(1, attempts)
for attempt in range(1, attempts + 1):
try:
await asyncio.wait_for(
self.cdp.send_raw("Target.closeTarget", {"targetId": tid}),
timeout=2,
)
log(f"closed owned tab {tid}")
self.owned_target_id = None
if self.target_id == tid:
self.target_id = None
Comment thread
MagMueller marked this conversation as resolved.
Outdated
return True
except Exception as e:
log(f"close owned tab {tid} (attempt {attempt}/{attempts}): {e}")
if attempt < attempts:
await asyncio.sleep(0.1)
return False

async def _enable_default_domains(self, session_id):
"""Enable Page/DOM/Runtime/Network on a CDP session.

Expand Down Expand Up @@ -572,10 +645,23 @@ async def disable_old():
return {"result": await self.cdp.send_raw(method, params, session_id=sid)}
except Exception as e:
msg = str(e)
if "Session with given id not found" in msg and sid == self.session and sid:
log(f"stale session {sid}, re-attaching")
if await self.attach_first_page():
return {"result": await self.cdp.send_raw(method, params, session_id=self.session)}
if "Session with given id not found" in msg and sid:
# IPC handlers run concurrently. Only one request may replace a
# stale session; later requests reuse that replacement instead
# of closing it and creating another dedicated tab.
async with self._attach_lock:
explicit_session = req.get("session_id")
if sid == self.session:
log(f"stale session {sid}, re-attaching")
if not await self._attach_first_page_unlocked():
return {"error": msg}
elif explicit_session:
return {"error": msg}
if self.session:
try:
return {"result": await self.cdp.send_raw(method, params, session_id=self.session)}
Comment thread
MagMueller marked this conversation as resolved.
Outdated
except Exception as retry_error:
return {"error": str(retry_error)}
return {"error": msg}


Expand Down Expand Up @@ -614,8 +700,14 @@ async def handler(reader, writer):

async def main():
d = Daemon()
await d.start()
await serve(d)
try:
await d.start()
await serve(d)
finally:
# Startup can fail after Target.createTarget has succeeded, and the
# selected tab can change during a normal run. Always clean up by the
# separately tracked owned target ID.
await d._close_owned_target(attempts=3)


def already_running():
Expand Down
228 changes: 228 additions & 0 deletions tests/unit/test_daemon.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import asyncio

import pytest

from browser_harness import daemon


Expand Down Expand Up @@ -293,3 +295,229 @@ def test_current_tab_meta_returns_not_attached_when_no_target_id():
assert result == {"error": "not_attached"}
# No CDP call should have been issued.
assert d.cdp.calls == []


class _AttachCDP(_FakeCDP):
"""FakeCDP with realistic responses for the attach flow."""

def __init__(self, targets=None, fail_method=None):
super().__init__()
self.targets = targets or []
self.created = 0
self.closed = []
self.fail_method = fail_method

async def send_raw(self, method, params=None, session_id=None):
self.calls.append((method, params, session_id))
if method == self.fail_method:
raise RuntimeError(f"simulated {method} failure")
if method == "Target.getTargets":
return {"targetInfos": self.targets}
if method == "Target.createTarget":
self.created += 1
return {"targetId": f"created-{self.created}"}
if method == "Target.attachToTarget":
return {"sessionId": f"session-for-{params['targetId']}"}
if method == "Target.closeTarget":
self.closed.append(params["targetId"])
return {}


def test_named_daemon_creates_dedicated_tab(monkeypatch):
"""Named local/CDP daemons must not fight over the first existing tab."""
monkeypatch.setattr(daemon, "NAME", "worker-a")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
existing = [{"targetId": "someone-elses-tab", "url": "https://example.com/", "type": "page"}]
d = daemon.Daemon()
d.cdp = _AttachCDP(existing)

page = asyncio.run(d.attach_first_page())

assert page["targetId"] == "created-1"
assert d.target_id == "created-1"
assert d.owned_target_id == "created-1"
assert d.session == "session-for-created-1"
attach_calls = [p for (m, p, _s) in d.cdp.calls if m == "Target.attachToTarget"]
assert attach_calls == [{"targetId": "created-1", "flatten": True}]
enabled = {m for (m, _p, s) in d.cdp.calls if s == d.session and m.endswith(".enable")}
assert enabled == {"Page.enable", "DOM.enable", "Runtime.enable", "Network.enable"}


def test_default_daemon_still_attaches_first_page(monkeypatch):
"""The default daemon keeps reusing the user's first real page."""
monkeypatch.setattr(daemon, "NAME", "default")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
existing = [{"targetId": "user-tab", "url": "https://example.com/", "type": "page"}]
d = daemon.Daemon()
d.cdp = _AttachCDP(existing)

page = asyncio.run(d.attach_first_page())

assert page["targetId"] == "user-tab"
assert d.owned_target_id is None
assert d.cdp.created == 0


def test_named_remote_daemon_keeps_first_page_attach(monkeypatch):
"""A cloud browser is exclusive, so a named cloud daemon needs no extra tab."""
monkeypatch.setattr(daemon, "NAME", "r7k2")
monkeypatch.setattr(daemon, "REMOTE_ID", "remote-browser-id")
monkeypatch.setattr(daemon, "BROWSER_KIND", "cloud")
existing = [{"targetId": "cloud-blank", "url": "about:blank", "type": "page"}]
d = daemon.Daemon()
d.cdp = _AttachCDP(existing)

page = asyncio.run(d.attach_first_page())

assert page["targetId"] == "cloud-blank"
assert d.owned_target_id is None
assert d.cdp.created == 0


def test_named_reattach_closes_previous_owned_tab_before_creating_another(monkeypatch):
"""A stale-session retry must not leak the old dedicated tab."""
monkeypatch.setattr(daemon, "NAME", "worker-a")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
d = daemon.Daemon()
d.cdp = _AttachCDP()

asyncio.run(d.attach_first_page())
d.target_id = "user-selected-tab"
asyncio.run(d.attach_first_page())

assert d.cdp.closed == ["created-1"]
assert d.target_id == "created-2"
assert d.owned_target_id == "created-2"


def test_concurrent_stale_requests_share_one_reattached_tab(monkeypatch):
"""Concurrent stale-session recovery must create only one replacement tab."""
class _ConcurrentStaleCDP(_AttachCDP):
def __init__(self):
super().__init__()
self.release_stale = None
self.stale_calls = 0
self.retry_sessions = []

async def send_raw(self, method, params=None, session_id=None):
if method == "Runtime.evaluate":
self.calls.append((method, params, session_id))
if session_id == "stale-session":
self.stale_calls += 1
if self.stale_calls == 2:
self.release_stale.set()
await self.release_stale.wait()
raise RuntimeError("Session with given id not found")
self.retry_sessions.append(session_id)
return {"session": session_id}
return await super().send_raw(method, params, session_id)

async def run():
monkeypatch.setattr(daemon, "NAME", "worker-a")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
d = daemon.Daemon()
d.cdp = _ConcurrentStaleCDP()
d.cdp.release_stale = asyncio.Event()
d.session = "stale-session"
d.target_id = "old-owned-tab"
d.owned_target_id = "old-owned-tab"
request = {"method": "Runtime.evaluate", "params": {"expression": "1"}}
results = await asyncio.gather(d.handle(request), d.handle(request))
return d, results

d, results = asyncio.run(run())

assert d.cdp.created == 1
assert d.cdp.closed == ["old-owned-tab"]
assert d.owned_target_id == "created-1"
assert d.cdp.retry_sessions == ["session-for-created-1", "session-for-created-1"]
assert results == [
{"result": {"session": "session-for-created-1"}},
{"result": {"session": "session-for-created-1"}},
]


def test_named_attach_failure_closes_created_tab(monkeypatch):
"""If attach fails after Target.createTarget, the new tab is cleaned up."""
monkeypatch.setattr(daemon, "NAME", "worker-a")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
d = daemon.Daemon()
d.cdp = _AttachCDP(fail_method="Target.attachToTarget")

with pytest.raises(RuntimeError, match="Target.attachToTarget"):
asyncio.run(d.attach_first_page())

assert d.cdp.closed == ["created-1"]
assert d.owned_target_id is None
assert d.session is None
assert d.target_id is None


def test_named_local_attach_cleans_inspect_tabs_before_return(monkeypatch):
"""The named-daemon early path must retain local inspect-tab cleanup."""
monkeypatch.setattr(daemon, "NAME", "worker-a")
monkeypatch.setattr(daemon, "REMOTE_ID", None)
monkeypatch.setattr(daemon, "BROWSER_KIND", "local")
monkeypatch.setattr(daemon, "harness_opened_inspect", lambda: True)
inspect = {"targetId": "inspect-tab", "url": "chrome://inspect/#remote-debugging", "type": "page"}
d = daemon.Daemon()
d.cdp = _AttachCDP([inspect])

asyncio.run(d.attach_first_page())

methods = [method for method, _params, _session in d.cdp.calls]
assert methods.index("Target.closeTarget") < methods.index("Target.createTarget")
assert d.cdp.closed == ["inspect-tab"]


def test_owned_cleanup_uses_owned_target_after_tab_switch():
"""Shutdown must close the daemon tab, not the user's selected tab."""
d = daemon.Daemon()
d.cdp = _AttachCDP()
d.owned_target_id = "daemon-owned-tab"
d.target_id = "user-selected-tab"

asyncio.run(d._close_owned_target())

assert d.cdp.closed == ["daemon-owned-tab"]
assert d.owned_target_id is None
assert d.target_id == "user-selected-tab"


def test_main_retries_owned_tab_cleanup_when_startup_fails(monkeypatch):
"""Shutdown retries transient close timeouts before exiting."""
class _FlakyCloseCDP(_AttachCDP):
def __init__(self):
super().__init__()
self.close_attempts = 0

async def send_raw(self, method, params=None, session_id=None):
if method == "Target.closeTarget":
self.calls.append((method, params, session_id))
self.close_attempts += 1
if self.close_attempts < 3:
raise TimeoutError("simulated close timeout")
self.closed.append(params["targetId"])
return {}
return await super().send_raw(method, params, session_id)

d = daemon.Daemon()
d.cdp = _FlakyCloseCDP()

async def failed_start():
d.owned_target_id = "created-before-startup-failure"
raise RuntimeError("simulated startup failure")

d.start = failed_start
monkeypatch.setattr(daemon, "Daemon", lambda: d)

with pytest.raises(RuntimeError, match="startup failure"):
asyncio.run(daemon.main())

assert d.cdp.close_attempts == 3
assert d.cdp.closed == ["created-before-startup-failure"]
assert d.owned_target_id is None