From 5bb5322f49b9d6235f0e9b44c4d7cdaa3d368cda Mon Sep 17 00:00:00 2001 From: QuanCheng <915158214@qq.com> Date: Tue, 14 Jul 2026 08:36:59 +0000 Subject: [PATCH] prevent shared browser session leaks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of the shared-browser session lifecycle fix. Atomic quota slots (unique-constraint claim on bf_key_fingerprint) are NOT yet implemented; the count-based pre-check remains advisory and the Browser Fabric server stays the final quota arbiter. - persist Browser Fabric credential identity without storing plaintext keys - preserve and track remote sessions across restart and worker changes - retry failed session cleanup with CAS-based maintenance claims - retain sessions after navigation or event-pipeline failures - add structured quota and credential errors - add browser leak and PostgreSQL-gated concurrency tests Tests: test_browser_tab_leak.py 38 passed; PostgreSQL concurrency tests 2 skipped (TEST_DATABASE_URL not configured); full backend suite 325 passed, 20 failed, 2 skipped — the 20 failures are byte-identical to the clean develop baseline (no local PostgreSQL, pre-existing mock/await and init.sql parity issues), zero regressions. --- .../026_browser_tab_session_tracking.py | 130 ++++ workspace/backend/app/browser.py | 201 +++-- workspace/backend/app/browser_creds.py | 129 ++++ workspace/backend/app/browser_maintenance.py | 279 +++++++ workspace/backend/app/main.py | 16 + workspace/backend/app/models.py | 10 + workspace/backend/app/routers/browser.py | 336 ++++++-- workspace/backend/pytest.ini | 3 + .../test_browser_postgres_concurrency.py | 131 ++++ .../backend/tests/test_browser_tab_leak.py | 727 ++++++++++++++++++ 10 files changed, 1840 insertions(+), 122 deletions(-) create mode 100644 workspace/backend/alembic/versions/026_browser_tab_session_tracking.py create mode 100644 workspace/backend/app/browser_creds.py create mode 100644 workspace/backend/app/browser_maintenance.py create mode 100644 workspace/backend/pytest.ini create mode 100644 workspace/backend/tests/test_browser_postgres_concurrency.py create mode 100644 workspace/backend/tests/test_browser_tab_leak.py diff --git a/workspace/backend/alembic/versions/026_browser_tab_session_tracking.py b/workspace/backend/alembic/versions/026_browser_tab_session_tracking.py new file mode 100644 index 000000000..d6b25fded --- /dev/null +++ b/workspace/backend/alembic/versions/026_browser_tab_session_tracking.py @@ -0,0 +1,130 @@ +# -*- coding: utf-8 -*- +"""Track BF credential reference + session release state on browser tabs. + +Revision ID: 026 +Revises: 025 +Create Date: 2026-07-14 + +Fixes the shared-browser session leak. Browser Fabric sessions created +with a per-workspace API key could never be reliably closed after a server +restart because the tab→key mapping lived only in process memory, and +closed-in-DB tabs whose remote close failed were silently forgotten. + +New columns on `browser_tabs`: + + - `bf_key_source` / `bf_key_fingerprint` — a *reference* to the key the + session was created with ('workspace' settings key or 'global' env key) + plus its SHA-256 fingerprint. The key itself is never stored here. + - `session_closed` (bool) — remote BF session confirmed released. + - `close_status` — none | open | closing | closed | close_failed | + retry_exhausted. Drives the maintenance sweeper's retry state machine. + - `close_attempts`, `last_close_attempt_at`, `last_close_error` — retry + bookkeeping (errors are redacted before storage). + - `last_error` — last init/navigation error surfaced to the caller. + +Backfill policy (deliberately NOT "assume old sessions are gone"): + + - rows with no remote session_id: nothing to release → + session_closed=TRUE, close_status='none'. + - active rows with a session_id: live session → close_status='open'. + - non-active rows WITH a session_id: the remote session may still exist + and be eating the per-key ephemeral quota → session_closed=FALSE, + close_status='close_failed' so the maintenance sweeper retries the + close. Load is bounded by the sweeper's per-pass action cap and + retry window, not by faking release here. + +Credential backfill resolves each workspace's key reference at migration +time (settings key → 'workspace', else global env key → 'global') and +stores only source + fingerprint. +""" + +import hashlib +import json +import os + +from alembic import op +import sqlalchemy as sa + + +revision = "026" +down_revision = "025" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("browser_tabs", sa.Column("bf_key_source", sa.Text(), nullable=True)) + op.add_column("browser_tabs", sa.Column("bf_key_fingerprint", sa.Text(), nullable=True)) + op.add_column( + "browser_tabs", + sa.Column("session_closed", sa.Boolean(), server_default=sa.text("FALSE"), nullable=False), + ) + op.add_column( + "browser_tabs", + sa.Column("close_status", sa.Text(), server_default=sa.text("'none'"), nullable=False), + ) + op.add_column( + "browser_tabs", + sa.Column("close_attempts", sa.Integer(), server_default=sa.text("0"), nullable=False), + ) + op.add_column("browser_tabs", sa.Column("last_close_attempt_at", sa.DateTime(timezone=True), nullable=True)) + op.add_column("browser_tabs", sa.Column("last_close_error", sa.Text(), nullable=True)) + op.add_column("browser_tabs", sa.Column("last_error", sa.Text(), nullable=True)) + + bind = op.get_bind() + + # ── Release-state backfill ── + # No remote session → nothing to release. + bind.execute(sa.text( + "UPDATE browser_tabs SET session_closed = TRUE, close_status = 'none' " + "WHERE session_id IS NULL" + )) + # Live sessions on active tabs. + bind.execute(sa.text( + "UPDATE browser_tabs SET session_closed = FALSE, close_status = 'open' " + "WHERE session_id IS NOT NULL AND status = 'active'" + )) + # Closed-in-DB tabs that still reference a remote session: the remote + # side may still hold the session — queue for retried release. + bind.execute(sa.text( + "UPDATE browser_tabs SET session_closed = FALSE, close_status = 'close_failed' " + "WHERE session_id IS NOT NULL AND status != 'active'" + )) + + # ── Credential-reference backfill ── + global_key = os.environ.get("BROWSERFABRIC_API_KEY", "") + global_fp = hashlib.sha256(global_key.encode("utf-8")).hexdigest() if global_key else None + + rows = bind.execute(sa.text("SELECT id, settings FROM workspaces")).fetchall() + for ws_id, settings in rows: + if isinstance(settings, str): + try: + settings = json.loads(settings) + except Exception: + settings = {} + settings = settings or {} + ws_key = settings.get("browserfabric_api_key") + if ws_key: + source, fp = "workspace", hashlib.sha256(ws_key.encode("utf-8")).hexdigest() + elif global_key: + source, fp = "global", global_fp + else: + continue # local mode — leave NULL + bind.execute( + sa.text( + "UPDATE browser_tabs SET bf_key_source = :source, bf_key_fingerprint = :fp " + "WHERE workspace_id = :ws AND session_id IS NOT NULL" + ), + {"source": source, "fp": fp, "ws": str(ws_id)}, + ) + + +def downgrade() -> None: + op.drop_column("browser_tabs", "last_error") + op.drop_column("browser_tabs", "last_close_error") + op.drop_column("browser_tabs", "last_close_attempt_at") + op.drop_column("browser_tabs", "close_attempts") + op.drop_column("browser_tabs", "close_status") + op.drop_column("browser_tabs", "session_closed") + op.drop_column("browser_tabs", "bf_key_fingerprint") + op.drop_column("browser_tabs", "bf_key_source") diff --git a/workspace/backend/app/browser.py b/workspace/backend/app/browser.py index 63ba53caf..ae4ff2c60 100644 --- a/workspace/backend/app/browser.py +++ b/workspace/backend/app/browser.py @@ -15,6 +15,8 @@ import httpx +from app.browser_creds import redact + logger = logging.getLogger(__name__) MAX_BROWSER_TABS = int(os.environ.get("MAX_BROWSER_TABS", "20")) @@ -52,6 +54,12 @@ def is_cloud(self) -> bool: def is_cloud_for(self, api_key: str = None) -> bool: return bool(api_key or BROWSERFABRIC_API_KEY) + def _is_cloud_tab(self, tab_id: str) -> bool: + """A tab is a cloud tab iff it has a BF session. Dispatching on this + (rather than the global env key) keeps per-workspace-key tabs working + even when no global BROWSERFABRIC_API_KEY is configured.""" + return tab_id in self._sessions + # ------------------------------------------------------------------ # Browser Fabric REST helpers # ------------------------------------------------------------------ @@ -62,6 +70,15 @@ def _key_for_tab(self, tab_id: str = None) -> str: return self._tab_keys[tab_id] return BROWSERFABRIC_API_KEY + def bind_tab_key(self, tab_id: str, api_key: Optional[str]) -> None: + """Refresh the in-process key cache from the router's credential + resolver, so per-tab ops use the verified key even when the tab was + opened by another process/before a restart.""" + if api_key: + self._tab_keys[tab_id] = api_key + else: + self._tab_keys.pop(tab_id, None) + async def _bf_call(self, tool_name: str, arguments: dict = None, session_id: str = None, api_key: str = None, tab_id: str = None) -> dict: """Call a Browser Fabric tool via REST API.""" key = api_key or (self._key_for_tab(tab_id) if tab_id else BROWSERFABRIC_API_KEY) @@ -130,12 +147,13 @@ async def _prune_dead_sessions(self) -> int: dead: list[str] = [] for tab_id, session_id in list(self._sessions.items()): try: - await self._bf_call("get_page_info", {}, session_id) + await self._bf_call("get_page_info", {}, session_id, tab_id=tab_id) except Exception: dead.append(tab_id) for tab_id in dead: - self._sessions.pop(tab_id, None) - self._live_urls.pop(tab_id, None) + # close_tab (not a bare pop) so a session that merely errored on + # the liveness probe still gets a best-effort release on BF. + await self.close_tab(tab_id) logger.info("Pruned dead BF session for tab %s", tab_id) return len(dead) @@ -144,36 +162,52 @@ async def open_tab(self, tab_id: str, url: str = "about:blank", bb_context_id: s if api_key: self._tab_keys[tab_id] = api_key async with self._global_lock: - active_count = len(self._sessions) if self.is_cloud else len(self._pages) + active_count = self.active_tab_count() if active_count >= MAX_BROWSER_TABS: - if self.is_cloud: + if self.is_cloud_for(api_key): await self._prune_dead_sessions() active_count = len(self._sessions) if active_count >= MAX_BROWSER_TABS: raise RuntimeError(f"Maximum browser tabs ({MAX_BROWSER_TABS}) reached") - if self.is_cloud: + if self.is_cloud_for(api_key): args: dict = {"headless": True} if bb_context_id: args["context_id"] = bb_context_id args["persist"] = True - result = await self._bf_call("create_session", args, tab_id=tab_id) + try: + result = await self._bf_call("create_session", args, tab_id=tab_id) + except Exception: + self._tab_keys.pop(tab_id, None) + raise session_data = result["result"] session_id = session_data["session_id"] self._sessions[tab_id] = session_id if session_data.get("share_url"): self._live_urls[tab_id] = session_data["share_url"] + # From here on the BF session exists and counts against the + # per-key quota — nothing below may raise, or the caller never + # records the session and it leaks. Failures are surfaced as + # warnings so the caller knows init didn't fully succeed. + warnings: list = [] + key_used = self._key_for_tab(tab_id) if url and url != "about:blank": try: await self._bf_call("navigate", {"url": url, "wait_until": "domcontentloaded"}, session_id, tab_id=tab_id) - except Exception: - pass + except Exception as e: + warnings.append(f"navigation_failed: {redact(str(e), key_used)}") - info = await self._bf_call("get_page_info", {}, session_id) - page_info = info.get("result", {}) - return {"url": page_info.get("url", url), "title": page_info.get("title", "")} + try: + info = await self._bf_call("get_page_info", {}, session_id, tab_id=tab_id) + page_info = info.get("result", {}) + except Exception as e: + logger.warning("get_page_info failed for new tab %s (session kept): %s", + tab_id, redact(str(e), key_used)) + warnings.append(f"page_info_failed: {redact(str(e), key_used)}") + page_info = {} + return {"url": page_info.get("url", url), "title": page_info.get("title", ""), "warnings": warnings} else: # Local mode async with self._global_lock: @@ -181,24 +215,25 @@ async def open_tab(self, tab_id: str, url: str = "about:blank", bb_context_id: s page = await self._browser.new_page() self._pages[tab_id] = page + warnings = [] if url and url != "about:blank": try: await page.goto(url, wait_until="domcontentloaded", timeout=30000) - except Exception: - pass + except Exception as e: + warnings.append(f"navigation_failed: {e}") title = await page.title() - return {"url": page.url, "title": title} + return {"url": page.url, "title": title, "warnings": warnings} async def navigate(self, tab_id: str, url: str) -> dict: """Navigate a tab to a URL. Returns {url, title}.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) try: - await self._bf_call("navigate", {"url": url, "wait_until": "domcontentloaded"}, session_id) + await self._bf_call("navigate", {"url": url, "wait_until": "domcontentloaded"}, session_id, tab_id=tab_id) except Exception: pass - info = await self._bf_call("get_page_info", {}, session_id) + info = await self._bf_call("get_page_info", {}, session_id, tab_id=tab_id) page_info = info.get("result", {}) return {"url": page_info.get("url", url), "title": page_info.get("title", "")} else: @@ -213,10 +248,10 @@ async def navigate(self, tab_id: str, url: str) -> dict: async def click(self, tab_id: str, selector: str) -> dict: """Click an element by CSS selector.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - await self._bf_call("click_element", {"selector": selector}, session_id) - info = await self._bf_call("get_page_info", {}, session_id) + await self._bf_call("click_element", {"selector": selector}, session_id, tab_id=tab_id) + info = await self._bf_call("get_page_info", {}, session_id, tab_id=tab_id) page_info = info.get("result", {}) return {"clicked": selector, "url": page_info.get("url", ""), "title": page_info.get("title", "")} else: @@ -228,9 +263,9 @@ async def click(self, tab_id: str, selector: str) -> dict: async def type_text(self, tab_id: str, selector: str, text: str, append: bool = False) -> dict: """Type text into an element.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - await self._bf_call("type_text", {"selector": selector, "text": text}, session_id) + await self._bf_call("type_text", {"selector": selector, "text": text}, session_id, tab_id=tab_id) return {"filled": selector, "text": text} else: page = self._get_page(tab_id) @@ -253,9 +288,9 @@ async def type_text(self, tab_id: str, selector: str, text: str, append: bool = async def press_key(self, tab_id: str, key: str) -> dict: """Press a keyboard key.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - await self._bf_call("press_key", {"key": key}, session_id) + await self._bf_call("press_key", {"key": key}, session_id, tab_id=tab_id) return {"pressed": key} else: page = self._get_page(tab_id) @@ -265,9 +300,9 @@ async def press_key(self, tab_id: str, key: str) -> dict: async def evaluate(self, tab_id: str, expression: str) -> dict: """Execute JavaScript in the page context.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - result = await self._bf_call("evaluate_js", {"expression": expression}, session_id) + result = await self._bf_call("evaluate_js", {"expression": expression}, session_id, tab_id=tab_id) return {"result": result.get("result", {}).get("result")} else: page = self._get_page(tab_id) @@ -277,9 +312,9 @@ async def evaluate(self, tab_id: str, expression: str) -> dict: async def screenshot(self, tab_id: str) -> bytes: """Take a PNG screenshot of the tab.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - result = await self._bf_call("take_screenshot", {"full_page": False}, session_id) + result = await self._bf_call("take_screenshot", {"full_page": False}, session_id, tab_id=tab_id) b64_data = result.get("result", {}).get("screenshot", "") if b64_data.startswith("data:"): b64_data = b64_data.split(",", 1)[1] @@ -291,9 +326,9 @@ async def screenshot(self, tab_id: str) -> bytes: async def snapshot(self, tab_id: str) -> str: """Get page content as a readable text snapshot.""" - if self.is_cloud: + if self._is_cloud_tab(tab_id): session_id = self._get_session(tab_id) - result = await self._bf_call("snapshot", {}, session_id) + result = await self._bf_call("snapshot", {}, session_id, tab_id=tab_id) return result.get("result", {}).get("snapshot", "(empty page)") else: page = self._get_page(tab_id) @@ -311,25 +346,55 @@ async def snapshot(self, tab_id: str) -> str: except Exception: return "(empty page)" - async def close_tab(self, tab_id: str, session_id_hint: str = None) -> None: - """Close a browser tab.""" - if self.is_cloud: - session_id = self._sessions.pop(tab_id, None) or session_id_hint - self._live_urls.pop(tab_id, None) - tab_key = self._tab_keys.pop(tab_id, None) - if session_id: - try: - await self._bf_call("close_session", {}, session_id, api_key=tab_key) - except Exception as e: - logger.warning("Failed to close BF session %s: %s", session_id, e) - else: - page = self._pages.pop(tab_id, None) - self._locks.pop(tab_id, None) - if page: - try: - await page.close() - except Exception: - pass + async def close_tab(self, tab_id: str, session_id_hint: str = None, api_key: str = None) -> tuple: + """Close a browser tab. + + Returns (released, error): + - (True, None) — BF session confirmed released: 2xx close success, + a typed HTTP 404 (session no longer exists), or + nothing to release. + - (False, error) — close failed / outcome unknown (timeout, network + error, non-404 HTTP error, success=false body). + The caller must keep the session marked open so + the maintenance sweeper retries. `error` is + redacted — safe to store and log. + + `api_key` is the key resolved from the tab's credential reference; + required for cross-restart closes where the in-memory `_tab_keys` + mapping is gone. + """ + session_id = self._sessions.pop(tab_id, None) or session_id_hint + tab_key = api_key or self._tab_keys.pop(tab_id, None) + if api_key: + self._tab_keys.pop(tab_id, None) + self._live_urls.pop(tab_id, None) + if session_id and self.is_cloud_for(tab_key): + try: + await self._bf_call("close_session", {}, session_id, api_key=tab_key) + except httpx.HTTPStatusError as e: + # Only a typed 404 reliably means "session no longer exists". + # Any other status is an unknown outcome — do NOT report + # released, the remote session may still hold quota. + if e.response is not None and e.response.status_code == 404: + logger.info("BF session %s already gone (404) — treating as released", session_id) + return True, None + err = redact(f"HTTP {e.response.status_code if e.response is not None else '?'} closing session", tab_key) + logger.warning("Failed to close BF session %s: %s", session_id, err) + return False, err + except Exception as e: + err = redact(str(e), tab_key) + logger.warning("Failed to close BF session %s: %s", session_id, err) + return False, err + return True, None + + page = self._pages.pop(tab_id, None) + self._locks.pop(tab_id, None) + if page: + try: + await page.close() + except Exception: + pass + return True, None async def shutdown(self) -> None: """Close all tabs and the browser.""" @@ -352,34 +417,38 @@ async def shutdown(self) -> None: # Reconnection (serverless / cold-start recovery) # ------------------------------------------------------------------ - async def reconnect(self, tab_id: str, session_id: str) -> None: + async def reconnect(self, tab_id: str, session_id: str, api_key: str = None) -> None: """Reconnect to an existing Browser Fabric session. - In REST-only mode, we just store the session_id mapping. - The next operation will use it to call the BF API. + In REST-only mode, we just store the session_id mapping (and the + API key the session was created with, so subsequent calls don't + fall back to the global key). The next operation will use it to + call the BF API. """ - if self.is_cloud: - if tab_id in self._sessions: - return - self._sessions[tab_id] = session_id - else: + if not self.is_cloud_for(api_key): raise KeyError(f"Cannot reconnect to local tab: {tab_id}") + if api_key: + self._tab_keys[tab_id] = api_key + if tab_id in self._sessions: + return + self._sessions[tab_id] = session_id # ------------------------------------------------------------------ # Persistent contexts # ------------------------------------------------------------------ - async def create_bb_context(self, session_id: str = None) -> str: + async def create_bb_context(self, session_id: str = None, tab_id: str = None) -> str: """Save the current session's state and return a Browser Fabric context ID. If session_id is provided, calls save_context on the active session so cookies/localStorage are captured before the session is closed. """ - if self.is_cloud and session_id: + if self.is_cloud_for(self._key_for_tab(tab_id) if tab_id else None) and session_id: result = await self._bf_call( "save_context", {"context_name": f"persist-{session_id[:8]}"}, session_id, + tab_id=tab_id, ) return result.get("result", {}).get("context_id", str(__import__("uuid").uuid4())) import uuid @@ -429,12 +498,10 @@ def get_session_id(self, tab_id: str) -> Optional[str]: async def get_current_url(self, tab_id: str) -> Optional[dict]: """Return the current {url, title} from the live page.""" - if self.is_cloud: - session_id = self._sessions.get(tab_id) - if not session_id: - return None + if self._is_cloud_tab(tab_id): + session_id = self._sessions[tab_id] try: - info = await self._bf_call("get_page_info", {}, session_id) + info = await self._bf_call("get_page_info", {}, session_id, tab_id=tab_id) page_info = info.get("result", {}) return {"url": page_info.get("url", ""), "title": page_info.get("title", "")} except Exception: @@ -451,6 +518,4 @@ async def get_current_url(self, tab_id: str) -> Optional[dict]: return None def active_tab_count(self) -> int: - if self.is_cloud: - return len(self._sessions) - return len(self._pages) + return len(self._sessions) + len(self._pages) diff --git a/workspace/backend/app/browser_creds.py b/workspace/backend/app/browser_creds.py new file mode 100644 index 000000000..533e637b8 --- /dev/null +++ b/workspace/backend/app/browser_creds.py @@ -0,0 +1,129 @@ +# -*- coding: utf-8 -*- +""" +Browser Fabric credential references — resolve, verify, redact. + +`browser_tabs` never stores a BF API key in plaintext. Each tab records: + + - `bf_key_source` — 'workspace' (workspace.settings.browserfabric_api_key) + or 'global' (BROWSERFABRIC_API_KEY env var); + NULL for local-mode tabs / pre-migration rows. + - `bf_key_fingerprint` — full SHA-256 hex of the key the session was + created with, for rotation detection and quota + attribution. A hash is stored, never the key. + +`resolve_tab_key` is the single path every session operation (close, +reconnect, persist, navigate, click, type, screenshot, snapshot, page +info) uses to turn that reference back into a real key. If the resolved +key no longer matches the stored fingerprint, the credential was rotated +and the operation MUST NOT proceed with the new key against the old +session — callers get `BrowserCredentialError` instead of a silent +wrong-key call (accepted residual risk: rotated-away sessions can no +longer be closed by us). + +Legacy rows (fingerprint NULL) skip the mismatch check but log that the +verification was bypassed. +""" + +import hashlib +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + +SOURCE_WORKSPACE = "workspace" +SOURCE_GLOBAL = "global" + + +class BrowserCredentialError(Exception): + """Credential reference cannot be resolved or no longer matches. + + `reason` is machine-readable: 'credential_missing' | 'credential_mismatch'. + The message never contains key material. + """ + + def __init__(self, reason: str, detail: str): + self.reason = reason + super().__init__(f"{reason}: {detail}") + + +def key_fingerprint(key: Optional[str]) -> Optional[str]: + """Full SHA-256 hex fingerprint of an API key (None for no key).""" + if not key: + return None + return hashlib.sha256(key.encode("utf-8")).hexdigest() + + +def redact(text: Optional[str], *keys: Optional[str]) -> Optional[str]: + """Strip any real key material from text before it is stored or logged.""" + if not text: + return text + for key in keys: + if key and key in text: + fp = key_fingerprint(key) + text = text.replace(key, f"") + return text + + +def _current_key_for_source(source: Optional[str], workspace) -> Optional[str]: + if source == SOURCE_WORKSPACE: + if workspace is None: + return None + return (workspace.settings or {}).get("browserfabric_api_key") + if source == SOURCE_GLOBAL: + # Read at call time (not import time) so env changes and test + # monkeypatching of app.browser.BROWSERFABRIC_API_KEY are honoured. + from app import browser as browser_module + return browser_module.BROWSERFABRIC_API_KEY or None + return None + + +def resolve_tab_key(tab, workspace) -> Optional[str]: + """Resolve the real BF API key for an existing tab's session. + + Returns the key (or None for local-mode tabs with no source). + Raises BrowserCredentialError when the reference is broken: + - credential_missing: the referenced source no longer holds a key; + - credential_mismatch: the source's current key is not the key the + session was created with (rotation) — do NOT use it on this session. + """ + source = getattr(tab, "bf_key_source", None) + stored_fp = getattr(tab, "bf_key_fingerprint", None) + + if source is None: + # Legacy / local-mode row: best-effort fallback (workspace key then + # global), preserving pre-migration behaviour. No fingerprint to + # verify against; log so ops can trace unverified key use. + key = _current_key_for_source(SOURCE_WORKSPACE, workspace) \ + or _current_key_for_source(SOURCE_GLOBAL, workspace) + if key and stored_fp is None: + logger.info( + "browser.credential.unverified tab=%s: legacy row without key source; " + "using fallback resolution", getattr(tab, "id", "?"), + ) + return key + + key = _current_key_for_source(source, workspace) + if not key: + raise BrowserCredentialError( + "credential_missing", + f"tab {getattr(tab, 'id', '?')} references {source} BF key but none is configured", + ) + + if stored_fp: + current_fp = key_fingerprint(key) + if current_fp != stored_fp: + logger.warning( + "browser.credential.rotated tab=%s source=%s stored_fp=%s current_fp=%s", + getattr(tab, "id", "?"), source, stored_fp[:12], current_fp[:12], + ) + raise BrowserCredentialError( + "credential_mismatch", + f"tab {getattr(tab, 'id', '?')}: {source} BF key was rotated; " + "refusing to operate on the old session with the new key", + ) + else: + logger.info( + "browser.credential.unverified tab=%s: no stored fingerprint; skipping check", + getattr(tab, "id", "?"), + ) + return key diff --git a/workspace/backend/app/browser_maintenance.py b/workspace/backend/app/browser_maintenance.py new file mode 100644 index 000000000..76a61f1eb --- /dev/null +++ b/workspace/backend/app/browser_maintenance.py @@ -0,0 +1,279 @@ +# -*- coding: utf-8 -*- +""" +Shared-browser maintenance sweep — the backstop against BF session leaks. + +Browser Fabric caps ephemeral sessions per API key (free tier: 3), so any +session the backend loses track of permanently eats a slot the user cannot +see or close from the UI. Each pass runs four steps: + +0. Stale-claim recovery — rows stuck in close_status='closing' (a worker + crashed mid-close) flip back to 'close_failed' so they can be retried. + +1. Idle reaper — active ephemeral (non-persistent) tabs with no activity + for BROWSER_TAB_IDLE_MINUTES are closed. Agent-opened tabs whose agent + crashed mid-task are the main source of these. + +2. Orphan release retry — tabs whose BF close failed (close_status= + 'close_failed', session_closed=FALSE) are retried with the key resolved + from the credential reference persisted on the row. + +3. Exhaust transition — rows still failing past + BROWSER_CLOSE_RETRY_WINDOW_HOURS become close_status='retry_exhausted' + with a searchable error log. session_closed stays FALSE: we do NOT + pretend an unconfirmed session was released (there is no evidence BF + expires them). These rows are left for manual review. + +Concurrency: the backend runs multiple replicas, so every row is claimed +via a conditional UPDATE (compare-and-swap on close_status AND session_id) +committed BEFORE the slow BF HTTP call, and the outcome is written with the +same session_id guard — if persist/reconnect swapped in a new session in +the meantime, the outcome write is dropped instead of clobbering the new +session's state. No row lock is held across a BF call. Work per pass is +capped by BROWSER_SWEEP_MAX_ACTIONS per step. +""" + +import logging +import os +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select, update + +from app.browser import BrowserManager +from app.browser_creds import BrowserCredentialError, resolve_tab_key +from app.models import BrowserTab, BrowserUsage, Workspace + +logger = logging.getLogger(__name__) + +BROWSER_TAB_IDLE_MINUTES = int(os.environ.get("BROWSER_TAB_IDLE_MINUTES", "30")) +BROWSER_CLOSE_RETRY_WINDOW_HOURS = int(os.environ.get("BROWSER_CLOSE_RETRY_WINDOW_HOURS", "6")) +BROWSER_CLOSING_STALE_MINUTES = int(os.environ.get("BROWSER_CLOSING_STALE_MINUTES", "10")) +MAX_SWEEP_ACTIONS = int(os.environ.get("BROWSER_SWEEP_MAX_ACTIONS", "10")) + + +def _finalize_usage(db, tab_id: str, now: datetime) -> None: + usage = db.execute( + select(BrowserUsage) + .where(BrowserUsage.tab_id == tab_id) + .where(BrowserUsage.ended_at.is_(None)) + ).scalar_one_or_none() + if usage: + usage.ended_at = now + if usage.started_at: + started = usage.started_at + if started.tzinfo is None: + started = started.replace(tzinfo=timezone.utc) + usage.duration_seconds = int((now - started).total_seconds()) + + +async def _emit_tab_closed(db, workspace: Workspace, tab: BrowserTab) -> None: + """Best-effort tab.closed event so connected UIs drop the tab.""" + try: + from app.routers.network import _emit_event + from openagents.core.onm_events import Event + + payload = {"tab_id": tab.id, "reason": "idle"} + if tab.context_id: + payload["context_id"] = tab.context_id + payload["persistent"] = True + event = Event( + type="workspace.browser.tab.closed", + source="system", + target="core", + payload=payload, + ) + await _emit_event(event, workspace, db, token=workspace.password_hash) + except Exception as e: + logger.warning("tab.closed event failed for reaped tab %s: %s", tab.id, e) + + +def _workspace_for(db, tab: BrowserTab): + return db.execute( + select(Workspace).where(Workspace.id == tab.workspace_id) + ).scalar_one_or_none() + + +async def _release_claimed(db, manager, tab: BrowserTab, session_snapshot: str, now: datetime) -> bool: + """Close the claimed tab's BF session and CAS the outcome back. + + The tab row is already in close_status='closing' (claimed by us). The + outcome UPDATE re-checks session_id so a concurrent persist/reconnect + swap is never overwritten. Returns True when the release was confirmed. + """ + workspace = _workspace_for(db, tab) + try: + key = resolve_tab_key(tab, workspace) + released, close_err = await manager.close_tab( + tab.id, session_id_hint=session_snapshot, api_key=key + ) + except BrowserCredentialError as e: + released, close_err = False, str(e) + except Exception as e: + logger.warning("Sweep close failed for tab %s: %s", tab.id, e) + released, close_err = False, "unexpected close error" + + values = { + "close_attempts": (tab.close_attempts or 0) + 1, + "last_close_attempt_at": now, + } + if released: + values.update({"close_status": "closed", "session_closed": True, "last_close_error": None}) + else: + values.update({"close_status": "close_failed", "session_closed": False, + "last_close_error": close_err or "unknown close failure"}) + + result = db.execute( + update(BrowserTab) + .where(BrowserTab.id == tab.id) + .where(BrowserTab.close_status == "closing") + .where(BrowserTab.session_id == session_snapshot) + .values(**values) + ) + if result.rowcount != 1: + logger.warning( + "Sweep outcome dropped for tab %s: session changed while closing " + "(persist/reconnect swap) — not touching the new session's state", tab.id, + ) + db.commit() + return released + + +async def sweep_browser_tabs() -> dict: + """One maintenance pass. Never raises; returns counters for logging/tests.""" + stats = {"reaped": 0, "released": 0, "release_failed": 0, "exhausted": 0, "stale_recovered": 0} + manager = BrowserManager.get() + # Call-time import so tests can monkeypatch app.database.SessionLocal + # (same convention as app.main._run_maintenance). + from app.database import SessionLocal + db = SessionLocal() + try: + now = datetime.now(timezone.utc) + retry_floor = now - timedelta(hours=BROWSER_CLOSE_RETRY_WINDOW_HOURS) + stale_cutoff = now - timedelta(minutes=BROWSER_CLOSING_STALE_MINUTES) + + # ── 0. Recover stale 'closing' claims (crashed workers) ── + recovered = db.execute( + update(BrowserTab) + .where(BrowserTab.close_status == "closing") + .where(BrowserTab.last_close_attempt_at < stale_cutoff) + .values(close_status="close_failed") + ) + if recovered.rowcount: + stats["stale_recovered"] = recovered.rowcount + logger.info("Recovered %d stale closing claim(s)", recovered.rowcount) + db.commit() + + # ── 1. Reap idle ephemeral tabs ── + idle_cutoff = now - timedelta(minutes=BROWSER_TAB_IDLE_MINUTES) + idle_tabs = db.execute( + select(BrowserTab) + .where(BrowserTab.status == "active") + .where(BrowserTab.context_id.is_(None)) + .where(BrowserTab.last_active_at < idle_cutoff) + .order_by(BrowserTab.last_active_at.asc()) + .limit(MAX_SWEEP_ACTIONS) + ).scalars().all() + + for tab in idle_tabs: + session_snapshot = tab.session_id + # Claim: active → closed(closing). The status guard means only one + # replica (and no user DELETE) wins this tab. + claim = db.execute( + update(BrowserTab) + .where(BrowserTab.id == tab.id) + .where(BrowserTab.status == "active") + .values(status="closed", close_status="closing", last_close_attempt_at=now) + ) + db.commit() + if claim.rowcount != 1: + continue # another replica or a user close got there first + + db.refresh(tab) + if session_snapshot: + released = await _release_claimed(db, manager, tab, session_snapshot, now) + else: + db.execute( + update(BrowserTab) + .where(BrowserTab.id == tab.id) + .values(close_status="none", session_closed=True) + ) + released = True + _finalize_usage(db, tab.id, now) + db.commit() + stats["reaped"] += 1 + logger.info("Reaped idle browser tab %s (idle > %dm, released=%s)", + tab.id, BROWSER_TAB_IDLE_MINUTES, released) + workspace = _workspace_for(db, tab) + if workspace: + await _emit_tab_closed(db, workspace, tab) + + # ── 2. Retry releasing orphaned BF sessions (within retry window) ── + orphans = db.execute( + select(BrowserTab) + .where(BrowserTab.status != "active") + .where(BrowserTab.close_status == "close_failed") + .where(BrowserTab.session_closed.is_(False)) + .where(BrowserTab.session_id.is_not(None)) + .where(BrowserTab.last_active_at >= retry_floor) + .order_by(BrowserTab.last_active_at.asc()) + .limit(MAX_SWEEP_ACTIONS) + ).scalars().all() + + for tab in orphans: + session_snapshot = tab.session_id + # Claim via CAS on close_status + session_id: only one replica + # retries, and never against a session that was since replaced. + claim = db.execute( + update(BrowserTab) + .where(BrowserTab.id == tab.id) + .where(BrowserTab.close_status == "close_failed") + .where(BrowserTab.session_id == session_snapshot) + .values(close_status="closing", last_close_attempt_at=now) + ) + db.commit() + if claim.rowcount != 1: + continue + + db.refresh(tab) + if await _release_claimed(db, manager, tab, session_snapshot, now): + stats["released"] += 1 + logger.info("Released orphaned BF session %s (tab %s)", session_snapshot, tab.id) + else: + stats["release_failed"] += 1 + + # ── 3. Exhaust transition: still failing past the retry window ── + exhausted_rows = db.execute( + select(BrowserTab.id, BrowserTab.session_id, BrowserTab.close_attempts, + BrowserTab.bf_key_fingerprint, BrowserTab.last_close_error) + .where(BrowserTab.close_status == "close_failed") + .where(BrowserTab.session_closed.is_(False)) + .where(BrowserTab.session_id.is_not(None)) + .where(BrowserTab.last_active_at < retry_floor) + .limit(MAX_SWEEP_ACTIONS) + ).all() + for row in exhausted_rows: + db.execute( + update(BrowserTab) + .where(BrowserTab.id == row.id) + .where(BrowserTab.close_status == "close_failed") + .values(close_status="retry_exhausted") + ) + # session_closed stays FALSE: the release was never confirmed and + # we have no evidence BF expires sessions on its own. + logger.error( + "browser.session.retry_exhausted tab=%s session=%s attempts=%s key_fp=%s last_error=%s " + "— remote session unconfirmed after %dh of retries; needs manual review", + row.id, row.session_id, row.close_attempts, + (row.bf_key_fingerprint or "")[:12], row.last_close_error, + BROWSER_CLOSE_RETRY_WINDOW_HOURS, + ) + stats["exhausted"] += 1 + db.commit() + + if any(stats.values()): + logger.info("Browser sweep: %s", stats) + return stats + except Exception: + logger.exception("Browser maintenance sweep failed") + return stats + finally: + db.close() diff --git a/workspace/backend/app/main.py b/workspace/backend/app/main.py index ae1989790..042436322 100644 --- a/workspace/backend/app/main.py +++ b/workspace/backend/app/main.py @@ -325,13 +325,22 @@ async def _timer_loop(): the 24-slot DB pool. Now the firing path uses a short-lived session and the heavy scans run off-loop via ``asyncio.to_thread``, far less often. """ + from app.browser_maintenance import sweep_browser_tabs + cycle = 0 + browser_sweep_task = None while True: try: await _fire_due() cycle += 1 if cycle % MAINTENANCE_EVERY_N_CYCLES == 0: await asyncio.to_thread(_run_maintenance) + # Browser sweep is async (BF HTTP calls) and self-contained; + # run it as its own task so slow BF responses never delay + # timer firing. Overlap guard: skip if the last one is + # still running. + if browser_sweep_task is None or browser_sweep_task.done(): + browser_sweep_task = asyncio.create_task(sweep_browser_tabs()) except Exception: logger.exception("Timer loop error") await asyncio.sleep(TIMER_LOOP_INTERVAL_SECONDS) @@ -367,6 +376,13 @@ async def lifespan(app: FastAPI): logger.info("LIFESPAN: creating timer task") timer_task = asyncio.create_task(_timer_loop()) + + # One-off browser sweep at boot: sessions leaked before a restart (or + # closes that failed mid-deploy) get released now instead of waiting + # for the first periodic maintenance cycle. + from app.browser_maintenance import sweep_browser_tabs + startup_browser_sweep = asyncio.create_task(sweep_browser_tabs()) # noqa: F841 — keep ref so it isn't GC'd + logger.info("LIFESPAN: yielding (startup complete)") yield timer_task.cancel() diff --git a/workspace/backend/app/models.py b/workspace/backend/app/models.py index 343e1883a..cc94aa9b1 100644 --- a/workspace/backend/app/models.py +++ b/workspace/backend/app/models.py @@ -306,6 +306,16 @@ class BrowserTab(Base): context_id = Column(Text, ForeignKey("browser_contexts.id", ondelete="SET NULL"), nullable=True) # persistent context session_id = Column(Text, nullable=True) # Browserbase session ID live_url = Column(Text, nullable=True) # Browserbase live view URL + # --- BF credential reference (never the key itself; see app/browser_creds.py) --- + bf_key_source = Column(Text, nullable=True) # 'workspace' | 'global' | NULL (local/legacy) + bf_key_fingerprint = Column(Text, nullable=True) # SHA-256 hex of the creating key + # --- Remote session release tracking --- + session_closed = Column(Boolean, nullable=False, default=False, server_default=text("FALSE")) # BF session confirmed released + close_status = Column(Text, nullable=False, default="none", server_default=text("'none'")) # none|open|closing|closed|close_failed|retry_exhausted + close_attempts = Column(Integer, nullable=False, default=0, server_default=text("0")) + last_close_attempt_at = Column(DateTime(timezone=True), nullable=True) + last_close_error = Column(Text, nullable=True) # redacted — never contains key material + last_error = Column(Text, nullable=True) # last init/navigation error (redacted) created_at = Column(DateTime(timezone=True), default=_now, server_default=text("NOW()")) last_active_at = Column(DateTime(timezone=True), default=_now, server_default=text("NOW()")) diff --git a/workspace/backend/app/routers/browser.py b/workspace/backend/app/routers/browser.py index 75fd6f71d..1af3c59ec 100644 --- a/workspace/backend/app/routers/browser.py +++ b/workspace/backend/app/routers/browser.py @@ -17,6 +17,7 @@ """ import logging +import os import uuid from datetime import datetime, timedelta, timezone from typing import Optional @@ -28,6 +29,13 @@ from sqlalchemy.orm import Session from app.browser import BROWSERFABRIC_API_KEY, BrowserManager +from app.browser_creds import ( + SOURCE_GLOBAL, + SOURCE_WORKSPACE, + BrowserCredentialError, + key_fingerprint, + resolve_tab_key, +) from app.database import get_db from app.models import BrowserContext, BrowserTab, BrowserUsage, Workspace from app.response import ResponseCode, json_response, success_response @@ -42,15 +50,34 @@ router = APIRouter(prefix="/v1/browser", tags=["Browser"]) +# Browser Fabric caps ephemeral (non-persistent) sessions per API key. +# Enforce it at the DB level BEFORE calling BF so the user gets an +# actionable error listing which tabs to close, instead of the raw BF 400. +# +# NOTE: this count-based pre-check is advisory only (display + early UX); +# two workers can race past it and the BF server remains the final quota +# arbiter (its 3/3 error is mapped to a structured 400 below). +# Phase-2 follow-ups, deliberately NOT in this change: +# TODO(browser-quota): atomic DB quota slots keyed on bf_key_fingerprint +# (unique-constraint claim), replacing this advisory count. +# TODO(browser-quota): PostgreSQL test — two replicas racing the last +# ephemeral slot, exactly one create_session allowed. +# TODO(browser-bf-api): investigate BF create_session request_id / +# idempotent creation (would make create timeouts trackable). +# TODO(browser-bf-api): investigate BF list_sessions / admin cleanup API +# (would allow reclaiming orphans that have no DB record at all). +BF_EPHEMERAL_TAB_LIMIT = int(os.environ.get("BF_EPHEMERAL_TAB_LIMIT", "3")) + # --------------------------------------------------------------------------- # Per-workspace BF API key resolution # --------------------------------------------------------------------------- -async def _resolve_bf_key(workspace: Workspace, db: Session) -> Optional[str]: - """Resolve the BF API key for a workspace. +async def _resolve_bf_key(workspace: Workspace, db: Session) -> tuple: + """Resolve the BF API key for a workspace, for creating NEW sessions. - Priority: + Returns (key, source) where source is 'workspace' | 'global' | None. + Priority (unchanged from the original deployment behaviour): 1. Custom key stored in workspace settings (user-provided) 2. Auto-provisioned key stored in workspace settings 3. Global BROWSERFABRIC_API_KEY env var (fallback) @@ -59,10 +86,10 @@ async def _resolve_bf_key(workspace: Workspace, db: Session) -> Optional[str]: settings = workspace.settings or {} stored_key = settings.get("browserfabric_api_key") if stored_key: - return stored_key + return stored_key, SOURCE_WORKSPACE if BROWSERFABRIC_API_KEY: - return BROWSERFABRIC_API_KEY + return BROWSERFABRIC_API_KEY, SOURCE_GLOBAL # Auto-provision from BF server new_key = await BrowserManager.provision_workspace_key(str(workspace.id)) @@ -72,9 +99,70 @@ async def _resolve_bf_key(workspace: Workspace, db: Session) -> Optional[str]: workspace.settings = current db.commit() logger.info("Auto-provisioned BF API key for workspace %s", workspace.id) - return new_key - - return None + return new_key, SOURCE_WORKSPACE + + return None, None + + +def _stamp_credential(tab: BrowserTab, key: Optional[str], source: Optional[str]) -> None: + """Record the credential reference for a freshly created session. + Stores source + fingerprint only — never the key itself.""" + tab.bf_key_source = source if key else None + tab.bf_key_fingerprint = key_fingerprint(key) + tab.session_closed = False + tab.close_status = "open" + tab.close_attempts = 0 + tab.last_close_error = None + + +def _record_close_outcome(tab: BrowserTab, released: bool, error: Optional[str]) -> None: + """Apply the result of a BF close attempt to the tab's release state. + A failed/unknown outcome keeps session_closed=False so the maintenance + sweeper retries; only a confirmed release marks the session closed.""" + now = datetime.now(timezone.utc) + if not tab.session_id: + tab.session_closed = True + tab.close_status = "none" + return + tab.close_attempts = (tab.close_attempts or 0) + 1 + tab.last_close_attempt_at = now + if released: + tab.session_closed = True + tab.close_status = "closed" + tab.last_close_error = None + else: + tab.session_closed = False + tab.close_status = "close_failed" + tab.last_close_error = error or "unknown close failure" + + +def _orphan_session_tombstone(db: Session, tab: BrowserTab, error: str) -> None: + """When a tab's old remote session could not be confirmed released before + being replaced (reconnect/persist swap), record it as a closed tab row so + the maintenance sweeper keeps retrying the release instead of the session + silently leaking against the per-key quota.""" + if not tab.session_id: + return + now = datetime.now(timezone.utc) + db.add(BrowserTab( + id=str(uuid.uuid4()), + workspace_id=tab.workspace_id, + url=tab.url or "about:blank", + title=tab.title, + status="closed", + created_by="system:orphaned-session", + shared_with=[], + session_id=tab.session_id, + bf_key_source=tab.bf_key_source, + bf_key_fingerprint=tab.bf_key_fingerprint, + session_closed=False, + close_status="close_failed", + close_attempts=1, + last_close_attempt_at=now, + last_close_error=error, + last_active_at=now, + )) + logger.warning("browser.session.orphaned tab=%s session=%s: %s", tab.id, tab.session_id, error) # --------------------------------------------------------------------------- @@ -137,6 +225,10 @@ def _tab_to_dict(tab: BrowserTab, context_name: str = None) -> dict: d["live_url"] = tab.live_url if tab.session_id: d["session_id"] = tab.session_id + if tab.last_error: + d["last_error"] = tab.last_error + # Deliberately NOT exposed: bf_key_source, bf_key_fingerprint and the + # close-retry bookkeeping — credential internals stay out of the API. if tab.context_id: d["context_id"] = tab.context_id d["persistent"] = True @@ -201,10 +293,24 @@ async def _ensure_connected(tab: BrowserTab, db: Session = None, workspace: Work if not tab.session_id and not manager.is_cloud: return # local mode, nothing to reconnect to - # --- Try reconnecting to the existing session first --- + # --- Resolve the credential reference for the EXISTING session --- + session_key = None + credential_rotated = False if tab.session_id: try: - await manager.reconnect(tab.id, tab.session_id) + session_key = resolve_tab_key(tab, workspace) + except BrowserCredentialError as e: + if e.reason == "credential_missing": + raise # no key at all — surface, don't guess + # credential_mismatch: the key was rotated. The old session must + # NOT be touched with the new key — record it for the sweeper + # (it will exhaust retries visibly) and recreate below. + credential_rotated = True + + # --- Try reconnecting to the existing session first --- + if tab.session_id and not credential_rotated: + try: + await manager.reconnect(tab.id, tab.session_id, api_key=session_key) # Sync URL/title from the live page live = await manager.get_current_url(tab.id) if live: @@ -221,12 +327,19 @@ async def _ensure_connected(tab: BrowserTab, db: Session = None, workspace: Work logger.info("Reconnect failed for tab %s (session %s), will create new session: %s", tab.id, tab.session_id, e) - # --- Session is dead — create a fresh one --- - # Clean up old session (best-effort) - try: - await manager.close_tab(tab.id, session_id_hint=tab.session_id) - except Exception: - pass + # --- Session is dead or unreachable — create a fresh one --- + if credential_rotated: + if db is not None: + _orphan_session_tombstone(db, tab, "credential_mismatch: key rotated before release") + tab.session_id = None + else: + # Clean up old session (best-effort); if the release isn't confirmed, + # keep it tracked so the sweeper retries instead of leaking it. + released, close_err = await manager.close_tab( + tab.id, session_id_hint=tab.session_id, api_key=session_key + ) + if tab.session_id and not released and db is not None: + _orphan_session_tombstone(db, tab, close_err or "close failed during session recreate") # Resolve persistent context (cookies/localStorage) if available bb_context_id = None @@ -239,14 +352,20 @@ async def _ensure_connected(tab: BrowserTab, db: Session = None, workspace: Work if ctx: bb_context_id = ctx.bb_context_id - bf_key = await _resolve_bf_key(workspace, db) if workspace and db else None + if workspace is not None and db is not None: + bf_key, bf_source = await _resolve_bf_key(workspace, db) + else: + bf_key, bf_source = session_key, tab.bf_key_source result = await manager.open_tab(tab.id, tab.url or "about:blank", bb_context_id=bb_context_id, api_key=bf_key) # Update the tab record with the new session info tab.session_id = manager.get_session_id(tab.id) tab.live_url = manager.get_live_url(tab.id) + _stamp_credential(tab, bf_key, bf_source) tab.url = result.get("url", tab.url) tab.title = result.get("title", tab.title) + warnings = result.get("warnings") or [] + tab.last_error = "; ".join(warnings) if warnings else None _touch(tab) logger.info("Tab %s auto-reconnected with new session %s", tab.id, tab.session_id) @@ -298,7 +417,26 @@ async def open_tab( tab_id = str(uuid.uuid4()) manager = BrowserManager.get() - bf_key = await _resolve_bf_key(workspace, db) + bf_key, bf_source = await _resolve_bf_key(workspace, db) + + # Ephemeral opens: enforce the per-key BF quota against the DB before + # spending a BF call, and tell the user which tabs can be closed. + if not body.context_id and manager.is_cloud_for(bf_key): + open_ephemeral = db.execute( + select(BrowserTab) + .where(BrowserTab.workspace_id == str(workspace.id)) + .where(BrowserTab.status == "active") + .where(BrowserTab.context_id.is_(None)) + .order_by(BrowserTab.last_active_at.asc()) + ).scalars().all() + if len(open_ephemeral) >= BF_EPHEMERAL_TAB_LIMIT: + return json_response( + ResponseCode.BAD_REQUEST, + f"Temporary tab limit reached ({len(open_ephemeral)}/{BF_EPHEMERAL_TAB_LIMIT}). " + "Close one of the open tabs first.", + data={"open_tabs": [_tab_to_dict(t) for t in open_ephemeral]}, + ) + try: result = await manager.open_tab(tab_id, body.url or "about:blank", bb_context_id=bb_context_id, api_key=bf_key) except RuntimeError as e: @@ -311,6 +449,8 @@ async def open_tab( if context_record: context_record.last_used_at = datetime.now(timezone.utc) + session_id = manager.get_session_id(tab_id) + warnings = result.get("warnings") or [] record = BrowserTab( id=tab_id, workspace_id=str(workspace.id), @@ -319,8 +459,13 @@ async def open_tab( created_by=body.source or "human:user", shared_with=[], context_id=body.context_id, - session_id=manager.get_session_id(tab_id), + session_id=session_id, live_url=manager.get_live_url(tab_id), + bf_key_source=bf_source if (bf_key and session_id) else None, + bf_key_fingerprint=key_fingerprint(bf_key) if session_id else None, + session_closed=not session_id, + close_status="open" if session_id else "none", + last_error="; ".join(warnings) if warnings else None, ) db.add(record) @@ -333,15 +478,28 @@ async def open_tab( ) db.add(usage) - event = Event( - type="workspace.browser.tab.opened", - source=body.source or "human:user", - target="core", - payload={"tab_id": tab_id, "url": record.url}, - ) - await _emit_event(event, workspace, db, token=x_workspace_token or workspace.password_hash) + # Commit BEFORE emitting the event: the BF session already exists, and a + # rejected/failed event pipeline must not roll back the only record of it + # (that's a leaked session the sweeper could never find). + db.commit() - return success_response(_tab_to_dict(record)) + try: + event = Event( + type="workspace.browser.tab.opened", + source=body.source or "human:user", + target="core", + payload={"tab_id": tab_id, "url": record.url}, + ) + await _emit_event(event, workspace, db, token=x_workspace_token or workspace.password_hash) + except Exception as e: + logger.warning("tab.opened event failed for %s (tab kept): %s", tab_id, e) + + # A partially failed init (session created, navigate/page-info failed) is + # still a created tab — the caller gets the tab plus explicit warnings. + data = _tab_to_dict(record) + if warnings: + data["warnings"] = warnings + return success_response(data) # --------------------------------------------------------------------------- @@ -454,7 +612,10 @@ async def navigate_tab( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: result = await manager.navigate(tab_id, body.url) @@ -502,11 +663,20 @@ async def reconnect_tab( manager = BrowserManager.get() - # Close old session gracefully (ignore errors — it's likely already dead) - try: - await manager.close_tab(tab_id, session_id_hint=tab.session_id) - except Exception: - pass + # Release the old session with ITS OWN credential; if the release can't + # be confirmed (failure or rotated key), leave a tombstone so the + # sweeper keeps the old session tracked instead of leaking it. + if tab.session_id: + try: + old_key = resolve_tab_key(tab, workspace) + released, close_err = await manager.close_tab(tab_id, session_id_hint=tab.session_id, api_key=old_key) + except BrowserCredentialError as e: + released, close_err = False, str(e) + manager._sessions.pop(tab_id, None) + manager._live_urls.pop(tab_id, None) + manager._tab_keys.pop(tab_id, None) + if not released: + _orphan_session_tombstone(db, tab, close_err or "close failed during reconnect") # Resolve persistent context if any bb_context_id = None @@ -520,22 +690,29 @@ async def reconnect_tab( bb_context_id = ctx.bb_context_id # Create a new session - bf_key = await _resolve_bf_key(workspace, db) + bf_key, bf_source = await _resolve_bf_key(workspace, db) try: result = await manager.open_tab(tab_id, tab.url or "about:blank", bb_context_id=bb_context_id, api_key=bf_key) except Exception as e: + db.commit() # keep any tombstone recorded above logger.error("Reconnect failed: %s", e) return json_response(ResponseCode.INTERNAL_ERROR, "Failed to reconnect browser tab") # Update DB record tab.session_id = manager.get_session_id(tab_id) tab.live_url = manager.get_live_url(tab_id) + _stamp_credential(tab, bf_key, bf_source) tab.url = result.get("url", tab.url) tab.title = result.get("title", tab.title) + warnings = result.get("warnings") or [] + tab.last_error = "; ".join(warnings) if warnings else None _touch(tab) db.commit() - return success_response(_tab_to_dict(tab)) + data = _tab_to_dict(tab) + if warnings: + data["warnings"] = warnings + return success_response(data) # --------------------------------------------------------------------------- @@ -560,7 +737,10 @@ async def click_tab( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: result = await manager.click(tab_id, body.selector) @@ -600,7 +780,10 @@ async def type_in_tab( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: await manager.type_text(tab_id, body.selector, body.text, append=body.append) @@ -638,7 +821,10 @@ async def press_key_in_tab( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: await manager.press_key(tab_id, body.key) @@ -676,7 +862,10 @@ async def evaluate_in_tab( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: result = await manager.evaluate(tab_id, body.expression) @@ -713,7 +902,10 @@ async def get_screenshot( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: data = await manager.screenshot(tab_id) @@ -765,7 +957,10 @@ async def get_snapshot( if not _verify_workspace_access(workspace, x_workspace_token, authorization): return json_response(ResponseCode.UNAUTHORIZED, "Invalid workspace credentials") - await _ensure_connected(tab, db, workspace) + try: + await _ensure_connected(tab, db, workspace) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) manager = BrowserManager.get() try: tree = await manager.snapshot(tab_id) @@ -862,24 +1057,40 @@ async def persist_tab( # Save current session state and create persistent context manager = BrowserManager.get() + bf_key, bf_source = await _resolve_bf_key(workspace, db) bb_context_id = None - if manager.is_cloud: + if manager.is_cloud_for(bf_key): try: await _ensure_connected(tab, db, workspace) - bb_context_id = await manager.create_bb_context(session_id=tab.session_id) + bb_context_id = await manager.create_bb_context(session_id=tab.session_id, tab_id=tab_id) + except BrowserCredentialError as e: + return json_response(ResponseCode.BAD_REQUEST, str(e)) except Exception as e: logger.error("Failed to create persistent context: %s", e) return json_response(ResponseCode.INTERNAL_ERROR, "Failed to create persistent context") # Close the current session and reopen with the context so that # future sessions restore cookies/localStorage from the saved state. - if manager.is_cloud and tab.session_id: + # The old and new sessions are distinct identities: the old one is + # released (or tombstoned for the sweeper) BEFORE tab.session_id is + # overwritten with the new one. + if manager.is_cloud_for(bf_key) and tab.session_id: try: current_url = tab.url - await manager.close_tab(tab_id, session_id_hint=tab.session_id) - result = await manager.open_tab(tab_id, current_url, bb_context_id=bb_context_id) + try: + old_key = resolve_tab_key(tab, workspace) + released, close_err = await manager.close_tab(tab_id, session_id_hint=tab.session_id, api_key=old_key) + except BrowserCredentialError as e: + released, close_err = False, str(e) + manager._sessions.pop(tab_id, None) + manager._live_urls.pop(tab_id, None) + manager._tab_keys.pop(tab_id, None) + if not released: + _orphan_session_tombstone(db, tab, close_err or "close failed during persist swap") + result = await manager.open_tab(tab_id, current_url, bb_context_id=bb_context_id, api_key=bf_key) tab.session_id = manager.get_session_id(tab_id) tab.live_url = manager.get_live_url(tab_id) + _stamp_credential(tab, bf_key, bf_source) tab.url = result.get("url", current_url) tab.title = result.get("title", tab.title) except Exception as e: @@ -1092,20 +1303,37 @@ async def close_tab( is_persistent = bool(tab.context_id) manager = BrowserManager.get() - await manager.close_tab(tab_id, session_id_hint=tab.session_id) + try: + close_key = resolve_tab_key(tab, workspace) + released, close_err = await manager.close_tab(tab_id, session_id_hint=tab.session_id, api_key=close_key) + except BrowserCredentialError as e: + # Never touch the old session with a wrong/rotated key. Record the + # failure; the sweeper will surface it as retry_exhausted. + released, close_err = False, str(e) + manager._sessions.pop(tab_id, None) + manager._live_urls.pop(tab_id, None) + manager._tab_keys.pop(tab_id, None) + # A failed/unknown BF close keeps session_closed=False so the maintenance + # sweeper retries — otherwise the session silently eats the per-key + # ephemeral quota while the UI shows no open tabs. + _record_close_outcome(tab, released, close_err) + db.commit() payload = {"tab_id": tab_id} if is_persistent: payload["context_id"] = tab.context_id payload["persistent"] = True - event = Event( - type="workspace.browser.tab.closed", - source="system", - target="core", - payload=payload, - ) - await _emit_event(event, workspace, db, token=x_workspace_token or workspace.password_hash) + try: + event = Event( + type="workspace.browser.tab.closed", + source="system", + target="core", + payload=payload, + ) + await _emit_event(event, workspace, db, token=x_workspace_token or workspace.password_hash) + except Exception as e: + logger.warning("tab.closed event failed for %s: %s", tab_id, e) return success_response({"id": tab_id, "status": "closed", "context_preserved": is_persistent}) diff --git a/workspace/backend/pytest.ini b/workspace/backend/pytest.ini new file mode 100644 index 000000000..895618d29 --- /dev/null +++ b/workspace/backend/pytest.ini @@ -0,0 +1,3 @@ +[pytest] +markers = + postgres: requires a real PostgreSQL server (set TEST_DATABASE_URL); skipped otherwise diff --git a/workspace/backend/tests/test_browser_postgres_concurrency.py b/workspace/backend/tests/test_browser_postgres_concurrency.py new file mode 100644 index 000000000..aa5def897 --- /dev/null +++ b/workspace/backend/tests/test_browser_postgres_concurrency.py @@ -0,0 +1,131 @@ +# -*- coding: utf-8 -*- +""" +Real-PostgreSQL concurrency tests for the browser maintenance CAS claims. + +These verify that the conditional-UPDATE claim used by the sweeper is a +genuine cross-connection arbiter, which SQLite cannot demonstrate (its +writer lock serialises everything). They need a real PostgreSQL server: + + export TEST_DATABASE_URL=postgresql://user:pass@localhost:5432/openagents_test + pytest -m postgres tests/test_browser_postgres_concurrency.py + +Without TEST_DATABASE_URL the whole module is skipped. SQLite-based tests +in test_browser_tab_leak.py cover the same logic as controlled-interleaving +simulations; only this module exercises true concurrent transactions. +""" + +import os +import threading +import uuid +from datetime import datetime, timezone + +import pytest + +TEST_DATABASE_URL = os.environ.get("TEST_DATABASE_URL", "") + +pytestmark = [ + pytest.mark.postgres, + pytest.mark.skipif( + not TEST_DATABASE_URL.startswith("postgresql"), + reason="TEST_DATABASE_URL not set to a PostgreSQL DSN", + ), +] + + +@pytest.fixture +def pg_engine(): + from sqlalchemy import create_engine + from app.database import Base + import app.models # noqa: F401 — register models + + engine = create_engine(TEST_DATABASE_URL) + Base.metadata.create_all(bind=engine) + yield engine + Base.metadata.drop_all(bind=engine) + engine.dispose() + + +def _insert_close_failed_tab(engine, ws_id: str) -> str: + from sqlalchemy import text + tab_id = str(uuid.uuid4()) + with engine.begin() as conn: + conn.execute(text( + "INSERT INTO workspaces (id, name, slug, password_hash, status, created_at) " + "VALUES (:id, 'pg-test', :slug, 'x', 'active', :now) ON CONFLICT DO NOTHING" + ), {"id": ws_id, "slug": f"pg-{ws_id[:8]}", "now": datetime.now(timezone.utc)}) + conn.execute(text( + "INSERT INTO browser_tabs (id, workspace_id, url, status, created_by, shared_with, " + "session_id, session_closed, close_status, close_attempts, created_at, last_active_at) " + "VALUES (:id, :ws, 'https://example.com', 'closed', 'human:user', '[]', " + "'sess-race', FALSE, 'close_failed', 0, :now, :now)" + ), {"id": tab_id, "ws": ws_id, "now": datetime.now(timezone.utc)}) + return tab_id + + +def test_only_one_worker_claims_a_close_failed_tab(pg_engine): + """Two connections race the same CAS claim; exactly one rowcount==1.""" + from sqlalchemy import text + + ws_id = str(uuid.uuid4()) + tab_id = _insert_close_failed_tab(pg_engine, ws_id) + + barrier = threading.Barrier(2) + results = [] + + def claim(): + with pg_engine.connect() as conn: + barrier.wait() + with conn.begin(): + r = conn.execute(text( + "UPDATE browser_tabs SET close_status='closing', last_close_attempt_at=:now " + "WHERE id=:id AND close_status='close_failed' AND session_id='sess-race'" + ), {"id": tab_id, "now": datetime.now(timezone.utc)}) + results.append(r.rowcount) + + threads = [threading.Thread(target=claim) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert sorted(results) == [0, 1], f"exactly one claim must win, got {results}" + + +def test_outcome_write_dropped_when_session_swapped(pg_engine): + """A persist/reconnect swap between claim and outcome invalidates the + outcome CAS — the new session's state is never clobbered.""" + from sqlalchemy import text + + ws_id = str(uuid.uuid4()) + tab_id = _insert_close_failed_tab(pg_engine, ws_id) + + with pg_engine.begin() as conn: + # Sweeper claims the row (session snapshot: sess-race) + r = conn.execute(text( + "UPDATE browser_tabs SET close_status='closing' " + "WHERE id=:id AND close_status='close_failed' AND session_id='sess-race'" + ), {"id": tab_id}) + assert r.rowcount == 1 + + with pg_engine.begin() as conn: + # Meanwhile a reconnect swaps in a fresh session + conn.execute(text( + "UPDATE browser_tabs SET session_id='sess-new', close_status='open', session_closed=FALSE " + "WHERE id=:id" + ), {"id": tab_id}) + + with pg_engine.begin() as conn: + # Sweeper writes its outcome with the stale snapshot — must be a no-op + r = conn.execute(text( + "UPDATE browser_tabs SET close_status='closed', session_closed=TRUE " + "WHERE id=:id AND close_status='closing' AND session_id='sess-race'" + ), {"id": tab_id}) + assert r.rowcount == 0 + + with pg_engine.connect() as conn: + row = conn.execute(text( + "SELECT session_id, close_status, session_closed FROM browser_tabs WHERE id=:id" + ), {"id": tab_id}).one() + assert row.session_id == "sess-new" + assert row.close_status == "open" + assert row.session_closed is False diff --git a/workspace/backend/tests/test_browser_tab_leak.py b/workspace/backend/tests/test_browser_tab_leak.py new file mode 100644 index 000000000..396a6df76 --- /dev/null +++ b/workspace/backend/tests/test_browser_tab_leak.py @@ -0,0 +1,727 @@ +# -*- coding: utf-8 -*- +""" +Tests for the shared-browser session-leak fixes. + +Covers the leak classes from the bug report and the follow-up review: + 1. BF session created but a later step failed → session must be recorded + and the caller must see explicit warnings (never silent success, never + a lost session). + 2. Close/reconnect after a restart resolve the key from the credential + reference persisted on the row (source + fingerprint, never plaintext); + rotated credentials fail loudly instead of touching the old session + with the new key. + 3. Failed/unknown closes keep session_closed=FALSE and step through the + close_status state machine (close_failed → retries → retry_exhausted, + never faked as released). + 4. The maintenance sweeper claims rows via CAS (close_status+session_id) + so multiple replicas never double-process, and a persist/reconnect + session swap is never clobbered. + +Concurrency here is simulated via controlled interleavings on SQLite; the +true multi-connection race is verified in test_browser_postgres_concurrency +(PostgreSQL only, gated on TEST_DATABASE_URL). +""" + +import asyncio +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +import app.database as database +from app.browser import BrowserManager +from app.browser_creds import BrowserCredentialError, key_fingerprint, redact, resolve_tab_key +from app.models import BrowserContext, BrowserTab, BrowserUsage, Workspace +from tests.conftest import TestingSessionLocal + +WS_KEY = "bf-secret-workspace-key-123456" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _create_workspace(client): + resp = client.post("/v1/workspaces", json={ + "name": "Leak Test Workspace", + "agent_name": "agent-leak", + "creator_email": "test@example.com", + }) + assert resp.status_code == 200 + data = resp.json()["data"] + return {"id": data["workspaceId"], "slug": data["slug"], "token": data["token"]} + + +def _set_workspace_key(db, ws_id, key=WS_KEY): + ws = db.query(Workspace).filter_by(id=ws_id).one() + settings = dict(ws.settings or {}) + settings["browserfabric_api_key"] = key + ws.settings = settings + db.commit() + + +def _mock_cloud_manager(session_id="sess-1"): + manager = MagicMock() + manager.is_cloud = True + manager.is_cloud_for = MagicMock(return_value=True) + manager.get_session_id.return_value = session_id + manager.get_live_url.return_value = None + manager.open_tab = AsyncMock(return_value={"url": "https://example.com", "title": "Example", "warnings": []}) + manager.close_tab = AsyncMock(return_value=(True, None)) + manager.reconnect = AsyncMock() + manager.get_current_url = AsyncMock(return_value={"url": "https://example.com", "title": "Example"}) + manager.delete_bb_context = MagicMock() + manager._pages = {} + manager._sessions = {} + manager._live_urls = {} + manager._tab_keys = {} + return manager + + +def _patch_manager_cls(MockManager, manager): + MockManager.get.return_value = manager + MockManager.provision_workspace_key = AsyncMock(return_value=None) + + +def _open_tab(client, ws, url="https://example.com"): + return client.post("/v1/browser/tabs", json={ + "url": url, + "network": ws["id"], + "source": "human:user", + }, headers={"X-Workspace-Token": ws["token"]}) + + +def _http_status_error(status_code): + req = httpx.Request("POST", "https://bf.example/api") + resp = httpx.Response(status_code, request=req) + return httpx.HTTPStatusError(f"HTTP {status_code}", request=req, response=resp) + + +# --------------------------------------------------------------------------- +# 1. Manager: no leak + explicit warnings when a step after create fails +# --------------------------------------------------------------------------- + +class TestManagerOpenTab: + def _manager(self, monkeypatch): + monkeypatch.setattr("app.browser.BROWSERFABRIC_API_KEY", "global-key") + return BrowserManager() + + def test_navigate_failure_keeps_session_and_reports_warning(self, monkeypatch): + manager = self._manager(monkeypatch) + + async def bf(tool_name, arguments=None, session_id=None, api_key=None, tab_id=None): + if tool_name == "create_session": + return {"success": True, "result": {"session_id": "sess-1", "share_url": "https://live"}} + if tool_name == "navigate": + raise RuntimeError("Browser Fabric error: nav boom") + return {"success": True, "result": {"url": "about:blank", "title": ""}} + + manager._bf_call = AsyncMock(side_effect=bf) + result = asyncio.run(manager.open_tab("tab-1", "https://example.com", api_key="ws-key")) + + assert manager.get_session_id("tab-1") == "sess-1" + assert any(w.startswith("navigation_failed:") for w in result["warnings"]) + + def test_page_info_failure_keeps_session_and_reports_warning(self, monkeypatch): + manager = self._manager(monkeypatch) + + async def bf(tool_name, arguments=None, session_id=None, api_key=None, tab_id=None): + if tool_name == "create_session": + return {"success": True, "result": {"session_id": "sess-1"}} + if tool_name == "navigate": + return {"success": True, "result": {}} + raise RuntimeError("Browser Fabric error: transient") + + manager._bf_call = AsyncMock(side_effect=bf) + result = asyncio.run(manager.open_tab("tab-1", "https://example.com", api_key="ws-key")) + + assert result["url"] == "https://example.com" + assert manager.get_session_id("tab-1") == "sess-1" + assert manager._tab_keys["tab-1"] == "ws-key" + assert any(w.startswith("page_info_failed:") for w in result["warnings"]) + + def test_create_failure_leaves_no_state(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(side_effect=RuntimeError("Browser Fabric error: limit reached")) + + with pytest.raises(RuntimeError): + asyncio.run(manager.open_tab("tab-1", "https://example.com", api_key="ws-key")) + + assert manager.get_session_id("tab-1") is None + assert "tab-1" not in manager._tab_keys + + def test_warning_text_never_contains_key(self, monkeypatch): + manager = self._manager(monkeypatch) + + async def bf(tool_name, arguments=None, session_id=None, api_key=None, tab_id=None): + if tool_name == "create_session": + return {"success": True, "result": {"session_id": "sess-1"}} + raise RuntimeError(f"Browser Fabric error: auth failed for ws-key") + + manager._bf_call = AsyncMock(side_effect=bf) + result = asyncio.run(manager.open_tab("tab-1", "https://example.com", api_key="ws-key")) + for w in result["warnings"]: + assert "ws-key" not in w + + +# --------------------------------------------------------------------------- +# 2. Manager: close outcome semantics (typed 404 vs unknown failures) +# --------------------------------------------------------------------------- + +class TestManagerClose: + def _manager(self, monkeypatch): + monkeypatch.setattr("app.browser.BROWSERFABRIC_API_KEY", "global-key") + return BrowserManager() + + def test_close_after_restart_uses_resolved_key(self, monkeypatch): + """Fresh manager (post-restart): close goes out with the explicitly + resolved credential, not the global one.""" + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(return_value={"success": True}) + + released, err = asyncio.run( + manager.close_tab("tab-1", session_id_hint="sess-1", api_key="ws-key") + ) + + assert (released, err) == (True, None) + manager._bf_call.assert_awaited_once_with("close_session", {}, "sess-1", api_key="ws-key") + + def test_close_404_is_confirmed_released(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(side_effect=_http_status_error(404)) + + released, err = asyncio.run(manager.close_tab("tab-1", session_id_hint="sess-1", api_key="k")) + assert (released, err) == (True, None) + + def test_close_non_404_http_error_not_released(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(side_effect=_http_status_error(502)) + + released, err = asyncio.run(manager.close_tab("tab-1", session_id_hint="sess-1", api_key="k")) + assert released is False + assert "502" in err + + def test_close_timeout_not_released(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(side_effect=httpx.ConnectTimeout("timed out")) + manager._sessions["tab-1"] = "sess-1" + + released, err = asyncio.run(manager.close_tab("tab-1")) + assert released is False + assert manager.get_session_id("tab-1") is None # local mapping still dropped + + def test_close_error_text_is_redacted(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._bf_call = AsyncMock(side_effect=RuntimeError("denied for key secret-key-9")) + + released, err = asyncio.run( + manager.close_tab("tab-1", session_id_hint="sess-1", api_key="secret-key-9") + ) + assert released is False + assert "secret-key-9" not in err + + def test_reconnect_restores_tab_key(self, monkeypatch): + manager = self._manager(monkeypatch) + asyncio.run(manager.reconnect("tab-1", "sess-1", api_key="ws-key")) + + assert manager.get_session_id("tab-1") == "sess-1" + assert manager._key_for_tab("tab-1") == "ws-key" + + def test_ops_pass_tab_id_for_key_resolution(self, monkeypatch): + manager = self._manager(monkeypatch) + manager._sessions["tab-1"] = "sess-1" + manager._tab_keys["tab-1"] = "ws-key" + manager._bf_call = AsyncMock(return_value={"success": True, "result": {"url": "u", "title": "t"}}) + + asyncio.run(manager.navigate("tab-1", "https://example.com")) + + for call in manager._bf_call.await_args_list: + assert call.kwargs.get("tab_id") == "tab-1" + + def test_per_workspace_key_without_global_env_is_cloud(self, monkeypatch): + monkeypatch.setattr("app.browser.BROWSERFABRIC_API_KEY", "") + manager = BrowserManager() + + async def bf(tool_name, arguments=None, session_id=None, api_key=None, tab_id=None): + return {"success": True, "result": {"session_id": "sess-1", "url": "u", "title": "t"}} + + manager._bf_call = AsyncMock(side_effect=bf) + asyncio.run(manager.open_tab("tab-1", "https://example.com", api_key="ws-key")) + + assert manager.get_session_id("tab-1") == "sess-1" # cloud branch, no Playwright + + +# --------------------------------------------------------------------------- +# 3. Credential reference: resolution, rotation, no plaintext anywhere +# --------------------------------------------------------------------------- + +class TestCredentialReference: + def test_redact_strips_key(self): + assert "sk-123" not in redact("error with sk-123 inside", "sk-123") + assert redact(None, "sk-123") is None + assert redact("no key here", None) == "no key here" + + def test_resolve_workspace_source(self, client, db): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"]) + workspace = db.query(Workspace).filter_by(id=ws["id"]).one() + tab = BrowserTab(workspace_id=ws["id"], url="u", created_by="human:user", + bf_key_source="workspace", bf_key_fingerprint=key_fingerprint(WS_KEY)) + assert resolve_tab_key(tab, workspace) == WS_KEY + + def test_resolve_rotated_key_raises_mismatch(self, client, db): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"], key="rotated-new-key") + workspace = db.query(Workspace).filter_by(id=ws["id"]).one() + tab = BrowserTab(workspace_id=ws["id"], url="u", created_by="human:user", + bf_key_source="workspace", bf_key_fingerprint=key_fingerprint(WS_KEY)) + with pytest.raises(BrowserCredentialError) as exc: + resolve_tab_key(tab, workspace) + assert exc.value.reason == "credential_mismatch" + assert "rotated-new-key" not in str(exc.value) + assert WS_KEY not in str(exc.value) + + def test_resolve_missing_key_raises(self, client, db): + ws = _create_workspace(client) + workspace = db.query(Workspace).filter_by(id=ws["id"]).one() + tab = BrowserTab(workspace_id=ws["id"], url="u", created_by="human:user", + bf_key_source="workspace", bf_key_fingerprint=key_fingerprint(WS_KEY)) + with pytest.raises(BrowserCredentialError) as exc: + resolve_tab_key(tab, workspace) + assert exc.value.reason == "credential_missing" + + @patch("app.routers.browser.BrowserManager") + def test_open_persists_reference_not_plaintext(self, MockManager, client, db, caplog): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"]) + manager = _mock_cloud_manager() + _patch_manager_cls(MockManager, manager) + + with caplog.at_level("DEBUG"): + resp = _open_tab(client, ws) + assert resp.status_code == 200 + + row = db.query(BrowserTab).filter_by(id=resp.json()["data"]["id"]).one() + assert row.bf_key_source == "workspace" + assert row.bf_key_fingerprint == key_fingerprint(WS_KEY) + # No plaintext key anywhere: DB row, API response, logs, repr + for value in row.__dict__.values(): + assert value != WS_KEY, "plaintext key stored in browser_tabs" + assert WS_KEY not in resp.text + assert WS_KEY not in repr(row) + for record in caplog.records: + assert WS_KEY not in record.getMessage() + + @patch("app.routers.browser.BrowserManager") + def test_close_after_rotation_fails_loudly_not_with_new_key(self, MockManager, client, db): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"]) + manager = _mock_cloud_manager(session_id="sess-old") + _patch_manager_cls(MockManager, manager) + + tab = _open_tab(client, ws).json()["data"] + _set_workspace_key(db, ws["id"], key="rotated-new-key") # rotate + + resp = client.delete(f"/v1/browser/tabs/{tab['id']}", + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200 + + manager.close_tab.assert_not_awaited() # never called BF with the new key + row = db.query(BrowserTab).filter_by(id=tab["id"]).one() + assert row.status == "closed" + assert row.session_closed is False + assert row.close_status == "close_failed" + assert "credential_mismatch" in row.last_close_error + assert "rotated-new-key" not in (row.last_close_error or "") + + @patch("app.routers.browser.BrowserManager") + def test_op_with_missing_credential_returns_structured_400(self, MockManager, client, db): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"]) + manager = _mock_cloud_manager(session_id="sess-1") + _patch_manager_cls(MockManager, manager) + tab = _open_tab(client, ws).json()["data"] + + _set_workspace_key(db, ws["id"], key=None) # key removed entirely + ws_row = db.query(Workspace).filter_by(id=ws["id"]).one() + settings = dict(ws_row.settings or {}) + settings.pop("browserfabric_api_key", None) + ws_row.settings = settings + db.commit() + + resp = client.post(f"/v1/browser/tabs/{tab['id']}/navigate", + json={"url": "https://other.com"}, + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 400 + assert "credential_missing" in resp.json()["message"] + + +# --------------------------------------------------------------------------- +# 4. Router: quota pre-check, warnings, close state machine, event failures +# --------------------------------------------------------------------------- + +class TestRouterOpenClose: + @patch("app.routers.browser.BrowserManager") + def test_ephemeral_quota_precheck(self, MockManager, client, monkeypatch): + monkeypatch.setattr("app.routers.browser.BF_EPHEMERAL_TAB_LIMIT", 2) + ws = _create_workspace(client) + manager = _mock_cloud_manager() + _patch_manager_cls(MockManager, manager) + + assert _open_tab(client, ws).status_code == 200 + assert _open_tab(client, ws).status_code == 200 + + resp = _open_tab(client, ws) + assert resp.status_code == 400 + body = resp.json() + assert "Temporary tab limit reached (2/2)" in body["message"] + assert len(body["data"]["open_tabs"]) == 2 + assert manager.open_tab.await_count == 2 # BF never called for the rejected open + + @patch("app.routers.browser.BrowserManager") + def test_bf_limit_error_maps_to_structured_400(self, MockManager, client): + ws = _create_workspace(client) + manager = _mock_cloud_manager() + manager.open_tab = AsyncMock(side_effect=RuntimeError( + "Browser Fabric error: 已到达临时标签限制（3/3）。先关闭一个标签页。")) + _patch_manager_cls(MockManager, manager) + + resp = _open_tab(client, ws) + assert resp.status_code == 400 + assert "临时标签限制" in resp.json()["message"] + + @patch("app.routers.browser.BrowserManager") + def test_persistent_open_skips_ephemeral_quota(self, MockManager, client, monkeypatch, db): + monkeypatch.setattr("app.routers.browser.BF_EPHEMERAL_TAB_LIMIT", 1) + ws = _create_workspace(client) + manager = _mock_cloud_manager() + _patch_manager_cls(MockManager, manager) + + assert _open_tab(client, ws).status_code == 200 # fills the quota + + ctx = BrowserContext(workspace_id=ws["id"], name="Reddit", bb_context_id="bbctx-1", + created_by="human:user", shared_with=[]) + db.add(ctx) + db.commit() + + resp = client.post("/v1/browser/tabs", json={ + "url": "https://reddit.com", "network": ws["id"], + "source": "human:user", "context_id": ctx.id, + }, headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200, resp.json() + + @patch("app.routers.browser.BrowserManager") + def test_open_with_init_warnings_returns_tab_plus_warnings(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager() + manager.open_tab = AsyncMock(return_value={ + "url": "https://example.com", "title": "", + "warnings": ["navigation_failed: Browser Fabric error: timeout"], + }) + _patch_manager_cls(MockManager, manager) + + resp = _open_tab(client, ws) + assert resp.status_code == 200 + data = resp.json()["data"] + assert data["id"] + assert data["status"] == "active" + assert any("navigation_failed" in w for w in data["warnings"]) + + row = db.query(BrowserTab).filter_by(id=data["id"]).one() + assert row.status == "active" + assert "navigation_failed" in row.last_error + assert row.session_id == "sess-1" # session recorded despite init failure + + @patch("app.routers.browser.BrowserManager") + def test_event_pipeline_failure_does_not_lose_tab(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager() + _patch_manager_cls(MockManager, manager) + + with patch("app.routers.browser._emit_event", AsyncMock(side_effect=RuntimeError("pipeline down"))): + resp = _open_tab(client, ws) + assert resp.status_code == 200 + + row = db.query(BrowserTab).filter_by(id=resp.json()["data"]["id"]).one() + assert row.session_id == "sess-1" # committed before the event + + @patch("app.routers.browser.BrowserManager") + def test_failed_bf_close_keeps_session_open_in_db(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager(session_id="sess-fail") + manager.close_tab = AsyncMock(return_value=(False, "HTTP 502 closing session")) + _patch_manager_cls(MockManager, manager) + + tab = _open_tab(client, ws).json()["data"] + resp = client.delete(f"/v1/browser/tabs/{tab['id']}", + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200 + + row = db.query(BrowserTab).filter_by(id=tab["id"]).one() + assert row.status == "closed" + assert row.session_closed is False + assert row.close_status == "close_failed" + assert row.close_attempts == 1 + assert "502" in row.last_close_error + + @patch("app.routers.browser.BrowserManager") + def test_successful_close_marks_session_closed(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager(session_id="sess-ok") + _patch_manager_cls(MockManager, manager) + + tab = _open_tab(client, ws).json()["data"] + resp = client.delete(f"/v1/browser/tabs/{tab['id']}", + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200 + + row = db.query(BrowserTab).filter_by(id=tab["id"]).one() + assert row.session_closed is True + assert row.close_status == "closed" + assert manager.close_tab.await_args.kwargs.get("session_id_hint") == "sess-ok" + + +# --------------------------------------------------------------------------- +# 5. Maintenance sweeper: claims, retries, exhaustion, swap safety +# --------------------------------------------------------------------------- + +class TestBrowserSweep: + def _run_sweep(self, manager, monkeypatch): + monkeypatch.setattr(database, "SessionLocal", TestingSessionLocal) + with patch("app.browser_maintenance.BrowserManager") as MockManager: + MockManager.get.return_value = manager + from app.browser_maintenance import sweep_browser_tabs + return asyncio.run(sweep_browser_tabs()) + + def _add_tab(self, db, ws_id, *, status="active", context_id=None, + session_id="sess-1", close_status="open", session_closed=False, + idle_minutes=0, closing_minutes=None, source=None, fingerprint=None): + tab = BrowserTab( + workspace_id=ws_id, + url="https://example.com", + status=status, + created_by="openagents:agent-leak", + shared_with=[], + context_id=context_id, + session_id=session_id, + session_closed=session_closed, + close_status=close_status, + bf_key_source=source, + bf_key_fingerprint=fingerprint, + last_active_at=datetime.now(timezone.utc) - timedelta(minutes=idle_minutes), + last_close_attempt_at=( + datetime.now(timezone.utc) - timedelta(minutes=closing_minutes) + if closing_minutes is not None else None + ), + ) + db.add(tab) + db.flush() + db.add(BrowserUsage( + workspace_id=ws_id, tab_id=tab.id, session_id=session_id, + opened_by="openagents:agent-leak", + )) + db.commit() + return tab.id + + def test_reaps_idle_ephemeral_tab(self, client, db, monkeypatch): + ws = _create_workspace(client) + _set_workspace_key(db, ws["id"]) + tab_id = self._add_tab(db, ws["id"], idle_minutes=120, + source="workspace", fingerprint=key_fingerprint(WS_KEY)) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["reaped"] == 1 + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.status == "closed" + assert row.close_status == "closed" + assert row.session_closed is True + usage = db.query(BrowserUsage).filter_by(tab_id=tab_id).one() + assert usage.ended_at is not None + # Reap resolved the key from the credential reference on the row + assert manager.close_tab.await_args.kwargs.get("api_key") == WS_KEY + + def test_skips_fresh_and_persistent_tabs(self, client, db, monkeypatch): + ws = _create_workspace(client) + ctx = BrowserContext(workspace_id=ws["id"], name="Reddit", bb_context_id="bbctx-1", + created_by="human:user", shared_with=[]) + db.add(ctx) + db.flush() + self._add_tab(db, ws["id"], idle_minutes=1) # fresh + self._add_tab(db, ws["id"], context_id=ctx.id, idle_minutes=120) # persistent + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["reaped"] == 0 + manager.close_tab.assert_not_awaited() + + def test_retries_orphaned_session_release(self, client, db, monkeypatch): + ws = _create_workspace(client) + tab_id = self._add_tab(db, ws["id"], status="closed", + close_status="close_failed", idle_minutes=30) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["released"] == 1 + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.session_closed is True + assert row.close_status == "closed" + + def test_failed_retry_stays_pending_with_bookkeeping(self, client, db, monkeypatch): + ws = _create_workspace(client) + tab_id = self._add_tab(db, ws["id"], status="closed", + close_status="close_failed", idle_minutes=30) + manager = _mock_cloud_manager() + manager.close_tab = AsyncMock(return_value=(False, "HTTP 502 closing session")) + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["release_failed"] == 1 + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.session_closed is False + assert row.close_status == "close_failed" # picked up again next sweep + assert row.close_attempts == 1 + assert row.last_close_attempt_at is not None + assert "502" in row.last_close_error + + def test_exhausts_after_retry_window_without_faking_release(self, client, db, monkeypatch, caplog): + ws = _create_workspace(client) + tab_id = self._add_tab(db, ws["id"], status="closed", + close_status="close_failed", idle_minutes=60 * 10) # 10h old + manager = _mock_cloud_manager() + + import logging + with caplog.at_level(logging.ERROR, logger="app.browser_maintenance"): + stats = self._run_sweep(manager, monkeypatch) + + assert stats["exhausted"] == 1 + assert stats["released"] == 0 + manager.close_tab.assert_not_awaited() + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.close_status == "retry_exhausted" + assert row.session_closed is False # NEVER faked as released + assert any("retry_exhausted" in r.getMessage() for r in caplog.records) + + def test_exhausted_rows_not_retried_again(self, client, db, monkeypatch): + ws = _create_workspace(client) + self._add_tab(db, ws["id"], status="closed", + close_status="retry_exhausted", idle_minutes=30) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["released"] == 0 and stats["exhausted"] == 0 + manager.close_tab.assert_not_awaited() + + def test_no_session_rows_produce_no_work(self, client, db, monkeypatch): + ws = _create_workspace(client) + self._add_tab(db, ws["id"], status="closed", session_id=None, + close_status="none", session_closed=True, idle_minutes=120) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert all(v == 0 for v in stats.values()) + manager.close_tab.assert_not_awaited() + + def test_row_claimed_by_other_worker_is_skipped(self, client, db, monkeypatch): + """A fresh 'closing' claim (another replica mid-close) is not touched.""" + ws = _create_workspace(client) + self._add_tab(db, ws["id"], status="closed", + close_status="closing", closing_minutes=1, idle_minutes=30) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["stale_recovered"] == 0 + manager.close_tab.assert_not_awaited() + + def test_stale_closing_claim_is_recovered_and_retried(self, client, db, monkeypatch): + """A 'closing' claim from a crashed worker flips back and gets retried.""" + ws = _create_workspace(client) + tab_id = self._add_tab(db, ws["id"], status="closed", + close_status="closing", closing_minutes=30, idle_minutes=30) + manager = _mock_cloud_manager() + + stats = self._run_sweep(manager, monkeypatch) + + assert stats["stale_recovered"] == 1 + assert stats["released"] == 1 + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.close_status == "closed" + + def test_session_swap_between_claim_and_outcome_is_not_clobbered(self, client, db, monkeypatch): + """If the session_id changes while the sweeper is closing (persist/ + reconnect swap), the outcome write must be dropped.""" + ws = _create_workspace(client) + tab_id = self._add_tab(db, ws["id"], idle_minutes=120, session_id="sess-old") + manager = _mock_cloud_manager() + + async def close_and_swap(*args, **kwargs): + swap_db = TestingSessionLocal() + try: + swap_db.query(BrowserTab).filter_by(id=tab_id).update({"session_id": "sess-new"}) + swap_db.commit() + finally: + swap_db.close() + return True, None + + manager.close_tab = AsyncMock(side_effect=close_and_swap) + + self._run_sweep(manager, monkeypatch) + + db.expire_all() + row = db.query(BrowserTab).filter_by(id=tab_id).one() + assert row.session_id == "sess-new" + assert row.session_closed is False # stale outcome dropped + assert row.close_status != "closed" + + +# --------------------------------------------------------------------------- +# 6. Session swaps leave tombstones instead of leaking the old session +# --------------------------------------------------------------------------- + +class TestOrphanTombstones: + @patch("app.routers.browser.BrowserManager") + def test_reconnect_with_failed_old_close_leaves_tombstone(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager(session_id="sess-old") + _patch_manager_cls(MockManager, manager) + tab = _open_tab(client, ws).json()["data"] + + manager.get_session_id.return_value = "sess-new" + manager.close_tab = AsyncMock(return_value=(False, "HTTP 502 closing session")) + + resp = client.post(f"/v1/browser/tabs/{tab['id']}/reconnect", + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200 + assert resp.json()["data"]["session_id"] == "sess-new" + + tombstone = db.query(BrowserTab).filter_by(session_id="sess-old", status="closed").one() + assert tombstone.created_by == "system:orphaned-session" + assert tombstone.close_status == "close_failed" + assert tombstone.session_closed is False # sweeper will retry it + + @patch("app.routers.browser.BrowserManager") + def test_reconnect_with_confirmed_old_close_leaves_no_tombstone(self, MockManager, client, db): + ws = _create_workspace(client) + manager = _mock_cloud_manager(session_id="sess-old") + _patch_manager_cls(MockManager, manager) + tab = _open_tab(client, ws).json()["data"] + + manager.get_session_id.return_value = "sess-new" + + resp = client.post(f"/v1/browser/tabs/{tab['id']}/reconnect", + headers={"X-Workspace-Token": ws["token"]}) + assert resp.status_code == 200 + + assert db.query(BrowserTab).filter_by(session_id="sess-old").count() == 0