"""Bot Mode roster probe — canonical Bot Chat system prompt section. When any profile carries ``ui_meta['hermes-bots']`` in profile.yaml (Bot-Mode-managed), a bot's canonical "Bot Chat" session — ONLY that session (agent/system_prompt.py enforces the ``BOT_CHAT_TITLE`` gate) — gets a "Messaging other agents" section. Silent (``""``) when no profile is managed or on any error. Older desktop builds appended a frozen copy of the section to SOUL.md; ``strip_legacy_protocol`` drops it at load time so the live roster here is the only copy any session sees. Cached per (process, home) so compression rebuilds produce identical bytes. Toggle: ``agent.bot_mode_protocol``. Also hosts path/roster helpers shared by ``bot_mode_dm`` and ``bot_relay``. """ from __future__ import annotations import os import re import threading from pathlib import Path _PROTOCOL_HEADING = "## Messaging other agents" # The legacy section through the next H2 heading (or EOF), plus the blank lines before it. _LEGACY_PROTOCOL_RE = re.compile(r"\n*" + re.escape(_PROTOCOL_HEADING) + r"[ \t]*\n.*?(?=\n## |\Z)", re.S) def strip_legacy_protocol(text: str) -> str: """SOUL text without the plugin-era "## Messaging other agents" section (idempotent).""" return _LEGACY_PROTOCOL_RE.sub("\n", text).strip() + "\n" if _PROTOCOL_HEADING in text else text # The only session title that receives the protocol section. Must match the # desktop plugin's createCanonicalChat title and the `-c "Bot Chat"` resume target. BOT_CHAT_TITLE = "Bot Chat" _lock = threading.Lock() _cached: dict[str, str] = {} # ── shared path / roster helpers ───────────────────────────────────────────── def _default_home() -> str: """Ambient process HERMES_HOME (env, else the platform default) as a string.""" from hermes_constants import get_process_hermes_home return str(get_process_hermes_home()) 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 def _profile_name(home: Path) -> str: return home.name if home.parent.name == "profiles" else "default" def _handle(name: str) -> str: # The mention middleware aliases the default profile as @hermes. return "hermes" if name == "default" else name def _roster(root: Path) -> list[tuple[str, Path]]: """(name, dir) for the default profile + every live named profile, sorted. Same identity predicate as ``profile list``: infra dirs (``sessions/``, ``logs/``) and tombstones are not teammates (#99392), and neither is a marker-carrying dir whose name is not a profile id — a parked backup or staging dir must never become a ``message_agent`` target (#116905).""" from hermes_constants import PROFILE_ID_RE, named_profile_is_live profiles = root / "profiles" named = _swallow( lambda: [ (c.name, c) for c in sorted(profiles.iterdir()) if c.name != "default" and PROFILE_ID_RE.match(c.name) and named_profile_is_live(c) ] if profiles.is_dir() else [], [], ) return [("default", root), *named] def _read_yaml_dict(path: Path, needle: str | None = None) -> dict | None: """YAML mapping at ``path``, or None when missing / not a mapping / unreadable. ``needle``: cheap substring precheck that skips the YAML parse on the dominant (unmanaged) path.""" def _load(): if not path.is_file(): return None raw = path.read_text(encoding="utf-8-sig", errors="replace") if needle is not None and needle not in raw: return None import hermes_yaml as yaml data = yaml.safe_load(raw) return data if isinstance(data, dict) else None return _swallow(_load, None) def _bots_meta(data: dict | None) -> dict | None: """The ``ui_meta['hermes-bots']`` block of a parsed profile.yaml, if a dict.""" ui_meta = data.get("ui_meta") if data else None bots = ui_meta.get("hermes-bots") if isinstance(ui_meta, dict) else None return bots if isinstance(bots, dict) else None def _is_bot_managed(profile_dir: Path) -> bool: 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. Never raises. The ``message_agent`` injection gate — deliberately independent of the protocol section's emptiness: a SOUL.md carrying the legacy protocol gets an empty section but still gets the tool.""" return _swallow(lambda: _any_managed(_hermes_root(_resolve_home(home))), 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; tells a teammate WHO to message for a job. A friendly ``display_name`` (``hermes profile rename``) that differs from both the folder id and the title leads the line, so an untagged "talk to Scribe" maps to the folder handle without a disk search (#100671). Single-line, ≤160 chars, "" when nothing. Never raises.""" def _role() -> str: data = _read_yaml_dict(profile_dir / "profile.yaml") or {} title = str((_bots_meta(data) or {}).get("title") or "").strip() display = str(data.get("display_name") or "").strip() if display.lower() in (profile_dir.name.lower(), title.lower()): display = "" line = _role_line(display, title, str(data.get("description") or "").strip()) return " ".join(line.split())[:160] return _swallow(_role, "") def _friendly_names(profile_dir: Path) -> tuple[str, str]: """(Bot Mode title, profile.yaml ``display_name``) for a profile, "" when unset. Never raises.""" def _read() -> tuple[str, str]: data = _read_yaml_dict(profile_dir / "profile.yaml") or {} return (str((_bots_meta(data) or {}).get("title") or "").strip(), str(data.get("display_name") or "").strip()) return _swallow(_read, ("", "")) def _display_name(name: str, profile_dir: Path) -> str: """Human-facing sender name, in the Desktop's ``botFriendlyNames`` order: Bot Mode title, then profile.yaml ``display_name`` (``hermes profile rename``), else the @handle — the renamed primary signs as ``Maia (@hermes)``, not ``hermes (@hermes)`` (#89720).""" return next((n for n in _friendly_names(profile_dir) if n), None) or _handle(name) # Tokens the Desktop mention parser reserves; a bot titled "Hermes" never hijacks @hermes. _RESERVED_ALIASES = frozenset({"all", "everyone", "user", "default", "hermes"}) def alias_forms(value: str) -> set[str]: """Lower-cased mention forms of a friendly name, mirroring the Desktop's ``mentionNameForms``: slugified (``"Dr. Foo"`` → ``dr-foo``, what autocomplete inserts) and collapsed (``drfoo``). Reserved tokens and empty forms are dropped.""" name = str(value or "").strip().lower() slug = re.sub(r"[^a-z0-9_-]+", "-", name).strip("-") collapsed = re.sub(r"[^a-z0-9_-]+", "", name) return {f for f in (slug, collapsed) if f and re.fullmatch(r"[a-z0-9][a-z0-9_-]*", f) and f not in _RESERVED_ALIASES} def local_alias_map(root: Path) -> dict[str, set[str]]: """``alias form → {folder ids}`` for every local profile's friendly names (profile.yaml ``display_name`` and the Bot Mode title). Folder ids themselves are not aliases: the caller matches those first, so a target that is an exact folder id always addresses that folder — a friendly name colliding with ANOTHER folder id never steals it. Ambiguity (one alias form shared by several profiles) surfaces as a multi-id set. Never raises.""" def _build() -> dict[str, set[str]]: aliases: dict[str, set[str]] = {} for name, profile_dir in _roster(root): for form in set().union(*(alias_forms(f) for f in _friendly_names(profile_dir))): aliases.setdefault(form, set()).add(name) return aliases return _swallow(_build, {}) def _peers(root: Path) -> list[str]: """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 [] 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 local_taken_forms(root: Path) -> set[str]: """Bare forms this gateway's own profiles answer to (handles + friendly-name slugs); a remote row must not be offered under any of them, since local resolution wins (``_resolve_local_name``).""" return {_handle(name) for name, _d in _roster(root)} | set(local_alias_map(root)) def _remote_paragraph(root: Path) -> str: """Addendum for agents on OTHER connected machines; only when the relay roster is non-empty.""" roster = _remote_roster(root) if not roster: return "" 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, local_taken_forms(root))) ] return ( "\n\nTeammates on OTHER connected machines (reachable through the " "Desktop relay — message them with message_agent exactly like local " "teammates; replies arrive as completion notifications the same " "way, or via reply_delivery=\"poll\" as below):\n" + "\n".join(lines) ) def _peer_paragraph(root: Path) -> str: """Addendum for cross-machine DMs — only when peers exist.""" peers = _peers(root) if not peers: return "" listed = ", ".join(f"`{p}`" for p in peers) return ( "\n\nTeammates on OTHER machines: this install also has peer gateways " f"registered ({listed}). Message an agent on a peer the same way — " 'message_agent with target "/" (or "" alone ' "for the peer's main agent). Run `hermes peer list` for the live " "peer list." ) def _build_section(home: Path) -> str: root = _hermes_root(home) me = _profile_name(home) if not _any_managed(root): return "" 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" "This install runs Bot Mode: each Hermes profile is an agent teammate with " 'one canonical "Bot Chat" conversation, and you have the `message_agent` ' "tool to DM any of them. It is FIRE-AND-FORGET: it delivers your message " "with your attribution prefixed automatically and returns an acknowledgement " "immediately — it never returns the reply. Send it, finish your turn, and " "the reply arrives later as a background-process completion notification " "that wakes you; relay it to the user then, attributed to that agent — unless " "the ack returns reply_delivery=\"poll\", in which case follow its " "process(action=\"wait\") instruction before ending the turn. " "COMPOSE every message yourself — say what YOU need from that agent; never " "forward the user's words verbatim, and never reveal private 1:1 chat " "content. When the user says \"ask \" or \"tell ...\", that is " "a handoff: pick the right teammate from the roster below, message them " "with message_agent, and report back naming which agent replied. Message " "ONE clearly relevant teammate; don't fan out to several unless the user " "explicitly asked.\n" f'When YOU receive a "Message from 🤖 (@):" message, a ' "teammate agent is talking to you (not the user): address them, reply " "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(me)}`. Your teammates (live roster; roles from their " "profiles):\n" f"{roster_block}" + _remote_paragraph(root) + _peer_paragraph(root) ) def get_bot_mode_protocol_section(home: str | os.PathLike | None = None, *, force_refresh: bool = False) -> str: """Cached probe entry point — one filesystem pass per (process, home). ``home`` should be the AGENT'S OWN resolved home (session-db derived), not ambient HERMES_HOME — build threads can lose the ContextVar override and the env var would then name the wrong profile.""" resolved = str(_resolve_home(home)) with _lock: if force_refresh or resolved not in _cached: _cached[resolved] = _swallow(lambda: _build_section(Path(resolved)), "") return _cached[resolved] # ── capability epoch ───────────────────────────────────────────────────────── # Bot Chat sessions are effectively eternal, so "build the prompt once" would strand # capability changes (skills, toolsets, MCP, SOUL, roster, peers, model capability # overrides that change the prompt) forever. The fingerprint hashes exactly that # surface; the built prompt embeds it and agent/conversation_loop.py rebuilds only # when the stored epoch differs from disk — once per change, never per-turn drift. _EPOCH_PREFIX = "Capability epoch: " _EPOCH_RE_TEXT = r"Capability epoch: ([0-9a-f]{12})" def _model_prompt_capability_surface(model_cfg: object) -> dict: """``model.*`` overrides whose flip changes a rebuilt prompt, coerced like their consumers. ``supports_vision`` uses image routing's strict bool so YAML ``yes`` and ``true`` share one epoch. ``context_length`` is the cap ``build_system_prompt_parts`` uses to truncate context files. Routing keys (provider, default, base_url) are identity lines, not this surface — and no model id is special-cased. """ from agent.image_routing import _coerce_capability_bool if not isinstance(model_cfg, dict): model_cfg = {} raw_ctx = model_cfg.get("context_length") ctx = None # bool is an int subclass; ``context_length: true`` is not a window. if isinstance(raw_ctx, bool): ctx = None elif isinstance(raw_ctx, int): ctx = raw_ctx if raw_ctx > 0 else None elif isinstance(raw_ctx, str) and raw_ctx.strip().isdigit(): parsed = int(raw_ctx.strip()) ctx = parsed if parsed > 0 else None return { "supports_vision": _coerce_capability_bool(model_cfg.get("supports_vision")), "context_length": ctx, } def capability_fingerprint(home: str | os.PathLike | None = None) -> str: """12-hex digest of the capability surface for ``home``'s profile: disabled skills + enabled toolsets + MCP config, model capability overrides that change the prompt (``supports_vision``, ``context_length``), SOUL.md bytes, installed skill names, the Bot-Mode roster (+ roles), peers and the relay roster. Deliberately NOT cached — the point is detecting on-disk drift against a stored prompt's epoch. Never raises ("unavailable" on failure).""" import hashlib import json resolved = _resolve_home(home) root = _hermes_root(resolved) surface: dict = {} try: # Canonical loader (managed overlay + env expansion + normalization), # scoped to the bot's home via the override the loaders already honor. from agent.skill_utils import parse_config_string_list from hermes_cli.config import load_config_readonly from hermes_constants import reset_hermes_home_override, set_hermes_home_override token = set_hermes_home_override(str(resolved)) try: cfg = load_config_readonly() or {} finally: reset_hermes_home_override(token) skills_cfg = cfg.get("skills") if isinstance(cfg.get("skills"), dict) else {} model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {} surface["model_capabilities"] = _model_prompt_capability_surface(model_cfg) surface["disabled_skills"] = sorted(str(s).lower() for s in (skills_cfg.get("disabled") or [])) # The live selection is platform_toolsets. minus agent.disabled_toolsets; # tools.enabled_toolsets is written by no surface, so watching it left Bot Chats # blind to `hermes tools enable/disable` (#124211). Raw slices: an edit that leaves # the effective selection unchanged costs one spurious rebuild at most. agent_cfg = cfg.get("agent") if isinstance(cfg.get("agent"), dict) else {} surface["platform_toolsets"] = json.dumps(cfg.get("platform_toolsets") or {}, sort_keys=True, default=str) surface["disabled_toolsets"] = sorted(parse_config_string_list(agent_cfg.get("disabled_toolsets"))) mcp = cfg.get("mcp_servers") surface["mcp"] = json.dumps(mcp, sort_keys=True, default=str) if isinstance(mcp, dict) else "" except Exception: pass def _soul() -> str: soul = resolved / "SOUL.md" return hashlib.sha256(soul.read_bytes()).hexdigest() if soul.is_file() else "" def _skills() -> list[str]: skills_root = resolved / "skills" if not skills_root.is_dir(): return [] # The same walk every other reader of this tree uses (skills_list/skill_view, the prompt's # skills index, skill_count): it prunes EXCLUDED_SKILL_DIRS — ``.archive``, ``.curator_backups``, # ``node_modules`` … — and each skill's support dirs. A raw ``**/SKILL.md`` glob counted files # the model can never invoke, so archiving a skill, or the curator writing a backup, flipped the # epoch and forced every Bot Chat to rebuild a system prompt whose skills index had not changed. # iter_skill_index_files is also org-token-gated, so the epoch moves on an org switch as well — # intended: a different org sees a different skills index, so it needs a different prompt. from agent.skill_utils import iter_skill_index_files return sorted(str(p.parent.relative_to(skills_root)) for p in iter_skill_index_files(skills_root, "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)) # Roles are part of the messaging surface: renaming a bot or editing a # description must refresh the roster block teammates pick recipients from. surface["roster_roles"] = sorted(f"{n}:{_profile_role(d)}" for n, d in roster) except Exception: surface["roster"] = [] # Protocol-text version salt: bumping it refreshes every eternal Bot Chat # 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. 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: """The epoch stamp appended to a Bot Chat prompt.""" return f"{_EPOCH_PREFIX}{capability_fingerprint(home)}" def stored_prompt_capability_stale(stored_prompt: str, home: str | os.PathLike | None = None) -> bool: """True when ``stored_prompt`` is a Bot Chat prompt whose embedded epoch no longer matches disk. Unstamped prompts are never stale. Fails closed to "not stale" — a broken probe must not become a rebuild-every-turn cache burner.""" import re 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: """True when a Bot Chat session's stored prompt PREDATES the epoch mechanism (no stamp — including SOUL-era prompts whose frozen roster rode in from SOUL.md). The caller must only ask for sessions titled "Bot Chat"; we rebuild only when the probe would actually emit a section, and every rebuilt prompt is stamped so this fires once. Fails closed.""" if _EPOCH_PREFIX in (stored_prompt or ""): return False return _swallow(lambda: bool(get_bot_mode_protocol_section(home)), False) def _reset_cache_for_tests() -> None: with _lock: _cached.clear()