diff --git a/SKILL.md b/SKILL.md
index 4633a690..102e71bc 100644
--- a/SKILL.md
+++ b/SKILL.md
@@ -26,6 +26,10 @@ PY
 - Invoke as `browser-harness`. Use heredocs for multi-line commands.
 - Helpers are pre-imported. `run.py` calls `ensure_daemon()` before `exec`.
 - First navigation is `new_tab(url)`, not `goto_url(url)`.
+- `new_tab()` and `switch_tab()` attach and move the horse marker without
+  changing Chrome's visible tab. Screenshots and normal CDP input work in the
+  background; call `activate_tab(target)` only when the user explicitly asks
+  or a page demonstrably pauses rendering while hidden.
 - The normal local flow attaches to the running Chrome/Chromium CDP endpoint. No browser ids or local profile selection.
 
 ## Local Chrome
diff --git a/interaction-skills/connection.md b/interaction-skills/connection.md
index 85e264c2..8bf8cc5b 100644
--- a/interaction-skills/connection.md
+++ b/interaction-skills/connection.md
@@ -4,7 +4,7 @@
 
 When Chrome opens fresh, the only CDP `type: "page"` targets are `chrome://inspect` and `chrome://omnibox-popup.top-chrome/` (a 1px invisible viewport). If the daemon attaches to the omnibox popup, all subsequent work — including `new_tab()` and `goto_url()` — happens on tabs that exist in CDP but may not be visible in the Chrome UI.
 
-The daemon's `attach_first_page()` handles this by creating an `about:blank` tab when no real pages exist. If you still end up on an invisible tab, use `switch_tab()` which calls `Target.activateTarget` to bring the tab to front.
+The daemon's `attach_first_page()` handles this by creating an `about:blank` tab when no real pages exist. If you still end up on an invisible tab, use `switch_tab()` to attach to the real tab; call `activate_tab()` only when Chrome must visibly show it.
 
 ## Startup sequence
 
@@ -12,7 +12,7 @@ The daemon's `attach_first_page()` handles this by creating an `about:blank` tab
 2. If stale sockets exist but daemon is dead, clean them up
 3. List open tabs with `list_tabs()` to see what's available
 4. `ensure_real_tab()` attaches to a real page
-5. `switch_tab(target_id)` both attaches AND activates (brings to front)
+5. `switch_tab(target_id)` attaches without changing the visible Chrome tab; use `activate_tab(target_id)` for an explicit visible switch
 
 ```python
 if not daemon_alive():
@@ -31,16 +31,20 @@ tab = ensure_real_tab()
 
 ## Bringing Chrome to front
 
-If Chrome is behind other windows or on another desktop:
+If Chrome is behind other windows or on another desktop and the user explicitly wants it shown:
 
 ```python
 import subprocess
 subprocess.run(["osascript", "-e", 'tell application "Google Chrome" to activate'])
 ```
 
+For normal agent work, do not activate Chrome. Screenshots and CDP input work
+on the attached background tab; activate only for a page that demonstrably
+pauses visibility-dependent rendering while hidden.
+
 ## Navigating
 
-Prefer navigating an existing tab over `new_tab()`. Tabs created via CDP's `Target.createTarget` are visible but may open behind the active tab.
+Prefer navigating an existing tab over `new_tab()`. Harness-created tabs open in the background.
 
 ```python
 tab = ensure_real_tab()
diff --git a/interaction-skills/tabs.md b/interaction-skills/tabs.md
index 39ed5b3b..357d1a1c 100644
--- a/interaction-skills/tabs.md
+++ b/interaction-skills/tabs.md
@@ -7,9 +7,9 @@ Use **CDP for control**, **UI automation for user-visible order**.
 ```python
 tabs = list_tabs()                    # includes chrome:// pages too
 real_tabs = list_tabs(include_chrome=False)
-tid = new_tab("https://example.com")  # create + attach
-switch_tab(tid)                       # attach harness to tab
-cdp("Target.activateTarget", targetId=tid)  # show it in Chrome
+tid = new_tab("https://example.com")  # create + attach in the background
+switch_tab(tid)                       # attach harness, move the horse marker
+activate_tab(tid)                     # optional: explicitly show it in Chrome
 print(current_tab())
 print(page_info())
 ```
@@ -61,7 +61,9 @@ Typical tools:
 
 ## Rules that held up in practice
 
-- `switch_tab()` is **not enough** if the user expects Chrome to visibly change.
+- `switch_tab()` intentionally does **not** change Chrome's visible tab.
+- Static screenshots and normal CDP input work on the attached background tab.
+- `activate_tab()` is the explicit opt-in for visibility-dependent rendering or a user-requested visible switch.
 - `Target.activateTarget` is the CDP-side "show this tab".
 - `list_tabs()` includes `chrome://newtab/` by default; ask for `include_chrome=False` when you want only real pages.
 - `chrome://omnibox-popup.top-chrome/` can appear as a fake page target; ignore it for user-facing tab lists.
diff --git a/src/browser_harness/daemon.py b/src/browser_harness/daemon.py
index de910171..3783ca4e 100644
--- a/src/browser_harness/daemon.py
+++ b/src/browser_harness/daemon.py
@@ -360,13 +360,56 @@ def __init__(self):
         self.cdp = None
         self.session = None
         self.target_id = None
+        self.dedicated_target_id = None
+        self._dedicated_target_lock = asyncio.Lock()
+        self._session_state_lock = asyncio.Lock()
+        self._session_replacements = {}
         self.events = deque(maxlen=BUF)
         self.dialog = None
         self.stop = None  # asyncio.Event, set inside start()
 
-    async def attach_first_page(self):
+    async def attach_first_page(self, replaces_session=None, enable_domains=True):
         """Attach to a real page (or any page). Sets self.session. Returns attached target or None."""
         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)
+            pages_by_id = {t["targetId"]: t for t in targets if t["type"] == "page"}
+            # A stale CDP session does not necessarily mean its tab disappeared.
+            # Reattach to the current tab first, then the daemon's dedicated tab.
+            page = pages_by_id.get(self.target_id) or pages_by_id.get(self.dedicated_target_id)
+            if page is None:
+                # Two stale IPC requests can recover concurrently. Recheck
+                # inside a narrow lock so they share one replacement tab.
+                async with self._dedicated_target_lock:
+                    refreshed = (await self.cdp.send_raw("Target.getTargets"))["targetInfos"]
+                    pages_by_id = {t["targetId"]: t for t in refreshed if t["type"] == "page"}
+                    page = pages_by_id.get(self.target_id) or pages_by_id.get(self.dedicated_target_id)
+                    if page is None:
+                        tid = (await self.cdp.send_raw(
+                            "Target.createTarget", {"url": "about:blank", "background": True}
+                        ))["targetId"]
+                        self.dedicated_target_id = tid
+                        log(f"named daemon {NAME}: created dedicated tab ({tid})")
+                        page = {"targetId": tid, "url": "about:blank", "type": "page"}
+            tid = page["targetId"]
+            self.session = (await self.cdp.send_raw(
+                "Target.attachToTarget", {"targetId": tid, "flatten": True}
+            ))["sessionId"]
+            self._record_session_replacement(replaces_session, self.session)
+            self.target_id = tid
+            log(f"attached {tid} ({page.get('url','')[:80]}) session={self.session}")
+            if enable_domains:
+                await self._enable_default_domains(self.session)
+            return page
+
         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
@@ -385,12 +428,15 @@ async def attach_first_page(self):
                 take_over = inspect_tabs[0]["targetId"]
         if not pages:
             # No usable pages - create one instead of attaching to omnibox popup.
-            tid = (await self.cdp.send_raw("Target.createTarget", {"url": "about:blank"}))["targetId"]
+            tid = (await self.cdp.send_raw(
+                "Target.createTarget", {"url": "about:blank", "background": True}
+            ))["targetId"]
             log(f"no real pages found, created about:blank ({tid})")
             pages = [{"targetId": tid, "url": "about:blank", "type": "page"}]
         self.session = (await self.cdp.send_raw(
             "Target.attachToTarget", {"targetId": pages[0]["targetId"], "flatten": True}
         ))["sessionId"]
+        self._record_session_replacement(replaces_session, self.session)
         self.target_id = pages[0]["targetId"]
         log(f"attached {pages[0]['targetId']} ({pages[0].get('url','')[:80]}) session={self.session}")
         if take_over:
@@ -401,7 +447,8 @@ async def attach_first_page(self):
                 log(f"take over inspect tab {take_over}: {e}")
         if BROWSER_KIND == "local":
             await self._close_inspect_tabs(targets)
-        await self._enable_default_domains(self.session)
+        if enable_domains:
+            await self._enable_default_domains(self.session)
         return pages[0]
 
     async def _close_inspect_tabs(self, targets):
@@ -444,6 +491,19 @@ async def enable_one(d):
                 log(f"enable {d} on {session_id}: {e}")
         await asyncio.gather(*(enable_one(d) for d in ("Page", "DOM", "Runtime", "Network")))
 
+    def _record_session_replacement(self, stale_session, replacement_session):
+        """Remember which recovered session still controls the same tab."""
+        if not stale_session or not replacement_session or stale_session == replacement_session:
+            return
+        # Preserve chains so requests delayed across multiple recoveries still
+        # land on their original tab, never whichever tab is current now.
+        for source, replacement in list(self._session_replacements.items()):
+            if replacement == stale_session:
+                self._session_replacements[source] = replacement_session
+        self._session_replacements[stale_session] = replacement_session
+        while len(self._session_replacements) > 32:
+            self._session_replacements.pop(next(iter(self._session_replacements)))
+
     async def start(self):
         self.stop = asyncio.Event()
         url = get_ws_url()
@@ -526,9 +586,11 @@ async def handle(self, req):
                 }
             return {"target_id": self.target_id, "session_id": self.session, "page": page}
         if meta == "set_session":
-            old_session = self.session
-            self.session = req.get("session_id")
-            self.target_id = req.get("target_id") or self.target_id
+            async with self._session_state_lock:
+                old_session = self.session
+                self.session = req.get("session_id")
+                self.target_id = req.get("target_id") or self.target_id
+                new_session = self.session
             # Run the old-session Network.disable (defense in depth — keeps
             # background-tab traffic out of the global event buffer; the
             # consumer-side filter in wait_for_network_idle is the actual
@@ -538,7 +600,7 @@ async def handle(self, req):
             # even on a remote daemon — sequentially these would have stacked
             # to ~22s worst case.
             tasks = []
-            if old_session and old_session != self.session:
+            if old_session and old_session != new_session:
                 async def disable_old():
                     try:
                         await asyncio.wait_for(
@@ -547,7 +609,7 @@ async def disable_old():
                         )
                     except Exception: pass
                 tasks.append(disable_old())
-            tasks.append(self._enable_default_domains(self.session))
+            tasks.append(self._enable_default_domains(new_session))
             await asyncio.gather(*tasks)
             # 🐴 tab-marker title prefix is purely cosmetic — fire-and-forget so
             # it doesn't add to the synchronous IPC budget.
@@ -555,11 +617,11 @@ async def disable_old():
                 self.cdp.send_raw(
                     "Runtime.evaluate",
                     {"expression": "if(!document.title.startsWith('\U0001F434'))document.title='\U0001F434 '+document.title"},
-                    session_id=self.session,
+                    session_id=new_session,
                 ),
                 timeout=2,
             )))
-            return {"session_id": self.session}
+            return {"session_id": new_session}
         if meta == "pending_dialog": return {"dialog": self.dialog}
         if meta == "shutdown":    self.stop.set(); return {"ok": True}
 
@@ -572,10 +634,34 @@ 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:
+                # Explicit session callers asked for that exact session; do not
+                # silently redirect them to the daemon's current tab.
+                if req.get("session_id"):
+                    return {"error": msg}
+                recovered_here = False
+                async with self._session_state_lock:
+                    replacement_session = self._session_replacements.get(sid)
+                    if replacement_session is None and sid == self.session:
+                        log(f"stale session {sid}, re-attaching")
+                        if not await self.attach_first_page(
+                            replaces_session=sid, enable_domains=False
+                        ):
+                            return {"error": msg}
+                        replacement_session = self._session_replacements.get(sid)
+                        recovered_here = replacement_session is not None
+                if recovered_here:
+                    await self._enable_default_domains(replacement_session)
+                # Retry only on a session known to replace this exact stale
+                # session. self.session may instead have changed because the
+                # user deliberately switched tabs while this request waited.
+                if replacement_session:
+                    try:
+                        return {"result": await self.cdp.send_raw(
+                            method, params, session_id=replacement_session
+                        )}
+                    except Exception as retry_error:
+                        return {"error": str(retry_error)}
             return {"error": msg}
 
 
diff --git a/src/browser_harness/helpers.py b/src/browser_harness/helpers.py
index f6ec182e..95a6dc31 100644
--- a/src/browser_harness/helpers.py
+++ b/src/browser_harness/helpers.py
@@ -291,15 +291,35 @@ def _mark_tab():
     try: cdp("Runtime.evaluate", expression="if(!document.title.startsWith('\U0001F434'))document.title='\U0001F434 '+document.title")
     except Exception: pass
 
-def switch_tab(target):
+def _target_id(target):
+    """Accept a raw target id or a tab dict returned by the helpers."""
+    return (target.get("targetId") or target.get("target_id")) if isinstance(target, dict) else target
+
+def activate_tab(target):
+    """Make a target the visible Chrome tab.
+
+    This is intentionally separate from switch_tab(): attaching the agent to a
+    target does not require taking over the user's visible Chrome tab.
+    """
+    target_id = _target_id(target)
+    cdp("Target.activateTarget", targetId=target_id)
+    return target_id
+
+def switch_tab(target, activate=False):
+    """Attach the agent without changing Chrome's visible tab by default.
+
+    Pass activate=True only when Chrome must visibly show the target. The horse
+    marker still moves to the attached target so the user can find it.
+    """
     # Accept either a raw targetId string or the dict returned by current_tab() / list_tabs(),
     # so `switch_tab(current_tab())` works without a manual ["targetId"] dance.
-    target_id = (target.get("targetId") or target.get("target_id")) if isinstance(target, dict) else target
+    target_id = _target_id(target)
     # Unmark old tab. Horse emoji is a surrogate pair in JS UTF-16 strings (2 code units),
     # plus the trailing space = 3 code units, so slice(3) cleanly removes the prefix.
     try: cdp("Runtime.evaluate", expression="if(document.title.startsWith('\U0001F434 '))document.title=document.title.slice(3)")
     except Exception: pass
-    cdp("Target.activateTarget", targetId=target_id)
+    if activate:
+        activate_tab(target_id)
     sid = cdp("Target.attachToTarget", targetId=target_id, flatten=True)["sessionId"]
     _send({"meta": "set_session", "session_id": sid, "target_id": target_id})
     _mark_tab()
@@ -323,7 +343,7 @@ def new_tab(url="about:blank"):
                 return cur.get("targetId") or cur.get("target_id")
         except Exception:
             pass
-    tid = cdp("Target.createTarget", url="about:blank")["targetId"]
+    tid = cdp("Target.createTarget", url="about:blank", background=True)["targetId"]
     switch_tab(tid)
     if url != "about:blank":
         goto_url(url)
@@ -332,7 +352,7 @@ def new_tab(url="about:blank"):
 def close_tab(target=None):
     """Close a tab. If `target` is omitted, closes the currently attached tab.
     Accepts a raw targetId string or a dict from list_tabs()/current_tab()."""
-    target_id = (target.get("targetId") or target.get("target_id")) if isinstance(target, dict) else target
+    target_id = _target_id(target)
     if target_id is None:
         target_id = current_tab()["targetId"]
     cdp("Target.closeTarget", targetId=target_id)
diff --git a/tests/unit/test_daemon.py b/tests/unit/test_daemon.py
index 90c5bc85..669b530a 100644
--- a/tests/unit/test_daemon.py
+++ b/tests/unit/test_daemon.py
@@ -1,5 +1,7 @@
 import asyncio
 
+import pytest
+
 from browser_harness import daemon
 
 
@@ -293,3 +295,431 @@ 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
+            tid = f"created-{self.created}"
+            self.targets.append({"targetId": tid, "url": "about:blank", "type": "page"})
+            return {"targetId": tid}
+        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.dedicated_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}]
+    create_calls = [p for (m, p, _s) in d.cdp.calls if m == "Target.createTarget"]
+    assert create_calls == [{"url": "about:blank", "background": 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.dedicated_target_id is None
+    assert d.cdp.created == 0
+
+
+def test_default_daemon_creates_missing_page_in_background(monkeypatch):
+    """Fallback tabs must not steal the user's foreground Chrome tab."""
+    monkeypatch.setattr(daemon, "NAME", "default")
+    monkeypatch.setattr(daemon, "REMOTE_ID", None)
+    monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
+    d = daemon.Daemon()
+    d.cdp = _AttachCDP()
+
+    page = asyncio.run(d.attach_first_page())
+
+    assert page["targetId"] == "created-1"
+    create_calls = [p for (m, p, _s) in d.cdp.calls if m == "Target.createTarget"]
+    assert create_calls == [{"url": "about:blank", "background": True}]
+
+
+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.dedicated_target_id is None
+    assert d.cdp.created == 0
+
+
+def test_named_reattach_reuses_dedicated_tab(monkeypatch):
+    """A stale CDP session should not replace a tab that still exists."""
+    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())
+    asyncio.run(d.attach_first_page())
+
+    assert d.cdp.created == 1
+    assert d.cdp.closed == []
+    assert d.target_id == "created-1"
+    assert d.dedicated_target_id == "created-1"
+
+
+def test_named_reattach_keeps_selected_tab_when_it_still_exists(monkeypatch):
+    """A deliberate switch_tab remains the active tab after session recovery."""
+    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.cdp.targets.append({"targetId": "selected-tab", "url": "https://example.com", "type": "page"})
+    d.target_id = "selected-tab"
+    asyncio.run(d.attach_first_page())
+
+    assert d.cdp.created == 1
+    assert d.cdp.closed == []
+    assert d.target_id == "selected-tab"
+    assert d.dedicated_target_id == "created-1"
+    assert d.session == "session-for-selected-tab"
+
+
+def test_named_reattach_creates_replacement_only_when_tab_is_gone(monkeypatch):
+    """If the user closes the dedicated tab, the daemon creates one replacement."""
+    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.cdp.targets = [t for t in d.cdp.targets if t["targetId"] != "created-1"]
+    asyncio.run(d.attach_first_page())
+
+    assert d.cdp.created == 2
+    assert d.cdp.closed == []
+    assert d.target_id == "created-2"
+    assert d.dedicated_target_id == "created-2"
+
+
+def test_concurrent_named_reattach_creates_one_replacement(monkeypatch):
+    """Concurrent recovery after a user closes the tab shares one replacement."""
+    class _ConcurrentAttachCDP(_AttachCDP):
+        def __init__(self):
+            super().__init__()
+            self.get_calls = 0
+            self.first_gets_done = asyncio.Event()
+
+        async def send_raw(self, method, params=None, session_id=None):
+            if method == "Target.getTargets":
+                self.calls.append((method, params, session_id))
+                snapshot = list(self.targets)
+                self.get_calls += 1
+                if self.get_calls <= 2:
+                    if self.get_calls == 2:
+                        self.first_gets_done.set()
+                    await self.first_gets_done.wait()
+                return {"targetInfos": snapshot}
+            return await super().send_raw(method, params, session_id)
+
+    async def run():
+        d = daemon.Daemon()
+        d.cdp = _ConcurrentAttachCDP()
+        pages = await asyncio.gather(d.attach_first_page(), d.attach_first_page())
+        return d, pages
+
+    monkeypatch.setattr(daemon, "NAME", "worker-a")
+    monkeypatch.setattr(daemon, "REMOTE_ID", None)
+    monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
+    d, pages = asyncio.run(run())
+
+    assert [page["targetId"] for page in pages] == ["created-1", "created-1"]
+    assert d.cdp.created == 1
+    assert d.cdp.closed == []
+    assert d.target_id == "created-1"
+    assert d.dedicated_target_id == "created-1"
+
+
+def test_named_attach_failure_reuses_created_tab_on_retry(monkeypatch):
+    """A transient attach failure leaves the tab available for the next retry."""
+    class _FailOnceAttachCDP(_AttachCDP):
+        def __init__(self):
+            super().__init__()
+            self.fail_attach = True
+
+        async def send_raw(self, method, params=None, session_id=None):
+            if method == "Target.attachToTarget" and self.fail_attach:
+                self.calls.append((method, params, session_id))
+                self.fail_attach = False
+                raise RuntimeError("simulated Target.attachToTarget failure")
+            return await super().send_raw(method, params, session_id)
+
+    monkeypatch.setattr(daemon, "NAME", "worker-a")
+    monkeypatch.setattr(daemon, "REMOTE_ID", None)
+    monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
+    d = daemon.Daemon()
+    d.cdp = _FailOnceAttachCDP()
+
+    with pytest.raises(RuntimeError, match="Target.attachToTarget"):
+        asyncio.run(d.attach_first_page())
+    page = asyncio.run(d.attach_first_page())
+
+    assert page["targetId"] == "created-1"
+    assert d.cdp.created == 1
+    assert d.cdp.closed == []
+    assert d.dedicated_target_id == "created-1"
+
+
+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_shutdown_leaves_dedicated_tab_open(monkeypatch):
+    """The real serve shutdown path never closes a working or user tab."""
+    d = daemon.Daemon()
+    d.cdp = _AttachCDP()
+
+    async def start():
+        d.dedicated_target_id = "daemon-tab"
+        d.target_id = "user-selected-tab"
+        d.stop = asyncio.Event()
+        d.stop.set()
+
+    async def wait_forever(*_args):
+        await asyncio.Event().wait()
+
+    d.start = start
+    monkeypatch.setattr(daemon, "Daemon", lambda: d)
+    monkeypatch.setattr(daemon.ipc, "serve", wait_forever)
+    monkeypatch.setattr(daemon.ipc, "sock_addr", lambda _name: "test-socket")
+    monkeypatch.setattr(daemon.ipc, "cleanup_endpoint", lambda _name: None)
+    monkeypatch.setattr(daemon, "log", lambda _message: None)
+
+    asyncio.run(daemon.main())
+
+    assert d.cdp.closed == []
+    assert d.dedicated_target_id == "daemon-tab"
+    assert d.target_id == "user-selected-tab"
+
+
+def test_delayed_stale_request_follows_recovery_during_domain_enable(monkeypatch):
+    """Publish the replacement before post-attach domain setup can yield."""
+    class _RecoveryWindowCDP(_FakeCDP):
+        def __init__(self):
+            super().__init__()
+            self.slow_started = None
+            self.release_slow = None
+            self.enable_started = None
+            self.release_enables = None
+
+        async def send_raw(self, method, params=None, session_id=None):
+            self.calls.append((method, params, session_id))
+            if method == "Runtime.evaluate" and session_id == "stale-session":
+                if params["expression"] == "slow":
+                    self.slow_started.set()
+                    await self.release_slow.wait()
+                raise RuntimeError("Session with given id not found")
+            if method == "Target.getTargets":
+                return {"targetInfos": [
+                    {"targetId": "same-tab", "url": "https://example.com", "type": "page"}
+                ]}
+            if method == "Target.attachToTarget":
+                return {"sessionId": "replacement-session"}
+            if method.endswith(".enable") and session_id == "replacement-session":
+                self.enable_started.set()
+                await self.release_enables.wait()
+                return {}
+            if method == "Runtime.evaluate" and session_id == "replacement-session":
+                return {"value": params["expression"]}
+            return {}
+
+    async def run():
+        d = daemon.Daemon()
+        d.cdp = _RecoveryWindowCDP()
+        d.cdp.slow_started = asyncio.Event()
+        d.cdp.release_slow = asyncio.Event()
+        d.cdp.enable_started = asyncio.Event()
+        d.cdp.release_enables = asyncio.Event()
+        d.session = "stale-session"
+        d.target_id = "same-tab"
+
+        slow = asyncio.create_task(d.handle({
+            "method": "Runtime.evaluate", "params": {"expression": "slow"}
+        }))
+        await d.cdp.slow_started.wait()
+        fast = asyncio.create_task(d.handle({
+            "method": "Runtime.evaluate", "params": {"expression": "fast"}
+        }))
+        await d.cdp.enable_started.wait()
+        # Recovery has attached but is still blocked enabling domains. The
+        # delayed request must already be able to find the replacement.
+        d.cdp.release_slow.set()
+        slow_result = await slow
+        d.cdp.release_enables.set()
+        return d, await fast, slow_result
+
+    monkeypatch.setattr(daemon, "NAME", "default")
+    monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
+    d, fast, slow = asyncio.run(run())
+
+    assert fast == {"result": {"value": "fast"}}
+    assert slow == {"result": {"value": "slow"}}
+    assert d._session_replacements == {"stale-session": "replacement-session"}
+
+
+def test_tab_switch_waits_for_recovery_and_keeps_old_action_on_old_tab(monkeypatch):
+    """A switch during target discovery cannot redirect the recovered action."""
+    class _SwitchRaceCDP(_FakeCDP):
+        def __init__(self):
+            super().__init__()
+            self.discovery_started = None
+            self.release_discovery = None
+
+        async def send_raw(self, method, params=None, session_id=None):
+            self.calls.append((method, params, session_id))
+            if (
+                method == "Runtime.evaluate"
+                and params.get("expression") == "old-tab-action"
+                and session_id == "old-session"
+            ):
+                raise RuntimeError("Session with given id not found")
+            if method == "Target.getTargets":
+                self.discovery_started.set()
+                await self.release_discovery.wait()
+                return {"targetInfos": [
+                    {"targetId": "old-tab", "url": "https://example.com", "type": "page"}
+                ]}
+            if method == "Target.attachToTarget":
+                return {"sessionId": "recovered-old-session"}
+            if (
+                method == "Runtime.evaluate"
+                and params.get("expression") == "old-tab-action"
+                and session_id == "recovered-old-session"
+            ):
+                return {"value": "old-tab-action"}
+            return {}
+
+    async def run():
+        d = daemon.Daemon()
+        d.cdp = _SwitchRaceCDP()
+        d.cdp.discovery_started = asyncio.Event()
+        d.cdp.release_discovery = asyncio.Event()
+        d.session = "old-session"
+        d.target_id = "old-tab"
+
+        request = asyncio.create_task(d.handle({
+            "method": "Runtime.evaluate",
+            "params": {"expression": "old-tab-action"},
+        }))
+        await d.cdp.discovery_started.wait()
+        switch = asyncio.create_task(d.handle({
+            "meta": "set_session",
+            "session_id": "new-session",
+            "target_id": "new-tab",
+        }))
+        await asyncio.sleep(0)  # let set_session wait on the recovery lock
+        d.cdp.release_discovery.set()
+        result, switch_result = await asyncio.gather(request, switch)
+        await asyncio.sleep(0)  # let the cosmetic marker task finish
+        return d, result, switch_result
+
+    monkeypatch.setattr(daemon, "NAME", "default")
+    monkeypatch.setattr(daemon, "BROWSER_KIND", "cdp")
+    d, result, switch_result = asyncio.run(run())
+
+    assert result == {"result": {"value": "old-tab-action"}}
+    assert switch_result == {"session_id": "new-session"}
+    assert d.session == "new-session"
+    assert d.target_id == "new-tab"
+    assert d._session_replacements == {"old-session": "recovered-old-session"}
+    redirected = [
+        (params, sid)
+        for method, params, sid in d.cdp.calls
+        if method == "Runtime.evaluate"
+        and params.get("expression") == "old-tab-action"
+        and sid == "new-session"
+    ]
+    assert redirected == []
+
+
+def test_explicit_stale_session_is_not_redirected():
+    """Explicit session requests retain their exact-session semantics."""
+    class _AlwaysStaleCDP(_FakeCDP):
+        async def send_raw(self, method, params=None, session_id=None):
+            self.calls.append((method, params, session_id))
+            raise RuntimeError("Session with given id not found")
+
+    d = daemon.Daemon()
+    d.cdp = _AlwaysStaleCDP()
+    d.session = "current-session"
+
+    result = asyncio.run(d.handle({
+        "method": "Runtime.evaluate",
+        "params": {"expression": "1"},
+        "session_id": "explicit-stale-session",
+    }))
+
+    assert result == {"error": "Session with given id not found"}
+    assert d.cdp.calls == [
+        ("Runtime.evaluate", {"expression": "1"}, "explicit-stale-session")
+    ]
diff --git a/tests/unit/test_helpers.py b/tests/unit/test_helpers.py
index 4a45ee07..9a099030 100644
--- a/tests/unit/test_helpers.py
+++ b/tests/unit/test_helpers.py
@@ -350,3 +350,57 @@ def fake_send(req):
         "session filter, the background rWS/lF pair would have updated "
         "last_activity and prevented the idle window from elapsing."
     )
+
+
+def test_switch_tab_keeps_visible_tab_unchanged_by_default(monkeypatch):
+    calls = []
+
+    def fake_cdp(method, **kwargs):
+        calls.append((method, kwargs))
+        if method == "Target.attachToTarget":
+            return {"sessionId": "session-new"}
+        return {}
+
+    monkeypatch.setattr(helpers, "cdp", fake_cdp)
+    monkeypatch.setattr(helpers, "_send", lambda request: calls.append(("ipc", request)) or {})
+    monkeypatch.setattr(helpers, "_mark_tab", lambda: None)
+
+    assert helpers.switch_tab({"target_id": "target-new"}) == "session-new"
+    assert not any(method == "Target.activateTarget" for method, _ in calls)
+
+
+def test_switch_tab_can_explicitly_activate_visible_tab(monkeypatch):
+    calls = []
+
+    def fake_cdp(method, **kwargs):
+        calls.append((method, kwargs))
+        if method == "Target.attachToTarget":
+            return {"sessionId": "session-new"}
+        return {}
+
+    monkeypatch.setattr(helpers, "cdp", fake_cdp)
+    monkeypatch.setattr(helpers, "_send", lambda request: calls.append(("ipc", request)) or {})
+    monkeypatch.setattr(helpers, "_mark_tab", lambda: None)
+
+    assert helpers.switch_tab("target-new", activate=True) == "session-new"
+    assert ("Target.activateTarget", {"targetId": "target-new"}) in calls
+
+
+def test_new_tab_creates_and_attaches_in_background(monkeypatch):
+    calls = []
+
+    def fake_cdp(method, **kwargs):
+        calls.append((method, kwargs))
+        if method == "Target.createTarget":
+            return {"targetId": "target-new"}
+        if method == "Target.attachToTarget":
+            return {"sessionId": "session-new"}
+        return {}
+
+    monkeypatch.setattr(helpers, "cdp", fake_cdp)
+    monkeypatch.setattr(helpers, "_send", lambda request: calls.append(("ipc", request)) or {})
+    monkeypatch.setattr(helpers, "_mark_tab", lambda: None)
+
+    assert helpers.new_tab() == "target-new"
+    assert ("Target.createTarget", {"url": "about:blank", "background": True}) in calls
+    assert not any(method == "Target.activateTarget" for method, _ in calls)