ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

refactor(tui_gateway): dedupe config/voice/profiles helpers, collapse ladders and hug payloads

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:02:07 -07:00
parent 113f04616b
commit f997235f1b
4 changed files with 282 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tui_gateway/methods_config.py
+36 -112
View File
@@ -31,7 +31,6 @@ def _(rid, params: dict) -> dict:
if db is None:
return _ok(rid, {"repos": []})
from hermes_cli import projects_db as pdb
policy = _repo_discovery_policy()
with pdb.connect_closing() as conn:
_reconcile_repo_discovery(pdb, conn, policy, _repo_discovery_policy_key(policy))
@@ -53,20 +52,14 @@ def _(rid, params: dict) -> dict:
crawl runs on the desktop), then return the merged repo list."""
try:
from hermes_cli import projects_db as pdb
policy = _repo_discovery_policy()
policy_key = _repo_discovery_policy_key(policy)
incoming_raw = params.get("discovery_policy")
incoming_policy = (
_repo_discovery_policy(incoming_raw) if isinstance(incoming_raw, dict) else None
)
incoming_policy = _repo_discovery_policy(incoming_raw) if isinstance(incoming_raw, dict) else None
incoming_matches = (
incoming_policy is not None
and _repo_discovery_policy_key(incoming_policy) == policy_key
)
accept_legacy_default = (
incoming_policy is None and _repo_discovery_policy_is_default(policy)
incoming_policy is not None and _repo_discovery_policy_key(incoming_policy) == policy_key
)
accept_legacy_default = incoming_policy is None and _repo_discovery_policy_is_default(policy)
pairs: list[tuple[str, str | None]] = []
for item in params.get("repos") or []:
@@ -84,16 +77,8 @@ def _(rid, params: dict) -> dict:
pdb.clear_discovered_repos(conn, policy_key=policy_key)
with _profile_db(params) as db:
return _ok(
rid,
{
"repos": _discover_repos_payload(db, include_cached=policy["enabled"])
if db is not None
else [],
"accepted": accepted,
"discovery_policy": policy,
},
)
repos = _discover_repos_payload(db, include_cached=policy["enabled"]) if db is not None else []
return _ok(rid, {"repos": repos, "accepted": accepted, "discovery_policy": policy})
except Exception as e:
return _err(rid, 5061, str(e))
@@ -101,7 +86,6 @@ def _(rid, params: dict) -> dict:
def _stamped_project_tree(db, params, **kwargs):
"""``_build_project_tree`` + profile stamping shared by the two tree RPCs."""
from tui_gateway.project_tree import stamp_profile
tree, active_id = _build_project_tree(db, **kwargs)
stamp_profile(tree["projects"], _response_profile_name(params.get("profile")))
return tree, active_id
@@ -120,21 +104,13 @@ def _(rid, params: dict) -> dict:
if db is None:
return _ok(rid, {"projects": [], "active_id": None, "scoped_session_ids": []})
tree, active_id = _stamped_project_tree(
db,
params,
preview_limit=int(params.get("preview_limit") or 3),
hydrate=False,
session_limit=int(params.get("session_limit") or 2000),
include_discovered=True,
)
return _ok(
rid,
{
"projects": tree["projects"],
"active_id": active_id,
"scoped_session_ids": tree["scoped_session_ids"],
},
db, params, preview_limit=int(params.get("preview_limit") or 3), hydrate=False,
session_limit=int(params.get("session_limit") or 2000), include_discovered=True,
)
return _ok(rid, {
"projects": tree["projects"], "active_id": active_id,
"scoped_session_ids": tree["scoped_session_ids"],
})
except Exception as e:
return _err(rid, 5061, str(e))
@@ -156,12 +132,8 @@ def _(rid, params: dict) -> dict:
# Drill-in only needs the entered project (which has sessions):
# skip the zero-session discovery tier.
tree, _active = _stamped_project_tree(
db,
params,
preview_limit=0,
hydrate=True,
session_limit=int(params.get("session_limit") or 5000),
include_discovered=False,
db, params, preview_limit=0, hydrate=True,
session_limit=int(params.get("session_limit") or 5000), include_discovered=False,
)
proj = next((p for p in tree["projects"] if p["id"] == project_id), None)
return _ok(rid, {"project": proj})
@@ -175,24 +147,17 @@ def _(rid, params: dict) -> dict:
# ---------------------------------------------------------------------------
def _display_cfg() -> dict:
display = _load_cfg().get("display")
return display if isinstance(display, dict) else {}
def _display_mode(cfg: dict, key: str, allowed: frozenset, default: str) -> str:
raw = str((cfg.get("display") or {}).get(key, default) or default).strip().lower()
return raw if raw in allowed else default
_DETAILS_MODES = frozenset({"hidden", "collapsed", "expanded"})
_THINKING_MODES = frozenset({"collapsed", "truncated", "full"})
def _cfg_get_provider(rid, params):
try:
from hermes_cli.models import list_available_providers, normalize_provider
model = _resolve_model()
parts = model.split("/", 1)
return {
@@ -206,7 +171,6 @@ def _cfg_get_provider(rid, params):
def _cfg_get_profile(rid, params):
from hermes_constants import display_hermes_home
return {"home": str(_hermes_home), "display": display_hermes_home()}
@@ -229,7 +193,6 @@ def _cfg_get_personality(rid, params):
# EFFECTIVE personality via the single owner — a stale/unknown name in
# config must not display as active.
from hermes_cli.personality import active_personality_name
return {"value": active_personality_name(_load_cfg()) or "none"}
@@ -238,13 +201,9 @@ def _cfg_get_reasoning(rid, params):
session = _sessions.get(params.get("session_id", ""))
reasoning_config = None
if session is not None:
if isinstance(session.get("create_reasoning_override"), dict):
reasoning_config = session.get("create_reasoning_override")
else:
agent_reasoning = getattr(session.get("agent"), "reasoning_config", None)
if isinstance(agent_reasoning, dict):
reasoning_config = agent_reasoning
reasoning_config = session.get("create_reasoning_override")
if not isinstance(reasoning_config, dict):
reasoning_config = getattr(session.get("agent"), "reasoning_config", None)
if isinstance(reasoning_config, dict):
if reasoning_config.get("enabled") is False:
effort = "none"
@@ -287,7 +246,7 @@ def _cfg_get_thinking_mode(rid, params):
raw = str((cfg.get("display") or {}).get("thinking_mode", "") or "").strip().lower()
if raw in _THINKING_MODES:
return {"value": raw}
dm = _display_mode(cfg, "details_mode", _DETAILS_MODES, "collapsed")
dm = _display_mode(cfg, "details_mode", _DETAIL_MODES, "collapsed")
return {"value": "full" if dm == "expanded" else "collapsed"}
@@ -332,7 +291,7 @@ def _config_getters() -> dict:
"approval_mode": _cfg_get_approval_mode,
"approvals.mode": _cfg_get_approval_mode,
"details_mode": lambda rid, params: {
"value": _display_mode(_load_cfg(), "details_mode", _DETAILS_MODES, "collapsed")
"value": _display_mode(_load_cfg(), "details_mode", _DETAIL_MODES, "collapsed")
},
"thinking_mode": _cfg_get_thinking_mode,
"density": lambda rid, params: {
@@ -376,12 +335,10 @@ def _readiness_profile_scope(params: dict):
quietly answer for the launch profile instead.
"""
import contextlib
profile = str(params.get("profile") or "").strip() if isinstance(params, dict) else ""
if not profile:
return "", contextlib.nullcontext()
from hermes_cli import profiles as profiles_mod
if not profiles_mod.profile_exists(profile):
raise FileNotFoundError(f"Profile '{profile}' does not exist on this backend.")
home = _profile_home(profile)
@@ -410,13 +367,9 @@ def _(rid, params: dict) -> dict:
"""Loose provider check; ``profile`` (optional) scopes it to that profile's home."""
try:
from hermes_cli.main import _has_any_provider_configured
def probe(profile):
configured = bool(_has_any_provider_configured(strict_profile_scope=bool(profile)))
payload = {"provider_configured": configured}
if profile:
payload["profile"] = profile
return payload
return {"provider_configured": configured, **({"profile": profile} if profile else {})}
return _readiness_check(rid, params, probe)
except Exception as e:
@@ -439,7 +392,6 @@ def _(rid, params: dict) -> dict:
from hermes_cli.runtime_provider import resolve_runtime_provider
from hermes_cli.auth import has_usable_secret
from hermes_cli.main import _has_any_provider_configured
requested = str(params.get("provider") or "").strip() or None
def probe(profile):
@@ -450,44 +402,24 @@ def _(rid, params: dict) -> dict:
scoped = {"profile": profile} if profile else {}
provider = runtime.get("provider") or "provider"
source = str(runtime.get("source") or "")
if (
not provider_configured
and provider == "bedrock"
and source in {"iam-role", "aws-sdk-default-chain"}
):
return {
"ok": False,
"provider": provider,
"model": runtime.get("model"),
"source": source,
"error": "No Hermes provider is configured.",
**scoped,
}
def fail(error, src):
return {"ok": False, "provider": provider, "model": runtime.get("model"),
"source": src, "error": error, **scoped}
if (not provider_configured and provider == "bedrock"
and source in {"iam-role", "aws-sdk-default-chain"}):
return fail("No Hermes provider is configured.", source)
api_key = runtime.get("api_key")
api_key_text = "" if callable(api_key) else str(api_key or "").strip()
credential_ok = (
callable(api_key)
or api_key_text in {"aws-sdk", "no-key-required"}
or has_usable_secret(api_key_text)
or bool(runtime.get("command"))
callable(api_key) or api_key_text in {"aws-sdk", "no-key-required"}
or has_usable_secret(api_key_text) or bool(runtime.get("command"))
)
if not credential_ok:
return {
"ok": False,
"provider": provider,
"model": runtime.get("model"),
"source": runtime.get("source"),
"error": f"No usable credentials found for {provider}.",
**scoped,
}
return {
"ok": True,
"provider": runtime.get("provider"),
"model": runtime.get("model"),
"source": runtime.get("source"),
**scoped,
}
return fail(f"No usable credentials found for {provider}.", runtime.get("source"))
return {"ok": True, "provider": runtime.get("provider"), "model": runtime.get("model"),
"source": runtime.get("source"), **scoped}
return _readiness_check(rid, params, probe)
except Exception as e:
@@ -512,7 +444,6 @@ def _(rid, params: dict) -> dict:
try:
from hermes_cli.debug import _redact_log_text, build_nous_bundle, collect_share_bundle
from hermes_cli.diagnostics_upload import share_to_nous
log_lines = params.get("log_lines")
if not isinstance(log_lines, int) or not (10 <= log_lines <= 2000):
log_lines = 200
@@ -548,18 +479,11 @@ def _(rid, params: dict) -> dict:
upload_id = res.get("id")
if not view_url and not upload_id:
# An upload the user can't reference is useless to support.
return _ok(
rid, {"ok": False, "error": "upload succeeded but returned no view URL or id"}
)
return _ok(
rid,
{
"ok": True,
"view_url": view_url,
"upload_id": upload_id,
"expires_at": res.get("expiresAt") or res.get("expires_at"),
},
)
return _ok(rid, {"ok": False, "error": "upload succeeded but returned no view URL or id"})
return _ok(rid, {
"ok": True, "view_url": view_url, "upload_id": upload_id,
"expires_at": res.get("expiresAt") or res.get("expires_at"),
})
except Exception as e:
return _ok(rid, {"ok": False, "error": str(e)})
tui_gateway/methods_config_set.py
+40 -59
View File
@@ -52,28 +52,28 @@ def _emit_all_session_info() -> None:
def _toggle_display_bool(rid, key, value, *, cfg_key, on_words, off_words):
"""Shared body of the on/off/toggle display booleans (``density``, ``battery``)."""
raw = str(value or "").strip().lower()
cur_b = bool(_display_section(_load_cfg()).get(cfg_key, False))
raw = _word(value)
cur_b = bool(_display_cfg().get(cfg_key, False))
if raw in {"", "toggle"}:
nv_b = not cur_b
elif raw in on_words:
nv_b = True
elif raw in off_words:
nv_b = False
elif raw in on_words or raw in off_words:
nv_b = raw in on_words
else:
return _err(rid, 4002, f"unknown {key} value: {value}")
_write_config_key(f"display.{cfg_key}", nv_b)
return _ok(rid, {"key": key, "value": "on" if nv_b else "off"})
def _word(value) -> str:
return str(value or "").strip().lower()
def _cfgset_await_agent(session, rid):
"""Wait for an in-progress agent build; the error envelope if it failed, else None."""
init_err = _wait_agent(session, rid)
if init_err:
return init_err
if session.get("agent") is None:
return _err(rid, 5032, "agent initialization failed")
return None
return _err(rid, 5032, "agent initialization failed") if session.get("agent") is None else None
def _cfgset_model_ok(rid, key, value, warning, confirm_required, confirm_message, scope, **extra):
@@ -92,7 +92,6 @@ def _set_model(rid, params, key, value, session):
confirmed = bool(params.get("confirm_expensive_model", False))
if session:
from hermes_cli.model_switch import parse_model_switch_args
sid = params.get("session_id", "")
# No live swap while a turn streams: agent.switch_model() mutates model/provider/
# base_url/client that the worker thread reads every iteration. Stash the pick and
@@ -138,8 +137,7 @@ def _set_model(rid, params, key, value, session):
)
if session.get("agent") is None and not explicit_provider.strip() and not failed_agent_init:
_start_agent_build(sid, session)
init_err = _cfgset_await_agent(session, rid)
if init_err:
if init_err := _cfgset_await_agent(session, rid):
return init_err
with _session_profile_runtime_scope(session):
result = _apply_model_switch(
@@ -147,8 +145,7 @@ def _set_model(rid, params, key, value, session):
)
if failed_agent_init and not result.get("confirm_required"):
_restart_completed_failed_agent_build(sid, session, failed_ready)
init_err = _cfgset_await_agent(session, rid)
if init_err:
if init_err := _cfgset_await_agent(session, rid):
return init_err
with _session_profile_runtime_scope(session):
_persist_live_session_runtime(session)
@@ -162,8 +159,11 @@ def _set_model(rid, params, key, value, session):
return _err(rid, 5001, str(e))
_FAST_WORDS = {"fast": "fast", "on": "fast", "normal": "normal", "off": "normal", "auto": "auto", "cold": "cold"}
def _set_fast(rid, params, key, value, session):
raw = str(value or "").strip().lower()
raw = _word(value)
agent = session.get("agent") if session else None
if agent is not None:
current_tier = getattr(agent, "service_tier", None)
@@ -174,23 +174,15 @@ def _set_fast(rid, params, key, value, session):
current_tier = _load_service_tier()
current_fast = current_tier == "priority"
if raw in {"status"}:
if raw == "status":
return _ok(rid, {"key": key, "value": {"priority": "fast", None: "normal"}.get(current_tier, current_tier)})
if raw in {"", "toggle"}:
nv = "normal" if current_fast else "fast"
elif raw in {"fast", "on"}:
nv = "fast"
elif raw in {"normal", "off"}:
nv = "normal"
elif raw in {"auto", "cold"}:
nv = raw
else:
nv = _FAST_WORDS.get(raw, ("normal" if current_fast else "fast") if raw in {"", "toggle"} else None)
if nv is None:
return _err(rid, 4002, f"unknown fast mode: {value}")
overrides = None
if nv == "fast":
from hermes_cli.models import resolve_fast_mode_overrides
if agent is not None:
target_model = getattr(agent, "model", None)
else:
@@ -200,8 +192,7 @@ def _set_fast(rid, params, key, value, session):
if not target_model:
return _err(rid, 4002, "fast mode is not available without a selected model")
overrides = resolve_fast_mode_overrides(
target_model, provider=getattr(agent, "provider", None), base_url=getattr(agent, "base_url", None)
)
target_model, provider=getattr(agent, "provider", None), base_url=getattr(agent, "base_url", None))
if overrides is None:
return _err(rid, 4002, "fast mode is not available for this model")
@@ -226,7 +217,7 @@ def _set_fast(rid, params, key, value, session):
def _set_busy(rid, params, key, value, session):
raw = str(value or "").strip().lower()
raw = _word(value)
if raw in {"", "status"}:
return _ok(rid, {"key": key, "value": _load_busy_input_mode()})
if raw not in {"queue", "steer", "interrupt"}:
@@ -248,9 +239,8 @@ def _set_verbose(rid, params, key, value, session):
_write_config_key("display.tool_progress", nv)
if session:
session["tool_progress_mode"] = nv
agent = session.get("agent")
if agent is not None:
agent.verbose_logging = nv == "verbose"
if session.get("agent") is not None:
session["agent"].verbose_logging = nv == "verbose"
return _ok(rid, {"key": key, "value": nv})
@@ -258,8 +248,7 @@ def _set_focus(rid, params, key, value, session):
# Focus view (/focus): display-only reduced output composed with tool_progress — enabling
# stashes the configured mode and pins tool_progress "off"; disabling restores the stash.
from hermes_cli.focus_view import FOCUS_TOOL_PROGRESS_MODE, normalize_tool_progress_mode, resolve_focus_arg
d_f = _display_section(_load_cfg())
d_f = _display_cfg()
cur_focus = bool(d_f.get("focus_view", False))
action, target = resolve_focus_arg(str(value or ""), cur_focus)
if action == "usage":
@@ -280,15 +269,14 @@ def _set_focus(rid, params, key, value, session):
if session:
session["focus_view"] = bool(target)
session["tool_progress_mode"] = effective
agent_f = session.get("agent")
if agent_f is not None:
if session.get("agent") is not None:
with contextlib.suppress(Exception):
agent_f.tool_progress_mode = effective
session["agent"].tool_progress_mode = effective
return _ok(rid, {"key": key, "value": "on" if target else "off", "tool_progress": effective})
def _set_approval_mode(rid, params, key, value, session):
raw = str(value or "").strip().lower()
raw = _word(value)
if raw not in _APPROVAL_MODES:
return _err(rid, 4002, f"unknown approval mode: {value}; pick one of manual|smart|off")
_write_config_key("approvals.mode", raw)
@@ -300,20 +288,17 @@ def _set_yolo(rid, params, key, value, session):
# Approval bypass. scope="session" (default; TUI Shift+Tab) toggles ONLY this session's flag.
# scope="global" (Shift+click the zap) flips persistent approvals.mode between "off" (bypass
# on) and "manual" (bypass off) for every surface, surviving restarts.
scope = str(params.get("scope") or "session").strip().lower()
scope = _word(params.get("scope") or "session")
try:
from tools.approval import disable_session_yolo, enable_session_yolo, is_session_yolo_enabled
raw = str(value or "").strip().lower()
raw = _word(value)
def _resolve_toggle(current: bool) -> bool:
return _BOOL_WORDS.get(raw, not current)
if scope == "global":
from tools.approval import _normalize_approval_mode
cfg = _load_cfg()
appr = cfg.get("approvals") if isinstance(cfg, dict) else None
appr = _load_cfg().get("approvals")
appr = appr if isinstance(appr, dict) else {}
enable = _resolve_toggle(_normalize_approval_mode(appr.get("mode", "manual")) == "off")
# Binary affordance: no restore of a prior "smart"/custom mode (those live in config.yaml).
@@ -351,9 +336,8 @@ _REASONING_DISPLAY_WORDS = (
def _set_reasoning(rid, params, key, value, session):
try:
from hermes_constants import parse_reasoning_effort
arg = str(value or "").strip().lower()
scope = str(params.get("scope") or "").strip().lower()
arg = _word(value)
scope = _word(params.get("scope"))
for words, reported, fields, thinking, show in _REASONING_DISPLAY_WORDS:
if arg in words:
_write_display_sections(sections={"thinking": thinking}, **fields)
@@ -382,7 +366,7 @@ def _set_reasoning(rid, params, key, value, session):
def _set_details_mode(rid, params, key, value, session):
nv = str(value or "").strip().lower()
nv = _word(value)
if nv not in _DETAIL_MODES:
return _err(rid, 4002, f"unknown details_mode: {value}")
_write_display_sections(sections={section: nv for section in _DETAIL_SECTION_NAMES}, details_mode=nv)
@@ -395,7 +379,7 @@ def _set_details_section(rid, params, key, value, session):
section = key.split(".", 1)[1]
if section not in _DETAIL_SECTION_NAMES:
return _err(rid, 4002, f"unknown section: {section}")
nv = str(value or "").strip().lower()
nv = _word(value)
if not nv:
_write_display_sections(drop_sections=(section,))
elif nv not in _DETAIL_MODES:
@@ -406,7 +390,7 @@ def _set_details_section(rid, params, key, value, session):
def _set_thinking_mode(rid, params, key, value, session):
nv = str(value or "").strip().lower()
nv = _word(value)
if nv not in {"collapsed", "truncated", "full"}:
return _err(rid, 4002, f"unknown thinking_mode: {value}")
_write_config_key("display.thinking_mode", nv)
@@ -427,7 +411,7 @@ def _set_battery(rid, params, key, value, session):
def _set_theme(rid, params, key, value, session):
# 'light'/'dark' pin beats background auto-detection (xterm.js hosts misreport OSC 11).
raw = str(value or "").strip().lower()
raw = _word(value)
if raw not in {"auto", "light", "dark"}:
return _err(rid, 4002, f"unknown theme value: {value} (use auto|light|dark)")
_write_config_key("display.tui_theme", raw)
@@ -435,14 +419,12 @@ def _set_theme(rid, params, key, value, session):
def _set_statusbar(rid, params, key, value, session):
raw = str(value or "").strip().lower()
current = _coerce_statusbar(_display_section(_load_cfg()).get("tui_statusbar", "top"))
raw = _word(value)
current = _coerce_statusbar(_display_cfg().get("tui_statusbar", "top"))
if raw in {"", "toggle"}:
nv = "top" if current == "off" else "off"
elif raw == "on":
nv = "top"
elif raw in _STATUSBAR_MODES:
nv = raw
elif raw == "on" or raw in _STATUSBAR_MODES:
nv = "top" if raw == "on" else raw
else:
return _err(rid, 4002, f"unknown statusbar value: {value}")
_write_config_key("display.tui_statusbar", nv)
@@ -453,7 +435,7 @@ def _set_mouse(rid, params, key, value, session):
# Explicit None check (not `value or ""`) so falsy non-string inputs (0, False from
# programmatic callers) reach the alias map as themselves (-> 'off') instead of toggling.
raw = ("" if value is None else str(value)).strip().lower()
current = _display_mouse_tracking(_display_section(_load_cfg()))
current = _display_mouse_tracking(_display_cfg())
if raw in {"", "toggle"}:
nv = "all" if current == "off" else "off"
elif raw in _MOUSE_TRACKING_ALIASES:
@@ -501,7 +483,6 @@ def _set_prompt_like(rid, params, key, value, session):
# Personality text is an in-session overlay; persistence goes through
# hermes_cli.personality (single owner), never the user-owned global system prompt.
from hermes_cli.personality import persist_personality
persist_personality(pname)
resp["value"] = str(value or "none")
history_reset, info = _apply_personality_to_session(params.get("session_id", ""), session, new_prompt, pname)
tui_gateway/methods_profiles.py
+32 -66
View File
@@ -169,11 +169,8 @@ def _canonical_session_row(db, profile_path):
tip_row = db.get_session(tip) or row
started = row.get("started_at") or 0
return {
"id": session_id,
"resolved_id": tip,
"root_title": row.get("title") or "",
"title": tip_row.get("title") or "",
"preview": _latest_message_preview(db, tip),
"id": session_id, "resolved_id": tip, "root_title": row.get("title") or "",
"title": tip_row.get("title") or "", "preview": _latest_message_preview(db, tip),
"started_at": tip_row.get("started_at") or started,
"last_active": tip_row.get("last_activity_at") or tip_row.get("started_at") or started,
"message_count": tip_row.get("message_count") or 0,
@@ -199,16 +196,13 @@ def _latest_profile_session_rows(db):
continue
if human is not None:
continue
# Rosters want "where the conversation IS": prefer the newest text.
human = {
"id": s["id"],
"title": title,
"preview": s.get("preview") or "",
"started_at": s.get("started_at") or 0,
"last_active": last_active,
"id": s["id"], "title": title,
"preview": _latest_message_preview(db, s["id"]) or s.get("preview") or "",
"started_at": s.get("started_at") or 0, "last_active": last_active,
"message_count": s.get("message_count") or 0,
}
# Rosters want "where the conversation IS": prefer the newest text.
human["preview"] = _latest_message_preview(db, s["id"]) or human["preview"]
if worker is not None:
break
return human, worker
@@ -259,14 +253,10 @@ def _(rid, params: dict) -> dict:
out = []
for p in list_profiles():
row = {
"name": p.name,
"path": str(p.path),
"is_default": bool(p.is_default),
"model": p.model,
"provider": p.provider,
"description": getattr(p, "description", "") or "",
"display_name": getattr(p, "display_name", "") or "",
"skill_count": getattr(p, "skill_count", 0) or 0,
"name": p.name, "path": str(p.path), "is_default": bool(p.is_default),
"model": p.model, "provider": p.provider,
"description": p.description or "", "display_name": p.display_name or "",
"skill_count": p.skill_count or 0,
}
if include_sessions:
_profile_session_fields(row, p.path)
@@ -413,10 +403,8 @@ def _(rid, params: dict) -> dict:
model_set = _best_effort(lambda: _pin_profile_model(path, provider, model))
elif is_truthy_value(params.get("mirror_credentials", True)):
mirrored["model_inherited"] = _try(lambda: _inherit_launch_model(path), False)
return _ok(
rid,
{"ok": True, "name": name, "path": str(path), "soul_written": soul_written, "model_set": model_set, "mirrored": mirrored},
)
return _ok(rid, {"ok": True, "name": name, "path": str(path), "soul_written": soul_written,
"model_set": model_set, "mirrored": mirrored})
def _describe_toolsets(cfg):
@@ -437,9 +425,8 @@ def _describe_toolsets(cfg):
if (ts_name in default_off or ts_name == "yuanbao") and not enabled:
continue
tool_count = _try(lambda: len(set(resolve_toolset(ts_name))), 0)
toolsets_out.append(
{"name": ts_name, "label": ts_label, "description": ts_desc or "", "tool_count": tool_count, "enabled": enabled}
)
toolsets_out.append({"name": ts_name, "label": ts_label, "description": ts_desc or "",
"tool_count": tool_count, "enabled": enabled})
return toolsets_out, pinned_set
@@ -448,19 +435,11 @@ def _describe_mcp_servers(cfg):
mcp_cfg = cfg.get("mcp_servers")
if not isinstance(mcp_cfg, dict):
return []
return _try(
lambda: [
{
"name": str(srv_name),
"enabled": not is_truthy_value(entry.get("disabled", False)),
"transport": str(entry.get("transport") or "http") if entry.get("url") else "stdio",
}
for srv_name in sorted(mcp_cfg.keys())
for entry in (mcp_cfg[srv_name],)
if isinstance(entry, dict)
],
[],
)
return _try(lambda: [
{"name": str(srv_name), "enabled": not is_truthy_value(entry.get("disabled", False)),
"transport": str(entry.get("transport") or "http") if entry.get("url") else "stdio"}
for srv_name in sorted(mcp_cfg.keys()) for entry in (mcp_cfg[srv_name],) if isinstance(entry, dict)
], [])
@_profile_handler("profiles.describe", 5063)
@@ -485,17 +464,13 @@ def _(rid, params: dict) -> dict:
mcp_out = _describe_mcp_servers(cfg)
model_cfg = cfg.get("model") if isinstance(cfg.get("model"), dict) else {}
meta = _try(lambda: _lazy("hermes_cli.profiles", "read_profile_meta")(profile_dir), {})
result = {
"name": name,
"description": str(meta.get("description") or ""),
"soul": soul,
"model": {"provider": str(model_cfg.get("provider") or ""), "default": str(model_cfg.get("default") or "")},
"skills": installed,
"toolsets": toolsets_out,
"toolsets_pinned": pinned_set is not None,
return _ok(rid, {
"name": name, "description": str(meta.get("description") or ""), "soul": soul,
"model": {"provider": str(model_cfg.get("provider") or ""),
"default": str(model_cfg.get("default") or "")},
"skills": installed, "toolsets": toolsets_out, "toolsets_pinned": pinned_set is not None,
"mcp_servers": mcp_out,
}
return _ok(rid, result)
})
def _configure_ui_meta(profile_dir, params, applied) -> None:
@@ -555,14 +530,9 @@ def _configure_model(profile_dir, params, applied):
if not (model and provider):
return None
if not is_truthy_value(params.get("confirm_expensive_model", False)):
confirm_message = _try(
lambda: getattr(
_lazy("hermes_cli.model_selection_guards", "combined_selection_warning")(model, provider=provider or None),
"message",
None,
),
None,
)
confirm_message = _try(lambda: getattr(
_lazy("hermes_cli.model_selection_guards", "combined_selection_warning")(model, provider=provider or None),
"message", None), None)
if confirm_message is None:
applied["model"] = _best_effort(lambda: _pin_profile_model(profile_dir, provider, model))
return confirm_message
@@ -589,9 +559,8 @@ def _configure_cfg_sections(profile_dir, params, applied) -> None:
if isinstance(params.get("enabled_toolsets"), list):
applied["toolsets"] = _best_effort(lambda: _save_toolset_pin(cfg, params["enabled_toolsets"], save_config))
if want_mcp:
applied["mcp_servers"] = _best_effort(
lambda: _save_mcp_toggles(load_config() or {}, params["enabled_mcp_servers"], launch_mcp, save_config)
)
applied["mcp_servers"] = _best_effort(lambda: _save_mcp_toggles(
load_config() or {}, params["enabled_mcp_servers"], launch_mcp, save_config))
def _clean_names(values) -> set:
@@ -640,11 +609,8 @@ def _(rid, params: dict) -> dict:
if isinstance(params.get("soul"), str):
applied["soul"] = _best_effort(lambda: (profile_dir / "SOUL.md").write_text(params["soul"], encoding="utf-8"))
if isinstance(params.get("description"), str):
applied["description"] = _best_effort(
lambda: _lazy("hermes_cli.profiles", "write_profile_meta")(
profile_dir, description=params["description"].strip(), description_auto=False
)
)
applied["description"] = _best_effort(lambda: _lazy("hermes_cli.profiles", "write_profile_meta")(
profile_dir, description=params["description"].strip(), description_auto=False))
confirm_message = _configure_model(profile_dir, params, applied)
if any(isinstance(params.get(k), list) for k in ("disabled_skills", "enabled_toolsets", "enabled_mcp_servers")):
_configure_cfg_sections(profile_dir, params, applied)
tui_gateway/methods_voice.py
+174 -273
View File
@@ -6,6 +6,7 @@ method_ctx.bind_module), so they reference server.py globals bare.
from __future__ import annotations
import contextlib
import threading
from .method_ctx import HandlerRegistry, bind_module
@@ -58,17 +59,12 @@ def _end_voice_chat(*, stop_loop: bool, stop_tts: bool) -> None:
os.environ["HERMES_VOICE"] = "0"
os.environ["HERMES_VOICE_TTS"] = "0"
if stop_loop:
try:
with contextlib.suppress(Exception):
from hermes_cli.voice import stop_continuous
stop_continuous()
except Exception:
pass
if stop_tts:
try:
with contextlib.suppress(Exception):
_tts_stream_stop(user_barge=False)
except Exception:
pass
def _tts_lease_async(lease: str, active: bool) -> None:
@@ -79,11 +75,7 @@ def _tts_lease_async(lease: str, active: bool) -> None:
def _run():
try:
from tools.tts_tool import acquire_tts_lease, release_tts_lease
if active:
acquire_tts_lease(lease)
else:
release_tts_lease(lease)
(acquire_tts_lease if active else release_tts_lease)(lease)
except Exception as e:
logger.debug("voice: tts lease %s active=%s failed: %s", lease, active, e)
@@ -114,27 +106,18 @@ def _tts_stream_begin() -> Optional[queue.Queue]:
return None
try:
from tools.tts_tool import check_tts_requirements, stream_tts_to_speaker
if not check_tts_requirements():
return None
except Exception:
return None
_tts_stream_stop()
text_queue: queue.Queue = queue.Queue()
stop = threading.Event()
done = threading.Event()
threading.Thread(
target=stream_tts_to_speaker, args=(text_queue, stop, done), daemon=True
).start()
stop, done = threading.Event(), threading.Event()
threading.Thread(target=stream_tts_to_speaker, args=(text_queue, stop, done), daemon=True).start()
global _tts_stream_state
with _tts_stream_lock:
_tts_stream_state = {"stop": stop, "done": done}
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
_arm_full_duplex_listener()
_arm_barge_listener_if_enabled()
return text_queue
@@ -148,21 +131,14 @@ def _tts_stream_stop(user_barge: bool = True) -> None:
return
if user_barge and not state["done"].is_set():
import traceback as _tb
logger.debug(
"TTS CUT: _tts_stream_stop(user_barge=True) — new turn or "
"interrupt cutting in-flight TTS\n%s",
"".join(_tb.format_stack()),
)
logger.debug("TTS CUT: _tts_stream_stop(user_barge=True) — new turn or "
"interrupt cutting in-flight TTS\n%s", "".join(_tb.format_stack()))
from tools.tts_streaming import mark_speech_interrupted
mark_speech_interrupted()
state["stop"].set()
try:
with contextlib.suppress(Exception):
from tools.voice_mode import stop_playback
stop_playback()
except Exception:
pass
# ── Full-duplex agent-turn listener (one mic, whole turn) ──────────────────
@@ -187,15 +163,24 @@ def _arm_full_duplex_listener() -> None:
threading.Thread(target=_full_duplex_listener, daemon=True, name="voice-full-duplex").start()
def _arm_barge_listener_if_enabled() -> None:
"""Arm the listener when voice mode is on and ``voice.barge_in`` isn't disabled."""
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
_arm_full_duplex_listener()
def _fd_speak_pipelines_snapshot() -> list:
with _fd_listener_lock:
return list(_fd_speak_pipelines)
def _fd_tts_pending() -> bool:
"""True while any TTS (streaming pipeline or fallback speak) is unfinished."""
with _tts_stream_lock:
state = _tts_stream_state
if state is not None and not state["done"].is_set():
return True
with _fd_listener_lock:
pipelines = list(_fd_speak_pipelines)
return any(not done.is_set() for _stop, done in pipelines)
return any(not done.is_set() for _stop, done in _fd_speak_pipelines_snapshot())
def _full_duplex_listener() -> None:
@@ -209,99 +194,34 @@ def _full_duplex_listener() -> None:
"""
global _fd_listener_active
try:
from tools.tts_streaming import mark_speech_interrupted
from tools.voice_mode import (
full_duplex_listen,
is_audio_output_active,
stop_playback,
transcribe_recording,
)
cfg = _voice_cfg_dict()
try:
_mult = float(cfg.get("barge_in_threshold_multiplier", 0) or 0)
except (TypeError, ValueError):
_mult = 0.0
try:
_grace_ms = int(float(cfg.get("barge_in_grace_seconds", 0.5)) * 1000)
except (TypeError, ValueError):
_grace_ms = 500
from tools.voice_mode import full_duplex_listen, is_audio_output_active, transcribe_recording
def _should_stop() -> bool:
if not _voice_mode_enabled():
return True
if _any_session_running():
return False
if _fd_tts_pending():
if _any_session_running() or _fd_tts_pending():
return False
return not is_audio_output_active()
tripped = threading.Event()
def _cut_all_tts() -> None:
_tts_stream_stop(user_barge=True)
with _fd_listener_lock:
pipelines = list(_fd_speak_pipelines)
for _stop, _done in pipelines:
_stop.set()
stop_playback()
def _on_trigger(phase: str) -> None:
tripped.set()
mark_speech_interrupted()
if phase == "playback":
logger.debug("TTS CUT: full-duplex listener tripped during playback")
_cut_all_tts()
else:
logger.debug(
"full-duplex listener tripped during generation — "
"interrupting running turn(s)"
)
# Cut pending TTS FIRST so the stale reply can never speak.
_cut_all_tts()
try:
with _sessions_lock:
running = [s for s in _sessions.values() if s.get("running")]
for s in running:
agent = s.get("agent")
if agent is not None and hasattr(agent, "interrupt"):
try:
agent.interrupt()
except Exception:
pass
except Exception as e:
logger.debug("voice interjection interrupt failed: %s", e)
_voice_emit("voice.interrupted")
_fd_trip(phase)
wav_path = full_duplex_listen(
_should_stop, is_playing=is_audio_output_active, on_trigger=_on_trigger,
multiplier=_mult or None, grace_ms=max(0, _grace_ms),
)
mult, grace_ms = _fd_barge_params(_voice_cfg_dict())
wav_path = full_duplex_listen(_should_stop, is_playing=is_audio_output_active,
on_trigger=_on_trigger, multiplier=mult or None, grace_ms=grace_ms)
if not (wav_path and tripped.is_set()):
return
try:
result = transcribe_recording(wav_path)
text = (result.get("transcript") or "").strip() if result.get("success") else ""
if text:
# Stop-check must never break transcript delivery (stubbed
# voice_mode in tests, partial installs) — treat as not-a-stop.
try:
from tools.voice_mode import is_voice_stop_phrase
_is_stop = is_voice_stop_phrase(text)
except Exception:
_is_stop = False
if _is_stop:
# Turn already interrupted / TTS cut at trip time; now end the chat.
_end_voice_chat(stop_loop=True, stop_tts=False)
_voice_emit("voice.transcript", {"stop_phrase": True, "text": text})
else:
_voice_emit("voice.transcript", {"text": text})
_deliver_fd_transcript(text)
finally:
try:
with contextlib.suppress(OSError):
os.unlink(wav_path)
except OSError:
pass
except Exception as e:
logger.debug("full-duplex listener failed: %s", e)
finally:
@@ -309,14 +229,79 @@ def _full_duplex_listener() -> None:
_fd_listener_active = False
def _fd_barge_params(cfg: dict) -> tuple[float, int]:
"""``(threshold multiplier, grace ms)`` from the voice config; malformed -> defaults."""
try:
mult = float(cfg.get("barge_in_threshold_multiplier", 0) or 0)
except (TypeError, ValueError):
mult = 0.0
try:
grace_ms = int(float(cfg.get("barge_in_grace_seconds", 0.5)) * 1000)
except (TypeError, ValueError):
grace_ms = 500
return mult, max(0, grace_ms)
def _cut_all_tts() -> None:
"""Cut streaming TTS, every fallback speak pipeline, and the file player."""
from tools.voice_mode import stop_playback
_tts_stream_stop(user_barge=True)
for _stop, _done in _fd_speak_pipelines_snapshot():
_stop.set()
stop_playback()
def _fd_trip(phase: str) -> None:
"""Listener tripped: latch the interruption, cut TTS, and during generation also
interrupt every running turn (the ``agent.interrupt()`` seam ``session.interrupt`` uses)."""
from tools.tts_streaming import mark_speech_interrupted
mark_speech_interrupted()
if phase == "playback":
logger.debug("TTS CUT: full-duplex listener tripped during playback")
_cut_all_tts()
else:
logger.debug("full-duplex listener tripped during generation — interrupting running turn(s)")
# Cut pending TTS FIRST so the stale reply can never speak.
_cut_all_tts()
try:
with _sessions_lock:
running = [s for s in _sessions.values() if s.get("running")]
for s in running:
agent = s.get("agent")
if agent is not None and hasattr(agent, "interrupt"):
with contextlib.suppress(Exception):
agent.interrupt()
except Exception as e:
logger.debug("voice interjection interrupt failed: %s", e)
_voice_emit("voice.interrupted")
def _deliver_fd_transcript(text: str) -> None:
"""Emit the captured interjection; a bare stop phrase also ends the voice chat."""
# Stop-check must never break transcript delivery (stubbed voice_mode in tests,
# partial installs) — treat as not-a-stop.
try:
from tools.voice_mode import is_voice_stop_phrase
is_stop = is_voice_stop_phrase(text)
except Exception:
is_stop = False
if is_stop:
# Turn already interrupted / TTS cut at trip time; now end the chat.
_end_voice_chat(stop_loop=True, stop_tts=False)
_voice_emit("voice.transcript", {"stop_phrase": True, "text": text})
else:
_voice_emit("voice.transcript", {"text": text})
def _speak_text_with_barge(text: str) -> None:
"""Speak via hermes_cli.voice.speak_text with spoken barge-in: the (stop, done)
pair is registered in ``_fd_speak_pipelines`` so the full-duplex listener can cut
it on a playback trip and keeps listening while it is pending."""
from hermes_cli.voice import speak_text
stop = threading.Event()
done = threading.Event()
stop, done = threading.Event(), threading.Event()
with _fd_listener_lock:
_fd_speak_pipelines.add((stop, done))
@@ -332,8 +317,7 @@ def _speak_text_with_barge(text: str) -> None:
_fd_speak_pipelines.discard((stop, done))
threading.Thread(target=_speak, daemon=True).start()
if _voice_mode_enabled() and _voice_cfg_dict().get("barge_in", True):
_arm_full_duplex_listener()
_arm_barge_listener_if_enabled()
def _voice_cfg_dict() -> dict:
@@ -341,7 +325,6 @@ def _voice_cfg_dict() -> dict:
root and ``voice`` may be any YAML scalar/list/None; malformed → {}."""
cfg = _load_cfg()
voice_cfg = cfg.get("voice") if isinstance(cfg, dict) else None
return voice_cfg if isinstance(voice_cfg, dict) else {}
@@ -354,7 +337,6 @@ def _voice_cfg_number(value, default):
def _voice_record_key() -> str:
"""Current ``voice.record_key`` value, documented default on error."""
record_key = _voice_cfg_dict().get("record_key")
return str(record_key) if isinstance(record_key, str) and record_key else "ctrl+b"
@@ -374,17 +356,21 @@ def _wake_owner_snapshot():
return _wake_owner_transport, _wake_owner_surface
def _set_wake_owner(transport, surface: str) -> None:
global _wake_owner_transport, _wake_owner_surface
with _wake_lock:
_wake_owner_transport, _wake_owner_surface = transport, surface
def _release_wake_for_transport(transport: "Transport") -> bool:
"""Release the wake lease iff ``transport`` is the current gateway owner."""
global _wake_owner_transport, _wake_owner_surface
with _wake_lock:
if _wake_owner_transport is not transport:
return False
_wake_owner_transport = None
_wake_owner_surface = ""
_wake_owner_transport, _wake_owner_surface = None, ""
try:
from tools.wake_word import stop_listening
stop_listening(owner=transport)
except Exception as e:
logger.debug("wake stop failed: %s", e)
@@ -438,10 +424,8 @@ def _wake_resume_if_owner(owner: "Transport", *, retry_seconds: float = 15.0,
continue
# False — detector gone or lease moved: stop, don't fight it.
return
logger.warning(
"wake: could not resume detector after voice turn "
"(microphone still busy?) — toggle the wake word to re-arm"
)
logger.warning("wake: could not resume detector after voice turn "
"(microphone still busy?) — toggle the wake word to re-arm")
finally:
with _wake_resume_retry_lock:
_wake_resume_retry_active = False
@@ -455,7 +439,6 @@ def _persist_wake_enabled(enabled: bool) -> bool:
(ear toggle, /wake on|off) — never passive auto-arm paths."""
try:
from cli import save_config_value
return bool(save_config_value("wake_word.enabled", enabled))
except Exception as e:
logger.warning("wake: failed to persist wake_word.enabled=%s: %s", enabled, e)
@@ -474,9 +457,7 @@ def _wake_probe(cfg: dict, prefer_client: bool) -> tuple[str, dict]:
from tools.wake_word import check_wake_word_requirements, resolve_capture_mode
capture_mode = resolve_capture_mode(cfg, prefer_client=prefer_client)
probe_cfg = dict(cfg)
probe_cfg["capture"] = capture_mode
return capture_mode, check_wake_word_requirements(probe_cfg)
return capture_mode, check_wake_word_requirements({**cfg, "capture": capture_mode})
def _wake_detect_handler(transport, sid: str, phrase: str, new_session: bool):
@@ -486,9 +467,7 @@ def _wake_detect_handler(transport, sid: str, phrase: str, new_session: bool):
def _on_detect() -> None:
from tools.wake_word import get_last_match, owns_listener, pause_listening
if not pause_listening(owner=transport):
return
if not owns_listener(transport):
if not pause_listening(owner=transport) or not owns_listener(transport):
return
if _transport_is_dead(transport):
_release_wake_for_transport(transport)
@@ -517,7 +496,6 @@ def _(rid, params: dict) -> dict:
withholds unless the guarantee is advertised. Sourced from the enforcing module,
never config: a believed-but-absent capability is worse than none."""
from hermes_cli.active_sessions import PER_SESSION_EXCLUSIVE_SUBMIT
return _ok(rid, {"per_session_exclusive_submit": bool(PER_SESSION_EXCLUSIVE_SUBMIT)})
@@ -543,13 +521,8 @@ def _(rid, params: dict) -> dict:
transport = _caller_transport()
try:
from tools.wake_word import (
WakeWordInUse,
detector_frame_info,
load_wake_word_config,
owns_listener,
start_listening,
wake_phrase,
wake_surface_enabled,
WakeWordInUse, detector_frame_info, load_wake_word_config, owns_listener, start_listening,
wake_phrase, wake_surface_enabled,
)
except Exception as e:
return _err(rid, 5026, f"wake module unavailable: {e}")
@@ -565,12 +538,9 @@ def _(rid, params: dict) -> dict:
"started": False, "reason": "unavailable", "hint": reqs.get("hint") or "",
"capture": capture_mode,
})
enabled_persisted = False
if persist and not cfg.get("enabled"):
enabled_persisted = _persist_wake_enabled(True)
if enabled_persisted:
cfg = dict(cfg)
cfg["enabled"] = True
enabled_persisted = bool(persist and not cfg.get("enabled") and _persist_wake_enabled(True))
if enabled_persisted:
cfg = {**cfg, "enabled": True}
if not wake_surface_enabled(surface, cfg):
# "disabled" (a persist:true retry can turn it on) vs "disabled_for_surface"
# (explicit wake_word.surface choice, which persist does NOT override).
@@ -580,36 +550,22 @@ def _(rid, params: dict) -> dict:
return _ok(rid, {"started": False, "reason": reason})
existing_owner, existing_surface = _wake_owner_snapshot()
if existing_owner is not None and (
_transport_is_dead(existing_owner) or not owns_listener(existing_owner)
):
if existing_owner is not None and (_transport_is_dead(existing_owner) or not owns_listener(existing_owner)):
_release_wake_for_transport(existing_owner)
existing_owner = None
existing_surface = ""
existing_owner, existing_surface = None, ""
if existing_owner is not None and existing_owner is not transport:
return _ok(rid, {"started": False, "reason": "owned", "owner_surface": existing_surface})
sid = str(params.get("session_id") or "")
try:
start_listening(
_wake_detect_handler(
transport, sid, wake_phrase(cfg), bool(cfg.get("start_new_session", True))
),
owner=transport,
config=cfg,
external_audio=external_audio,
)
on_detect = _wake_detect_handler(transport, sid, wake_phrase(cfg), bool(cfg.get("start_new_session", True)))
start_listening(on_detect, owner=transport, config=cfg, external_audio=external_audio)
except WakeWordInUse:
return _ok(rid, {
"started": False, "reason": "owned", "owner_surface": existing_surface or None,
})
return _ok(rid, {"started": False, "reason": "owned", "owner_surface": existing_surface or None})
except Exception as e:
logger.warning("wake.start(%s): failed to start listener: %s", surface, e)
return _err(rid, 5026, str(e))
global _wake_owner_transport, _wake_owner_surface
with _wake_lock:
_wake_owner_transport = transport
_wake_owner_surface = surface
_set_wake_owner(transport, surface)
frame = detector_frame_info()
logger.info(
"wake.start(%s): listening for %r (%s) capture=%s frame=%s",
@@ -633,7 +589,6 @@ def _(rid, params: dict) -> dict:
if bool(params.get("persist")):
try:
from tools.wake_word import load_wake_word_config
currently_enabled = bool(load_wake_word_config().get("enabled"))
except Exception:
currently_enabled = True
@@ -651,7 +606,6 @@ def _(rid, params: dict) -> dict:
transport = _caller_transport()
try:
from tools.wake_word import pause_listening
paused = pause_listening(owner=transport)
logger.info("wake.pause: detector paused=%s", paused)
except Exception as e:
@@ -672,18 +626,12 @@ def _(rid, params: dict) -> dict:
def _(rid, params: dict) -> dict:
try:
from tools.wake_word import (
audio_is_silent,
detector_frame_info,
get_input_device_status,
is_listening,
load_wake_word_config,
owns_listener,
silent_audio_hint,
audio_is_silent, detector_frame_info, get_input_device_status, is_listening,
load_wake_word_config, owns_listener, silent_audio_hint,
)
cfg = load_wake_word_config()
probe_capture, reqs = _wake_probe(
cfg, _wake_prefers_client(params, str(params.get("surface") or "").strip().lower())
)
surface = str(params.get("surface") or "").strip().lower()
probe_capture, reqs = _wake_probe(cfg, _wake_prefers_client(params, surface))
transport = _caller_transport()
owner, owner_surface = _wake_owner_snapshot()
owned_by_caller = owns_listener(transport)
@@ -706,23 +654,17 @@ def _(rid, params: dict) -> dict:
else:
capture = probe_capture or reqs.get("capture") or str(cfg.get("capture") or "auto")
return _ok(rid, {
"listening": listening,
"owned_by_caller": owned_by_caller,
"listening": listening, "owned_by_caller": owned_by_caller,
"owner_surface": owner_surface if owner is not None else None,
"phrase": reqs["phrase"],
"provider": reqs["provider"],
"phrase": reqs["phrase"], "provider": reqs["provider"],
"configured_surface": str(cfg.get("surface") or "auto"),
"input_device": input_device,
"available": reqs["available"],
"hint": hint,
"input_device": input_device, "available": reqs["available"], "hint": hint,
# Config truth: clients re-arm after a voice turn ("permanent on") from this.
"enabled": bool(cfg.get("enabled")),
# Armed but deaf despite an open stream; see platform-specific hint.
"audio_silent": silent,
"capture": capture,
"audio_silent": silent, "capture": capture,
"local_input_available": bool(reqs.get("local_input_available")),
"sample_rate": frame.get("sample_rate", 16000),
"frame_length": frame.get("frame_length", 1280),
"sample_rate": frame.get("sample_rate", 16000), "frame_length": frame.get("frame_length", 1280),
})
except Exception as e:
return _err(rid, 5026, str(e))
@@ -762,19 +704,12 @@ def _(rid, params: dict) -> dict:
def _voice_toggle_status(rid, params: dict) -> dict:
# Mirrors CLI _show_voice_status: STT/TTS availability tells the user WHY
# voice isn't working; record_key lets the TUI bind and display the shortcut.
payload: dict = {
"enabled": _voice_mode_enabled(),
"record_key": _voice_record_key(),
"tts": _voice_tts_enabled(),
}
payload: dict = {"enabled": _voice_mode_enabled(), "record_key": _voice_record_key(), "tts": _voice_tts_enabled()}
try:
from tools.voice_mode import check_voice_requirements
reqs = check_voice_requirements()
payload["available"] = bool(reqs.get("available"))
payload["audio_available"] = bool(reqs.get("audio_available"))
payload["stt_available"] = bool(reqs.get("stt_available"))
payload["details"] = reqs.get("details") or ""
payload.update(available=bool(reqs.get("available")), audio_available=bool(reqs.get("audio_available")),
stt_available=bool(reqs.get("stt_available")), details=reqs.get("details") or "")
except Exception as e:
# Optional transcription deps — /voice status must always answer.
logger.warning("voice.toggle status: requirements probe failed: %s", e)
@@ -790,44 +725,30 @@ def _voice_toggle_mode(rid, params: dict) -> dict:
stop_hint = ""
if enabled:
# Spoken-stop hint for the client; sourced from voice.stop_phrases,
# empty when the feature is disabled.
# Spoken-stop hint for the client; sourced from voice.stop_phrases, empty when disabled.
try:
from tools.voice_mode import voice_stop_hint
stop_hint = voice_stop_hint()
except Exception:
stop_hint = ""
# Speech output already on → warm the engine now, not on the first reply.
if _voice_tts_enabled():
_tts_lease_async("tui:voice-tts", True)
if not enabled:
else:
# The continuous loop holds the microphone; tear it down with the mode.
try:
from hermes_cli.voice import stop_continuous
stop_continuous()
except ImportError:
pass
except Exception as e:
logger.warning("voice: stop_continuous failed during toggle off: %s", e)
# Clear TTS so it can be toggled independently later; silence live speech.
os.environ["HERMES_VOICE_TTS"] = "0"
_tts_stream_stop(user_barge=False)
_tts_lease_async("tui:voice-tts", False)
return _ok(
rid,
{
"enabled": enabled,
"record_key": _voice_record_key(),
"tts": _voice_tts_enabled(),
"stop_hint": stop_hint,
},
)
return _ok(rid, {"enabled": enabled, "record_key": _voice_record_key(), "tts": _voice_tts_enabled(),
"stop_hint": stop_hint})
def _voice_toggle_tts(rid, params: dict) -> dict:
@@ -907,63 +828,46 @@ def _(rid, params: dict) -> dict:
try:
global _voice_event_sid, _voice_wake_owner
if action == "start":
if not _voice_mode_enabled():
return _err(rid, 4015, "voice mode is off — enable with /voice on")
with _voice_sid_lock:
_voice_event_sid = params.get("session_id") or _voice_event_sid
from hermes_cli.voice import start_continuous
# Busy probe holds the no-speech counter during long agent turns.
# Safe to re-register every start; older wrappers lack the setter.
try:
from hermes_cli.voice import set_voice_busy_probe
set_voice_busy_probe(_any_session_running)
except Exception:
pass
# Shape-safe: malformed voice YAML falls back to documented defaults.
voice_cfg = _voice_cfg_dict()
safe_threshold = _voice_cfg_number(voice_cfg.get("silence_threshold"), 200)
safe_duration = _voice_cfg_number(voice_cfg.get("silence_duration"), 3.0)
# max_recording_seconds: explicit numeric <= 0 disables the cap (0.0).
max_rec = _voice_cfg_number(voice_cfg.get("max_recording_seconds"), 120.0)
safe_max_rec = max_rec if max_rec > 0 else 0.0
# Hand the mic to STT if the wake detector holds it; resume on a
# terminal capture event so wake-triggered and manual captures coexist.
try:
from tools.wake_word import pause_listening
wake_paused = pause_listening(owner=transport)
except Exception:
wake_paused = False
if wake_paused:
with _voice_sid_lock:
_voice_wake_owner = transport
started = start_continuous(
on_transcript=_vr_on_transcript, on_status=_vr_on_status,
on_silent_limit=_vr_on_silent, silence_threshold=safe_threshold,
silence_duration=safe_duration, auto_restart=False,
max_recording_seconds=safe_max_rec, on_stop_phrase=_vr_on_stop_phrase,
)
if started is False:
_resume_voice_wake()
return _ok(rid, {"status": "busy"})
return _ok(rid, {"status": "recording"})
# action == "stop"
if action == "start" and not _voice_mode_enabled():
return _err(rid, 4015, "voice mode is off — enable with /voice on")
with _voice_sid_lock:
_voice_event_sid = params.get("session_id") or _voice_event_sid
if action == "stop":
from hermes_cli.voice import stop_continuous
stop_continuous(force_transcribe=True)
_resume_voice_wake()
return _ok(rid, {"status": "stopped"})
from hermes_cli.voice import stop_continuous
stop_continuous(force_transcribe=True)
_resume_voice_wake()
return _ok(rid, {"status": "stopped"})
from hermes_cli.voice import start_continuous
# Busy probe holds the no-speech counter during long agent turns.
# Safe to re-register every start; older wrappers lack the setter.
with contextlib.suppress(Exception):
from hermes_cli.voice import set_voice_busy_probe
set_voice_busy_probe(_any_session_running)
# Shape-safe: malformed voice YAML falls back to documented defaults.
# max_recording_seconds: explicit numeric <= 0 disables the cap (0.0).
voice_cfg = _voice_cfg_dict()
max_rec = _voice_cfg_number(voice_cfg.get("max_recording_seconds"), 120.0)
# Hand the mic to STT if the wake detector holds it; resume on a terminal
# capture event so wake-triggered and manual captures coexist.
try:
from tools.wake_word import pause_listening
wake_paused = pause_listening(owner=transport)
except Exception:
wake_paused = False
if wake_paused:
with _voice_sid_lock:
_voice_wake_owner = transport
started = start_continuous(
on_transcript=_vr_on_transcript, on_status=_vr_on_status, on_silent_limit=_vr_on_silent,
silence_threshold=_voice_cfg_number(voice_cfg.get("silence_threshold"), 200),
silence_duration=_voice_cfg_number(voice_cfg.get("silence_duration"), 3.0), auto_restart=False,
max_recording_seconds=max_rec if max_rec > 0 else 0.0, on_stop_phrase=_vr_on_stop_phrase,
)
if started is False:
_resume_voice_wake()
return _ok(rid, {"status": "busy"})
return _ok(rid, {"status": "recording"})
except Exception as e:
if wake_paused or action == "stop":
_resume_voice_wake()
@@ -981,10 +885,7 @@ def _(rid, params: dict) -> dict:
# Import check up front so a missing voice module returns 5026 instead
# of failing silently in the thread.
import hermes_cli.voice # noqa: F401
threading.Thread(
target=_speak_text_with_barge, args=(text,), daemon=True
).start()
threading.Thread(target=_speak_text_with_barge, args=(text,), daemon=True).start()
return _ok(rid, {"status": "speaking"})
except ImportError:
return _err(rid, 5026, "voice module not available")