mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
unsupportedpastels
0b388277f9
The fallback marker re-derived the resolver's answer from a few local sources, so a window a custom endpoint reported through /models (or a local probe, provider profile, live catalog) read as "fallback" whenever it equalled the default. Its 60s memo was also keyed without the profile home, so one multiplexed profile's verdict labelled another's window. get_model_context_length now records when it ended on a static default (the generic 256K or Bedrock's 128K) per (profile home, model, base_url), and the marker reads that record. The memo is gone. Gateway /model drops the configured pin for a changed route before both the number and its label, through the same helper resolve_display_context_length uses, so a discarded 128K pin no longer shows "(pinned)" over a fallback. /status and /context now carry the same label as /usage for resident agents and non-resident sessions. Addresses Greptile review on #148801.
709 lines
38 KiB
Python
709 lines
38 KiB
Python
"""Read-only gateway introspection commands: /status, /context, /usage, /agents, /insights, /topup.
|
|
Bound onto ``GatewayRunner`` through ``GatewaySlashCommandsMixin``."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import asyncio
|
|
import hashlib
|
|
import os
|
|
import re
|
|
import time
|
|
from typing import Any
|
|
|
|
from agent.account_usage import fetch_account_usage, render_account_usage_lines
|
|
from agent.i18n import t
|
|
from gateway.config import Platform
|
|
from gateway.platforms.event import MessageEvent
|
|
from gateway.session_transcript import TranscriptReadError
|
|
from hermes_cli.status_report import build_status_fields
|
|
|
|
# Log-record parity with gateway/run.py and the origin module.
|
|
logger = logging.getLogger("gateway.run")
|
|
|
|
_LIST_CAP = 12 # /agents shows at most this many rows per section
|
|
|
|
|
|
def _clean_str(value: Any) -> str:
|
|
"""Strip and return a non-empty string value, or empty string."""
|
|
return value.strip() if isinstance(value, str) and value.strip() else ""
|
|
|
|
|
|
def _int_value(value: Any) -> int:
|
|
"""Safely coerce to int."""
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _n(obj, attr: str):
|
|
return getattr(obj, attr, 0) or 0
|
|
|
|
|
|
def _fmt(n) -> str:
|
|
return f"{n:,}"
|
|
|
|
|
|
def _pct(used, total) -> float: # clamped occupancy percentage; 0 for an unknown window
|
|
return min(100, used / total * 100) if total else 0
|
|
|
|
|
|
def _clip(text: str, limit: int) -> str:
|
|
return text[: limit - 3] + "..." if len(text) > limit else text
|
|
|
|
|
|
def _transcript_estimate(history) -> tuple[int, int]:
|
|
"""``(approx_tokens, message_count)`` over the user/assistant messages of a transcript."""
|
|
from agent.model_metadata import estimate_messages_tokens_rough
|
|
msgs = [m for m in history if m.get("role") in {"user", "assistant"} and m.get("content")]
|
|
return estimate_messages_tokens_rough(msgs), len(msgs)
|
|
|
|
|
|
async def _quiet(call, default=None):
|
|
"""Await ``call()`` fail-open: any exception (sync or in the awaitable) yields *default*."""
|
|
try:
|
|
return await call()
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
# English source for the transcript-read-failure reply; callers render ``history_unreadable()``
|
|
# (``gateway.shared.history_unreadable``) so the active language applies.
|
|
HISTORY_UNREADABLE = ("⚠️ I can't read this conversation's history right now (your earlier messages "
|
|
"exist but cannot be loaded). Run `hermes doctor --fix` on the host, or use /new "
|
|
"to start fresh.")
|
|
|
|
|
|
def history_unreadable() -> str:
|
|
return t("gateway.shared.history_unreadable")
|
|
|
|
|
|
def _configured_provider() -> str:
|
|
"""``model.provider`` from the gateway config ("" when unset)."""
|
|
from gateway.run import _load_gateway_config
|
|
user_config = _load_gateway_config()
|
|
model_cfg = user_config.get("model", {}) if isinstance(user_config, dict) else {}
|
|
return _clean_str(model_cfg.get("provider")) if isinstance(model_cfg, dict) else ""
|
|
|
|
|
|
def _quiet_sync(call, default=None):
|
|
"""Sync twin of ``_quiet``."""
|
|
try:
|
|
return call()
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def _status_model_route(
|
|
status_agent, active_override: dict, persisted_route: dict, session_row: dict, session_entry
|
|
):
|
|
"""``(model, provider, context_used, context_total, route)`` for /status.
|
|
|
|
Order: live/cached agent route -> active session override -> persisted recent route ->
|
|
SessionDB row -> gateway config (only loaded when something is still missing). ``route`` carries
|
|
the ``provider`` / ``base_url`` / ``api_key`` of the winning source only, so a later context-window
|
|
lookup queries the endpoint that serves the displayed model (never a losing route's endpoint);
|
|
a winner without a ``base_url`` leaves the lookup on the default runtime route.
|
|
"""
|
|
from gateway.run import _AGENT_PENDING_SENTINEL, _load_gateway_config, _resolve_gateway_model
|
|
context_used = context_total = 0
|
|
routes: list[tuple[str, str, dict]] = []
|
|
if status_agent is not None and status_agent is not _AGENT_PENDING_SENTINEL:
|
|
routes.append((_clean_str(getattr(status_agent, "model", "")),
|
|
_clean_str(getattr(status_agent, "provider", "")),
|
|
{"base_url": _clean_str(getattr(status_agent, "base_url", "")),
|
|
"api_key": _clean_str(getattr(status_agent, "api_key", ""))}))
|
|
ctx = getattr(status_agent, "context_compressor", None)
|
|
if ctx is not None:
|
|
context_used = max(0, _int_value(getattr(ctx, "last_prompt_tokens", 0)))
|
|
context_total = _int_value(getattr(ctx, "context_length", 0))
|
|
routes.append((_clean_str(active_override.get("model")),
|
|
_clean_str(active_override.get("provider")),
|
|
{"base_url": _clean_str(active_override.get("base_url")),
|
|
"api_key": _clean_str(active_override.get("api_key"))}))
|
|
routes.append((_clean_str(persisted_route.get("model")),
|
|
_clean_str(persisted_route.get("billing_provider")), {}))
|
|
row_route = (_clean_str(session_row.get("model")), _clean_str(session_row.get("billing_provider")), {})
|
|
# First fully-resolved (model AND provider) route wins; the SessionDB row is used even if partial.
|
|
model_name, provider_name, route = next((r for r in routes if r[0] and r[1]), row_route)
|
|
context_used = context_used or _int_value(getattr(session_entry, "last_prompt_tokens", 0))
|
|
user_config: dict[str, Any] = {}
|
|
if not model_name or not provider_name:
|
|
user_config = _quiet_sync(_load_gateway_config, {})
|
|
model_cfg = user_config.get("model", {}) if isinstance(user_config, dict) else {}
|
|
model_cfg = model_cfg if isinstance(model_cfg, dict) else {}
|
|
model_name = model_name or _resolve_gateway_model(user_config)
|
|
provider_name = provider_name or _clean_str(model_cfg.get("provider"))
|
|
# No raw ``model.context_length`` pin here: the resolver applies it only while the displayed
|
|
# route still matches the configured one (a session /model switch must not inherit it).
|
|
return model_name, provider_name, context_used, context_total, {"provider": provider_name, **route}
|
|
|
|
|
|
def _context_compressor_lines(agent, ctx, used: int) -> list[str]:
|
|
"""/context full view: auto-compression threshold/headroom, compression count + last savings,
|
|
and cumulative throughput (labelled as throughput, NOT context size)."""
|
|
lines: list[str] = []
|
|
from agent.context_breakdown import context_display_source
|
|
mark = "~" if context_display_source(ctx) != "provider_usage" else ""
|
|
threshold = _n(ctx, "threshold_tokens")
|
|
threshold_pct = f"{_n(ctx, 'threshold_percent') * 100:.0f}"
|
|
if threshold > 0:
|
|
if used >= threshold:
|
|
lines.append(t("gateway.context.over_threshold", threshold=_fmt(threshold),
|
|
threshold_pct=threshold_pct))
|
|
else:
|
|
lines.append(t("gateway.context.threshold", threshold=_fmt(threshold),
|
|
threshold_pct=threshold_pct, to_go=mark + _fmt(threshold - used)))
|
|
compressions = _n(ctx, "compression_count")
|
|
lines.append(t("gateway.context.compressions", count=compressions))
|
|
savings = getattr(ctx, "_last_compression_savings_pct", None) if compressions else None
|
|
if savings is not None:
|
|
lines.append(t("gateway.context.last_savings", savings=f"{savings:.0f}"))
|
|
lines += [
|
|
"",
|
|
t("gateway.context.totals_header", calls=_n(agent, "session_api_calls")),
|
|
t("gateway.context.totals_line",
|
|
input=_fmt(_n(agent, "session_input_tokens")),
|
|
output=_fmt(_n(agent, "session_output_tokens")),
|
|
reasoning=_fmt(_n(agent, "session_reasoning_tokens"))),
|
|
t("gateway.context.total_billed", total=_fmt(_n(agent, "session_total_tokens"))),
|
|
t("gateway.context.throughput_note"),
|
|
]
|
|
return lines
|
|
|
|
|
|
def _agents_delegation_lines(d: dict) -> list[str]:
|
|
"""/agents rows for one background delegation. Live per-child activity comes from the
|
|
registry's progress sampler: api calls, current tool, seconds since last activity."""
|
|
goal = _clip(" ".join(str(d.get("goal") or "").split()), 70)
|
|
status = d.get("status", "?")
|
|
row = f"- `{d.get('delegation_id', '?')}` · {status}"
|
|
quiet = d.get("stalled_after_quiet_seconds")
|
|
if status == "stalling" and quiet is not None:
|
|
row += t("gateway.agents.no_progress", seconds=f"{quiet:.0f}")
|
|
elif status != "stalling" and d.get("seconds_since_progress", 0) >= 60:
|
|
row += t("gateway.agents.quiet_for", seconds=f"{d['seconds_since_progress']:.0f}")
|
|
if goal:
|
|
row += f" · {goal}"
|
|
lines = [row]
|
|
for i, child in enumerate(d.get("children_activity") or []):
|
|
if not isinstance(child, dict):
|
|
continue
|
|
tool = child.get("current_tool")
|
|
doing = f"`{tool}`" if tool else t("gateway.agents.between_turns")
|
|
part = t("gateway.agents.child_row", index=i + 1, calls=child.get("api_calls", "?"), doing=doing)
|
|
idle = child.get("seconds_since_activity")
|
|
lines.append(part + (t("gateway.agents.active_ago", seconds=f"{idle:.0f}") if idle is not None else ""))
|
|
return lines
|
|
|
|
|
|
def _usage_agent_stats_lines(agent) -> list[str]:
|
|
"""/usage session block for a live agent: rate limits, token breakdown (matches the CLI),
|
|
context window and compression count."""
|
|
lines: list[str] = []
|
|
rl_state = agent.get_rate_limit_state()
|
|
if rl_state and rl_state.has_data:
|
|
from agent.rate_limit_tracker import format_rate_limit_compact
|
|
lines += [t("gateway.usage.rate_limits", state=format_rate_limit_compact(rl_state)), ""]
|
|
lines += [
|
|
t("gateway.usage.header_session"),
|
|
t("gateway.usage.label_model", model=agent.model),
|
|
t("gateway.usage.label_input_tokens", count=_fmt(_n(agent, "session_input_tokens"))),
|
|
t("gateway.usage.label_output_tokens", count=_fmt(_n(agent, "session_output_tokens"))),
|
|
t("gateway.usage.label_total", count=_fmt(agent.session_total_tokens)),
|
|
t("gateway.usage.label_api_calls", count=agent.session_api_calls),
|
|
]
|
|
ctx = agent.context_compressor
|
|
if ctx.last_prompt_tokens > 0:
|
|
pct = _pct(ctx.last_prompt_tokens, ctx.context_length)
|
|
from agent.context_breakdown import context_display_source
|
|
from agent.context_pin import agent_context_suffix
|
|
mark = "~" if context_display_source(ctx) != "provider_usage" else ""
|
|
lines.append(t("gateway.usage.label_context", used=mark + _fmt(ctx.last_prompt_tokens),
|
|
total=_fmt(ctx.context_length), pct=f"{mark}{pct:.0f}") + agent_context_suffix(agent))
|
|
if ctx.compression_count:
|
|
lines.append(t("gateway.usage.label_compressions", count=ctx.compression_count))
|
|
return lines
|
|
|
|
|
|
def _capped_rows(items: list, render) -> list[str]:
|
|
"""Render up to ``_LIST_CAP`` items via *render* (list of lines each) plus an overflow line."""
|
|
lines: list[str] = []
|
|
for item in items[:_LIST_CAP]:
|
|
lines.extend(render(item))
|
|
if len(items) > _LIST_CAP:
|
|
lines.append(t("gateway.agents.more", count=len(items) - _LIST_CAP))
|
|
return lines
|
|
|
|
|
|
class GatewayStatusCommandsMixin:
|
|
"""Read-only gateway introspection commands: /status, /context, /usage, /agents, /insights, /topup."""
|
|
|
|
async def _handle_status_command(self, event: MessageEvent) -> str:
|
|
"""Handle /status command."""
|
|
from gateway.run import _AGENT_PENDING_SENTINEL
|
|
source = event.source
|
|
session_entry = await self.async_session_store.get_or_create_session(source)
|
|
session_key = session_entry.session_key
|
|
# Keep the sentinel distinct: a starting/pending run is not a usable agent for
|
|
# model/context display, but it still occupies the session slot.
|
|
agent = self._running_agents.get(session_key)
|
|
is_running = agent is not None and agent is not _AGENT_PENDING_SENTINEL
|
|
# Pending /queue follow-ups (slot + overflow).
|
|
adapter = self.adapters.get(source.platform) if source else None
|
|
queue_depth = self._queue_depth(session_key, adapter=adapter)
|
|
title, session_row, db_total_tokens, persisted_route = await self._status_session_db_facts(
|
|
session_entry.session_id
|
|
)
|
|
# Prefer the live or cached agent (actual runtime route + context compressor); fall back
|
|
# to an active /model override, then SessionDB metadata + last_prompt_tokens so /status
|
|
# stays useful between turns. Rehydrate first so this precedence survives gateway restarts.
|
|
status_agent = agent if is_running else self._cached_agent_for(session_key)
|
|
self._rehydrate_session_model_override(session_key)
|
|
active_override = self._session_model_override(session_key) or {}
|
|
model_name, provider_name, context_used, context_total, route = _status_model_route(
|
|
status_agent, active_override, persisted_route, session_row, session_entry
|
|
)
|
|
if not context_total and model_name:
|
|
# Same resolver /context uses (off-loop: it can probe /models). A window the resolver only
|
|
# invented (unknown model → DEFAULT_FALLBACK_CONTEXT) stays hidden rather than being shown
|
|
# as a real limit; the occupancy-only line below is honest for that case.
|
|
resolved = await self._resolve_route_context(source, model_name, route)
|
|
if resolved is not None and resolved.context_source != "default":
|
|
context_total = _int_value(resolved.context_length)
|
|
|
|
fields = build_status_fields(
|
|
session_entry.session_id, None, session_row, title=title, model=model_name, provider=provider_name,
|
|
created=session_entry.created_at, last_activity=session_entry.updated_at,
|
|
tokens=db_total_tokens, agent_running=is_running,
|
|
)
|
|
lines = [t("gateway.status.header"), "",
|
|
t("gateway.status.session_id", session_id=fields["session_id"])]
|
|
if fields["title"]:
|
|
lines.append(t("gateway.status.title", title=fields["title"]))
|
|
lines += [t("gateway.status.created", timestamp=fields["created"]),
|
|
t("gateway.status.last_activity", timestamp=fields["last_activity"])]
|
|
if fields["model"] and fields["provider"]:
|
|
lines.append(t("gateway.status.model_provider", model=fields["model"], provider=fields["provider"]))
|
|
elif fields["model"]:
|
|
lines.append(t("gateway.status.model", model=fields["model"]))
|
|
from agent.context_breakdown import context_display_source
|
|
mark = "~" if context_display_source(getattr(status_agent, "context_compressor", None)) != "provider_usage" else ""
|
|
if context_total:
|
|
pct = min(100, round((context_used / context_total) * 100))
|
|
from agent.context_pin import agent_context_suffix # (pinned) / (fallback) for the agent's own window
|
|
compressor_total = _n(getattr(status_agent, "context_compressor", None), "context_length")
|
|
suffix = agent_context_suffix(status_agent) if compressor_total and compressor_total == context_total else ""
|
|
lines.append(t("gateway.status.context", used=mark + _fmt(context_used), total=_fmt(context_total),
|
|
pct=f"{mark}{pct}") + suffix)
|
|
elif context_used:
|
|
lines.append(t("gateway.status.context_used", used=mark + _fmt(context_used)))
|
|
state = t("gateway.status.state_yes") if fields["agent_running"] else t("gateway.status.state_no")
|
|
lines += [t("gateway.status.tokens", tokens=fields["tokens"]),
|
|
t("gateway.status.agent_running", state=state)]
|
|
if queue_depth:
|
|
lines.append(t("gateway.status.queued", count=queue_depth))
|
|
if source.platform == Platform.MATRIX:
|
|
scope = getattr(self.adapters.get(Platform.MATRIX), "_matrix_session_scope",
|
|
os.getenv("MATRIX_SESSION_SCOPE", "auto"))
|
|
lines += [
|
|
"",
|
|
t("gateway.status.matrix_scope_header"),
|
|
t("gateway.status.matrix_scope_room", room=source.chat_name or source.chat_id),
|
|
t("gateway.status.matrix_scope_room_id", room_id=source.chat_id),
|
|
t("gateway.status.matrix_scope_thread", thread_id=source.thread_id or t("gateway.shared.none_value")),
|
|
t("gateway.status.matrix_scope_mode", scope=scope),
|
|
t("gateway.status.matrix_scope_key",
|
|
session_key=self._redact_matrix_session_key(session_key)),
|
|
]
|
|
lines += ["", t("gateway.status.platforms", platforms=', '.join(p.value for p in self.adapters))]
|
|
return "\n".join(lines)
|
|
|
|
async def _status_session_db_facts(self, session_id: str):
|
|
"""``(title, session_row, db_total_tokens, persisted_route)`` for /status; each fail-open.
|
|
|
|
Token totals come from the SQLite session DB, not SessionStore: run_agent.py persists per-turn
|
|
token deltas into sessions_db, never into SessionEntry (its total_tokens is always 0).
|
|
"""
|
|
db = self._session_db
|
|
if not db:
|
|
return None, {}, 0, {}
|
|
title = await _quiet(lambda: db.get_session_title(session_id))
|
|
row = await _quiet(lambda: db.get_session(session_id))
|
|
session_row = row if isinstance(row, dict) else {}
|
|
db_total_tokens = sum(
|
|
_int_value(session_row.get(k))
|
|
for k in ("input_tokens", "output_tokens", "cache_read_tokens", "cache_write_tokens", "reasoning_tokens")
|
|
)
|
|
route = await _quiet(lambda: db.get_recent_session_model_route(session_id))
|
|
return title, session_row, db_total_tokens, route if isinstance(route, dict) else {}
|
|
|
|
@staticmethod
|
|
def _redact_matrix_session_key(session_key: str) -> str:
|
|
"""Return a stable Matrix session-key fingerprint for shared room status."""
|
|
digest = hashlib.sha256(str(session_key or "").encode("utf-8")).hexdigest()[:12]
|
|
return f"sha256:{digest}"
|
|
|
|
async def _handle_context_command(self, event: MessageEvent) -> str:
|
|
"""Handle /context — the deep context-window view (/status has the one-line summary).
|
|
|
|
Gauge, auto-compression threshold/headroom, compression count + last savings, and cumulative
|
|
throughput (labelled as throughput, NOT context size). Resolution: running agent -> cached
|
|
agent -> SessionStore/SessionDB metadata -> transcript estimate. ``all`` adds listings.
|
|
"""
|
|
source = event.source
|
|
session_entry = await self.async_session_store.get_or_create_session(source)
|
|
expanded = event.get_command_args().strip().lower() in {"all", "full", "details"}
|
|
# Running agent first (mid-turn), then cached agent (between turns).
|
|
agent = self._resident_agent_for(self._session_key_for_source(source)) or None
|
|
ctx = getattr(agent, "context_compressor", None) if agent else None
|
|
used, context_length, model_name, window_suffix = await self._resolve_context_figures(
|
|
agent, ctx, session_entry, source
|
|
)
|
|
from agent.context_breakdown import context_display_source
|
|
mark = "~" if context_display_source(ctx) != "provider_usage" else ""
|
|
# Gauge path: preserve the provenance of the selected occupancy figure.
|
|
if used > 0 and context_length > 0:
|
|
pct = _pct(used, context_length)
|
|
filled = round(pct / 100 * 24)
|
|
lines = [
|
|
t("gateway.context.header"), "",
|
|
t("gateway.context.model", model=model_name or "?"),
|
|
t("gateway.context.window", total=_fmt(context_length)) + window_suffix,
|
|
t("gateway.context.in_use", used=mark + _fmt(used), total=_fmt(context_length), pct=f"{mark}{pct:.0f}"),
|
|
t("gateway.context.bar", bar="█" * max(0, filled) + "░" * max(0, 24 - filled)),
|
|
t("gateway.context.headroom", headroom=mark + _fmt(max(0, context_length - used))),
|
|
"",
|
|
]
|
|
# Full view — compression / throughput need the live agent.
|
|
lines += _context_compressor_lines(agent, ctx, used) if ctx is not None else [
|
|
t("gateway.context.detail_after_first")]
|
|
# Per-category estimated breakdown (+ optional expanded listings). Same chars/4 engine
|
|
# the desktop popover and /usage use; plain text (monospace isn't guaranteed on
|
|
# messaging platforms). Fail-open: rendering errors never break /context.
|
|
breakdown = await asyncio.to_thread(self._context_breakdown_block, agent, source, expanded) if agent else []
|
|
return "\n".join(lines + ([""] + breakdown if breakdown else []))
|
|
# Last resort: rough estimate from transcript
|
|
try:
|
|
history = await self.async_session_store.load_transcript(session_entry.session_id)
|
|
except TranscriptReadError:
|
|
return history_unreadable()
|
|
if not history:
|
|
return t("gateway.context.no_data")
|
|
approx, count = _transcript_estimate(history)
|
|
return "\n".join([
|
|
t("gateway.context.header"), "",
|
|
t("gateway.context.estimated", count=_fmt(approx), messages=count),
|
|
t("gateway.context.detail_after_first"),
|
|
])
|
|
|
|
async def _resolve_context_figures(self, agent, ctx, session_entry, source):
|
|
"""``(used, context_length, model_name, window_suffix)`` for /context: used = compressor -> SessionStore;
|
|
model = agent -> SessionDB row; window = compressor -> gateway model route -> model metadata. The suffix
|
|
is `` (pinned)`` / `` (fallback)`` from whichever source supplied the window."""
|
|
from agent.context_pin import agent_context_suffix, context_window_suffix
|
|
used = max(0, _n(ctx, "last_prompt_tokens")) or max(0, _int_value(getattr(session_entry, "last_prompt_tokens", 0)))
|
|
context_length = _n(ctx, "context_length")
|
|
suffix = agent_context_suffix(agent) if context_length else ""
|
|
model_name = _clean_str(getattr(agent, "model", "")) if agent is not None else ""
|
|
if not model_name and self._session_db:
|
|
row = await _quiet(lambda: self._session_db.get_session(session_entry.session_id))
|
|
model_name = _clean_str(row.get("model", "")) if isinstance(row, dict) else ""
|
|
if not context_length:
|
|
resolved = await self._resolve_route_context(source, model_name)
|
|
if resolved is not None:
|
|
model_name = model_name or resolved.model
|
|
context_length = _int_value(resolved.context_length)
|
|
suffix = {"config": " (pinned)", "default": " (fallback)"}.get(resolved.context_source, "")
|
|
if not context_length and model_name:
|
|
from agent.model_metadata import get_model_context_length
|
|
context_length = _int_value(
|
|
await _quiet(lambda: asyncio.to_thread(get_model_context_length, model_name))
|
|
)
|
|
suffix = context_window_suffix(context_length, None, model=model_name)
|
|
return used, context_length, model_name, suffix
|
|
|
|
async def _resolve_route_context(self, source, model_name: str, route: dict | None = None):
|
|
"""``_resolve_gateway_model_context`` for a non-resident session, run off the event loop inside
|
|
the profile serving ``source`` (multiplex). Fail-open: None on any error."""
|
|
from gateway.run import _profile_runtime_scope, _resolve_gateway_model_context
|
|
|
|
def _resolve():
|
|
if getattr(getattr(self, "config", None), "multiplex_profiles", False):
|
|
with _profile_runtime_scope(self._resolve_profile_home_for_source(source)):
|
|
return _resolve_gateway_model_context(model_name or None, route)
|
|
return _resolve_gateway_model_context(model_name or None, route)
|
|
return await _quiet(lambda: asyncio.to_thread(_resolve))
|
|
|
|
async def _handle_agents_command(self, event: MessageEvent) -> str:
|
|
"""Handle /agents command - list active agents and running tasks."""
|
|
from gateway.run import _AGENT_PENDING_SENTINEL
|
|
from tools.process_registry import format_uptime_short, process_registry
|
|
now = time.time()
|
|
current_session_key = self._session_key_for_source(event.source)
|
|
running_started: dict = getattr(self, "_running_agents_ts", {}) or {}
|
|
agent_rows: list[dict] = []
|
|
for session_key, agent in (getattr(self, "_running_agents", {}) or {}).items():
|
|
pending = agent is _AGENT_PENDING_SENTINEL
|
|
agent_rows.append({
|
|
"session_key": session_key,
|
|
"elapsed": max(0, int(now - float(running_started.get(session_key, now)))),
|
|
"state": t("gateway.agents.state_starting") if pending else t("gateway.agents.state_running"),
|
|
"session_id": "" if pending else str(getattr(agent, "session_id", "") or ""),
|
|
"model": "" if pending else str(getattr(agent, "model", "") or ""),
|
|
})
|
|
agent_rows.sort(key=lambda row: row["elapsed"], reverse=True)
|
|
procs = _quiet_sync(process_registry.list_sessions, [])
|
|
running_processes = [p for p in procs if p.get("status") == "running"]
|
|
background_tasks = [task for task in (getattr(self, "_background_tasks", set()) or set())
|
|
if hasattr(task, "done") and not task.done()]
|
|
|
|
# Background (async) delegations — delegate_task(background=true).
|
|
# Live per-child activity comes from the registry's progress sampler (#51690): api calls, current
|
|
# tool, seconds since last activity.
|
|
from tools.async_delegation import list_async_delegations
|
|
delegations = [d for d in _quiet_sync(list_async_delegations, [])
|
|
if d.get("status") in ("running", "stalling", "finalizing")]
|
|
|
|
def _agent_row(idx_row):
|
|
idx, row = idx_row
|
|
current = t("gateway.agents.this_chat") if row["session_key"] == current_session_key else ""
|
|
sid = f" · `{row['session_id']}`" if row["session_id"] else ""
|
|
model = f" · `{row['model']}`" if row["model"] else ""
|
|
return [f"{idx}. `{row['session_key']}` · {row['state']} · "
|
|
f"{format_uptime_short(row['elapsed'])}{sid}{model}{current}"]
|
|
|
|
def _proc_row(proc):
|
|
cmd = _clip(" ".join(str(proc.get("command", "")).split()), 90)
|
|
return [f"- `{proc.get('session_id', '?')}` · "
|
|
f"{format_uptime_short(int(proc.get('uptime_seconds', 0)))} · `{cmd}`"]
|
|
|
|
lines = [t("gateway.agents.header"), "", t("gateway.agents.active_agents", count=len(agent_rows))]
|
|
lines += _capped_rows(list(enumerate(agent_rows, 1)), _agent_row)
|
|
lines += ["", t("gateway.agents.running_processes", count=len(running_processes))]
|
|
lines += _capped_rows(running_processes, _proc_row)
|
|
lines += ["", t("gateway.agents.async_jobs", count=len(background_tasks))]
|
|
if delegations:
|
|
lines += ["", t("gateway.agents.background_delegations", count=len(delegations))]
|
|
lines += _capped_rows(delegations, _agents_delegation_lines)
|
|
if not (agent_rows or running_processes or background_tasks or delegations):
|
|
lines += ["", t("gateway.agents.none")]
|
|
return "\n".join(lines)
|
|
|
|
async def _handle_topup_command(self, event: MessageEvent) -> str:
|
|
"""Handle /topup -- show the Nous balance and hand off to the portal. Does NOT charge, confirm,
|
|
or track payment (that happens in the browser; the next /topup shows the new balance)."""
|
|
from agent.account_usage import build_credits_view
|
|
view = await _quiet(lambda: asyncio.to_thread(build_credits_view, markdown=True))
|
|
if view is None or not view.logged_in:
|
|
return t("gateway.credits.not_logged_in")
|
|
# Drop the helper's 📈 header; we print our own.
|
|
lines = [t("gateway.topup.header")] + [ln for ln in view.balance_lines if not ln.lstrip().startswith("📈")]
|
|
if view.identity_line:
|
|
lines += ["", view.identity_line]
|
|
if view.topup_url:
|
|
lines += ["", t("gateway.topup.portal_link", url=view.topup_url), t("gateway.topup.hint")]
|
|
return "\n".join(lines)
|
|
|
|
def _context_breakdown_block(self, agent, source, expanded: bool) -> list[str]:
|
|
"""/context per-category block (plain text, chars/4 estimate, same engine as /usage).
|
|
Runs in a thread; returns [] and never raises."""
|
|
try:
|
|
from agent.context_breakdown import compute_context_details, render_context_breakdown_lines
|
|
try:
|
|
payload = self._session_context_breakdown(agent, source)
|
|
except TranscriptReadError:
|
|
return [history_unreadable()] # a read failure is not an empty transcript
|
|
if not (payload.get("categories") or []):
|
|
return []
|
|
details = _quiet_sync(lambda: compute_context_details(agent), {"skills": [], "toolsets": []}) if expanded else None
|
|
from agent.context_file_sources import context_file_sources_for_agent, render_context_file_lines
|
|
file_lines = _quiet_sync(lambda: render_context_file_lines(context_file_sources_for_agent(agent)), [])
|
|
return render_context_breakdown_lines(payload, details=details, grid=False) + ([""] + file_lines if file_lines else [])
|
|
except Exception:
|
|
return []
|
|
|
|
def _session_context_breakdown(self, agent, source) -> dict:
|
|
"""Per-category context estimate (chars/4) for *agent* over the session transcript (sync).
|
|
Raises ``TranscriptReadError`` (unreadable rows must not pass as an empty transcript)."""
|
|
from agent.context_breakdown import compute_session_context_breakdown
|
|
store = self.session_store
|
|
try:
|
|
history = store.load_transcript(store.get_or_create_session(source).session_id) or []
|
|
except TranscriptReadError:
|
|
raise
|
|
except Exception:
|
|
history = []
|
|
return compute_session_context_breakdown(agent, history)
|
|
|
|
def _context_breakdown_lines(self, agent, source) -> list[str]:
|
|
"""/usage per-category context breakdown (chars/4 estimate). Returns [] and never raises."""
|
|
try:
|
|
try:
|
|
payload = self._session_context_breakdown(agent, source)
|
|
except TranscriptReadError:
|
|
return [history_unreadable()]
|
|
categories = payload.get("categories") or []
|
|
if not categories:
|
|
return []
|
|
total = payload.get("estimated_total") or 0
|
|
out = [t("gateway.usage.breakdown_header")]
|
|
for cat in categories:
|
|
tokens = int(cat.get("tokens") or 0)
|
|
if tokens <= 0:
|
|
continue
|
|
cat_id = str(cat.get("id") or "")
|
|
label = t(f"gateway.usage.breakdown_cat_{cat_id}")
|
|
if label.endswith(f"breakdown_cat_{cat_id}"): # missing key: t() echoes it back
|
|
label = str(cat.get("label") or cat_id)
|
|
pct = round(tokens / total * 100) if total else 0
|
|
out.append(t("gateway.usage.breakdown_line", label=label, count=_fmt(tokens), pct=f"~{pct}"))
|
|
return out if len(out) > 1 else []
|
|
except Exception:
|
|
return []
|
|
|
|
async def _handle_usage_command(self, event: MessageEvent) -> str:
|
|
"""Handle /usage -- token usage for the current session (live or cached agent) plus
|
|
account/credit blocks; ``/usage reset [--force]`` redeems a banked Codex reset credit."""
|
|
source = event.source
|
|
session_key = self._session_key_for_source(source)
|
|
raw_args = event.get_command_args().strip()
|
|
args = [a.lower() for a in raw_args.split()] if raw_args else []
|
|
wants_reset = bool(args) and args[0] == "reset"
|
|
if args and not wants_reset:
|
|
return t("gateway.usage.unknown_subcommand", args=raw_args)
|
|
|
|
# Running agent first (mid-turn), then cached agent (between turns).
|
|
agent = self._resident_agent_for(session_key)
|
|
|
|
# Provider/base_url/api_key for the account-usage fetch: live agent first, else persisted
|
|
# billing data on the SessionDB row so `/usage` still returns account info between turns.
|
|
provider, base_url, api_key = (
|
|
getattr(agent, k, None) if agent else None for k in ("provider", "base_url", "api_key")
|
|
)
|
|
if not provider and getattr(self, "_session_db", None) is not None:
|
|
provider, base_url = await self._persisted_billing_route(source)
|
|
if not provider:
|
|
# Fresh or evicted session with no persisted route (e.g. /usage right after login):
|
|
# fall back to the configured provider, as /status does, so account limits such as
|
|
# Codex subscription windows still render from on-disk credentials (#15167).
|
|
provider = await _quiet(lambda: asyncio.to_thread(_configured_provider)) or None
|
|
if wants_reset:
|
|
if str(provider or "").strip().lower() != "openai-codex":
|
|
return t("gateway.usage.reset_wrong_provider")
|
|
from agent.account_usage import redeem_codex_reset_credit
|
|
result = await asyncio.to_thread(
|
|
redeem_codex_reset_credit, base_url=base_url, api_key=api_key, force="--force" in args[1:],
|
|
)
|
|
return result.message
|
|
|
|
# Account usage off the event loop so slow provider APIs don't block the gateway;
|
|
# failures are non-fatal (account_lines stays []).
|
|
account_snapshot = provider and await _quiet(
|
|
lambda: asyncio.to_thread(fetch_account_usage, provider, base_url=base_url, api_key=api_key)
|
|
)
|
|
account_lines = (
|
|
render_account_usage_lines(account_snapshot, markdown=True) if account_snapshot else []
|
|
)
|
|
|
|
# Nous credits + monthly-grant gauge (shared with CLI/TUI). Gates on "a Nous account is
|
|
# logged in" — NOT the inference provider — so a Nous user inferring elsewhere still sees
|
|
# a balance. Fail-open: never break /usage.
|
|
from agent.account_usage import nous_credits_lines
|
|
credits_state = getattr(agent, "_credits_state", None) if agent else None
|
|
credits_lines = await _quiet(
|
|
lambda: asyncio.to_thread(nous_credits_lines, markdown=True, credits_state=credits_state),
|
|
[],
|
|
)
|
|
# Window plans: a chat's /topup has no Keep going control, so point the setting at the Portal.
|
|
from agent.plan_limits import portal_wording
|
|
credits_lines = [portal_wording(line) for line in credits_lines]
|
|
|
|
def _with_account_blocks(lines: list[str]) -> str:
|
|
# Each block is preceded by a blank divider only when something precedes it.
|
|
for block in (account_lines, credits_lines):
|
|
if block:
|
|
if lines:
|
|
lines.append("")
|
|
lines.extend(block)
|
|
return "\n".join(lines)
|
|
if agent and hasattr(agent, "session_total_tokens") and agent.session_api_calls > 0:
|
|
lines = _usage_agent_stats_lines(agent)
|
|
# Per-category breakdown (chars/4 estimate, same engine as the desktop popover): prompt
|
|
# / tools / skills / memory off the live agent, conversation from the transcript.
|
|
breakdown_lines = await asyncio.to_thread(self._context_breakdown_lines, agent, source)
|
|
if breakdown_lines:
|
|
lines += [""] + breakdown_lines
|
|
return _with_account_blocks(lines)
|
|
|
|
# No agent at all -- rough count from session history
|
|
session_entry = await self.async_session_store.get_or_create_session(source)
|
|
try:
|
|
history = await self.async_session_store.load_transcript(session_entry.session_id)
|
|
except TranscriptReadError:
|
|
return history_unreadable()
|
|
if history:
|
|
approx, count = _transcript_estimate(history)
|
|
return _with_account_blocks([
|
|
t("gateway.usage.header_session_info"),
|
|
t("gateway.usage.label_messages", count=count),
|
|
t("gateway.usage.label_estimated_context", count=_fmt(approx)),
|
|
t("gateway.usage.detailed_after_first"),
|
|
])
|
|
if account_lines or credits_lines:
|
|
return _with_account_blocks([])
|
|
return t("gateway.usage.no_data")
|
|
|
|
async def _persisted_billing_route(self, source):
|
|
"""``(provider, base_url)`` from the SessionDB row / most recent route when no agent is resident."""
|
|
async def _rows():
|
|
entry = await self.async_session_store.get_or_create_session(source)
|
|
persisted = await self._session_db.get_session(entry.session_id) or {}
|
|
route = await self._session_db.get_recent_session_model_route(entry.session_id)
|
|
return persisted, route if isinstance(route, dict) else {}
|
|
persisted, recent = await _quiet(_rows, ({}, {}))
|
|
row = recent if recent.get("billing_provider") else persisted
|
|
return row.get("billing_provider"), row.get("billing_base_url")
|
|
|
|
async def _handle_insights_command(self, event: MessageEvent) -> str:
|
|
"""Handle /insights [N | --days N] [--source S] -- usage insights and analytics."""
|
|
# Normalize Unicode dashes (Telegram/iOS auto-converts -- to em/en dash)
|
|
args = re.sub(r'[\u2012\u2013\u2014\u2015](days|source)', r'--\1', event.get_command_args().strip())
|
|
days, source = 30, None
|
|
parts = args.split()
|
|
i = 0
|
|
while i < len(parts):
|
|
flag, value = parts[i], parts[i + 1] if i + 1 < len(parts) else None
|
|
if flag == "--days" and value is not None:
|
|
try:
|
|
days = int(value)
|
|
except ValueError:
|
|
return t("gateway.insights.invalid_days", value=value)
|
|
i += 2
|
|
elif flag == "--source" and value is not None:
|
|
source, i = value, i + 2
|
|
else:
|
|
days = int(flag) if flag.isdigit() else days
|
|
i += 1
|
|
try:
|
|
from hermes_state_registry import acquire
|
|
from agent.insights import InsightsEngine
|
|
|
|
def _run_insights():
|
|
db = acquire()
|
|
try:
|
|
engine = InsightsEngine(db)
|
|
return engine.format_gateway(engine.generate(days=days, source=source))
|
|
finally:
|
|
from hermes_state_registry import release_or_close
|
|
release_or_close(db)
|
|
|
|
# Not a bare hop: ``SessionDB()`` resolves ``get_hermes_home()`` at call time, a
|
|
# contextvar set by ``_profile_runtime_scope``; a default-executor hop starts with an
|
|
# EMPTY context and would read the DEFAULT profile's state.db.
|
|
return await self._run_in_executor_with_context(_run_insights)
|
|
except Exception as e:
|
|
logger.error("Insights command error: %s", e, exc_info=True)
|
|
return t("gateway.insights.error", error=e)
|