refactor(tools): compact bot-mode DM/relay/probe and browser supervisor modules

This commit is contained in:
Teknium
2026-09-02 22:26:43 -07:00
parent 113f04616b
commit 88d1423e99
9 changed files with 629 additions and 1391 deletions
+39 -122
View File
@@ -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
+122 -317
View File
@@ -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 <name> chat --in ~ -c
"Bot Chat" --create-if-missing -Q --query-file <tmp>`` (one turn, reply on
stdout); peer → ``hermes peer dm <peer>[/<name>] < <tmp>``. 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 <name> chat --in ~ -c "Bot Chat" --create-if-missing -Q
--query-file <tmp>``; peer → ``hermes peer dm <peer>[/<name>] < <tmp>``; 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: '<peer>/<agent>' 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 <profile> 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 ""
+107 -153
View File
@@ -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:
+117 -285
View File
@@ -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 ``<root>/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 ``<root>/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.<key>`` 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;
- ``<handle>@<connection-id>`` / ``<profile>@<connection-id>`` — 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
``<handle|profile>@<connection-id>``; ``"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 ``<root>/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 ``<root>/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)
+2 -10
View File
@@ -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="💬",
)
+16 -45
View File
@@ -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,
)
+126 -286
View File
@@ -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": <value>, "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",
]
+75 -137
View File
@@ -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]] = {
+25 -36
View File
@@ -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: