From 3494f7cf232f67b7eae26322d1e5abb1594cce11 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:58:31 -0700 Subject: [PATCH] refactor(hermes_cli): AST-neutral closer hugging across r3-17 slice files --- hermes_cli/build_info.py | 3 +- hermes_cli/bundles.py | 28 ++-- hermes_cli/checkpoints.py | 3 +- hermes_cli/claw.py | 9 +- hermes_cli/cli_agent_setup_mixin.py | 6 +- hermes_cli/cli_info_mixin.py | 3 +- hermes_cli/cli_model_switch_mixin.py | 129 ++++++----------- hermes_cli/cli_session_mixin.py | 141 +++++++------------ hermes_cli/cli_status_bar_mixin.py | 48 +++---- hermes_cli/cli_stream_mixin.py | 72 ++++------ hermes_cli/cli_tui_mixin.py | 48 +++---- hermes_cli/cli_voice_mixin.py | 33 ++--- hermes_cli/clipboard.py | 7 +- hermes_cli/codex_models.py | 15 +- hermes_cli/codex_runtime_plugin_migration.py | 6 +- 15 files changed, 182 insertions(+), 369 deletions(-) diff --git a/hermes_cli/build_info.py b/hermes_cli/build_info.py index 3fc7e8eb8a..b18aa12de5 100644 --- a/hermes_cli/build_info.py +++ b/hermes_cli/build_info.py @@ -103,8 +103,7 @@ def get_code_identity(refresh: bool = False) -> dict: "sha": sha, "short_sha": sha[:8] if sha else None, "version": version, - "source": source, - } + "source": source} return dict(_code_identity_cache) diff --git a/hermes_cli/bundles.py b/hermes_cli/bundles.py index dc92b0af91..0a8cacc362 100644 --- a/hermes_cli/bundles.py +++ b/hermes_cli/bundles.py @@ -31,8 +31,7 @@ def _cmd_list(args) -> None: c.print( f"[dim]No bundles installed yet. Create one with:\n" f" hermes bundles create --skill skill1 --skill skill2[/]\n" - f"Bundles directory: [bold]{_bundles_dir()}[/]" - ) + f"Bundles directory: [bold]{_bundles_dir()}[/]") return table = Table(title=f"Skill Bundles ({len(bundles)})", show_lines=False) @@ -46,8 +45,7 @@ def _cmd_list(args) -> None: f"/{info['slug']}", info["name"], str(len(info.get("skills", []))), - info.get("description") or "", - ) + info.get("description") or "") c.print(table) c.print(f"\n[dim]Bundles directory: {_bundles_dir()}[/]") @@ -76,8 +74,7 @@ def _cmd_create(args) -> None: # Interactive prompt for skills if none were passed on the CLI. c.print( "[dim]No skills passed via --skill. Enter one skill name per line.\n" - "Submit an empty line to finish.[/]" - ) + "Submit an empty line to finish.[/]") try: while True: line = line_input("skill> ").strip() @@ -92,8 +89,7 @@ def _cmd_create(args) -> None: try: path = save_bundle( name, skills, description=args.description or "", instruction=args.instruction or "", - overwrite=bool(args.force), - ) + overwrite=bool(args.force)) except FileExistsError as exc: _fail(c, f"[bold red]{exc}[/]\n[dim]Pass --force to overwrite.[/]") except ValueError as exc: @@ -104,8 +100,7 @@ def _cmd_create(args) -> None: if info: c.print( f" Invoke with: [bold cyan]/{info['slug']}[/] " - f"(loads {len(info['skills'])} skills)" - ) + f"(loads {len(info['skills'])} skills)") def _cmd_delete(args) -> None: @@ -151,22 +146,17 @@ def register_cli(subparser) -> None: help="Create a new skill bundle", description=( "Create a new bundle. Skills can be passed via --skill (repeat for " - "multiple) or entered interactively when omitted." - ), - ) + "multiple) or entered interactively when omitted.")) p_create.add_argument("name", help="Bundle name (becomes the /slash command)") p_create.add_argument( "--skill", "-s", action="append", default=[], - help="Skill name to include (repeat for multiple)", - ) + help="Skill name to include (repeat for multiple)") p_create.add_argument( "--description", "-d", default="", - help="Human-readable description shown in /help and `hermes bundles list`", - ) + help="Human-readable description shown in /help and `hermes bundles list`") p_create.add_argument( "--instruction", "-i", default="", - help="Extra guidance prepended to the loaded skill content", - ) + help="Extra guidance prepended to the loaded skill content") p_create.add_argument( "--force", "-f", action="store_true", help="Overwrite an existing bundle with the same name" ) diff --git a/hermes_cli/checkpoints.py b/hermes_cli/checkpoints.py index 2d1f29c372..72388936da 100644 --- a/hermes_cli/checkpoints.py +++ b/hermes_cli/checkpoints.py @@ -109,8 +109,7 @@ def cmd_prune(args: argparse.Namespace) -> int: retention_days=args.retention_days, delete_orphans=delete_orphans, max_total_size_mb=args.max_size_mb, - orphan_allowlist=orphan_allowlist, - ) + orphan_allowlist=orphan_allowlist) print(f"Scanned: {result['scanned']}") print(f"Deleted orphan: {result['deleted_orphan']}") print(f"Deleted stale: {result['deleted_stale']}") diff --git a/hermes_cli/claw.py b/hermes_cli/claw.py index 5f83cf28b6..def2dee09b 100644 --- a/hermes_cli/claw.py +++ b/hermes_cli/claw.py @@ -32,16 +32,14 @@ _OPENCLAW_DIR_NAMES = (".openclaw", ".clawdbot", ".moltbot") _MIGRATE_ARG_DEFAULTS = ( ("source", None), ("dry_run", False), ("preset", "full"), ("overwrite", False), ("migrate_secrets", False), ("workspace_target", None), ("skill_conflict", "skip"), - ("no_backup", False), ("yes", False), -) + ("no_backup", False), ("yes", False)) # (status, heading, color, default reason) — printed in this order after migrated items. _REPORT_REASON_GROUPS = ( ("conflict", " ⚠ Conflicts (skipped — use --overwrite to force):", Colors.YELLOW, "already exists"), ("skipped", " ─ Skipped:", Colors.DIM, ""), - ("error", " ✗ Errors:", Colors.RED, "unknown error"), -) + ("error", " ✗ Errors:", Colors.RED, "unknown error")) # Summary-line counters after the migrated count: (summary key, label). _SUMMARY_COUNT_LABELS = (("conflict", "conflict(s)"), ("skipped", "skipped"), ("error", "error(s)")) @@ -49,8 +47,7 @@ _SUMMARY_COUNT_LABELS = (("conflict", "conflict(s)"), ("skipped", "skipped"), (" _WORKSPACE_MARKERS = ("todo.json", "SOUL.md", "MEMORY.md", "USER.md") _WORKSPACE_ITEM_LABELS = ( ("todo.json", "todo.json", Path.exists), ("sessions", "sessions/", Path.is_dir), - ("SOUL.md", "SOUL.md", Path.exists), ("MEMORY.md", "MEMORY.md", Path.exists), -) + ("SOUL.md", "SOUL.md", Path.exists), ("MEMORY.md", "MEMORY.md", Path.exists)) def _print_banner(title: str) -> None: diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index 237287c5ec..e6b93b8afd 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -93,8 +93,7 @@ def _tool_calls_summary(tool_calls) -> str: _RESUME_EVENT_TEXT = { "model_switch": "model changed", "async_delegation_complete": "background delegation completed", - "auto_continue": "resumed interrupted turn", -} + "auto_continue": "resumed interrupted turn"} def _collect_resume_entries(display_history, disp: dict, clean_assistant): """Displayable ``(role, text)`` recap entries from stored history, truncated per the @@ -153,8 +152,7 @@ def _collect_resume_entries(display_history, disp: dict, clean_assistant): # (skin key, fallback) for recap panel colors: body text, session label, border, assistant label. _RESUME_SKIN_COLORS = ( ("banner_text", "#FFF8DC"), ("session_label", "#DAA520"), ("session_border", "#8B8682"), - ("ui_ok", "#8FBC8F"), -) + ("ui_ok", "#8FBC8F")) def _resume_panel_colors() -> tuple: diff --git a/hermes_cli/cli_info_mixin.py b/hermes_cli/cli_info_mixin.py index f938b66c6a..154fa33022 100644 --- a/hermes_cli/cli_info_mixin.py +++ b/hermes_cli/cli_info_mixin.py @@ -863,8 +863,7 @@ class CLIInfoMixin: diff = { "Added": connected_servers - old_servers, "Removed": old_servers - connected_servers, - "Reconnected": connected_servers & old_servers, - } + "Reconnected": connected_servers & old_servers} for label, icon in (("Reconnected", "♻️ "), ("Added", "➕"), ("Removed", "➖")): if diff[label]: print(f" {icon} {label}: {', '.join(sorted(diff[label]))}") diff --git a/hermes_cli/cli_model_switch_mixin.py b/hermes_cli/cli_model_switch_mixin.py index 20906f0ed2..7b0ce6403e 100644 --- a/hermes_cli/cli_model_switch_mixin.py +++ b/hermes_cli/cli_model_switch_mixin.py @@ -21,8 +21,7 @@ from utils import base_url_host_matches # override and restored wholesale on rollback. _RUNTIME_FIELDS = ( "model", "provider", "requested_provider", "_explicit_api_key", "_explicit_base_url", - "api_key", "base_url", "api_mode", -) + "api_key", "base_url", "api_mode") def _runtime_fields(cli) -> dict: @@ -57,8 +56,7 @@ def _merge_preflight_warning(cli, result, custom_providers) -> None: messages=list(cli.conversation_history or []), custom_providers=custom_providers if custom_providers is not None else getattr(cli.agent, "_custom_providers", None), - config_context_length=getattr(cli.agent, "_config_context_length", None), - ) + config_context_length=getattr(cli.agent, "_config_context_length", None)) except Exception as exc: logger.debug("preflight-compression switch warning failed: %s", exc) @@ -78,8 +76,7 @@ def _print_switch_summary(cli, result, old_model, *, one_turn: bool, strict_cont f"[Note: model was just switched from {_display_old} to {_display_new} " f"via {result.provider_label or result.target_provider}. " f"{'This override applies to the next turn only. ' if one_turn else ''}" - f"Adjust your self-identification accordingly.]" - ) + f"Adjust your self-identification accordingly.]") _cprint(f" ✓ Model switched: {_display_new}") _cprint(f" Provider: {result.provider_label or result.target_provider}") @@ -93,8 +90,7 @@ def _print_switch_summary(cli, result, old_model, *, one_turn: bool, strict_cont base_url=result.base_url or cli.base_url or "", api_key=result.api_key or cli.api_key or "", model_info=mi, config_context_length=getattr(agent, "_config_context_length", None) if agent else None, - custom_providers=getattr(agent, "_custom_providers", None) if agent else None, - ) + custom_providers=getattr(agent, "_custom_providers", None) if agent else None) except Exception: if strict_context: raise @@ -107,8 +103,7 @@ def _print_switch_summary(cli, result, old_model, *, one_turn: bool, strict_cont _cprint(f" Capabilities: {mi.format_capabilities()}") cache_enabled = ( (base_url_host_matches(result.base_url or "", "openrouter.ai") and "claude" in result.new_model.lower()) - or result.api_mode == "anthropic_messages" - ) + or result.api_mode == "anthropic_messages") if cache_enabled: _cprint(" Prompt caching: enabled") if result.warning_message: @@ -116,16 +111,14 @@ def _print_switch_summary(cli, result, old_model, *, one_turn: bool, strict_cont def _switch_model_from( - cli, raw_input, *, is_global, explicit_provider, user_providers, custom_providers -): + cli, raw_input, *, is_global, explicit_provider, user_providers, custom_providers): """``switch_model`` seeded with this CLI's live route.""" from hermes_cli.model_switch import switch_model return switch_model( raw_input=raw_input, current_provider=cli.provider or "", current_model=cli.model or "", current_base_url=cli.base_url or "", current_api_key=cli.api_key or "", is_global=is_global, explicit_provider=explicit_provider, user_providers=user_providers, - custom_providers=custom_providers, - ) + custom_providers=custom_providers) def _run_confirm_and_apply(cli, target, *args) -> None: @@ -142,8 +135,7 @@ def _run_confirm_and_apply(cli, target, *args) -> None: def _commit_model_switch( - cli, result, *, persist_global: bool, one_turn: bool = False, picker: bool = False -) -> None: + cli, result, *, persist_global: bool, one_turn: bool = False, picker: bool = False) -> None: """Stage + swap, print the summary, persist (session row unless --once; config on --global). ``picker``: tolerate context-resolution errors and label the config write "(--global)"; the typed path additionally records the one-turn restore snapshot.""" @@ -207,8 +199,7 @@ def _show_model_picker(cli, ctx, force_refresh: bool) -> None: cli._open_model_picker( providers, cli.model or "unknown", get_label(cli.provider) if cli.provider else "unknown", user_provs=ctx.user_providers if ctx is not None else None, - custom_provs=ctx.custom_providers if ctx is not None else None, - ) + custom_provs=ctx.custom_providers if ctx is not None else None) class CLIModelSwitchMixin: @@ -247,15 +238,12 @@ class CLIModelSwitchMixin: try: from hermes_cli.model_normalize import ( - _AGGREGATOR_PROVIDERS, normalize_model_for_provider - ) + _AGGREGATOR_PROVIDERS, normalize_model_for_provider) if resolved_provider not in _AGGREGATOR_PROVIDERS: _adopt( normalize_model_for_provider(current_model, resolved_provider), lambda new: ( - f"Normalized model '{current_model}' to '{new}' for {resolved_provider}." - ), - ) + f"Normalized model '{current_model}' to '{new}' for {resolved_provider}.")) except Exception: pass @@ -264,8 +252,7 @@ class CLIModelSwitchMixin: return _adopt_with_mode( lambda m: normalize_copilot_model_id(m, api_key=self.api_key), lambda m: copilot_model_api_mode(m, api_key=self.api_key), - lambda new: f"Normalized Copilot model '{current_model}' to '{new}'.", - ) + lambda new: f"Normalized Copilot model '{current_model}' to '{new}'.") from hermes_cli.models import opencode_provider_family if opencode_provider_family(resolved_provider) is not None: @@ -275,9 +262,7 @@ class CLIModelSwitchMixin: lambda m: opencode_model_api_mode(resolved_provider, m), lambda new: ( f"Stripped provider prefix from '{current_model}'; " - f"using '{new}' for {resolved_provider}." - ), - ) + f"using '{new}' for {resolved_provider}.")) if resolved_provider != "openai-codex": return changed @@ -288,8 +273,7 @@ class CLIModelSwitchMixin: if not self._model_is_default: self._console_print( f"[yellow]⚠️ Stripped provider prefix from '{current_model}'; " - f"using '{slug}' for OpenAI Codex.[/]" - ) + f"using '{slug}' for OpenAI Codex.[/]") self.model = slug current_model = slug changed = True @@ -327,8 +311,7 @@ class CLIModelSwitchMixin: result.target_provider, base_url=result.base_url, model=result.new_model, ) or None, "base_url": result.base_url or None, - "api_mode": result.api_mode or None, - } + "api_mode": result.api_mode or None} try: db.update_session_model(sid, result.new_model) db.patch_session_model_config(sid, {"gateway_runtime": route, **route}) @@ -355,8 +338,7 @@ class CLIModelSwitchMixin: # Stricter than the TUI gateway's recovery (which keeps bare "custom" when a # base_url exists) — the CLI's resolve path would hard-fail on it. stored_provider = _heal_bare_custom_provider( - _stored_runtime.get("provider") or None, base_url=stored_base_url, model=stored_model, - ) + _stored_runtime.get("provider") or None, base_url=stored_base_url, model=stored_model) model_changed = stored_model != self.model provider_changed = bool(stored_provider) and stored_provider != self.provider if not model_changed and not provider_changed: @@ -389,16 +371,14 @@ class CLIModelSwitchMixin: logger.debug( "Credential re-resolution for resumed session provider " "%s failed; keeping ambient credentials", - stored_provider, exc_info=True, - ) + stored_provider, exc_info=True) # Mid-chat /resume swaps the live agent; on startup --resume _init_agent picks up # self.model / self.provider. if self.agent is not None: try: self.agent.switch_model( new_model=self.model, new_provider=self.provider, api_key=self.api_key or "", - base_url=self.base_url or "", api_mode=self.api_mode or "", - ) + base_url=self.base_url or "", api_mode=self.api_mode or "") except Exception: logger.debug("In-place agent model swap on resume failed", exc_info=True) msg = f"Model restored from session: {stored_model}" @@ -420,8 +400,7 @@ class CLIModelSwitchMixin: "current_provider": current_provider, "user_provs": user_provs, "custom_provs": custom_provs, - "filter": "", - } + "filter": ""} self._invalidate(min_interval=0.0) def _confirm_expensive_model_switch(self, result) -> bool: @@ -433,32 +412,27 @@ class CLIModelSwitchMixin: warning = combined_selection_warning( result.new_model, provider=result.target_provider, base_url=result.base_url or self.base_url or "", - api_key=result.api_key or self.api_key or "", model_info=result.model_info, - ) + api_key=result.api_key or self.api_key or "", model_info=result.model_info) except Exception: warning = None if warning is None: return True choices = [ ("once", "Switch anyway", "Use this model for the current Hermes session."), - ("cancel", "Cancel", "Keep the current model."), - ] + ("cancel", "Cancel", "Keep the current model.")] raw = self._prompt_text_input_modal( - title=f"!!! {warning.title} !!!", detail=warning.message, choices=choices, timeout=120, - ) + title=f"!!! {warning.title} !!!", detail=warning.message, choices=choices, timeout=120) return self._normalize_slash_confirm_choice(raw, choices) == "once" def _confirm_and_apply_model_switch_result( - self, result, persist_global: bool, custom_providers=None - ) -> None: + self, result, persist_global: bool, custom_providers=None) -> None: from cli import _cprint try: if result.success and not self._confirm_expensive_model_switch(result): _cprint(" Model switch cancelled.") return self._apply_model_switch_result( - result, persist_global, custom_providers=custom_providers - ) + result, persist_global, custom_providers=custom_providers) except Exception as exc: _cprint(f" ✗ Model selection failed: {exc}") @@ -474,8 +448,7 @@ class CLIModelSwitchMixin: **_runtime_fields(self), "agent_primary_runtime": copy.deepcopy( getattr(agent, "_primary_runtime", None) - ) if agent is not None else None, - } + ) if agent is not None else None} def _restore_model_runtime_snapshot(self, snapshot: dict | None) -> None: """Restore a model runtime captured before a one-turn override.""" @@ -505,8 +478,7 @@ class CLIModelSwitchMixin: new_model=snapshot.get("model", ""), new_provider=snapshot.get("provider", ""), api_key=snapshot.get("api_key", ""), base_url=snapshot.get("base_url", ""), api_mode=snapshot.get("api_mode", ""), - capabilities=snapshot.get("capabilities"), - ) + capabilities=snapshot.get("capabilities")) except Exception as exc: logger.warning("CLI one-turn model restore failed: %s", exc) @@ -529,8 +501,7 @@ class CLIModelSwitchMixin: @staticmethod def _compute_model_picker_viewport( selected: int, scroll_offset: int, n: int, term_rows: int, reserved_below: int = 6, - panel_chrome: int = 6, min_visible: int = 3, - ) -> tuple[int, int]: + panel_chrome: int = 6, min_visible: int = 3) -> tuple[int, int]: """Resolve (scroll_offset, visible) for the /model picker viewport. ``reserved_below`` matches the approval/clarify panels (input, status bar, separators); ``panel_chrome`` is borders + blanks + hint row. The offset slides to keep ``selected`` on screen.""" @@ -557,8 +528,7 @@ class CLIModelSwitchMixin: if should_clear_context_pin( model_cfg.get("default") or model_cfg.get("model"), result.new_model, model_cfg.get("base_url"), result.base_url, - model_cfg.get("provider"), result.target_provider, - ): + model_cfg.get("provider"), result.target_provider): save_config_value("model.context_length", None) except Exception: save_config_value("model.context_length", None) @@ -591,21 +561,18 @@ class CLIModelSwitchMixin: self.agent.switch_model( new_model=result.new_model, new_provider=result.target_provider, api_key=result.api_key, base_url=result.base_url, api_mode=result.api_mode, - capabilities=getattr(result, "runtime_capabilities", None), - ) + capabilities=getattr(result, "runtime_capabilities", None)) except Exception as exc: for _k, _v in _cli_snapshot.items(): setattr(self, _k, _v) _cprint( f" ⚠ Model switch to {result.new_model} failed ({exc}); " - f"staying on {old_model}." - ) + f"staying on {old_model}.") return False return True def _apply_model_switch_result( - self, result, persist_global: bool, custom_providers=None - ) -> None: + self, result, persist_global: bool, custom_providers=None) -> None: """Picker-path commit (see _commit_model_switch).""" from cli import _cprint if not result.success: @@ -637,8 +604,7 @@ class CLIModelSwitchMixin: pass state.update( stage="model", provider_data=provider_data, model_list=model_list, - selected=0, filter="", _filtered_pairs=None, - ) + selected=0, filter="", _filtered_pairs=None) self._invalidate(min_interval=0.0) return if stage == "model": @@ -654,8 +620,7 @@ class CLIModelSwitchMixin: state.update( stage="provider", filter="", _filtered_pairs=None, selected=next((i for i, p in enumerate(state.get("providers") or []) - if p.get("slug") == provider_data.get("slug")), 0), - ) + if p.get("slug") == provider_data.get("slug")), 0)) self._invalidate(min_interval=0.0) return if selected > back_idx: # cancel row (and anything past it) @@ -666,15 +631,13 @@ class CLIModelSwitchMixin: self, visible_labels[selected], is_global=persist_global, explicit_provider=provider_data.get("slug"), user_providers=state.get("user_provs"), - custom_providers=state.get("custom_provs"), - ) + custom_providers=state.get("custom_provs")) # Capture before close — picker state is cleared on close. _picker_custom_provs = state.get("custom_provs") self._close_model_picker() _run_confirm_and_apply( self, self._confirm_and_apply_model_switch_result, - result, persist_global, _picker_custom_provs, - ) + result, persist_global, _picker_custom_provs) return self._close_model_picker() @@ -704,8 +667,7 @@ class CLIModelSwitchMixin: one_turn = request.is_once persist_global = resolve_persist_behavior( request.is_global, request.is_session, is_once=one_turn, - explicit_provider=request.explicit_provider, - ) + explicit_provider=request.explicit_provider) # --refresh: wipe the picker cache so every authed provider's /v1/models is re-fetched. if request.force_refresh: @@ -721,8 +683,7 @@ class CLIModelSwitchMixin: try: ctx = load_picker_context().with_overrides( current_provider=self.provider or "", current_model=self.model or "", - current_base_url=self.base_url or "", - ) + current_base_url=self.base_url or "") except Exception: ctx = None # switch_model() + _open_model_picker still need the raw provider dicts. @@ -735,20 +696,17 @@ class CLIModelSwitchMixin: result = _switch_model_from( self, request.target, is_global=persist_global, explicit_provider=request.explicit_provider, - user_providers=user_provs, custom_providers=custom_provs, - ) + user_providers=user_provs, custom_providers=custom_provs) if not result.success: _cprint(f" ✗ {result.error_message}") return _merge_preflight_warning(self, result, custom_provs) _run_confirm_and_apply( self, self._confirm_and_apply_cli_model_switch, - result, persist_global, one_turn, custom_provs, - ) + result, persist_global, one_turn, custom_provs) def _confirm_and_apply_cli_model_switch( - self, result, persist_global: bool, one_turn: bool, custom_provs=None - ) -> None: + self, result, persist_global: bool, one_turn: bool, custom_provs=None) -> None: """Confirm an expensive model switch and apply it (typed /model path). Runs on a worker thread when the TUI is active (see _run_confirm_and_apply) so the modal can render.""" from cli import _cprint @@ -782,8 +740,7 @@ class CLIModelSwitchMixin: return result = crs.apply( load_config(), new_value, - persist_callback=(save_config if new_value is not None else None), - ) + persist_callback=(save_config if new_value is not None else None)) prefix = "✓" if result.success else "✗" for line in result.message.splitlines(): _cprint(f" {prefix} {line}" if line.startswith("openai_runtime") else f" {line}") @@ -817,9 +774,7 @@ class CLIModelSwitchMixin: self._pending_moa_restore_model = { key: getattr(self, key, None) for key in ( - "requested_provider", "provider", "model", "api_key", "base_url", "api_mode", - ) - } + "requested_provider", "provider", "model", "api_key", "base_url", "api_mode")} self.requested_provider = "moa" self.provider = "moa" self.model = preset diff --git a/hermes_cli/cli_session_mixin.py b/hermes_cli/cli_session_mixin.py index b67137ce63..9613686c52 100644 --- a/hermes_cli/cli_session_mixin.py +++ b/hermes_cli/cli_session_mixin.py @@ -28,8 +28,7 @@ def _user_turn_indices(history: list) -> list[int]: return [ i for i, m in enumerate(history) - if not _is_ephemeral_scaffolding(m) and user_originated_turn_view(m) is not None - ] + if not _is_ephemeral_scaffolding(m) and user_originated_turn_view(m) is not None] def _timestamp_or(value, default): @@ -84,16 +83,14 @@ def _reset_model_to_config_default(cli, silent: bool) -> None: current_base_url=cli.base_url or "", current_api_key=cli.api_key or "", is_global=False, - explicit_provider=_config_provider or "", - ) + explicit_provider=_config_provider or "") if not r.success: return if cli.agent: cli.agent.switch_model( new_model=r.new_model, new_provider=r.target_provider, api_key=r.api_key, base_url=r.base_url, api_mode=r.api_mode, - capabilities=getattr(r, "runtime_capabilities", None), - ) + capabilities=getattr(r, "runtime_capabilities", None)) cli.model = r.new_model cli.provider = r.target_provider cli.requested_provider = r.target_provider @@ -175,8 +172,7 @@ class CLISessionMixin: try: from hermes_state import SessionDB from tools.approval import ( - _YOLO_MODE_FROZEN, enable_session_yolo, is_session_yolo_enabled, - ) + _YOLO_MODE_FROZEN, enable_session_yolo, is_session_yolo_enabled) except Exception: return if _YOLO_MODE_FROZEN or not SessionDB.session_yolo_enabled(session_meta): @@ -187,8 +183,7 @@ class CLISessionMixin: enable_session_yolo(session_key) _dim_notice(self, "⚡ YOLO mode restored from session — all commands auto-approved. /yolo to turn off.", - quiet, - ) + quiet) def _render_resume_history_panel_lines(self, panel) -> list[str]: """Render the resume panel at the current terminal width for resize replay.""" @@ -198,8 +193,7 @@ class CLISessionMixin: buf = StringIO() console = Console( file=buf, force_terminal=True, color_system="truecolor", highlight=False, - width=shutil.get_terminal_size((80, 24)).columns, - ) + width=shutil.get_terminal_size((80, 24)).columns) with _suspend_output_history(): console.print(panel) return buf.getvalue().rstrip("\n").splitlines() @@ -246,8 +240,7 @@ class CLISessionMixin: provider_info += f"{sep}[dim]auth: {self._provider_source}[/]" self._console_print( f" {api_indicator} [{accent_color}]{model_short}[/]{sep}" - f"[bold {label_color}]{tool_status}[/]{toolsets_info}{provider_info}" - ) + f"[bold {label_color}]{tool_status}[/]{toolsets_info}{provider_info}") def _show_session_status(self): """Show gateway-style status for the current CLI session.""" @@ -320,8 +313,7 @@ class CLISessionMixin: f"Created: {created_at.strftime('%Y-%m-%d %H:%M')}", f"Last Activity: {updated_at.strftime('%Y-%m-%d %H:%M')}", f"Tokens: {total_tokens:,}", - f"Agent Running: {'Yes' if is_running else 'No'}", - ]) + f"Agent Running: {'Yes' if is_running else 'No'}"]) self._console_print("\n".join(lines), highlight=False, markup=False) def _list_recent_sessions(self, limit: int = 10) -> list[dict[str, Any]]: @@ -334,8 +326,7 @@ class CLISessionMixin: return query_session_listing( self._session_db, source="cli", current_session_id=self.session_id, include_all_sources=False, include_unnamed=True, limit=limit, - exclude_sources=["kanban", "tool"], - ) + exclude_sources=["kanban", "tool"]) except Exception: return [] @@ -448,8 +439,7 @@ class CLISessionMixin: context = { "session_id": self.agent.session_id if self.agent else None, "platform": getattr(self, "platform", None) or "cli", - "reason": "new_session" if event_type == "on_session_reset" else "session_boundary", - } + "reason": "new_session" if event_type == "on_session_reset" else "session_boundary"} if event_type == "on_session_finalize": finalize_session(**context) else: @@ -469,15 +459,13 @@ class CLISessionMixin: try: from hermes_constants import get_hermes_home as _ghh return self._session_db.delete_session_if_empty( - session_id, sessions_dir=_ghh() / "sessions" - ) + session_id, sessions_dir=_ghh() / "sessions") except Exception: logger.debug("Could not prune empty session %s", session_id, exc_info=True) return False def _launch_session_boundary_memory_flush( - self, history_snapshot: list, *, session_id: Optional[str] = None, - ) -> Optional[list]: + self, history_snapshot: list, *, session_id: Optional[str] = None) -> Optional[list]: """Stage old-session memory extraction so /new stays responsive. The context-engine ``on_session_end`` is delivered synchronously here: cheap (no LLM) @@ -508,8 +496,7 @@ class CLISessionMixin: """Start a fresh session with a new session ID and cleared agent state.""" from cli import ( CLI_CONFIG, _parse_reasoning_config, _parse_service_tier_config, - _sync_process_session_id, datetime, - ) + _sync_process_session_id, datetime) old_session_id = self.session_id _boundary_snapshot = None if self.agent: @@ -517,8 +504,7 @@ class CLISessionMixin: # Context-engine boundary now; provider extraction is queued below (after # rotation) so /new never blocks on the LLM-bound call. _boundary_snapshot = self._launch_session_boundary_memory_flush( - list(self.conversation_history), session_id=old_session_id, - ) + list(self.conversation_history), session_id=old_session_id) self._notify_session_boundary("on_session_finalize") if self._session_db and old_session_id: @@ -527,8 +513,7 @@ class CLISessionMixin: if self.agent: with contextlib.suppress(Exception): self.agent._flush_messages_to_session_db( - self.conversation_history, conversation_history=self.conversation_history, - ) + self.conversation_history, conversation_history=self.conversation_history) with contextlib.suppress(Exception): self._session_db.end_session(old_session_id, "new_session") self._discard_session_if_empty(old_session_id) @@ -543,8 +528,7 @@ class CLISessionMixin: # An explicit -m/--model was for the previous session only. self._explicit_model_override = False self.reasoning_config = _parse_reasoning_config( - CLI_CONFIG["agent"].get("reasoning_effort", "") - ) + CLI_CONFIG["agent"].get("reasoning_effort", "")) # Session-scoped overrides (/model --session, /fast, one-turn restores) don't carry over. self._pending_one_turn_model_restore = None self.service_tier = _parse_service_tier_config(CLI_CONFIG["agent"].get("service_tier", "")) @@ -574,8 +558,7 @@ class CLISessionMixin: model=self.model, model_config={ "max_iterations": self.max_turns, "reasoning_config": self.reasoning_config, - }, - ) + }) self.agent._session_db_created = True if title: title = _apply_new_session_title(self, title) @@ -588,13 +571,11 @@ class CLISessionMixin: if _mm is not None and _boundary_snapshot: _mm.commit_session_boundary_async( _boundary_snapshot, new_session_id=self.session_id, - parent_session_id=old_session_id or "", reason="new_session", - ) + parent_session_id=old_session_id or "", reason="new_session") elif _mm is not None: _mm.on_session_switch( self.session_id, parent_session_id=old_session_id or "", - reset=True, reason="new_session", - ) + reset=True, reason="new_session") self._notify_session_boundary("on_session_reset") if not silent: @@ -638,8 +619,7 @@ class CLISessionMixin: """ from cli import datetime from hermes_cli.session_export import ( - SAVE_USAGE, normalize_save_format, render_session_for_save, - ) + SAVE_USAGE, normalize_save_format, render_session_for_save) parts = cmd.split()[1:] redact = bool(parts) and parts[-1].lower() in ("redact", "--redact") @@ -672,8 +652,7 @@ class CLISessionMixin: return session_data = { "id": self.session_id, "model": self.model, - "started_at": self.session_start.timestamp(), "messages": self.conversation_history, - } + "started_at": self.session_start.timestamp(), "messages": self.conversation_history} if redact: from hermes_cli.session_export_md import redact_session_data @@ -718,8 +697,7 @@ class CLISessionMixin: from agent.context_compressor import ( history_before_user_originated_turn, split_user_originated_turn, - user_originated_turn_view, - ) + user_originated_turn_view) from agent.memory_manager import sanitize_context from agent.tool_dispatch_helpers import _is_multimodal_tool_result, _multimodal_text_summary from run_agent import _is_ephemeral_scaffolding @@ -734,8 +712,7 @@ class CLISessionMixin: if isinstance(part, dict) and part.get("type") == "text": text_parts.append(str(part.get("text", ""))) elif isinstance(part, dict) and part.get("type") in { - "image", "image_url", "input_image", - }: + "image", "image_url", "input_image"}: text_parts.append("[screenshot]") return "\n".join(text_parts) if text_parts else None return content @@ -752,8 +729,7 @@ class CLISessionMixin: changed = RuntimeError("session history changed before the rewind could be persisted") expected_active_ids = self._session_db.get_active_message_ids(self.session_id) durable = self._session_db.get_messages_as_conversation( - self.session_id, include_row_ids=True - ) + self.session_id, include_row_ids=True) warm_persistence_history = [m for m in warm_history if not _is_ephemeral_scaffolding(m)] warm_user_indices = _user_indices(warm_persistence_history) durable_user_indices = _user_indices(durable) @@ -763,13 +739,11 @@ class CLISessionMixin: raise RuntimeError("persisted rewind target is no longer available") warm_prefix, _ = history_before_user_originated_turn( - warm_persistence_history, warm_user_indices[user_ordinal] - ) + warm_persistence_history, warm_user_indices[user_ordinal]) durable_target_index = durable_user_indices[user_ordinal] durable_target = durable[durable_target_index] durable_prefix, durable_live_view = history_before_user_originated_turn( - durable, durable_target_index - ) + durable, durable_target_index) if _comparison_content(durable_live_view) != _comparison_content(warm_live_view): raise changed target_row_id = durable_target.get("_row_id") @@ -780,8 +754,7 @@ class CLISessionMixin: self.session_id, target_row_id, preserve_compaction_handoff=scaffold is not None, expected_active_ids=expected_active_ids, - expected_target_content=durable_live_view.get("content"), - ) + expected_target_content=durable_live_view.get("content")) if scaffold is not None: replacement_id = result.get("replacement_message_id") if not isinstance(replacement_id, int) or not durable_prefix: @@ -817,8 +790,7 @@ class CLISessionMixin: return None from agent.context_compressor import ( - history_before_user_originated_turn, retryable_user_text, - ) + history_before_user_originated_turn, retryable_user_text) from agent.memory_manager import sanitize_context warm_history = list(self.conversation_history) @@ -833,8 +805,7 @@ class CLISessionMixin: # by /retry, so fail closed before archiving anything. try: truncated, live_view = history_before_user_originated_turn( - warm_history, user_indices[-1] - ) + warm_history, user_indices[-1]) live_content = live_view.get("content") if isinstance(live_content, str): live_content = sanitize_context(live_content).strip() @@ -850,8 +821,7 @@ class CLISessionMixin: truncated, _, _ = self._rewind_persisted_user_turn( warm_history=warm_history, user_ordinal=len(user_indices) - 1, - warm_live_view=live_view, - ) + warm_live_view=live_view) except Exception as exc: print(f"(x_x) Retry rewind failed; history was not changed: {exc}") return None @@ -896,8 +866,7 @@ class CLISessionMixin: truncated, durable_live_view, result = self._rewind_persisted_user_turn( warm_history=warm_history, user_ordinal=target_ordinal, - warm_live_view=live_view, - ) + warm_live_view=live_view) # Canonical editable prefill: the raw carrier holds the reference-summary wrapper. durable_text = self._undo_content_to_text(durable_live_view.get("content")) if durable_text: @@ -919,8 +888,7 @@ class CLISessionMixin: turn_word = "turn" if turns_undone == 1 else "turns" print( f"(^_^)b Undid {turns_undone} {turn_word} ({rewound_rows or removed_count} message(s)). " - f"Backed up to: \"{removed_text[:60]}{'...' if len(removed_text) > 60 else ''}\"" - ) + f"Backed up to: \"{removed_text[:60]}{'...' if len(removed_text) > 60 else ''}\"") print(f" {len(self.conversation_history)} message(s) remaining in history.") # Editable, not auto-sent (Claude-Code-style). if prefill and removed_text: @@ -955,8 +923,7 @@ class CLISessionMixin: return try: from tools.approval import ( - disable_session_yolo, enable_session_yolo, is_session_yolo_enabled, - ) + disable_session_yolo, enable_session_yolo, is_session_yolo_enabled) except Exception: return if is_session_yolo_enabled(old_session_id): @@ -993,8 +960,7 @@ class CLISessionMixin: from cli import _cprint from hermes_cli.colors import Colors as _Colors from tools.approval import ( - _YOLO_MODE_FROZEN, disable_session_yolo, enable_session_yolo, is_session_yolo_enabled, - ) + _YOLO_MODE_FROZEN, disable_session_yolo, enable_session_yolo, is_session_yolo_enabled) # A frozen process-level bypass short-circuits the approval gate ahead of the session # check — toggling "OFF" would be a false safety claim. Say so instead. @@ -1003,8 +969,7 @@ class CLISessionMixin: f" ⚡ YOLO is {_Colors.BOLD}{_Colors.RED}locked ON{_Colors.RESET}" " for this process (started with --yolo / HERMES_YOLO_MODE)." " /yolo cannot disable it — restart without the flag to" - " re-enable approvals." - ) + " re-enable approvals.") return session_key = self.session_id or "default" @@ -1016,16 +981,14 @@ class CLISessionMixin: _persist(session_key, False) _cprint( f" ⚠ YOLO mode {_Colors.BOLD}{_Colors.RED}OFF{_Colors.RESET}" - " — dangerous commands will require approval." - ) + " — dangerous commands will require approval.") else: enable_session_yolo(session_key) if _persist: _persist(session_key, True) _cprint( f" ⚡ YOLO mode {_Colors.BOLD}{_Colors.GREEN}ON{_Colors.RESET}" - " — all commands auto-approved. Use with caution." - ) + " — all commands auto-approved. Use with caution.") def _persist_session_yolo(self, session_key: str, enabled: bool) -> None: """Persist the YOLO flag to the session row so --resume restores it. Best-effort; the @@ -1056,8 +1019,7 @@ class CLISessionMixin: from hermes_cli.partial_compress import ( extract_compress_flags, parse_partial_compress_args, rejoin_compressed_head_and_tail, - split_history_for_partial_compress, summarize_compress_preview, - ) + split_history_for_partial_compress, summarize_compress_preview) from agent.conversation_compression import finalize_context_engine_compression_notification from agent.model_metadata import estimate_request_tokens_rough @@ -1080,13 +1042,11 @@ class CLISessionMixin: # understates real request pressure and can even appear to grow after compression. _estimate_kw = { "system_prompt": getattr(self.agent, "_cached_system_prompt", "") or "", - "tools": getattr(self.agent, "tools", None) or None, - } + "tools": getattr(self.agent, "tools", None) or None} if preview: approx_tokens = estimate_request_tokens_rough(self.conversation_history, **_estimate_kw) report = summarize_compress_preview( - self.conversation_history, partial, keep_last, focus_topic or None, approx_tokens, - ) + self.conversation_history, partial, keep_last, focus_topic or None, approx_tokens) for line in report["lines"]: print(f"🗜️ {line}") return @@ -1122,8 +1082,7 @@ class CLISessionMixin: # passing _cached_system_prompt duplicated the identity block. compressed, _ = self.agent._compress_context( head, None, approx_tokens=approx_tokens, focus_topic=focus_topic or None, - force=True, defer_context_engine_notification=True, - ) + force=True, defer_context_engine_notification=True) # Unchanged because a concurrent compression lock is held: say so instead of # the misleading "No changes" no-op text. Type-pinned check (is True / str) — @@ -1154,17 +1113,14 @@ class CLISessionMixin: self.agent._flush_messages_to_session_db(self.conversation_history, None) finalize_context_engine_compression_notification(self.agent, committed=True) new_tokens = estimate_request_tokens_rough( - self.conversation_history, **_estimate_kw - ) + self.conversation_history, **_estimate_kw) summary = summarize_manual_compression( original_history, self.conversation_history, approx_tokens, new_tokens, - compression_state=getattr(self.agent, "context_compressor", None), - ) + compression_state=getattr(self.agent, "context_compressor", None)) if ( summary.get("aborted") or summary.get("fallback_used") - or summary.get("refused_would_grow") - ): + or summary.get("refused_would_grow")): icon = "⚠️" else: icon = "🗜️" if summary["noop"] else "✅" @@ -1226,8 +1182,7 @@ class CLISessionMixin: if not isinstance(messages, list): return if isinstance(pending_cli_message, dict) and not any( - m is pending_cli_message for m in messages - ): + m is pending_cli_message for m in messages): # The UI accepted a new input but the worker still exposes its prior snapshot. messages = [*messages, pending_cli_message] if not messages: @@ -1241,8 +1196,7 @@ class CLISessionMixin: if ( isinstance(conversation_history, list) and conversation_history - and conversation_history[-1] is pending_cli_message - ): + and conversation_history[-1] is pending_cli_message): # Accepted but not yet durable: exclude it from the resumed-history baseline. conversation_history = conversation_history[:-1] elif not isinstance(conversation_history, list) or conversation_history is messages: @@ -1295,8 +1249,7 @@ class CLISessionMixin: user_msgs = len([m for m in self.conversation_history if m.get("role") == "user"]) tool_calls = len([ - m for m in self.conversation_history if m.get("role") == "tool" or m.get("tool_calls") - ]) + m for m in self.conversation_history if m.get("role") == "tool" or m.get("tool_calls")]) elapsed = datetime.now() - self.session_start hours, remainder = divmod(int(elapsed.total_seconds()), 3600) minutes, seconds = divmod(remainder, 60) diff --git a/hermes_cli/cli_status_bar_mixin.py b/hermes_cli/cli_status_bar_mixin.py index a57543e293..00d0a72198 100644 --- a/hermes_cli/cli_status_bar_mixin.py +++ b/hermes_cli/cli_status_bar_mixin.py @@ -22,8 +22,7 @@ _STRONG = "class:status-bar-strong" _AGENT_COUNTERS = ( "session_input_tokens", "session_output_tokens", "session_cache_read_tokens", "session_cache_write_tokens", "session_prompt_tokens", "session_completion_tokens", - "session_total_tokens", "session_api_calls", -) + "session_total_tokens", "session_api_calls") def _threshold_style(value, ladder, fallback: str) -> str: @@ -135,8 +134,7 @@ class CLIStatusBarMixin: @staticmethod def _format_prompt_elapsed( - prompt_start_time: Optional[float], prompt_duration: float, live: bool = False - ) -> str: + prompt_start_time: Optional[float], prompt_duration: float, live: bool = False) -> str: """Per-prompt elapsed time. Always a string (``⏲ 0s`` on fresh start); seconds stay visible at every scale so it increments smoothly (``1m 59s → 2m → 2m 1s``). ⏱ while live, ⏲ frozen — width-1 glyphs (no variation selector) keep the bar aligned.""" @@ -196,11 +194,9 @@ class CLIStatusBarMixin: "duration": format_duration_compact(elapsed_seconds), "session_title": self._get_status_bar_session_title(), "prompt_elapsed": self._format_prompt_elapsed( - prompt_start, getattr(self, "_prompt_duration", 0.0), live=turn_live, - ), + prompt_start, getattr(self, "_prompt_duration", 0.0), live=turn_live), "idle_since": self._format_idle_since( - getattr(self, "_last_turn_finished_at", None), turn_live=turn_live, - ), + getattr(self, "_last_turn_finished_at", None), turn_live=turn_live), "context_tokens": 0, "context_length": None, "context_percent": None, @@ -214,15 +210,13 @@ class CLIStatusBarMixin: "focus_label": "", # /focus badge: the reduced-output mode is never invisible. "goal_active": False, "goal_turns_used": 0, - "goal_max_turns": 0, - } + "goal_max_turns": 0} try: from hermes_cli.focus_view import focus_statusbar_segment snapshot["focus_label"] = focus_statusbar_segment( - bool(getattr(self, "_focus_view_enabled", False)) - ) + bool(getattr(self, "_focus_view_enabled", False))) except Exception: pass @@ -291,8 +285,7 @@ class CLIStatusBarMixin: _anchored = anchored_context_tokens( _msgs if isinstance(_msgs, list) else [], getattr(agent, "_turn_base_usage_anchor", None), - charge_stale_thinking=False, - ) + charge_stale_thinking=False) if _anchored is not None and _anchored > 0: context_tokens = _anchored except Exception: @@ -542,8 +535,7 @@ class CLIStatusBarMixin: from agent.turn_summary import format_token_flow produced = (getattr(agent, "session_output_tokens", 0) or 0) - ( - getattr(self, "_turn_token_baseline", 0) or 0 - ) + getattr(self, "_turn_token_baseline", 0) or 0) return format_token_flow(produced) except Exception: return "" @@ -651,11 +643,9 @@ class CLIStatusBarMixin: or self._pet_slug != pet.slug or self._pet_cols != cols or self._pet_scale != scale - or self._pet_renderer.mode != renderer_mode - ): + or self._pet_renderer.mode != renderer_mode): self._pet_renderer = pet_render.PetRenderer( - str(pet.spritesheet), mode=renderer_mode, scale=scale, unicode_cols=cols - ) + str(pet.spritesheet), mode=renderer_mode, scale=scale, unicode_cols=cols) self._pet_slug = pet.slug self._pet_cols = cols self._pet_scale = scale @@ -711,8 +701,7 @@ class CLIStatusBarMixin: or self._clarify_state or self._sudo_state or self._secret_state - or getattr(self, "_slash_confirm_state", None) - ) + or getattr(self, "_slash_confirm_state", None)) return derive_pet_state( awaiting_input=awaiting_input, busy=getattr(self, "_agent_running", False), @@ -968,8 +957,7 @@ class CLIStatusBarMixin: return result def _status_bar_segments( - self, snapshot, width: int, field_set, yolo_active: bool, *, styled: bool - ) -> list: + self, snapshot, width: int, field_set, yolo_active: bool, *, styled: bool) -> list: """Ordered status-bar segments for one width tier (<52 / <76 / wide), each a list of ``(style, text)`` fragments. Shared by the plain-text and prompt_toolkit renderers so the two can never drift; ``styled`` selects the graphical context bar.""" @@ -1026,8 +1014,7 @@ class CLIStatusBarMixin: add("cache_hit", self._cache_hit_rate_style(cache[0]), cache[1]) if wide: for name, key, glyph in ( - ("latency", "avg_latency_label", "◷"), ("tps", "avg_velocity_label", "↑"), - ): + ("latency", "avg_latency_label", "◷"), ("tps", "avg_velocity_label", "↑")): label = snapshot.get(key) or "" if label: add(name, _DIM, f"{glyph} {label}") @@ -1067,8 +1054,7 @@ class CLIStatusBarMixin: show_title = field_set is None or "title" in field_set session_title = (snapshot.get("session_title") or "") if show_title else "" segs = self._status_bar_segments( - snapshot, width, field_set, self._is_session_yolo_active(), styled=False - ) + snapshot, width, field_set, self._is_session_yolo_active(), styled=False) parts = ["".join(t for _, t in seg) for seg in segs] or [f"⚕ {model_short}"] # Narrow bars always join the battery with │; wider tiers use the tier separator. if battery_label: @@ -1085,8 +1071,7 @@ class CLIStatusBarMixin: if ( not self._status_bar_visible or getattr(self, "_model_picker_state", None) - or getattr(self, "_command_palette_state", None) - ): + or getattr(self, "_command_palette_state", None)): return [] try: snapshot = self._get_status_bar_snapshot() @@ -1100,8 +1085,7 @@ class CLIStatusBarMixin: session_title = (snapshot.get("session_title") or "") if _ok("title") else "" segs = self._status_bar_segments( - snapshot, width, field_set, self._is_session_yolo_active(), styled=True - ) + snapshot, width, field_set, self._is_session_yolo_active(), styled=True) sep = " · " if width < 76 else " │ " frags: list = [] for seg in segs or [[(_SB, " ⚕ "), (_STRONG, snapshot["model_short"])]]: diff --git a/hermes_cli/cli_stream_mixin.py b/hermes_cli/cli_stream_mixin.py index 13071b2681..60c719522b 100644 --- a/hermes_cli/cli_stream_mixin.py +++ b/hermes_cli/cli_stream_mixin.py @@ -20,8 +20,7 @@ from rich.markup import escape as _escape # Model-generated reasoning tags: suppressed during streaming (they'd display as raw XML; # the agent strips them from final_response too) unless show_reasoning routes them to the box. _OPEN_TAGS = ( - "", "", "", "", "", "", -) + "", "", "", "", "", "") _CLOSE_TAGS = tuple(" int: @@ -145,18 +142,15 @@ class CLIStreamMixin: min_newline_flush = max(16, target_width // 3) if line_break != -1 and ( line_break >= min_newline_flush - or buf.endswith(("\n\n", ".\n", "!\n", "?\n", ":\n")) - ): + or buf.endswith(("\n\n", ".\n", "!\n", "?\n", ":\n"))): flush_text, buf = buf[: line_break + 1], buf[line_break + 1 :] elif len(buf) >= target_width: search_start = max(20, target_width // 2) search_end = min( - len(buf), max(target_width + (target_width // 3), target_width + 8) - ) + len(buf), max(target_width + (target_width // 3), target_width + 8)) cut = max( buf.rfind(b, search_start, search_end) - for b in (" ", "\t", ".", "!", "?", ",", ";", ":") - ) + for b in (" ", "\t", ".", "!", "?", ",", ";", ":")) if cut != -1: flush_text, buf = buf[: cut + 1], buf[cut + 1 :] @@ -169,8 +163,7 @@ class CLIStreamMixin: from cli import _accent_hex, datetime ts_suffix = ( f" [dim]{datetime.now().strftime(getattr(self, 'timestamp_format', '%H:%M'))}[/]" - if getattr(self, "show_timestamps", False) else "" - ) + if getattr(self, "show_timestamps", False) else "") lines = user_input.split("\n") if len(lines) <= 1: return f"[bold {_accent_hex()}]●[/] [bold]{_escape(user_input)}[/]{ts_suffix}" @@ -305,8 +298,7 @@ class CLIStreamMixin: # Boundary: only whitespace since the last newline — or, with no newline # buffered yet, since the last emit (which must have ended a line). is_block_boundary = preceding[preceding.rfind("\n") + 1:].strip() == "" and ( - "\n" in preceding or getattr(self, "_stream_last_was_newline", True) - ) + "\n" in preceding or getattr(self, "_stream_last_was_newline", True)) if is_block_boundary: if preceding: self._emit_stream_text(preceding) @@ -363,15 +355,13 @@ class CLIStreamMixin: from cli import _RST, _STREAM_PAD, _cprint _tc = getattr(self, "_stream_text_ansi", "") _cprint( - f"{_STREAM_PAD}{_tc}{printed_line}{_RST}" if _tc else f"{_STREAM_PAD}{printed_line}" - ) + f"{_STREAM_PAD}{_tc}{printed_line}{_RST}" if _tc else f"{_STREAM_PAD}{printed_line}") def _flush_stream_table_buf(self) -> None: """Emit the held table block re-aligned as a whole. Cell-level markdown is stripped FIRST so the realigner pads to the final visible width, not the marker-decorated width.""" from cli import ( - _strip_markdown_syntax, _terminal_width_for_streaming, realign_markdown_tables - ) + _strip_markdown_syntax, _terminal_width_for_streaming, realign_markdown_tables) buf = self._stream_table_buf self._stream_table_buf = [] self._in_stream_table = False @@ -388,8 +378,7 @@ class CLIStreamMixin: """Emit filtered text to the streaming display.""" from cli import ( HermesCLI, _ACCENT, _RST, _STREAM_PARTIAL_PREVIEW_LEN, _cprint, _strip_markdown_syntax, - datetime, is_table_divider, looks_like_table_row, - ) + datetime, is_table_divider, looks_like_table_row) if not text: return # Defer content while the reasoning box renders so reasoning always lands BEFORE it. @@ -448,8 +437,7 @@ class CLIStreamMixin: self._stream_buf and not self._in_stream_table and not self._stream_buf.lstrip().startswith("|") - and len(self._stream_buf) >= 80 - ): + and len(self._stream_buf) >= 80): preview = self._stream_buf[-int(_STREAM_PARTIAL_PREVIEW_LEN):] cut = preview.find(" ") if 0 < cut < len(preview) - 1: @@ -463,8 +451,7 @@ class CLIStreamMixin: def _flush_stream(self) -> None: """Emit any remaining partial line from the stream buffer and close the box.""" from cli import ( - _ACCENT, _RST, _cprint, _strip_markdown_syntax, is_table_divider, looks_like_table_row - ) + _ACCENT, _RST, _cprint, _strip_markdown_syntax, is_table_divider, looks_like_table_row) # Still inside a "reasoning block" at end-of-stream = false positive (the model # mentioned a tag in prose and never closed it): recover the buffer as regular text. if getattr(self, "_in_reasoning_block", False) and getattr(self, "_stream_prefilt", ""): @@ -477,8 +464,7 @@ class CLIStreamMixin: if ( self._stream_buf and getattr(self, "_in_stream_table", False) - and (looks_like_table_row(self._stream_buf) or is_table_divider(self._stream_buf)) - ): + and (looks_like_table_row(self._stream_buf) or is_table_divider(self._stream_buf))): self._stream_table_buf.append(self._stream_buf) self._stream_buf = "" if getattr(self, "_stream_table_buf", None): @@ -554,8 +540,7 @@ class CLIStreamMixin: analysis_prompt = ( "Describe everything visible in this image in thorough detail. " "Include any text, code, data, objects, people, layout, colors, " - "and any other notable visual information." - ) + "and any other notable visual information.") enriched_parts = [] for img_path in images: if not img_path.exists(): @@ -565,32 +550,28 @@ class CLIStreamMixin: _cprint(f" {_DIM}👁️ analyzing {img_path.name} ({size_kb}KB)...{_RST}") try: result_json = _asyncio.run( - vision_analyze_tool(image_url=str(img_path), user_prompt=analysis_prompt) - ) + vision_analyze_tool(image_url=str(img_path), user_prompt=analysis_prompt)) result = json.loads(result_json) if result.get("success"): description = result.get("analysis", "") enriched_parts.append( f"[The user attached an image. Here's what it contains:\n{description}]\n" f"[If you need a closer look, use vision_analyze with " - f"image_url: {img_path}]" - ) + f"image_url: {img_path}]") if announce: _cprint(f" {_DIM}✓ image analyzed{_RST}") else: enriched_parts.append( f"[The user attached an image but it couldn't be analyzed. " f"You can try examining it with vision_analyze using " - f"image_url: {img_path}]" - ) + f"image_url: {img_path}]") if announce: _cprint(f" {_DIM}⚠ vision analysis failed — path included for retry{_RST}") except Exception as e: enriched_parts.append( f"[The user attached an image but analysis failed ({e}). " f"You can try examining it with vision_analyze using " - f"image_url: {img_path}]" - ) + f"image_url: {img_path}]") if announce: _cprint(f" {_DIM}⚠ vision analysis error — path included for retry{_RST}") @@ -663,8 +644,7 @@ class CLIStreamMixin: if event_type == "tool.completed": self._tool_start_time = 0.0 self._turn_summary_record( - function_name, kwargs.get("result"), kwargs.get("is_error", False) - ) + function_name, kwargs.get("result"), kwargs.get("is_error", False)) # Focus view: count the hidden scrollback line for the post-turn recovery report. if getattr(self, "_focus_view_enabled", False): try: @@ -697,11 +677,9 @@ class CLIStreamMixin: if ( not getattr(self, "_long_tool_hint_fired", False) and self.tool_progress_mode == "all" - and duration >= 30.0 - ): + and duration >= 30.0): from agent.onboarding import ( - TOOL_PROGRESS_FLAG, is_seen, mark_seen, tool_progress_hint_cli - ) + TOOL_PROGRESS_FLAG, is_seen, mark_seen, tool_progress_hint_cli) if not is_seen(CLI_CONFIG, TOOL_PROGRESS_FLAG): self._long_tool_hint_fired = True _cprint(f" {_DIM}{tool_progress_hint_cli()}{_RST}") @@ -723,8 +701,7 @@ class CLIStreamMixin: self._tool_start_time = time.monotonic() # Store args for stacked scrollback line on completion self._pending_tool_info.setdefault(function_name, []).append( - function_args if function_args is not None else {} - ) + function_args if function_args is not None else {}) self._invalidate() def _on_tool_start(self, tool_call_id: str, function_name: str, function_args: dict): @@ -760,7 +737,6 @@ class CLIStreamMixin: from agent.display import render_edit_diff_with_delta render_edit_diff_with_delta( function_name, function_result, function_args=function_args, snapshot=snapshot, - print_fn=_cprint, - ) + print_fn=_cprint) except Exception: logger.debug("Edit diff preview failed for %s", function_name, exc_info=True) diff --git a/hermes_cli/cli_tui_mixin.py b/hermes_cli/cli_tui_mixin.py index a75739931a..c5aa0021fb 100644 --- a/hermes_cli/cli_tui_mixin.py +++ b/hermes_cli/cli_tui_mixin.py @@ -393,8 +393,7 @@ class CLITuiMixin: input_area, input_rule_bot, voice_status_bar, - completions_menu, - ] + completions_menu] return [item for item in ordered if item is not None] def _tui_spinner_loop(self): @@ -630,8 +629,7 @@ class CLITuiMixin: choices.append("Cancel") hint = ( f"Current: {state.get('current_model', 'unknown')} " - f"on {state.get('current_provider', 'unknown')}" - ) + f"on {state.get('current_provider', 'unknown')}") else: provider_data = state.get("provider_data") or {} model_list = state.get("model_list") or [] @@ -652,8 +650,7 @@ class CLITuiMixin: else: hint = "No models listed for this provider. Use Back or Cancel." return self._render_scroll_list_panel( - state, title, hint, choices, min_width=46, max_width=84, indent=' ', - ) + state, title, hint, choices, min_width=46, max_width=84, indent=' ') def _get_command_palette_display_fragments(self): state = self._command_palette_state @@ -774,8 +771,7 @@ class CLITuiMixin: try: _stash = self._prompt_stash return self._render_stash_panel( - _stash.panel_rows(), _stash.panel_cursor, self._get_tui_terminal_width(), - ) + _stash.panel_rows(), _stash.panel_cursor, self._get_tui_terminal_width()) except Exception: return [] @@ -1314,8 +1310,7 @@ class CLITuiMixin: from cli import ( _apply_backslash_line_continuation, _is_backslash_line_continuation, - _looks_like_slash_command, - ) + _looks_like_slash_command) if self._tui_enter_overlay(event): return buf = event.app.current_buffer @@ -1431,8 +1426,7 @@ class CLITuiMixin: with open(_hermes_home / "interrupt_debug.log", "a", encoding="utf-8") as _f: _f.write( f"{time.strftime('%H:%M:%S')} ENTER: queued interrupt msg={str(payload)[:60]!r}, " - f"agent_running={self._agent_running}\n" - ) + f"agent_running={self._agent_running}\n") except Exception: pass # First-touch onboarding: one-line tip about the /busy knob on the first busy-while- @@ -1816,8 +1810,7 @@ class CLITuiMixin: CLI_CONFIG, _bind_prompt_submit_keys, _cli_multiline_shortcuts_enabled, - _preserve_ctrl_enter_newline, - ) + _preserve_ctrl_enter_newline) from prompt_toolkit.keys import Keys kb = KeyBindings() _multiline_shortcuts_enabled = _cli_multiline_shortcuts_enabled(self.config or CLI_CONFIG) @@ -1825,8 +1818,7 @@ class CLITuiMixin: kb.add(Keys.Ignore, eager=True)(self._tui_handle_ignored_terminal_sequence) _bind_prompt_submit_keys( - kb, self._tui_handle_enter, multiline_shortcuts_enabled=_multiline_shortcuts_enabled, - ) + kb, self._tui_handle_enter, multiline_shortcuts_enabled=_multiline_shortcuts_enabled) kb.add('escape', 'enter')(self._tui_insert_newline) # Ctrl+J inserts a newline (Claude Code / Codex / OpenCode). Windows Terminal delivers # Ctrl+Enter as the same c-j code. display.cli_multiline_shortcuts: false restores legacy @@ -1852,8 +1844,7 @@ class CLITuiMixin: kb.add('c-q')(self._tui_handle_ctrl_q) kb.add('c-d')(self._tui_handle_ctrl_d) _modal_prompt_active = Condition( - lambda: bool(self._secret_state or self._sudo_state or self._slash_confirm_state) - ) + lambda: bool(self._secret_state or self._sudo_state or self._slash_confirm_state)) kb.add('escape', filter=_modal_prompt_active, eager=True)(self._tui_handle_escape_modal) kb.add('escape', 'escape', filter=~_modal_prompt_active)(self._tui_handle_double_escape) kb.add('c-z')(self._tui_handle_ctrl_z) @@ -1869,11 +1860,9 @@ class CLITuiMixin: # unbound there and arrives as ('escape', 'g') — register it as a fallback. _editor_filter = Condition( lambda: not self._clarify_state and not self._approval_state - and not self._sudo_state and not self._secret_state - ) + and not self._sudo_state and not self._secret_state) kb.add('c-g', filter=_editor_filter)( - kb.add('escape', 'g', filter=_editor_filter)(self._tui_handle_open_in_editor) - ) + kb.add('escape', 'g', filter=_editor_filter)(self._tui_handle_open_in_editor)) # Ctrl+S prompt stash: park a draft, send something else, bring it back. Suppressed while # a modal prompt owns the composer so Ctrl+S can't stash a password. _stash_filter = Condition( @@ -1896,8 +1885,7 @@ class CLITuiMixin: _clarify_nav = Condition(lambda: bool(self._clarify_state) and not self._clarify_freetext) _clarify_batch = Condition( lambda: bool(self._clarify_state) and bool(self._clarify_state.get("questions")) - and not self._clarify_freetext - ) + and not self._clarify_freetext) kb.add('up', filter=_clarify_nav)(self._tui_clarify_up) kb.add('down', filter=_clarify_nav)(self._tui_clarify_down) # Multi-select: Space toggles the checkbox under the cursor. @@ -2005,16 +1993,14 @@ class CLITuiMixin: spinner_widget = Window( content=FormattedTextControl(self._tui_spinner_text), height=self._tui_spinner_height, - wrap_lines=True, - ) + wrap_lines=True) # Petdex mascot — right-aligned Kitty placeholder or half-block sprite above the prompt; # height 0 when no pet is enabled. The animation thread queues virtual Kitty frames; # after_render writes them out-of-band while prompt_toolkit owns the placeholder grid. self._pet_widget = Window( content=FormattedTextControl(self._pet_fragments), height=self._pet_widget_height, - align=WindowAlign.RIGHT, - ) + align=WindowAlign.RIGHT) # Hint line above the input: only for interactive prompts that need extra instructions # (sudo countdown, approval navigation, clarify); the agent-running hint is the placeholder. spacer = Window(content=FormattedTextControl(self._tui_hint_text), height=self._tui_hint_height) @@ -2023,11 +2009,9 @@ class CLITuiMixin: secret_widget = self._tui_overlay_widget(self._get_secret_display_fragments, "_secret_state") approval_widget = self._tui_overlay_widget(self._get_approval_display_fragments, "_approval_state") slash_confirm_widget = self._tui_overlay_widget( - self._get_slash_confirm_display_fragments, "_slash_confirm_state", - ) + self._get_slash_confirm_display_fragments, "_slash_confirm_state") model_picker_widget = self._tui_overlay_widget( - self._get_model_picker_display_fragments, "_model_picker_state", - ) + self._get_model_picker_display_fragments, "_model_picker_state") command_palette_widget = self._tui_overlay_widget( self._get_command_palette_display_fragments, "_command_palette_state") # Rules above/below the input; narrow terminals hide the bottom one to recover a row. diff --git a/hermes_cli/cli_voice_mixin.py b/hermes_cli/cli_voice_mixin.py index 16e28bf144..c6ec8aaff8 100644 --- a/hermes_cli/cli_voice_mixin.py +++ b/hermes_cli/cli_voice_mixin.py @@ -82,16 +82,14 @@ class CLIVoiceMixin: ) raise RuntimeError( "Voice mode requires sounddevice and numpy.\n" - f"Install with: {sys.executable} -m pip install sounddevice numpy" - ) + f"Install with: {sys.executable} -m pip install sounddevice numpy") if not reqs.get("stt_available", reqs.get("stt_key_set")): raise RuntimeError( "Voice mode requires an STT provider for transcription.\n" "Option 1: uv pip install faster-whisper " "(free, local; `pip install faster-whisper` also works if pip is on PATH)\n" "Option 2: Set GROQ_API_KEY (free tier)\n" - "Option 3: Set VOICE_TOOLS_OPENAI_KEY (paid)" - ) + "Option 3: Set VOICE_TOOLS_OPENAI_KEY (paid)") # Prevent double-start from concurrent threads (atomic check-and-set) with self._voice_lock: @@ -223,8 +221,7 @@ class CLIVoiceMixin: if self._voice_stt_provider() == "local": _cprint( f"{_DIM}Preparing local STT model '{stt_model}' " - f"(first use may download it from Hugging Face)...{_RST}" - ) + f"(first use may download it from Hugging Face)...{_RST}") else: _cprint(f"{_DIM}Transcribing...{_RST}") from tools.voice_mode import is_voice_stop_phrase, transcribe_recording @@ -270,8 +267,7 @@ class CLIVoiceMixin: _tts_done = getattr(self, "_voice_tts_done", None) _activity_hold = bool( getattr(self, "_agent_running", False) - or (_tts_done is not None and not _tts_done.is_set()) - ) + or (_tts_done is not None and not _tts_done.is_set())) if submitted: self._no_speech_count = 0 elif not _activity_hold: @@ -287,8 +283,7 @@ class CLIVoiceMixin: self._voice_continuous and not submitted and not self._voice_recording - and not stop_continuous_restart - ): + and not stop_continuous_restart): self._voice_restart_recording_async() def _voice_speak_response_async(self, text: str) -> None: @@ -419,8 +414,7 @@ class CLIVoiceMixin: # Generation phase: no audio to cut — interrupt the in-flight agent turn. logger.debug( "full-duplex listener tripped during generation — " - "interrupting agent turn" - ) + "interrupting agent turn") if _pipe_stop is not None: _pipe_stop.set() # never let the stale reply speak try: @@ -432,8 +426,7 @@ class CLIVoiceMixin: wav_path = full_duplex_listen( _should_stop, is_playing=is_audio_output_active, on_trigger=_on_trigger, - multiplier=_mult or None, grace_ms=max(0, _grace_ms), - ) + multiplier=_mult or None, grace_ms=max(0, _grace_ms)) if wav_path and self._voice_barge_capture.is_set(): self._voice_submit_barge_utterance(wav_path) else: @@ -464,8 +457,7 @@ class CLIVoiceMixin: from tools.voice_mode import is_tts_echo if is_tts_echo(transcript, getattr(self, "_voice_last_tts_text", "")): logger.debug( - "Dropping playback-phase barge transcript as TTS echo: %r", transcript - ) + "Dropping playback-phase barge transcript as TTS echo: %r", transcript) _cprint(f"\n{_DIM}Ignored likely TTS echo (not queued).{_RST}") return self._pending_input.put(_VoiceInputMessage(transcript)) @@ -617,8 +609,7 @@ class CLIVoiceMixin: say = _cprint if announce else (lambda *_a: None) try: from tools.wake_word import ( - check_wake_word_requirements, load_wake_word_config, owns_listener, start_listening - ) + check_wake_word_requirements, load_wake_word_config, owns_listener, start_listening) except Exception as e: say(f"{_DIM}Wake word unavailable: {e}{_RST}") return False @@ -749,8 +740,7 @@ class CLIVoiceMixin: self._agent_running or self._voice_recording or getattr(self, "_voice_processing", False) - or not self._pending_input.empty() - ) + or not self._pending_input.empty()) if busy: idle_polls = 0 continue @@ -777,8 +767,7 @@ class CLIVoiceMixin: from cli import _ACCENT, _BOLD, _DIM, _RST, _cprint from tools.wake_word import ( audio_is_silent, check_wake_word_requirements, is_listening, load_wake_word_config, - owns_listener, - ) + owns_listener) cfg = load_wake_word_config() reqs = check_wake_word_requirements(cfg) diff --git a/hermes_cli/clipboard.py b/hermes_cli/clipboard.py index 50b6ed9843..93cd6b233e 100644 --- a/hermes_cli/clipboard.py +++ b/hermes_cli/clipboard.py @@ -50,8 +50,7 @@ def _linux_backends(): return ( (_is_wsl(), _wsl_has_image, _wsl_save), (bool(os.environ.get("WAYLAND_DISPLAY")), _wayland_has_image, _wayland_save), - (True, _xclip_has_image, _xclip_save), - ) + (True, _xclip_has_image, _xclip_save)) def save_clipboard_image(dest: Path) -> bool: @@ -206,9 +205,7 @@ _PS_IMAGE_STRATEGIES = ( _PS_FILEDROP_HIT + "if ($null -eq $hit) { exit 1 }" "[System.Convert]::ToBase64String([System.IO.File]::ReadAllBytes($hit))" - "} catch { exit 1 }", - ), -) + "} catch { exit 1 }")) def _ps_clipboard(exe: str, timeout: int, label: str, dest: Path | None = None) -> bool: diff --git a/hermes_cli/codex_models.py b/hermes_cli/codex_models.py index 7191806ef2..f3f477115c 100644 --- a/hermes_cli/codex_models.py +++ b/hermes_cli/codex_models.py @@ -28,8 +28,7 @@ DEFAULT_CODEX_MODELS: List[str] = [ # not in the public API, so it stays out of the "openai" catalog in hermes_cli/models.py. # The backend reports ``supported_in_api: false`` for it; that flag describes API # availability, not Codex availability, so fetch/cache paths must not filter on it. - "gpt-5.3-codex-spark", -] + "gpt-5.3-codex-spark"] _FORWARD_COMPAT_TEMPLATE_MODELS: List[tuple[str, tuple[str, ...]]] = [ ("gpt-5.6-sol", ("gpt-5.5", "gpt-5.4")), @@ -40,8 +39,7 @@ _FORWARD_COMPAT_TEMPLATE_MODELS: List[tuple[str, tuple[str, ...]]] = [ ("gpt-5.4", ("gpt-5.3-codex",)), # Spark surfaces whenever a compatible template is present; the backend (not Hermes) # gates real availability by ChatGPT Pro entitlement. - ("gpt-5.3-codex-spark", ("gpt-5.3-codex",)), -] + ("gpt-5.3-codex-spark", ("gpt-5.3-codex",))] def _dedupe(model_ids) -> List[str]: @@ -103,8 +101,7 @@ def _extract_chatgpt_account_id(access_token: str) -> Optional[str]: acct_id = ( claims.get("https://api.openai.com/auth", {}).get("chatgpt_account_id") if isinstance(claims, dict) - else None - ) + else None) return acct_id if isinstance(acct_id, str) and acct_id else None except Exception: return None @@ -145,8 +142,7 @@ def _fetch_models_from_api(access_token: str) -> List[str]: resp = httpx.get( "https://chatgpt.com/backend-api/codex/models?client_version=1.0.0", headers=headers, - timeout=10, - ) + timeout=10) if resp.status_code != 200: return [] data = resp.json() @@ -194,5 +190,4 @@ def get_codex_model_ids(access_token: Optional[str] = None) -> List[str]: default_model = _read_default_model(codex_home) return _finalize_codex_models(_dedupe([ *([default_model] if default_model else []), *_read_cache_models(codex_home), - *DEFAULT_CODEX_MODELS, - ])) + *DEFAULT_CODEX_MODELS])) diff --git a/hermes_cli/codex_runtime_plugin_migration.py b/hermes_cli/codex_runtime_plugin_migration.py index 9002d89347..e712f18e91 100644 --- a/hermes_cli/codex_runtime_plugin_migration.py +++ b/hermes_cli/codex_runtime_plugin_migration.py @@ -70,8 +70,7 @@ _KEYS_DROPPED_WITH_WARNING = {"sampling"} # (hermes key, codex key, skip note) — timeouts are emitted as floats or skipped when non-numeric. _TIMEOUT_KEYS = ( ("timeout", "tool_timeout_sec", "timeout (not numeric)"), - ("connect_timeout", "startup_timeout_sec", "connect_timeout (not numeric)"), -) + ("connect_timeout", "startup_timeout_sec", "connect_timeout (not numeric)")) def _str_map(d: dict) -> dict[str, str]: @@ -130,8 +129,7 @@ def _translate_one_server(name: str, hermes_cfg: dict) -> tuple[Optional[dict], # env-var passthrough (HERMES_HOME, PYTHONPATH) could carry one in pathological cases. _TOML_ESCAPES = ( ("\\", "\\\\"), ('"', '\\"'), ("\b", "\\b"), ("\t", "\\t"), - ("\n", "\\n"), ("\f", "\\f"), ("\r", "\\r"), -) + ("\n", "\\n"), ("\f", "\\f"), ("\r", "\\r")) def _escape_toml_string(value: str) -> str: