5eea87882a
* refactor(desktop): split clarify-tool.tsx into a clarify/ folder
Pure moves, no behaviour change. Parsing, the question-card core
(shell, choice rows, question block), the delivery watchdog, and each
pending/settled card get their own file so the core can be reused.
* feat(clarify): one questions[] shape with per-question status and one outcome
The clarify tool now takes only questions=[{question, choices?, multi_select?}].
A wrong shape is a tool error that names the right one, so a model that
sends the old top-level question/choices corrects itself on the next call.
Every result has the same shape on every surface: each response carries
status (answered, skipped, unanswered) with user_response null unless
answered, and the result carries one outcome (submitted, cancelled,
timed_out, undelivered) plus an optional surface-written notice. The
timeout and cancel sentences that used to pose as the user's answer are
gone, and the compressor reads status instead of matching their prefixes.
The callback contract is callback(questions) -> {answers, outcome, notice?}
(None = skipped, missing qid = unanswered). tui_gateway settles a batch the
same way for the last lock, a cancel, an interrupt and the deadline, and
clarify.lock keeps a null answer as a skip. A multi-select answer that is
not a JSON array counts as one typed ("Other") answer. The tool-row preview
reads the first question.
* feat(clarify): every surface asks through the one question card
CLI: the batch panel is the only panel; Enter on an empty field skips a
question, Ctrl+C cancels and keeps the locked answers, the deadline returns
timed_out. -q and -z return undelivered with a no-user notice.
Ink TUI: the single-question prompt is gone; the card supports multi-select
(Space or a digit toggles, Enter locks, typed Other text joins the picks),
an empty submit skips, Esc cancels, and "Batch" names are dropped.
Desktop: the single-question card is deleted and its keyboard handling moved
into the one card (arrows, letters and digits, auto-advance, Enter picks then
confirms). Confirm enables at one answer; blank questions lock null; Skip and
a composer message cancel; the settled card shows No answer for unanswered
questions. The bots room card follows the same contract.
Messaging: one card per question with a "Reply skip to skip" line in every
locale; timeouts and delivery failures map to timed_out and undelivered.
* fix(clarify): cancelled for a released messaging card, labels win over the skip word
A prose reply to a messaging choice card, /new and session end released
the wait with "", which read as timed_out. They now resolve with a
CANCELLED marker and the tool gets outcome cancelled.
A typed reply that matches a choice label ("Skip") now resolves to that
choice; the skip word applies only when no choice matches.
The bots room card sends the picked labels as they are; the tool already
strips the recommendation label. Unused OUTCOMES and a new docstring and
comment are removed.
* fix(tui): re-editing a multi-select answer keeps its picks
The Ink card stored a multi-select answer as display text ("A, B"), so
revisiting the question put the whole string into Other and re-locked it
as one item. The answers map now keeps the raw JSON array, the card
restores the picks on revisit, and only the display lines format it.
Comments that earlier edits reworded are restored to their original
words, minus the phrases the change made wrong. The compute-host clarify
lock accepts None for a skipped question.
* test(clarify): align three checks with the one-answer confirm and the regenerated keys
The desktop Cmd+Enter test now expects a send with one answer (the blank
question locks null), the compressor test expects the single-answer summary
the code gives, and locales/_keys.desktop.json is regenerated for the
removed and added clarify strings.
* test(tui): the one-question card header is singular
868 lines
47 KiB
Python
868 lines
47 KiB
Python
"""Agent construction + session-resume display for ``HermesCLI``: credential resolution,
|
|
per-turn agent config, first-use build, resume preload + recap. ``cli.py`` helpers are
|
|
imported lazily inside each method (import cycle)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
|
|
from rich.markup import escape as _escape
|
|
|
|
from agent.i18n import t
|
|
from utils import base_url_host_matches
|
|
|
|
|
|
def _single_query_clarify_callback(questions: list) -> dict:
|
|
"""Headless clarify answer for ``hermes chat -q``.
|
|
|
|
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.
|
|
|
|
The oneshot path answers immediately via ``_oneshot_clarify_callback``; single-query turns need the same
|
|
headless behavior (#94943).
|
|
"""
|
|
return {"answers": {}, "outcome": "undelivered", "notice": (
|
|
"single-query mode: no user available to answer. Pick the best choices using your own "
|
|
"judgment, or 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 (
|
|
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 _cooldown_cause(entry) -> str:
|
|
"""Why a benched (exhausted) row is cooling down, from what the pool recorded: a rate-limit or
|
|
quota response, a failed token refresh, or another HTTP failure."""
|
|
reason = (entry.last_error_reason or "").lower()
|
|
if entry.last_error_code in (402, 429) or any(k in reason for k in ("rate", "quota", "insufficient")):
|
|
return t("cli.startup.cooldown_cause_rate_limit")
|
|
if entry.last_error_code is None or "refresh" in reason:
|
|
return t("cli.startup.cooldown_cause_token_refresh")
|
|
return t("cli.startup.cooldown_cause_http", code=entry.last_error_code)
|
|
|
|
|
|
def _credential_pool_notice(provider: str) -> tuple:
|
|
"""``(cooling, lines)`` on why *provider*'s pool has nothing selectable right now, for the
|
|
startup notice. *cooling* is True when the first line is a live cooldown with its remaining
|
|
time; a dead (quarantined) sign-in adds a line naming the re-login."""
|
|
import time
|
|
from agent.credential_pool import STATUS_DEAD, STATUS_EXHAUSTED, load_pool
|
|
try:
|
|
pool = load_pool(provider)
|
|
if not pool.has_credentials() or pool.has_available():
|
|
return False, []
|
|
next_at = pool.next_available_at()
|
|
entries = pool.entries()
|
|
except Exception:
|
|
return False, []
|
|
lines = []
|
|
if next_at is not None:
|
|
minutes = max(1, int((next_at - time.time() + 59) // 60))
|
|
benched = [e for e in entries if e.last_status == STATUS_EXHAUSTED]
|
|
cause = _cooldown_cause(benched[0]) if benched else t("cli.startup.cooldown_cause_failed_request")
|
|
lines.append(t("cli.startup.credential_cooling_down", provider=provider, cause=cause, minutes=minutes))
|
|
dead = [e for e in entries if e.last_status == STATUS_DEAD]
|
|
if dead:
|
|
reason = dead[0].last_error_message or dead[0].last_error_reason or t("cli.startup.sign_in_lost")
|
|
lines.append(t("cli.startup.sign_in_was_lost", provider=provider, reason=reason))
|
|
return next_at is not None, lines
|
|
|
|
|
|
def _resume_counts(user_count: int, total: int) -> str:
|
|
"""``(N user message(s), M total messages)`` suffix of the resumed-session line."""
|
|
key = "cli.resume.counts_one" if user_count == 1 else "cli.resume.counts_other"
|
|
return t(key, count=user_count, total=total)
|
|
|
|
|
|
def _keyless_custom_base(base_url) -> bool:
|
|
"""Custom/local endpoints (llama.cpp, ollama, vLLM) often need no auth; only a
|
|
non-OpenRouter base_url qualifies."""
|
|
return bool(
|
|
isinstance(base_url, str)
|
|
and base_url
|
|
and not base_url_host_matches(base_url, "openrouter.ai"))
|
|
|
|
|
|
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
|
|
|
|
|
|
def _user_display_text(content) -> str:
|
|
"""Recap text for a user row; multimodal lists become text parts + ``[image]`` markers."""
|
|
if isinstance(content, list):
|
|
return " ".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"))
|
|
return "" if content is None else str(content)
|
|
|
|
|
|
def _tool_calls_summary(tool_calls) -> str:
|
|
"""``[N tool call(s): name, ...]`` with up to 4 distinct 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 else "")
|
|
key = "cli.resume.tool_calls_one" if len(tool_calls) == 1 else "cli.resume.tool_calls_other"
|
|
return t(key, count=len(tool_calls), names=names_str)
|
|
|
|
|
|
# display_kind -> catalog key of the recap event line (resolved at call time); ``hidden`` rows
|
|
# are skipped before this lookup.
|
|
_RESUME_EVENT_KEYS = {
|
|
"model_switch": "cli.resume.event_model_changed",
|
|
"async_delegation_complete": "cli.resume.event_delegation_completed",
|
|
"process_complete": "cli.resume.event_process_finished",
|
|
"auto_continue": "cli.resume.event_resumed_interrupted"}
|
|
|
|
def _collect_resume_entries(display_history, disp: dict, clean_assistant):
|
|
"""Displayable ``(role, text)`` recap entries from stored history, truncated per the
|
|
``display.resume_*`` config; system and tool-result rows are skipped. Returns
|
|
``(entries, index of last assistant entry, its un-truncated text)``.
|
|
|
|
Stored history is untrusted for display: text is sanitized so replay can't clear the
|
|
screen, retitle the window or restyle the panel. Pure-reasoning assistant rows with no
|
|
visible output are skipped, as are tool-call-only rows when ``resume_skip_tool_only``.
|
|
"""
|
|
from tools.ansi_strip import sanitize_display_text as _sanitize_display_text
|
|
max_user_len = int(disp.get("resume_max_user_chars", 300))
|
|
max_asst_len = int(disp.get("resume_max_assistant_chars", 200))
|
|
max_asst_lines = int(disp.get("resume_max_assistant_lines", 3))
|
|
skip_tool_only = disp.get("resume_skip_tool_only", True)
|
|
entries: list = []
|
|
last_asst_idx = None
|
|
last_asst_full = None
|
|
for msg in display_history:
|
|
role = msg.get("role", "")
|
|
display_kind = msg.get("display_kind")
|
|
content = msg.get("content")
|
|
tool_calls = msg.get("tool_calls") or []
|
|
if display_kind == "hidden":
|
|
continue
|
|
if display_kind in _RESUME_EVENT_KEYS:
|
|
metadata = msg.get("display_metadata") or {}
|
|
label = metadata.get("display_text") if display_kind in ("async_delegation_complete", "process_complete") else None
|
|
entries.append(("event", _sanitize_display_text(label or t(_RESUME_EVENT_KEYS[display_kind]))))
|
|
continue
|
|
if role == "user":
|
|
text = _sanitize_display_text(_user_display_text(content))
|
|
if len(text) > max_user_len:
|
|
text = text[:max_user_len] + "..."
|
|
entries.append(("user", text))
|
|
elif role == "assistant":
|
|
text = clean_assistant("" if content is None else str(content))
|
|
parts, full_parts = [], []
|
|
if text:
|
|
full_parts.append(text)
|
|
lines = text.splitlines()
|
|
if len(lines) > max_asst_lines:
|
|
text = "\n".join(lines[:max_asst_lines]) + " ..."
|
|
if len(text) > max_asst_len:
|
|
text = text[:max_asst_len] + "..."
|
|
parts.append(text)
|
|
if tool_calls:
|
|
parts.append(_tool_calls_summary(tool_calls))
|
|
full_parts.append(parts[-1])
|
|
if not text and (skip_tool_only or not tool_calls):
|
|
continue
|
|
entries.append(("assistant", " ".join(parts)))
|
|
last_asst_idx = len(entries) - 1
|
|
last_asst_full = " ".join(full_parts)
|
|
return entries, last_asst_idx, last_asst_full
|
|
|
|
|
|
# (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"))
|
|
|
|
|
|
def _resume_panel_colors() -> tuple:
|
|
"""Active-skin colors for ``_RESUME_SKIN_COLORS`` (fallbacks when no skin loads)."""
|
|
try:
|
|
from hermes_cli.skin_engine import get_active_skin
|
|
_skin = get_active_skin()
|
|
return tuple(_skin.get_color(key, default) for key, default in _RESUME_SKIN_COLORS)
|
|
except Exception:
|
|
return tuple(default for _, default in _RESUME_SKIN_COLORS)
|
|
|
|
|
|
def _retire_agent(cli) -> None:
|
|
"""Drop ``cli.agent`` so the next turn rebuilds it, releasing its LLM clients first: the Codex
|
|
app-server child (and MCP descendants) belongs to the instance, so ``self.agent = None`` alone
|
|
orphans it for the CLI process lifetime (#72548). Session tool state is kept (soft release)."""
|
|
agent = cli.agent
|
|
if agent is not None and hasattr(agent, "release_clients"):
|
|
agent.release_clients()
|
|
cli.agent = None
|
|
|
|
|
|
class CLIAgentSetupMixin:
|
|
"""Agent construction + session-resume display methods for ``HermesCLI``."""
|
|
|
|
def _ensure_runtime_credentials(self) -> bool:
|
|
"""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, logger
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider, format_runtime_provider_error
|
|
_primary_exc = None
|
|
runtime = None
|
|
_model_at_entry = self.model
|
|
self._credentials_rate_limited = False
|
|
self._credentials_terminal = False
|
|
try:
|
|
# target_model: the ladder's model-keyed rungs (Zen/Go api_mode, Copilot/Nous
|
|
# api_mode) must see the model this CLI will actually send, not config's `default`,
|
|
# or `hermes -m mimo-v2.5 --provider opencode-go` resolves an api_mode/base_url the
|
|
# sent model cannot use (#112600).
|
|
runtime = resolve_runtime_provider(
|
|
requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
|
|
explicit_base_url=self._explicit_base_url, target_model=self.model or None)
|
|
except Exception as exc:
|
|
_primary_exc = exc
|
|
if _primary_exc is not None:
|
|
runtime = self._resolve_fallback_runtime(_primary_exc)
|
|
if runtime is not None:
|
|
_primary_exc = None
|
|
if runtime is None:
|
|
from hermes_cli.auth import AuthError, is_rate_limited_auth_error
|
|
self._credentials_rate_limited = bool(_primary_exc) and is_rate_limited_auth_error(_primary_exc)
|
|
# Only an explicit re-authentication requirement is terminal. Unknown
|
|
# resolver/network failures must not acquire the sticky Kanban block.
|
|
self._credentials_terminal = isinstance(_primary_exc, AuthError) and _primary_exc.relogin_required
|
|
message = format_runtime_provider_error(_primary_exc) if _primary_exc else t("cli.startup.provider_resolution_failed")
|
|
if getattr(self, "tool_progress_mode", "full") == "off":
|
|
print(message, file=sys.stderr) # quiet/stream-json: stdout is machine-readable
|
|
else:
|
|
ChatConsole().print(f"[bold red]{message}[/]")
|
|
return False
|
|
api_key = runtime.get("api_key")
|
|
base_url = runtime.get("base_url")
|
|
resolved_provider = runtime.get("provider", "openrouter")
|
|
if resolved_provider != "nous":
|
|
# An explicit provider carries inference. The free-tier identity (for connectors) was
|
|
# created by the boot bootstrap before this point, never here; this prints the one-time
|
|
# "free tier is here" notice the first time an identity is seen beside an own key.
|
|
self._maybe_print_free_tier_available_notice()
|
|
resolved_routing = (
|
|
resolved_provider, runtime.get("api_mode", self.api_mode), runtime.get("command"),
|
|
list(runtime.get("args") or []))
|
|
# 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.
|
|
if not callable(api_key) and not (isinstance(api_key, str) and api_key):
|
|
if _keyless_custom_base(base_url):
|
|
# Placeholder key so the SDK doesn't reject the keyless local endpoint.
|
|
api_key = "no-key-required"
|
|
logger.debug(
|
|
"No API key for custom endpoint %s (source=%s), "
|
|
"using placeholder — local servers typically ignore auth",
|
|
base_url, runtime.get("source", ""))
|
|
else:
|
|
_prov = (resolved_provider or self.requested_provider or "").strip()
|
|
if _prov and _prov != "auto":
|
|
print(f"\n{t('cli.startup.no_api_key_for_provider', provider=_prov)}")
|
|
else:
|
|
print(f"\n{t('cli.startup.no_provider_configured')}")
|
|
print(f" {t('cli.startup.run_model_or_setup')}")
|
|
return False
|
|
if not isinstance(base_url, str) or not base_url:
|
|
print(f"\n{t('cli.startup.empty_base_url')}")
|
|
return False
|
|
credentials_changed = api_key != self.api_key or base_url != self.base_url
|
|
routing_changed = resolved_routing != (self.provider, self.api_mode, self.acp_command, self.acp_args)
|
|
self.provider, self.api_mode, self.acp_command, self.acp_args = resolved_routing
|
|
self._credential_pool = runtime.get("credential_pool")
|
|
self._provider_source = runtime.get("source")
|
|
self.api_key = api_key
|
|
self.base_url = base_url
|
|
|
|
# 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) and (
|
|
not self.model or self.model == self.provider or self.model == runtime.get("name")):
|
|
self.model = runtime_model
|
|
|
|
# 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
|
|
_default = get_default_model_for_provider(resolved_provider)
|
|
if _default:
|
|
self.model = _default
|
|
logger.info(
|
|
"No model configured — defaulting to %s for provider %s",
|
|
_default, resolved_provider)
|
|
except Exception:
|
|
pass
|
|
|
|
# Normalize model for the resolved provider (e.g. swap non-Codex models on openai-codex).
|
|
# Fixes #651.
|
|
model_changed = self._normalize_model_for_provider(resolved_provider)
|
|
|
|
# Startup resolved reasoning_config for the launch model; whichever path above moved
|
|
# self.model (auth fallback, custom-entry model, provider default, normalization) leaves a
|
|
# per-model contract the lazily built agent would otherwise miss (an always-thinking model
|
|
# 400s on the primary's effort). Same chokepoint as /model, /new and --resume; an explicit
|
|
# --reasoning is the user's intent for this run and outranks the new model's config.
|
|
if self.model != _model_at_entry and getattr(self, "_explicit_reasoning_config", None) is None:
|
|
from hermes_cli.cli_model_switch_mixin import _resolve_cli_reasoning
|
|
_resolve_cli_reasoning(self)
|
|
logger.info("Model moved to %s: reasoning_config resolved: %s", self.model, self.reasoning_config)
|
|
|
|
# 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:
|
|
_retire_agent(self)
|
|
self._active_agent_route_signature = None
|
|
return True
|
|
|
|
def _maybe_print_free_tier_available_notice(self) -> None:
|
|
"""One-time notice for installs whose inference is carried by an explicit provider: the free
|
|
tier (inference + connectors) now exists. Printed the first time an identity is present, then
|
|
flagged on that identity so it never repeats. Never blocks or raises."""
|
|
from cli import logger
|
|
try:
|
|
from hermes_cli import anon_auth
|
|
if not anon_auth.guest_notice_pending():
|
|
return
|
|
self._console_print(f"[dim]{anon_auth.FREE_TIER_AVAILABLE_NOTICE}[/]")
|
|
anon_auth.mark_guest_notice_shown()
|
|
except Exception as exc:
|
|
logger.debug("free tier availability notice skipped: %s", exc)
|
|
|
|
def _resolve_fallback_runtime(self, primary_exc):
|
|
"""Primary provider resolution failed: on an AuthError try each fallback entry in
|
|
order and switch the CLI's requested_provider/model to the first that resolves.
|
|
None when the error is not auth-related or no fallback resolves."""
|
|
from cli import _cprint, logger
|
|
from hermes_cli.auth import AuthError, primary_failure_wording
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
if not isinstance(primary_exc, AuthError):
|
|
return None
|
|
_fb_chain = self._fallback_model if isinstance(self._fallback_model, list) else []
|
|
for _fb in _fb_chain:
|
|
_fb_provider = (_fb.get("provider") or "").strip().lower()
|
|
_fb_model = (_fb.get("model") or "").strip()
|
|
if not _fb_provider or not _fb_model:
|
|
continue
|
|
try:
|
|
from hermes_cli.fallback_config import resolve_entry_api_key
|
|
# target_model: the fallback entry names the model that will be sent; without it the
|
|
# ladder keys off config `default` (see _ensure_runtime_credentials, #112600).
|
|
_fb_kwargs = {"requested": _fb_provider, "target_model": _fb_model}
|
|
if _fb.get("base_url"):
|
|
_fb_kwargs["explicit_base_url"] = _fb["base_url"]
|
|
_fb_api_key = resolve_entry_api_key(_fb)
|
|
if _fb_api_key:
|
|
_fb_kwargs["explicit_api_key"] = _fb_api_key
|
|
runtime = resolve_runtime_provider(**_fb_kwargs)
|
|
_why_log, _why = primary_failure_wording(primary_exc) # #117482: quota is not auth
|
|
logger.warning(
|
|
"Primary provider %s (%s). Falling through to fallback: %s/%s",
|
|
_why_log, primary_exc, _fb_provider, _fb_model)
|
|
from gateway.warning_notifications import render_notification
|
|
render_notification(
|
|
lambda: _cprint(t("cli.startup.switching_to_fallback", reason=_why, provider=_fb_provider, model=_fb_model)),
|
|
platform="cli")
|
|
self.requested_provider = _fb_provider
|
|
self.model = _fb_model
|
|
# reasoning_config follows the swap in _ensure_runtime_credentials (the only caller).
|
|
return runtime
|
|
except Exception:
|
|
continue
|
|
return None
|
|
|
|
def _runtime_credentials_ready(self) -> bool:
|
|
"""Silently probe whether any inference provider can be resolved.
|
|
|
|
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.
|
|
|
|
See #62935.
|
|
"""
|
|
return self._probe_runtime_credentials()[0]
|
|
|
|
def _probe_runtime_credentials(self) -> tuple:
|
|
"""``(ready, error)``: *error* is the exception that stopped resolution — raised, or
|
|
swallowed by the "auto" ladder and stamped on a keyless fallback — ``None`` when a provider
|
|
resolved (usable or merely keyless). Never prints or mutates CLI state."""
|
|
from hermes_cli.runtime_provider import resolve_runtime_provider
|
|
try:
|
|
runtime = resolve_runtime_provider(
|
|
requested=self.requested_provider, explicit_api_key=self._explicit_api_key,
|
|
explicit_base_url=self._explicit_base_url)
|
|
except Exception as exc:
|
|
return False, exc
|
|
if not isinstance(runtime, dict):
|
|
return False, None
|
|
api_key = runtime.get("api_key")
|
|
base_url = runtime.get("base_url")
|
|
if callable(api_key) or (isinstance(api_key, str) and api_key):
|
|
return bool(base_url), None
|
|
return _keyless_custom_base(base_url), runtime.get("auth_error")
|
|
|
|
def _maybe_offer_first_run_setup(self) -> None:
|
|
"""Interactive startup gate: a blank install goes to the provider wizard; a configured
|
|
profile whose credential is benched or signed out gets the reason instead (#113720)."""
|
|
if not sys.stdin.isatty():
|
|
return
|
|
ready, error = self._probe_runtime_credentials()
|
|
if not ready and not self._explain_unusable_credentials(error):
|
|
self._offer_first_run_setup()
|
|
|
|
def _explain_unusable_credentials(self, error) -> bool:
|
|
"""A configured profile whose credential is benched, quarantined or signed out is not a
|
|
blank install: print what is wrong (and the remaining cooldown) instead of the first-run
|
|
wizard, whose "nothing is configured" claim sends operators into a second login that can
|
|
rotate a single-use OAuth grant away from the session that was working (#113720).
|
|
|
|
True when the failure was explained; False when nothing is configured (the wizard's case).
|
|
"""
|
|
from cli import _cprint
|
|
from hermes_cli.auth import format_auth_error
|
|
if error is None or getattr(error, "code", None) == "no_provider_configured":
|
|
return False
|
|
provider = getattr(error, "provider", None) or self.requested_provider
|
|
cooling, lines = _credential_pool_notice(provider) if provider and provider != "auto" else (False, [])
|
|
_cprint("")
|
|
if cooling:
|
|
# A live cooldown is a wait, not a lost login: lead with it and skip the re-auth hint.
|
|
_cprint(f"⚠️ {_escape(lines.pop(0))}")
|
|
_cprint(f" {_escape(str(error))}")
|
|
else:
|
|
_cprint(f"⚠️ {_escape(format_auth_error(error))}")
|
|
for line in lines:
|
|
_cprint(f" {_escape(line)}")
|
|
return True
|
|
|
|
def _offer_first_run_setup(self) -> bool:
|
|
"""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("")
|
|
_cprint(t("cli.startup.first_run_no_provider"))
|
|
_cprint(f" {t('cli.startup.first_run_pick_provider')}")
|
|
try:
|
|
answer = input(f" {t('cli.startup.first_run_prompt')} ").strip().lower()
|
|
except (KeyboardInterrupt, EOFError):
|
|
print()
|
|
answer = "n"
|
|
if answer in {"n", "no"}:
|
|
_cprint(f" {t('cli.startup.first_run_skipped')}")
|
|
return False
|
|
try:
|
|
from hermes_cli.main import select_provider_and_model
|
|
from hermes_cli.observability.shared_metrics_setup import provider_setup_surface
|
|
with provider_setup_surface("cli_setup"):
|
|
select_provider_and_model()
|
|
except (KeyboardInterrupt, EOFError, SystemExit):
|
|
print()
|
|
_cprint(f" {t('cli.startup.first_run_cancelled')}")
|
|
return False
|
|
except Exception as exc:
|
|
logger.debug("first-run provider setup failed: %s", exc)
|
|
_cprint(f" {t('cli.startup.provider_setup_failed', error=exc)}")
|
|
_cprint(f" {t('cli.startup.run_model_to_retry')}")
|
|
return False
|
|
|
|
# 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 {})
|
|
if isinstance(_model_cfg, dict):
|
|
self.requested_provider = (_model_cfg.get("provider") or "").strip() or self.requested_provider
|
|
_new_model = (_model_cfg.get("default") or _model_cfg.get("model") or "").strip()
|
|
self.model = _new_model or self.model
|
|
# The picker's model has its own per-model reasoning contract (see
|
|
# _resolve_cli_reasoning); an explicit --reasoning stays the user's intent.
|
|
if _new_model and getattr(self, "_explicit_reasoning_config", None) is None:
|
|
from hermes_cli.cli_model_switch_mixin import _resolve_cli_reasoning
|
|
_resolve_cli_reasoning(self)
|
|
except Exception as exc:
|
|
logger.debug("first-run config re-sync failed: %s", exc)
|
|
# Force credential re-resolution + agent rebuild on next use.
|
|
_retire_agent(self)
|
|
self._active_agent_route_signature = None
|
|
if self._runtime_credentials_ready():
|
|
_cprint(f" {t('cli.startup.provider_configured')}")
|
|
return True
|
|
_cprint(f" {t('cli.startup.provider_setup_incomplete')}")
|
|
return False
|
|
|
|
def _resolve_turn_agent_config(self, user_message: str) -> dict:
|
|
"""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 = _current_runtime(self)
|
|
route = {"model": self.model, "runtime": runtime, "signature": _route_signature(self.model, runtime)}
|
|
overrides = None
|
|
if getattr(self, "service_tier", None) == "priority":
|
|
try:
|
|
overrides = resolve_fast_mode_overrides(
|
|
route["model"], provider=runtime["provider"], base_url=runtime["base_url"])
|
|
except Exception:
|
|
pass
|
|
route["request_overrides"] = overrides
|
|
return route
|
|
|
|
def _follow_compression_chain(self, session_meta, announce):
|
|
"""If the resumed id is an empty compression-chain head, announce and switch to
|
|
the descendant holding the messages; returns the (possibly refreshed) meta."""
|
|
resolved_id = _compression_descendant(self._session_db, self.session_id)
|
|
if resolved_id:
|
|
announce(resolved_id)
|
|
self.session_id = resolved_id
|
|
session_meta = self._session_db.get_session(self.session_id) or session_meta
|
|
return session_meta
|
|
|
|
def _restore_session_state(self, session_meta, *, quiet: bool = False) -> None:
|
|
"""Restore cwd / yolo / model from the resumed session's metadata."""
|
|
self._restore_session_cwd(session_meta, quiet=quiet)
|
|
self._restore_session_yolo(session_meta, quiet=quiet)
|
|
self._restore_session_model(session_meta, quiet=quiet)
|
|
|
|
def _reopen_session(self) -> None:
|
|
"""Clear ended_at so the resumed session is active again (best effort)."""
|
|
try:
|
|
self._session_db.reopen_session(self.session_id)
|
|
except Exception:
|
|
pass
|
|
|
|
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 ...)`.
|
|
# Without this, the resume banner pollutes captured stdout. See #11793.
|
|
_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:
|
|
hint = t("cli.resume.session_id_hint")
|
|
not_found = t("cli.resume.session_not_found", session_id=self.session_id)
|
|
if _quiet_mode:
|
|
print(not_found, file=sys.stderr)
|
|
print(hint, file=sys.stderr)
|
|
else:
|
|
_cprint(f"\033[1;31m{not_found}{_RST}")
|
|
_cprint(f"{_DIM}{hint}{_RST}")
|
|
return False
|
|
session_meta = self._follow_compression_chain(
|
|
session_meta,
|
|
lambda rid: ChatConsole().print(
|
|
f"[dim]{_escape(t('cli.resume.compressed_into', session_id=self.session_id, descendant=rid))}[/dim]"))
|
|
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(
|
|
t("cli.resume.cannot_resume", error=resume_limit_error),
|
|
f"[bold red]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {_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 = _resume_counts(msg_count, len(restored))
|
|
_say(
|
|
f"{t('cli.resume.resumed_session')} {self.session_id}{title_part} {counts}",
|
|
f"[bold {_accent_hex()}]{_escape(t('cli.resume.resumed_session'))}[/] [bold]{_escape(self.session_id)}[/]"
|
|
f"[bold {_accent_hex()}]{_escape(title_part)}[/] {counts}")
|
|
self._restore_session_state(session_meta, quiet=_quiet_mode)
|
|
else:
|
|
no_messages = t("cli.resume.session_no_messages", session_id=self.session_id)
|
|
_say(no_messages, f"[bold {_accent_hex()}]{_escape(no_messages)}[/]")
|
|
self._reopen_session()
|
|
return True
|
|
|
|
def _init_agent(self, *, model_override: str = None, runtime_override: dict = None, request_overrides: dict | None = None) -> bool:
|
|
"""Build the agent on first use; when resuming, restore history from SQLite.
|
|
Returns True on success."""
|
|
from cli import ChatConsole, _cprint, _prepare_deferred_agent_startup, logger
|
|
from run_agent import AIAgent
|
|
if self.agent is not None:
|
|
return True
|
|
|
|
# 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()
|
|
self._install_tool_callbacks()
|
|
self._ensure_tirith_security()
|
|
if not self._ensure_runtime_credentials():
|
|
return False
|
|
from hermes_cli.mcp_startup import ensure_mcp_discovery_before_agent_build
|
|
ensure_mcp_discovery_before_agent_build(
|
|
logger=logger, single_query=getattr(self, "_single_query_mode", False))
|
|
if self._session_db is None:
|
|
try:
|
|
from hermes_state_registry import acquire
|
|
self._session_db = acquire()
|
|
except Exception as e:
|
|
logger.warning("SQLite session store not available — session will NOT be indexed: %s", e)
|
|
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 _current_runtime(self)
|
|
effective_model = model_override or self.model
|
|
# -q never builds the prompt_toolkit app, so the clarify modal can't be
|
|
# answered — answer headless instead of polling until clarify_timeout.
|
|
single_query_mode = getattr(self, "_single_query_mode", False)
|
|
clarify_callback = (
|
|
# See #94943.
|
|
_single_query_clarify_callback
|
|
if single_query_mode
|
|
else self._clarify_callback)
|
|
connection_callback = None if single_query_mode else self._connection_callback
|
|
self.agent = AIAgent(
|
|
model=effective_model, api_key=runtime.get("api_key"),
|
|
base_url=runtime.get("base_url"), provider=runtime.get("provider"),
|
|
requested_provider=runtime.get("requested_provider"),
|
|
api_mode=runtime.get("api_mode"), acp_command=runtime.get("command"),
|
|
acp_args=runtime.get("args"), credential_pool=runtime.get("credential_pool"),
|
|
max_iterations=self.max_turns,
|
|
run_budget_seconds=getattr(self, "run_budget_seconds", None),
|
|
enabled_toolsets=self.enabled_toolsets, disabled_toolsets=self.disabled_toolsets,
|
|
verbose_logging=self.verbose, quiet_mode=not self.verbose,
|
|
tool_progress_mode=getattr(self, "tool_progress_mode", "all"),
|
|
ephemeral_system_prompt=self.system_prompt if self.system_prompt else None,
|
|
prefill_messages=self.prefill_messages or None,
|
|
reasoning_config=self.reasoning_config, service_tier=self.service_tier,
|
|
request_overrides=request_overrides, providers_allowed=self._providers_only,
|
|
providers_ignored=self._providers_ignore, providers_order=self._providers_order,
|
|
provider_sort=self._provider_sort,
|
|
provider_require_parameters=self._provider_require_params,
|
|
provider_data_collection=self._provider_data_collection,
|
|
openrouter_min_coding_score=self._openrouter_min_coding_score,
|
|
session_id=self.session_id, platform="cli", session_db=self._session_db,
|
|
clarify_callback=clarify_callback, connection_callback=connection_callback,
|
|
reasoning_callback=self._current_reasoning_callback(),
|
|
fallback_model=self._fallback_model, thinking_callback=self._on_thinking,
|
|
checkpoints_enabled=self.checkpoints_enabled,
|
|
checkpoint_max_snapshots=self.checkpoint_max_snapshots,
|
|
checkpoint_max_total_size_mb=self.checkpoint_max_total_size_mb,
|
|
checkpoint_max_file_size_mb=self.checkpoint_max_file_size_mb,
|
|
pass_session_id=self.pass_session_id, skip_context_files=self.ignore_rules,
|
|
skip_memory=self.ignore_rules, tool_progress_callback=self._on_tool_progress,
|
|
tool_start_callback=self._on_tool_start if self._inline_diffs_enabled else None,
|
|
tool_complete_callback=self._on_tool_complete if self._inline_diffs_enabled else None,
|
|
stream_delta_callback=self._stream_delta if self.streaming_enabled else None,
|
|
tool_gen_callback=self._on_tool_gen_start if self.streaming_enabled else None,
|
|
notice_callback=self._on_notice, notice_clear_callback=self._on_notice_clear,
|
|
reaction_callback=self._on_reaction)
|
|
# 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.
|
|
# 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).
|
|
import cli as _cli
|
|
_cli._active_agent_ref = self.agent
|
|
# Seed the agent's once-per-lifecycle auto_load cache with the bytes the preload
|
|
# thread rendered, so the shared prompt path never re-reads config or skill files.
|
|
_auto_result = getattr(self, "_auto_load_skills_result", None)
|
|
if _auto_result is not None:
|
|
self.agent._auto_load_skills_result = _auto_result
|
|
self.agent._auto_load_skills_resolved = True
|
|
# Route agent status output through prompt_toolkit so ANSI escapes aren't garbled by
|
|
# patch_stdout's StdoutProxy (#2262), holding lines while a response box streams so a
|
|
# subagent/background completion notice never splits the reply mid-paragraph.
|
|
self.agent._print_fn = self._agent_status_print
|
|
# 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 = _route_signature(effective_model, runtime)
|
|
|
|
# Force-create DB row on /title intent, then apply title.
|
|
if self._pending_title and self._session_db:
|
|
try:
|
|
self.agent._ensure_db_session()
|
|
if self.agent._session_db_created:
|
|
self._session_db.set_session_title(self.session_id, self._pending_title)
|
|
_cprint(f" {t('cli.resume.title_applied', title=self._pending_title)}")
|
|
self._pending_title = None
|
|
# else: row creation failed transiently — keep _pending_title for retry
|
|
except Exception as e:
|
|
_cprint(f" {t('cli.resume.title_apply_failed', error=e)}")
|
|
# Keep _pending_title so it can be retried after row creation succeeds
|
|
return True
|
|
except Exception as e:
|
|
console = ChatConsole()
|
|
from hermes_cli.cli_chat_error_copy import agent_init_failure_message
|
|
console.print(f"[bold red]{_escape(agent_init_failure_message(e))}[/]")
|
|
from hermes_constants import partial_update_hint
|
|
for line in partial_update_hint(e):
|
|
console.print(line)
|
|
return False
|
|
|
|
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 — 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 cli import logger
|
|
from hermes_state import SessionResumeTooLargeError
|
|
try:
|
|
safety_check = getattr(self._session_db, "assert_resume_safe", None)
|
|
if not callable(safety_check):
|
|
return None
|
|
safety_check(self.session_id, **({"tip_only": True} if tip_only else {}))
|
|
except SessionResumeTooLargeError as exc:
|
|
return str(exc)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Resume safety check failed for %s (proceeding without guard): %s",
|
|
self.session_id, exc)
|
|
return None
|
|
|
|
def _preload_resumed_session(self) -> bool:
|
|
"""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]{_escape(t('cli.resume.session_not_found', session_id=self.session_id))}[/]")
|
|
self._console_print(f"[dim]{_escape(t('cli.resume.session_id_hint'))}[/]")
|
|
return False
|
|
session_meta = self._follow_compression_chain(
|
|
session_meta,
|
|
lambda rid: self._console_print(
|
|
f"[dim]{_escape(t('cli.resume.compressed_into', session_id=self.session_id, descendant=rid))}[/]"))
|
|
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]{_escape(t('cli.resume.cannot_resume_prefix'))}[/] {resume_limit_error}")
|
|
return False
|
|
restored, display_history = self._session_db.get_resume_conversations(self.session_id)
|
|
accent_color = _accent_hex()
|
|
if not restored:
|
|
self._console_print(
|
|
f"[{accent_color}]{_escape(t('cli.resume.session_no_messages', session_id=self.session_id))}[/]")
|
|
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}]{_escape(t('cli.resume.resumed_session'))} [bold]{self.session_id}[/bold]"
|
|
f"{title_part} {_resume_counts(msg_count, len(restored))}[/]")
|
|
self._restore_session_state(session_meta)
|
|
self._reopen_session()
|
|
return True
|
|
|
|
def _display_resumed_history(self):
|
|
"""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 or self.resume_display == "minimal":
|
|
return
|
|
_disp = CLI_CONFIG.get("display", {})
|
|
entries, _last_asst_idx, _last_asst_full = _collect_resume_entries(
|
|
display_history, _disp, lambda t: _sanitize_display_text(_strip_reasoning_tags(t)))
|
|
if not entries:
|
|
return
|
|
skipped = max(0, len(entries) - int(_disp.get("resume_exchanges", 10)) * 2)
|
|
entries = entries[skipped:]
|
|
# 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)
|
|
from rich.panel import Panel
|
|
from rich.text import Text
|
|
_history_text_c, _session_label_c, _session_border_c, _assistant_label_c = (
|
|
_resume_panel_colors())
|
|
|
|
# role -> (label, label style, body style, continuation indent)
|
|
you_label = f" {t('cli.resume.label_you')} "
|
|
assistant_label = f" {t('cli.resume.label_assistant', agent_name='Hermes')} "
|
|
role_styles = {
|
|
"user": (you_label, f"dim bold {_session_label_c}", "dim", " " * len(you_label)),
|
|
"assistant": (assistant_label, f"dim bold {_assistant_label_c}", "dim", " " * len(assistant_label)),
|
|
"assistant_last": (assistant_label, f"bold {_assistant_label_c}", "", " " * len(assistant_label)), # full, non-dim
|
|
}
|
|
lines = Text()
|
|
if skipped:
|
|
lines.append(f" {t('cli.resume.earlier_messages', count=skipped)}\n\n", style="dim italic")
|
|
for i, (role, text) in enumerate(entries):
|
|
if role == "event":
|
|
lines.append(f" ◈ {text}\n", style="dim italic")
|
|
else:
|
|
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
|
|
panel = Panel(
|
|
lines, title=f"[dim {_session_label_c}]{_escape(t('cli.resume.panel_title'))}[/]",
|
|
border_style=f"dim {_session_border_c}", padding=(0, 1), style=_history_text_c)
|
|
_record_output_history_entry(lambda: self._render_resume_history_panel_lines(panel))
|
|
with _suspend_output_history():
|
|
self._console_print(panel)
|