From 11889a28379c19fa2cd9eecb0b3c7a1bc082e961 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:40:00 -0700 Subject: [PATCH] =?UTF-8?q?refactor(hermes=5Fcli):=20skin=5Fengine/slack?= =?UTF-8?q?=5Fcli/slash=5Fexec/telegram=5Fmanaged=5Fbot/suggestions=5Fcmd/?= =?UTF-8?q?skills=5Fconfig/skin=5Fcmd=20=E2=80=94=20collapse=20mapping/nor?= =?UTF-8?q?malization=20helpers,=20pack=20data=20tables,=20compact=20docst?= =?UTF-8?q?rings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hermes_cli/skills_config.py | 109 ++---- hermes_cli/skin_cmd.py | 23 +- hermes_cli/skin_engine.py | 607 ++++++++--------------------- hermes_cli/slack_cli.py | 243 ++++-------- hermes_cli/slash_exec.py | 106 ++--- hermes_cli/suggestions_cmd.py | 84 ++-- hermes_cli/telegram_managed_bot.py | 73 +--- 7 files changed, 341 insertions(+), 904 deletions(-) diff --git a/hermes_cli/skills_config.py b/hermes_cli/skills_config.py index 8aef1e4293..83e6dd882d 100644 --- a/hermes_cli/skills_config.py +++ b/hermes_cli/skills_config.py @@ -5,19 +5,13 @@ from hermes_cli.config import cfg_get, load_config, save_config from hermes_cli.colors import Colors, color from hermes_cli.platforms import PLATFORMS as _PLATFORMS -# Backward-compatible view: {key: label_string} so existing code that -# iterates ``PLATFORMS.items()`` or calls ``PLATFORMS.get(key)`` keeps -# working without changes to every call site. +# Backward-compatible {key: label} view for callers using ``PLATFORMS.items()`` / ``.get(key)``. PLATFORMS = {k: info.label for k, info in _PLATFORMS.items() if k != "api_server"} -# ─── Config Helpers ─────────────────────────────────────────────────────────── def _normalize_skill_names(values) -> Set[str]: - """Normalize a config value into a set of skill names. - - Mirrors ``agent.skill_utils._normalize_string_set``: ``None`` (YAML null) means empty, and a - bare scalar (``disabled: my-skill``) means a single-item list — NOT a set of its characters. - """ + """Config value -> set of skill names (mirrors ``agent.skill_utils._normalize_string_set``): + ``None`` (YAML null) is empty and a bare scalar is a single-item list, NOT its characters.""" if values is None: return set() if isinstance(values, str): @@ -29,33 +23,23 @@ def _normalize_skill_names(values) -> Set[str]: def get_disabled_skills(config: dict, platform: Optional[str] = None) -> Set[str]: - """Return disabled skill names: the global list unioned with the platform list when given. - - A globally-disabled skill stays disabled on every platform, so the platform list adds to the - global list rather than replacing it (mirrors - ``agent.skill_utils.get_disabled_skill_names``). - """ + """Disabled skill names: the global list unioned with the platform list when given (globally + disabled stays disabled everywhere; mirrors ``agent.skill_utils.get_disabled_skill_names``).""" skills_cfg = config.get("skills") or {} if not isinstance(skills_cfg, dict): return set() from agent.skill_utils import ESSENTIAL_SKILLS - global_disabled = _normalize_skill_names(skills_cfg.get("disabled")) - if platform is None: - return global_disabled - ESSENTIAL_SKILLS - platform_disabled = cfg_get(skills_cfg, "platform_disabled", platform) - if platform_disabled is None: - return global_disabled - ESSENTIAL_SKILLS - return ( - global_disabled | _normalize_skill_names(platform_disabled) - ) - ESSENTIAL_SKILLS + disabled = _normalize_skill_names(skills_cfg.get("disabled")) + if platform is not None: + platform_disabled = cfg_get(skills_cfg, "platform_disabled", platform) + if platform_disabled is not None: + disabled = disabled | _normalize_skill_names(platform_disabled) + return disabled - ESSENTIAL_SKILLS def save_disabled_skills(config: dict, disabled: Set[str], platform: Optional[str] = None): - """Persist disabled skill names to config. - - Essential skills (e.g. ``hermes-agent``) are silently dropped from the list — they cannot be - disabled from any surface. - """ + """Persist disabled skill names to config; essential skills (e.g. ``hermes-agent``) are + silently dropped — they cannot be disabled from any surface.""" from agent.skill_utils import ESSENTIAL_SKILLS disabled = set(disabled) - ESSENTIAL_SKILLS config.setdefault("skills", {}) @@ -67,8 +51,6 @@ def save_disabled_skills(config: dict, disabled: Set[str], platform: Optional[st save_config(config) -# ─── Skill Discovery ───────────────────────────────────────────────────────── - def _list_all_skills() -> List[dict]: """Return all installed skills (ignoring disabled state).""" try: @@ -83,10 +65,8 @@ def _get_categories(skills: List[dict]) -> List[str]: return sorted({s["category"] or "uncategorized" for s in skills}) -# ─── Platform Selection ────────────────────────────────────────────────────── - def _select_platform() -> Optional[str]: - """Ask user which platform to configure, or global.""" + """Ask which platform to configure; None means global.""" options = [("global", "All platforms (global default)")] + list(PLATFORMS.items()) print() print(color(" Configure skills for:", Colors.BOLD)) @@ -97,67 +77,49 @@ def _select_platform() -> Optional[str]: raw = input(color(" Select [1]: ", Colors.YELLOW)).strip() except (KeyboardInterrupt, EOFError): return None - if not raw: - return None # global try: - idx = int(raw) - 1 - if 0 <= idx < len(options): - key = options[idx][0] - return None if key == "global" else key + idx = int(raw) - 1 # empty input -> ValueError -> global except ValueError: - pass + return None + if 0 <= idx < len(options) and options[idx][0] != "global": + return options[idx][0] return None -# ─── Category Toggle ───────────────────────────────────────────────────────── - def _toggle_by_category(skills: List[dict], disabled: Set[str]) -> Set[str]: """Toggle all skills in a category at once.""" from hermes_cli.curses_ui import curses_checklist categories = _get_categories(skills) - cat_labels = [] + cat_skills = [{s["name"] for s in skills if (s["category"] or "uncategorized") == cat} + for cat in categories] + cat_labels = [f"{cat} ({len(names)} skills)" for cat, names in zip(categories, cat_skills)] # A category is "enabled" (checked) when NOT all its skills are disabled - pre_selected = set() - for i, cat in enumerate(categories): - cat_skills = [s["name"] for s in skills if (s["category"] or "uncategorized") == cat] - cat_labels.append(f"{cat} ({len(cat_skills)} skills)") - if not all(s in disabled for s in cat_skills): - pre_selected.add(i) - - chosen = curses_checklist( - "Categories — toggle entire categories", - cat_labels, pre_selected, cancel_returns=pre_selected, - ) - + pre_selected = {i for i, names in enumerate(cat_skills) + if not all(s in disabled for s in names)} + chosen = curses_checklist("Categories — toggle entire categories", + cat_labels, pre_selected, cancel_returns=pre_selected) new_disabled = set(disabled) - for i, cat in enumerate(categories): - cat_skills = {s["name"] for s in skills if (s["category"] or "uncategorized") == cat} + for i, names in enumerate(cat_skills): if i in chosen: - new_disabled -= cat_skills # category enabled → remove from disabled + new_disabled -= names # category enabled → remove from disabled else: - new_disabled |= cat_skills # category disabled → add to disabled + new_disabled |= names # category disabled → add to disabled return new_disabled -# ─── Entry Point ────────────────────────────────────────────────────────────── - def skills_command(args=None): """Entry point for `hermes skills`.""" from hermes_cli.curses_ui import curses_checklist config = load_config() skills = _list_all_skills() - if not skills: print(color(" No skills installed.", Colors.DIM)) return - # Step 1: Select platform platform = _select_platform() platform_label = PLATFORMS.get(platform, "All platforms") if platform else "All platforms" - - # Step 2: Select mode — individual or by category print() print(color(f" Configure for: {platform_label}", Colors.DIM)) print() @@ -170,22 +132,15 @@ def skills_command(args=None): return disabled = get_disabled_skills(config, platform) - if mode == "2": new_disabled = _toggle_by_category(skills, disabled) else: - # Build labels and map indices → skill names - labels = [ - f"{s['name']} ({s['category'] or 'uncategorized'}) — {s['description'][:55]}" - for s in skills - ] + labels = [f"{s['name']} ({s['category'] or 'uncategorized'}) — {s['description'][:55]}" + for s in skills] # "selected" = enabled (not disabled) — matches the [✓] convention pre_selected = {i for i, s in enumerate(skills) if s["name"] not in disabled} - chosen = curses_checklist( - f"Skills for {platform_label}", - labels, pre_selected, cancel_returns=pre_selected, - ) - # Anything NOT chosen is disabled + chosen = curses_checklist(f"Skills for {platform_label}", + labels, pre_selected, cancel_returns=pre_selected) new_disabled = {skills[i]["name"] for i in range(len(skills)) if i not in chosen} if new_disabled == disabled: diff --git a/hermes_cli/skin_cmd.py b/hermes_cli/skin_cmd.py index f7606072a5..7c7daf6fec 100644 --- a/hermes_cli/skin_cmd.py +++ b/hermes_cli/skin_cmd.py @@ -49,35 +49,27 @@ def _skin_set(key: str, value: str, skin: str | None) -> int: data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} target = name else: - # Built-in (or missing): fork into an editable copy that keeps its full - # palette, under a fresh name so the built-in stays intact for revert. + # Built-in (or missing): fork into an editable copy that keeps its full palette, under a + # fresh name so the built-in stays intact for revert. from hermes_cli.skin_engine import load_skin resolved = load_skin(name) target = f"{name}-custom" path = _skins_dir() / f"{target}.yaml" - data = { - "name": target, - "description": f"{name} + custom {key}", - "colors": dict(resolved.colors), - "branding": dict(resolved.branding), - "tool_prefix": resolved.tool_prefix, - } - + data = {"name": target, "description": f"{name} + custom {key}", + "colors": dict(resolved.colors), "branding": dict(resolved.branding), + "tool_prefix": resolved.tool_prefix} if not isinstance(data.get("colors"), dict): data["colors"] = {} data["colors"][key] = value data.setdefault("name", target) - - # Atomic write: write_text truncates with no fsync; safe_load("") → None - # → {} would permanently lose the palette on the next set (#51356, #16743). + # Atomic write: write_text truncates with no fsync; safe_load("") → None → {} would + # permanently lose the palette on the next set. from utils import atomic_yaml_write atomic_yaml_write(path, data, sort_keys=False) - if target != name: _use(target) - print(f"✓ {key} = {value} in {display_hermes_home()}/skins/{target}.yaml (live within ~1s)") return 0 @@ -95,7 +87,6 @@ def _skin_list() -> int: def skin_command(args) -> None: """Dispatch ``hermes skin ``.""" verb = getattr(args, "skin_command", None) - if verb == "set": sys.exit(_skin_set(args.key, args.value, getattr(args, "skin", None))) elif verb == "use": diff --git a/hermes_cli/skin_engine.py b/hermes_cli/skin_engine.py index 92f6c6be03..eb70927a06 100644 --- a/hermes_cli/skin_engine.py +++ b/hermes_cli/skin_engine.py @@ -10,22 +10,15 @@ from hermes_constants import get_hermes_home logger = logging.getLogger(__name__) -# ============================================================================= -# Skin data structure -# ============================================================================= - @dataclass class SkinConfig: """Complete skin configuration.""" name: str description: str = "" colors: Dict[str, str] = field(default_factory=dict) - # Paired palettes for terminals whose background polarity differs from the - # one `colors` was authored against (mirrors the desktop app's - # colors/darkColors pairing). A consumer that knows the terminal is light - # prefers `light_colors` (falling back to `colors`), and vice versa for - # `dark_colors`. Both merge over the default skin's matching block, so - # partial user skins still resolve to a complete palette. + # Paired palettes for the opposite background polarity (mirrors the desktop app's + # colors/darkColors pairing): a light terminal prefers `light_colors` (falling back to + # `colors`), and vice versa for `dark_colors`. light_colors: Dict[str, str] = field(default_factory=dict) dark_colors: Dict[str, str] = field(default_factory=dict) spinner: Dict[str, Any] = field(default_factory=dict) @@ -36,152 +29,80 @@ class SkinConfig: banner_hero: str = "" # Rich-markup hero art (replaces HERMES_CADUCEUS) def get_color(self, key: str, fallback: str = "") -> str: - """Get a color value with fallback.""" return self.colors.get(key, fallback) def get_spinner_wings(self) -> List[Tuple[str, str]]: - """Get spinner wing pairs, or empty list if none.""" - return [ - (str(pair[0]), str(pair[1])) - for pair in self.spinner.get("wings", []) - if isinstance(pair, (list, tuple)) and len(pair) == 2 - ] + """Spinner wing pairs, or empty list if none.""" + return [(str(pair[0]), str(pair[1])) for pair in self.spinner.get("wings", []) + if isinstance(pair, (list, tuple)) and len(pair) == 2] def get_branding(self, key: str, fallback: str = "") -> str: - """Get a branding value with fallback.""" return self.branding.get(key, fallback) -# ============================================================================= -# Built-in skin definitions -# ============================================================================= - # Branding shared by every Hermes-named built-in (mono/daylight override help_header). _HERMES_BRANDING: Dict[str, str] = { "agent_name": "Hermes Agent", "welcome": "Welcome to Hermes Agent! Type your message or /help for commands.", - "goodbye": "Goodbye! ⚕", - "response_label": " ⚕ Hermes ", - "prompt_symbol": "❯", - "help_header": "(^_^)? Available Commands", -} + "goodbye": "Goodbye! ⚕", "response_label": " ⚕ Hermes ", "prompt_symbol": "❯", + "help_header": "(^_^)? Available Commands"} _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "default": { "name": "default", "description": "Classic Hermes — gold and kawaii", - # Dark-authored. Values match the TUI's DARK_THEME so the classic CLI - # and the TUI render the same Hermes gold. + # Dark-authored; values match the TUI's DARK_THEME so both render the same gold. "colors": { - "banner_border": "#CD7F32", - "banner_title": "#FFD700", - "banner_accent": "#FFBF00", - "banner_dim": "#B8860B", - "banner_text": "#FFF8DC", - "ui_accent": "#FFBF00", - "ui_label": "#DAA520", - "ui_ok": "#4caf50", - "ui_error": "#ef5350", - "ui_warn": "#ffa726", - "prompt": "#FFF8DC", - "input_rule": "#CD7F32", - "response_border": "#FFD700", - "status_bar_bg": "#1a1a2e", - "status_bar_text": "#C0C0C0", - "status_bar_strong": "#FFD700", - "status_bar_dim": "#8A7A4A", - "status_bar_good": "#8FBC8F", - "status_bar_warn": "#FFD700", - "status_bar_bad": "#FF8C00", - "status_bar_critical": "#FF6B6B", - "session_label": "#DAA520", - "session_border": "#8B8682", - "completion_menu_bg": "#1a1a2e", - "completion_menu_current_bg": "#333355", - "selection_bg": "#3a3a55", - "shell_dollar": "#4dabf7", - "voice_status_bg": "#1a1a2e", - }, - # Light overlay (merged onto `colors`; dark mode renders the vivid - # block above untouched). The goldenrod ladder: on white, the vivid - # #FFD700/#FFBF00 read as glare and WCAG-darkened mustard (#867000) - # reads as mud — the sweet spot is the statusbar's goldenrod family - # (#B8860B/#DAA520): hue kept, saturation tamed, mid luminance. - # Hierarchy on white: ink body 8.9:1 > fade 5.2 > label 3.7 > - # muted 3.3 > title 2.7 > headers 2.4 (accents recede last, like - # slate's pastels — the raw-canon look, just not neon). + "banner_border": "#CD7F32", "banner_title": "#FFD700", "banner_accent": "#FFBF00", + "banner_dim": "#B8860B", "banner_text": "#FFF8DC", "ui_accent": "#FFBF00", + "ui_label": "#DAA520", "ui_ok": "#4caf50", "ui_error": "#ef5350", "ui_warn": "#ffa726", + "prompt": "#FFF8DC", "input_rule": "#CD7F32", "response_border": "#FFD700", + "status_bar_bg": "#1a1a2e", "status_bar_text": "#C0C0C0", + "status_bar_strong": "#FFD700", "status_bar_dim": "#8A7A4A", + "status_bar_good": "#8FBC8F", "status_bar_warn": "#FFD700", "status_bar_bad": "#FF8C00", + "status_bar_critical": "#FF6B6B", "session_label": "#DAA520", + "session_border": "#8B8682", "completion_menu_bg": "#1a1a2e", + "completion_menu_current_bg": "#333355", "selection_bg": "#3a3a55", + "shell_dollar": "#4dabf7", "voice_status_bg": "#1a1a2e"}, + # Light overlay (merged onto `colors`). Goldenrod ladder: on white the vivid + # #FFD700/#FFBF00 read as glare and WCAG-darkened mustard (#867000) as mud; the + # statusbar's goldenrod family (#B8860B/#DAA520) keeps the hue, tames saturation. + # Hierarchy on white: ink body 8.9:1 > fade 5.2 > label 3.7 > muted 3.3 > title 2.7 > + # headers 2.4. Fills (*_bg) flip the dark navy surfaces to light polarity. "light_colors": { - "banner_title": "#C8961E", - "banner_accent": "#D89B04", - "banner_dim": "#B8860B", - "banner_text": "#5C4718", - "ui_accent": "#D89B04", - "ui_label": "#A97E10", - "ui_ok": "#2E7D32", - "ui_error": "#C62828", - "ui_warn": "#D97706", - "prompt": "#5C4718", - "response_border": "#C8961E", - "session_label": "#A97E10", - "status_bar_text": "#6F6F6F", - "status_bar_strong": "#C8961E", - "status_bar_dim": "#9A8A5A", - "status_bar_good": "#2E7D32", - "status_bar_warn": "#C8961E", - "status_bar_bad": "#C2410C", - "status_bar_critical": "#B91C1C", - "shell_dollar": "#1E6FC0", - # Fills: flip the dark navy surfaces to light polarity. - "completion_menu_bg": "#F5F5F5", - "completion_menu_current_bg": "#E0D1BF", - "selection_bg": "#D4E4F7", - "status_bar_bg": "#F5F5F5", - "voice_status_bg": "#F5F5F5", - }, + "banner_title": "#C8961E", "banner_accent": "#D89B04", "banner_dim": "#B8860B", + "banner_text": "#5C4718", "ui_accent": "#D89B04", "ui_label": "#A97E10", + "ui_ok": "#2E7D32", "ui_error": "#C62828", "ui_warn": "#D97706", "prompt": "#5C4718", + "response_border": "#C8961E", "session_label": "#A97E10", "status_bar_text": "#6F6F6F", + "status_bar_strong": "#C8961E", "status_bar_dim": "#9A8A5A", + "status_bar_good": "#2E7D32", "status_bar_warn": "#C8961E", "status_bar_bad": "#C2410C", + "status_bar_critical": "#B91C1C", "shell_dollar": "#1E6FC0", + "completion_menu_bg": "#F5F5F5", "completion_menu_current_bg": "#E0D1BF", + "selection_bg": "#D4E4F7", "status_bar_bg": "#F5F5F5", "voice_status_bg": "#F5F5F5"}, "spinner": {}, # empty = hardcoded defaults in display.py "branding": _HERMES_BRANDING, - "tool_prefix": "┊", - }, + "tool_prefix": "┊"}, "ares": { "name": "ares", "description": "War-god theme — crimson and bronze", "colors": { - "banner_border": "#A93333", - "banner_title": "#C7A96B", - "banner_accent": "#DD4A3A", - "banner_dim": "#905151", - "banner_text": "#F1E6CF", - "ui_accent": "#DD4A3A", - "ui_label": "#C7A96B", - "ui_ok": "#4caf50", - "ui_error": "#ef5350", - "ui_warn": "#ffa726", - "prompt": "#F1E6CF", - "input_rule": "#A93333", - "response_border": "#C7A96B", - "status_bar_bg": "#2A1212", - "status_bar_text": "#F1E6CF", - "status_bar_strong": "#C7A96B", - "status_bar_dim": "#756054", - "status_bar_good": "#7BC96F", - "status_bar_warn": "#C7A96B", - "status_bar_bad": "#DD4A3A", - "status_bar_critical": "#EF5350", - "session_label": "#C7A96B", - "session_border": "#6E584B", - "completion_menu_bg": "#2A1212", - "completion_menu_current_bg": "#5C221D", - "selection_bg": "#692620", - "shell_dollar": "#DD4A3A", - "voice_status_bg": "#2A1212", - }, + "banner_border": "#A93333", "banner_title": "#C7A96B", "banner_accent": "#DD4A3A", + "banner_dim": "#905151", "banner_text": "#F1E6CF", "ui_accent": "#DD4A3A", + "ui_label": "#C7A96B", "ui_ok": "#4caf50", "ui_error": "#ef5350", "ui_warn": "#ffa726", + "prompt": "#F1E6CF", "input_rule": "#A93333", "response_border": "#C7A96B", + "status_bar_bg": "#2A1212", "status_bar_text": "#F1E6CF", + "status_bar_strong": "#C7A96B", "status_bar_dim": "#756054", + "status_bar_good": "#7BC96F", "status_bar_warn": "#C7A96B", "status_bar_bad": "#DD4A3A", + "status_bar_critical": "#EF5350", "session_label": "#C7A96B", + "session_border": "#6E584B", "completion_menu_bg": "#2A1212", + "completion_menu_current_bg": "#5C221D", "selection_bg": "#692620", + "shell_dollar": "#DD4A3A", "voice_status_bg": "#2A1212"}, "spinner": { "waiting_faces": ["(⚔)", "(⛨)", "(▲)", "(<>)", "(/)"], "thinking_faces": ["(⚔)", "(⛨)", "(▲)", "(⌁)", "(<>)"], "thinking_verbs": [ "forging", "marching", "sizing the field", "holding the line", - "hammering plans", "tempering steel", "plotting impact", "raising the shield", - ], + "hammering plans", "tempering steel", "plotting impact", "raising the shield"], "wings": [["⟪⚔", "⚔⟫"], ["⟪▲", "▲⟫"], ["⟪╸", "╺⟫"], ["⟪⛨", "⛨⟫"]], }, "branding": { @@ -190,8 +111,7 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "goodbye": "Farewell, warrior! ⚔", "response_label": " ⚔ Ares ", "prompt_symbol": "⚔", - "help_header": "(⚔) Available Commands", - }, + "help_header": "(⚔) Available Commands"}, "tool_prefix": "╎", "banner_logo": """[bold #A3261F] █████╗ ██████╗ ███████╗███████╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/] [bold #B73122]██╔══██╗██╔══██╗██╔════╝██╔════╝ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/] @@ -218,195 +138,98 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "name": "mono", "description": "Monochrome — clean grayscale", "colors": { - "banner_border": "#5E5E5E", - "banner_title": "#e6edf3", - "banner_accent": "#aaaaaa", - "banner_dim": "#606060", - "banner_text": "#c9d1d9", - "ui_accent": "#aaaaaa", - "ui_label": "#888888", - "ui_ok": "#888888", - "ui_error": "#cccccc", - "ui_warn": "#999999", - "prompt": "#c9d1d9", - "input_rule": "#606060", - "response_border": "#aaaaaa", - "status_bar_bg": "#1F1F1F", - "status_bar_text": "#C9D1D9", - "status_bar_strong": "#E6EDF3", - "status_bar_dim": "#777777", - "status_bar_good": "#B5B5B5", - "status_bar_warn": "#AAAAAA", - "status_bar_bad": "#D0D0D0", - "status_bar_critical": "#F0F0F0", - "session_label": "#888888", - "session_border": "#5E5E5E", - "completion_menu_bg": "#1F1F1F", - "completion_menu_current_bg": "#464646", - "selection_bg": "#505050", - "shell_dollar": "#aaaaaa", - "voice_status_bg": "#1F1F1F", - }, + "banner_border": "#5E5E5E", "banner_title": "#e6edf3", "banner_accent": "#aaaaaa", + "banner_dim": "#606060", "banner_text": "#c9d1d9", "ui_accent": "#aaaaaa", + "ui_label": "#888888", "ui_ok": "#888888", "ui_error": "#cccccc", "ui_warn": "#999999", + "prompt": "#c9d1d9", "input_rule": "#606060", "response_border": "#aaaaaa", + "status_bar_bg": "#1F1F1F", "status_bar_text": "#C9D1D9", + "status_bar_strong": "#E6EDF3", "status_bar_dim": "#777777", + "status_bar_good": "#B5B5B5", "status_bar_warn": "#AAAAAA", "status_bar_bad": "#D0D0D0", + "status_bar_critical": "#F0F0F0", "session_label": "#888888", + "session_border": "#5E5E5E", "completion_menu_bg": "#1F1F1F", + "completion_menu_current_bg": "#464646", "selection_bg": "#505050", + "shell_dollar": "#aaaaaa", "voice_status_bg": "#1F1F1F"}, "spinner": {}, "branding": {**_HERMES_BRANDING, "help_header": "[?] Available Commands"}, - "tool_prefix": "┊", - }, + "tool_prefix": "┊"}, "slate": { "name": "slate", "description": "Cool blue — developer-focused", "colors": { - "banner_border": "#4169e1", - "banner_title": "#7eb8f6", - "banner_accent": "#8EA8FF", - "banner_dim": "#545E6B", - "banner_text": "#c9d1d9", - "ui_accent": "#7eb8f6", - "ui_label": "#8EA8FF", - "ui_ok": "#63D0A6", - "ui_error": "#F7A072", - "ui_warn": "#e6a855", - "prompt": "#c9d1d9", - "input_rule": "#4169e1", - "response_border": "#7eb8f6", - "status_bar_bg": "#151C2F", - "status_bar_text": "#C9D1D9", - "status_bar_strong": "#7EB8F6", - "status_bar_dim": "#5D6672", - "status_bar_good": "#63D0A6", - "status_bar_warn": "#E6A855", - "status_bar_bad": "#F7A072", - "status_bar_critical": "#FF7A7A", - "session_label": "#7eb8f6", - "session_border": "#545E6B", - "completion_menu_bg": "#151C2F", - "completion_menu_current_bg": "#324867", - "selection_bg": "#3A5375", - "shell_dollar": "#7eb8f6", - "voice_status_bg": "#151C2F", - }, + "banner_border": "#4169e1", "banner_title": "#7eb8f6", "banner_accent": "#8EA8FF", + "banner_dim": "#545E6B", "banner_text": "#c9d1d9", "ui_accent": "#7eb8f6", + "ui_label": "#8EA8FF", "ui_ok": "#63D0A6", "ui_error": "#F7A072", "ui_warn": "#e6a855", + "prompt": "#c9d1d9", "input_rule": "#4169e1", "response_border": "#7eb8f6", + "status_bar_bg": "#151C2F", "status_bar_text": "#C9D1D9", + "status_bar_strong": "#7EB8F6", "status_bar_dim": "#5D6672", + "status_bar_good": "#63D0A6", "status_bar_warn": "#E6A855", "status_bar_bad": "#F7A072", + "status_bar_critical": "#FF7A7A", "session_label": "#7eb8f6", + "session_border": "#545E6B", "completion_menu_bg": "#151C2F", + "completion_menu_current_bg": "#324867", "selection_bg": "#3A5375", + "shell_dollar": "#7eb8f6", "voice_status_bg": "#151C2F"}, "spinner": {}, "branding": _HERMES_BRANDING, - "tool_prefix": "┊", - }, + "tool_prefix": "┊"}, "daylight": { "name": "daylight", "description": "Light theme for bright terminals with dark text and cool blue accents", "colors": { - "banner_border": "#2563EB", - "banner_title": "#0F172A", - "banner_accent": "#1D4ED8", - "banner_dim": "#475569", - "banner_text": "#111827", - "ui_accent": "#2563EB", - "ui_label": "#0F766E", - "ui_ok": "#15803D", - "ui_error": "#B91C1C", - "ui_warn": "#B45309", - "prompt": "#111827", - "input_rule": "#6E94BE", - "response_border": "#2563EB", - "status_bar_bg": "#E5EDF8", - "status_bar_text": "#111827", - "status_bar_strong": "#2563EB", - "status_bar_dim": "#838890", - "status_bar_good": "#15803D", - "status_bar_warn": "#B45309", - "status_bar_bad": "#B45309", - "status_bar_critical": "#B91C1C", - "session_label": "#1D4ED8", - "session_border": "#64748B", - "completion_menu_bg": "#F8FAFC", - "completion_menu_current_bg": "#DBEAFE", - "completion_menu_meta_bg": "#EEF2FF", - "completion_menu_meta_current_bg": "#BFDBFE", - "selection_bg": "#D3E0FB", - "shell_dollar": "#2563EB", - "voice_status_bg": "#E5EDF8", - }, + "banner_border": "#2563EB", "banner_title": "#0F172A", "banner_accent": "#1D4ED8", + "banner_dim": "#475569", "banner_text": "#111827", "ui_accent": "#2563EB", + "ui_label": "#0F766E", "ui_ok": "#15803D", "ui_error": "#B91C1C", "ui_warn": "#B45309", + "prompt": "#111827", "input_rule": "#6E94BE", "response_border": "#2563EB", + "status_bar_bg": "#E5EDF8", "status_bar_text": "#111827", + "status_bar_strong": "#2563EB", "status_bar_dim": "#838890", + "status_bar_good": "#15803D", "status_bar_warn": "#B45309", "status_bar_bad": "#B45309", + "status_bar_critical": "#B91C1C", "session_label": "#1D4ED8", + "session_border": "#64748B", "completion_menu_bg": "#F8FAFC", + "completion_menu_current_bg": "#DBEAFE", "completion_menu_meta_bg": "#EEF2FF", + "completion_menu_meta_current_bg": "#BFDBFE", "selection_bg": "#D3E0FB", + "shell_dollar": "#2563EB", "voice_status_bg": "#E5EDF8"}, "spinner": {}, "branding": {**_HERMES_BRANDING, "help_header": "[?] Available Commands"}, - "tool_prefix": "│", - }, + "tool_prefix": "│"}, "warm-lightmode": { "name": "warm-lightmode", "description": "Warm light mode — dark brown/gold text for light terminal backgrounds", "colors": { - "banner_border": "#8B6914", - "banner_title": "#5C3D11", - "banner_accent": "#8B4513", - "banner_dim": "#8B7355", - "banner_text": "#2C1810", - "ui_accent": "#8B4513", - "ui_label": "#5C3D11", - "ui_ok": "#2E7D32", - "ui_error": "#C62828", - "ui_warn": "#E65100", - "prompt": "#2C1810", - "input_rule": "#8B6914", - "response_border": "#8B6914", - "status_bar_bg": "#F5F0E8", - "status_bar_text": "#2C1810", - "status_bar_strong": "#8B4513", - "status_bar_dim": "#8A8F98", - "status_bar_good": "#2E7D32", - "status_bar_warn": "#E65100", - "status_bar_bad": "#DA4D00", - "status_bar_critical": "#C62828", - "session_label": "#5C3D11", - "session_border": "#A0845C", - "completion_menu_bg": "#F5EFE0", - "completion_menu_current_bg": "#E8DCC8", - "completion_menu_meta_bg": "#F0E8D8", - "completion_menu_meta_current_bg": "#DFCFB0", - "selection_bg": "#E8DAD0", - "shell_dollar": "#8B4513", - "voice_status_bg": "#F5F0E8", - }, + "banner_border": "#8B6914", "banner_title": "#5C3D11", "banner_accent": "#8B4513", + "banner_dim": "#8B7355", "banner_text": "#2C1810", "ui_accent": "#8B4513", + "ui_label": "#5C3D11", "ui_ok": "#2E7D32", "ui_error": "#C62828", "ui_warn": "#E65100", + "prompt": "#2C1810", "input_rule": "#8B6914", "response_border": "#8B6914", + "status_bar_bg": "#F5F0E8", "status_bar_text": "#2C1810", + "status_bar_strong": "#8B4513", "status_bar_dim": "#8A8F98", + "status_bar_good": "#2E7D32", "status_bar_warn": "#E65100", "status_bar_bad": "#DA4D00", + "status_bar_critical": "#C62828", "session_label": "#5C3D11", + "session_border": "#A0845C", "completion_menu_bg": "#F5EFE0", + "completion_menu_current_bg": "#E8DCC8", "completion_menu_meta_bg": "#F0E8D8", + "completion_menu_meta_current_bg": "#DFCFB0", "selection_bg": "#E8DAD0", + "shell_dollar": "#8B4513", "voice_status_bg": "#F5F0E8"}, "spinner": {}, "branding": _HERMES_BRANDING, - "tool_prefix": "┊", - }, + "tool_prefix": "┊"}, "poseidon": { "name": "poseidon", "description": "Ocean-god theme — deep blue and seafoam", "colors": { - "banner_border": "#2A6FB9", - "banner_title": "#A9DFFF", - "banner_accent": "#5DB8F5", - "banner_dim": "#44638F", - "banner_text": "#EAF7FF", - "ui_accent": "#5DB8F5", - "ui_label": "#A9DFFF", - "ui_ok": "#4caf50", - "ui_error": "#ef5350", - "ui_warn": "#ffa726", - "prompt": "#EAF7FF", - "input_rule": "#2A6FB9", - "response_border": "#5DB8F5", - "status_bar_bg": "#0F2440", - "status_bar_text": "#EAF7FF", - "status_bar_strong": "#A9DFFF", - "status_bar_dim": "#52708A", - "status_bar_good": "#6ED7B0", - "status_bar_warn": "#5DB8F5", - "status_bar_bad": "#3576BC", - "status_bar_critical": "#D94F4F", - "session_label": "#A9DFFF", - "session_border": "#496884", - "completion_menu_bg": "#0F2440", - "completion_menu_current_bg": "#254D73", - "selection_bg": "#2A587F", - "shell_dollar": "#5DB8F5", - "voice_status_bg": "#0F2440", - }, + "banner_border": "#2A6FB9", "banner_title": "#A9DFFF", "banner_accent": "#5DB8F5", + "banner_dim": "#44638F", "banner_text": "#EAF7FF", "ui_accent": "#5DB8F5", + "ui_label": "#A9DFFF", "ui_ok": "#4caf50", "ui_error": "#ef5350", "ui_warn": "#ffa726", + "prompt": "#EAF7FF", "input_rule": "#2A6FB9", "response_border": "#5DB8F5", + "status_bar_bg": "#0F2440", "status_bar_text": "#EAF7FF", + "status_bar_strong": "#A9DFFF", "status_bar_dim": "#52708A", + "status_bar_good": "#6ED7B0", "status_bar_warn": "#5DB8F5", "status_bar_bad": "#3576BC", + "status_bar_critical": "#D94F4F", "session_label": "#A9DFFF", + "session_border": "#496884", "completion_menu_bg": "#0F2440", + "completion_menu_current_bg": "#254D73", "selection_bg": "#2A587F", + "shell_dollar": "#5DB8F5", "voice_status_bg": "#0F2440"}, "spinner": { "waiting_faces": ["(≈)", "(Ψ)", "(∿)", "(◌)", "(◠)"], "thinking_faces": ["(Ψ)", "(∿)", "(≈)", "(⌁)", "(◌)"], "thinking_verbs": [ "charting currents", "sounding the depth", "reading foam lines", "steering the trident", "tracking undertow", "plotting sea lanes", - "calling the swell", "measuring pressure", - ], + "calling the swell", "measuring pressure"], "wings": [["⟪≈", "≈⟫"], ["⟪Ψ", "Ψ⟫"], ["⟪∿", "∿⟫"], ["⟪◌", "◌⟫"]], }, "branding": { @@ -415,8 +238,7 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "goodbye": "Fair winds! Ψ", "response_label": " Ψ Poseidon ", "prompt_symbol": "Ψ", - "help_header": "(Ψ) Available Commands", - }, + "help_header": "(Ψ) Available Commands"}, "tool_prefix": "│", "banner_logo": """[bold #B8E8FF]██████╗ ██████╗ ███████╗███████╗██╗██████╗ ██████╗ ███╗ ██╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/] [bold #97D6FF]██╔══██╗██╔═══██╗██╔════╝██╔════╝██║██╔══██╗██╔═══██╗████╗ ██║ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/] @@ -442,43 +264,24 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "name": "sisyphus", "description": "Sisyphean theme — austere grayscale with persistence", "colors": { - "banner_border": "#B7B7B7", - "banner_title": "#F5F5F5", - "banner_accent": "#E7E7E7", - "banner_dim": "#5C5C5C", - "banner_text": "#D3D3D3", - "ui_accent": "#E7E7E7", - "ui_label": "#D3D3D3", - "ui_ok": "#919191", - "ui_error": "#E7E7E7", - "ui_warn": "#B7B7B7", - "prompt": "#F5F5F5", - "input_rule": "#656565", - "response_border": "#B7B7B7", - "status_bar_bg": "#202020", - "status_bar_text": "#D3D3D3", - "status_bar_strong": "#F5F5F5", - "status_bar_dim": "#6D6D6D", - "status_bar_good": "#B7B7B7", - "status_bar_warn": "#D3D3D3", - "status_bar_bad": "#E7E7E7", - "status_bar_critical": "#F5F5F5", - "session_label": "#919191", - "session_border": "#656565", - "completion_menu_bg": "#202020", - "completion_menu_current_bg": "#585858", - "selection_bg": "#666666", - "shell_dollar": "#E7E7E7", - "voice_status_bg": "#202020", - }, + "banner_border": "#B7B7B7", "banner_title": "#F5F5F5", "banner_accent": "#E7E7E7", + "banner_dim": "#5C5C5C", "banner_text": "#D3D3D3", "ui_accent": "#E7E7E7", + "ui_label": "#D3D3D3", "ui_ok": "#919191", "ui_error": "#E7E7E7", "ui_warn": "#B7B7B7", + "prompt": "#F5F5F5", "input_rule": "#656565", "response_border": "#B7B7B7", + "status_bar_bg": "#202020", "status_bar_text": "#D3D3D3", + "status_bar_strong": "#F5F5F5", "status_bar_dim": "#6D6D6D", + "status_bar_good": "#B7B7B7", "status_bar_warn": "#D3D3D3", "status_bar_bad": "#E7E7E7", + "status_bar_critical": "#F5F5F5", "session_label": "#919191", + "session_border": "#656565", "completion_menu_bg": "#202020", + "completion_menu_current_bg": "#585858", "selection_bg": "#666666", + "shell_dollar": "#E7E7E7", "voice_status_bg": "#202020"}, "spinner": { "waiting_faces": ["(◉)", "(◌)", "(◬)", "(⬤)", "(::)"], "thinking_faces": ["(◉)", "(◬)", "(◌)", "(○)", "(●)"], "thinking_verbs": [ "finding traction", "measuring the grade", "resetting the boulder", "counting the ascent", "testing leverage", "setting the shoulder", - "pushing uphill", "enduring the loop", - ], + "pushing uphill", "enduring the loop"], "wings": [["⟪◉", "◉⟫"], ["⟪◬", "◬⟫"], ["⟪◌", "◌⟫"], ["⟪⬤", "⬤⟫"]], }, "branding": { @@ -487,8 +290,7 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "goodbye": "The boulder waits. ◉", "response_label": " ◉ Sisyphus ", "prompt_symbol": "◉", - "help_header": "(◉) Available Commands", - }, + "help_header": "(◉) Available Commands"}, "tool_prefix": "│", "banner_logo": """[bold #F5F5F5]███████╗██╗███████╗██╗ ██╗██████╗ ██╗ ██╗██╗ ██╗███████╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/] [bold #E7E7E7]██╔════╝██║██╔════╝╚██╗ ██╔╝██╔══██╗██║ ██║██║ ██║██╔════╝ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/] @@ -515,45 +317,25 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "name": "charizard", "description": "Volcanic theme — burnt orange and ember", "colors": { - "banner_border": "#C75B1D", - "banner_title": "#FFD39A", - "banner_accent": "#F29C38", - "banner_dim": "#C58A45", - "banner_text": "#FFF0D4", - "ui_accent": "#F29C38", - "ui_label": "#FFD39A", - "ui_ok": "#4caf50", - "ui_error": "#ef5350", - "ui_warn": "#ffa726", - "prompt": "#FFF0D4", - "input_rule": "#C75B1D", - "response_border": "#F29C38", - "status_bar_bg": "#2B160E", - "status_bar_text": "#FFF0D4", - "status_bar_strong": "#FFD39A", - "status_bar_dim": "#826144", - "status_bar_good": "#6BCB77", - "status_bar_warn": "#F29C38", - "status_bar_bad": "#E2832B", - "status_bar_critical": "#EF5350", - "session_label": "#FFD39A", - "session_border": "#7B593A", - "completion_menu_bg": "#0B0503", - "completion_menu_current_bg": "#4A1B07", - "completion_menu_meta_bg": "#120806", - "completion_menu_meta_current_bg": "#5A260D", - "selection_bg": "#5A260D", - "shell_dollar": "#F29C38", - "voice_status_bg": "#2B160E", - }, + "banner_border": "#C75B1D", "banner_title": "#FFD39A", "banner_accent": "#F29C38", + "banner_dim": "#C58A45", "banner_text": "#FFF0D4", "ui_accent": "#F29C38", + "ui_label": "#FFD39A", "ui_ok": "#4caf50", "ui_error": "#ef5350", "ui_warn": "#ffa726", + "prompt": "#FFF0D4", "input_rule": "#C75B1D", "response_border": "#F29C38", + "status_bar_bg": "#2B160E", "status_bar_text": "#FFF0D4", + "status_bar_strong": "#FFD39A", "status_bar_dim": "#826144", + "status_bar_good": "#6BCB77", "status_bar_warn": "#F29C38", "status_bar_bad": "#E2832B", + "status_bar_critical": "#EF5350", "session_label": "#FFD39A", + "session_border": "#7B593A", "completion_menu_bg": "#0B0503", + "completion_menu_current_bg": "#4A1B07", "completion_menu_meta_bg": "#120806", + "completion_menu_meta_current_bg": "#5A260D", "selection_bg": "#5A260D", + "shell_dollar": "#F29C38", "voice_status_bg": "#2B160E"}, "spinner": { "waiting_faces": ["(✦)", "(▲)", "(◇)", "(<>)", "(🔥)"], "thinking_faces": ["(✦)", "(▲)", "(◇)", "(⌁)", "(🔥)"], "thinking_verbs": [ "banking into the draft", "measuring burn", "reading the updraft", "tracking ember fall", "setting wing angle", "holding the flame core", - "plotting a hot landing", "coiling for lift", - ], + "plotting a hot landing", "coiling for lift"], "wings": [["⟪✦", "✦⟫"], ["⟪▲", "▲⟫"], ["⟪◌", "◌⟫"], ["⟪◇", "◇⟫"]], }, "branding": { @@ -562,8 +344,7 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { "goodbye": "Flame out! ✦", "response_label": " ✦ Charizard ", "prompt_symbol": "✦", - "help_header": "(✦) Available Commands", - }, + "help_header": "(✦) Available Commands"}, "tool_prefix": "│", "banner_logo": """[bold #FFF0D4] ██████╗██╗ ██╗ █████╗ ██████╗ ██╗███████╗ █████╗ ██████╗ ██████╗ █████╗ ██████╗ ███████╗███╗ ██╗████████╗[/] [bold #FFD39A]██╔════╝██║ ██║██╔══██╗██╔══██╗██║╚══███╔╝██╔══██╗██╔══██╗██╔══██╗ ██╔══██╗██╔════╝ ██╔════╝████╗ ██║╚══██╔══╝[/] @@ -584,25 +365,18 @@ _BUILTIN_SKINS: Dict[str, Dict[str, Any]] = { [#F29C38]⠀⠀⠀⠀⠀⠀⠀⠀⣰⡿⢿⣆⠀⠀⠀⠀⠀⠀⠀⠀⠀[/] [#F29C38]⠀⠀⠀⠀⠀⠀⠀⣼⡟⠀⠀⢻⣧⠀⠀⠀⠀⠀⠀⠀⠀[/] [dim #7A3511]⠀⠀⠀⠀⠀⠀⠀tail flame lit⠀⠀⠀⠀⠀⠀⠀⠀[/]""", - }, -} - - -# ============================================================================= -# Skin loading and management -# ============================================================================= + }} _active_skin: Optional[SkinConfig] = None _active_skin_name: str = "default" def _skins_dir() -> Path: - """User skins directory.""" return get_hermes_home() / "skins" def _load_skin_from_yaml(path: Path) -> Optional[Dict[str, Any]]: - """Load a skin definition from a YAML file.""" + """Load a skin definition from a YAML file; None on any failure.""" try: import yaml with open(path, "r", encoding="utf-8") as f: @@ -614,68 +388,45 @@ def _load_skin_from_yaml(path: Path) -> Optional[Dict[str, Any]]: return None -def _mapping_or_empty(value: Any, *, section: str, skin_name: str) -> Dict[str, Any]: - """Return a mapping value or an empty dict when the section type is invalid.""" - if isinstance(value, dict): - return value - if value is None: - return {} - logger.warning( - "Skin '%s' has invalid '%s' section type (%s); ignoring section", - skin_name, - section, - type(value).__name__, - ) - return {} - - def _build_skin_config(data: Dict[str, Any]) -> SkinConfig: """Build a SkinConfig from a raw dict (built-in or loaded from YAML).""" - # Start with default values as base for missing keys default = _BUILTIN_SKINS["default"] skin_name = str(data.get("name", "unknown")) def section(key: str) -> Dict[str, Any]: - return _mapping_or_empty(data.get(key), section=key, skin_name=skin_name) + value = data.get(key) + if isinstance(value, dict): + return value + if value is not None: + logger.warning("Skin '%s' has invalid '%s' section type (%s); ignoring section", + skin_name, key, type(value).__name__) + return {} def merged(key: str) -> Dict[str, Any]: return {**default.get(key, {}), **section(key)} - colors, spinner, branding = merged("colors"), merged("spinner"), merged("branding") - tool_emojis = section("tool_emojis") - # Paired palettes are NOT merged over the default skin's blocks: an empty - # block means "this skin has no hand-tuned variant for that polarity", and - # consumers (the TUI) fall back to `colors` + automatic adaptation. Merging - # the default's gold light palette under a crimson skin would be worse - # than adapting the crimson. + # Paired palettes are NOT merged over the default skin's blocks: an empty block means + # "no hand-tuned variant for that polarity" and consumers (the TUI) fall back to `colors` + # + automatic adaptation, which beats the default's gold light palette under a crimson skin. return SkinConfig( - name=skin_name, - description=data.get("description", ""), - colors=colors, - light_colors=section("light_colors"), - dark_colors=section("dark_colors"), - spinner=spinner, - branding=branding, + name=skin_name, description=data.get("description", ""), colors=merged("colors"), + light_colors=section("light_colors"), dark_colors=section("dark_colors"), + spinner=merged("spinner"), branding=merged("branding"), tool_prefix=data.get("tool_prefix", default.get("tool_prefix", "┊")), - tool_emojis=tool_emojis, - banner_logo=data.get("banner_logo", ""), - banner_hero=data.get("banner_hero", ""), - ) + tool_emojis=section("tool_emojis"), banner_logo=data.get("banner_logo", ""), + banner_hero=data.get("banner_hero", "")) def list_skins() -> List[Dict[str, str]]: - """List all available skins (built-in + user-installed).""" - result = [ - {"name": name, "description": data.get("description", ""), "source": "builtin"} - for name, data in _BUILTIN_SKINS.items() - ] + """List all available skins (built-in + user-installed); user skins never shadow built-ins.""" + result = [{"name": name, "description": data.get("description", ""), "source": "builtin"} + for name, data in _BUILTIN_SKINS.items()] skins_path = _skins_dir() if skins_path.is_dir(): for f in sorted(skins_path.glob("*.yaml")): data = _load_skin_from_yaml(f) if data: skin_name = data.get("name", f.stem) - # Skip if it shadows a built-in if any(s["name"] == skin_name for s in result): continue result.append({"name": skin_name, "description": data.get("description", ""), "source": "user"}) @@ -683,26 +434,20 @@ def list_skins() -> List[Dict[str, str]]: def load_skin(name: str) -> SkinConfig: - """Load a skin by name. Checks user skins first, then built-in.""" - # Check user skins directory - skins_path = _skins_dir() - user_file = skins_path / f"{name}.yaml" + """Load a skin by name: user skins first, then built-in, then default.""" + user_file = _skins_dir() / f"{name}.yaml" if user_file.is_file(): data = _load_skin_from_yaml(user_file) if data: return _build_skin_config(data) - - # Check built-in skins if name in _BUILTIN_SKINS: return _build_skin_config(_BUILTIN_SKINS[name]) - - # Fallback to default logger.warning("Skin '%s' not found, using default", name) return _build_skin_config(_BUILTIN_SKINS["default"]) def get_active_skin() -> SkinConfig: - """Get the currently active skin config (cached).""" + """Currently active skin config (cached).""" global _active_skin if _active_skin is None: _active_skin = load_skin(_active_skin_name) @@ -718,7 +463,6 @@ def set_active_skin(name: str) -> SkinConfig: def get_active_skin_name() -> str: - """Get the name of the currently active skin.""" return _active_skin_name @@ -729,11 +473,6 @@ def init_skin_from_config(config: dict) -> None: set_active_skin(skin_name.strip() if isinstance(skin_name, str) and skin_name.strip() else "default") -# ============================================================================= -# Convenience helpers for CLI modules -# ============================================================================= - - def _active_branding(key: str, fallback: str) -> str: try: return get_active_skin().get_branding(key, fallback) @@ -742,22 +481,16 @@ def _active_branding(key: str, fallback: str) -> str: def get_active_prompt_symbol(fallback: str = "❯") -> str: - """Return the interactive prompt symbol with a single trailing space. - - Skins store ``prompt_symbol`` as a bare token; the space is appended here so callers can drop it - straight into a rendered prompt. - """ + """Interactive prompt symbol (skins store a bare token) plus a single trailing space.""" cleaned = (_active_branding("prompt_symbol", fallback) or fallback).strip() return f"{cleaned or fallback.strip()} " def get_active_help_header(fallback: str = "(^_^)? Available Commands") -> str: - """Get the /help header from the active skin.""" return _active_branding("help_header", fallback) def get_active_goodbye(fallback: str = "Goodbye! ⚕") -> str: - """Get the goodbye line from the active skin.""" return _active_branding("goodbye", fallback) @@ -767,11 +500,8 @@ def get_prompt_toolkit_style_overrides() -> Dict[str, str]: skin = get_active_skin() except Exception: return {} - - # Input/prompt: leave unset by default so the typed text inherits - # the terminal's foreground color (readable in both light and dark - # color schemes). Skins can opt into a colored prompt by setting - # `prompt` explicitly in their YAML. + # `prompt` is unset by default so typed text inherits the terminal's foreground (readable + # on light and dark schemes); skins opt into a colored prompt symbol via `prompt` in YAML. c = skin.get_color prompt = c("prompt", "") input_rule = c("input_rule", "#CD7F32") @@ -796,11 +526,7 @@ def get_prompt_toolkit_style_overrides() -> Dict[str, str]: menu_meta_current_bg = c("completion_menu_meta_current_bg", menu_current_bg) return { - # Typed input always uses terminal default fg/bg so it's - # readable in both light and dark Terminal.app modes. The - # skin's `prompt` color (if any) only styles the prompt symbol, - # NOT the user's typed text. - "input-area": "", + "input-area": "", # terminal default fg/bg — `prompt` styles the symbol, NOT typed text "placeholder": f"{dim} italic", "prompt": prompt, "prompt-working": f"{dim} italic", @@ -837,5 +563,4 @@ def get_prompt_toolkit_style_overrides() -> Dict[str, str]: "approval-choice": dim, "approval-selected": f"{title} bold", "voice-status": f"bg:{voice_bg} {label}", - "voice-status-recording": f"bg:{voice_bg} {error} bold", - } + "voice-status-recording": f"bg:{voice_bg} {error} bold"} diff --git a/hermes_cli/slack_cli.py b/hermes_cli/slack_cli.py index df6c7f80cc..19cb5057c8 100644 --- a/hermes_cli/slack_cli.py +++ b/hermes_cli/slack_cli.py @@ -1,33 +1,22 @@ -"""``hermes slack ...`` CLI subcommands. - -Today only ``hermes slack manifest`` is implemented — it generates the Slack app manifest JSON for -registering every gateway command as a native Slack slash (``/btw``, ``/stop``, ``/model``, …) so -users get the same first-class slash UX Discord and Telegram already have. -""" +"""``hermes slack manifest`` — generate the Slack app manifest JSON that registers every gateway +command as a native Slack slash (``/btw``, ``/stop``, ``/model``, …).""" from __future__ import annotations import json -import os import sys from pathlib import Path - SLACK_LONG_DESCRIPTION_MIN_CHARACTERS = 175 SLACK_LONG_DESCRIPTION_MAX_CHARACTERS = 4000 def _build_full_manifest( - bot_name: str, - bot_description: str, - include_assistant: bool = True, - messaging_experience: str | None = None, - long_description: str | None = None, -) -> dict: - """Build a full Slack manifest merging display info + our slash list. + bot_name: str, bot_description: str, include_assistant: bool = True, + messaging_experience: str | None = None, long_description: str | None = None) -> dict: + """Build a full Slack manifest: display info + slash list from ``COMMAND_REGISTRY``. - The slash-command list is always generated from ``COMMAND_REGISTRY`` so it stays in sync with - the rest of Hermes. Other manifest sections (display info, OAuth scopes, socket mode) are set to - sensible defaults for a Hermes deployment — users can tweak them in the Slack UI after pasting. + Other sections (OAuth scopes, socket mode) are sensible Hermes defaults, tweakable in the Slack + UI after pasting. """ from hermes_cli.commands import slack_app_manifest @@ -35,74 +24,34 @@ def _build_full_manifest( messaging_experience = "assistant" if include_assistant else "none" messaging_experience = str(messaging_experience).strip().lower() if messaging_experience not in {"assistant", "agent", "none"}: - raise ValueError( - "messaging_experience must be one of: assistant, agent, none" - ) - - partial = slack_app_manifest() - slashes = partial["features"]["slash_commands"] + raise ValueError("messaging_experience must be one of: assistant, agent, none") features = { "app_home": { "home_tab_enabled": False, "messages_tab_enabled": True, - "messages_tab_read_only_enabled": False, - }, - "bot_user": { - "display_name": bot_name[:80], - "always_online": True, - }, - "slash_commands": slashes, - } + "messages_tab_read_only_enabled": False}, + "bot_user": {"display_name": bot_name[:80], "always_online": True}, + "slash_commands": slack_app_manifest()["features"]["slash_commands"]} bot_scopes = [ - "app_mentions:read", - "channels:history", - "channels:read", - "chat:write", - "commands", - "files:read", - "files:write", - "groups:history", - "groups:read", - "im:history", - "im:read", - "im:write", - "mpim:history", - "mpim:read", - "reactions:read", - "users:read", - ] - + "app_mentions:read", "channels:history", "channels:read", "chat:write", "commands", + "files:read", "files:write", "groups:history", "groups:read", "im:history", "im:read", + "im:write", "mpim:history", "mpim:read", "reactions:read", "users:read"] bot_events = [ - "app_mention", - "message.channels", - "message.groups", - "message.im", - "message.mpim", - "reaction_added", - "reaction_removed", - ] + "app_mention", "message.channels", "message.groups", "message.im", "message.mpim", + "reaction_added", "reaction_removed"] if messaging_experience == "assistant": features["assistant_view"] = { - "assistant_description": "Chat with Hermes in threads and DMs.", - } + "assistant_description": "Chat with Hermes in threads and DMs."} bot_scopes.append("assistant:write") - bot_events.extend( - [ - "assistant_thread_context_changed", - "assistant_thread_started", - ] - ) + bot_events.extend(["assistant_thread_context_changed", "assistant_thread_started"]) elif messaging_experience == "agent": - features["agent_view"] = { - "agent_description": "Chat with Hermes in Slack Messages.", - } + features["agent_view"] = {"agent_description": "Chat with Hermes in Slack Messages."} bot_scopes.append("assistant:write") - # Slack includes current viewing context in Agent DM events only after - # this subscription is enabled; the adapter consumes that context to - # preserve the referred channel across the agent turn. + # Slack includes current viewing context in Agent DM events only after this subscription + # is enabled; the adapter uses it to preserve the referred channel across the agent turn. bot_events.extend(["app_context_changed", "app_home_opened"]) bot_scopes.sort() @@ -111,97 +60,52 @@ def _build_full_manifest( display_information = { "name": bot_name[:35], "description": (bot_description or "Your Hermes agent on Slack")[:140], - "background_color": "#1a1a2e", - } + "background_color": "#1a1a2e"} if long_description is not None: display_information["long_description"] = long_description return { - "_metadata": { - "major_version": 1, - "minor_version": 1, - }, + "_metadata": {"major_version": 1, "minor_version": 1}, "display_information": display_information, "features": features, - "oauth_config": { - "scopes": { - "bot": bot_scopes, - }, - }, + "oauth_config": {"scopes": {"bot": bot_scopes}}, "settings": { - "event_subscriptions": { - "bot_events": bot_events, - }, - "interactivity": { - "is_enabled": True, - }, + "event_subscriptions": {"bot_events": bot_events}, + "interactivity": {"is_enabled": True}, "org_deploy_enabled": False, "socket_mode_enabled": True, - "token_rotation_enabled": False, - }, - } + "token_rotation_enabled": False}} def slack_manifest_command(args) -> int: - """Print or write a Slack app manifest JSON. - - Flags (all parsed in ``hermes_cli/main.py``): --write [PATH] Write to file instead of stdout - (default path: ``$HERMES_HOME/slack-manifest.json``) --name NAME Override the bot display name - (default: "Hermes") --description DESC Override the bot description --long-description TEXT - Override the long app description (175-4,000 characters) --long-description-file PATH Read the - long app description from a UTF-8 file --slashes-only Emit only the ``features.slash_commands`` - array (for merging into an existing manifest manually) --no-assistant Omit Slack AI Assistant - mode (assistant_view feature, assistant:write scope, assistant_thread_* events) so DMs render as - a flat chat where bare slash commands work inline instead of the Assistant thread pane. - """ + """Print or write a Slack app manifest JSON (flags documented in ``hermes_cli/main.py``).""" name = getattr(args, "name", None) or "Hermes" description = getattr(args, "description", None) or "Your Hermes agent on Slack" long_description = getattr(args, "long_description", None) long_description_file = getattr(args, "long_description_file", None) - if getattr(args, "slashes_only", False) and ( - long_description is not None or long_description_file is not None - ): - print( - "hermes slack manifest: long description options cannot be used " - "with --slashes-only", - file=sys.stderr, - ) + slashes_only = getattr(args, "slashes_only", False) + + def fail(msg: str) -> int: + print(f"hermes slack manifest: {msg}", file=sys.stderr) return 2 + + if slashes_only and (long_description is not None or long_description_file is not None): + return fail("long description options cannot be used with --slashes-only") if long_description_file is not None: source_arg = str(long_description_file) try: - source = Path(source_arg).expanduser() - with source.open("r", encoding="utf-8", newline="") as handle: + with Path(source_arg).expanduser().open("r", encoding="utf-8", newline="") as handle: long_description = handle.read() except (OSError, UnicodeError, RuntimeError) as exc: - print( - f"hermes slack manifest: cannot read long description from " - f"{source_arg}: {exc}", - file=sys.stderr, - ) - return 2 - if ( - long_description is not None - and len(long_description) < SLACK_LONG_DESCRIPTION_MIN_CHARACTERS - ): - print( - "hermes slack manifest: long description must be at least " - f"{SLACK_LONG_DESCRIPTION_MIN_CHARACTERS} characters " - f"(got {len(long_description)})", - file=sys.stderr, - ) - return 2 - if ( - long_description is not None - and len(long_description) > SLACK_LONG_DESCRIPTION_MAX_CHARACTERS - ): - print( - "hermes slack manifest: long description must be at most " - f"{SLACK_LONG_DESCRIPTION_MAX_CHARACTERS} characters " - f"(got {len(long_description)})", - file=sys.stderr, - ) - return 2 + return fail(f"cannot read long description from {source_arg}: {exc}") + if long_description is not None: + n = len(long_description) + if n < SLACK_LONG_DESCRIPTION_MIN_CHARACTERS: + return fail(f"long description must be at least " + f"{SLACK_LONG_DESCRIPTION_MIN_CHARACTERS} characters (got {n})") + if n > SLACK_LONG_DESCRIPTION_MAX_CHARACTERS: + return fail(f"long description must be at most " + f"{SLACK_LONG_DESCRIPTION_MAX_CHARACTERS} characters (got {n})") if getattr(args, "agent_view", False): messaging_experience = "agent" elif getattr(args, "no_assistant", False): @@ -209,45 +113,38 @@ def slack_manifest_command(args) -> int: else: messaging_experience = "assistant" - if getattr(args, "slashes_only", False): + if slashes_only: from hermes_cli.commands import slack_app_manifest manifest = slack_app_manifest()["features"]["slash_commands"] else: manifest = _build_full_manifest( - name, - description, - messaging_experience=messaging_experience, - long_description=long_description, - ) - + name, description, messaging_experience=messaging_experience, + long_description=long_description) payload = json.dumps(manifest, indent=2, ensure_ascii=False) + "\n" write_target = getattr(args, "write", None) - if write_target is not None: - if isinstance(write_target, bool) and write_target: - # --write with no value → default location - from hermes_constants import get_hermes_home - - target = Path(get_hermes_home()) / "slack-manifest.json" - else: - target = Path(write_target).expanduser() - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(payload, encoding="utf-8") - print(f"Slack manifest written to: {target}", file=sys.stderr) - print( - "\nNext steps:\n" - " 1. Open https://api.slack.com/apps and pick your Hermes app\n" - " (or create a new one: Create New App → From an app manifest).\n" - f" 2. Features → App Manifest → paste the contents of\n" - f" {target}\n" - " 3. Save; Slack will prompt to reinstall the app if scopes or\n" - " slash commands changed.\n" - " 4. Make sure Socket Mode is enabled and you have a bot token\n" - " (xoxb-...) and app token (xapp-...) configured via\n" - " `hermes setup`.\n", - file=sys.stderr, - ) - else: + if write_target is None: sys.stdout.write(payload) + return 0 + if isinstance(write_target, bool) and write_target: # bare --write → default location + from hermes_constants import get_hermes_home + + target = Path(get_hermes_home()) / "slack-manifest.json" + else: + target = Path(write_target).expanduser() + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(payload, encoding="utf-8") + print(f"Slack manifest written to: {target}", file=sys.stderr) + print( + "\nNext steps:\n" + " 1. Open https://api.slack.com/apps and pick your Hermes app\n" + " (or create a new one: Create New App → From an app manifest).\n" + f" 2. Features → App Manifest → paste the contents of\n" + f" {target}\n" + " 3. Save; Slack will prompt to reinstall the app if scopes or\n" + " slash commands changed.\n" + " 4. Make sure Socket Mode is enabled and you have a bot token\n" + " (xoxb-...) and app token (xapp-...) configured via\n" + " `hermes setup`.\n", file=sys.stderr) return 0 diff --git a/hermes_cli/slash_exec.py b/hermes_cli/slash_exec.py index 28c8163372..3536537ee0 100644 --- a/hermes_cli/slash_exec.py +++ b/hermes_cli/slash_exec.py @@ -11,20 +11,10 @@ from collections.abc import Callable, Mapping from dataclasses import dataclass, field from typing import Any -__all__ = [ - "CommandContext", - "CommandReply", - "EXECUTORS", - "execute_command", - "resolve_executor", - "run_execute", -] +__all__ = ["CommandContext", "CommandReply", "EXECUTORS", "execute_command", "resolve_executor", + "run_execute"] -# --------------------------------------------------------------------------- -# Context / reply dataclasses -# --------------------------------------------------------------------------- - @dataclass(frozen=True) class CommandContext: """Surface-provided inputs for a shared command executor.""" @@ -37,22 +27,15 @@ class CommandContext: @dataclass(frozen=True) class CommandReply: - """Canonical result of a shared executor. - - ``text`` is the surface-independent core text. ``data`` carries the structured values the - executor derived so a surface may re-render them with its own decoration (Rich columns, markdown - bullets) without duplicating the computation. ``format`` is a rendering hint only. - """ + """Canonical result of a shared executor: surface-independent ``text`` plus the structured + ``data`` it derived so a surface can re-render with its own decoration.""" text: str data: Mapping[str, Any] = field(default_factory=dict) format: str = "plain" # "plain" | "markdown" (hint, not a contract) -# --------------------------------------------------------------------------- -# Executors — pure formatters, no agent/session mutation -# --------------------------------------------------------------------------- - +# Executors — pure formatters, no agent/session mutation. def _exec_version(ctx: CommandContext) -> CommandReply: """Core /version text — the banner version label.""" from hermes_cli.banner import format_banner_version_label @@ -70,13 +53,11 @@ def _exec_egress(ctx: CommandContext) -> CommandReply: def _exec_profile(ctx: CommandContext) -> CommandReply: """Core /profile data — active profile name + home directory. - A multiplexed gateway may pre-resolve the per-source profile/home and pass them via - ``options`` (``profile_name`` / ``home_display``); otherwise process-level values are used, - matching the old CLI and non-multiplex gateway behavior. + A multiplexed gateway may pre-resolve the per-source profile/home via ``options`` + (``profile_name`` / ``home_display``); otherwise process-level values are used. """ profile_name = str(ctx.options.get("profile_name") or "").strip() home_display = str(ctx.options.get("home_display") or "").strip() - if not profile_name: from hermes_cli.profiles import get_active_profile_name @@ -85,26 +66,17 @@ def _exec_profile(ctx: CommandContext) -> CommandReply: from hermes_constants import display_hermes_home home_display = display_hermes_home() - - # Presentation-only display name (profile.yaml). `data.profile` stays - # the canonical id — consumers route on it; only the text gets the label. + # Presentation-only display name (profile.yaml); `data.profile` stays the canonical id. label = profile_name try: - from hermes_cli.profiles import ( - format_profile_label, - get_profile_dir, - read_profile_meta, - ) + from hermes_cli.profiles import format_profile_label, get_profile_dir, read_profile_meta display = read_profile_meta(get_profile_dir(profile_name)).get("display_name", "") label = format_profile_label(profile_name, display) except Exception: pass - - return CommandReply( - f"Profile: {label}\nHome: {home_display}", - data={"profile": profile_name, "home": home_display}, - ) + return CommandReply(f"Profile: {label}\nHome: {home_display}", + data={"profile": profile_name, "home": home_display}) def _exec_bundles(ctx: CommandContext) -> CommandReply: @@ -112,11 +84,7 @@ def _exec_bundles(ctx: CommandContext) -> CommandReply: try: from agent.skill_bundles import _bundles_dir, list_bundles except Exception as exc: # pragma: no cover - env-specific - return CommandReply( - f"Bundles subsystem unavailable: {exc}", - data={"error": str(exc)}, - ) - + return CommandReply(f"Bundles subsystem unavailable: {exc}", data={"error": str(exc)}) bundles = list_bundles() bundles_dir = str(_bundles_dir()) if not bundles: @@ -124,21 +92,15 @@ def _exec_bundles(ctx: CommandContext) -> CommandReply: "No skill bundles installed.\n" "Create one with: hermes bundles create --skill --skill \n" f"Directory: {bundles_dir}", - data={"bundles": [], "dir": bundles_dir}, - ) - + data={"bundles": [], "dir": bundles_dir}) lines = [f"Skill Bundles ({len(bundles)} installed):"] for info in bundles: skill_count = len(info.get("skills", [])) desc = info.get("description") or f"Load {skill_count} skills" lines.append(f"/{info['slug']} — {desc} ({skill_count} skills)") - for s in info.get("skills", []): - lines.append(f" · {s}") + lines.extend(f" · {s}" for s in info.get("skills", [])) lines.append("Invoke a bundle with / to load all its skills.") - return CommandReply( - "\n".join(lines), - data={"bundles": bundles, "dir": bundles_dir}, - ) + return CommandReply("\n".join(lines), data={"bundles": bundles, "dir": bundles_dir}) def _skill_commands() -> dict: @@ -156,18 +118,14 @@ def _exec_help(ctx: CommandContext) -> CommandReply: from agent.i18n import t from hermes_cli.commands import gateway_help_lines - lines = [ - t("gateway.help.header"), - *gateway_help_lines(), - ] + lines = [t("gateway.help.header"), *gateway_help_lines()] skill_cmds = _skill_commands() try: if skill_cmds: lines.append(t("gateway.help.skill_header", count=len(skill_cmds))) - # Show first 10, then point to /commands for the rest - sorted_cmds = sorted(skill_cmds) - for cmd in sorted_cmds[:10]: - lines.append(f"`{cmd}` — {skill_cmds[cmd]['description']}") + sorted_cmds = sorted(skill_cmds) # first 10, then point to /commands for the rest + lines.extend(f"`{cmd}` — {skill_cmds[cmd]['description']}" + for cmd in sorted_cmds[:10]) if len(sorted_cmds) > 10: lines.append(t("gateway.help.more_use_commands", count=len(sorted_cmds) - 10)) except Exception: @@ -178,8 +136,7 @@ def _exec_help(ctx: CommandContext) -> CommandReply: def _exec_commands(ctx: CommandContext) -> CommandReply: """Core gateway /commands body — paginated command + skill listing. - ``ctx.options["page_size"]`` is a surface parameter (Telegram uses 15, everything else 20) — for - a fixed context the text is surface-invariant. + ``ctx.options["page_size"]`` is a surface parameter (Telegram uses 15, everything else 20). """ from agent.i18n import t from hermes_cli.commands import gateway_help_lines @@ -193,7 +150,6 @@ def _exec_commands(ctx: CommandContext) -> CommandReply: else: requested_page = 1 - # Build combined entry list: built-in commands + skill commands entries = list(gateway_help_lines()) skill_cmds = _skill_commands() try: @@ -216,13 +172,8 @@ def _exec_commands(ctx: CommandContext) -> CommandReply: total_pages = max(1, (len(entries) + page_size - 1) // page_size) page = max(1, min(requested_page, total_pages)) start = (page - 1) * page_size - page_entries = entries[start:start + page_size] - - lines = [ - t("gateway.commands.header", total=len(entries), page=page, total_pages=total_pages), - "", - *page_entries, - ] + lines = [t("gateway.commands.header", total=len(entries), page=page, total_pages=total_pages), + "", *entries[start:start + page_size]] if total_pages > 1: nav_parts = [] if page > 1: @@ -235,18 +186,13 @@ def _exec_commands(ctx: CommandContext) -> CommandReply: return CommandReply("\n".join(lines), format="markdown") -# --------------------------------------------------------------------------- -# Registry + resolution -# --------------------------------------------------------------------------- - EXECUTORS: dict[str, Callable[[CommandContext], CommandReply]] = { "version": _exec_version, "egress": _exec_egress, "profile": _exec_profile, "bundles": _exec_bundles, "gateway_help": _exec_help, - "gateway_commands": _exec_commands, -} + "gateway_commands": _exec_commands} def resolve_executor(cmd_def: Any) -> Callable[[CommandContext], CommandReply] | None: @@ -261,11 +207,7 @@ def run_execute(cmd_def: Any, ctx: CommandContext) -> CommandReply | None: def execute_command(name: str, ctx: CommandContext) -> CommandReply: - """Run the shared executor for the command named ``name``. - - Raises ``LookupError`` when the command is unknown or not migrated — call sites use this only - for commands they know carry ``execute``. - """ + """Run the shared executor for ``name``; ``LookupError`` when unknown or not migrated.""" from hermes_cli.commands import resolve_command cmd_def = resolve_command(name) diff --git a/hermes_cli/suggestions_cmd.py b/hermes_cli/suggestions_cmd.py index 6ad01104a9..2eaa90a7c7 100644 --- a/hermes_cli/suggestions_cmd.py +++ b/hermes_cli/suggestions_cmd.py @@ -1,13 +1,7 @@ """Shared ``/suggestions`` command logic for CLI and gateway. -Both surfaces call ``handle_suggestions_command(args, origin=...)`` and present the returned text -however they present command output. Keeping the logic here (not in cli.py / gateway/run.py) means -the two surfaces can never drift. - -Subcommands: /suggestions list pending suggestions (numbered) /suggestions accept create the -cron job for that suggestion /suggestions dismiss dismiss it (latched, never re-offered) -/suggestions catalog seed the curated starter automations as pending /suggestions clear drop -accepted records (housekeeping) +Both surfaces call ``handle_suggestions_command(args, origin=...)`` and present the returned text, +so the two surfaces can never drift. Subcommands are listed in ``_USAGE``. """ from __future__ import annotations @@ -20,17 +14,13 @@ logger = logging.getLogger(__name__) def _fmt_pending(pending: list) -> str: if not pending: - return ( - "No suggested automations right now.\n" - "Try `/suggestions catalog` to see the curated starter set, or " - "install a blueprint skill to get one." - ) + return ("No suggested automations right now.\n" + "Try `/suggestions catalog` to see the curated starter set, or " + "install a blueprint skill to get one.") lines = ["Suggested automations — `/suggestions accept N` or `dismiss N`:\n"] for i, s in enumerate(pending, 1): - spec = s.get("job_spec", {}) or {} - sched = spec.get("schedule", "?") - src = s.get("source", "?") - lines.append(f" {i}. {s.get('title', '(untitled)')} [{sched}] ({src})") + sched = (s.get("job_spec", {}) or {}).get("schedule", "?") + lines.append(f" {i}. {s.get('title', '(untitled)')} [{sched}] ({s.get('source', '?')})") desc = s.get("description", "").strip() if desc: lines.append(f" {desc}") @@ -38,12 +28,8 @@ def _fmt_pending(pending: list) -> str: def _resolve_origin() -> Optional[Dict[str, Any]]: - """Best-effort current-chat origin from session env (CLI and gateway both set it). - - Mirrors cron's ``_origin_from_env`` so an accepted suggestion's job delivers back to the chat - where it was accepted. Returns None if unavailable, in which case create_job falls back to a - configured home channel. - """ + """Best-effort current-chat origin from session env (mirrors cron's ``_origin_from_env``) so an + accepted job delivers back to the accepting chat; None lets create_job use the home channel.""" try: from gateway.session_context import get_session_env @@ -54,8 +40,7 @@ def _resolve_origin() -> Optional[Dict[str, Any]]: "platform": platform, "chat_id": chat_id, "chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None, - "thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None, - } + "thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None} except Exception: pass return None @@ -74,11 +59,8 @@ def _accept(store, rest: str, origin, surface: str) -> str: return f"No pending suggestion matches '{rest}'. Run /suggestions to list them." sched = job.get("schedule_display") or (job.get("job_spec", {}) or {}).get("schedule", "") name = job.get("name", "automation") - manage = ( - "Manage it with /cron." - if surface == "cli" - else "Ask me to list, pause, or remove it any time." - ) + manage = ("Manage it with /cron." if surface == "cli" + else "Ask me to list, pause, or remove it any time.") return f"Scheduled '{name}'" + (f" ({sched})" if sched else "") + f". {manage}" @@ -99,10 +81,8 @@ def _catalog(store, rest: str, origin, surface: str) -> str: logger.debug("catalog seed failed: %s", e) return "Couldn't load the catalog." if not created: - return ( - "No new catalog automations to add (already offered, dismissed, " - "or your suggestion list is full). Run /suggestions to see pending." - ) + return ("No new catalog automations to add (already offered, dismissed, " + "or your suggestion list is full). Run /suggestions to see pending.") added = ", ".join(c.get("title", "?") for c in created) return f"Added {len(created)} suggestion(s): {added}.\nRun /suggestions to review." @@ -113,15 +93,9 @@ def _clear(store, rest: str, origin, surface: str) -> str: _SUBCOMMANDS = { "": lambda store, rest, origin, surface: _fmt_pending(store.list_pending()), - "accept": _accept, - "add": _accept, - "schedule": _accept, - "dismiss": _dismiss, - "no": _dismiss, - "reject": _dismiss, - "catalog": _catalog, - "clear": _clear, -} + "accept": _accept, "add": _accept, "schedule": _accept, + "dismiss": _dismiss, "no": _dismiss, "reject": _dismiss, + "catalog": _catalog, "clear": _clear} _USAGE = ( "Usage:\n" @@ -129,22 +103,13 @@ _USAGE = ( " /suggestions accept N schedule suggestion N\n" " /suggestions dismiss N dismiss suggestion N\n" " /suggestions catalog add curated starter automations\n" - " /suggestions clear housekeeping" -) + " /suggestions clear housekeeping") def handle_suggestions_command( - args: str, - *, - origin: Optional[Dict[str, Any]] = None, - surface: str = "cli", -) -> str: - """Dispatch a ``/suggestions`` invocation. Returns text to show the user. - - ``args`` is everything after ``/suggestions`` (already stripped of the command word). ``origin`` - is the platform/chat dict so an accepted job's "origin" delivery routes back to where the user - accepted; when omitted it is resolved from the session environment. - """ + args: str, *, origin: Optional[Dict[str, Any]] = None, surface: str = "cli") -> str: + """Dispatch a ``/suggestions`` invocation (``args`` = text after the command word); returns + text to show the user. ``origin`` defaults to the session environment's chat.""" if origin is None: origin = _resolve_origin() try: @@ -154,10 +119,7 @@ def handle_suggestions_command( return "Suggestions are unavailable in this build." parts = (args or "").strip().split() - sub = parts[0].lower() if parts else "" - rest = " ".join(parts[1:]).strip() - - handler = _SUBCOMMANDS.get(sub) + handler = _SUBCOMMANDS.get(parts[0].lower() if parts else "") if handler is None: return _USAGE - return handler(store, rest, origin, surface) + return handler(store, " ".join(parts[1:]).strip(), origin, surface) diff --git a/hermes_cli/telegram_managed_bot.py b/hermes_cli/telegram_managed_bot.py index 1d625a973b..44551067d5 100644 --- a/hermes_cli/telegram_managed_bot.py +++ b/hermes_cli/telegram_managed_bot.py @@ -1,9 +1,5 @@ -"""Telegram Managed Bot onboarding client. - -Uses Telegram's Managed Bots feature to create a user-owned child bot without manual BotFather token -copy-paste. Hermes talks only to the Nous onboarding service; the raw Telegram token is saved -locally after one-time retrieval. -""" +"""Telegram Managed Bot onboarding client: creates a user-owned child bot via the Nous onboarding +service (no BotFather copy-paste); the raw Telegram token is saved locally after one retrieval.""" from __future__ import annotations @@ -17,15 +13,12 @@ from typing import Optional import httpx -# Default pairing API base URL (Nous-hosted Cloudflare Worker). -# Override for PoC/staging with TELEGRAM_ONBOARDING_URL. +# Nous-hosted pairing API; override for PoC/staging with TELEGRAM_ONBOARDING_URL. DEFAULT_API_URL = "https://setup.hermes-agent.nousresearch.com" TELEGRAM_ONBOARDING_URL_ENV = "TELEGRAM_ONBOARDING_URL" - -# The Nous-hosted manager bot username (without @). The backend returns the -# actual deep link, so this is only used by local helpers/tests. +# Manager bot username (without @); the backend returns the real deep link, so this is only +# used by local helpers/tests. DEFAULT_MANAGER_BOT = "HermesSetupBot" - DEFAULT_BOT_NAME = "Hermes Agent" DEFAULT_POLL_TIMEOUT = 180 POLL_INTERVAL = 2 @@ -37,7 +30,6 @@ _TELEGRAM_BOT_TOKEN_RE = re.compile(r"^\d+:[A-Za-z0-9_-]{30,}$") @dataclass(frozen=True) class TelegramPairing: """Pairing record returned by the Telegram onboarding service.""" - pairing_id: str poll_token: str suggested_username: str @@ -49,7 +41,6 @@ class TelegramPairing: @dataclass(frozen=True) class TelegramBotSetupResult: """Successful Telegram onboarding result returned by the setup service.""" - token: str bot_username: str | None = None owner_user_id: int | None = None @@ -77,7 +68,6 @@ def render_qr_terminal(url: str) -> str: """Render a URL as a QR code string suitable for terminal output.""" try: import io - import qrcode # type: ignore[import-untyped] except ImportError: return "" @@ -102,11 +92,8 @@ def generate_username_slug(length: int = 16) -> str: def generate_bot_username(profile_name: Optional[str] = None) -> str: - """Generate a secure suggested bot username like ``hermes__bot``. - - ``profile_name`` is accepted for backward compatibility but intentionally not embedded: the - username must carry enough entropy for backend correlation. - """ + """Suggested bot username ``hermes__bot``; ``profile_name`` is accepted for backward + compatibility but not embedded (the slug must carry the entropy for backend correlation).""" _ = profile_name return f"hermes_{generate_username_slug()}_bot" @@ -114,11 +101,8 @@ def generate_bot_username(profile_name: Optional[str] = None) -> str: def create_pairing( api_url: str | None = None, bot_name: str = DEFAULT_BOT_NAME, timeout: float = 10.0 ) -> TelegramPairing | None: - """Create a Telegram onboarding pairing. - - ``POST /v1/telegram/pairings`` returns the deep link, QR payload, public pairing id, and secret - poll token. The token is only used as a bearer credential while polling. - """ + """POST a pairing; returns deep link, QR payload, public id and the secret poll token (used + only as a bearer credential while polling).""" try: resp = httpx.post(f"{_api_url(api_url)}/v1/telegram/pairings", json={"bot_name": bot_name}, timeout=timeout) if resp.status_code not in (200, 201): @@ -126,35 +110,25 @@ def create_pairing( data = resp.json() except (httpx.HTTPError, ValueError): return None - required = ("pairing_id", "poll_token", "suggested_username", "deep_link") if not all(isinstance(data.get(key), str) and data.get(key) for key in required): return None - qr_payload = data.get("qr_payload") or data["deep_link"] if not isinstance(qr_payload, str): return None - expires_at = data.get("expires_at") - return TelegramPairing( - *(data[key] for key in required), - qr_payload=qr_payload, - expires_at=expires_at if isinstance(expires_at, str) else None, - ) + return TelegramPairing(*(data[key] for key in required), qr_payload=qr_payload, + expires_at=expires_at if isinstance(expires_at, str) else None) def poll_pairing_result_once( api_url: str | None, pairing: TelegramPairing, timeout: float = 10.0 ) -> TelegramBotSetupResult | None: """Poll the onboarding service once. Returns setup metadata when ready.""" - resp = httpx.get( - f"{_api_url(api_url)}/v1/telegram/pairings/{pairing.pairing_id}", - headers={"Authorization": f"Bearer {pairing.poll_token}"}, - timeout=timeout, - ) + resp = httpx.get(f"{_api_url(api_url)}/v1/telegram/pairings/{pairing.pairing_id}", + headers={"Authorization": f"Bearer {pairing.poll_token}"}, timeout=timeout) if resp.status_code != 200: return None - data = resp.json() token = data.get("token") if data.get("status") != "ready" or not is_valid_telegram_bot_token(token): @@ -163,8 +137,7 @@ def poll_pairing_result_once( return TelegramBotSetupResult( token=token, bot_username=bot_username if isinstance(bot_username, str) and bot_username else None, - owner_user_id=_parse_owner_user_id(data.get("owner_user_id")), - ) + owner_user_id=_parse_owner_user_id(data.get("owner_user_id"))) def _try_poll(api_url: str | None, pairing: TelegramPairing) -> TelegramBotSetupResult | None: @@ -176,11 +149,8 @@ def _try_poll(api_url: str | None, pairing: TelegramPairing) -> TelegramBotSetup def poll_for_setup_result( - api_url: str | None, - pairing: TelegramPairing, - timeout: float = DEFAULT_POLL_TIMEOUT, - interval: float = POLL_INTERVAL, -) -> Optional[TelegramBotSetupResult]: + api_url: str | None, pairing: TelegramPairing, timeout: float = DEFAULT_POLL_TIMEOUT, + interval: float = POLL_INTERVAL) -> Optional[TelegramBotSetupResult]: """Poll the pairing API until setup metadata is available or timeout.""" deadline = time.monotonic() + timeout while time.monotonic() < deadline: @@ -191,10 +161,8 @@ def poll_for_setup_result( def auto_setup_telegram_bot_result( - api_url: str | None = None, - manager_bot: str = DEFAULT_MANAGER_BOT, - profile_name: Optional[str] = None, - poll_timeout: float = DEFAULT_POLL_TIMEOUT, + api_url: str | None = None, manager_bot: str = DEFAULT_MANAGER_BOT, + profile_name: Optional[str] = None, poll_timeout: float = DEFAULT_POLL_TIMEOUT, ) -> Optional[TelegramBotSetupResult]: """Run the full automatic Telegram bot creation flow.""" _ = manager_bot, profile_name @@ -221,15 +189,12 @@ def auto_setup_telegram_bot_result( start = time.monotonic() deadline = start + poll_timeout idx = 0 - while time.monotonic() < deadline: char = spinner_chars[idx % len(spinner_chars)] - elapsed = int(time.monotonic() - start) - remaining = max(0, int(poll_timeout - elapsed)) + remaining = max(0, int(poll_timeout - int(time.monotonic() - start))) sys.stdout.write(f"\r {char} Waiting for bot creation... ({remaining}s remaining) ") sys.stdout.flush() idx += 1 - if result := _try_poll(resolved_api_url, pairing): sys.stdout.write("\r ✓ Bot created successfully! \n") sys.stdout.flush()