refactor(cli): simplify CLI mixins (billing, agent-setup, commands) — dedupe helpers, dispatch tables, god-method splits
hermes_cli/cli_billing_mixin.py (1566 -> 1173): shared _modal_choice / _step_up_remote_spending / _print_org_line / _print_portal_line / _print_total_spendable / _try_usage_model / _billing_patch_auto_top_up; copy tables for charge-failed / charge-error / upgrade-status. hermes_cli/cli_agent_setup_mixin.py (962 -> 787): _current_runtime + _route_signature (3 sites), _compression_descendant (2 sites), resume event/skin tables, _load_resumed_history_late lifted out of _init_agent; fixes latent undefined 'logger' in _resume_history_limit_error. hermes_cli/cli_commands_mixin.py (4175 -> 3634): _end_current_session + _sync_agent_to_session (/resume + /branch), _print_side_result_panel (/bg + /btw), _queue_prompt_turn (/learn /plan /init), _toggle_target + _command_arg (/footer /timestamps /focus /voice /wake), _print_diff_body, _summarize_paths, _split_scope_flags + _scope_outcome (/reasoning /fast); dispatch tables for /cron flags + subcommands, /busy, /fast, /voice; /browser, /cron, /goal god-methods split into helpers. Docstrings/comments compacted by hand; rationale kept.
This commit is contained in:
Teknium
@@ -1,15 +1,10 @@
|
||||
"""Agent-construction and session-resume display methods for ``HermesCLI``.
|
||||
|
||||
Extracted from ``cli.py`` as part of the god-file decomposition campaign
|
||||
(``~/.hermes/plans/god-file-decomposition.md``, Phase 4 step 2). This mixin holds
|
||||
the agent lifecycle/setup cluster: runtime-credential resolution, per-turn agent
|
||||
config, first-use agent construction, and resumed-session preload + history recap.
|
||||
|
||||
Behavior-neutral: every method is lifted verbatim from ``HermesCLI``. ``self.*``
|
||||
calls resolve unchanged via the MRO. Neutral dependencies are imported at module
|
||||
top level; ``cli.py``-internal helpers/constants are imported lazily inside each
|
||||
method (``from cli import ...`` resolves at call time, when ``cli`` is fully
|
||||
loaded) so this module never imports ``cli`` at import time -> no import cycle.
|
||||
Holds the agent lifecycle cluster lifted from ``cli.py``: runtime-credential
|
||||
resolution, per-turn agent config, first-use agent construction, and resumed-session
|
||||
preload + history recap. ``cli.py``-internal helpers are imported lazily inside each
|
||||
method (``from cli import ...`` resolves once ``cli`` is fully loaded) so this module
|
||||
never imports ``cli`` at import time -> no import cycle.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -22,48 +17,82 @@ from utils import base_url_host_matches
|
||||
|
||||
|
||||
def _single_query_clarify_callback(question: str, choices=None, multi_select=False) -> str:
|
||||
"""Clarify has no interactive surface in a single-query (-q) turn.
|
||||
"""Headless clarify answer for ``hermes chat -q``.
|
||||
|
||||
``hermes chat -q`` runs one turn without ever building the
|
||||
prompt_toolkit application, so the interactive clarify modal can never
|
||||
be painted or answered — the CLI callback would poll its response queue
|
||||
until ``agent.clarify_timeout`` expires (default 3600 s, 0 = unlimited)
|
||||
while the gateway/cron/kanban-dispatcher caller sees a silent hang. The
|
||||
oneshot path answers immediately via ``_oneshot_clarify_callback``;
|
||||
single-query turns need the same headless behavior (#94943)."""
|
||||
A -q turn never builds the prompt_toolkit app, so the interactive clarify modal
|
||||
can never be painted or answered — the CLI callback would poll until
|
||||
``agent.clarify_timeout`` while the caller sees a silent hang. Mirror the oneshot
|
||||
path and answer immediately instead."""
|
||||
prefix = f"[single-query mode: no user available to answer {question!r}. "
|
||||
if choices:
|
||||
if multi_select:
|
||||
return (
|
||||
f"[single-query mode: no user available to answer {question!r}. "
|
||||
f"Pick the best subset from {choices} using your own judgment "
|
||||
f"and continue.]"
|
||||
)
|
||||
return (
|
||||
f"[single-query mode: no user available to answer {question!r}. "
|
||||
f"Pick the best option from {choices} using your own judgment "
|
||||
f"and continue.]"
|
||||
)
|
||||
what = "subset" if multi_select else "option"
|
||||
return f"{prefix}Pick the best {what} from {choices} using your own judgment and continue.]"
|
||||
return f"{prefix}Make the most reasonable assumption you can and continue.]"
|
||||
|
||||
|
||||
def _current_runtime(cli) -> dict:
|
||||
"""Snapshot the CLI's resolved provider routing as an AIAgent runtime dict.
|
||||
|
||||
getattr guards stay: tests build minimal shells lacking these attributes."""
|
||||
return {
|
||||
"api_key": cli.api_key,
|
||||
"base_url": cli.base_url,
|
||||
"provider": cli.provider,
|
||||
"requested_provider": getattr(cli, "requested_provider", cli.provider),
|
||||
"api_mode": cli.api_mode,
|
||||
"command": cli.acp_command,
|
||||
"args": list(cli.acp_args or []),
|
||||
"credential_pool": getattr(cli, "_credential_pool", None),
|
||||
}
|
||||
|
||||
|
||||
def _route_signature(model, runtime: dict) -> tuple:
|
||||
"""Hashable identity of (model, routing) used to detect when the agent must be rebuilt."""
|
||||
return (
|
||||
f"[single-query mode: no user available to answer {question!r}. Make "
|
||||
f"the most reasonable assumption you can and continue.]"
|
||||
model,
|
||||
runtime.get("provider"),
|
||||
runtime.get("requested_provider"),
|
||||
runtime.get("base_url"),
|
||||
runtime.get("api_mode"),
|
||||
runtime.get("command"),
|
||||
tuple(runtime.get("args") or ()),
|
||||
)
|
||||
|
||||
|
||||
def _compression_descendant(session_db, session_id):
|
||||
"""If ``session_id`` is the (empty) head of a compression chain, return the
|
||||
descendant that actually holds the messages; else None. Fails open on DB errors."""
|
||||
try:
|
||||
resolved_id = session_db.resolve_resume_session_id(session_id)
|
||||
except Exception:
|
||||
return None
|
||||
return resolved_id if resolved_id and resolved_id != session_id else None
|
||||
|
||||
|
||||
# display_kind -> recap event line; ``hidden`` rows are skipped before this lookup.
|
||||
_RESUME_EVENT_TEXT = {
|
||||
"model_switch": "model changed",
|
||||
"async_delegation_complete": "background delegation completed",
|
||||
"auto_continue": "resumed interrupted turn",
|
||||
}
|
||||
|
||||
# (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"),
|
||||
)
|
||||
|
||||
|
||||
class CLIAgentSetupMixin:
|
||||
"""Agent construction + session-resume display methods for ``HermesCLI``."""
|
||||
|
||||
def _ensure_runtime_credentials(self) -> bool:
|
||||
"""
|
||||
Ensure runtime credentials are resolved before agent use.
|
||||
Re-resolves provider credentials so key rotation and token refresh
|
||||
are picked up without restarting the CLI.
|
||||
Returns True if credentials are ready, False on auth failure.
|
||||
"""
|
||||
"""Re-resolve provider credentials before agent use so key rotation / token
|
||||
refresh are picked up without restarting the CLI. False on auth failure."""
|
||||
from cli import ChatConsole, _cprint, logger
|
||||
from hermes_cli.runtime_provider import (
|
||||
resolve_runtime_provider,
|
||||
format_runtime_provider_error,
|
||||
)
|
||||
from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error
|
||||
|
||||
_primary_exc = None
|
||||
runtime = None
|
||||
@@ -119,18 +148,12 @@ class CLIAgentSetupMixin:
|
||||
resolved_api_mode = runtime.get("api_mode", self.api_mode)
|
||||
resolved_acp_command = runtime.get("command")
|
||||
resolved_acp_args = list(runtime.get("args") or [])
|
||||
resolved_credential_pool = runtime.get("credential_pool")
|
||||
# A callable api_key is a bearer-token provider (Azure Foundry
|
||||
# Entra ID — ``azure_identity_adapter.build_token_provider``).
|
||||
# The OpenAI SDK accepts ``Callable[[], str]`` for ``api_key`` and
|
||||
# invokes it before every request. Skip the string-only validation
|
||||
# and placeholder substitution for callables.
|
||||
# A callable api_key is a bearer-token provider (Azure Entra ID): the OpenAI SDK
|
||||
# invokes it per request, so skip string validation / placeholder substitution.
|
||||
_is_callable_provider = callable(api_key) and not isinstance(api_key, str)
|
||||
if not _is_callable_provider and (not isinstance(api_key, str) or not api_key):
|
||||
# Custom / local endpoints (llama.cpp, ollama, vLLM, etc.) often
|
||||
# don't require authentication. When a base_url IS configured but
|
||||
# no API key was found, use a placeholder so the OpenAI SDK
|
||||
# doesn't reject the request and local servers just ignore it.
|
||||
# Custom/local endpoints (llama.cpp, ollama, vLLM) often need no auth: with a
|
||||
# non-OpenRouter base_url use a placeholder key so the SDK doesn't reject it.
|
||||
_source = runtime.get("source", "")
|
||||
_has_custom_base = (
|
||||
isinstance(base_url, str)
|
||||
@@ -169,30 +192,22 @@ class CLIAgentSetupMixin:
|
||||
self.api_mode = resolved_api_mode
|
||||
self.acp_command = resolved_acp_command
|
||||
self.acp_args = resolved_acp_args
|
||||
self._credential_pool = resolved_credential_pool
|
||||
self._credential_pool = runtime.get("credential_pool")
|
||||
self._provider_source = runtime.get("source")
|
||||
self.api_key = api_key
|
||||
self.base_url = base_url
|
||||
|
||||
# When a custom_provider entry carries an explicit `model` field,
|
||||
# use it as the effective model name. Without this, running
|
||||
# `hermes chat --model <provider-name>` sends the provider name
|
||||
# (e.g. "my-provider") as the model string to the API instead of
|
||||
# the configured model (e.g. "qwen3.6-plus"), causing 400 errors.
|
||||
# A custom_provider entry's explicit `model` wins when the CLI model is unset or
|
||||
# is just the provider slug/display name (`hermes chat --model <provider-name>`
|
||||
# would otherwise send the provider name as the model string -> 400).
|
||||
runtime_model = runtime.get("model")
|
||||
if runtime_model and isinstance(runtime_model, str):
|
||||
# Only use runtime model if: model is unset, or model equals provider name
|
||||
should_use_runtime_model = (
|
||||
not self.model or # No model configured yet
|
||||
self.model == self.provider or # Model is the provider slug
|
||||
self.model == runtime.get("name") # Model matches provider display name
|
||||
)
|
||||
if should_use_runtime_model:
|
||||
self.model = runtime_model
|
||||
if runtime_model and isinstance(runtime_model, str) and (
|
||||
not self.model or self.model == self.provider or self.model == runtime.get("name")
|
||||
):
|
||||
self.model = runtime_model
|
||||
|
||||
# If model is still empty (e.g. user ran `hermes auth add openai-codex`
|
||||
# without `hermes model`), fall back to the provider's first catalog
|
||||
# model so the API call doesn't fail with "model must be non-empty".
|
||||
# Still empty (e.g. `hermes auth add` without `hermes model`): fall back to the
|
||||
# provider's first catalog model so the API doesn't reject an empty model.
|
||||
if not self.model and resolved_provider:
|
||||
try:
|
||||
from hermes_cli.models import get_default_model_for_provider
|
||||
@@ -206,12 +221,10 @@ class CLIAgentSetupMixin:
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Normalize model for the resolved provider (e.g. swap non-Codex
|
||||
# models when provider is openai-codex). Fixes #651.
|
||||
# Normalize model for the resolved provider (e.g. swap non-Codex models on openai-codex).
|
||||
model_changed = self._normalize_model_for_provider(resolved_provider)
|
||||
|
||||
# AIAgent/OpenAI client holds auth at init time, so rebuild if key,
|
||||
# routing, or the effective model changed.
|
||||
# AIAgent/OpenAI client holds auth at init, so rebuild on key/routing/model change.
|
||||
if (credentials_changed or routing_changed or model_changed) and self.agent is not None:
|
||||
self.agent = None
|
||||
self._active_agent_route_signature = None
|
||||
@@ -221,12 +234,8 @@ class CLIAgentSetupMixin:
|
||||
def _runtime_credentials_ready(self) -> bool:
|
||||
"""Silently probe whether any inference provider can be resolved.
|
||||
|
||||
Unlike ``_ensure_runtime_credentials`` this never prints and never
|
||||
mutates CLI state — it exists so the interactive first-run path can
|
||||
detect a completely unconfigured install *before* the user types a
|
||||
message into a chat that cannot work (#62935-adjacent UX class:
|
||||
keyless first run must route into onboarding, not a broken chat).
|
||||
"""
|
||||
Never prints or mutates CLI state, so the interactive first-run path can route a
|
||||
keyless install into onboarding before the user types into a chat that can't work."""
|
||||
from hermes_cli.runtime_provider import resolve_runtime_provider
|
||||
|
||||
try:
|
||||
@@ -241,9 +250,7 @@ class CLIAgentSetupMixin:
|
||||
return False
|
||||
api_key = runtime.get("api_key")
|
||||
base_url = runtime.get("base_url")
|
||||
if callable(api_key) and not isinstance(api_key, str):
|
||||
return bool(base_url)
|
||||
if isinstance(api_key, str) and api_key:
|
||||
if (callable(api_key) and not isinstance(api_key, str)) or (isinstance(api_key, str) and api_key):
|
||||
return bool(base_url)
|
||||
# Keyless custom/local endpoints (ollama, llama.cpp, vLLM…) are fine.
|
||||
return bool(
|
||||
@@ -253,15 +260,9 @@ class CLIAgentSetupMixin:
|
||||
)
|
||||
|
||||
def _offer_first_run_setup(self) -> bool:
|
||||
"""Offer the provider picker when no provider is configured at all.
|
||||
|
||||
Called from the interactive startup path when
|
||||
``_runtime_credentials_ready()`` is False and stdin is a TTY. Runs the
|
||||
exact same flow as ``hermes model`` (which fronts Quick Setup / Nous
|
||||
Portal OAuth as the first, recommended option) so there is a single
|
||||
source of truth for provider onboarding. Returns True when a provider
|
||||
was configured.
|
||||
"""
|
||||
"""Offer the provider picker when no provider is configured at all (interactive
|
||||
startup, TTY). Runs the same flow as ``hermes model`` so onboarding has a single
|
||||
source of truth. True when a provider was configured."""
|
||||
from cli import _cprint, logger
|
||||
|
||||
_cprint("")
|
||||
@@ -290,8 +291,7 @@ class CLIAgentSetupMixin:
|
||||
_cprint(" Run 'hermes model' to try again.")
|
||||
return False
|
||||
|
||||
# Re-sync CLI state from what the picker persisted so the very next
|
||||
# turn uses the new provider without a restart.
|
||||
# Re-sync CLI state from what the picker persisted so the next turn uses it without a restart.
|
||||
try:
|
||||
from hermes_cli.config import load_config
|
||||
_model_cfg = (load_config().get("model") or {})
|
||||
@@ -299,9 +299,7 @@ class CLIAgentSetupMixin:
|
||||
_new_provider = (_model_cfg.get("provider") or "").strip()
|
||||
if _new_provider:
|
||||
self.requested_provider = _new_provider
|
||||
_new_model = (
|
||||
_model_cfg.get("default") or _model_cfg.get("model") or ""
|
||||
).strip()
|
||||
_new_model = (_model_cfg.get("default") or _model_cfg.get("model") or "").strip()
|
||||
if _new_model:
|
||||
self.model = _new_model
|
||||
except Exception as exc:
|
||||
@@ -317,74 +315,108 @@ class CLIAgentSetupMixin:
|
||||
return False
|
||||
|
||||
def _resolve_turn_agent_config(self, user_message: str) -> dict:
|
||||
"""Build the effective model/runtime config for a single user turn.
|
||||
|
||||
Always uses the session's primary model/provider. If the user has
|
||||
toggled `/fast` on and the current model supports Priority
|
||||
Processing / Anthropic fast mode, attach `request_overrides` so the
|
||||
API call is marked accordingly.
|
||||
"""
|
||||
"""Effective model/runtime config for one turn — always the session's primary
|
||||
provider. With `/fast` on (service_tier == "priority") attach request_overrides;
|
||||
auto/cold tiers are applied per request by agent.fast_mode instead."""
|
||||
from hermes_cli.models import resolve_fast_mode_overrides
|
||||
|
||||
runtime = {
|
||||
"api_key": self.api_key,
|
||||
"base_url": self.base_url,
|
||||
"provider": self.provider,
|
||||
"requested_provider": getattr(
|
||||
self, "requested_provider", self.provider
|
||||
),
|
||||
"api_mode": self.api_mode,
|
||||
"command": self.acp_command,
|
||||
"args": list(self.acp_args or []),
|
||||
"credential_pool": getattr(self, "_credential_pool", None),
|
||||
}
|
||||
route = {
|
||||
"model": self.model,
|
||||
"runtime": runtime,
|
||||
"signature": (
|
||||
self.model,
|
||||
runtime["provider"],
|
||||
runtime["requested_provider"],
|
||||
runtime["base_url"],
|
||||
runtime["api_mode"],
|
||||
runtime["command"],
|
||||
tuple(runtime["args"]),
|
||||
),
|
||||
}
|
||||
runtime = _current_runtime(self)
|
||||
route = {"model": self.model, "runtime": runtime, "signature": _route_signature(self.model, runtime)}
|
||||
|
||||
service_tier = getattr(self, "service_tier", None)
|
||||
if service_tier != "priority":
|
||||
# None (normal) or auto/cold — the bounded window is applied per
|
||||
# request by agent.fast_mode, not pinned into request_overrides.
|
||||
if getattr(self, "service_tier", None) != "priority":
|
||||
route["request_overrides"] = None
|
||||
return route
|
||||
|
||||
try:
|
||||
overrides = resolve_fast_mode_overrides(
|
||||
route["model"],
|
||||
provider=runtime["provider"],
|
||||
base_url=runtime["base_url"],
|
||||
route["model"], provider=runtime["provider"], base_url=runtime["base_url"],
|
||||
)
|
||||
except Exception:
|
||||
overrides = None
|
||||
route["request_overrides"] = overrides
|
||||
return route
|
||||
|
||||
def _load_resumed_history_late(self) -> bool:
|
||||
"""Late resume path: validate the session and load its history from the DB when
|
||||
_preload_resumed_session() (called from run()) did not already populate it.
|
||||
False when the resume must abort (missing session / over the safe-resume limit)."""
|
||||
from cli import ChatConsole, _DIM, _RST, _accent_hex, _cprint
|
||||
session_meta = self._session_db.get_session(self.session_id)
|
||||
# Quiet mode (tool_progress_mode == "off") routes resume status lines to
|
||||
# stderr so stdout stays machine-readable for `$(hermes chat -Q --resume ...)`.
|
||||
_quiet_mode = getattr(self, "tool_progress_mode", "full") == "off"
|
||||
|
||||
def _say(plain: str, rich: str) -> None:
|
||||
if _quiet_mode:
|
||||
print(plain, file=sys.stderr)
|
||||
else:
|
||||
ChatConsole().print(rich)
|
||||
|
||||
if not session_meta:
|
||||
if _quiet_mode:
|
||||
print(f"Session not found: {self.session_id}", file=sys.stderr)
|
||||
print("Use a session ID from a previous CLI run (hermes sessions list).", file=sys.stderr)
|
||||
else:
|
||||
_cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}")
|
||||
_cprint(f"{_DIM}Use a session ID from a previous CLI run (hermes sessions list).{_RST}")
|
||||
return False
|
||||
resolved_id = _compression_descendant(self._session_db, self.session_id)
|
||||
if resolved_id:
|
||||
ChatConsole().print(
|
||||
f"[dim]Session {_escape(self.session_id)} was compressed into "
|
||||
f"{_escape(resolved_id)}; resuming the descendant with your "
|
||||
f"transcript.[/dim]"
|
||||
)
|
||||
self.session_id = resolved_id
|
||||
session_meta = self._session_db.get_session(self.session_id) or session_meta
|
||||
if getattr(self, "_resume_history_error", None):
|
||||
return False
|
||||
# Only the TIP session's rows are loaded here (no ancestors), so use the
|
||||
# tip-only count — the full-lineage count would over-reject compressed sessions.
|
||||
resume_limit_error = self._resume_history_limit_error(tip_only=True)
|
||||
if resume_limit_error:
|
||||
self._resume_history_error = resume_limit_error
|
||||
_say(
|
||||
f"Cannot resume session: {resume_limit_error}",
|
||||
f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}",
|
||||
)
|
||||
return False
|
||||
restored = self._session_db.get_messages_as_conversation(self.session_id, repair_alternation=True)
|
||||
if restored:
|
||||
restored = [m for m in restored if m.get("role") != "session_meta"]
|
||||
self.conversation_history = restored
|
||||
msg_count = len([m for m in restored if m.get("role") == "user"])
|
||||
title_part = f" \"{session_meta['title']}\"" if session_meta.get("title") else ""
|
||||
counts = f"({msg_count} user message{'s' if msg_count != 1 else ''}, {len(restored)} total messages)"
|
||||
_say(
|
||||
f"↻ Resumed session {self.session_id}{title_part} {counts}",
|
||||
f"[bold {_accent_hex()}]↻ Resumed session[/] [bold]{_escape(self.session_id)}[/]"
|
||||
f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}",
|
||||
)
|
||||
self._restore_session_cwd(session_meta, quiet=_quiet_mode)
|
||||
self._restore_session_yolo(session_meta, quiet=_quiet_mode)
|
||||
self._restore_session_model(session_meta, quiet=_quiet_mode)
|
||||
else:
|
||||
_say(
|
||||
f"Session {self.session_id} found but has no messages. Starting fresh.",
|
||||
f"[bold {_accent_hex()}]Session {_escape(self.session_id)} found but has no messages. Starting fresh.[/]",
|
||||
)
|
||||
# Re-open the session (clear ended_at so it's active again)
|
||||
try:
|
||||
self._session_db.reopen_session(self.session_id)
|
||||
except Exception:
|
||||
pass
|
||||
return True
|
||||
|
||||
def _init_agent(self, *, model_override: str = None, runtime_override: dict = None, request_overrides: dict | None = None) -> bool:
|
||||
"""
|
||||
Initialize the agent on first use.
|
||||
When resuming a session, restores conversation history from SQLite.
|
||||
|
||||
Returns:
|
||||
bool: True if successful, False otherwise
|
||||
"""
|
||||
from cli import AIAgent, ChatConsole, _DIM, _RST, _accent_hex, _cprint, _prepare_deferred_agent_startup, logger
|
||||
"""Build the agent on first use; when resuming, restore history from SQLite.
|
||||
Returns True on success."""
|
||||
from cli import AIAgent, ChatConsole, _cprint, _prepare_deferred_agent_startup, logger
|
||||
if self.agent is not None:
|
||||
return True
|
||||
|
||||
# Join the background preloaded-skills load (cli.py cmd_chat starts
|
||||
# it when --skills/-s is passed) BEFORE the agent snapshots
|
||||
# self.system_prompt below. No-op when nothing was requested.
|
||||
# Join the background preloaded-skills load (--skills/-s) BEFORE the agent
|
||||
# snapshots self.system_prompt below. No-op when nothing was requested.
|
||||
self.finalize_preloaded_skills()
|
||||
|
||||
_prepare_deferred_agent_startup()
|
||||
@@ -401,126 +433,21 @@ class CLIAgentSetupMixin:
|
||||
single_query=getattr(self, "_single_query_mode", False),
|
||||
)
|
||||
|
||||
# Initialize SQLite session store for CLI sessions (if not already done in __init__)
|
||||
if self._session_db is None:
|
||||
try:
|
||||
from hermes_state import SessionDB
|
||||
self._session_db = SessionDB()
|
||||
except Exception as e:
|
||||
logger.warning("SQLite session store not available — session will NOT be indexed: %s", e)
|
||||
|
||||
# If resuming, validate the session exists and load its history.
|
||||
# _preload_resumed_session() may have already loaded it (called from
|
||||
# run() for immediate display). In that case, conversation_history
|
||||
# is non-empty and we skip the DB round-trip.
|
||||
if self._resumed and self._session_db and not self.conversation_history:
|
||||
session_meta = self._session_db.get_session(self.session_id)
|
||||
# In quiet mode (`hermes chat -Q` / --quiet, surfaced via
|
||||
# tool_progress_mode == "off"), resume status lines go to stderr
|
||||
# so stdout stays machine-readable for automation wrappers that
|
||||
# do `$(hermes chat -Q --resume <id> -q "...")`. Without this,
|
||||
# the resume banner pollutes captured stdout. See #11793.
|
||||
_quiet_mode = getattr(self, "tool_progress_mode", "full") == "off"
|
||||
if not session_meta:
|
||||
if _quiet_mode:
|
||||
print(f"Session not found: {self.session_id}", file=sys.stderr)
|
||||
print(
|
||||
"Use a session ID from a previous CLI run (hermes sessions list).",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
_cprint(f"\033[1;31mSession not found: {self.session_id}{_RST}")
|
||||
_cprint(f"{_DIM}Use a session ID from a previous CLI run (hermes sessions list).{_RST}")
|
||||
return False
|
||||
# If the requested session is the (empty) head of a compression
|
||||
# chain, walk to the descendant that actually holds the messages.
|
||||
# See #15000 and SessionDB.resolve_resume_session_id.
|
||||
try:
|
||||
resolved_id = self._session_db.resolve_resume_session_id(self.session_id)
|
||||
except Exception:
|
||||
resolved_id = self.session_id
|
||||
if resolved_id and resolved_id != self.session_id:
|
||||
ChatConsole().print(
|
||||
f"[dim]Session {_escape(self.session_id)} was compressed into "
|
||||
f"{_escape(resolved_id)}; resuming the descendant with your "
|
||||
f"transcript.[/dim]"
|
||||
)
|
||||
self.session_id = resolved_id
|
||||
resolved_meta = self._session_db.get_session(self.session_id)
|
||||
if resolved_meta:
|
||||
session_meta = resolved_meta
|
||||
prior_resume_error = getattr(self, "_resume_history_error", None)
|
||||
if prior_resume_error:
|
||||
return False
|
||||
# This path loads only the TIP session's rows (no ancestors),
|
||||
# so guard with a tip-only count — the full-lineage count would
|
||||
# over-reject heavily-compressed sessions with a small tip.
|
||||
resume_limit_error = self._resume_history_limit_error(tip_only=True)
|
||||
if resume_limit_error:
|
||||
self._resume_history_error = resume_limit_error
|
||||
if _quiet_mode:
|
||||
print(f"Cannot resume session: {resume_limit_error}", file=sys.stderr)
|
||||
else:
|
||||
ChatConsole().print(
|
||||
f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}"
|
||||
)
|
||||
return False
|
||||
restored = self._session_db.get_messages_as_conversation(
|
||||
self.session_id, repair_alternation=True
|
||||
)
|
||||
if restored:
|
||||
restored = [m for m in restored if m.get("role") != "session_meta"]
|
||||
self.conversation_history = restored
|
||||
msg_count = len([m for m in restored if m.get("role") == "user"])
|
||||
title_part = ""
|
||||
if session_meta.get("title"):
|
||||
title_part = f" \"{session_meta['title']}\""
|
||||
if _quiet_mode:
|
||||
print(
|
||||
f"↻ Resumed session {self.session_id}{title_part} "
|
||||
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
|
||||
f"{len(restored)} total messages)",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
ChatConsole().print(
|
||||
f"[bold {_accent_hex()}]↻ Resumed session[/] "
|
||||
f"[bold]{_escape(self.session_id)}[/]"
|
||||
f"[bold {_accent_hex()}]{_escape(title_part)}[/] "
|
||||
f"({msg_count} user message{'s' if msg_count != 1 else ''}, {len(restored)} total messages)"
|
||||
)
|
||||
self._restore_session_cwd(session_meta, quiet=_quiet_mode)
|
||||
self._restore_session_yolo(session_meta, quiet=_quiet_mode)
|
||||
self._restore_session_model(session_meta, quiet=_quiet_mode)
|
||||
else:
|
||||
if _quiet_mode:
|
||||
print(
|
||||
f"Session {self.session_id} found but has no messages. Starting fresh.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
else:
|
||||
ChatConsole().print(
|
||||
f"[bold {_accent_hex()}]Session {_escape(self.session_id)} found but has no messages. Starting fresh.[/]"
|
||||
)
|
||||
# Re-open the session (clear ended_at so it's active again)
|
||||
try:
|
||||
self._session_db.reopen_session(self.session_id)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
if (
|
||||
self._resumed and self._session_db and not self.conversation_history
|
||||
and not self._load_resumed_history_late()
|
||||
):
|
||||
return False
|
||||
|
||||
try:
|
||||
runtime = runtime_override or {
|
||||
"api_key": self.api_key,
|
||||
"base_url": self.base_url,
|
||||
"provider": self.provider,
|
||||
"requested_provider": getattr(
|
||||
self, "requested_provider", self.provider
|
||||
),
|
||||
"api_mode": self.api_mode,
|
||||
"command": self.acp_command,
|
||||
"args": list(self.acp_args or []),
|
||||
"credential_pool": getattr(self, "_credential_pool", None),
|
||||
}
|
||||
runtime = runtime_override or _current_runtime(self)
|
||||
effective_model = model_override or self.model
|
||||
self.agent = AIAgent(
|
||||
model=effective_model,
|
||||
@@ -555,17 +482,14 @@ class CLIAgentSetupMixin:
|
||||
session_id=self.session_id,
|
||||
platform="cli",
|
||||
session_db=self._session_db,
|
||||
# A -q turn never builds the prompt_toolkit application, so
|
||||
# the interactive modal can never be painted or answered —
|
||||
# answer headless instead of polling until clarify_timeout
|
||||
# (#94943; mirrors _oneshot_clarify_callback on the -z path).
|
||||
# -q never builds the prompt_toolkit app, so the clarify modal can't be
|
||||
# answered — answer headless instead of polling until clarify_timeout.
|
||||
clarify_callback=(
|
||||
_single_query_clarify_callback
|
||||
if getattr(self, "_single_query_mode", False)
|
||||
else self._clarify_callback
|
||||
),
|
||||
reasoning_callback=self._current_reasoning_callback(),
|
||||
|
||||
fallback_model=self._fallback_model,
|
||||
thinking_callback=self._on_thinking,
|
||||
checkpoints_enabled=self.checkpoints_enabled,
|
||||
@@ -584,39 +508,23 @@ class CLIAgentSetupMixin:
|
||||
notice_clear_callback=self._on_notice_clear,
|
||||
reaction_callback=self._on_reaction,
|
||||
)
|
||||
# Store reference for atexit memory provider shutdown.
|
||||
# NOTE: this MUST write to the ``cli`` module's global, not a
|
||||
# local module global. ``_run_cleanup`` (in cli.py) reads
|
||||
# ``cli._active_agent_ref`` to decide whether to fire the memory
|
||||
# provider's ``on_session_end`` hook. When this code lived in
|
||||
# cli.py a bare ``global _active_agent_ref`` worked; after the
|
||||
# god-file extraction into this mixin a ``global`` here would bind
|
||||
# *this module's* namespace, leaving ``cli._active_agent_ref`` None
|
||||
# forever — so memory shutdown never ran on /exit (#49287).
|
||||
# Reference for atexit memory-provider shutdown: ``_run_cleanup`` in cli.py
|
||||
# reads ``cli._active_agent_ref``, so this MUST write the ``cli`` module's
|
||||
# global — a ``global`` statement here would bind this module's namespace.
|
||||
import cli as _cli
|
||||
_cli._active_agent_ref = self.agent
|
||||
# Route agent status output through prompt_toolkit so ANSI escape
|
||||
# sequences aren't garbled by patch_stdout's StdoutProxy (#2262).
|
||||
# Route agent status output through prompt_toolkit so ANSI escapes
|
||||
# aren't garbled by patch_stdout's StdoutProxy.
|
||||
self.agent._print_fn = _cprint
|
||||
# Hydrate credits notices at session OPEN (parity with the TUI), so a
|
||||
# depletion / usage-band warning shows before the first message. The
|
||||
# notice_callback is bound above → _on_notice renders the line. Idempotent
|
||||
# + fail-open inside the helper; harmless for non-Nous providers.
|
||||
# Hydrate credits notices at session OPEN (parity with the TUI) so a depletion
|
||||
# warning shows before the first message. Idempotent + fail-open in the helper.
|
||||
try:
|
||||
from agent.credits_tracker import seed_credits_at_session_start
|
||||
|
||||
seed_credits_at_session_start(self.agent)
|
||||
except Exception:
|
||||
pass
|
||||
self._active_agent_route_signature = (
|
||||
effective_model,
|
||||
runtime.get("provider"),
|
||||
runtime.get("requested_provider"),
|
||||
runtime.get("base_url"),
|
||||
runtime.get("api_mode"),
|
||||
runtime.get("command"),
|
||||
tuple(runtime.get("args") or ()),
|
||||
)
|
||||
self._active_agent_route_signature = _route_signature(effective_model, runtime)
|
||||
|
||||
# Force-create DB row on /title intent, then apply title.
|
||||
if self._pending_title and self._session_db and self.agent:
|
||||
@@ -643,17 +551,13 @@ class CLIAgentSetupMixin:
|
||||
def _resume_history_limit_error(self, tip_only: bool = False):
|
||||
"""Return a safe-resume error without materializing transcript rows.
|
||||
|
||||
``tip_only`` matches call sites that load only the tip session's rows
|
||||
(``get_messages_as_conversation`` without ancestors) — counting the
|
||||
full lineage there would over-reject heavily-compressed sessions
|
||||
whose tip is small. Generic guard failures fail OPEN (resume
|
||||
proceeds) — only a genuine over-limit result blocks.
|
||||
"""
|
||||
``tip_only`` matches call sites that load only the tip session's rows — counting
|
||||
the full lineage there would over-reject heavily-compressed sessions with a small
|
||||
tip. Generic guard failures fail OPEN; only a genuine over-limit result blocks."""
|
||||
if not self._session_db:
|
||||
return None
|
||||
from hermes_state import (
|
||||
SessionResumeTooLargeError,
|
||||
)
|
||||
from cli import logger
|
||||
from hermes_state import SessionResumeTooLargeError
|
||||
|
||||
try:
|
||||
safety_check = getattr(self._session_db, "assert_resume_safe", None)
|
||||
@@ -674,95 +578,63 @@ class CLIAgentSetupMixin:
|
||||
return None
|
||||
|
||||
def _preload_resumed_session(self) -> bool:
|
||||
"""Load a resumed session's history from the DB early (before first chat).
|
||||
|
||||
Called from run() so the conversation history is available for display
|
||||
before the user sends their first message. Sets
|
||||
``self.conversation_history`` and prints the one-liner status. Returns
|
||||
True if history was loaded, False otherwise.
|
||||
|
||||
The corresponding block in ``_init_agent()`` checks whether history is
|
||||
already populated and skips the DB round-trip.
|
||||
"""
|
||||
"""Load a resumed session's history early (from run(), before the first chat) so
|
||||
it can be displayed; ``_init_agent()`` then skips its own DB round-trip. Sets
|
||||
``self.conversation_history`` and prints the status line. True if history loaded."""
|
||||
from cli import _accent_hex
|
||||
if not self._resumed or not self._session_db:
|
||||
return False
|
||||
|
||||
session_meta = self._session_db.get_session(self.session_id)
|
||||
if not session_meta:
|
||||
self._console_print(
|
||||
f"[bold red]Session not found: {self.session_id}[/]"
|
||||
)
|
||||
self._console_print(
|
||||
"[dim]Use a session ID from a previous CLI run "
|
||||
"(hermes sessions list).[/]"
|
||||
)
|
||||
self._console_print(f"[bold red]Session not found: {self.session_id}[/]")
|
||||
self._console_print("[dim]Use a session ID from a previous CLI run (hermes sessions list).[/]")
|
||||
return False
|
||||
|
||||
# If the requested session is the (empty) head of a compression chain,
|
||||
# walk to the descendant that actually holds the messages. See #15000.
|
||||
try:
|
||||
resolved_id = self._session_db.resolve_resume_session_id(self.session_id)
|
||||
except Exception:
|
||||
resolved_id = self.session_id
|
||||
if resolved_id and resolved_id != self.session_id:
|
||||
resolved_id = _compression_descendant(self._session_db, self.session_id)
|
||||
if resolved_id:
|
||||
self._console_print(
|
||||
f"[dim]Session {self.session_id} was compressed into "
|
||||
f"{resolved_id}; resuming the descendant with your transcript.[/]"
|
||||
)
|
||||
self.session_id = resolved_id
|
||||
resolved_meta = self._session_db.get_session(self.session_id)
|
||||
if resolved_meta:
|
||||
session_meta = resolved_meta
|
||||
session_meta = self._session_db.get_session(self.session_id) or session_meta
|
||||
|
||||
resume_limit_error = self._resume_history_limit_error()
|
||||
if resume_limit_error:
|
||||
self._resume_history_error = resume_limit_error
|
||||
self._console_print(
|
||||
f"[bold red]Cannot resume session:[/] {resume_limit_error}"
|
||||
)
|
||||
self._console_print(f"[bold red]Cannot resume session:[/] {resume_limit_error}")
|
||||
return False
|
||||
|
||||
model_history, display_history = self._session_db.get_resume_conversations(self.session_id)
|
||||
restored = model_history
|
||||
if restored:
|
||||
restored = [m for m in restored if m.get("role") != "session_meta"]
|
||||
self.conversation_history = restored
|
||||
self._resume_display_history = [
|
||||
m for m in display_history if m.get("role") != "session_meta"
|
||||
]
|
||||
from agent.context_compressor import is_user_originated_turn
|
||||
|
||||
# Count only user-originated turns (#80622): legacy compaction
|
||||
# handoffs are durable role=user rows without display_kind.
|
||||
msg_count = len(
|
||||
[
|
||||
m
|
||||
for m in self._resume_display_history
|
||||
if is_user_originated_turn(m)
|
||||
]
|
||||
)
|
||||
title_part = ""
|
||||
if session_meta.get("title"):
|
||||
title_part = f' "{session_meta["title"]}"'
|
||||
accent_color = _accent_hex()
|
||||
self._console_print(
|
||||
f"[{accent_color}]↻ Resumed session [bold]{self.session_id}[/bold]"
|
||||
f"{title_part} "
|
||||
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
|
||||
f"{len(restored)} total messages)[/]"
|
||||
)
|
||||
self._restore_session_cwd(session_meta)
|
||||
self._restore_session_yolo(session_meta)
|
||||
self._restore_session_model(session_meta)
|
||||
else:
|
||||
accent_color = _accent_hex()
|
||||
accent_color = _accent_hex()
|
||||
if not restored:
|
||||
self._console_print(
|
||||
f"[{accent_color}]Session {self.session_id} found but has no "
|
||||
f"messages. Starting fresh.[/]"
|
||||
)
|
||||
return False
|
||||
|
||||
restored = [m for m in restored if m.get("role") != "session_meta"]
|
||||
self.conversation_history = restored
|
||||
self._resume_display_history = [m for m in display_history if m.get("role") != "session_meta"]
|
||||
from agent.context_compressor import is_user_originated_turn
|
||||
|
||||
# Count only user-originated turns: legacy compaction handoffs are durable
|
||||
# role=user rows without display_kind.
|
||||
msg_count = len([m for m in self._resume_display_history if is_user_originated_turn(m)])
|
||||
title_part = f' "{session_meta["title"]}"' if session_meta.get("title") else ""
|
||||
self._console_print(
|
||||
f"[{accent_color}]↻ Resumed session [bold]{self.session_id}[/bold]"
|
||||
f"{title_part} "
|
||||
f"({msg_count} user message{'s' if msg_count != 1 else ''}, "
|
||||
f"{len(restored)} total messages)[/]"
|
||||
)
|
||||
self._restore_session_cwd(session_meta)
|
||||
self._restore_session_yolo(session_meta)
|
||||
self._restore_session_model(session_meta)
|
||||
|
||||
# Re-open the session (clear ended_at so it's active again)
|
||||
try:
|
||||
self._session_db.reopen_session(self.session_id)
|
||||
@@ -772,24 +644,14 @@ class CLIAgentSetupMixin:
|
||||
return True
|
||||
|
||||
def _display_resumed_history(self):
|
||||
"""Render a compact recap of previous conversation messages.
|
||||
|
||||
Uses Rich markup with dim/muted styling so the recap is visually
|
||||
distinct from the active conversation. Caps the display at the
|
||||
last ``MAX_DISPLAY_EXCHANGES`` user/assistant exchanges and shows
|
||||
an indicator for earlier hidden messages.
|
||||
"""
|
||||
"""Render a dim Rich-panel recap of the previous conversation, capped at the last
|
||||
``resume_exchanges`` user/assistant exchanges with a hidden-count indicator."""
|
||||
from cli import CLI_CONFIG, _record_output_history_entry, _strip_reasoning_tags, _suspend_output_history
|
||||
from tools.ansi_strip import sanitize_display_text as _sanitize_display_text
|
||||
display_history = getattr(self, "_resume_display_history", self.conversation_history)
|
||||
if not display_history:
|
||||
if not display_history or self.resume_display == "minimal":
|
||||
return
|
||||
|
||||
# Check config: resume_display setting
|
||||
if self.resume_display == "minimal":
|
||||
return
|
||||
|
||||
# Read limits from config (with hardcoded defaults)
|
||||
_disp = CLI_CONFIG.get("display", {})
|
||||
MAX_DISPLAY_EXCHANGES = int(_disp.get("resume_exchanges", 10))
|
||||
MAX_USER_LEN = int(_disp.get("resume_max_user_chars", 300))
|
||||
@@ -797,8 +659,8 @@ class CLIAgentSetupMixin:
|
||||
MAX_ASST_LINES = int(_disp.get("resume_max_assistant_lines", 3))
|
||||
SKIP_TOOL_ONLY = _disp.get("resume_skip_tool_only", True)
|
||||
|
||||
# Collect displayable entries (skip system, tool-result messages)
|
||||
entries = [] # list of (role, display_text)
|
||||
# Collect displayable (role, text) entries; system and tool-result rows are skipped.
|
||||
entries = []
|
||||
_last_asst_idx = None # index of last assistant entry
|
||||
_last_asst_full = None # un-truncated display text for last assistant
|
||||
for msg in display_history:
|
||||
@@ -809,36 +671,22 @@ class CLIAgentSetupMixin:
|
||||
|
||||
if display_kind == "hidden":
|
||||
continue
|
||||
if display_kind == "model_switch":
|
||||
entries.append(("event", "model changed"))
|
||||
if display_kind in _RESUME_EVENT_TEXT:
|
||||
entries.append(("event", _RESUME_EVENT_TEXT[display_kind]))
|
||||
continue
|
||||
if display_kind == "async_delegation_complete":
|
||||
entries.append(("event", "background delegation completed"))
|
||||
continue
|
||||
if display_kind == "auto_continue":
|
||||
entries.append(("event", "resumed interrupted turn"))
|
||||
continue
|
||||
|
||||
if role == "system":
|
||||
continue
|
||||
if role == "tool":
|
||||
if role in ("system", "tool"):
|
||||
continue
|
||||
|
||||
if role == "user":
|
||||
text = "" if content is None else str(content)
|
||||
# Handle multimodal content (list of dicts)
|
||||
if isinstance(content, list):
|
||||
parts = []
|
||||
for part in content:
|
||||
if isinstance(part, dict) and part.get("type") == "text":
|
||||
parts.append(part.get("text", ""))
|
||||
elif isinstance(part, dict) and part.get("type") == "image_url":
|
||||
parts.append("[image]")
|
||||
text = " ".join(parts)
|
||||
# Stored history is untrusted for display: strip escape
|
||||
# sequences/control chars so replaying a message can't
|
||||
# clear the screen, retitle the window, or restyle the
|
||||
# recap panel (see tools/ansi_strip.sanitize_display_text).
|
||||
if isinstance(content, list): # multimodal: text parts + [image] markers
|
||||
text = " ".join(
|
||||
part.get("text", "") if part.get("type") == "text" else "[image]"
|
||||
for part in content
|
||||
if isinstance(part, dict) and part.get("type") in ("text", "image_url")
|
||||
)
|
||||
# Stored history is untrusted for display: strip escape sequences/control
|
||||
# chars so replay can't clear the screen, retitle the window or restyle the panel.
|
||||
text = _sanitize_display_text(text)
|
||||
if len(text) > MAX_USER_LEN:
|
||||
text = text[:MAX_USER_LEN] + "..."
|
||||
@@ -858,27 +706,20 @@ class CLIAgentSetupMixin:
|
||||
text = text[:MAX_ASST_LEN] + "..."
|
||||
parts.append(text)
|
||||
if tool_calls:
|
||||
tc_count = len(tool_calls)
|
||||
# Extract tool names
|
||||
names = []
|
||||
for tc in tool_calls:
|
||||
fn = tc.get("function", {})
|
||||
name = fn.get("name", "unknown") if isinstance(fn, dict) else "unknown"
|
||||
if name not in names:
|
||||
names.append(name)
|
||||
names_str = ", ".join(names[:4])
|
||||
if len(names) > 4:
|
||||
names_str += ", ..."
|
||||
noun = "call" if tc_count == 1 else "calls"
|
||||
tc_summary = f"[{tc_count} tool {noun}: {names_str}]"
|
||||
names_str = ", ".join(names[:4]) + (", ..." if len(names) > 4 else "")
|
||||
noun = "call" if len(tool_calls) == 1 else "calls"
|
||||
tc_summary = f"[{len(tool_calls)} tool {noun}: {names_str}]"
|
||||
parts.append(tc_summary)
|
||||
full_parts.append(tc_summary)
|
||||
if not parts:
|
||||
# Skip pure-reasoning messages that have no visible output
|
||||
continue
|
||||
# Skip tool-call-only entries when SKIP_TOOL_ONLY is enabled
|
||||
has_text = bool(text)
|
||||
if SKIP_TOOL_ONLY and not has_text and tool_calls:
|
||||
# Skip pure-reasoning messages with no visible output, and tool-call-only
|
||||
# entries when SKIP_TOOL_ONLY is enabled.
|
||||
if not text and (SKIP_TOOL_ONLY or not tool_calls):
|
||||
continue
|
||||
entries.append(("assistant", " ".join(parts)))
|
||||
_last_asst_idx = len(entries) - 1
|
||||
@@ -887,66 +728,52 @@ class CLIAgentSetupMixin:
|
||||
if not entries:
|
||||
return
|
||||
|
||||
# Determine if we need to truncate
|
||||
skipped = 0
|
||||
if len(entries) > MAX_DISPLAY_EXCHANGES * 2:
|
||||
skipped = len(entries) - MAX_DISPLAY_EXCHANGES * 2
|
||||
entries = entries[skipped:]
|
||||
|
||||
# Replace last assistant entry with full (un-truncated) text
|
||||
# so the user can see where they left off without wasting tokens.
|
||||
# Show the last assistant entry in full so the user sees where they left off.
|
||||
if _last_asst_idx is not None and _last_asst_full:
|
||||
adj_idx = _last_asst_idx - skipped
|
||||
if 0 <= adj_idx < len(entries):
|
||||
entries[adj_idx] = ("assistant_last", _last_asst_full)
|
||||
|
||||
# Build the display using Rich
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
try:
|
||||
from hermes_cli.skin_engine import get_active_skin
|
||||
_skin = get_active_skin()
|
||||
_history_text_c = _skin.get_color("banner_text", "#FFF8DC")
|
||||
_session_label_c = _skin.get_color("session_label", "#DAA520")
|
||||
_session_border_c = _skin.get_color("session_border", "#8B8682")
|
||||
_assistant_label_c = _skin.get_color("ui_ok", "#8FBC8F")
|
||||
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
|
||||
_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS
|
||||
)
|
||||
except Exception:
|
||||
_history_text_c = "#FFF8DC"
|
||||
_session_label_c = "#DAA520"
|
||||
_session_border_c = "#8B8682"
|
||||
_assistant_label_c = "#8FBC8F"
|
||||
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
|
||||
default for _, default in _RESUME_SKIN_COLORS
|
||||
)
|
||||
|
||||
# role -> (label, label style, body style, continuation indent)
|
||||
role_styles = {
|
||||
"user": (" ● You: ", f"dim bold {_session_label_c}", "dim", " " * 9),
|
||||
"assistant": (" ◆ Hermes: ", f"dim bold {_assistant_label_c}", "dim", " " * 12),
|
||||
"assistant_last": (" ◆ Hermes: ", f"bold {_assistant_label_c}", "", " " * 12), # full, non-dim
|
||||
}
|
||||
|
||||
lines = Text()
|
||||
if skipped:
|
||||
lines.append(
|
||||
f" ... {skipped} earlier messages ...\n\n",
|
||||
style="dim italic",
|
||||
)
|
||||
lines.append(f" ... {skipped} earlier messages ...\n\n", style="dim italic")
|
||||
|
||||
for i, (role, text) in enumerate(entries):
|
||||
if role == "event":
|
||||
lines.append(f" ◈ {text}\n", style="dim italic")
|
||||
elif role == "user":
|
||||
lines.append(" ● You: ", style=f"dim bold {_session_label_c}")
|
||||
# Show first line inline, indent rest
|
||||
msg_lines = text.splitlines() or [""]
|
||||
lines.append(msg_lines[0] + "\n", style="dim")
|
||||
for ml in msg_lines[1:]:
|
||||
lines.append(f" {ml}\n", style="dim")
|
||||
elif role == "assistant_last":
|
||||
# Last assistant response shown in full, non-dim
|
||||
lines.append(" ◆ Hermes: ", style=f"bold {_assistant_label_c}")
|
||||
msg_lines = text.splitlines() or [""]
|
||||
lines.append(msg_lines[0] + "\n", style="")
|
||||
for ml in msg_lines[1:]:
|
||||
lines.append(f" {ml}\n", style="")
|
||||
else:
|
||||
lines.append(" ◆ Hermes: ", style=f"dim bold {_assistant_label_c}")
|
||||
msg_lines = text.splitlines() or [""]
|
||||
lines.append(msg_lines[0] + "\n", style="dim")
|
||||
for ml in msg_lines[1:]:
|
||||
lines.append(f" {ml}\n", style="dim")
|
||||
label, label_style, body_style, indent = role_styles[role]
|
||||
lines.append(label, style=label_style)
|
||||
first, *rest = text.splitlines() or [""] # first line inline, rest indented
|
||||
lines.append(first + "\n", style=body_style)
|
||||
for ml in rest:
|
||||
lines.append(f"{indent}{ml}\n", style=body_style)
|
||||
if i < len(entries) - 1:
|
||||
lines.append("") # small gap
|
||||
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user