proxy.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""browser-broker proxy — a transparent Chrome DevTools Protocol endpoint that
enforces tab ownership for EVERY client, with zero changes to any of them.
The real browser runs on an upstream port (default 9229). This process
listens on the port every tool already expects (default 9222) and speaks
CDP back at them:
agent-browser --cdp 9222, chrome-devtools-mcp --browser-url, Playwright's
connectOverCDP, Puppeteer.connect, browser-use, a curl to /json, or a raw
websocket — all unchanged.
What each of them gets is a browser in which the ONLY tabs that exist are the
ones this client was given. A tool that "grabs the first page" grabs its own
leased page, because that is the only page it can see. The human's tabs, and
every other agent's tabs, are simply not there.
Identity comes for free from the connection:
* a browser-level websocket (/devtools/browser/...) is one client for as
long as it stays connected; its tabs are released when it goes away.
* a legacy HTTP client (GET /json, PUT /json/new, then a page websocket)
claims a tab by attaching to it; the tab is released when the page socket
closes. A pool tab is handed to one /json caller and one page socket only.
* the REST lease API on 9223 (broker_client.py) still works and is the way
to hold a tab across reconnects, with a TTL.
Focus-stealing commands (Target.activateTarget, Page.bringToFront,
/json/activate) are answered with success and never forwarded, so no agent
can bring itself in front of the human. Agent tabs are ordinary background
tabs in the window that is already open — they never take focus, and they
keep rendering because the browser is launched with backgrounding disabled
(launch-browser.sh).
Ownership rule, deliberately dumb: this proxy can only hand out tabs it
created. It never enumerates, attaches to, or closes anything else.
Browser-wide commands that would hit the human (Browser.close, Browser.crash,
Browser.setWindowBounds) are swallowed too. Puppeteer's browser.close() sends
Browser.close, and through a plain CDP port that quits the human's whole browser.
"""
import asyncio, concurrent.futures, json, os, stat, sys, time, uuid, threading, subprocess
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs, unquote
import websockets, logging
if os.environ.get("BROKER_DEBUG"):
logging.basicConfig(level=logging.DEBUG, format="%(name)s %(message)s")
else:
# Handshake failures from port probes printed a full traceback each, and grew
# launchd.err to 28 MB on the M5 (2026-10-06). They are noise, not errors.
logging.getLogger("websockets.server").setLevel(logging.CRITICAL)
from websockets.asyncio.client import connect as ws_connect
from websockets.asyncio.server import serve as ws_serve
from websockets.http11 import Response
from websockets.datastructures import Headers
UPSTREAM = int(os.environ.get("BROKER_UPSTREAM_PORT", 9229)) # the real browser
LISTEN = int(os.environ.get("BROKER_PROXY_PORT", 9222)) # what tools dial
API = int(os.environ.get("BROKER_PORT", 9223)) # REST lease API
DEFAULT_TTL = int(os.environ.get("BROKER_TTL", 300))
UNATTACHED_TTL = 60 # a tab handed to a legacy client that never attached
HEARTBEAT_EVERY = int(os.environ.get("BROKER_HEARTBEAT", 3600))
LOG = os.path.expanduser(os.environ.get("BROKER_LOG",
os.path.join(os.path.dirname(os.path.abspath(__file__)), "broker.log")))
FOCUS_STEALERS = {"Target.activateTarget", "Page.bringToFront"}
# Commands that act on the human's browser as a whole. Answered with success and
# never forwarded. Browser.close is the one that bit: Puppeteer's browser.close()
# sends it, and it quit Matt's whole Brave, every window and tab.
BROWSER_KILLERS = {"Browser.close", "Browser.crash", "Browser.crashGpuProcess",
"Browser.setWindowBounds"}
LOG_MAX = 5 * 1024 * 1024 # broker.log rolls to broker.log.1 past this
STDIO_MAX = 10 * 1024 * 1024 # launchd.err / proxy.err get emptied past this
STATE = os.path.join(os.path.dirname(os.path.abspath(LOG)), "tabs.json")
POOL_HOLD = 10 # seconds a listed pool tab is reserved for its /json caller
BROWSER_APP = os.environ.get("BROWSER_APP", "Brave Browser") # for the one allowed activate
STAGE_DWELL = int(os.environ.get("BROKER_STAGE_DWELL", 20)) # seconds a shown page is protected
STAGE_WAIT = int(os.environ.get("BROKER_STAGE_WAIT", 30)) # seconds a caller will queue
# The macOS app-activation step of the stage. The test suite turns it off so a
# test run never brings a real window forward on the machine running it.
STAGE_ACTIVATE_APP = os.environ.get("BROKER_STAGE_ACTIVATE_APP", "1") != "0"
# The five rules (2026-10-23). See "When the human helps" in the README.
HUMAN_QUIET = int(os.environ.get("BROKER_HUMAN_QUIET", 20)) # seconds his last input protects a tab
AGENT_ECHO = 1.5 # trusted events this soon after an agent's own CDP input are the agent's
CHURN_WINDOW = int(os.environ.get("BROKER_CHURN_WINDOW", 600)) # seconds an open/close cycle is remembered
CHURN_LIMIT = int(os.environ.get("BROKER_CHURN_LIMIT", 2)) # cycles allowed before "stop and ask"
PORT_GRACE = 8 # seconds a restarting proxy waits for the old one to let go
def log(msg):
line = f"[{datetime.now().isoformat(timespec='seconds')}] {msg}"
print(line, flush=True)
try:
if os.path.getsize(LOG) > LOG_MAX:
os.replace(LOG, LOG + ".1")
except OSError:
pass
try:
with open(LOG, "a") as f:
f.write(line + "\n")
except OSError:
pass
# ---------------------------------------------------------------------------
# upstream browser connection (the proxy's own control channel)
# ---------------------------------------------------------------------------
class Upstream:
def __init__(self):
self.ws = None
self._id = 0
self._pending = {}
self._lock = asyncio.Lock()
self.version = {}
async def _http(self, path):
import urllib.request
def get():
with urllib.request.urlopen(f"http://127.0.0.1:{UPSTREAM}{path}", timeout=10) as r:
return json.load(r)
return await asyncio.to_thread(get)
async def connect(self):
self.version = await self._http("/json/version")
self.ws = await ws_connect(self.version["webSocketDebuggerUrl"], max_size=None)
asyncio.create_task(self._reader())
async def _reader(self):
try:
async for raw in self.ws:
m = json.loads(raw)
fut = self._pending.pop(m.get("id"), None)
if fut and not fut.done():
fut.set_result(m)
except Exception as e:
log(f"upstream control socket dropped: {e}")
self.ws = None
# Anyone waiting on a reply would otherwise sit out the full 20s timeout.
pending, self._pending = self._pending, {}
for fut in pending.values():
if not fut.done():
fut.set_exception(ConnectionError("browser control socket dropped"))
async def send(self, method, **params):
async with self._lock:
if self.ws is None:
await self.connect()
self._id += 1
mid = self._id
fut = asyncio.get_running_loop().create_future()
self._pending[mid] = fut
try:
await self.ws.send(json.dumps({"id": mid, "method": method, "params": params}))
m = await asyncio.wait_for(fut, 20)
finally:
self._pending.pop(mid, None)
if "error" in m:
raise RuntimeError(m["error"].get("message"))
return m.get("result", {})
async def alive(self):
try:
await self.send("Browser.getVersion")
return True
except Exception:
return False
# ---------------------------------------------------------------------------
# registry: which targets we made, who holds them
# ---------------------------------------------------------------------------
class Registry:
def __init__(self, up):
self.up = up
self.tabs = {} # target_id -> dict(owner, kind, hidden, created, expires, attached, lease_id)
self.loop = None
def _save(self):
"""Write down which tabs are ours, so a proxy that crashed or restarted
can close them next time instead of leaving them in Matt's window with
nobody owning them."""
try:
tmp = STATE + ".tmp"
with open(tmp, "w") as f:
json.dump(list(self.tabs), f)
os.replace(tmp, STATE)
except OSError:
pass
async def close_leftovers(self):
"""Startup: close tabs a previous run of this proxy made and never closed.
Only IDs from our own record are touched, never anything else."""
try:
with open(STATE) as f:
old = set(json.load(f))
except (OSError, ValueError):
return
try:
live = {t["targetId"] for t in (await self.up.send("Target.getTargets")).get("targetInfos", [])}
except Exception:
return
closed = 0
for tid in old & live:
# 2026-10-23: a broker restarted mid-task closed a tab the human had
# been working in. The listener's record lives in the page itself, so
# it outlasts the broker: a leftover he ever typed in is his now.
try:
touched = bool(json.loads(await page_eval(tid, "JSON.stringify("
"(window.__bbHuman && window.__bbHuman.events) || [])", 3) or "[]"))
except Exception:
touched = False
if touched:
log(f"left tab {tid[:12]} from the last run open: the human worked in it")
continue
try:
await self.up.send("Target.closeTarget", targetId=tid)
closed += 1
except Exception:
pass
if closed:
log(f"closed {closed} tab(s) left over from the last run")
self._save()
async def create(self, owner, url, hidden=True, kind="rest", ttl=None, context=None):
# A plain background tab in the window that is already open. Matt asked
# for this on 2026-09-07: parking agent work in its own off-screen window
# shoved his browser aside every time an agent ran. background=True means
# it opens behind whatever he is looking at and never takes focus, and
# the ownership filter below -- not the window -- is what keeps agents
# out of each other's tabs, so nothing about the isolation changes.
args = {"url": url or "about:blank", "background": True}
if context:
args["browserContextId"] = context
res = await self.up.send("Target.createTarget", **args)
tid = res["targetId"]
# Register ownership BEFORE parking the window: Chrome tells auto-attached
# clients about the new target immediately, and their filter must be
# able to see that it is theirs.
self.tabs[tid] = {"owner": owner, "kind": kind, "hidden": hidden,
"created": time.time(), "attached": 0,
"expires": time.time() + (ttl or UNATTACHED_TTL),
"lease_id": uuid.uuid4().hex[:12] if kind == "rest" else None,
"listed_until": 0}
self._save()
HUMANS.watch(tid)
log(f"tab {tid[:12]} -> {owner} ({kind}{', hidden' if hidden else ''})")
return tid
async def adopt(self, tid, owner, ttl):
"""Lease an existing job tab instead of opening a copy of it (rule 1).
The tab stays the job's: releasing this lease never closes it, bb-close does."""
self.tabs[tid] = {"owner": owner, "kind": "rest", "hidden": False, "created": time.time(),
"attached": 1, "expires": time.time() + ttl,
"lease_id": uuid.uuid4().hex[:12], "listed_until": 0, "shared": True}
self._save()
HUMANS.watch(tid)
log(f"tab {tid[:12]} -> {owner} (lease on its own shown job tab, no copy opened)")
return tid
def owned_by(self, owner):
return [t for t, i in self.tabs.items() if i["owner"] == owner]
def free_pool(self):
"""Tabs made for legacy /json clients that nobody has attached to yet."""
now = time.time()
return [t for t, i in self.tabs.items()
if i["kind"] == "pool" and i["attached"] == 0 and i["listed_until"] <= now]
async def close(self, tid, why=""):
"""Close a tab the broker made. True if it closed.
Two exceptions, and both leave the tab open: it is also an open job tab
on the human's screen (bb-close finishes that), or the human is working
in it right now (rule 4: it is his, and closing it would wipe his work)."""
info = self.tabs.pop(tid, None)
if info is None:
return False
self._save()
if STAGE is not None and STAGE.job_of(tid):
log(f"tab {tid[:12]} lease ended ({info['owner']}, {why or 'released'}); kept open, "
f"it is the shown job tab of {STAGE.job_of(tid)}")
return False
if HUMANS.active(tid):
HUMANS.forget(tid)
log(f"tab {tid[:12]} left open for the human ({info['owner']}, {why or 'released'}): "
f"he is working in it")
return False
HUMANS.forget(tid)
try:
await self.up.send("Target.closeTarget", targetId=tid)
except Exception:
pass
log(f"tab {tid[:12]} closed ({info['owner']}{', ' + why if why else ''})")
return True
async def release_owner(self, owner, why="client disconnected"):
for tid in self.owned_by(owner):
await self.close(tid, why)
async def reap(self):
now = time.time()
for tid, i in list(self.tabs.items()):
if i["kind"] == "ws":
continue # bound to a live connection
if i["kind"] == "rest" and i["attached"]:
if i["expires"] <= now:
await self.close(tid, "lease expired")
elif i["attached"] == 0 and i["expires"] <= now:
await self.close(tid, "never attached")
# drop bookkeeping for tabs that vanished under us (browser restart)
try:
live = {t["targetId"] for t in (await self.up.send("Target.getTargets")).get("targetInfos", [])}
except Exception:
return
gone = [tid for tid in self.tabs if tid not in live]
for tid in gone:
self.tabs.pop(tid, None)
HUMANS.forget(tid)
if gone:
self._save()
def status(self):
now = time.time()
return {"tabs": [{"target": t[:12], "owner": i["owner"], "kind": i["kind"],
"hidden": i["hidden"], "attached": i["attached"],
"expires_in": round(i["expires"] - now, 1) if i["kind"] != "ws" else None,
"lease_id": i["lease_id"]} for t, i in self.tabs.items()],
"count": len(self.tabs)}
def target_info(self, tid):
"""Shape a /json entry for a tab we own."""
return {"id": tid, "type": "page", "title": "", "url": "",
"webSocketDebuggerUrl": f"ws://127.0.0.1:{LISTEN}/devtools/page/{tid}",
"devtoolsFrontendUrl": f"/devtools/inspector.html?ws=127.0.0.1:{LISTEN}/devtools/page/{tid}"}
# ---------------------------------------------------------------------------
# the human: notice when he is working in a tab (rules 2-4, 2026-10-23)
# ---------------------------------------------------------------------------
# Matt, 2026-10-23: agents opened a tab, started a form, closed it, opened it again
# "over 10 times right before my eyes". When he typed in the form or solved the
# captcha himself, the agent never noticed: it reloaded or re-showed the page and
# wiped his work. Everything below exists so the broker can tell, in code, that a
# person is using a tab, and refuse to pull it out from under him.
# Injected into every tab the broker opens or shows. It records trusted input
# only (event.isTrusted: a real keyboard or mouse; a page script cannot fake it)
# and reports it two ways: a window variable any reader can poll, and a CDP
# binding that pushes each event to the broker the moment it happens, which
# also survives the page navigating away (a submitted form). Focus moving into
# an iframe counts too, because a cross-site captcha frame never lets its own
# clicks bubble up to the page. CDP Input events an agent sends are trusted as
# well; the broker sees those go through and discounts them (AGENT_ECHO).
HUMAN_BINDING = "__bbHumanEvent"
HUMAN_JS = r"""(() => {
if (window.__bbHuman) return 'already';
const s = {v: 1, events: []};
try { Object.defineProperty(window, '__bbHuman', {value: s}); } catch (e) { return 'blocked'; }
let last = 0;
const send = (kind) => {
const t = Date.now() / 1000;
if (t - last < 0.25) return;
last = t;
s.events.push([t, kind]);
if (s.events.length > 50) s.events.shift();
try { if (typeof window.__bbHumanEvent === 'function') window.__bbHumanEvent(JSON.stringify([t, kind])); } catch (e) {}
};
const note = (e) => { if (e.isTrusted) send(e.type); };
for (const k of ['pointerdown', 'keydown', 'input', 'paste', 'change']) addEventListener(k, note, true);
addEventListener('blur', (e) => {
if (!e.isTrusted || e.target !== window) return;
setTimeout(() => { const a = document.activeElement; if (a && a.tagName === 'IFRAME') send('frame'); }, 0);
}, true);
return 'installed';
})()"""
def page_key(url):
"""What "the same page" means everywhere here: the URL without its #fragment or trailing slash."""
return (url or "").split("#", 1)[0].rstrip("/")
def site_of(url):
"""The site a job is pinned to: the host without www. (the page itself when there is no host)."""
try:
host = (urlparse(url or "").hostname or "").lower()
except ValueError:
host = ""
return (host[4:] if host.startswith("www.") else host) or page_key(url)
async def page_eval(tid, expr, timeout=5):
"""Evaluate `expr` in tab `tid` over a short-lived page socket and return its value.
One Runtime.evaluate, nothing else: no navigation, no focus, no input."""
async with ws_connect(f"ws://127.0.0.1:{UPSTREAM}/devtools/page/{tid}", max_size=None) as ws:
await ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate",
"params": {"expression": expr, "returnByValue": True}}))
while True:
m = json.loads(await asyncio.wait_for(ws.recv(), timeout))
if m.get("id") == 1:
return m.get("result", {}).get("result", {}).get("value")
class HumanBusy(Exception):
"""An agent asked to reload, navigate, close or type into a tab the human is using."""
def __init__(self, tid, seconds_ago, action):
self.tid, self.seconds_ago, self.action = tid, seconds_ago, action
super().__init__(
f"the human is working in this tab right now (his last input was {seconds_ago:.0f}s ago), "
f"so it was not {action}. Re-read the page (bb-peek) and continue from what it shows; "
f"do not start over.")
class Humans:
"""Per tab: when did a person last type, click or paste in it?
A tab counts as in use for HUMAN_QUIET seconds after his last input. While it
is, no agent may reload it, navigate it, close it or send it input, through any
path the broker controls."""
def __init__(self):
self.events = {} # tid -> [(epoch seconds, kind)], oldest first
self.agent = {} # tid -> [epoch seconds an agent sent CDP input]
self.tasks = {} # tid -> the watcher task holding the listener in place
# ---- what happened -------------------------------------------------------
def agent_input(self, tid, t=None):
a = self.agent.setdefault(tid, [])
a.append(time.time() if t is None else t)
del a[:-200]
def _is_echo(self, tid, t):
return any(a - 0.25 <= t <= a + AGENT_ECHO for a in self.agent.get(tid, ()))
def note(self, tid, t, kind):
"""Record one trusted input event. False when it was really the agent's own input."""
if self._is_echo(tid, t):
return False
ev = self.events.setdefault(tid, [])
if (t, kind) in ev:
return False
ev.append((t, kind))
ev.sort()
del ev[:-100]
return True
def merge(self, tid, events):
"""Fold in events read back from the page's own window.__bbHuman record."""
for e in events or ():
try:
self.note(tid, float(e[0]), str(e[1])[:20])
except (TypeError, ValueError, IndexError):
pass
# ---- what that means -----------------------------------------------------
def last(self, tid):
ev = self.events.get(tid)
return ev[-1][0] if ev else None
def seconds_ago(self, tid, now=None):
t = self.last(tid)
return None if t is None else max(0.0, (now or time.time()) - t)
def active(self, tid, now=None):
ago = self.seconds_ago(tid, now)
return ago is not None and ago < HUMAN_QUIET
def since(self, tid):
"""Start of his current stretch of work in this tab (inputs less than HUMAN_QUIET apart)."""
ev = self.events.get(tid)
if not ev:
return None
start = ev[-1][0]
for t, _ in reversed(ev[:-1]):
if start - t >= HUMAN_QUIET:
break
start = t
return start
def info(self, tid, now=None):
now = now or time.time()
ago = self.seconds_ago(tid, now)
return {"active": self.active(tid, now),
"human_active_since": round(self.since(tid), 3) if self.active(tid, now) else None,
"last_input_seconds_ago": None if ago is None else round(ago, 1),
"inputs": len(self.events.get(tid, ())),
"watched": tid in self.tasks}
def guard(self, tid, action):
"""Raise HumanBusy if the human is using `tid`. `action` reads "reloaded", "closed"..."""
if tid and self.active(tid):
raise HumanBusy(tid, self.seconds_ago(tid), action)
# ---- the listener --------------------------------------------------------
def watch(self, tid):
"""Put the listener in tab `tid` and keep it there across navigations.
A script added with Page.addScriptToEvaluateOnNewDocument lives only as
long as the session that added it, so this holds one quiet page socket
open per watched tab, for as long as the tab exists."""
if tid and tid not in self.tasks:
self.tasks[tid] = asyncio.ensure_future(self._watch(tid))
def forget(self, tid):
task = self.tasks.pop(tid, None)
if task is not None:
task.cancel()
self.events.pop(tid, None)
self.agent.pop(tid, None)
async def _watch(self, tid):
me = asyncio.current_task()
try:
async with ws_connect(f"ws://127.0.0.1:{UPSTREAM}/devtools/page/{tid}", max_size=None) as ws:
setup = [("Runtime.addBinding", {"name": HUMAN_BINDING}),
("Runtime.enable", {}),
("Page.enable", {}),
("Page.addScriptToEvaluateOnNewDocument", {"source": HUMAN_JS}),
("Runtime.evaluate", {"expression": HUMAN_JS})]
for i, (method, params) in enumerate(setup, 1):
await ws.send(json.dumps({"id": i, "method": method, "params": params}))
async for raw in ws:
m = json.loads(raw)
p = m.get("params") or {}
if m.get("method") == "Runtime.bindingCalled" and p.get("name") == HUMAN_BINDING:
try:
t, kind = json.loads(p.get("payload") or "[]")
if self.note(tid, float(t), str(kind)[:20]):
log(f"human input in tab {tid[:12]} ({kind})")
except (TypeError, ValueError):
pass
except asyncio.CancelledError:
raise
except Exception:
pass # the tab closed, or the browser went away: nothing left to watch
finally:
if self.tasks.get(tid) is me:
self.tasks.pop(tid, None)
HUMANS = Humans()
async def agent_gate(tid, method, params):
"""The refusal text when an agent may not send `method` to tab `tid` right now, else None.
Rule 3: while the human is working in a tab, agents keep their hands off it.
Rule 4: a tab that is on screen for a job is never reloaded or reopened mid-job.
Called for every command an agent sends through the proxy to a tab."""
if not method:
return None
if method.startswith("Input."):
if HUMANS.active(tid):
return str(HumanBusy(tid, HUMANS.seconds_ago(tid), "typed or clicked in"))
HUMANS.agent_input(tid)
return None
if method in ("Page.reload", "Page.navigate", "Target.closeTarget"):
verb = {"Page.reload": "reloaded", "Page.navigate": "navigated", "Target.closeTarget": "closed"}[method]
if HUMANS.active(tid):
return str(HumanBusy(tid, HUMANS.seconds_ago(tid), verb))
owner = STAGE.job_of(tid) if STAGE is not None else None
if owner is None:
return None
if method == "Page.navigate":
try:
cur = (await UP.send("Target.getTargetInfo", targetId=tid)).get("targetInfo", {}).get("url", "")
except Exception:
cur = ""
if page_key(cur) != page_key((params or {}).get("url")):
return None # going somewhere new is work, not a reload
return (f"this tab is on screen for the job \"{owner}\" and that job is not done, so it was not "
f"{verb}. Read what it shows now and continue from there. If you are stuck, stop and ask "
f"the human in one line. The job's tab closes with bb-close {owner}.")
return None
# ---------------------------------------------------------------------------
# the stage: the human's screen, held by one agent at a time
# ---------------------------------------------------------------------------
class StageBusy(Exception):
def __init__(self, holder, seconds_left, url):
self.holder, self.seconds_left, self.url = holder, seconds_left, url
super().__init__(f"stage held by {holder} for another {seconds_left}s")
class StopAndAsk(Exception):
"""Rule 5: an agent that keeps opening and closing the same page is stuck. Refuse it."""
def __init__(self, owner, url, times, minutes):
self.owner, self.url = owner, url
super().__init__(
f"stop: \"{owner}\" has opened and closed {url} {times} times in the last {minutes} minutes. "
f"It was not opened again. Do not retry. Ask the human in one line what is wrong "
f"(bb-peek {owner} shows what the page says right now).")
PEEK_JS = r"""(() => {
const n = %d;
const clean = (s) => (s || '').replace(/\s+/g, ' ').trim();
const label = (el) => clean(el.getAttribute('aria-label')
|| (el.labels && el.labels.length ? el.labels[0].innerText : '')
|| el.placeholder || el.name || el.id || el.type).slice(0, 60);
const secret = (el, name) => el.type === 'password'
|| /^cc-|one-time-code|password/.test(el.autocomplete || '')
|| /card|cvv|cvc|ssn|social security|security code|passcode|\bpin\b/i.test(name);
const fields = [];
for (const el of document.querySelectorAll('input, textarea, select')) {
const t = (el.type || '').toLowerCase();
if (['hidden', 'submit', 'button', 'reset', 'image', 'file'].includes(t)) continue;
const v = (t === 'checkbox' || t === 'radio') ? (el.checked ? 'checked' : '') : el.value;
if (!v) continue;
const name = label(el);
fields.push({field: name, value: secret(el, name) ? '(hidden)' : String(v).slice(0, 80)});
if (fields.length >= 30) break;
}
const h = window.__bbHuman;
return JSON.stringify({url: location.href, title: document.title,
text: clean(document.body && document.body.innerText).slice(0, n), fields: fields,
human: h && Array.isArray(h.events) ? h.events.slice(-50) : []});
})()"""
class Stage:
"""Everything else in this file exists to keep agents OUT of the human's way.
This is the opposite: the one path that deliberately interrupts him, for the
times an agent genuinely has something to show. Because it is an interruption,
it is the one thing that has to take turns.
Two Claude Code sessions ran `open -a "Brave Browser" <url>` four minutes
apart on 2026-09-20 and threw Matt's window back and forth between them. The
proxy never saw it: `open -a` and AppleScript `activate` are macOS calls, not
CDP, so swallowing Target.activateTarget did nothing. The fix is not to try to
intercept them, which is not possible from here. It is to make a queued path
that is easier to use than the rude one, and to make the rude one loud.
Holding the stage does not block the caller: show() returns as soon as the
page is up, and the hold decays on its own. A second agent that arrives while
the page is fresh waits its turn instead of yanking the window mid-sentence.
Jobs (2026-10-23). The owner name an agent passes is its job. A job owns
exactly one tab, pinned to one site, from its first show until bb-close:
* showing the same site again returns that same tab, as it is now;
* it is never reloaded unless the caller asks, and never while the human
is working in it;
* opening and closing the same page over and over is refused (rule 5).
"""
def __init__(self, up):
self.up = up
self._lock = asyncio.Lock()
self._serial = 0
self.holder = None
self.url = None
self.purpose = None
self.until = 0
self.tabs = {} # owner -> that owner's job tab; nobody closes another's
self.jobs = {} # owner -> {"site", "asked", "keys", "since"}
self.created = set() # job tabs the broker itself opened; only these may be closed
self.churn = {} # (owner, page key) -> times that job was closed or reloaded
def state(self):
left = round(max(0.0, self.until - time.time()), 1)
return {"held_by": self.holder if left else None, "showing": self.url if left else None,
"purpose": self.purpose if left else None, "seconds_left": left,
"tab": (self.tabs.get(self.holder) or "")[:12] or None,
"tabs": {o: t[:12] for o, t in self.tabs.items()},
"jobs": {o: {"site": j["site"], "tab": self.tabs.get(o, "")[:12],
"open_for": round(time.time() - j["since"]),
"human": HUMANS.info(self.tabs.get(o, ""))}
for o, j in self.jobs.items()}}
def job_of(self, tid):
"""The owner whose open job this tab is, or None."""
for o, t in self.tabs.items():
if t == tid:
return o
return None
def job_tab(self, owner, url):
"""The tab of `owner`'s open job when that job is on the same site as `url`."""
job = self.jobs.get(owner)
if job and self.tabs.get(owner) and job["site"] == site_of(url):
return self.tabs[owner]
return None
# ---- rule 5: churn -------------------------------------------------------
def _churn_count(self, owner, key):
now = time.time()
hits = [t for t in self.churn.get((owner, key), ()) if now - t < CHURN_WINDOW]
if hits:
self.churn[(owner, key)] = hits
else:
self.churn.pop((owner, key), None)
return len(hits)
def _refuse_churn(self, owner, url):
n = self._churn_count(owner, page_key(url))
if n >= CHURN_LIMIT:
log(f"stage: refused {owner} reopening {url}: {n} open/close cycles in {CHURN_WINDOW}s")
raise StopAndAsk(owner, page_key(url), n, max(1, CHURN_WINDOW // 60))
def _note_churn(self, owner, keys):
for k in keys:
self.churn.setdefault((owner, k), []).append(time.time())
async def _front(self, tid):
"""Bring the page, and on macOS the app, to the front. Exactly once.
This is the only place in the project allowed to do this, which is the
whole point: one door, with a queue in front of it.
"""
try:
await self.up.send("Target.activateTarget", targetId=tid)
except Exception as e:
log(f"stage: activateTarget failed ({e})")
if sys.platform == "darwin" and STAGE_ACTIVATE_APP:
try:
pr = await asyncio.create_subprocess_exec(
"osascript", "-e", f'tell application "{BROWSER_APP}" to activate',
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
await asyncio.wait_for(pr.wait(), 5)
except Exception as e:
log(f"stage: activate failed ({e})")
async def show(self, owner, url, purpose=None, wait=None, dwell=None, reload=False):
wait = STAGE_WAIT if wait is None else int(wait)
dwell = STAGE_DWELL if dwell is None else int(dwell)
try:
if wait <= 0:
# wait_for(..., 0) times out even on a free lock, because the
# acquire never gets to run. "Don't queue" checks the lock itself.
if self._lock.locked():
raise asyncio.TimeoutError
await self._lock.acquire()
else:
await asyncio.wait_for(self._lock.acquire(), timeout=wait)
except asyncio.TimeoutError:
raise StageBusy(self.holder, round(max(0.0, self.until - time.time()), 1), self.url)
self._serial += 1
mine = self._serial
try:
return await self._show_locked(mine, owner, url, purpose, dwell, reload)
except BaseException:
# A bad URL, Brave being down, a refusal, or the caller giving up must
# not leave the screen locked: every later /show would get 409.
self.free(mine)
raise
async def _live(self):
"""targetId -> targetInfo for every open page."""
try:
infos = (await self.up.send("Target.getTargets")).get("targetInfos", [])
except Exception:
return {}
return {t["targetId"]: t for t in infos if t.get("type") == "page"}
def _tab_showing(self, url, live, owner):
"""A tab to reuse instead of opening a copy, or None.
In order: a job tab already on this page; this owner's own hidden (leased)
tab on this page or this site, so a job never hops between a hidden and a
visible copy; then any tab on this page, the human's own included."""
want, site = page_key(url), site_of(url)
shown = set(self.tabs.values())
leased = [t for t, i in (REG.tabs.items() if REG is not None else ()) if i["owner"] == owner]
same = [t for t, i in live.items() if page_key(i.get("url", "")) == want]
for pool in ([t for t in same if t in shown], [t for t in same if t in leased],
[t for t in leased if t in live and site_of(live[t].get("url", "")) == site], same):
if pool:
return pool[0]
return None
async def _navigate(self, tid, url, reload=False):
"""Move tab `tid` to `url` in place; with reload=True, a fresh copy (cache bypassed)."""
async with ws_connect(f"ws://127.0.0.1:{UPSTREAM}/devtools/page/{tid}", max_size=None) as ws:
await ws.send(json.dumps({"id": 1, "method": "Page.navigate", "params": {"url": url}}))
await asyncio.wait_for(ws.recv(), 10)
if reload:
await ws.send(json.dumps({"id": 2, "method": "Page.reload", "params": {"ignoreCache": True}}))
await asyncio.wait_for(ws.recv(), 10)
async def peek_tab(self, tid, chars=400):
"""What tab `tid` shows right now: final URL, title, the start of its visible
text, the form fields that hold something (secrets masked), and whether
the human is working in it.
Read only. It evaluates one expression; it never navigates, reloads,
focuses or types. Matt's rule, 2026-10-23: look at a page before touching
it, because he may already have moved it forward (filled the form, solved
the captcha, hit send).
"""
chars = max(0, min(int(chars), 5000))
info = {}
try:
info = (await self.up.send("Target.getTargetInfo", targetId=tid)).get("targetInfo", {})
except Exception:
pass
out = {"target": tid[:12], "url": info.get("url", ""), "title": info.get("title", ""),
"text": "", "fields": []}
try:
v = json.loads(await page_eval(tid, PEEK_JS % chars) or "{}")
if isinstance(v, dict):
out.update({k: v[k] for k in ("url", "title", "text") if isinstance(v.get(k), str)})
if isinstance(v.get("fields"), list):
out["fields"] = v["fields"][:30]
HUMANS.merge(tid, v.get("human"))
except Exception as e:
out["note"] = f"could not read the page ({type(e).__name__})"
out["human"] = HUMANS.info(tid)
out["human_active_since"] = out["human"]["human_active_since"]
return out
async def peek(self, match, chars=400):
"""Every open page tab whose job owner, lease owner or target id is `match`,
or whose URL contains it. Includes tabs the human opened himself. Changes nothing."""
if not match:
return []
live = await self._live()
owners = {}
for o, t in self.tabs.items():
owners.setdefault(t, []).append(o)
lessees = {t: i["owner"] for t, i in (REG.tabs.items() if REG is not None else ())}
tids = [t for t, i in live.items()
if match in owners.get(t, []) or match == lessees.get(t) or match in i.get("url", "")
or (len(match) >= 8 and t.startswith(match))]
hits = await asyncio.gather(*(self.peek_tab(tid, chars) for tid in tids))
for tid, p in zip(tids, hits):
p["shown_by"] = owners.get(tid) or None
p["leased_by"] = lessees.get(tid)
p["opened_by_broker"] = tid in self.created or tid in lessees
return list(hits)
async def _show_locked(self, mine, owner, url, purpose, dwell, reload=False):
# History of this method, because each line of it is a lesson:
# 2026-09-16: re-showing a page closes the old tab first, so there is never
# a row of half-stale copies of the same page to pick from.
# 2026-10-02: only ever THIS OWNER's tab. One shared stage tab let each
# agent's show close the page another agent had just put up.
# 2026-10-22: "refresh the page that is already open, never open new tabs
# of the same thing." Reuse any tab already on the page.
# 2026-10-23: "I filled out TruePeopleSearch and hit send ... you didn't
# recognize that on the screen and you're doing it again." Reloading that
# tab wiped his form, his captcha and his success page. Now a job owns one
# tab; showing it again brings it forward AS IS and hands back what it
# shows. It is reloaded only on request, and never while he works in it.
key, site = page_key(url), site_of(url)
live = await self._live()
tid, job = self.tabs.get(owner), self.jobs.get(owner)
if tid and tid not in live: # he closed it himself: that job is over
self.tabs.pop(owner, None)
self.jobs.pop(owner, None)
self.created.discard(tid)
tid = job = None
if tid and job and job["site"] != site:
# The owner moved on to another site, so the old job is done.
await self._end_job(owner, live)
tid = job = None
if tid: # rule 1: this job already has its tab
if reload:
HUMANS.guard(tid, "reloaded")
self._refuse_churn(owner, url)
how = "reloaded"
elif job and key != job["asked"]:
HUMANS.guard(tid, "moved to another page")
how = "moved"
else:
how = "as is"
else:
self._refuse_churn(owner, url)
tid = self._tab_showing(url, live, owner)
if tid and reload:
HUMANS.guard(tid, "reloaded")
how = ("reloaded" if reload else "as is") if tid else "new"
self.holder, self.url, self.purpose = owner, url, purpose
self.until = time.time() + dwell
if how == "new":
res = await self.up.send("Target.createTarget", url=url, background=False)
tid = res["targetId"]
self.created.add(tid)
elif how in ("reloaded", "moved"):
await self._navigate(tid, url, reload=(how == "reloaded"))
if how == "reloaded":
self._note_churn(owner, [key])
if REG is not None and tid in REG.tabs:
self.created.add(tid) # the broker made it, as a hidden lease
for o, t in list(self.tabs.items()):
if t == tid and o != owner: # one job per tab
self.tabs.pop(o, None)
self.jobs.pop(o, None)
if self.tabs.get(owner) != tid or owner not in self.jobs:
self.jobs[owner] = {"site": site, "asked": key, "keys": set(), "since": time.time()}
self.tabs[owner] = tid
self.jobs[owner]["asked"] = key
self.jobs[owner]["keys"].add(key)
HUMANS.watch(tid)
await self._front(tid)
words = {"new": "new tab", "as is": "already open, brought forward as is",
"reloaded": "reloaded in place (asked for)", "moved": "same job tab, moved to this page"}
log(f"stage -> {owner}: {url} ({purpose or 'no purpose given'}), {words[how]}, held {dwell}s")
asyncio.create_task(self._decay(mine, dwell))
r = {"shown": url, "target_id": tid, "held_for": dwell, "owner": owner,
"already_open": how != "new", "reloaded": how == "reloaded", "moved": how == "moved"}
if how == "as is":
r["page"] = await self.peek_tab(tid, 200)
return r
async def _end_job(self, owner, live=None):
"""Finish `owner`'s job: close its tab if the broker opened it. Returns
("closed", info) or ("left", info). A tab the human is working in is never
closed: it is handed to him, and the broker forgets it."""
tid = self.tabs.pop(owner, None)
job = self.jobs.pop(owner, None)
if not tid:
return None
if job:
self._note_churn(owner, job["keys"])
url = (live or {}).get(tid, {}).get("url", "")
if tid not in self.created:
return ("left", {"owner": owner, "target": tid[:12], "why": "the human's own tab"})
self.created.discard(tid)
if REG is not None and REG.tabs.pop(tid, None) is not None:
REG._save()
if HUMANS.active(tid):
HUMANS.forget(tid)
log(f"stage tab {tid[:12]} left open for the human ({owner}): he is working in it")
return ("left", {"owner": owner, "target": tid[:12],
"why": "the human is working in it, so it is his now"})
HUMANS.forget(tid)
try:
await self.up.send("Target.closeTarget", targetId=tid)
except Exception:
pass
log(f"stage tab {tid[:12]} closed ({owner}, job done)")
return ("closed", {"owner": owner, "target": tid[:12], "url": url})
async def unshow(self, match):
"""Finish the job(s) whose owner is `match`, or whose page URL contains it.
Matt's rule, 2026-10-23: "quit leaving tabs open after you do things." An
agent that put a page up closes it when the task is done (bb-close). Only
tabs the broker itself opened are closed. A tab the human already had open,
or is working in right now, is left to him.
"""
if not match:
return {"closed": [], "left_open": []}
live = await self._live()
closed, left = [], []
for o, tid in list(self.tabs.items()):
if o != match and match not in live.get(tid, {}).get("url", ""):
continue
res = await self._end_job(o, live)
if res:
(closed if res[0] == "closed" else left).append(res[1])
if o == self.holder:
self.free(self._serial)
return {"closed": closed, "left_open": left}
async def _decay(self, serial, dwell):
await asyncio.sleep(dwell)
self.free(serial)
def free(self, serial=None):
if serial is not None and serial != self._serial:
return False # someone else's turn by now
if self._lock.locked():
self.holder = None
self.url = None
self.purpose = None
self.until = 0
self._lock.release()
return True
return False
REG: Registry = None
UP: Upstream = None
STAGE: Stage = None
# ---------------------------------------------------------------------------
# the CDP-compatible front door on LISTEN
# ---------------------------------------------------------------------------
def _resp(status, reason, body=b"", ctype="application/json"):
return Response(status, reason, Headers([("Content-Type", ctype),
("Content-Length", str(len(body))),
("Connection", "close")]), body)
async def _pool_entries():
"""What a legacy client sees on GET /json: one unclaimed tab we made, reserved
for this caller for a few seconds so a second caller can't be handed the same
one before the first attaches. Made fresh if none is free."""
free = REG.free_pool()
tid = free[0] if free else await REG.create("pool", "about:blank", hidden=True, kind="pool")
REG.tabs[tid]["listed_until"] = time.time() + POOL_HOLD
e = REG.target_info(tid)
try:
for t in (await UP.send("Target.getTargets")).get("targetInfos", []):
if t["targetId"] == tid:
e["title"], e["url"] = t.get("title", ""), t.get("url", "")
except Exception:
pass
return [e]
async def route_http(method, path):
"""HTTP endpoints of the DevTools remote-debugging surface.
Returns (status line, content type, body bytes)."""
u = urlparse(path)
# Playwright asks for /json/version/ with a trailing slash; Chrome tolerates it.
u = u._replace(path=(u.path.rstrip("/") or "/"))
j = lambda obj: ("200 OK", "application/json", json.dumps(obj, indent=2).encode())
if u.path == "/json/version":
v = dict(UP.version or {})
v["Browser"] = (v.get("Browser", "") + " (browser-broker)").strip()
v["webSocketDebuggerUrl"] = f"ws://127.0.0.1:{LISTEN}/devtools/browser/{uuid.uuid4().hex}"
return j(v)
if u.path in ("/json", "/json/list"):
return j(await _pool_entries())
if u.path == "/json/new": # Chrome 111+ wants PUT; GET still accepted here
url = unquote(u.query) if u.query else "about:blank"
tid = await REG.create("pool", url, hidden=True, kind="pool")
REG.tabs[tid]["listed_until"] = time.time() + POOL_HOLD
return j(REG.target_info(tid))
if u.path.startswith("/json/activate/"):
return ("200 OK", "text/plain", b"Target activated (noop)")
if u.path.startswith("/json/close/"):
tid = u.path.rsplit("/", 1)[-1]
# only unleased pool tabs: a REST lease or a browser client's tab is not
# some other caller's to close just because it knows the id
if REG.tabs.get(tid, {}).get("kind") == "pool":
await REG.close(tid, "client closed it")
return ("200 OK", "text/plain", b"Target is closing")
return ("404 Not Found", "text/plain", b"No such target id")
if u.path == "/json/protocol":
return j(await UP._http("/json/protocol"))
return ("404 Not Found", "text/plain", b"Not Found")
async def process_request(connection, request):
"""The inner websocket server only ever sees upgrades (the front door routes
plain HTTP itself); anything else is refused."""
if request.path.startswith("/devtools/"):
return None
return _resp(404, "Not Found")
INNER_PORT = None # where the websocket server really listens (ephemeral, loopback)
async def _splice(r, w):
try:
while True:
data = await r.read(65536)
if not data:
break
w.write(data)
await w.drain()
except Exception:
pass
finally:
try:
w.close()
except Exception:
pass
async def front_door(reader, writer):
"""Port 9222. The websockets library refuses any HTTP method but GET before
our code runs, so PUT /json/new (what Chrome 111+ clients send) failed. Plain
HTTP is answered here; websocket upgrades are piped byte-for-byte to the
inner websocket server. A bare port probe just gets closed, quietly."""
try:
head = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 10)
except Exception:
writer.close()
return
lines = head.decode("latin-1").split("\r\n")
parts = lines[0].split(" ")
if len(parts) < 2:
writer.close()
return
method, path = parts[0].upper(), parts[1]
headers = {}
for l in lines[1:]:
if ":" in l:
k, v = l.split(":", 1)
headers[k.strip().lower()] = v.strip()
if "websocket" in headers.get("upgrade", "").lower():
try:
ur, uw = await asyncio.open_connection("127.0.0.1", INNER_PORT)
except Exception:
writer.close()
return
uw.write(head)
await asyncio.gather(_splice(reader, uw), _splice(ur, writer))
return
try:
status, ctype, body = await route_http(method, path)
except Exception as e:
status, ctype, body = "500 Internal Server Error", "text/plain", f"{type(e).__name__}: {e}".encode()
try:
writer.write((f"HTTP/1.1 {status}\r\nContent-Type: {ctype}\r\n"
f"Content-Length: {len(body)}\r\nConnection: close\r\n\r\n").encode() + body)
await writer.drain()
except Exception:
pass
finally:
writer.close()
async def _pump(src, dst, transform=None):
"""Copy messages src -> dst, optionally rewriting/dropping via transform."""
try:
async for raw in src:
if transform:
out = await transform(raw)
if out is None:
continue
raw = out
await dst.send(raw)
except websockets.ConnectionClosed:
pass
finally:
try:
await dst.close()
except Exception:
pass
async def handle_ws(connection):
path = connection.request.path
if path.startswith("/devtools/page/"):
await _page_session(connection, path.rsplit("/", 1)[-1])
elif path.startswith("/devtools/browser/"):
await _browser_session(connection)
else:
await connection.close(1008, "unknown endpoint")
async def _page_session(client, tid):
info = REG.tabs.get(tid)
if info is None:
await client.close(1008, "not your tab")
return
if info["kind"] == "pool" and info["attached"] > 0:
await client.close(1008, "tab already in use") # a pool tab is one client's
return
info["attached"] += 1
owner = info["owner"]
try:
up = await ws_connect(f"ws://127.0.0.1:{UPSTREAM}/devtools/page/{tid}", max_size=None)
except Exception as e:
info["attached"] -= 1
await client.close(1011, f"browser unreachable: {e}")
return
async def c2u(raw):
try:
m = json.loads(raw)
except Exception:
return raw
if m.get("method") in FOCUS_STEALERS or m.get("method") in BROWSER_KILLERS:
out = {"id": m.get("id"), "result": {}}
if "sessionId" in m:
out["sessionId"] = m["sessionId"]
await client.send(json.dumps(out))
return None
no = await agent_gate(tid, m.get("method"), m.get("params"))
if no:
out = {"id": m.get("id"), "error": {"code": -32000, "message": "browser-broker: " + no}}
if "sessionId" in m:
out["sessionId"] = m["sessionId"]
await client.send(json.dumps(out))
return None
return raw
try:
await asyncio.gather(_pump(client, up, c2u), _pump(up, client))
finally:
info = REG.tabs.get(tid)
if info is not None:
info["attached"] -= 1
if info["kind"] == "pool" and info["attached"] <= 0:
await REG.close(tid, "page socket closed")
elif info["kind"] == "rest":
pass # REST lease outlives the socket; the reaper handles TTL
log(f"page socket closed for {tid[:12]} ({owner})")
async def _browser_session(client):
"""One browser-level client == one owner. It sees only what it created."""
owner = "ws:" + uuid.uuid4().hex[:8]
try:
up = await ws_connect(UP.version["webSocketDebuggerUrl"], max_size=None)
except Exception as e:
await client.close(1011, f"browser unreachable: {e}")
return
sessions = {} # sessionId -> targetId (flattened attach)
seen = set() # every target this client was ever shown, so it hears when they close
contexts = set() # browser contexts this client created; disposed when it leaves
log(f"browser client {owner} connected")
def mine(tid):
return tid in REG.tabs and REG.tabs[tid]["owner"] == owner
async def settled_mine(tid):
"""A target Chrome just announced may be ours but not yet registered
(createTarget reply still in flight on the control socket). Wait for it
ONLY while this client has a create in flight — otherwise decide now, so
a foreign tab never stalls the event pump."""
if tid in REG.tabs:
return mine(tid)
if not creating["n"]:
return False
for _ in range(40):
await asyncio.sleep(0.025)
if tid in seen:
return True # ours, even if the client has already closed it
if tid in REG.tabs:
return mine(tid)
return False
async def make(url, context=None):
creating["n"] += 1
try:
tid = await REG.create(owner, url or "about:blank", hidden=True, kind="ws", context=context)
finally:
creating["n"] -= 1
seen.add(tid)
return tid
async def reply(m, **body):
out = {"id": m.get("id"), **body}
if "sessionId" in m:
out["sessionId"] = m["sessionId"]
await client.send(json.dumps(out))
async def c2u(raw):
try:
m = json.loads(raw)
except Exception:
return raw
try:
return await _c2u(m, raw)
except websockets.ConnectionClosed:
raise
except Exception as e:
# A slow or failed upstream call answers THIS command with an error.
# It used to escape the pump and end the session, closing every tab
# the client had.
if m.get("id") is not None:
await reply(m, error={"code": -32000, "message": f"browser-broker: {e}"})
return None
async def _c2u(m, raw):
method, mid, p = m.get("method"), m.get("id"), m.get("params") or {}
if method in FOCUS_STEALERS:
await reply(m, result={})
return None
if method in BROWSER_KILLERS:
log(f"{owner} sent {method} -- swallowed, the browser is the human's")
await reply(m, result={})
if method == "Browser.close":
# what the client wanted is to be done; let it go so its close() resolves
asyncio.ensure_future(client.close())
return None
if method == "Target.attachToBrowserTarget":
await reply(m, error={"code": -32000, "message": "browser-broker: browser target is not shared"})
return None
if method == "Target.createBrowserContext":
res = await UP.send(method, **p)
contexts.add(res.get("browserContextId"))
await reply(m, result=res)
return None
if method == "Target.disposeBrowserContext":
if p.get("browserContextId") not in contexts:
await reply(m, error={"code": -32000, "message": "Failed to find context with id"})
return None
contexts.discard(p.get("browserContextId"))
for tid, i in list(REG.tabs.items()):
if i["owner"] == owner and i.get("context") == p.get("browserContextId"):
REG.tabs.pop(tid, None)
REG._save()
await reply(m, result=await UP.send(method, **p))
return None
if method == "Target.setAutoAttach":
auto_attach["on"] = bool(p.get("autoAttach"))
if method == "Target.createTarget":
ctx = p.get("browserContextId")
if ctx and ctx not in contexts:
await reply(m, error={"code": -32000, "message": "Failed to find browser context with id " + str(ctx)})
return None
tid = await make(p.get("url"), ctx)
if ctx:
REG.tabs[tid]["context"] = ctx
if auto_attach["on"]:
# Playwright/Puppeteer expect attachedToTarget for the new page to
# arrive BEFORE the createTarget reply. Chrome guarantees that on
# one socket; we are two sockets, so hold the reply until the
# attach event has gone through.
for _ in range(80):
if tid in sessions.values():
break
await asyncio.sleep(0.025)
await reply(m, result={"targetId": tid})
return None
if method == "Target.getTargets":
infos = (await UP.send("Target.getTargets")).get("targetInfos", [])
infos = [t for t in infos if mine(t["targetId"])]
if not infos:
# nothing yet -> give this client one so "first page" works
tid = await make("about:blank")
infos = [t for t in (await UP.send("Target.getTargets")).get("targetInfos", []) if t["targetId"] == tid]
await reply(m, result={"targetInfos": infos})
return None
if method == "Target.setDiscoverTargets":
# Forwarded, so the client gets Chrome's real targetCreated /
# targetInfoChanged / targetDestroyed stream; u2c drops everyone else's.
discover["on"] = bool(p.get("discover"))
return raw
if method in ("Target.attachToTarget", "Target.closeTarget", "Target.detachFromTarget",
"Browser.getWindowForTarget"):
tid = p.get("targetId")
if method == "Target.detachFromTarget":
tid = sessions.get(p.get("sessionId"), tid)
if tid and not mine(tid):
await reply(m, error={"code": -32000, "message": "No target with given id found"})
return None
if method == "Target.closeTarget":
no = await agent_gate(tid, method, p)
if no:
await reply(m, error={"code": -32000, "message": "browser-broker: " + no})
return None
await REG.close(tid, "client closed it")
await reply(m, result={"success": True})
return None
if method == "Target.setAutoAttach" and p.get("flatten") is not None:
pass # allowed; auto-attach events are filtered below
if m.get("sessionId") in sessions:
no = await agent_gate(sessions[m["sessionId"]], method, p)
if no:
await reply(m, error={"code": -32000, "message": "browser-broker: " + no})
return None
return raw
async def u2c(raw):
try:
m = json.loads(raw)
except Exception:
return raw
method, p, sid = m.get("method"), m.get("params") or {}, m.get("sessionId")
if method == "Target.attachedToTarget":
child, tid = p.get("sessionId"), (p.get("targetInfo") or {}).get("targetId")
# nested under one of our sessions (OOPIF, worker) or one of our pages
if sid in sessions or await settled_mine(tid):
sessions[child] = tid
seen.add(tid)
return raw
try: # someone else's tab auto-attached to us: let go quietly
await up.send(json.dumps({"id": 0, "method": "Target.detachFromTarget",
"params": {"sessionId": child}}))
except Exception:
pass
return None
if method == "Target.detachedFromTarget":
known = p.get("sessionId") in sessions
sessions.pop(p.get("sessionId"), None)
return raw if known else None
if sid is not None:
return raw if sid in sessions else None # session traffic: ours or nothing
if method in ("Target.targetCreated", "Target.targetInfoChanged", "Target.targetDestroyed"):
tid = p.get("targetId") or (p.get("targetInfo") or {}).get("targetId")
# REG forgets a tab the moment it is closed, so its targetDestroyed
# would be dropped without `seen` and the client never hears it went.
if tid in seen or tid in sessions.values() or await settled_mine(tid):
seen.add(tid)
if method == "Target.targetDestroyed":
seen.discard(tid)
return raw
return None
return raw
discover = {"on": False}
auto_attach = {"on": False}
creating = {"n": 0}
try:
await asyncio.gather(_pump(client, up, c2u), _pump(up, client, u2c))
finally:
await REG.release_owner(owner)
for ctx in contexts:
try:
await UP.send("Target.disposeBrowserContext", browserContextId=ctx)
except Exception:
pass
log(f"browser client {owner} disconnected")
# ---------------------------------------------------------------------------
# REST lease API on API port (kept compatible with broker_client.py)
# ---------------------------------------------------------------------------
class ApiHandler(BaseHTTPRequestHandler):
loop = None
def log_message(self, *a):
pass
def _reply(self, code, obj):
body = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _run(self, coro, timeout=30):
fut = asyncio.run_coroutine_threadsafe(coro, self.loop)
try:
return fut.result(timeout)
except concurrent.futures.TimeoutError:
# Without this the coroutine kept running: a /show whose caller had
# given up still put the page up later, on top of bb-show's fallback.
fut.cancel()
raise
def do_GET(self):
if self.path.startswith("/status"):
s = REG.status()
s["brave_alive"] = self._run(UP.alive())
s["leases"] = [{"lease_id": t["lease_id"], "owner": t["owner"], "purpose": None,
"target": t["target"], "expires_in": t["expires_in"]}
for t in s["tabs"] if t["kind"] == "rest"]
return self._reply(200, s)
if self.path.startswith("/health"):
return self._reply(200, {"ok": True, "brave": self._run(UP.alive())})
if self.path.startswith("/stage"):
return self._reply(200, STAGE.state())
if self.path.startswith("/peek"):
# Read only: what matching tabs show right now, the human's own included.
from urllib.parse import urlsplit, parse_qs
q = parse_qs(urlsplit(self.path).query)
match = (q.get("match") or [""])[0]
if not match:
return self._reply(400, {"error": "peek needs ?match=<owner|url-substring>"})
try:
chars = int((q.get("chars") or ["400"])[0])
except ValueError:
chars = 400
try:
return self._reply(200, {"match": match, "tabs": self._run(STAGE.peek(match, chars))})
except concurrent.futures.TimeoutError:
return self._reply(504, {"error": "browser did not answer in time"})
self._reply(404, {"error": "no such endpoint"})
def do_POST(self):
try:
n = int(self.headers.get("Content-Length") or 0)
b = json.loads(self.rfile.read(n) or b"{}")
if self.path.startswith("/lease"):
ttl = int(b.get("ttl", DEFAULT_TTL))
owner = b.get("owner", "anonymous")
# Rule 1: a job that is already on screen for this owner and site
# is leased as it is. A hidden second copy is what agents used to
# fill in while the human was looking at (and typing in) the first.
shared = STAGE.job_tab(owner, b["url"]) if b.get("url") else None
if shared and shared not in REG.tabs:
tid = self._run(REG.adopt(shared, owner, ttl))
else:
shared = None
tid = self._run(REG.create(owner, b.get("url"), hidden=bool(b.get("hidden", True)),
kind="rest", ttl=ttl))
i = REG.tabs[tid]
i["attached"] = 1 # a REST lease counts as held from grant
i["purpose"] = b.get("purpose")
return self._reply(200, {"lease_id": i["lease_id"], "target_id": tid,
"ws_url": f"ws://127.0.0.1:{LISTEN}/devtools/page/{tid}",
"expires_in": ttl, "shared_with_shown_job": bool(shared)})
if self.path.startswith("/renew"):
# Pool and browser-client tabs carry lease_id None: a body with no
# lease_id must not match them.
if not b.get("lease_id"):
return self._reply(400, {"error": "lease_id is required"})
for tid, i in list(REG.tabs.items()):
if i.get("lease_id") == b.get("lease_id"):
i["expires"] = time.time() + int(b.get("ttl", DEFAULT_TTL))
return self._reply(200, {"lease_id": b["lease_id"], "expires_in": int(b.get("ttl", DEFAULT_TTL))})
return self._reply(404, {"error": "unknown lease"})
if self.path.startswith("/release"):
if not b.get("lease_id"):
return self._reply(400, {"error": "lease_id is required"})
for tid, i in list(REG.tabs.items()):
if i.get("lease_id") == b.get("lease_id"):
closed = self._run(REG.close(tid, "released"))
r = {"released": b["lease_id"], "tab_closed": bool(closed)}
if not closed:
r["left_open"] = ("it is a job tab on screen; bb-close finishes the job"
if STAGE.job_of(tid) else "the human is working in it")
return self._reply(200, r)
return self._reply(404, {"error": "unknown lease"})
if self.path.startswith("/show"):
if not b.get("url"):
return self._reply(400, {"error": "show needs a url"})
try:
wait = STAGE_WAIT if b.get("wait") is None else int(b["wait"])
return self._reply(200, self._run(STAGE.show(
b.get("owner", "anonymous"), b["url"], b.get("purpose"),
wait, b.get("dwell"), bool(b.get("reload"))), timeout=wait + 15))
except StageBusy as e:
return self._reply(409, {"error": str(e), "held_by": e.holder,
"seconds_left": e.seconds_left, "showing": e.url})
except HumanBusy as e:
# Rule 3: hand back what he has done so far, so the agent picks up from there.
page = None
try:
page = self._run(STAGE.peek_tab(e.tid, 400), timeout=10)
except Exception:
pass
return self._reply(423, {"error": str(e), "human_active": True,
"seconds_ago": round(e.seconds_ago, 1), "page": page})
except StopAndAsk as e:
return self._reply(429, {"error": str(e), "stop_and_ask": True})
if self.path.startswith("/unshow"):
match = b.get("match") or b.get("owner") or b.get("url")
if not match:
return self._reply(400, {"error": "unshow needs an owner or a url substring (match)"})
return self._reply(200, self._run(STAGE.unshow(match)))
if self.path.startswith("/unstage"):
return self._reply(200, {"freed": STAGE.free()})
if self.path.startswith("/adopt"):
return self._reply(410, {"error": "adopt is not supported by the proxy — agents only ever get tabs the broker made"})
self._reply(404, {"error": "no such endpoint"})
except concurrent.futures.TimeoutError:
self._reply(504, {"error": "browser did not answer in time"})
except Exception as e:
self._reply(500, {"error": f"{type(e).__name__}: {e}"})
def _trim_stdio():
"""launchd / Task Scheduler point our stdout and stderr at plain files that
nothing rotates. Empty them when they get big."""
for fd in (1, 2):
try:
st = os.fstat(fd)
if stat.S_ISREG(st.st_mode) and st.st_size > STDIO_MAX:
os.ftruncate(fd, 0)
os.lseek(fd, 0, os.SEEK_SET)
except OSError:
pass
async def background():
last_hb = 0
while True:
try:
_trim_stdio()
await REG.reap()
if time.time() - last_hb > HEARTBEAT_EVERY:
s = REG.status()
log(f"heartbeat — {s['count']} broker tabs, browser {'up' if await UP.alive() else 'DOWN'}")
last_hb = time.time()
except Exception as e:
log(f"background error: {type(e).__name__}: {e}")
await asyncio.sleep(5)
def _claim_ports():
"""Take LISTEN and API before anything else happens.
2026-10-07: a second supervisor started a second proxy every 10 seconds.
Each one connected upstream, ran close_leftovers() — which closed every tab
the RUNNING proxy owned — and only then failed to bind and exited. Agents
lost their tabs mid-run every 10 seconds and the cause was invisible, because
the duplicate never lived long enough to be in the process list. A doomed
instance must now die before it can touch a single tab."""
import socket
# 2026-10-23: after `launchctl kickstart -k` the new proxy decided a live proxy
# owned its ports, exited 0, and launchd (restart only on failure) left the
# broker down. Two causes, both handled: TIME_WAIT leftovers (below) and an
# old process still letting go, which gets a few seconds of grace. A genuine
# duplicate still exits without touching anything.
deadline = time.time() + PORT_GRACE
while True:
held, busy = [], None
for port in (LISTEN, API):
s = socket.socket()
# The real servers bind with SO_REUSEADDR, so the probe must too.
# Without it, the old process's connections still in TIME_WAIT
# look exactly like a live proxy for half a minute after a restart.
# An actual listener on the port still makes this bind fail.
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
s.bind(("127.0.0.1", port))
except OSError:
s.close()
busy = port
break
held.append(s)
for h in held: # hand the ports back to the real servers below
h.close()
if busy is None:
return
if time.time() >= deadline:
log(f"another proxy already owns port {busy} — exiting without "
f"touching any tabs")
sys.exit(0)
time.sleep(0.5)
async def main():
global REG, UP, STAGE, INNER_PORT
_claim_ports()
UP = Upstream()
try:
await UP.connect()
except Exception as e:
log(f"no browser answering CDP on {UPSTREAM} ({e}) — launch it with "
f"--remote-debugging-port={UPSTREAM} first (see launch-browser.sh)")
sys.exit(1)
REG = Registry(UP)
STAGE = Stage(UP)
loop = asyncio.get_running_loop()
ApiHandler.loop = loop
api = ThreadingHTTPServer(("127.0.0.1", API), ApiHandler)
threading.Thread(target=api.serve_forever, daemon=True).start()
asyncio.create_task(background())
async with ws_serve(handle_ws, "127.0.0.1", 0, process_request=process_request,
max_size=None, ping_interval=None) as inner:
INNER_PORT = inner.sockets[0].getsockname()[1]
front = await asyncio.start_server(front_door, "127.0.0.1", LISTEN, limit=1 << 20)
async with front:
# Only once we genuinely own the ports is it safe to clean up after
# a previous run. Before the bind, "leftovers" may be another live
# proxy's working tabs.
await REG.close_leftovers()
log(f"proxy up: CDP on {LISTEN} (upstream browser {UPSTREAM}), lease API on {API}")
await asyncio.Future()
if __name__ == "__main__":
asyncio.run(main())