From 88d1423e9958ff6a26f19cd3734af435beb34aba Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:26:43 -0700 Subject: [PATCH] refactor(tools): compact bot-mode DM/relay/probe and browser supervisor modules --- tools/bot_failure_reasons.py | 161 +++------- tools/bot_mode_dm.py | 439 ++++++++-------------------- tools/bot_mode_probe.py | 260 +++++++--------- tools/bot_relay.py | 402 ++++++++----------------- tools/browser_dialog_tool.py | 12 +- tools/browser_extension_router.py | 61 +--- tools/browser_supervisor.py | 412 ++++++++------------------ tools/browser_supervisor_dialogs.py | 212 +++++--------- tools/browser_supervisor_frames.py | 61 ++-- 9 files changed, 629 insertions(+), 1391 deletions(-) diff --git a/tools/bot_failure_reasons.py b/tools/bot_failure_reasons.py index 2fa5975f93..7e9359d4df 100644 --- a/tools/bot_failure_reasons.py +++ b/tools/bot_failure_reasons.py @@ -1,37 +1,25 @@ -"""Typed failure-reason codes for bot turns and relay replies (#93091). +"""Typed failure-reason codes for bot turns and relay replies. A closed vocabulary of machine-readable reason codes carried ALONGSIDE the -existing free-text ``error`` fields (additive schema — old consumers keep -working). Platform-side codes are assigned by the transport/relay layer; -agent-side codes are derived from raw agent/provider error text via -``classify_agent_error``. - -Classifier precedence (deterministic, documented, tested): - 1. auth — an explicit ``authentication_error`` type, a 401/403 status, - or "invalid api key" wins over everything else. Rationale: real - provider 401 bodies (e.g. Anthropic) say "invalid, blocked or out of - funds" — quota words inside an auth error must not misclassify it. - 2. quota — 402 / out of funds / quota / balance. - 3. rate — 429 / rate limit. - 4. server — 5xx / server error / overloaded. - 5. context — context length / context_overflow / maximum context. - 6. config — No LLM provider configured / missing config / No access token. - 7. model — model not found / does not exist. - 8. unknown — anything else (including empty text). +free-text ``error`` fields (additive — old consumers keep working). Platform-side +codes are assigned by the transport/relay layer; agent-side codes are derived +from raw agent/provider error text via ``classify_agent_error``. Classifier +precedence is the order of ``_RULES``: auth outranks quota by design — real +provider 401 bodies (e.g. Anthropic) say "invalid, blocked or out of funds". """ from __future__ import annotations import re -# ── platform-side reason codes ─────────────────────────────────────────────── +# platform-side RUNTIME_OFFLINE = "runtime_offline" QUEUED_EXPIRED = "queued_expired" DELIVERY_TIMEOUT = "delivery_timeout" AGENT_BLOCKED = "agent_blocked" CANCELLED = "cancelled" -# ── agent-side reason codes ────────────────────────────────────────────────── +# agent-side PROVIDER_AUTH_OR_ACCESS = "provider_auth_or_access" PROVIDER_QUOTA_LIMIT = "provider_quota_limit" PROVIDER_RATE_LIMIT = "provider_rate_limit" @@ -41,62 +29,32 @@ MISSING_CONFIG = "missing_config" MODEL_UNAVAILABLE = "model_unavailable" UNKNOWN = "unknown" -ALL_REASONS = frozenset( - { - RUNTIME_OFFLINE, - QUEUED_EXPIRED, - DELIVERY_TIMEOUT, - AGENT_BLOCKED, - CANCELLED, - PROVIDER_AUTH_OR_ACCESS, - PROVIDER_QUOTA_LIMIT, - PROVIDER_RATE_LIMIT, - PROVIDER_SERVER_ERROR, - CONTEXT_OVERFLOW, - MISSING_CONFIG, - MODEL_UNAVAILABLE, - UNKNOWN, - } -) +ALL_REASONS = frozenset({ + RUNTIME_OFFLINE, QUEUED_EXPIRED, DELIVERY_TIMEOUT, AGENT_BLOCKED, CANCELLED, + PROVIDER_AUTH_OR_ACCESS, PROVIDER_QUOTA_LIMIT, PROVIDER_RATE_LIMIT, + PROVIDER_SERVER_ERROR, CONTEXT_OVERFLOW, MISSING_CONFIG, MODEL_UNAVAILABLE, UNKNOWN, +}) #: Reasons a supervisor may retry automatically without human intervention. -AUTO_RETRYABLE = frozenset( - {RUNTIME_OFFLINE, DELIVERY_TIMEOUT, PROVIDER_RATE_LIMIT, PROVIDER_SERVER_ERROR} -) +AUTO_RETRYABLE = frozenset({RUNTIME_OFFLINE, DELIVERY_TIMEOUT, PROVIDER_RATE_LIMIT, PROVIDER_SERVER_ERROR}) def is_auto_retryable(reason: str) -> bool: - """True when ``reason`` is safe to retry automatically.""" return reason in AUTO_RETRYABLE -# ── retry session policy (#93091 item 5) ───────────────────────────────────── -# -# Maintainer ruling (2026-08-23, #93091): a retried bot turn NEVER mints a -# fresh session. Transient classes resume the session as-is. context_overflow -# runs context compression — the one sanctioned context mutation, already in -# the agent core — on the same session and retries against the compacted -# context. Everything else (auth/quota/config/model/unknown) is not -# auto-retried at all: surface the typed reason and stop. - -#: Retry actions returned by :func:`retry_action`. +# Retry session policy: a retried bot turn NEVER mints a fresh session. Transient +# classes resume as-is; context_overflow runs context compression (the one +# sanctioned context mutation) on the same session first; everything else +# (auth/quota/config/model/unknown) is never auto-retried — it can't be fixed by +# a retry and only burns quota. RETRY_RESUME = "resume" RETRY_COMPRESS_THEN_RESUME = "compress_then_resume" RETRY_NONE = "none" def retry_action(reason: str) -> str: - """Map a failure reason to the bot-turn retry action. - - - transient (:data:`AUTO_RETRYABLE`) → ``'resume'``: retry the same - session unchanged, bounded by the caller's backoff ladder. - - :data:`CONTEXT_OVERFLOW` → ``'compress_then_resume'``: run context - compression on the session, then retry the same session. Resending - the identical overflowing context would fail identically, and a - fresh-session escape hatch is explicitly not wanted. - - anything else → ``'none'``: never auto-retry auth/quota/config - failures; a retry cannot fix them and only burns quota. - """ + """Map a failure reason to the bot-turn retry action (see policy above).""" if reason in AUTO_RETRYABLE: return RETRY_RESUME if reason == CONTEXT_OVERFLOW: @@ -104,69 +62,28 @@ def retry_action(reason: str) -> str: return RETRY_NONE -# Ordered (pattern, code) rules — first match wins. See module docstring for -# the precedence rationale (auth beats quota by design). -_RULES: tuple[tuple[re.Pattern[str], str], ...] = ( - ( - re.compile( - r"authentication_error|invalid api key" - r"|(?:error code:?\s*|status(?:\s*code)?:?\s*|http\s*)(?:401|403)\b", - re.IGNORECASE, - ), - PROVIDER_AUTH_OR_ACCESS, - ), - ( - re.compile( - r"(?:error code:?\s*|status(?:\s*code)?:?\s*|http\s*)402\b" - r"|out of funds|quota|balance", - re.IGNORECASE, - ), - PROVIDER_QUOTA_LIMIT, - ), - ( - re.compile( - r"(?:error code:?\s*|status(?:\s*code)?:?\s*|http\s*)429\b|rate.?limit", - re.IGNORECASE, - ), - PROVIDER_RATE_LIMIT, - ), - ( - re.compile( - r"(?:error code:?\s*|status(?:\s*code)?:?\s*|http\s*)5\d{2}\b" - r"|server error|overloaded", - re.IGNORECASE, - ), - PROVIDER_SERVER_ERROR, - ), - ( - re.compile(r"context length|context_overflow|maximum context", re.IGNORECASE), - CONTEXT_OVERFLOW, - ), - ( - re.compile( - r"no llm provider configured|missing config|no access token", - re.IGNORECASE, - ), - MISSING_CONFIG, - ), - ( - re.compile(r"model .*(not found|does not exist)|model_not_found", re.IGNORECASE), - MODEL_UNAVAILABLE, - ), +_STATUS = r"(?:error code:?\s*|status(?:\s*code)?:?\s*|http\s*)" + +# Ordered (pattern, code) — first match wins. +_RULES: tuple[tuple[re.Pattern[str], str], ...] = tuple( + (re.compile(pat, re.IGNORECASE), code) + for pat, code in ( + (rf"authentication_error|invalid api key|{_STATUS}(?:401|403)\b", PROVIDER_AUTH_OR_ACCESS), + (rf"{_STATUS}402\b|out of funds|quota|balance", PROVIDER_QUOTA_LIMIT), + (rf"{_STATUS}429\b|rate.?limit", PROVIDER_RATE_LIMIT), + (rf"{_STATUS}5\d{{2}}\b|server error|overloaded", PROVIDER_SERVER_ERROR), + (r"context length|context_overflow|maximum context", CONTEXT_OVERFLOW), + (r"no llm provider configured|missing config|no access token", MISSING_CONFIG), + (r"model .*(not found|does not exist)|model_not_found", MODEL_UNAVAILABLE), + ) ) def classify_agent_error(text: str) -> str: - """Map raw agent/provider error text to a closed reason code. - - First matching rule in ``_RULES`` wins; anything unmatched (or empty) - is ``unknown``. Auth intentionally outranks quota: a 401 body that also - mentions "out of funds" is still an auth/access failure. - """ + """Map raw agent/provider error text to a closed reason code (``unknown`` when unmatched/empty).""" raw = str(text or "") - if not raw.strip(): - return UNKNOWN - for pattern, code in _RULES: - if pattern.search(raw): - return code + if raw.strip(): + for pattern, code in _RULES: + if pattern.search(raw): + return code return UNKNOWN diff --git a/tools/bot_mode_dm.py b/tools/bot_mode_dm.py index 4b0f91c5ed..e0e3b9f31f 100644 --- a/tools/bot_mode_dm.py +++ b/tools/bot_mode_dm.py @@ -1,29 +1,16 @@ """Bot Mode agent-to-agent DM tool — ``message_agent``. -Lets a Bot Mode agent message a teammate (another profile on this install, an -agent on a registered peer gateway, or one on another Desktop-connected machine) -as a real tool call instead of a hand-assembled ``hermes -p …`` shellout, whose -invocation was fragile (quoting traps, temp-file choreography, dead-profile -races, raw user text forwarded verbatim): the message is a parameter, the target -is validated against the live roster, the attribution prefix is applied -server-side, and the reply arrives later through the background-process -completion notification (fire-and-forget — never blocks the sender's turn). - -Containment contract (all three MUST hold — reviewers check each): -- The schema is injected ONLY into a bot's canonical "Bot Chat" session on a - Bot-Mode-managed install — the same gate as ``tools/bot_mode_probe.py``. It is - not in the registry or any toolset; CLI, ordinary gateway, group-room - (titled "Group: …"), cron and subagent sessions never see it. -- Dispatch re-checks the title gate (defense in depth): a forged call returns a - structured error instead of delivering. -- Everything is additive; the legacy ``hermes -p`` / ``hermes peer dm`` - transports keep working for older prompts. - -Transports (unchanged and proven): local → ``hermes -p chat --in ~ -c -"Bot Chat" --create-if-missing -Q --query-file `` (one turn, reply on -stdout); peer → ``hermes peer dm [/] < ``. Both run via -``terminal_tool(background=True, notify_on_complete=True)`` so the reply wakes -the sender on its NEXT turn — the wake shape every Bot Mode agent already knows. +Lets a Bot Mode agent message a teammate (a profile on this install, an agent on +a registered peer gateway, or one on another Desktop-connected machine): the +target is validated against the live roster, the attribution prefix is applied +server-side, and the reply arrives later via the background-process completion +notification (fire-and-forget). Containment: the schema is injected ONLY into a +bot's canonical "Bot Chat" session on a Bot-Mode-managed install (same gate as +``tools/bot_mode_probe.py``; never in the registry or any toolset), and dispatch +re-checks that gate so a forged call returns a structured error. Transports: +local → ``hermes -p chat --in ~ -c "Bot Chat" --create-if-missing -Q +--query-file ``; peer → ``hermes peer dm [/] < ``; both via +``terminal_tool(background=True, notify_on_complete=True)``. """ from __future__ import annotations @@ -125,14 +112,9 @@ def message_agent_tool_schema() -> dict: def ensure_message_agent_tool(agent: Any) -> bool: - """Inject the ``message_agent`` schema into a Bot Chat agent's tool list. - - Called once per turn. Idempotent and deterministic for the life of a session - (the gate — canonical Bot Chat title on a Bot-Mode-managed install — is - stable from the first turn), so the tool list is byte-identical across turns - — prompt-cache safe. Every non-Bot-Chat session fails the gate on every turn - and never sees the schema. Never raises. - """ + """Inject the ``message_agent`` schema into a Bot Chat agent's tool list (once per turn). + Idempotent and deterministic for the session's life (the gate is stable from the + first turn), so the tool list is byte-identical across turns — prompt-cache safe. Never raises.""" try: if not getattr(agent, "_bot_mode_protocol", True): return False @@ -144,11 +126,9 @@ def ensure_message_agent_tool(agent: Any) -> bool: return True from tools.bot_mode_probe import BOT_CHAT_TITLE, is_bot_mode_managed - if _session_title(agent) != BOT_CHAT_TITLE: - return False # Managed-install check, NOT section non-emptiness: a SOUL.md carrying the # legacy protocol text gets an empty section but must still get the tool. - if not is_bot_mode_managed(_agent_home(agent)): + if _session_title(agent) != BOT_CHAT_TITLE or not is_bot_mode_managed(_agent_home(agent)): return False if agent.tools is None: agent.tools = [] @@ -162,30 +142,12 @@ def ensure_message_agent_tool(agent: Any) -> bool: return False -# ── roster resolution ──────────────────────────────────────────────────────── - - -def _local_roster(root: Path) -> list[str]: - """Profile names on this install: default + every named profile.""" - from tools.bot_mode_probe import _roster - - return [name for name, _dir in _roster(root)] - - def _resolve_local_name(target: str, roster: list[str]) -> Optional[str]: """Map a target handle to a profile name ('hermes' → 'default').""" - want = target.strip() - if not want: - return None - if want.lower() == "hermes": + want = target.strip().lower() + if want == "hermes": return "default" if "default" in roster else None - for name in roster: - if name.lower() == want.lower(): - return name - return None - - -# ── the tool ───────────────────────────────────────────────────────────────── + return next((name for name in roster if name.lower() == want), None) if want else None def _err(message: str, *, roster: list[str] | None = None, peers: list[str] | None = None) -> str: @@ -199,64 +161,48 @@ def _err(message: str, *, roster: list[str] | None = None, peers: list[str] | No return json.dumps(payload) -def message_agent_tool( - target: str = "", - message: str = "", - task_id: Optional[str] = None, - agent: Any = None, -) -> str: +def message_agent_tool(target: str = "", message: str = "", task_id: Optional[str] = None, agent: Any = None) -> str: """Deliver ``message`` to ``target``'s Bot Chat. Returns a JSON ack/error. - - ``agent`` is the calling AIAgent (threaded by the executor) — used for the - Bot Chat gate, the sender identity, and the session key. - """ - # Defense-in-depth gate: only a canonical Bot Chat may deliver. + ``agent`` is the calling AIAgent — used for the Bot Chat gate and sender identity.""" home = _agent_home(agent) try: from tools.bot_mode_probe import ( - BOT_CHAT_TITLE, - _handle, - _hermes_root, - _peers, - _profile_name as _self_profile_name, + BOT_CHAT_TITLE, _handle, _hermes_root, _peers, _profile_name as _self_profile_name, _roster, is_bot_mode_managed, ) from tools.bot_relay import BOT_CHAT_TURN_ARGS if _session_title(agent) != BOT_CHAT_TITLE: - return _err( - "message_agent is only available in a Bot Mode 'Bot Chat' session. " - "This session is not one; do not retry." - ) + return _err("message_agent is only available in a Bot Mode 'Bot Chat' session. " + "This session is not one; do not retry.") if not is_bot_mode_managed(home): - return _err( - "This install is not Bot-Mode-managed (no bot roster); " - "message_agent is unavailable. Do not retry." - ) + return _err("This install is not Bot-Mode-managed (no bot roster); " + "message_agent is unavailable. Do not retry.") except Exception as exc: # pragma: no cover — defensive return _err(f"Bot Mode gate check failed: {exc}") root = _hermes_root(Path(home)) me = _self_profile_name(Path(home)) - roster = _local_roster(root) + roster = [name for name, _dir in _roster(root)] peers = _peers(root) teammates = [_handle(n) for n in roster if n != me] + def _roster_err(msg: str) -> str: + return _err(msg, roster=teammates, peers=peers) + body = str(message or "").strip() if not body: return _err("message is required — compose what you want to say to that agent.") if len(body) > MESSAGE_MAX_CHARS: - return _err( - f"message too long ({len(body)} chars > {MESSAGE_MAX_CHARS}). " - "Send the essentials; share large content as a file path instead." - ) + return _err(f"message too long ({len(body)} chars > {MESSAGE_MAX_CHARS}). " + "Send the essentials; share large content as a file path instead.") raw_target = str(target or "").strip().lstrip("@") if not raw_target: - return _err("target is required.", roster=teammates, peers=peers) + return _roster_err("target is required.") sender_handle = _handle(me) - prefix = f"Message from 🤖 {sender_handle} (@{sender_handle}): " + content = f"Message from 🤖 {sender_handle} (@{sender_handle}): " + body # Peer target: '/' or a bare registered peer name. peer_match = _PEER_TARGET_RE.match(raw_target) @@ -265,118 +211,65 @@ def message_agent_tool( peer_name = peer_match.group(1) if peer_match else bare_peer peer_profile = peer_match.group(2) if peer_match else None if peer_name not in peers: - return _err( - f"No registered peer named '{peer_name}'.", roster=teammates, peers=peers - ) + return _roster_err(f"No registered peer named '{peer_name}'.") dm_target = f"{peer_name}/{peer_profile}" if peer_profile else peer_name - label = f"@{peer_profile or peer_name} on peer '{peer_name}'" # Pin the registry-owning profile: `hermes peer` resolves bot_peers via # the profile-scoped load_config(), while the roster above reads the # machine-root config — the CLI must run in that same profile or a # secondary-profile bot sees an empty registry ("No peer named"). - return _start_delivery( - ["hermes", "-p", _self_profile_name(root), "peer", "dm", dm_target], - prefix + body, - label, - stdin_file=True, - task_id=task_id, - agent=agent, - ) + return _start_delivery(["hermes", "-p", _self_profile_name(root), "peer", "dm", dm_target], content, + f"@{peer_profile or peer_name} on peer '{peer_name}'", + stdin_file=True, task_id=task_id, agent=agent) # Local teammate. - if not _LOCAL_TARGET_RE.match(raw_target) and "@" not in raw_target: - return _err(f"Invalid target: {raw_target!r}.", roster=teammates, peers=peers) - resolved = _resolve_local_name(raw_target, roster) if _LOCAL_TARGET_RE.match(raw_target) else None + is_local_shape = bool(_LOCAL_TARGET_RE.match(raw_target)) + if not is_local_shape and "@" not in raw_target: + return _roster_err(f"Invalid target: {raw_target!r}.") + resolved = _resolve_local_name(raw_target, roster) if is_local_shape else None if resolved is None or resolved == me: # Unknown locally, or same-name target on ANOTHER connection (this # gateway's 'default' messaging the cloud 'default'): every gateway # connected to the user's Desktop is reachable via the relay roster, so # try that before reporting a resolution failure / self-message. - relayed = _try_relay_delivery( - root, raw_target, prefix + body, me, task_id=task_id, agent=agent - ) + relayed = _try_relay_delivery(root, raw_target, content, me, task_id=task_id, agent=agent) if relayed is not None: return relayed if resolved == me: return _err("You can't message yourself. Pick a teammate from the roster.") - return _err( - f"No teammate named '{raw_target}' on this install, on a connected " - "machine, or on a registered peer. Pick a name from the roster " - "(roles are listed in your system prompt).", - roster=teammates, - peers=peers, - ) - - return _start_delivery( - ["hermes", "-p", resolved, *BOT_CHAT_TURN_ARGS], - prefix + body, - f"@{_handle(resolved)}", - stdin_file=False, - task_id=task_id, - agent=agent, - ) + return _roster_err(f"No teammate named '{raw_target}' on this install, on a connected " + "machine, or on a registered peer. Pick a name from the roster " + "(roles are listed in your system prompt).") + return _start_delivery(["hermes", "-p", resolved, *BOT_CHAT_TURN_ARGS], content, f"@{_handle(resolved)}", + stdin_file=False, task_id=task_id, agent=agent) -def _try_relay_delivery( - root: Path, - raw_target: str, - content: str, - me: str, - *, - task_id: Optional[str], - agent: Any, -) -> Optional[str]: - """Cross-connection delivery via the Desktop relay; None when the target - doesn't resolve against the relay roster. ``content`` is the attributed - message (prefix already applied). - - The envelope is queued on disk; the Desktop drains it over RPC and delivers - on the target connection's own socket. A background waiter is spawned - immediately so the relayed reply wakes the sender through the standard - completion-notification path — identical UX to a local DM. - """ +def _try_relay_delivery(root: Path, raw_target: str, content: str, me: str, *, + task_id: Optional[str], agent: Any) -> Optional[str]: + """Cross-connection delivery via the Desktop relay; None when the target doesn't + resolve against the relay roster. The envelope is queued on disk for the Desktop + to drain; a background waiter is spawned immediately so the relayed reply wakes + the sender through the standard completion-notification path.""" try: from tools.bot_mode_probe import _handle from tools.bot_relay import ( - EnvelopeRefusedError, - enqueue_envelope, - read_remote_roster, - resolve_remote_target, - waiter_command, + EnvelopeRefusedError, enqueue_envelope, read_remote_roster, resolve_remote_target, waiter_command, ) roster = read_remote_roster(root) - if not roster: - return None - match = resolve_remote_target(raw_target, roster) + match = resolve_remote_target(raw_target, roster) if roster else None if match is None: return None if match == "ambiguous": - forms = ", ".join( - f"{r['handle']}@{r['connection_id']}" - for r in roster - if r["handle"].lower() == raw_target.strip().lstrip("@").lower() - ) - return _err( - f"'{raw_target}' exists on several connected machines — " - f"disambiguate with one of: {forms}." - ) + want = raw_target.strip().lstrip("@").lower() + forms = ", ".join(f"{r['handle']}@{r['connection_id']}" for r in roster if r["handle"].lower() == want) + return _err(f"'{raw_target}' exists on several connected machines — disambiguate with one of: {forms}.") try: - envelope = enqueue_envelope( - root, - target=match, - message=content, - sender_profile=me, - sender_handle=_handle(me), - ) + envelope = enqueue_envelope(root, target=match, message=content, sender_profile=me, sender_handle=_handle(me)) except EnvelopeRefusedError as exc: # Fail fast: target definitively offline — nothing was queued. - # Structured 'runtime_offline' refusal, distinct from a resolution error. return json.dumps({"error": str(exc), "reason": exc.reason}) label = f"@{match['handle']} on {match['connection_label'] or match['connection_id']}" - return _spawn_delivery( - waiter_command(root, envelope), label, task_id=task_id, agent=agent - ) + return _spawn_delivery(waiter_command(root, envelope), label, task_id=task_id, agent=agent) except Exception: logger.debug("relay delivery attempt failed", exc_info=True) return None @@ -388,7 +281,6 @@ def _dm_dir() -> Path: dirname = f"{_DM_DIR_NAME}-{uid}" if uid is not None else _DM_DIR_NAME path = Path(tempfile.gettempdir()) / dirname path.mkdir(mode=0o700, exist_ok=True) - # Shared POSIX temp roots need a per-user directory. Fail closed if an # attacker pre-created the expected path or replaced it with a symlink. info = path.lstat() @@ -401,32 +293,25 @@ def _dm_dir() -> Path: return path -def cleanup_bot_dm_cache( - max_age_hours: float = _DM_STALE_SECONDS / 3600, *, now: float | None = None -) -> int: +def cleanup_bot_dm_cache(max_age_hours: float = _DM_STALE_SECONDS / 3600, *, now: float | None = None) -> int: """Delete orphaned DM payload files older than *max_age_hours*; returns count. - - Same contract as the other ``cleanup_*_cache`` helpers so the gateway - housekeeping loop can prune hourly even on installs that never send another - DM (the in-band sweep in ``_write_dm_file`` only runs when a DM is written). - """ + Same contract as the other ``cleanup_*_cache`` helpers (hourly gateway housekeeping); + legacy temp-root locations from versions predating the dedicated directory are swept too.""" cutoff = (time.time() if now is None else now) - max_age_hours * 3600 - # Legacy temp-root locations too, so upgrades clean files from versions - # predating the dedicated directory. temp_root = Path(tempfile.gettempdir()) - locations: list[tuple[Path, str]] = [ - (temp_root, "hermes-dm-*.txt"), - (temp_root, "hermes-relay-dm-*.txt"), - ] - try: + locations = [(temp_root, "hermes-dm-*.txt"), (temp_root, "hermes-relay-dm-*.txt")] + with contextlib.suppress(OSError): locations.append((_dm_dir(), "*.txt")) - except OSError: - pass from tools.bot_relay import unlink_files_older_than return sum(unlink_files_older_than(d, pattern, cutoff) for d, pattern in locations) +def _unlink_dm_file(path: str) -> None: + with contextlib.suppress(OSError): + os.unlink(path) + + def _write_dm_file(content: str) -> str: """The message rides a temp file — never inline shell text.""" cleanup_bot_dm_cache() @@ -437,41 +322,22 @@ def _write_dm_file(content: str) -> str: except BaseException: # fdopen owns the descriptor once it succeeds, but if fdopen itself # failed the raw descriptor is still ours. Closing twice is harmless. - try: + with contextlib.suppress(OSError): os.close(fd) - except OSError: - pass _unlink_dm_file(path) raise return path -def _unlink_dm_file(path: str) -> None: - try: - os.unlink(path) - except OSError: - pass - - def _delivery_lock(argv: list[str], *, stdin_file: bool): - """Per-profile turn lock context for a LOCAL teammate delivery. - - Local deliveries (``hermes -p chat …``) collide with relay - deliveries into the same profile — both run a Bot Chat turn on this install — - so the turn window is serialized on the cross-process lock in - ``tools.bot_relay``. Peer transports (stdin mode) are locked on the remote - gateway by its own deliver path. - """ + """Per-profile turn lock for a LOCAL teammate delivery: local and relay deliveries + into one profile both run a Bot Chat turn here, so the turn window is serialized on + ``tools.bot_relay``'s cross-process lock. Peer transports (stdin mode) are locked + on the remote gateway by its own deliver path.""" # Match the CLI element by basename: argv[0] may be an absolute venv path - # (service contexts lack PATH) and carries .exe on Windows; split on both - # separators so the shape matches whichever platform built the argv. + # (service contexts lack PATH) and carries .exe on Windows; split on both separators. cli = (argv[0] if argv else "").rsplit("\\", 1)[-1].rsplit("/", 1)[-1] - if ( - stdin_file - or len(argv) < 3 - or cli not in ("hermes", "hermes.exe") - or argv[1] != "-p" - ): + if stdin_file or len(argv) < 3 or cli not in ("hermes", "hermes.exe") or argv[1] != "-p": return contextlib.nullcontext() from tools.bot_mode_probe import _hermes_root from tools.bot_relay import acquire_turn_lock @@ -480,17 +346,14 @@ def _delivery_lock(argv: list[str], *, stdin_file: bool): def _run_local_turn(argv: list[str], dm_file: str) -> int: - """One Bot Chat turn via ``--query-file`` (plus the policy-gated retry); - re-emits the transport's streams and returns its exit code.""" + """One Bot Chat turn via ``--query-file`` (plus one policy-gated retry); re-emits + the transport's streams and returns its exit code. Transient failures re-run the + same session; a context_overflow re-run lets the retried turn's pre-API compaction + compact the transcript first (no fresh session is ever minted). Auth/quota/config never retry.""" def _turn(): - return subprocess.run( - [*argv, "--query-file", dm_file], - check=False, - stdin=subprocess.DEVNULL, - capture_output=True, - text=True, - ) + return subprocess.run([*argv, "--query-file", dm_file], check=False, stdin=subprocess.DEVNULL, + capture_output=True, text=True) proc = _turn() if proc.returncode != 0: @@ -500,9 +363,8 @@ def _run_local_turn(argv: list[str], dm_file: str) -> int: if retry_action(classify_agent_error(detail)) != RETRY_NONE: proc = _turn() if proc.returncode != 0 and "already has a live owner" in (proc.stderr or ""): - # The target's Bot Chat is held live by another surface (Desktop). - # The turn never ran — tell the sender plainly instead of leaking - # a raw lease error + exit code. + # The target's Bot Chat is held live by another surface (Desktop); the turn + # never ran — tell the sender plainly instead of leaking a raw lease error. who = argv[argv.index("-p") + 1] if "-p" in argv[:-1] else "the teammate" print(json.dumps({ "error": f"Delivery failed: @{who}'s Bot Chat is open on another " @@ -512,29 +374,17 @@ def _run_local_turn(argv: list[str], dm_file: str) -> int: return 1 # Re-emit the transport's streams: stdout is the reply text the # completion notification carries back to the sending agent. - if proc.stdout: - sys.stdout.write(proc.stdout) - sys.stdout.flush() - if proc.stderr: - sys.stderr.write(proc.stderr) - sys.stderr.flush() + for stream, text in ((sys.stdout, proc.stdout), (sys.stderr, proc.stderr)): + if text: + stream.write(text) + stream.flush() return proc.returncode def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: - """Run one DM transport and remove its plaintext file after consumption. - - The turn execution window (not the enqueue) holds the target profile's - cross-process lock, so two deliveries into one profile queue instead of - racing; a bounded wait ends in a structured 'target_busy' refusal. - - Local (query-file) turns get one policy-gated retry: transient failures - re-run the same session, and a context_overflow re-run lets the retried - turn's pre-API compaction compact the Bot Chat transcript first — the - sanctioned compression lever; no fresh session is ever minted. - Auth/quota/config failures never retry. Peer transports retry on their own - gateway's deliver path, not here. - """ + """Run one DM transport and remove its plaintext file after consumption. The turn + window (not the enqueue) holds the target profile's cross-process lock, so two + deliveries into one profile queue; a bounded wait ends in a 'target_busy' refusal.""" try: with _delivery_lock(argv, stdin_file=stdin_file): if not stdin_file: @@ -549,14 +399,8 @@ def _run_delivery(argv: list[str], dm_file: str, *, stdin_file: bool) -> int: def _delivery_command(argv: list[str], dm_file: str, *, stdin_file: bool) -> str: """Build an argv-safe command for the cleanup-owning background runner.""" - runner_argv = [ - sys.executable, - str(Path(__file__).resolve()), - "--run-delivery", - "stdin" if stdin_file else "query-file", - dm_file, - *argv, - ] + runner_argv = [sys.executable, str(Path(__file__).resolve()), "--run-delivery", + "stdin" if stdin_file else "query-file", dm_file, *argv] if sys.platform == "win32": # The tracked local backend uses Git Bash on native Windows: forward # slashes keep native drive paths executable there; backslash paths are @@ -565,15 +409,8 @@ def _delivery_command(argv: list[str], dm_file: str, *, stdin_file: bool) -> str return shlex.join(runner_argv) -def _start_delivery( - argv: list[str], - content: str, - label: str, - *, - stdin_file: bool, - task_id: Optional[str], - agent: Any, -) -> str: +def _start_delivery(argv: list[str], content: str, label: str, *, stdin_file: bool, + task_id: Optional[str], agent: Any) -> str: """Create a DM file and transfer its cleanup ownership to the runner.""" dm_file = _write_dm_file(content) try: @@ -584,31 +421,17 @@ def _start_delivery( return _spawn_delivery(command, label, dm_file=dm_file, task_id=task_id, agent=agent) -def _spawn_delivery( - command: str, - label: str, - *, - dm_file: Optional[str] = None, - task_id: Optional[str], - agent: Any, -) -> str: - """Launch the cleanup-owning runner and transfer file ownership on ack. - - ``dm_file`` is None for relay deliveries: the waiter watches a reply file and - the envelope artifacts are owned/swept by ``tools/bot_relay.py``. - """ +def _spawn_delivery(command: str, label: str, *, dm_file: Optional[str] = None, + task_id: Optional[str], agent: Any) -> str: + """Launch the cleanup-owning runner and transfer file ownership on ack. ``dm_file`` + is None for relay deliveries (the waiter watches a reply file; envelope artifacts + are owned/swept by ``tools/bot_relay.py``).""" transferred = False try: from tools.terminal_tool import terminal_tool - raw = terminal_tool( - command, - background=True, - notify_on_complete=True, - task_id=task_id, - workdir=str(Path(__file__).resolve().parent.parent), - _host_local=True, - ) + raw = terminal_tool(command, background=True, notify_on_complete=True, task_id=task_id, + workdir=str(Path(__file__).resolve().parent.parent), _host_local=True) try: parsed = json.loads(raw) except (ValueError, TypeError): @@ -621,20 +444,15 @@ def _spawn_delivery( # From here the background runner owns the file and removes it only # after the local query-file or peer stdin consumer has finished. transferred = True - return json.dumps( - { - "status": "sent", - "to": label, - "detail": ( - f"Message dispatched to {label}. This is asynchronous — do NOT wait " - "or poll. Finish your turn now; when the delivery completes, its " - "notification carries the reply — relay it then, attributed to " - "that agent." - ), - **({"process_id": proc_id} if proc_id else {}), - "sent_at": int(time.time()), - } - ) + return json.dumps({ + "status": "sent", + "to": label, + "detail": (f"Message dispatched to {label}. This is asynchronous — do NOT wait " + "or poll. Finish your turn now; when the delivery completes, its " + "notification carries the reply — relay it then, attributed to that agent."), + "process_id": proc_id, + "sent_at": int(time.time()), + }) except Exception as exc: logger.error("message_agent delivery spawn failed: %s", exc, exc_info=True) return _err(f"Delivery to {label} could not be started: {exc}") @@ -644,39 +462,29 @@ def _spawn_delivery( def _delivery_main(args: list[str]) -> int: - if len(args) < 3 or args[0] != "--run-delivery": + if len(args) < 3 or args[0] != "--run-delivery" or args[1] not in ("stdin", "query-file"): return 2 - stdin_file = args[1] == "stdin" - if not stdin_file and args[1] != "query-file": - return 2 - dm_file = args[2] try: - return _run_delivery(args[3:], dm_file, stdin_file=stdin_file) + return _run_delivery(args[3:], args[2], stdin_file=args[1] == "stdin") except Exception as exc: - # 'target_busy': the queued delivery gave up after its bounded wait — - # surface the structured payload on stdout so the completion - # notification carries it back to the sending agent. + # 'target_busy': the queued delivery gave up after its bounded wait — surface the + # structured payload on stdout so the completion notification carries it back. if getattr(exc, "reason", "") == "target_busy": print(json.dumps({"error": str(exc), "reason": "target_busy"})) - return 1 - print( - f"message_agent delivery failed: {type(exc).__name__}: {exc}", - file=sys.stderr, - ) + else: + print(f"message_agent delivery failed: {type(exc).__name__}: {exc}", file=sys.stderr) return 1 -# ── agent-context helpers (mirror system_prompt.py's resolution) ───────────── +# agent-context helpers (mirror system_prompt.py's resolution) def _agent_home(agent: Any) -> str: """The calling agent's OWN home (session-db derived), not ambient env.""" - try: + with contextlib.suppress(Exception): db_path = getattr(getattr(agent, "_session_db", None), "db_path", None) if db_path: return str(Path(db_path).parent) - except Exception: - pass return _default_home() @@ -684,13 +492,10 @@ def _session_title(agent: Any) -> str: title = str(getattr(agent, "_session_title_hint", "") or "").strip() if title: return title - try: - sdb = getattr(agent, "_session_db", None) - sid = getattr(agent, "session_id", None) + with contextlib.suppress(Exception): + sdb, sid = getattr(agent, "_session_db", None), getattr(agent, "session_id", None) if sdb and sid: return str(sdb.get_session_title(sid) or "").strip() - except Exception: - pass return "" diff --git a/tools/bot_mode_probe.py b/tools/bot_mode_probe.py index ca79f9966b..d19cbed2f7 100644 --- a/tools/bot_mode_probe.py +++ b/tools/bot_mode_probe.py @@ -2,20 +2,13 @@ When any profile on this install carries ``ui_meta['hermes-bots']`` in its profile.yaml (Bot-Mode-managed), a bot's canonical "Bot Chat" session — and ONLY -that session (the caller, agent/system_prompt.py, enforces the title gate against -``BOT_CHAT_TITLE``) — gets a "Messaging other agents" section so the bot can -receive teammate DMs, reply with attribution, and hand off @mentions. Regular -sessions never carry it; the desktop's composer middleware owns the @mention -send path there. - -The protocol is injected by the core at prompt-build time instead of the old -plugin-side SOUL.md append. Silent (``""``) when no profile is managed, when the -profile's SOUL.md already carries the heading (legacy plugin-appended text must -never double up), or on any error — a prompt build must never crash. Cached per -(process, home) so compression-triggered rebuilds produce identical bytes. -Toggle: ``agent.bot_mode_protocol`` in config.yaml (default True). - -Also hosts the path/roster helpers shared by ``bot_mode_dm`` and ``bot_relay``. +that session (agent/system_prompt.py enforces the ``BOT_CHAT_TITLE`` gate) — gets +a "Messaging other agents" section. Silent (``""``) when no profile is managed, +when the profile's SOUL.md already carries the heading (legacy plugin-appended +text must never double up), or on any error — a prompt build must never crash. +Cached per (process, home) so compression-triggered rebuilds produce identical +bytes. Toggle: ``agent.bot_mode_protocol`` (default True). Also hosts the +path/roster helpers shared by ``bot_mode_dm`` and ``bot_relay``. """ from __future__ import annotations @@ -46,6 +39,14 @@ def _resolve_home(home: str | os.PathLike | None) -> Path: return Path(str(home) if home else _default_home()) +def _swallow(fn, default): + """``fn()`` or ``default`` on any exception — the probe must never crash a prompt build.""" + try: + return fn() + except Exception: + return default + + def _hermes_root(home: Path) -> Path: """Root ~/.hermes for both the default profile and named profiles.""" return home.parent.parent if home.parent.name == "profiles" else home @@ -62,14 +63,9 @@ def _handle(name: str) -> str: def _roster(root: Path) -> list[tuple[str, Path]]: """(name, dir) for the default profile + every named profile, sorted.""" - entries: list[tuple[str, Path]] = [("default", root)] - try: - profiles = root / "profiles" - if profiles.is_dir(): - entries.extend((c.name, c) for c in sorted(profiles.iterdir()) if c.is_dir()) - except Exception: - pass - return entries + profiles = root / "profiles" + named = _swallow(lambda: [(c.name, c) for c in sorted(profiles.iterdir()) if c.is_dir()] if profiles.is_dir() else [], []) + return [("default", root), *named] def _read_yaml_dict(path: Path, needle: str | None = None) -> dict | None: @@ -78,7 +74,7 @@ def _read_yaml_dict(path: Path, needle: str | None = None) -> dict | None: ``needle``: cheap substring precheck that skips the YAML parse on the dominant (unmanaged) path — the key is absent from most installs. """ - try: + def _load(): if not path.is_file(): return None raw = path.read_text(encoding="utf-8", errors="replace") @@ -88,8 +84,8 @@ def _read_yaml_dict(path: Path, needle: str | None = None) -> dict | None: data = yaml.safe_load(raw) return data if isinstance(data, dict) else None - except Exception: - return None + + return _swallow(_load, None) def _bots_meta(data: dict | None) -> dict | None: @@ -100,96 +96,81 @@ def _bots_meta(data: dict | None) -> dict | None: def _is_bot_managed(profile_dir: Path) -> bool: - """True when profile.yaml carries a ui_meta['hermes-bots'] block.""" return _bots_meta(_read_yaml_dict(profile_dir / "profile.yaml", "hermes-bots")) is not None +def _any_managed(root: Path) -> bool: + return any(_is_bot_managed(d) for _n, d in _roster(root)) + + def is_bot_mode_managed(home: str | os.PathLike | None = None) -> bool: - """True when ANY profile on this install is Bot-Mode-managed. + """True when ANY profile on this install is Bot-Mode-managed. Never raises. The tool-injection gate for ``message_agent`` — deliberately independent of - :func:`get_bot_mode_protocol_section`'s emptiness: a profile whose SOUL.md - carries the legacy protocol gets an empty section but must still get the - tool. Never raises. + the protocol section's emptiness: a SOUL.md carrying the legacy protocol + gets an empty section but must still get the tool. """ - try: - root = _hermes_root(_resolve_home(home)) - return any(_is_bot_managed(d) for _n, d in _roster(root)) - except Exception: - return False + return _swallow(lambda: _any_managed(_hermes_root(_resolve_home(home))), False) def _soul_has_protocol(profile_dir: Path) -> bool: - try: - soul = profile_dir / "SOUL.md" - return soul.is_file() and _PROTOCOL_HEADING in soul.read_text(encoding="utf-8", errors="replace") - except Exception: - return False + soul = profile_dir / "SOUL.md" + return _swallow(lambda: soul.is_file() and _PROTOCOL_HEADING in soul.read_text(encoding="utf-8", errors="replace"), False) + + +def _role_line(*parts: str) -> str: + """'title — description' from the non-empty parts (either may be absent).""" + return " — ".join(p for p in parts if p) + + +def _bullet(handle: str, *parts: str) -> str: + """Roster line: '- `handle`' plus ' — part' for each non-empty part.""" + return _role_line(f"- `{handle}`", *parts) def _profile_role(profile_dir: Path) -> str: - """Teammate role line: Bot Mode title — profile description (either may be - absent). The title is the name the user gave the bot in Bot Mode; the - description is the profile's stated purpose — either tells a teammate WHO - to message for a job. Single-line, ≤160 chars, "" when neither. Never raises.""" - try: - data = _read_yaml_dict(profile_dir / "profile.yaml") - if not data: - return "" - bots = _bots_meta(data) or {} - title = str(bots.get("title") or "").strip() - description = str(data.get("description") or "").strip() - line = " — ".join(p for p in (title, description) if p) + """Teammate role line: Bot Mode title — profile description; tells a teammate + WHO to message for a job. Single-line, ≤160 chars, "" when neither. Never raises.""" + def _role() -> str: + data = _read_yaml_dict(profile_dir / "profile.yaml") or {} + line = _role_line(str((_bots_meta(data) or {}).get("title") or "").strip(), + str(data.get("description") or "").strip()) return " ".join(line.split())[:160] - except Exception: - return "" - -def _roster_lines(root: Path, me: str) -> list[str]: - """One '- `@handle` — role' line per teammate (excluding ``me``).""" - lines = [] - for name, profile_dir in _roster(root): - if name == me: - continue - role = _profile_role(profile_dir) - lines.append(f"- `@{_handle(name)}`" + (f" — {role}" if role else "")) - return lines + return _swallow(_role, "") def _peers(root: Path) -> list[str]: - """Registered peer gateway names (``hermes peer``) from config.yaml. + """Registered peer gateway names (``hermes peer``) from config.yaml, read + directly (no config-loader import; the section is absent on most installs). Never raises.""" + def _names() -> list[str]: + peers = (_read_yaml_dict(root / "config.yaml", "bot_peers") or {}).get("bot_peers") + return sorted(str(n) for n in peers if str(n).strip()) if isinstance(peers, dict) else [] - Read directly (no config-loader import — cheap; the section is absent on - most installs). Never raises. - """ - try: - data = _read_yaml_dict(root / "config.yaml", "bot_peers") - peers = data.get("bot_peers") if data else None - if not isinstance(peers, dict): - return [] - return sorted(str(name) for name in peers if str(name).strip()) - except Exception: - return [] + return _swallow(_names, []) + + +def _remote_roster(root: Path) -> list[dict]: + """Desktop relay roster (``tools/bot_relay.py``); [] on any failure.""" + def _read(): + from tools.bot_relay import read_remote_roster + + return read_remote_roster(root) + + return _swallow(_read, []) def _remote_paragraph(root: Path) -> str: - """Addendum for agents on OTHER connected machines (Desktop relay roster, - ``tools/bot_relay.py``). Rendered only when the relay roster is non-empty.""" - try: - from tools.bot_relay import read_remote_roster, remote_target_forms - - roster = read_remote_roster(root) - except Exception: - return "" + """Addendum for agents on OTHER connected machines; only when the relay roster is non-empty.""" + roster = _remote_roster(root) if not roster: return "" - lines = [] - for row, form in zip(roster, remote_target_forms(roster)): - where = row["connection_label"] or row["connection_id"] - role = " — ".join(p for p in (row["title"], row["description"]) if p) - lines.append( - f"- `@{form}` — on {where}" + (f" — {role}" if role else "") - ) + from tools.bot_relay import remote_target_forms + + lines = [ + _bullet(f"@{form}", f"on {row['connection_label'] or row['connection_id']}", row["title"], row["description"]) + for row, form in zip(roster, remote_target_forms(roster)) + ] return ( "\n\nTeammates on OTHER connected machines (reachable through the " "Desktop relay — message them with message_agent exactly like local " @@ -216,17 +197,14 @@ def _peer_paragraph(root: Path) -> str: def _build_section(home: Path) -> str: root = _hermes_root(home) me = _profile_name(home) - - if not any(_is_bot_managed(d) for _n, d in _roster(root)): + if not _any_managed(root): return "" - # An older plugin build may have appended the protocol to SOUL.md — never double it. - my_dir = home if me == "default" else root / "profiles" / me - if _soul_has_protocol(my_dir): + if _soul_has_protocol(home if me == "default" else root / "profiles" / me): return "" - handle = _handle(me) - roster_block = "\n".join(_roster_lines(root, me)) or "- (no teammates yet)" + roster_lines = [_bullet(f"@{_handle(name)}", _profile_role(d)) for name, d in _roster(root) if name != me] + roster_block = "\n".join(roster_lines) or "- (no teammates yet)" return ( f"{_PROTOCOL_HEADING}\n" @@ -249,7 +227,7 @@ def _build_section(home: Path) -> str: "concisely via message_agent to their handle, and if it is a pure FYI " "with nothing to add, staying silent is fine — never ping-pong " "acknowledgements.\n" - f"You are `@{handle}`. Your teammates (live roster; roles from their " + f"You are `@{_handle(me)}`. Your teammates (live roster; roles from their " "profiles):\n" f"{roster_block}" + _remote_paragraph(root) @@ -267,10 +245,7 @@ def get_bot_mode_protocol_section(home: str | os.PathLike | None = None, *, forc resolved = str(_resolve_home(home)) with _lock: if force_refresh or resolved not in _cached: - try: - _cached[resolved] = _build_section(Path(resolved)) - except Exception: - _cached[resolved] = "" + _cached[resolved] = _swallow(lambda: _build_section(Path(resolved)), "") return _cached[resolved] @@ -280,8 +255,7 @@ def get_bot_mode_protocol_section(home: str | os.PathLike | None = None, *, forc # strand capability changes (skills, toolsets, MCP, SOUL, roster, peers) forever. # The fingerprint hashes exactly that surface; the built Bot Chat prompt embeds # it and agent/conversation_loop.py rebuilds only when the stored epoch differs -# from disk — a loud, user-initiated, once-per-change cache break, never per-turn -# drift (unchanged state hashes identically; stored bytes are reused verbatim). +# from disk — a loud, once-per-change cache break, never per-turn drift. _EPOCH_PREFIX = "Capability epoch: " _EPOCH_RE_TEXT = r"Capability epoch: ([0-9a-f]{12})" @@ -300,6 +274,7 @@ def capability_fingerprint(home: str | os.PathLike | None = None) -> str: import json resolved = _resolve_home(home) + root = _hermes_root(resolved) surface: dict = {} try: # Canonical loader (managed overlay + env expansion + normalization), @@ -320,22 +295,19 @@ def capability_fingerprint(home: str | os.PathLike | None = None) -> str: surface["mcp"] = json.dumps(mcp, sort_keys=True, default=str) if isinstance(mcp, dict) else "" except Exception: pass - try: + + def _soul() -> str: soul = resolved / "SOUL.md" - surface["soul"] = hashlib.sha256(soul.read_bytes()).hexdigest() if soul.is_file() else "" - except Exception: - surface["soul"] = "" - try: + return hashlib.sha256(soul.read_bytes()).hexdigest() if soul.is_file() else "" + + def _skills() -> list[str]: skills_root = resolved / "skills" - names = ( - [str(p.parent.relative_to(skills_root)) for p in skills_root.glob("**/SKILL.md")] - if skills_root.is_dir() - else [] - ) - surface["skills"] = sorted(names) - except Exception: - surface["skills"] = [] - root = _hermes_root(resolved) + if not skills_root.is_dir(): + return [] + return sorted(str(p.parent.relative_to(skills_root)) for p in skills_root.glob("**/SKILL.md")) + + surface["soul"] = _swallow(_soul, "") + surface["skills"] = _swallow(_skills, []) try: roster = _roster(root) surface["roster"] = sorted(n for n, d in roster if _is_bot_managed(d)) @@ -348,25 +320,15 @@ def capability_fingerprint(home: str | os.PathLike | None = None) -> str: # prompt ONCE so existing bots adopt a new protocol section. surface["protocol_version"] = 2 # Peer gateways and the Desktop relay roster are part of the messaging - # surface too: registering a peer or (dis)connecting a machine must show up - # on the next message. - try: - surface["peers"] = _peers(root) - except Exception: - surface["peers"] = [] - try: - from tools.bot_relay import read_remote_roster - - surface["remote_roster"] = sorted( - f"{r['connection_id']}:{r['profile']}:{r['title']}" for r in read_remote_roster(root) - ) - except Exception: - surface["remote_roster"] = [] - try: - blob = json.dumps(surface, sort_keys=True).encode("utf-8") - return hashlib.sha256(blob).hexdigest()[:12] - except Exception: - return "unavailable" + # surface too: registering a peer or (dis)connecting a machine must show up. + surface["peers"] = _peers(root) + surface["remote_roster"] = sorted( + f"{r['connection_id']}:{r['profile']}:{r['title']}" for r in _remote_roster(root) + ) + return _swallow( + lambda: hashlib.sha256(json.dumps(surface, sort_keys=True).encode("utf-8")).hexdigest()[:12], + "unavailable", + ) def epoch_line(home: str | os.PathLike | None = None) -> str: @@ -380,16 +342,11 @@ def stored_prompt_capability_stale(stored_prompt: str, home: str | os.PathLike | "not stale" — a broken probe must not become a rebuild-every-turn cache burner.""" import re - try: - m = re.search(_EPOCH_RE_TEXT, stored_prompt or "") - if not m: - return False - current = capability_fingerprint(home) - if current == "unavailable": - return False - return m.group(1) != current - except Exception: + m = re.search(_EPOCH_RE_TEXT, stored_prompt or "") + if not m: return False + current = _swallow(lambda: capability_fingerprint(home), "unavailable") + return current != "unavailable" and m.group(1) != current def stored_bot_chat_prompt_needs_upgrade(stored_prompt: str, home: str | os.PathLike | None = None) -> bool: @@ -402,13 +359,10 @@ def stored_bot_chat_prompt_needs_upgrade(stored_prompt: str, home: str | os.Path already carries the legacy protocol yields an empty section, and rebuilding would produce another unstamped prompt and loop. Fails closed to "no upgrade". """ - try: - text = stored_prompt or "" - if _EPOCH_PREFIX in text or _PROTOCOL_HEADING in text: - return False - return bool(get_bot_mode_protocol_section(home)) - except Exception: + text = stored_prompt or "" + if _EPOCH_PREFIX in text or _PROTOCOL_HEADING in text: return False + return _swallow(lambda: bool(get_bot_mode_protocol_section(home)), False) def _reset_cache_for_tests() -> None: diff --git a/tools/bot_relay.py b/tools/bot_relay.py index 1306debc02..ff0ba85712 100644 --- a/tools/bot_relay.py +++ b/tools/bot_relay.py @@ -1,24 +1,14 @@ """Bot Mode cross-connection relay — connections ARE the peer set. -Gateway-side half of the relay that lets agents on ANY Desktop-connected -gateway (local, remote URL, SSH, Hermes Cloud, docker) message agents on ANY -other, with ``message_agent`` as the one send path — connections ARE the peer -set. Plain file plumbing under ``/bot_relay/`` — no network; the gateway -never holds another connection's credentials, the Desktop owns every socket and -does all cross-connection I/O: - -- ``roster.json`` — union roster of agents on OTHER connections, pushed by the - Desktop (``bot_relay.roster.sync``); folded into the Bot Chat protocol section - and used to resolve cross-connection targets. -- ``outbox/`` — envelopes queued by ``message_agent``; the Desktop drains them - (``bot_relay.outbox.drain``) and delivers on the target connection. -- ``replies/`` — one JSON per envelope (``bot_relay.reply``); a background - waiter spawned at send time watches it so the reply wakes the sender via the - same completion-notification path local DMs use. - +Gateway-side half of the relay letting agents on ANY Desktop-connected gateway +message agents on ANY other. Plain file plumbing under ``/bot_relay/`` — +no network; the Desktop owns every socket: ``roster.json`` (union roster of +agents on OTHER connections, pushed via ``bot_relay.roster.sync``), ``outbox/`` +(envelopes queued by ``message_agent``, drained via ``bot_relay.outbox.drain``), +``replies/`` (one JSON per envelope via ``bot_relay.reply``; a waiter spawned at +send time watches it so the reply wakes the sender like a local DM). Public helpers never raise, except ``enqueue_envelope`` → ``EnvelopeRefusedError`` -when the target is definitively offline (fail fast instead of queueing a DM -nobody will drain). +when the target is definitively offline (fail fast instead of queueing a DM nobody will drain). """ from __future__ import annotations @@ -48,35 +38,20 @@ CLAIMED_DIR = "claimed" REPLIES_DIR = "replies" LOCKS_DIR = "locks" -# Fallback wait budget for a queued delivery turn when config is unreadable -# (real knob: ``bot_mode.turn_wait_seconds``). +# Config fallbacks (real knobs: ``bot_mode.turn_wait_seconds`` / ``bot_mode.envelope_ttl_seconds``). TURN_WAIT_SECONDS_FALLBACK = 120 - -# Waiter give-up budget. Cross-connection turns can be slow (remote model, -# cold gateway) — generous, but bounded. +DEFAULT_ENVELOPE_TTL_SECONDS = 900 # older envelopes are refused at drain with 'queued_expired' +# Waiter give-up budget: cross-connection turns can be slow — generous, but bounded. REPLY_WAIT_SECONDS = 900 - -# Envelopes/replies older than this are stale artifacts (Desktop closed, -# connection died) and are swept opportunistically. +# Envelopes/replies older than this are stale artifacts (Desktop closed) and are swept. STALE_AFTER_SECONDS = 6 * 3600 - -# Fallback envelope TTL when config is unreachable — mirrors the -# ``bot_mode.envelope_ttl_seconds`` default. Older envelopes are refused at -# drain time with a 'queued_expired' error reply instead of delivered late. -DEFAULT_ENVELOPE_TTL_SECONDS = 900 - -# A roster older than this proves nothing about who is offline: the Desktop -# re-pushes roster.sync on connection-state changes, so only a recent roster -# is authoritative for the fail-fast check. +# Only a recent roster is authoritative for the fail-fast offline check: the +# Desktop re-pushes roster.sync on connection-state changes. ROSTER_FRESH_SECONDS = 600 class EnvelopeRefusedError(RuntimeError): - """``enqueue_envelope`` refused to queue — nothing was written to disk. - - ``reason`` is a stable machine code ('runtime_offline'); ``str(exc)`` is the - human text. - """ + """``enqueue_envelope`` refused to queue (nothing written); ``reason`` is a stable machine code.""" def __init__(self, reason: str, message: str): super().__init__(message) @@ -104,56 +79,48 @@ def _ensure_dirs(root: Path | str) -> Path: def _atomic_write_json(target: Path, payload: Any, *, prefix: str, sort_keys: bool = False) -> None: - """Write ``payload`` to ``target`` via tempfile + os.replace (readers never see - a partial file). The tempfile is removed if the write fails.""" + """tempfile + os.replace so readers never see a partial file; tempfile removed on failure.""" fd, tmp = tempfile.mkstemp(dir=str(target.parent), prefix=prefix, suffix=".tmp") try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(payload, f, ensure_ascii=False, sort_keys=sort_keys) os.replace(tmp, target) except Exception: - try: + with contextlib.suppress(OSError): os.unlink(tmp) - except OSError: - pass raise -# ── remote roster ──────────────────────────────────────────────────────────── +def _bot_mode_cfg(key: str, *, loader: str) -> Any: + """``bot_mode.`` from config, read lazily (tools/ must not import CLI + config at import time); None when absent or the config is unreadable.""" + try: + import hermes_cli.config as cfgmod + + cfg = getattr(cfgmod, loader)() or {} + return (cfg.get("bot_mode") or {}).get(key) + except Exception: + logger.debug("bot_mode.%s config read failed", key, exc_info=True) + return None def _normalize_roster_row(row: Any) -> Optional[dict]: - """Validated, minimal roster row or None. - - Rows come from the Desktop over RPC — treat as untrusted input. A row names - an agent on another connection: profile, taggable handle, owning connection - id/label, and optional title/description for the protocol section. - """ + """Validated, minimal roster row or None. Rows come from the Desktop over + RPC — treat as untrusted input.""" if not isinstance(row, dict): return None profile = str(row.get("profile") or "").strip() - handle = str(row.get("handle") or "").strip().lstrip("@") + handle = str(row.get("handle") or "").strip().lstrip("@") or ("hermes" if profile == "default" else profile) connection_id = str(row.get("connection_id") or "").strip() - if not profile or not connection_id: - return None - if not handle: - handle = "hermes" if profile == "default" else profile - if ( - not _HANDLE_RE.match(handle) - or not _HANDLE_RE.match(profile) - or not _HANDLE_RE.match(connection_id) - ): + if not profile or not connection_id or not all(_HANDLE_RE.match(v) for v in (handle, profile, connection_id)): return None out = { - "profile": profile, - "handle": handle, - "connection_id": connection_id, + "profile": profile, "handle": handle, "connection_id": connection_id, "connection_label": str(row.get("connection_label") or "").strip()[:80], "title": str(row.get("title") or "").strip()[:120], "description": " ".join(str(row.get("description") or "").split())[:160], } - # Optional liveness flag, kept only when a real bool so absent stays - # distinguishable from false: absent == unknown == fail-open on enqueue. + # Liveness kept only when a real bool: absent == unknown == fail-open on enqueue. if isinstance(row.get("online"), bool): out["online"] = row["online"] return out @@ -176,8 +143,7 @@ def write_remote_roster(root: Path | str, rows: Any) -> int: def read_remote_roster(root: Path | str) -> list[dict]: """The current remote roster (possibly empty). Never raises.""" try: - raw = (relay_root(root) / ROSTER_FILE).read_text(encoding="utf-8") - data = json.loads(raw) + data = json.loads((relay_root(root) / ROSTER_FILE).read_text(encoding="utf-8")) agents = data.get("agents") if isinstance(data, dict) else None if not isinstance(agents, list): return [] @@ -190,140 +156,87 @@ def read_remote_roster(root: Path | str) -> list[dict]: def resolve_remote_target(raw_target: str, roster: list[dict]) -> Any: - """Resolve ``raw_target`` against the remote roster. - - Accepted forms: - - bare handle/profile (``moxie``) — must be unique across connections; - - ``@`` / ``@`` — exact. - - Returns the matched row, the string ``"ambiguous"`` when a bare form - matches agents on several connections, or None for no match. - """ + """Matched row for a bare handle/profile (unique across connections) or + ``@``; ``"ambiguous"`` for a bare form on several connections; None otherwise.""" want = str(raw_target or "").strip().lstrip("@") if not want: return None conn: Optional[str] = None if "@" in want: want, _, conn = want.partition("@") - want = want.strip() - conn = conn.strip() + want, conn = want.strip(), conn.strip() if not want or not conn: return None - matches = [ - row - for row in roster - if want.lower() in (row["handle"].lower(), row["profile"].lower()) - and (not conn or row["connection_id"].lower() == conn.lower()) - ] + matches = [row for row in roster if want.lower() in (row["handle"].lower(), row["profile"].lower()) + and (not conn or row["connection_id"].lower() == conn.lower())] if not matches: return None return matches[0] if len(matches) == 1 else "ambiguous" def remote_target_forms(roster: list[dict]) -> list[str]: - """Human/agent-facing target strings: bare handle when unique across - connections, else ``handle@connection`` (mirrors ``resolve_remote_target``).""" + """Target strings: bare handle when unique across connections, else + ``handle@connection`` (mirrors ``resolve_remote_target``).""" handles = [row["handle"].lower() for row in roster] - return [ - f"{row['handle']}@{row['connection_id']}" if handles.count(h) > 1 else row["handle"] - for row, h in zip(roster, handles) - ] - - -# ── outbox / replies ───────────────────────────────────────────────────────── + return [f"{row['handle']}@{row['connection_id']}" if handles.count(h) > 1 else row["handle"] + for row, h in zip(roster, handles)] def _envelope_ttl_seconds() -> int: - """Configured drain TTL (``bot_mode.envelope_ttl_seconds``), read per-drain - (tools/ must not import CLI config at import time); falls back to - ``DEFAULT_ENVELOPE_TTL_SECONDS``. ``0`` (or negative) disables expiry.""" - try: - from hermes_cli.config import load_config_readonly - - cfg = load_config_readonly() or {} - val = (cfg.get("bot_mode") or {}).get("envelope_ttl_seconds") - if val is not None: - return int(val) - except Exception: - logger.debug("bot_relay TTL config read failed", exc_info=True) - return DEFAULT_ENVELOPE_TTL_SECONDS + """Configured drain TTL (``bot_mode.envelope_ttl_seconds``), read per-drain. + ``0`` (or negative) disables expiry.""" + val = _bot_mode_cfg("envelope_ttl_seconds", loader="load_config_readonly") + return DEFAULT_ENVELOPE_TTL_SECONDS if val is None else int(val) def _target_liveness(root: Path | str, target: dict) -> Optional[bool]: - """Tri-state liveness for ``target``: True / False / None (unknown). - - 'Definitively offline' = explicit ``online: false`` on the row, or the - target ABSENT from a *fresh* roster (the Desktop re-pushes the whole roster - on connection-state changes). A missing, unreadable, empty or stale roster - proves nothing → None, and callers fail open. Never raises. - """ + """Tri-state liveness: True / False / None (unknown → callers fail open). Offline = + explicit ``online: false`` or ABSENT from a *fresh* roster; a missing, unreadable, + empty or stale roster proves nothing → None. Never raises.""" try: - roster_path = relay_root(root) / ROSTER_FILE try: - age = time.time() - roster_path.stat().st_mtime + age = time.time() - (relay_root(root) / ROSTER_FILE).stat().st_mtime except OSError: - return None # no roster ever synced — unknown + return None if age > ROSTER_FRESH_SECONDS: - return None # stale view — unknown + return None roster = read_remote_roster(root) if not roster: - return None # empty/corrupt roster — treat as unknown, fail open + return None key = (str(target.get("connection_id") or ""), str(target.get("profile") or "")) for row in roster: if (row["connection_id"], row["profile"]) == key: online = row.get("online") - if online is False: - return False - return True if online is True else None + return online if isinstance(online, bool) else None return False # fresh roster no longer lists the target — offline except Exception: logger.debug("bot_relay liveness check failed", exc_info=True) return None -def enqueue_envelope( - root: Path | str, - *, - target: dict, - message: str, - sender_profile: str, - sender_handle: str, -) -> dict: - """Queue a cross-connection DM for the Desktop relay. Returns envelope. - - Raises ``EnvelopeRefusedError`` ('runtime_offline') without writing when the - target is definitively offline; unknown liveness enqueues (fail-open). - """ +def enqueue_envelope(root: Path | str, *, target: dict, message: str, sender_profile: str, sender_handle: str) -> dict: + """Queue a cross-connection DM for the Desktop relay; returns the envelope. Raises + ``EnvelopeRefusedError`` ('runtime_offline') without writing when the target is + definitively offline; unknown liveness enqueues (fail-open).""" if _target_liveness(root, target) is False: - label = ( - f"@{target.get('handle') or target.get('profile') or '?'} on " - f"{target.get('connection_label') or target.get('connection_id') or '?'}" - ) - raise EnvelopeRefusedError( - "runtime_offline", - f"{label} is offline right now — the message was NOT queued. " - "Try again once that machine reconnects to the Desktop.", - ) + label = (f"@{target.get('handle') or target.get('profile') or '?'} on " + f"{target.get('connection_label') or target.get('connection_id') or '?'}") + raise EnvelopeRefusedError("runtime_offline", f"{label} is offline right now — the message was NOT queued. " + "Try again once that machine reconnects to the Desktop.") base = _ensure_dirs(root) envelope = { - "id": uuid.uuid4().hex, - "created_at": int(time.time()), - "from_profile": sender_profile, - "from_handle": sender_handle, - "target_connection": target["connection_id"], - "target_profile": target["profile"], - "target_handle": target["handle"], - "message": message, + "id": uuid.uuid4().hex, "created_at": int(time.time()), + "from_profile": sender_profile, "from_handle": sender_handle, + "target_connection": target["connection_id"], "target_profile": target["profile"], + "target_handle": target["handle"], "message": message, } _atomic_write_json(base / OUTBOX_DIR / f"{envelope['id']}.json", envelope, prefix=".env-") return envelope def _expire_if_stale(root: Path | str, path: Path, ttl: float, now: float) -> bool: - """True when the outbox envelope at ``path`` is older than ``ttl``; writes the - 'queued_expired' error reply so the sender's waiter resolves (best effort — - an invalid id still counts as expired). Unreadable envelopes are left for - the claim attempt to deal with.""" + """True when the outbox envelope is older than ``ttl``; writes the 'queued_expired' + reply so the sender's waiter resolves (best effort). Unreadable envelopes are left for the claim.""" try: env = json.loads(path.read_text(encoding="utf-8")) created = float(env.get("created_at") or path.stat().st_mtime) @@ -331,30 +244,17 @@ def _expire_if_stale(root: Path | str, path: Path, ttl: float, now: float) -> bo return False if now - created <= ttl: return False - handle = str(env.get("target_handle") or "?") - conn = str(env.get("target_connection") or "?") with contextlib.suppress(OSError, ValueError): - write_reply( - root, - str(env.get("id") or ""), - error=( - f"queued message to @{handle} on {conn} expired after " - f"{ttl}s waiting for the Desktop to drain it — it was " - "NOT delivered. Resend once the Desktop reconnects." - ), - reason="queued_expired", - ) + write_reply(root, str(env.get("id") or ""), reason="queued_expired", error=( + f"queued message to @{env.get('target_handle') or '?'} on {env.get('target_connection') or '?'} " + f"expired after {ttl}s waiting for the Desktop to drain it — it was NOT delivered. " + "Resend once the Desktop reconnects.")) return True def claim_pending_envelopes(root: Path | str) -> list[dict]: - """Drain the outbox (rename → claimed/, so a second drain can't double- - deliver). Sweeps stale claimed/reply artifacts opportunistically. - - Envelopes older than ``bot_mode.envelope_ttl_seconds`` are NOT delivered: - each gets a 'queued_expired' error reply (so the sender's waiter resolves) - and its outbox file is removed. - """ + """Drain the outbox (rename → claimed/ so a second drain can't double-deliver). + TTL-expired envelopes get a 'queued_expired' reply and are removed instead.""" base = _ensure_dirs(root) _sweep_stale(base) ttl = _envelope_ttl_seconds() @@ -366,98 +266,65 @@ def claim_pending_envelopes(root: Path | str) -> list[dict]: path.unlink() continue claimed = base / CLAIMED_DIR / path.name - try: + with contextlib.suppress(OSError, ValueError): os.replace(path, claimed) # atomic claim out.append(json.loads(claimed.read_text(encoding="utf-8"))) - except (OSError, ValueError): - continue return out -def write_reply( - root: Path | str, envelope_id: str, *, reply: str = "", error: str = "", reason: str = "" -) -> Path: - """Persist the relayed reply (or delivery error) for the waiter. - - ``reason`` is an optional typed failure code (``tools.bot_failure_reasons``); - when omitted and ``error`` is non-empty it is classified from the text. The - waiter only surfaces the human ``error`` (plus the code as a tag). - """ +def write_reply(root: Path | str, envelope_id: str, *, reply: str = "", error: str = "", reason: str = "") -> Path: + """Persist the relayed reply (or delivery error) for the waiter. ``reason`` (typed + code, ``tools.bot_failure_reasons``) is classified from ``error`` when omitted.""" base = _ensure_dirs(root) safe = str(envelope_id or "").strip() if not re.match(r"^[0-9a-f]{32}$", safe): raise ValueError(f"invalid envelope id: {envelope_id!r}") - err = str(error or "") - code = str(reason or "") + err, code = str(error or ""), str(reason or "") if not code and err: from tools.bot_failure_reasons import classify_agent_error code = classify_agent_error(err) path = base / REPLIES_DIR / f"{safe}.json" - payload = {"id": safe, "at": int(time.time()), "reply": str(reply or ""), "error": err, "reason": code} - _atomic_write_json(path, payload, prefix=".rep-") + _atomic_write_json(path, {"id": safe, "at": int(time.time()), "reply": str(reply or ""), "error": err, "reason": code}, + prefix=".rep-") return path def unlink_files_older_than(directory: Path, pattern: str, cutoff: float) -> int: - """Remove regular files under ``directory`` matching ``pattern`` with mtime - before ``cutoff``; returns the count. Never raises (missing dir → 0).""" + """Unlink regular files matching ``pattern`` with mtime before ``cutoff``; returns count. Never raises.""" removed = 0 - try: + with contextlib.suppress(OSError): for path in directory.glob(pattern): - try: + with contextlib.suppress(OSError): if path.is_file() and path.stat().st_mtime < cutoff: path.unlink() removed += 1 - except OSError: - continue - except OSError: - pass return removed def _sweep_stale(base: Path, *, now: float | None = None) -> int: cutoff = (time.time() if now is None else now) - STALE_AFTER_SECONDS - return sum( - unlink_files_older_than(base / sub, "*.json", cutoff) - for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR) - ) + return sum(unlink_files_older_than(base / sub, "*.json", cutoff) for sub in (CLAIMED_DIR, REPLIES_DIR, OUTBOX_DIR)) def cleanup_bot_relay_artifacts(max_age_hours: float | None = None) -> int: - """Sweep stale relay artifacts (envelopes/replies hold DM plaintext). - - ``_sweep_stale`` otherwise runs only on Desktop drains — if the Desktop never - reconnects, plaintext would sit on disk forever. Same contract as the - ``cleanup_*_cache`` helpers (hourly housekeeping). ``max_age_hours`` is - accepted for signature compatibility; ``STALE_AFTER_SECONDS`` governs. - """ + """Hourly sweep of stale relay artifacts (DM plaintext; ``_sweep_stale`` otherwise runs + only on Desktop drains). ``max_age_hours`` is for ``cleanup_*_cache`` signature parity only.""" del max_age_hours try: base = relay_root(_hermes_root(Path(_default_home()))) - if not base.is_dir(): - return 0 - return _sweep_stale(base) + return _sweep_stale(base) if base.is_dir() else 0 except Exception: logger.debug("bot_relay artifact sweep failed", exc_info=True) return 0 -# ── waiter (runs on the sender gateway via terminal background process) ───── - - def waiter_command(root: Path | str, envelope: dict) -> str: - """Shell command that blocks until the reply file appears, then prints it. - - Spawned via ``terminal_tool(background=True, notify_on_complete=True)`` so - its stdout — the reply — arrives as the same completion notification local - DMs use. Stdlib-only; runs under the sender gateway's interpreter. - """ + """Shell command that blocks until the reply file appears, then prints it; spawned + via ``terminal_tool(background=True, notify_on_complete=True)`` so its stdout arrives + as the same completion notification local DMs use. Stdlib-only.""" reply_path = str(relay_root(root) / REPLIES_DIR / f"{envelope['id']}.json") - label = ( - f"@{envelope.get('target_handle', '')} " - f"on {envelope.get('target_connection', '')}" - ) + label = f"@{envelope.get('target_handle', '')} on {envelope.get('target_connection', '')}" # !r keeps roster fields from breaking out of the generated python -c source. # The r-prefix keeps Windows paths viable: the Windows execution layer folds # repr's "\\" back to "\", turning "\U" into an invalid unicode escape; a @@ -490,21 +357,11 @@ def waiter_command(root: Path | str, envelope: dict) -> str: return f"{shlex.quote(sys.executable or 'python3')} -c {shlex.quote(code)}" -# ── delivery command (used by the deliver RPC on the TARGET gateway) ──────── - - def _hermes_cli() -> str: - """Resolve the hermes CLI beside this gateway's own interpreter. - - Service contexts (systemd, desktop launchers, non-login SSH) lack PATH, so a - bare "hermes" died with ENOENT; the venv sibling wins, then ``shutil.which`` - (honors whatever PATH exists), then the bare name. - """ - exe = Path(sys.executable or "") - sibling = exe.parent / ("hermes.exe" if sys.platform == "win32" else "hermes") - if sibling.is_file(): - return str(sibling) - return shutil.which("hermes") or "hermes" + """hermes CLI beside this interpreter, then ``shutil.which``, then the bare name + (service contexts lack PATH, so a bare "hermes" died with ENOENT).""" + sibling = Path(sys.executable or "").parent / ("hermes.exe" if sys.platform == "win32" else "hermes") + return str(sibling) if sibling.is_file() else shutil.which("hermes") or "hermes" def local_delivery_command(profile: str, query_file: str) -> list[str]: @@ -512,46 +369,29 @@ def local_delivery_command(profile: str, query_file: str) -> list[str]: return [_hermes_cli(), "-p", profile, *BOT_CHAT_TURN_ARGS, "--query-file", query_file] -# ── per-profile turn lock ──────────────────────────────────────────────────── -# -# Two deliveries into the SAME profile must never run Bot Chat turns -# concurrently. Deliveries are separate ``hermes`` subprocesses, so the lock is -# a per-profile lockfile under ``/bot_relay/locks/`` held with -# ``fcntl.flock`` for exactly the turn window; the kernel releases it on fd -# close (including process death), so a crashed turn can never wedge the -# profile. Waiters are bounded by ``bot_mode.turn_wait_seconds`` and then fail -# with a structured 'target_busy' refusal. +# Two deliveries into the SAME profile must never run Bot Chat turns concurrently. +# Deliveries are separate ``hermes`` subprocesses, so the lock is a per-profile +# lockfile under ``/bot_relay/locks/`` held with ``fcntl.flock`` for exactly +# the turn window; the kernel releases it on fd close (incl. process death), so a +# crashed turn can never wedge the profile. class TurnBusyError(RuntimeError): - """A delivery turn is already running for the target profile. - - ``waited_seconds``: roughly how long the caller queued before giving up. - """ + """A delivery turn is already running for the target profile (``waited_seconds`` ≈ time queued).""" reason = "target_busy" def __init__(self, profile: str, waited_seconds: float): - self.profile = profile - self.waited_seconds = waited_seconds - super().__init__( - f"target_busy: another delivery turn is already running for " - f"profile '{profile}' — queued behind it for ~{int(round(waited_seconds))}s " - "without it finishing. The message was NOT delivered; retry shortly." - ) + self.profile, self.waited_seconds = profile, waited_seconds + super().__init__(f"target_busy: another delivery turn is already running for profile '{profile}' — " + f"queued behind it for ~{int(round(waited_seconds))}s without it finishing. " + "The message was NOT delivered; retry shortly.") def turn_wait_seconds() -> float: """Wait budget for a queued delivery turn (config, lazily read).""" - try: - from hermes_cli.config import cfg_get, load_config - - val = cfg_get(load_config(), "bot_mode", "turn_wait_seconds", default=None) - if val is not None: - return max(0.0, float(val)) - except Exception: - logger.debug("bot_mode.turn_wait_seconds read failed", exc_info=True) - return float(TURN_WAIT_SECONDS_FALLBACK) + val = _bot_mode_cfg("turn_wait_seconds", loader="load_config") + return float(TURN_WAIT_SECONDS_FALLBACK) if val is None else max(0.0, float(val)) def turn_lock_path(root: Path | str, profile: str) -> Path: @@ -561,18 +401,12 @@ def turn_lock_path(root: Path | str, profile: str) -> Path: @contextlib.contextmanager -def acquire_turn_lock( - root: Path | str, profile: str, timeout_seconds: float | None = None -) -> Iterator[Path]: - """Hold ``profile``'s cross-process turn lock for the ``with`` body. - - Non-blocking flock probe + short-sleep retry up to the budget - (``bot_mode.turn_wait_seconds`` unless ``timeout_seconds`` is given). No - ordering among waiters, but every waiter is bounded — no deadlock. Raises - :class:`TurnBusyError` when the budget is exhausted. Without ``fcntl`` - (Windows) the lock is a no-op — those installs never had this race path - in production. - """ +def acquire_turn_lock(root: Path | str, profile: str, timeout_seconds: float | None = None) -> Iterator[Path]: + """Hold ``profile``'s cross-process turn lock for the ``with`` body: non-blocking + flock probe + short-sleep retry up to the budget (``bot_mode.turn_wait_seconds`` + unless ``timeout_seconds``); raises :class:`TurnBusyError` when exhausted. No + ordering among waiters, but every waiter is bounded. Without ``fcntl`` (Windows) + the lock is a no-op — those installs never had this race path.""" try: import fcntl except ImportError: # pragma: no cover — Windows @@ -599,9 +433,7 @@ def acquire_turn_lock( try: yield path finally: - try: + with contextlib.suppress(OSError): # kernel releases on close anyway fcntl.flock(fd, fcntl.LOCK_UN) - except OSError: # pragma: no cover — kernel releases on close anyway - pass finally: os.close(fd) diff --git a/tools/browser_dialog_tool.py b/tools/browser_dialog_tool.py index b1e3d7a8c5..ecd6dffa1e 100644 --- a/tools/browser_dialog_tool.py +++ b/tools/browser_dialog_tool.py @@ -8,15 +8,11 @@ appears together with ``browser_cdp``. Design: ``website/docs/developer-guide/br from __future__ import annotations import json -import logging from typing import Any, Dict, Optional from tools.browser_supervisor import SUPERVISOR_REGISTRY from tools.registry import registry -logger = logging.getLogger(__name__) - - BROWSER_DIALOG_SCHEMA: Dict[str, Any] = { "name": "browser_dialog", "description": ( @@ -106,12 +102,8 @@ registry.register( name="browser_dialog", toolset="browser-cdp", schema=BROWSER_DIALOG_SCHEMA, - handler=lambda args, **kw: browser_dialog( - action=args.get("action", ""), - prompt_text=args.get("prompt_text"), - dialog_id=args.get("dialog_id"), - task_id=kw.get("task_id"), - ), + handler=lambda args, **kw: browser_dialog(action=args.get("action", ""), prompt_text=args.get("prompt_text"), + dialog_id=args.get("dialog_id"), task_id=kw.get("task_id")), check_fn=_browser_dialog_check, emoji="💬", ) diff --git a/tools/browser_extension_router.py b/tools/browser_extension_router.py index cdf5ac7eab..626fcc26b4 100644 --- a/tools/browser_extension_router.py +++ b/tools/browser_extension_router.py @@ -5,7 +5,6 @@ invocation, whether an attached extension controller (via :mod:`gateway.browser_control_broker`) or the legacy backend executes it. Routing contract (see ``tests/tools/test_browser_extension_router.py``): - - Feature off ⇒ legacy, broker never touched, ``fallback()`` called exactly once. - No server-bound identity ⇒ legacy. - Bound identity ⇒ authoritative extension lane; missing/ambiguous scope, @@ -31,10 +30,10 @@ def _bound_identity() -> Tuple[Optional[str], Optional[str], Optional[str]]: """(session_id, principal_id, transport_family) from the session context.""" from gateway.session_context import get_session_env - return ( - get_session_env("HERMES_SESSION_ID", "") or None, - get_session_env("HERMES_BROWSER_CONTROL_PRINCIPAL", "") or None, - get_session_env("HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY", "") or None, + return tuple( # type: ignore[return-value] + get_session_env(key, "") or None + for key in ("HERMES_SESSION_ID", "HERMES_BROWSER_CONTROL_PRINCIPAL", + "HERMES_BROWSER_CONTROL_TRANSPORT_FAMILY") ) @@ -46,10 +45,8 @@ def _controller_unavailable(message: str) -> Exception: def extension_controller_available(action: str) -> bool: """Whether this request owns one exact controller capable of ``action``. - Runs during tool-schema assembly inside the request's session context; - consults the process-local broker directly and fails closed on any gap. - """ + consults the process-local broker directly and fails closed on any gap.""" try: from gateway.browser_control_broker import browser_control_enabled, get_browser_control_broker @@ -59,9 +56,7 @@ def extension_controller_available(action: str) -> bool: if not session_id or not principal_id or not transport_family: return False broker = get_browser_control_broker() - scope = broker.scope_for_session( - session_id=session_id, principal_id=principal_id, transport_family=transport_family, - ) + scope = broker.scope_for_session(session_id=session_id, principal_id=principal_id, transport_family=transport_family) return scope is not None and broker.select(scope, action) is not None except Exception: logger.debug("browser extension availability check failed for %s", action, exc_info=True) @@ -69,17 +64,9 @@ def extension_controller_available(action: str) -> bool: def route_browser_tool( - action: str, - args: Dict[str, Any], - *, - fallback: Callable[[], Any], - broker: Any, - enabled: bool, - session_id: Optional[str] = None, - task_id: Optional[str] = None, - principal_id: Optional[str] = None, - transport_family: Optional[str] = None, - tool_call_id: Optional[str] = "", + action: str, args: Dict[str, Any], *, fallback: Callable[[], Any], broker: Any, enabled: bool, + session_id: Optional[str] = None, task_id: Optional[str] = None, principal_id: Optional[str] = None, + transport_family: Optional[str] = None, tool_call_id: Optional[str] = "", ) -> Any: """Route one browser action through the extension-control broker. @@ -91,9 +78,7 @@ def route_browser_tool( if not enabled or not str(principal_id or "").strip() or not str(transport_family or "").strip(): return fallback() - identity = dict( - session_id=session_id, task_id=task_id, principal_id=principal_id, transport_family=transport_family, - ) + identity = dict(session_id=session_id, task_id=task_id, principal_id=principal_id, transport_family=transport_family) scope = broker.scope_for_session(**identity) if scope is None: # A stamped identity only becomes authoritative once a controller has @@ -125,20 +110,12 @@ def current_tool_call_id() -> str: def routed_browser_handler( - action: str, - args: Dict[str, Any], - *, - fallback: Callable[[], Any], - task_id: Optional[str] = None, - session_id: Optional[str] = None, - principal_id: Optional[str] = None, - transport_family: Optional[str] = None, - tool_call_id: Optional[str] = None, + action: str, args: Dict[str, Any], *, fallback: Callable[[], Any], task_id: Optional[str] = None, + session_id: Optional[str] = None, principal_id: Optional[str] = None, + transport_family: Optional[str] = None, tool_call_id: Optional[str] = None, ) -> Any: """Lazy registry-handler route wrapper for ``browser_*`` tools. - - Feature off (or gateway unimportable) ⇒ the legacy handler runs unchanged. - """ + Feature off (or gateway unimportable) ⇒ the legacy handler runs unchanged.""" try: from gateway.browser_control_broker import browser_control_enabled, get_browser_control_broker except Exception as exc: # pragma: no cover - defensive, gateway always present @@ -153,14 +130,8 @@ def routed_browser_handler( env_session = env_principal = env_transport = None return route_browser_tool( - action, - args, - fallback=fallback, - broker=get_browser_control_broker(), - enabled=True, - session_id=session_id or env_session, - task_id=task_id, - principal_id=principal_id or env_principal, + action, args, fallback=fallback, broker=get_browser_control_broker(), enabled=True, + session_id=session_id or env_session, task_id=task_id, principal_id=principal_id or env_principal, transport_family=transport_family or env_transport, tool_call_id=current_tool_call_id() if tool_call_id is None else tool_call_id, ) diff --git a/tools/browser_supervisor.py b/tools/browser_supervisor.py index da646199fa..67ab90d7e5 100644 --- a/tools/browser_supervisor.py +++ b/tools/browser_supervisor.py @@ -6,19 +6,17 @@ It holds one persistent WebSocket, subscribes to ``Page`` / ``Runtime`` / worker targets), and exposes pending dialogs + frame tree through a thread-safe snapshot that tool handlers read synchronously. -Not in the agent's tool schema. Output reaches the agent via -``browser_snapshot`` (merges supervisor state, see ``tools/browser_tool.py``) -and ``browser_dialog`` (calls ``respond_to_dialog()``). +Not in the agent's tool schema: output reaches the agent via ``browser_snapshot`` +and ``browser_dialog``. Dialog capture lives in ``browser_supervisor_dialogs``, +frame tracking in ``browser_supervisor_frames``; both are mixed into +``CDPSupervisor`` and their public names are re-exported here. Design spec: ``website/docs/developer-guide/browser-supervisor.md``. - -Dialog capture lives in ``tools.browser_supervisor_dialogs``, frame tracking in -``tools.browser_supervisor_frames``; both are mixed into ``CDPSupervisor`` and -their public names are re-exported here. """ from __future__ import annotations import asyncio +import contextlib import json import logging import threading @@ -27,31 +25,16 @@ from dataclasses import dataclass from typing import Any, Callable, Dict, List, Optional, Tuple, TYPE_CHECKING from tools.browser_supervisor_dialogs import ( # noqa: F401 — re-exported - DEFAULT_DIALOG_POLICY, - DEFAULT_DIALOG_TIMEOUT_S, - DIALOG_BRIDGE_HOST, - DIALOG_BRIDGE_URL_PATTERN, - DIALOG_POLICY_AUTO_ACCEPT, - DIALOG_POLICY_AUTO_DISMISS, - DIALOG_POLICY_MUST_RESPOND, - RECENT_DIALOGS_MAX, - _DIALOG_BRIDGE_SCRIPT, - _VALID_POLICIES, - DialogRecord, - DialogSupervisionMixin, - PendingDialog, - _redact_supervisor_text, - _trim_ring, + DEFAULT_DIALOG_POLICY, DEFAULT_DIALOG_TIMEOUT_S, DIALOG_BRIDGE_HOST, DIALOG_BRIDGE_URL_PATTERN, + DIALOG_POLICY_AUTO_ACCEPT, DIALOG_POLICY_AUTO_DISMISS, DIALOG_POLICY_MUST_RESPOND, RECENT_DIALOGS_MAX, + _DIALOG_BRIDGE_SCRIPT, _VALID_POLICIES, DialogRecord, DialogSupervisionMixin, PendingDialog, + _redact_supervisor_text, _trim_ring, ) from tools.browser_supervisor_frames import ( # noqa: F401 — re-exported - FRAME_TREE_MAX_ENTRIES, - FRAME_TREE_MAX_OOPIF_DEPTH, - FrameInfo, - FrameTrackingMixin, + FRAME_TREE_MAX_ENTRIES, FRAME_TREE_MAX_OOPIF_DEPTH, FrameInfo, FrameTrackingMixin, ) -# ``websockets`` costs ~22 ms at import and is only needed once a supervisor -# connects; with postponed annotations the type import stays under TYPE_CHECKING. +# ``websockets`` costs ~22 ms at import and is only needed once a supervisor connects. if TYPE_CHECKING: from websockets.asyncio.client import ClientConnection @@ -62,9 +45,9 @@ def _redact_cdp_error_text(exc: object) -> str: """Redact CDP endpoint credentials from an exception's (or URL's) string form. ``websockets`` bakes the raw target URL (``?token=`` / ``user:pass@``) into - its exception messages. Every egress point that turns such an exception into - log text or a re-raised message MUST route through here; falls back to a - fixed sentinel if redaction itself raises, erring toward masking. + its exception messages, so every egress point that turns such an exception + into log text or a re-raised message MUST route through here; falls back to + a fixed sentinel if redaction itself raises, erring toward masking. """ try: from agent.redact import redact_cdp_url @@ -88,13 +71,17 @@ def _schedule(coro, loop, *, timeout: float): return fut.result(timeout=timeout) +def _fail(error: str) -> Dict[str, Any]: + return {"ok": False, "error": error} + + def _err(exc: BaseException) -> Dict[str, Any]: - return {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + return _fail(f"{type(exc).__name__}: {exc}") @dataclass(frozen=True) class SupervisorSnapshot: - """Read-only (frozen) snapshot of supervisor state for tool handlers.""" + """Read-only snapshot of supervisor state for tool handlers.""" pending_dialogs: Tuple[PendingDialog, ...] recent_dialogs: Tuple[DialogRecord, ...] @@ -105,41 +92,22 @@ class SupervisorSnapshot: def to_dict(self) -> Dict[str, Any]: """Serialize for inclusion in ``browser_snapshot`` output.""" - out: Dict[str, Any] = { - "pending_dialogs": [d.to_dict() for d in self.pending_dialogs], - "frame_tree": self.frame_tree, - } + out: Dict[str, Any] = {"pending_dialogs": [d.to_dict() for d in self.pending_dialogs], "frame_tree": self.frame_tree} if self.recent_dialogs: out["recent_dialogs"] = [d.to_dict() for d in self.recent_dialogs] return out -# ── Supervisor core ─────────────────────────────────────────────────────────── - - class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): - """One supervisor per (task_id, cdp_url) pair. + """One supervisor per (task_id, cdp_url) pair. ``start()`` spawns a daemon thread + running its own asyncio loop, connects, attaches to the first page target, enables + domains and auto-attach. ``snapshot()`` / ``respond_to_dialog()`` / ``evaluate_runtime()`` + are sync, thread-safe bridges onto that loop; all CDP I/O lives on the loop.""" - ``start()`` spawns a daemon thread running its own asyncio loop, connects, - attaches to the first page target, enables domains and auto-attach. - ``snapshot()`` / ``respond_to_dialog()`` / ``evaluate_runtime()`` are sync, - thread-safe bridges onto that loop; ``stop()`` tears it down. All CDP I/O - lives on the supervisor's own loop. - """ - - def __init__( - self, - task_id: str, - cdp_url: str, - *, - dialog_policy: str = DEFAULT_DIALOG_POLICY, - dialog_timeout_s: float = DEFAULT_DIALOG_TIMEOUT_S, - ) -> None: + def __init__(self, task_id: str, cdp_url: str, *, dialog_policy: str = DEFAULT_DIALOG_POLICY, + dialog_timeout_s: float = DEFAULT_DIALOG_TIMEOUT_S) -> None: if dialog_policy not in _VALID_POLICIES: - raise ValueError( - f"Invalid dialog_policy {dialog_policy!r}; " - f"must be one of {sorted(_VALID_POLICIES)}" - ) + raise ValueError(f"Invalid dialog_policy {dialog_policy!r}; must be one of {sorted(_VALID_POLICIES)}") self.task_id = task_id self.cdp_url = cdp_url self.dialog_policy = dialog_policy @@ -151,20 +119,17 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): self._recent_dialogs: List[DialogRecord] = [] self._frames: Dict[str, FrameInfo] = {} self._active = False - # Supervisor loop machinery — populated in start(). self._loop: Optional[asyncio.AbstractEventLoop] = None self._thread: Optional[threading.Thread] = None self._ready_event = threading.Event() self._start_error: Optional[BaseException] = None self._stop_requested = False - # CDP call tracking (runs on supervisor loop only). self._next_call_id = 1 self._pending_calls: Dict[int, asyncio.Future] = {} self._ws: Optional[ClientConnection] = None self._page_session_id: Optional[str] = None - # Dialog auto-dismiss watchdog handles (per dialog id) + id generator. self._dialog_watchdogs: Dict[str, asyncio.TimerHandle] = {} self._dialog_seq = 0 @@ -172,52 +137,40 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): # ── Public sync API ────────────────────────────────────────────────────── def start(self, timeout: float = 15.0) -> None: - """Launch the background loop and block until attachment completes. - - Raises whatever attach failed with (redacted). On return, dialog events - are already being captured. - """ + """Launch the background loop and block until attachment completes; raises what attach failed with (redacted).""" if self._thread and self._thread.is_alive(): return self._ready_event.clear() - self._start_error = None - self._stop_requested = False - self._thread = threading.Thread( - target=self._thread_main, name=f"cdp-supervisor-{self.task_id}", daemon=True, - ) + self._start_error, self._stop_requested = None, False + self._thread = threading.Thread(target=self._thread_main, name=f"cdp-supervisor-{self.task_id}", daemon=True) self._thread.start() if not self._ready_event.wait(timeout=timeout): self.stop() - raise TimeoutError( - f"CDP supervisor did not attach within {timeout}s " - f"(cdp_url={_redact_cdp_error_text(self.cdp_url)[:80]}...)" - ) + raise TimeoutError(f"CDP supervisor did not attach within {timeout}s " + f"(cdp_url={_redact_cdp_error_text(self.cdp_url)[:80]}...)") if self._start_error is not None: err = self._start_error self.stop() - # ``err`` is a raw ``websockets`` exception embedding the full cdp_url - # (token / userinfo). Re-raise redacted and suppress the cause - # (``from None``) so nothing leaks via message OR traceback chain. - raise RuntimeError( - f"CDP supervisor failed to start: {_redact_cdp_error_text(err)}" - ) from None + # ``err`` is a raw ``websockets`` exception embedding the full cdp_url (token / + # userinfo): re-raise redacted, ``from None`` so the traceback chain leaks nothing. + raise RuntimeError(f"CDP supervisor failed to start: {_redact_cdp_error_text(err)}") from None def stop(self, timeout: float = 5.0) -> None: """Cancel the supervisor task and join the thread.""" self._stop_requested = True loop = self._loop if loop is not None and loop.is_running(): - # Close the WebSocket from inside the loop so ``async for raw in - # self._ws`` returns cleanly, ``_run`` hits its ``finally``, pending - # tasks cancel in order, THEN the thread exits. - try: + # Close the WebSocket from inside the loop so ``async for raw in self._ws`` + # returns cleanly, ``_run`` hits its ``finally``, THEN the thread exits. + with contextlib.suppress(Exception): # loop already shutting down / close timed out _schedule(self._close_ws(), loop, timeout=2.0) - except Exception: - pass # loop already shutting down / close timed out if self._thread is not None: self._thread.join(timeout=timeout) + self._set_active(False) + + def _set_active(self, value: bool) -> None: with self._state_lock: - self._active = False + self._active = value def snapshot(self) -> SupervisorSnapshot: """Return an immutable snapshot of current state.""" @@ -226,120 +179,73 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): pending_dialogs=tuple(self._pending_dialogs.values()), recent_dialogs=tuple(self._recent_dialogs[-RECENT_DIALOGS_MAX:]), frame_tree=self._build_frame_tree_locked(), - active=self._active, - cdp_url=self.cdp_url, - task_id=self.task_id, + active=self._active, cdp_url=self.cdp_url, task_id=self.task_id, ) - def respond_to_dialog( - self, - action: str, - *, - prompt_text: Optional[str] = None, - dialog_id: Optional[str] = None, - timeout: float = 10.0, - ) -> Dict[str, Any]: + def respond_to_dialog(self, action: str, *, prompt_text: Optional[str] = None, + dialog_id: Optional[str] = None, timeout: float = 10.0) -> Dict[str, Any]: """Accept/dismiss a pending dialog (sync bridge onto the supervisor loop). Returns ``{"ok": True, "dialog": {...}}`` or ``{"ok": False, "error": ...}`` for recoverable errors (no dialog, ambiguous dialog_id, inactive). """ if action not in {"accept", "dismiss"}: - return {"ok": False, "error": f"action must be 'accept' or 'dismiss', got {action!r}"} - + return _fail(f"action must be 'accept' or 'dismiss', got {action!r}") with self._state_lock: if not self._active: - return {"ok": False, "error": "supervisor is not active"} + return _fail("supervisor is not active") pending = list(self._pending_dialogs.values()) if not pending: - return {"ok": False, "error": "no dialog is currently open"} + return _fail("no dialog is currently open") if dialog_id: dialog = self._pending_dialogs.get(dialog_id) if dialog is None: - return { - "ok": False, - "error": f"dialog_id {dialog_id!r} not found " - f"(known: {sorted(self._pending_dialogs)})", - } + return _fail(f"dialog_id {dialog_id!r} not found (known: {sorted(self._pending_dialogs)})") elif len(pending) > 1: - return { - "ok": False, - "error": ( - f"{len(pending)} pending dialogs; specify dialog_id. " - f"Candidates: {[d.id for d in pending]}" - ), - } + return _fail(f"{len(pending)} pending dialogs; specify dialog_id. Candidates: {[d.id for d in pending]}") else: dialog = pending[0] - loop = self._loop if loop is None: - return {"ok": False, "error": "supervisor loop is not running"} - + return _fail("supervisor loop is not running") try: - _schedule( - self._handle_dialog_cdp( - dialog, accept=(action == "accept"), prompt_text=prompt_text or "" - ), - loop, - timeout=timeout, - ) + coro = self._handle_dialog_cdp(dialog, accept=(action == "accept"), prompt_text=prompt_text or "") + _schedule(coro, loop, timeout=timeout) except _LoopUnavailable as e: - return {"ok": False, "error": str(e)} + return _fail(str(e)) except Exception as e: return _err(e) return {"ok": True, "dialog": dialog.to_dict()} - def evaluate_runtime( - self, - expression: str, - *, - return_by_value: bool = True, - await_promise: bool = True, - timeout: float = 10.0, - ) -> Dict[str, Any]: + def evaluate_runtime(self, expression: str, *, return_by_value: bool = True, + await_promise: bool = True, timeout: float = 10.0) -> Dict[str, Any]: """Evaluate ``expression`` in the page's Runtime context over the live WS. - - Zero subprocess cost vs the agent-browser CLI ``eval``. Returns - ``{"ok": True, "result": , "result_type": ...}`` or - ``{"ok": False, "error": ...}``. ``return_by_value=True`` JSON-serializes - the result (DevTools-console semantics); non-serializable objects come - back as a description string. - """ + Returns ``{"ok": True, "result", "result_type"}`` or ``{"ok": False, "error"}``. + ``return_by_value=True`` JSON-serializes the result (DevTools-console + semantics); non-serializable objects come back as a description string.""" loop = self._loop if loop is None or not loop.is_running(): - return {"ok": False, "error": "supervisor loop is not running"} - + return _fail("supervisor loop is not running") with self._state_lock: if not self._active: - return {"ok": False, "error": "supervisor is not active"} + return _fail("supervisor is not active") session_id = self._page_session_id - if not session_id: - return {"ok": False, "error": "supervisor has no attached page session"} + return _fail("supervisor has no attached page session") def _run_eval(by_value: bool) -> Dict[str, Any]: - coro = self._cdp( - "Runtime.evaluate", - { - "expression": expression, - "returnByValue": by_value, - "awaitPromise": await_promise, - # userGesture: clipboard / fullscreen APIs need user activation. - "userGesture": True, - }, - session_id=session_id, - timeout=timeout, - ) + # userGesture: clipboard / fullscreen APIs need user activation. + params = {"expression": expression, "returnByValue": by_value, + "awaitPromise": await_promise, "userGesture": True} + coro = self._cdp("Runtime.evaluate", params, session_id=session_id, timeout=timeout) return _schedule(coro, loop, timeout=timeout + 1) try: response = _run_eval(return_by_value) except Exception as exc: # Deep-serializing live DOM nodes / NodeLists / Window can blow past - # CDP's recursion guard with the protocol-level error ``Object - # reference chain is too long``. Retry once with returnByValue=False - # so Chrome returns the description string instead of failing. + # CDP's recursion guard (``Object reference chain is too long``). + # Retry once with returnByValue=False so Chrome returns the description. if not (return_by_value and "reference chain is too long" in str(exc).lower()): return _err(exc) try: @@ -353,9 +259,7 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): if exception_details: exc_text = exception_details.get("text") or "JavaScript exception" description = (exception_details.get("exception") or {}).get("description") - if description: - exc_text = f"{exc_text}: {description}" - return {"ok": False, "error": exc_text} + return _fail(f"{exc_text}: {description}" if description else exc_text) result_obj = result_payload.get("result", {}) result_type = result_obj.get("type", "undefined") @@ -364,8 +268,7 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): elif result_type == "undefined": value = None else: - # Non-serializable (functions, DOM nodes…) — give the model the - # browser's description so it gets *something*. + # Non-serializable (functions, DOM nodes…) — give the model the browser's description. value = result_obj.get("description") or result_obj.get("unserializableValue") return {"ok": True, "result": value, "result_type": result_type} @@ -384,17 +287,14 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): finally: # Cancel + flush remaining tasks before closing the loop to avoid # "Task was destroyed but it is pending" warnings. - try: + with contextlib.suppress(Exception): pending = [t for t in asyncio.all_tasks(loop) if not t.done()] for t in pending: t.cancel() if pending: loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) loop.close() - except Exception: - pass - with self._state_lock: - self._active = False + self._set_active(False) def _fail_start(self, e: BaseException) -> bool: """Propagate ``e`` to ``start()`` if we never got ready; True if it was consumed.""" @@ -408,50 +308,37 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): """Detach and close the current WebSocket, swallowing close errors.""" ws, self._ws = self._ws, None if ws is not None: - try: + with contextlib.suppress(Exception): await ws.close() - except Exception: - pass async def _run(self) -> None: - """Top-level reconnecting supervisor coroutine. - - Browserbase tears down the CDP socket every time a short-lived client - (e.g. agent-browser's per-command CDP client) disconnects, so on drop we - reset per-session ids, re-attach, and keep going. A failure before the - first successful attach is fatal for ``start()``. - """ - attempt = 0 - last_success_at = 0.0 - backoff = 0.5 + """Top-level reconnecting supervisor coroutine. Browserbase tears down the CDP + socket whenever a short-lived client (agent-browser's per-command CDP client) + disconnects, so on drop we reset per-session ids, re-attach, and keep going. + A failure before the first successful attach is fatal for ``start()``.""" + attempt, last_success_at, backoff = 0, 0.0, 0.5 import websockets # deferred: only supervisors that connect pay the import while not self._stop_requested: try: - self._ws = await asyncio.wait_for( - websockets.connect(self.cdp_url, max_size=50 * 1024 * 1024), timeout=10.0, - ) + self._ws = await asyncio.wait_for(websockets.connect(self.cdp_url, max_size=50 * 1024 * 1024), timeout=10.0) except Exception as e: attempt += 1 if self._fail_start(e): return - logger.warning( - "CDP supervisor %s: connect failed (attempt %s): %s", - self.task_id, attempt, _redact_cdp_error_text(e), - ) + logger.warning("CDP supervisor %s: connect failed (attempt %s): %s", + self.task_id, attempt, _redact_cdp_error_text(e)) await asyncio.sleep(min(backoff, 10.0)) backoff = min(backoff * 2, 10.0) continue reader_task = asyncio.create_task(self._read_loop(), name="cdp-reader") try: - # Reset the per-connection page session id. ``_pending_dialogs`` - # and ``_frames`` are deliberately kept — they reconcile as fresh - # events arrive; worst case a stale dialog entry is rejected - # with "no dialog is showing" (logged, not surfaced). + # Reset the per-connection page session id; ``_pending_dialogs`` / ``_frames`` + # are deliberately kept — they reconcile as fresh events arrive (worst case a + # stale dialog entry is rejected with "no dialog is showing", logged only). self._page_session_id = None await self._attach_initial_page() - with self._state_lock: - self._active = True + self._set_active(True) last_success_at = time.time() backoff = 0.5 # reset after a successful attach self._ready_event.set() @@ -459,19 +346,14 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): except BaseException as e: if self._fail_start(e): raise - logger.warning( - "CDP supervisor %s: session dropped after %.1fs: %s", - self.task_id, time.time() - last_success_at, _redact_cdp_error_text(e), - ) + logger.warning("CDP supervisor %s: session dropped after %.1fs: %s", + self.task_id, time.time() - last_success_at, _redact_cdp_error_text(e)) finally: - with self._state_lock: - self._active = False + self._set_active(False) if not reader_task.done(): reader_task.cancel() - try: + with contextlib.suppress(asyncio.CancelledError, Exception): await reader_task - except (asyncio.CancelledError, Exception): - pass for handle in self._dialog_watchdogs.values(): handle.cancel() self._dialog_watchdogs.clear() @@ -485,37 +367,23 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): async def _attach_initial_page(self) -> None: """Find (or create) a page target, attach flattened, enable domains, install dialog bridge.""" - resp = await self._cdp("Target.getTargets") - targets = resp.get("result", {}).get("targetInfos", []) + targets = (await self._cdp("Target.getTargets")).get("result", {}).get("targetInfos", []) page_target = next((t for t in targets if t.get("type") == "page"), None) if page_target is None: - created = await self._cdp("Target.createTarget", {"url": "about:blank"}) - target_id = created["result"]["targetId"] - else: - target_id = page_target["targetId"] - attach = await self._cdp("Target.attachToTarget", {"targetId": target_id, "flatten": True}) - self._page_session_id = attach["result"]["sessionId"] - await self._enable_page_domains(self._page_session_id, timeout=10.0) - await self._install_dialog_bridge(self._page_session_id) + page_target = (await self._cdp("Target.createTarget", {"url": "about:blank"}))["result"] + attach = await self._cdp("Target.attachToTarget", {"targetId": page_target["targetId"], "flatten": True}) + self._page_session_id = sid = attach["result"]["sessionId"] + await self._enable_page_domains(sid, timeout=10.0) + await self._install_dialog_bridge(sid) - async def _cdp( - self, - method: str, - params: Optional[Dict[str, Any]] = None, - *, - session_id: Optional[str] = None, - timeout: float = 10.0, - ) -> Dict[str, Any]: + async def _cdp(self, method: str, params: Optional[Dict[str, Any]] = None, *, + session_id: Optional[str] = None, timeout: float = 10.0) -> Dict[str, Any]: """Send a CDP command and await its response.""" if self._ws is None: raise RuntimeError("supervisor WebSocket is not connected") - call_id = self._next_call_id - self._next_call_id += 1 + call_id, self._next_call_id = self._next_call_id, self._next_call_id + 1 payload: Dict[str, Any] = {"id": call_id, "method": method} - if params: - payload["params"] = params - if session_id: - payload["sessionId"] = session_id + payload.update({k: v for k, v in (("params", params), ("sessionId", session_id)) if v}) fut: asyncio.Future = asyncio.get_running_loop().create_future() self._pending_calls[call_id] = fut await self._ws.send(json.dumps(payload)) @@ -538,17 +406,16 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): continue if "id" in msg: fut = self._pending_calls.pop(msg["id"], None) - if fut is not None and not fut.done(): - if "error" in msg: - fut.set_exception(RuntimeError(f"CDP error on id={msg['id']}: {msg['error']}")) - else: - fut.set_result(msg) - elif "method" in msg: - handler = self._EVENT_HANDLERS.get(msg["method"]) - if handler is not None: - result = handler(self, msg.get("params", {}), msg.get("sessionId")) - if result is not None: - await result + if fut is None or fut.done(): + continue + if "error" in msg: + fut.set_exception(RuntimeError(f"CDP error on id={msg['id']}: {msg['error']}")) + else: + fut.set_result(msg) + elif handler := self._EVENT_HANDLERS.get(msg.get("method")): + result = handler(self, msg.get("params", {}), msg.get("sessionId")) + if result is not None: + await result except Exception as e: logger.debug("CDP read loop exited: %s", e) @@ -559,56 +426,40 @@ class CDPSupervisor(DialogSupervisionMixin, FrameTrackingMixin): } -# ── Registry ───────────────────────────────────────────────────────────────── - - class _SupervisorRegistry: """Process-global (task_id → supervisor) map with idempotent start/stop. - - One instance, exposed as ``SUPERVISOR_REGISTRY``; mutations go through ``_lock``. - """ + One instance, exposed as ``SUPERVISOR_REGISTRY``; mutations go through ``_lock``.""" def __init__(self) -> None: self._lock = threading.Lock() self._by_task: Dict[str, CDPSupervisor] = {} def get(self, task_id: str) -> Optional[CDPSupervisor]: - """Return the supervisor for ``task_id`` if running, else ``None``.""" with self._lock: return self._by_task.get(task_id) - def get_or_start( - self, - task_id: str, - cdp_url: str, - *, - dialog_policy: str = DEFAULT_DIALOG_POLICY, - dialog_timeout_s: float = DEFAULT_DIALOG_TIMEOUT_S, - start_timeout: float = 15.0, - ) -> CDPSupervisor: - """Idempotently ensure a supervisor is running for ``(task_id, cdp_url)``. + def _pop(self, task_id: str) -> Optional[CDPSupervisor]: + with self._lock: + return self._by_task.pop(task_id, None) + def get_or_start(self, task_id: str, cdp_url: str, *, dialog_policy: str = DEFAULT_DIALOG_POLICY, + dialog_timeout_s: float = DEFAULT_DIALOG_TIMEOUT_S, start_timeout: float = 15.0) -> CDPSupervisor: + """Idempotently ensure a supervisor is running for ``(task_id, cdp_url)``. An existing supervisor bound to a different ``cdp_url`` (or unhealthy: - dead thread / stopped loop) is stopped and replaced. - """ + dead thread / stopped loop) is stopped and replaced.""" with self._lock: existing = self._by_task.get(task_id) if existing is not None: thread, loop = existing._thread, existing._loop - if ( - existing.cdp_url == cdp_url - and thread is not None and thread.is_alive() - and loop is not None and loop.is_running() - ): + healthy = thread is not None and thread.is_alive() and loop is not None and loop.is_running() + if existing.cdp_url == cdp_url and healthy: return existing self._by_task.pop(task_id, None) if existing is not None: existing.stop() - supervisor = CDPSupervisor( - task_id=task_id, cdp_url=cdp_url, - dialog_policy=dialog_policy, dialog_timeout_s=dialog_timeout_s, - ) + supervisor = CDPSupervisor(task_id=task_id, cdp_url=cdp_url, + dialog_policy=dialog_policy, dialog_timeout_s=dialog_timeout_s) supervisor.start(timeout=start_timeout) with self._lock: # Guard against a concurrent get_or_start from another thread. @@ -620,9 +471,7 @@ class _SupervisorRegistry: return supervisor def stop(self, task_id: str) -> None: - """Stop and discard the supervisor for ``task_id`` if it exists.""" - with self._lock: - supervisor = self._by_task.pop(task_id, None) + supervisor = self._pop(task_id) if supervisor is not None: supervisor.stop() @@ -639,16 +488,7 @@ SUPERVISOR_REGISTRY = _SupervisorRegistry() __all__ = [ - "CDPSupervisor", - "DEFAULT_DIALOG_POLICY", - "DEFAULT_DIALOG_TIMEOUT_S", - "DIALOG_POLICY_AUTO_ACCEPT", - "DIALOG_POLICY_AUTO_DISMISS", - "DIALOG_POLICY_MUST_RESPOND", - "DialogRecord", - "FrameInfo", - "PendingDialog", - "SUPERVISOR_REGISTRY", - "SupervisorSnapshot", - "_SupervisorRegistry", + "CDPSupervisor", "DEFAULT_DIALOG_POLICY", "DEFAULT_DIALOG_TIMEOUT_S", "DIALOG_POLICY_AUTO_ACCEPT", + "DIALOG_POLICY_AUTO_DISMISS", "DIALOG_POLICY_MUST_RESPOND", "DialogRecord", "FrameInfo", "PendingDialog", + "SUPERVISOR_REGISTRY", "SupervisorSnapshot", "_SupervisorRegistry", ] diff --git a/tools/browser_supervisor_dialogs.py b/tools/browser_supervisor_dialogs.py index cb825c509b..4150610e17 100644 --- a/tools/browser_supervisor_dialogs.py +++ b/tools/browser_supervisor_dialogs.py @@ -1,17 +1,15 @@ """Dialog capture + response half of the CDP supervisor. -Two capture paths feed the same ``PendingDialog`` queue: +Two capture paths feed the same ``PendingDialog`` queue: native +``Page.javascriptDialogOpening`` events (answered with +``Page.handleJavaScriptDialog``), and the injected *dialog bridge* — a page +script that rewrites alert/confirm/prompt into a sync XHR to a magic host we +intercept via the CDP ``Fetch`` domain and answer with ``Fetch.fulfillRequest``. +The bridge works on Browserbase, whose CDP proxy auto-dismisses real native +dialogs, because the native dialog never fires. -* native ``Page.javascriptDialogOpening`` events (answered with - ``Page.handleJavaScriptDialog``), and -* the injected *dialog bridge*: a page script that rewrites alert/confirm/prompt - into a sync XHR to a magic host we intercept via the CDP ``Fetch`` domain and - answer with ``Fetch.fulfillRequest``. Works on Browserbase, whose CDP proxy - auto-dismisses real native dialogs, because the native dialog never fires. - -``DialogSupervisionMixin`` is mixed into ``tools.browser_supervisor.CDPSupervisor`` -and relies on the state that class initialises (``_state_lock``, -``_pending_dialogs``, ``_recent_dialogs``, ``_dialog_watchdogs``, ``_cdp`` ...). +``DialogSupervisionMixin`` relies on state ``CDPSupervisor.__init__`` sets +(``_state_lock``, ``_pending_dialogs``, ``_recent_dialogs``, ``_dialog_watchdogs``, ``_cdp`` ...). """ from __future__ import annotations @@ -41,7 +39,13 @@ def _trim_ring(events: list, keep: int) -> list: return events[-keep:] if len(events) > keep * 2 else events -# ── Policy / config defaults ───────────────────────────────────────────────── +_REDACTED_FIELDS = frozenset({"message", "default_prompt"}) + + +def _dialog_dict(obj: Any, keys: tuple) -> Dict[str, Any]: + """Snapshot dict of ``keys`` with page-originated text fields redacted.""" + return {k: _redact_supervisor_text(getattr(obj, k)) if k in _REDACTED_FIELDS else getattr(obj, k) for k in keys} + DIALOG_POLICY_MUST_RESPOND = "must_respond" DIALOG_POLICY_AUTO_DISMISS = "auto_dismiss" @@ -110,9 +114,6 @@ _DIALOG_BRIDGE_SCRIPT = r""" """ -# ── Data model ──────────────────────────────────────────────────────────────── - - @dataclass class PendingDialog: """A JS dialog currently open on some frame's session.""" @@ -129,14 +130,7 @@ class PendingDialog: bridge_request_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: - return { - "id": self.id, - "type": self.type, - "message": _redact_supervisor_text(self.message), - "default_prompt": _redact_supervisor_text(self.default_prompt), - "opened_at": self.opened_at, - "frame_id": self.frame_id, - } + return _dialog_dict(self, ("id", "type", "message", "default_prompt", "opened_at", "frame_id")) @dataclass @@ -152,18 +146,7 @@ class DialogRecord: frame_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: - return { - "id": self.id, - "type": self.type, - "message": _redact_supervisor_text(self.message), - "opened_at": self.opened_at, - "closed_at": self.closed_at, - "closed_by": self.closed_by, - "frame_id": self.frame_id, - } - - -# ── Mixin ───────────────────────────────────────────────────────────────────── + return _dialog_dict(self, ("id", "type", "message", "opened_at", "closed_at", "closed_by", "frame_id")) class DialogSupervisionMixin: @@ -180,41 +163,30 @@ class DialogSupervisionMixin: async def _install_dialog_bridge(self, session_id: str) -> None: """Install the dialog-bridge init script + Fetch interceptor on a session. - The JS override runs in every frame before page scripts; Fetch.enable - scoped to the bridge URL catches the XHRs, which surface as pending - dialogs and are fulfilled when the agent responds. Idempotent at the CDP - level (Chromium de-dupes identical add-script calls; Fetch.enable - replaces prior patterns). The final Runtime.evaluate injects into the - already-loaded document so existing pages pick up the override on reconnect. + Idempotent at the CDP level (Chromium de-dupes identical add-script + calls; Fetch.enable replaces prior patterns). The final Runtime.evaluate + injects into the already-loaded document so existing pages pick up the + override on reconnect. """ sid = (session_id or "")[:16] - await self._cdp_quiet( - "Page.addScriptToEvaluateOnNewDocument", - {"source": _DIALOG_BRIDGE_SCRIPT, "runImmediately": True}, - session_id=session_id, timeout=5.0, what=f"dialog bridge sid={sid}", - ) - await self._cdp_quiet( - "Fetch.enable", - {"patterns": [{"urlPattern": DIALOG_BRIDGE_URL_PATTERN, "requestStage": "Request"}], - "handleAuthRequests": False}, - session_id=session_id, timeout=5.0, what=f"dialog bridge sid={sid}", - ) - await self._cdp_quiet( - "Runtime.evaluate", - {"expression": _DIALOG_BRIDGE_SCRIPT, "returnByValue": True}, - session_id=session_id, timeout=3.0, what=f"dialog bridge inject sid={sid}", + steps = ( + ("Page.addScriptToEvaluateOnNewDocument", {"source": _DIALOG_BRIDGE_SCRIPT, "runImmediately": True}, + 5.0, f"dialog bridge sid={sid}"), + ("Fetch.enable", {"patterns": [{"urlPattern": DIALOG_BRIDGE_URL_PATTERN, "requestStage": "Request"}], + "handleAuthRequests": False}, 5.0, f"dialog bridge sid={sid}"), + ("Runtime.evaluate", {"expression": _DIALOG_BRIDGE_SCRIPT, "returnByValue": True}, + 3.0, f"dialog bridge inject sid={sid}"), ) + for method, params, timeout, what in steps: + await self._cdp_quiet(method, params, session_id=session_id, timeout=timeout, what=what) # ── Capture ────────────────────────────────────────────────────────────── async def _on_dialog_opening(self, params: Dict[str, Any], session_id: Optional[str]) -> None: - self._admit_dialog(self._new_dialog( - type=str(params.get("type") or ""), - message=str(params.get("message") or ""), - default_prompt=str(params.get("defaultPrompt") or ""), - session_id=session_id, - frame_id=params.get("frameId"), - )) + self._admit_dialog( + type=str(params.get("type") or ""), message=str(params.get("message") or ""), + default_prompt=str(params.get("defaultPrompt") or ""), session_id=session_id, frame_id=params.get("frameId"), + ) async def _on_fetch_paused(self, params: Dict[str, Any], session_id: Optional[str]) -> None: """Bridge XHR captured mid-flight — materialize as a pending dialog. @@ -231,40 +203,28 @@ class DialogSupervisionMixin: await self._cdp_quiet("Fetch.continueRequest", {"requestId": request_id}, session_id=session_id, timeout=3.0, what="passthrough") return - q = parse_qs(urlparse(url).query) - self._admit_dialog(self._new_dialog( - type=q.get("kind", [""])[0] or "alert", - message=q.get("message", [""])[0], - default_prompt=q.get("default_prompt", [""])[0], - session_id=session_id, - frame_id=params.get("frameId"), - bridge_request_id=str(request_id), - )) - - def _new_dialog(self, *, type: str, message: str, default_prompt: str, session_id: Optional[str], - frame_id: Optional[str], bridge_request_id: Optional[str] = None) -> PendingDialog: - self._dialog_seq += 1 - return PendingDialog( - id=f"d-{self._dialog_seq}", - type=type, - message=message, - default_prompt=default_prompt, - opened_at=time.time(), - cdp_session_id=session_id or self._page_session_id or "", - frame_id=frame_id, - bridge_request_id=bridge_request_id, + q = {k: v[0] for k, v in parse_qs(urlparse(url).query).items()} + self._admit_dialog( + type=q.get("kind") or "alert", message=q.get("message", ""), default_prompt=q.get("default_prompt", ""), + session_id=session_id, frame_id=params.get("frameId"), bridge_request_id=str(request_id), ) - def _admit_dialog(self, dialog: PendingDialog) -> None: - """Apply the dialog policy: auto-respond, or queue + arm the watchdog. + def _admit_dialog(self, *, type: str, message: str, default_prompt: str, session_id: Optional[str], + frame_id: Optional[str], bridge_request_id: Optional[str] = None) -> None: + """Create the dialog and apply the policy: auto-respond, or queue + arm the watchdog. Auto policies archive FIRST (tagged ``auto_policy``) so the ``closed`` event that follows our own response isn't re-archived as ``remote``. """ - auto = { - DIALOG_POLICY_AUTO_DISMISS: (False, ""), - DIALOG_POLICY_AUTO_ACCEPT: (True, dialog.default_prompt), - }.get(self.dialog_policy) + self._dialog_seq += 1 + dialog = PendingDialog( + id=f"d-{self._dialog_seq}", type=type, message=message, default_prompt=default_prompt, + opened_at=time.time(), cdp_session_id=session_id or self._page_session_id or "", + frame_id=frame_id, bridge_request_id=bridge_request_id, + ) + auto = {DIALOG_POLICY_AUTO_DISMISS: (False, ""), DIALOG_POLICY_AUTO_ACCEPT: (True, default_prompt)}.get( + self.dialog_policy + ) if auto is not None: with self._state_lock: self._archive_dialog_locked(dialog, "auto_policy") @@ -280,37 +240,26 @@ class DialogSupervisionMixin: # ── Responding ─────────────────────────────────────────────────────────── async def _respond(self, dialog: PendingDialog, *, accept: bool, prompt_text: Optional[str]) -> None: - """Bridge-fulfill for XHR-captured dialogs, else native CDP. - - Native path sends ``promptText`` only for prompt dialogs when - ``prompt_text`` is given, and raises on CDP failure; the bridge path - (Fetch.fulfillRequest so the page unblocks) swallows failures. - """ + """Bridge-fulfill for XHR-captured dialogs (swallows failures so the page + unblocks), else native CDP — ``promptText`` only for prompt dialogs when + given; raises on CDP failure.""" + session_id = dialog.cdp_session_id or None if dialog.bridge_request_id: - body = json.dumps({ - "accept": bool(accept), - "prompt_text": (prompt_text or "") if dialog.type == "prompt" else "", - "dialog_id": dialog.id, - }).encode() + body = json.dumps({"accept": bool(accept), "dialog_id": dialog.id, + "prompt_text": (prompt_text or "") if dialog.type == "prompt" else ""}).encode() await self._cdp_quiet( "Fetch.fulfillRequest", - { - "requestId": dialog.bridge_request_id, - "responseCode": 200, - "responseHeaders": [ - {"name": "Content-Type", "value": "application/json"}, - {"name": "Access-Control-Allow-Origin", "value": "*"}, - ], - "body": base64.b64encode(body).decode(), - }, - session_id=dialog.cdp_session_id or None, timeout=5.0, what=f"bridge fulfill {dialog.id}", + {"requestId": dialog.bridge_request_id, "responseCode": 200, + "responseHeaders": [{"name": "Content-Type", "value": "application/json"}, + {"name": "Access-Control-Allow-Origin", "value": "*"}], + "body": base64.b64encode(body).decode()}, + session_id=session_id, timeout=5.0, what=f"bridge fulfill {dialog.id}", ) return params: Dict[str, Any] = {"accept": accept} if prompt_text is not None and dialog.type == "prompt": params["promptText"] = prompt_text - await self._cdp("Page.handleJavaScriptDialog", params, - session_id=dialog.cdp_session_id or None, timeout=5.0) + await self._cdp("Page.handleJavaScriptDialog", params, session_id=session_id, timeout=5.0) async def _respond_quiet(self, dialog: PendingDialog, *, accept: bool, prompt_text: Optional[str]) -> None: """Auto-policy / watchdog response (already archived by the caller); failures logged only.""" @@ -320,11 +269,8 @@ class DialogSupervisionMixin: logger.debug("auto response failed for %s: %s", dialog.id, e) async def _handle_dialog_cdp(self, dialog: PendingDialog, *, accept: bool, prompt_text: str) -> None: - """Agent response path. - - The dialog is retired regardless of outcome — a CDP error usually means - it already closed (browser auto-dismissed after navigation, etc.). - """ + """Agent response path. The dialog is retired regardless of outcome — a CDP + error usually means it already closed (browser auto-dismissed after navigation).""" try: await self._respond(dialog, accept=accept, prompt_text=prompt_text) finally: @@ -335,14 +281,10 @@ class DialogSupervisionMixin: dialog = self._pending_dialogs.get(dialog_id) if dialog is None: return - logger.warning( - "CDP supervisor %s: dialog %s (%s) auto-dismissed after %ss timeout", - self.task_id, dialog_id, dialog.type, self.dialog_timeout_s, - ) + logger.warning("CDP supervisor %s: dialog %s (%s) auto-dismissed after %ss timeout", + self.task_id, dialog_id, dialog.type, self.dialog_timeout_s) # Archive with watchdog tag BEFORE unblocking the page. - with self._state_lock: - if self._pending_dialogs.pop(dialog_id, None) is not None: - self._archive_dialog_locked(dialog, "watchdog") + self._retire_dialog(dialog_id, "watchdog") await self._respond_quiet(dialog, accept=False, prompt_text=None) # ── Bookkeeping ────────────────────────────────────────────────────────── @@ -359,11 +301,9 @@ class DialogSupervisionMixin: def _archive_dialog_locked(self, dialog: PendingDialog, closed_by: str) -> None: """Move a pending dialog to the recent_dialogs ring buffer. Must hold state_lock.""" - self._recent_dialogs.append(DialogRecord( - id=dialog.id, type=dialog.type, message=dialog.message, opened_at=dialog.opened_at, - closed_at=time.time(), closed_by=closed_by, frame_id=dialog.frame_id, - )) - self._recent_dialogs = _trim_ring(self._recent_dialogs, RECENT_DIALOGS_MAX) + record = DialogRecord(id=dialog.id, type=dialog.type, message=dialog.message, opened_at=dialog.opened_at, + closed_at=time.time(), closed_by=closed_by, frame_id=dialog.frame_id) + self._recent_dialogs = _trim_ring([*self._recent_dialogs, record], RECENT_DIALOGS_MAX) async def _on_dialog_closed(self, params: Dict[str, Any], session_id: Optional[str]) -> None: # ``Page.javascriptDialogClosed`` carries only ``result``/``userInput``, not @@ -371,12 +311,10 @@ class DialogSupervisionMixin: # it — the JS thread blocks while a dialog is up, so at most one is in # flight per session. Bridge dialogs resolve via Fetch.fulfillRequest. with self._state_lock: - candidate_ids = [ - d.id for d in self._pending_dialogs.values() - if d.cdp_session_id == session_id and d.bridge_request_id is None - ] - if candidate_ids: - self._retire_dialog(candidate_ids[0], "remote") + candidate = next((d.id for d in self._pending_dialogs.values() + if d.cdp_session_id == session_id and d.bridge_request_id is None), None) + if candidate: + self._retire_dialog(candidate, "remote") # CDP event → handler(self, params, session_id); merged into CDPSupervisor._EVENT_HANDLERS. EVENT_HANDLERS: Dict[str, Callable[..., Any]] = { diff --git a/tools/browser_supervisor_frames.py b/tools/browser_supervisor_frames.py index 482c76d9c8..5832b8b0fc 100644 --- a/tools/browser_supervisor_frames.py +++ b/tools/browser_supervisor_frames.py @@ -25,12 +25,9 @@ _AUTO_ATTACH_PARAMS = {"autoAttach": True, "waitForDebuggerOnStart": False, "fla @dataclass class FrameInfo: - """One frame in the page's frame tree. - - ``is_oopif`` frames have their own CDP target (reachable via - ``cdp_session_id``); same-origin / srcdoc iframes share the parent process - and have ``is_oopif=False`` + ``cdp_session_id=None``. - """ + """One frame in the page's frame tree. ``is_oopif`` frames have their own CDP + target (reachable via ``cdp_session_id``); same-origin / srcdoc iframes share + the parent process (``is_oopif=False``, ``cdp_session_id=None``).""" frame_id: str url: str @@ -42,12 +39,10 @@ class FrameInfo: def to_dict(self) -> Dict[str, Any]: d = {"frame_id": self.frame_id, "url": self.url, "origin": self.origin, "is_oopif": self.is_oopif} - if self.cdp_session_id: - d["session_id"] = self.cdp_session_id - if self.parent_frame_id: - d["parent_frame_id"] = self.parent_frame_id - if self.name: - d["name"] = self.name + for key, value in (("session_id", self.cdp_session_id), ("parent_frame_id", self.parent_frame_id), + ("name", self.name)): + if value: + d[key] = value return d @@ -62,13 +57,13 @@ class FrameTrackingMixin: def _on_frame_attached(self, params: Dict[str, Any], session_id: Optional[str]) -> None: frame_id = params.get("frameId") - if not frame_id: - return + if frame_id: + self._set_frame(FrameInfo(frame_id=frame_id, url="", origin="", parent_frame_id=params.get("parentFrameId"), + is_oopif=False, cdp_session_id=session_id)) + + def _set_frame(self, frame: FrameInfo) -> None: with self._state_lock: - self._frames[frame_id] = FrameInfo( - frame_id=frame_id, url="", origin="", parent_frame_id=params.get("parentFrameId"), - is_oopif=False, cdp_session_id=session_id, - ) + self._frames[frame.frame_id] = frame def _on_frame_navigated(self, params: Dict[str, Any], session_id: Optional[str]) -> None: frame = params.get("frame") or {} @@ -78,8 +73,7 @@ class FrameTrackingMixin: with self._state_lock: old = self._frames.get(frame_id) self._frames[frame_id] = FrameInfo( - frame_id=frame_id, - url=str(frame.get("url") or ""), + frame_id=frame_id, url=str(frame.get("url") or ""), origin=str(frame.get("securityOrigin") or frame.get("origin") or ""), parent_frame_id=frame.get("parentId") or (old.parent_frame_id if old else None), is_oopif=bool(old.is_oopif if old else False), @@ -91,10 +85,10 @@ class FrameTrackingMixin: """Drop a frame only when it's truly gone. ``reason="swap"`` means the frame is migrating processes (e.g. promoted - to an OOPIF) — dropping it would hide the iframe, so it's a no-op. Even - with ``reason="remove"`` the parent only knows the child left ITS - process; if we hold a live child session for that frame_id it is still - alive, so keep it until Target.detached + a later frameDetached clear it. + to an OOPIF) — dropping it would hide the iframe. Even with ``remove`` + the parent only knows the child left ITS process; if we hold a live + child session for that frame_id it is still alive, so keep it until + Target.detached + a later frameDetached clear it. """ frame_id = params.get("frameId") if not frame_id or str(params.get("reason") or "remove").lower() == "swap": @@ -118,9 +112,8 @@ class FrameTrackingMixin: with self._state_lock: old = self._frames.get(target_id) self._frames[target_id] = FrameInfo( - frame_id=target_id, url=str(info.get("url") or ""), origin="", - parent_frame_id=(old.parent_frame_id if old else None), is_oopif=True, - cdp_session_id=sid, name=str(info.get("title") or (old.name if old else "")), + frame_id=target_id, url=str(info.get("url") or ""), origin="", is_oopif=True, cdp_session_id=sid, + parent_frame_id=(old.parent_frame_id if old else None), name=str(info.get("title") or (old.name if old else "")), ) # Enable child domains off-loop: awaiting the replies here would deadlock # because only the reader can resolve those Futures. @@ -138,8 +131,7 @@ class FrameTrackingMixin: """Clear the session binding of frames on a detached child session. Frames are deliberately NOT dropped: Browserbase fires transient detaches - during page transitions while the iframe is still visible, and dropping - would hide OOPIFs until the next ``Target.attachedToTarget``. Clearing + during page transitions while the iframe is still visible. Clearing ``cdp_session_id`` just stops stale routing; ``Page.frameDetached`` cleans up if the iframe truly goes away. """ @@ -152,12 +144,9 @@ class FrameTrackingMixin: self._frames[fid] = replace(frame, cdp_session_id=None) def _build_frame_tree_locked(self) -> Dict[str, Any]: - """Build the capped frame_tree payload. Must be called under state lock. - - Top frame = one with no parent, preferring oopif=False. BFS from it, - capped by FRAME_TREE_MAX_ENTRIES and FRAME_TREE_MAX_OOPIF_DEPTH for - OOPIF branches. - """ + """Capped frame_tree payload (must hold state lock). Top frame = one with + no parent, preferring oopif=False; BFS from it, capped by + FRAME_TREE_MAX_ENTRIES and FRAME_TREE_MAX_OOPIF_DEPTH for OOPIF branches.""" frames = self._frames tops = [f for f in frames.values() if not f.parent_frame_id] top = next((f for f in tops if not f.is_oopif), tops[0] if tops else None) @@ -167,7 +156,7 @@ class FrameTrackingMixin: children: List[Dict[str, Any]] = [] truncated = False queue: List[Tuple[FrameInfo, int]] = [(f, 1) for f in frames.values() if f.parent_frame_id == top.frame_id] - visited: set[str] = {top.frame_id} + visited = {top.frame_id} while queue and len(children) < FRAME_TREE_MAX_ENTRIES: frame, depth = queue.pop(0) if frame.frame_id in visited: