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

refactor(tui_gateway): tighten W5 modules (host_supervisor, agent_callbacks, model_switch, change_watcher, compute_host_bridge, browser, hosted-room adapters, git_probe, method_ctx, bot_relay) 3037->2680 LOC, zero behavior change

Browse Source
This commit is contained in:
Teknium
2026-09-03 02:28:12 -07:00
parent e1dc1ff22d
commit e7e287cdd8
11 changed files with 493 additions and 850 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
tui_gateway/agent_callbacks.py
View File

@@ -1,9 +1,6 @@
"""Agent callback wiring: child-session live mirror, per-session agent callbacks,
personality overlay, background/preview agent kwargs, agent reset.
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
"""Agent callback wiring: child-session live mirror, per-session agent callbacks, personality
overlay, background/preview agent kwargs, agent reset. Bodies are rebound onto server.py's
globals at install time (method_ctx.bind_module), so they reference server.py globals bare."""
from __future__ import annotations
@@ -12,18 +9,18 @@ import threading
from .method_ctx import bind_module
# Child-session live mirror: a delegated child's activity reaches the gateway only
# as relayed ``subagent.*`` events on the PARENT sid, so a window opened on the
# child's own session would sit silent until the run persists. Translate them into
# the native stream events emitted on the CHILD sid (write_json routes by sid).
# Child-session live mirror: a delegated child's activity reaches the gateway only as
# relayed ``subagent.*`` events on the PARENT sid; translate them into native stream
# events on the CHILD sid (write_json routes by sid) so its own window is not silent.
_child_mirrors: dict[str, dict] = {}
_child_mirrors_lock = threading.Lock()
# Child session ids with a run in flight (refreshed per relayed event, popped on
# complete) so a lazy watch resume reports running=true during a silent long tool.
# Child sids with a run in flight (refreshed per relayed event, popped on complete) so a
# lazy watch resume reports running=true during a silent long tool.
_active_child_runs: dict[str, float] = {}
# Anything quiet this long lost its completion event (callback raised, parent
# crashed) — don't pin "running".
# Anything quiet this long lost its completion event — don't pin "running".
_CHILD_RUN_STALE_S = 3600.0
_CHILD_DELTA_EVENTS = {"subagent.thinking": "reasoning.delta", "subagent.text": "message.delta",
"subagent.start": "message.delta"}
def _child_run_active(child_key: str) -> bool:
@@ -35,15 +32,13 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
child_key = str(payload.get("child_session_id") or "")
if not child_key:
return
# Liveness registry first: accurate with no window open, so one opened mid-run
# immediately knows the child is busy.
# Liveness registry first: accurate with no window open (one opened mid-run knows busy).
if event_type == "subagent.complete":
_active_child_runs.pop(child_key, None)
else:
_active_child_runs[child_key] = time.time()
# Mirror only into a live watch session NOT upgraded to a full agent: an
# upgraded one owns a real native stream and mirroring would interleave two
# turns on one sid. Either way drop state so a reopened window starts fresh.
# Mirror only into a live watch session NOT upgraded to a full agent (an upgraded one owns
# a real native stream). Either way drop state so a reopened window starts fresh.
live = _find_live_session_by_key(child_key)
if live is None or live[1].get("agent") is not None:
with _child_mirrors_lock:
@@ -51,19 +46,16 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
return
csid = live[0]
text = str(payload.get("text") or "")
# thinking/text/start (the child's goal, as a one-time header) are plain deltas.
delta = {"subagent.thinking": "reasoning.delta", "subagent.text": "message.delta",
"subagent.start": "message.delta"}
with _child_mirrors_lock:
st = _child_mirrors.setdefault(child_key, {"seq": 0, "open_tool": None, "started": False})
if not st["started"]:
st["started"] = True
_emit("message.start", csid)
if event_type in delta:
# thinking/text/start (the child's goal, as a one-time header) are plain deltas.
if event_type in _CHILD_DELTA_EVENTS:
if text:
if event_type == "subagent.start":
text = f"{text}\n"
_emit(delta[event_type], csid, {"text": text})
_emit(_CHILD_DELTA_EVENTS[event_type], csid,
{"text": f"{text}\n" if event_type == "subagent.start" else text})
return
if event_type not in ("subagent.tool", "subagent.complete"):
return
@@ -85,11 +77,11 @@ def _mirror_subagent_to_child(event_type: str, payload: dict) -> None:
def _agent_cbs(sid: str) -> dict:
def _read_block(event: str, timeout: int):
# read_terminal / read_preview (desktop GUI): blocking bridge like clarify; the
# preview read gets longer since a URL tab extracts text from a live page.
# read_terminal / read_preview (desktop GUI): blocking bridge like clarify; the preview
# read gets longer since a URL tab extracts text from a live page.
return lambda start=None, count=None: _block(
event, sid, {k: v for k, v in (("start", start), ("count", count)) if v is not None}, timeout=timeout
)
event, sid, {k: v for k, v in (("start", start), ("count", count)) if v is not None},
timeout=timeout)
callbacks = {
"tool_start_callback": lambda tc_id, name, args: _on_tool_start(sid, tc_id, name, args),
@@ -98,49 +90,42 @@ def _agent_cbs(sid: str) -> dict:
sid, event_type, name, preview, args, **kwargs),
"tool_gen_callback": lambda name: _tool_progress_enabled(sid) and _emit("tool.generating", sid, {"name": name}),
"thinking_callback": lambda text: _emit("thinking.delta", sid, {"text": text}),
# Affection reaction (ily / <3 / good bot) → hearts; core-detected so TUI and desktop share it.
# Affection reaction (ily / <3 / good bot) → hearts; core-detected so TUI/desktop share it.
"reaction_callback": lambda kind: _emit("reaction", sid, {"kind": kind}),
"reasoning_callback": lambda text: _emit(
"reasoning.delta", sid, {"text": text, **({"verbose": True} if _session_verbose(sid) else {})}
),
"reasoning.delta", sid, {"text": text, **({"verbose": True} if _session_verbose(sid) else {})}),
"status_callback": lambda kind, text=None: _status_update(sid, str(kind), None if text is None else str(text)),
# Credits/notice spine: AgentNotice → notification.show; recovery clear → notification.clear.
# Credits/notice spine: AgentNotice → notification.show; recovery → notification.clear.
"notice_callback": lambda n: _emit(
"notification.show", sid,
{"text": n.text, "level": n.level, "kind": n.kind, "ttl_ms": n.ttl_ms, "key": n.key, "id": n.id},
),
{"text": n.text, "level": n.level, "kind": n.kind, "ttl_ms": n.ttl_ms, "key": n.key, "id": n.id}),
"notice_clear_callback": lambda key: _emit("notification.clear", sid, {"key": key}),
"clarify_callback": lambda q, c, multi_select=False, questions=None: (
_clarify_block(sid, q, c, multi_select=multi_select, questions=questions)),
"read_terminal_callback": _read_block("terminal.read.request", 30),
"read_preview_callback": _read_block("preview.read.request", 45),
# drive_preview / annotate_preview (desktop GUI): renderer drives the preview webview and
# answers with outcome + refreshed element inventory; same budget as the preview read it ends with.
# drive_preview / annotate_preview (desktop GUI): same budget as the preview read it ends with.
"drive_preview_callback": lambda payload: _block("preview.act.request", sid, dict(payload), timeout=45),
# read_window_below (desktop GUI): main process enumerates native windows.
"read_window_below_callback": lambda: _block("window.read.request", sid, {}, timeout=30),
# setup_mcp (desktop GUI): consent card + install/enable/OAuth. Long timeout on purpose (typing
# an API key, browser OAuth); like clarify, timeout returns "unanswered" and a late answer is tolerated.
# setup_mcp (desktop GUI): consent card + install/enable/OAuth; long timeout on purpose
# (typing an API key, browser OAuth) and, like clarify, a late answer is tolerated.
"setup_mcp_callback": lambda server, action, reason: _block(
"mcp.setup.request", sid, {"server": server, "action": action, "reason": reason}, timeout=600
),
"mcp.setup.request", sid, {"server": server, "action": action, "reason": reason}, timeout=600),
# tour (desktop GUI): renderer drives driver.js and answers tour.respond.
"tour_callback": lambda payload: _tour_request(sid, payload)}
# Interim assistant commentary (text alongside tool calls). Gated on
# display.interim_assistant_messages (default true); _run_prompt_submit overwrites
# it per turn and clears it in its finally so a stale closure can't fire.
# Interim assistant commentary (text alongside tool calls), gated on display.interim_assistant_
# messages; _run_prompt_submit overwrites it per turn and clears it so a stale closure can't fire.
if _load_interim_assistant_messages():
callbacks["interim_assistant_callback"] = lambda text, *, already_streamed=False: _emit(
"message.interim", sid, {"text": str(text), "already_streamed": bool(already_streamed)})
return callbacks
def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
"""Intentional workspace move from the project_* tools: re-anchor the live
session's cwd and push session.info so the desktop follows. This is the ONLY
auto-cwd path — driven by an explicit tool call, never a terminal `cd`."""
"""Intentional workspace move from the project_* tools: re-anchor the live session's cwd
and push session.info. The ONLY auto-cwd path — an explicit tool call, never a `cd`."""
if not path:
return
# task_id is the durable session_key; _sessions (and desktop event routing) key by sid.
@@ -149,24 +134,19 @@ def _apply_project_workspace(task_id: str, path: str, _name: str = "") -> None:
sid, session = (key, _sessions[key]) if key in _sessions else next(
((s, c) for s, c in _sessions.items()
if c.get("session_key") == key or getattr(c.get("agent"), "session_id", None) == key),
("", None),
)
if session is None:
return
("", None))
resolved = os.path.abspath(os.path.expanduser(str(path)))
if not os.path.isdir(resolved):
if session is None or not os.path.isdir(resolved):
return
session["cwd"] = resolved
session["explicit_cwd"] = True
session["cwd_from_settle"] = False # explicit switch supersedes a settle-adopted cwd
# explicit switch supersedes a settle-adopted cwd
session.update(cwd=resolved, explicit_cwd=True, cwd_from_settle=False)
_register_session_cwd(session)
_persist_session_cwd_and_schedule_git_meta(session, resolved)
try:
agent = session.get("agent")
info = _session_info(agent, session) if agent is not None else {
"cwd": resolved, "branch": _git_branch_for_cwd(resolved),
"project": _project_info_for_cwd(resolved), "lazy": True,
}
"project": _project_info_for_cwd(resolved), "lazy": True}
_emit("session.info", sid, info)
except Exception:
logger.debug("failed to emit session.info after project workspace move", exc_info=True)
@@ -176,19 +156,17 @@ def _wire_callbacks(sid: str):
from tools.terminal_tool import set_sudo_password_callback
from tools.skills_tool import set_secret_capture_callback
from tools.project_tools import set_project_workspace_callback
set_sudo_password_callback(lambda: _block("sudo.request", sid, {}, timeout=120))
set_project_workspace_callback(_apply_project_workspace)
def secret_cb(env_var, prompt, metadata=None):
pl = {"prompt": prompt, "env_var": env_var}
if metadata:
pl["metadata"] = metadata
pl = {"prompt": prompt, "env_var": env_var, **({"metadata": metadata} if metadata else {})}
val = _block("secret.request", sid, pl)
if not val:
return {"success": True, "stored_as": env_var, "validated": False, "skipped": True, "message": "skipped"}
from hermes_cli.config import save_env_value_secure
return {**save_env_value_secure(env_var, val), "skipped": False, "message": "ok"}
set_sudo_password_callback(lambda: _block("sudo.request", sid, {}, timeout=120))
set_project_workspace_callback(_apply_project_workspace)
set_secret_capture_callback(secret_cb)
@@ -199,12 +177,10 @@ def _available_personalities(cfg: dict | None = None) -> dict:
def _validate_personality(value: str, cfg: dict | None = None) -> tuple[str, str]:
"""Resolve a requested personality to (name, prompt) or raise ValueError. Same
contract as hermes_cli.personality.resolve_personality, but goes through the
module-level _available_personalities so tests keep a single patch point."""
"""(name, prompt) for a requested personality or ValueError; like resolve_personality but
via the module-level _available_personalities so tests keep a single patch point."""
from hermes_cli.personality import normalize_personality_name, render_personality_prompt
name = normalize_personality_name(value)
if not name:
if not (name := normalize_personality_name(value)):
return "", ""
personalities = _available_personalities(cfg)
if name not in personalities:
@@ -221,16 +197,13 @@ def _prompt_text(value) -> str:
def _apply_personality_to_session(
sid: str, session: dict, new_prompt: str, personality: str = "") -> tuple[bool, dict | None]:
"""Apply a personality change to a live session without resetting history: the
ephemeral system prompt is updated in place (appended at API-call time, so
prompt-cache hits survive) plus a pivot marker so the model stops pattern-matching
its earlier tone. Returns (history_reset=False, info)."""
"""Apply a personality change without resetting history: the ephemeral system prompt is
updated in place (appended at API-call time, so prompt-cache hits survive) plus a pivot
marker so the model stops pattern-matching its earlier tone. Returns (False, info)."""
if not session:
return False, None
session["personality"] = personality
agent = session.get("agent")
if not agent:
if not (agent := session.get("agent")):
return False, None
agent.ephemeral_system_prompt = new_prompt or None
marker = (
@@ -239,12 +212,10 @@ def _apply_personality_to_session(
f"accordingly: {new_prompt}]"
if new_prompt else
"[System: The user has cleared the personality overlay. "
"From this point forward, respond in your normal default style.]"
)
# Like the model-switch marker: role=user so strict providers accept it
# mid-conversation, but `display_kind` keeps it out of the
# `truncate_before_user_ordinal` addressing space (untagged, every rewind would
# land one turn early and `replace_messages` hard-delete the difference).
"From this point forward, respond in your normal default style.]")
# Like the model-switch marker: role=user so strict providers accept it mid-conversation,
# but `display_kind` keeps it out of the `truncate_before_user_ordinal` addressing space
# (untagged, every rewind would land one turn early and hard-delete the difference).
with session["history_lock"]:
session["history"].append({"role": "user", "content": marker, "display_kind": "personality_switch"})
session["history_version"] = int(session.get("history_version", 0)) + 1
@@ -256,8 +227,7 @@ def _apply_personality_to_session(
def _cfg_max_turns(cfg: dict, default: int) -> int:
from hermes_cli.config import resolve_turn_limit as _resolve_turn_limit
# Env override wins; resolve_turn_limit makes "none"/"unlimited"/0 first-class spellings.
env_val = os.environ.get("HERMES_TUI_MAX_TURNS")
if env_val:
if env_val := os.environ.get("HERMES_TUI_MAX_TURNS"):
return _resolve_turn_limit(env_val, default=default)
raw = (cfg.get("agent") or {}).get("max_turns")
if raw is None:
@@ -267,12 +237,7 @@ def _cfg_max_turns(cfg: dict, default: int) -> int:
def _parse_tui_skills_env() -> list[str]:
raw = os.environ.get("HERMES_TUI_SKILLS", "")
skills: list[str] = []
for part in raw.replace("\n", ",").split(","):
item = part.strip()
if item and item not in skills:
skills.append(item)
return skills
return list(dict.fromkeys(p.strip() for p in raw.replace("\n", ",").split(",") if p.strip()))
def _load_fallback_model():
@@ -282,40 +247,33 @@ def _load_fallback_model():
return get_fallback_chain(_load_cfg())
def _agent_fallback_model(agent):
"""Return an agent's fallback chain without rehydrating deliberately empty chains."""
if hasattr(agent, "_fallback_chain"):
return agent._fallback_chain or []
return agent._fallback_model if hasattr(agent, "_fallback_model") else _load_fallback_model()
def _background_agent_kwargs(agent, task_id: str) -> dict:
cfg = _load_cfg()
def g(name, default=None):
return getattr(agent, name, default)
kwargs = {k: g(k) or None for k in (
"base_url", "api_key", "provider", "api_mode", "acp_command", "acp_args",
"ephemeral_system_prompt")}
kwargs.update({k: g(k) for k in (
"providers_allowed", "providers_ignored", "providers_order", "provider_sort",
"provider_data_collection", "openrouter_min_coding_score")})
kwargs.update(
model=g("model") or _resolve_model(),
max_iterations=_cfg_max_turns(cfg, 25),
# Detached tasks declare platform="tui" (no UI sid for renderer-routed
# events), so resolve toolsets against it — never GUI schema they can't use.
enabled_toolsets=g("enabled_toolsets") or _load_enabled_toolsets("tui"),
quiet_mode=True, verbose_logging=False,
provider_require_parameters=g("provider_require_parameters", False),
session_id=task_id,
reasoning_config=g("reasoning_config") or _load_reasoning_config(str(g("model", "") or "")),
service_tier=g("service_tier") or _load_service_tier(),
request_overrides=dict(g("request_overrides", {}) or {}),
platform="tui", session_db=_get_db(), fallback_model=_agent_fallback_model(agent),
)
return kwargs
# Don't rehydrate a deliberately empty fallback chain.
if hasattr(agent, "_fallback_chain"):
fallback = agent._fallback_chain or []
else:
fallback = (agent._fallback_model if hasattr(agent, "_fallback_model")
else _load_fallback_model())
# Detached tasks declare platform="tui" (no UI sid for renderer-routed events), so resolve
# toolsets against it — never GUI schema they can't use.
return {
**{k: g(k) or None for k in ("base_url", "api_key", "provider", "api_mode", "acp_command",
"acp_args", "ephemeral_system_prompt")},
**{k: g(k) for k in ("providers_allowed", "providers_ignored", "providers_order", "provider_sort",
"provider_data_collection", "openrouter_min_coding_score")},
"model": g("model") or _resolve_model(), "max_iterations": _cfg_max_turns(cfg, 25),
"enabled_toolsets": g("enabled_toolsets") or _load_enabled_toolsets("tui"),
"quiet_mode": True, "verbose_logging": False,
"provider_require_parameters": g("provider_require_parameters", False), "session_id": task_id,
"reasoning_config": g("reasoning_config") or _load_reasoning_config(str(g("model", "") or "")),
"service_tier": g("service_tier") or _load_service_tier(),
"request_overrides": dict(g("request_overrides", {}) or {}),
"platform": "tui", "session_db": _get_db(), "fallback_model": fallback}
def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:
@@ -323,13 +281,10 @@ def _ephemeral_preview_agent_kwargs(agent, task_id: str) -> dict:
"enabled_toolsets": ["terminal", "file"], "session_db": None, "skip_memory": True}
_PREVIEW_HISTORY_ROLES = ("user", "assistant", "tool", "system")
def _preview_restart_history(session: dict, max_messages: int = 24, max_tool_chars: int = 1200) -> list[dict]:
"""Distill recent parent history for the ephemeral preview-restart agent (else it
guesses app/server/cwd/port from the bare URL). Keeps the last ``max_messages``
(always back to the last user turn); tool results truncated to ``max_tool_chars``."""
"""Distill recent parent history for the ephemeral preview-restart agent (else it guesses
app/cwd/port from the bare URL): last ``max_messages`` back to the last user turn, tool
results truncated to ``max_tool_chars``."""
try:
with session["history_lock"]:
history = list(session.get("history") or [])
@@ -337,14 +292,13 @@ def _preview_restart_history(session: dict, max_messages: int = 24, max_tool_cha
history = list(session.get("history") or [])
if not history:
return []
last_user = next((i for i in range(len(history) - 1, -1, -1) if history[i].get("role") == "user"), None)
start = max(0, len(history) - max_messages)
for idx in range(len(history) - 1, -1, -1):
if history[idx].get("role") == "user":
start = min(start, idx)
break
if last_user is not None:
start = min(start, last_user)
trimmed: list[dict] = []
for msg in history[start:]:
if not isinstance(msg, dict) or msg.get("role") not in _PREVIEW_HISTORY_ROLES:
if not isinstance(msg, dict) or msg.get("role") not in ("user", "assistant", "tool", "system"):
continue
copy = {k: v for k, v in msg.items() if k != "reasoning"}
content = copy.get("content")
@@ -358,7 +312,7 @@ def _preview_tool_result_preview(name: str, result: str) -> str:
try:
data = json.loads(result)
except Exception:
return ""
data = None
if not isinstance(data, dict):
return ""
if name == "terminal":
@@ -375,8 +329,7 @@ def _preview_restart_callbacks(parent: str, task_id: str) -> dict:
started_at: dict[str, float] = {}
def progress(message: str, level: str = "info") -> None:
text = str(message or "").strip()
if text:
if text := str(message or "").strip():
_emit("preview.restart.progress", parent, {"task_id": task_id, "level": level, "text": text})
def tool_start(tool_call_id: str, name: str, args: dict) -> None:
@@ -391,27 +344,22 @@ def _preview_restart_callbacks(parent: str, task_id: str) -> dict:
progress(summary + (f"\n{output}" if output else ""))
def tool_progress(event_type: str, name: str | None = None, preview: str | None = None, **_kwargs) -> None:
if preview:
progress(str(preview))
elif name:
progress(f"{event_type.replace('.', ' ')}: {name}")
if preview or name:
progress(str(preview) if preview else f"{event_type.replace('.', ' ')}: {name}")
return {
"tool_start_callback": tool_start, "tool_complete_callback": tool_complete,
"tool_progress_callback": tool_progress,
"tool_gen_callback": lambda name: progress(f"Preparing {name}"),
"status_callback": lambda kind, text=None: progress(text if text is not None else kind),
}
"status_callback": lambda kind, text=None: progress(text if text is not None else kind)}
def _reset_session_agent(sid: str, session: dict) -> dict:
tokens = _set_session_context(session["session_key"])
try:
# /new is a full conversation boundary: session-scoped runtime overrides
# (/model, /reasoning, /fast) do NOT carry forward — the fresh agent
# re-derives them from config.yaml, and the pins are cleared so a rebuild
# can't resurrect them. Global process state is never touched (see the
# cross-session-contamination note in _apply_model_switch).
# /new is a full conversation boundary: session-scoped runtime overrides (/model,
# /reasoning, /fast) do NOT carry forward and the pins are cleared so a rebuild can't
# resurrect them. Global process state is never touched (see _apply_model_switch).
for k in ("model_override", "create_reasoning_override", "create_service_tier_override", "one_turn_model_restore"):
session.pop(k, None)
new_agent = _make_agent(
@@ -425,8 +373,7 @@ def _reset_session_agent(sid: str, session: dict) -> dict:
queued_prompt=None,
_queued_prompt_generation=int(session.get("_queued_prompt_generation", 0)) + 1,
edit_snapshots={}, image_counter=0, running=False, show_reasoning=_load_show_reasoning(),
tool_progress_mode=_load_tool_progress_mode(), tool_started_at={},
)
tool_progress_mode=_load_tool_progress_mode(), tool_started_at={})
session.pop("queued_prompts", None)
with session["history_lock"]:
session["history"] = []

110
tui_gateway/change_watcher.py
View File

@@ -1,8 +1,6 @@
"""Skin + config-change watcher: signatures for skin/pet/cron/sessions/platforms/pairing/bot-relay state and the broadcast loop that pushes *.changed events.
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
"""Skin + config-change watcher: on-disk signatures for skin/pet/cron/sessions/platforms/
pairing/bot-relay state and the broadcast loop that pushes *.changed events. Bodies are
rebound onto server.py's globals at install time (method_ctx.bind_module)."""
from __future__ import annotations
@@ -14,12 +12,11 @@ _registry = HandlerRegistry()
def resolve_skin() -> dict:
try:
from hermes_cli.skin_engine import init_skin_from_config, get_active_skin
init_skin_from_config(_load_cfg())
skin = get_active_skin()
# light/dark are paired palettes: the TUI prefers the block matching terminal polarity.
return {
"name": skin.name, "colors": skin.colors,
# Paired palettes: the TUI prefers the block matching terminal polarity.
"light_colors": skin.light_colors, "dark_colors": skin.dark_colors,
"branding": skin.branding, "banner_logo": skin.banner_logo,
"banner_hero": skin.banner_hero, "tool_prefix": skin.tool_prefix,
@@ -47,10 +44,13 @@ def _watcher_mtime_ns(path: Path):
return None
def _home_mtime_ns(*parts: str):
return _watcher_mtime_ns(_watcher_home().joinpath(*parts))
def _newest_mtime_ns(paths) -> int | None:
"""Max ``st_mtime_ns`` across ``paths`` (unstat-able ignored); None when none stat'ed."""
mtimes = (_watcher_mtime_ns(p) for p in paths)
return max((m for m in mtimes if m is not None), default=None)
return max((m for m in map(_watcher_mtime_ns, paths) if m is not None), default=None)
def _skin_sig() -> tuple[str, float | None]:
@@ -58,10 +58,9 @@ def _skin_sig() -> tuple[str, float | None]:
their name moves; a user skin's mtime lets an in-place color edit repaint too."""
name = str((_load_cfg().get("display") or {}).get("skin") or "default")
try:
mtime: float | None = (_watcher_home() / "skins" / f"{name}.yaml").stat().st_mtime
return name, (_watcher_home() / "skins" / f"{name}.yaml").stat().st_mtime
except OSError:
mtime = None
return name, mtime
return name, None
def _note_skin_broadcast() -> None:
@@ -75,14 +74,11 @@ def _broadcast_skin_if_changed() -> None:
"""Emit ``skin.changed`` when the active skin moved, via the SAME live path as
``/skin`` so every surface repaints. The check is a dict lookup + one stat."""
global _last_skin_sig
try:
sig = _skin_sig()
except Exception:
return
if sig == _last_skin_sig:
return
_last_skin_sig = sig
with contextlib.suppress(Exception):
sig = _skin_sig()
if sig == _last_skin_sig:
return
_last_skin_sig = sig
_broadcast_global_event("skin.changed", resolve_skin())
@@ -99,55 +95,39 @@ def _pet_sig() -> tuple:
if not pet_cfg or not is_truthy_value(pet_cfg.get("enabled"), default=False):
return ("off",)
try:
active = _active_pet()
if not active:
return ("off",)
pet, scale = active
return (pet.slug, _pet_sheet_revision(pet.spritesheet), scale)
if active := _active_pet():
pet, scale = active
return (pet.slug, _pet_sheet_revision(pet.spritesheet), scale)
except Exception: # noqa: BLE001 - cosmetic, never break the watcher
return ("off",)
pass
return ("off",)
def _pet_changed_payload() -> dict:
"""``pet.info.meta``-shaped payload so the renderer can decide whether to refetch sprites."""
try:
active = _active_pet()
if not active:
return {"enabled": False}
pet, scale = active
return {
"enabled": True, "slug": pet.slug, "displayName": pet.display_name, "scale": scale,
"spritesheetRevision": _pet_sheet_revision(pet.spritesheet)}
if active := _active_pet():
pet, scale = active
return {"enabled": True, "slug": pet.slug, "displayName": pet.display_name,
"scale": scale, "spritesheetRevision": _pet_sheet_revision(pet.spritesheet)}
except Exception: # noqa: BLE001 - cosmetic, never break the watcher
return {"enabled": False}
def _cron_sig():
"""mtime of cron/jobs.json — moves on edits AND scheduler tick bookkeeping."""
return _watcher_mtime_ns(_watcher_home() / "cron" / "jobs.json")
pass
return {"enabled": False}
def _sessions_sig():
"""Newest mtime across state.db + WAL: the one thing messaging-gateway turns and
cron runs (which never touch this gateway's transports) all move. Served sibling
profile homes are probed too, else a routed profile's Bot Chat never refreshes."""
"""Newest mtime across state.db + WAL: the one thing messaging-gateway turns and cron runs
all move. Served sibling profile homes are probed too, else a routed Bot Chat never refreshes."""
return _newest_mtime_ns(
root / name
for root in (_watcher_home(), *_served_profile_homes)
for name in ("state.db", "state.db-wal")
)
def _platforms_sig():
"""mtime of gateway_state.json — where the messaging gateway persists platform
connect/disconnect/health, i.e. the Messaging page's status-changed signal."""
return _watcher_mtime_ns(_watcher_home() / "gateway_state.json")
for name in ("state.db", "state.db-wal"))
def _pairing_sig():
"""Newest mtime across every profile's pairing ledgers (legacy ``pairing/`` and
``platforms/pairing/``). Pending codes are written by the gateway process, so the
files are the only shared signal; a pairing request moves nothing in gateway_state.json."""
``platforms/pairing/``): the gateway process writes pending codes, so the files are the only
shared signal (a pairing request moves nothing in gateway_state.json)."""
home = _watcher_home()
roots = [home / "pairing", home / "platforms" / "pairing"]
with contextlib.suppress(OSError):
@@ -168,29 +148,27 @@ _bot_relay_outbox_seen = 0
def _bot_relay_outbox_sig():
"""Newest mtime across pending bot-relay outbox envelopes (monotone). Written by
the AGENT process, so the files are the only shared signal; the Desktop reacts
to ``bot_relay.outbox.pending`` with an immediate debounced drain."""
"""Newest mtime across pending bot-relay outbox envelopes (monotone). Written by the AGENT
process, so the files are the only shared signal; the Desktop reacts with a debounced drain."""
global _bot_relay_outbox_seen
home = _watcher_home()
root = home.parent.parent if home.parent.name == "profiles" else home
newest = 0
with contextlib.suppress(OSError):
for entry in (root / "bot_relay" / "outbox").iterdir():
if entry.name.endswith(".json"):
newest = max(newest, _watcher_mtime_ns(entry) or 0)
if newest > _bot_relay_outbox_seen:
_bot_relay_outbox_seen = newest
_bot_relay_outbox_seen = max(_bot_relay_outbox_seen, _watcher_mtime_ns(entry) or 0)
return _bot_relay_outbox_seen or None
# event → (check interval, signature fn, payload fn). Signatures are stat-cheap;
# the interval keeps pricier probes (pet resolves the sheet off disk) off the 0.5s tick.
# event → (check interval, signature fn, payload fn). Signatures are stat-cheap; the interval
# keeps pricier probes (pet resolves the sheet off disk) off the 0.5s tick. cron/jobs.json
# moves on edits AND scheduler ticks; gateway_state.json is where the messaging gateway
# persists platform connect/disconnect/health (the Messaging page's status signal).
_CHANGE_WATCHES: dict[str, tuple[float, Any, Any]] = {
"pet.changed": (2.0, _pet_sig, _pet_changed_payload),
"cron.changed": (1.0, _cron_sig, lambda: {}),
"cron.changed": (1.0, lambda: _home_mtime_ns("cron", "jobs.json"), lambda: {}),
"sessions.changed": (0.5, _sessions_sig, lambda: {}),
"platforms.changed": (2.0, _platforms_sig, lambda: {}),
"platforms.changed": (2.0, lambda: _home_mtime_ns("gateway_state.json"), lambda: {}),
"pairing.changed": (2.0, _pairing_sig, lambda: {}),
# 1s so a queued DM envelope reaches the Desktop's push-triggered drain fast.
"bot_relay.outbox.pending": (1.0, _bot_relay_outbox_sig, lambda: {})}
@@ -220,9 +198,9 @@ def _broadcast_watched_changes(now: float | None = None) -> None:
if event not in _change_sigs:
_change_sigs[event] = sig
continue
floor = _CHANGE_BROADCAST_FLOOR_S.get(event, 0.0)
if sig == _change_sigs[event]:
continue
floor = _CHANGE_BROADCAST_FLOOR_S.get(event, 0.0)
if floor and now - _change_broadcast_at.get(event, -floor) < floor:
continue # floored: old signature stays so it re-fires when the window opens
_change_sigs[event] = sig
@@ -235,9 +213,8 @@ _skin_watcher_started = False
def _ensure_skin_watcher() -> None:
"""Start the process's one change watcher (named for its original skin-only
duty): cheap on-disk signatures → broadcast events, so skin/pet/cron/cross-process
changes go live everywhere within seconds without client polling. Idempotent."""
"""Start the process's one change watcher (named for its original skin-only duty): cheap
on-disk signatures → broadcast events, so changes go live without client polling. Idempotent."""
global _skin_watcher_started
if _skin_watcher_started:
return
@@ -249,7 +226,6 @@ def _ensure_skin_watcher() -> None:
time.sleep(0.5)
_broadcast_skin_if_changed()
_broadcast_watched_changes()
threading.Thread(target=_loop, name="hermes-change-watcher", daemon=True).start()

104
tui_gateway/compute_host_bridge.py
View File

@@ -19,23 +19,18 @@ _compute_host_supervisor_lock = threading.Lock()
_COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS = 630.0
def _inside_compute_host_child() -> bool:
return os.environ.get("HERMES_COMPUTE_HOST_CHILD") == "1"
def _turn_isolation_enabled(cfg: dict | None = None) -> bool:
if _inside_compute_host_child():
if os.environ.get("HERMES_COMPUTE_HOST_CHILD") == "1":
return False
return bool((cfg or _load_dashboard_process_isolation_config()).get("turn_isolation"))
def _session_uses_compute_host(session: dict, cfg: dict | None = None) -> bool:
if not _turn_isolation_enabled(cfg):
return False
# Routes lazy sessions whose AIAgent was never built in-process; already-built
# sessions keep the in-process path unless a prior isolated turn marked host ownership.
return bool(session.get("_compute_host_active")) or (
session.get("agent") is None and session.get("agent_ready") is not None)
return _turn_isolation_enabled(cfg) and (
bool(session.get("_compute_host_active"))
or (session.get("agent") is None and session.get("agent_ready") is not None))
def _get_compute_host_supervisor(cfg: dict | None = None):
@@ -57,8 +52,7 @@ def _compute_host_turn_frame(
with session["history_lock"]:
history = list(session.get("history", []))
history_version = int(session.get("history_version", 0))
attached_images = list(
image_paths if image_paths is not None else session.get("attached_images", []))
attached_images = list(image_paths if image_paths is not None else session.get("attached_images", []))
return {
"type": "turn.start", "sid": sid, "request_id": rid,
"session_key": session.get("session_key") or sid, "text": text,
@@ -93,8 +87,8 @@ def _compute_host_adopt_frame_meta(session: dict, frame: dict) -> None:
session["session_key"] = str(frame.get("session_key"))
if frame.get("history_version") is not None:
with contextlib.suppress(Exception):
session["history_version"] = max(
int(session.get("history_version", 0)), int(frame.get("history_version") or 0))
session["history_version"] = max(int(session.get("history_version", 0)),
int(frame.get("history_version") or 0))
def _relay_compute_host_rpc(message: dict) -> bool:
@@ -106,7 +100,7 @@ def _relay_compute_host_rpc(message: dict) -> bool:
payload = params.get("payload")
request_id = payload.get("request_id") if isinstance(payload, dict) else None
if session is not None and request_id:
with session.get("history_lock", threading.Lock()):
with _history_lock(session):
if kind == "clarify.request":
session["_compute_host_pending_clarify"] = dict(payload)
elif _pending_clarify_matches(session, request_id):
@@ -114,6 +108,10 @@ def _relay_compute_host_rpc(message: dict) -> bool:
return write_json(message)
def _history_lock(session: dict):
return session.get("history_lock", threading.Lock())
def _pending_clarify_matches(session: dict, request_id) -> bool:
"""Whether ``session``'s mirrored pending clarify is ``request_id``. Caller holds
history_lock."""
@@ -123,32 +121,26 @@ def _pending_clarify_matches(session: dict, request_id) -> bool:
def _compute_host_clarify_session(request_id: str) -> tuple[str, dict] | None:
"""Find the parent mirror for one host-owned clarify request."""
if not request_id:
return None
for sid, session in list(_sessions.items()):
with session.get("history_lock", threading.Lock()):
for sid, session in list(_sessions.items()) if request_id else ():
with _history_lock(session):
if _pending_clarify_matches(session, request_id):
return sid, session
return None
def _update_compute_host_clarify_snapshot(
sid: str, session: dict, params: dict, result: dict) -> None:
def _update_compute_host_clarify_snapshot(sid: str, session: dict, params: dict, result: dict) -> None:
"""Keep reconnect snapshots accurate while a batch clarify is answered."""
request_id = str(params.get("request_id") or "")
with session.get("history_lock", threading.Lock()):
question_id = str(params.get("question_id") or "")
with _history_lock(session):
if not _pending_clarify_matches(session, request_id):
return
pending = session["_compute_host_pending_clarify"]
expired = result.get("status") == "expired"
if expired or not result.get("remaining") and not params.get("question_id"):
if result.get("status") == "expired" or not result.get("remaining") and not question_id:
session.pop("_compute_host_pending_clarify", None)
return
question_id = str(params.get("question_id") or "")
if question_id and isinstance(result.get("remaining"), list):
answers = dict(pending.get("answers") or {})
answers[question_id] = str(params.get("answer") or "")
pending["answers"] = answers
elif question_id and isinstance(result.get("remaining"), list):
pending["answers"] = {**(pending.get("answers") or {}),
question_id: str(params.get("answer") or "")}
if not result["remaining"]:
session.pop("_compute_host_pending_clarify", None)
@@ -156,11 +148,9 @@ def _update_compute_host_clarify_snapshot(
def _respond_compute_host_clarify(rid: str, params: dict) -> dict | None:
"""Proxy a clarify answer into the process that owns its pending Event."""
located = _compute_host_clarify_session(str(params.get("request_id") or ""))
if located is None:
if located is None or not _session_uses_compute_host(located[1]):
return None
sid, session = located
if not _session_uses_compute_host(session):
return None
try:
ack = _get_compute_host_supervisor().respond(sid, params)
except Exception as exc:
@@ -172,9 +162,8 @@ def _respond_compute_host_clarify(rid: str, params: dict) -> dict | None:
return _err(rid, 5019, "compute-host clarify response returned an invalid response")
if "error" in response:
error = response["error"] if isinstance(response["error"], dict) else {}
return _err(
rid, int(error.get("code") or 5000),
str(error.get("message") or "clarify response failed"))
return _err(rid, int(error.get("code") or 5000),
str(error.get("message") or "clarify response failed"))
result = response.get("result")
if not isinstance(result, dict):
return _err(rid, 5019, "compute-host clarify response returned an invalid result")
@@ -187,7 +176,7 @@ def _apply_compute_host_metadata_mirror(session: dict, frame: dict | None) -> No
writer of live agent/history state, and UI reads must not build a second agent."""
if not isinstance(frame, dict):
return
with session.get("history_lock", threading.Lock()):
with _history_lock(session):
_compute_host_adopt_frame_meta(session, frame)
if frame.get("message_count") is not None:
with contextlib.suppress(Exception):
@@ -219,17 +208,15 @@ def _submit_prompt_to_compute_host(
rid: str, sid: str, session: dict, text: Any, image_paths: list[str] | None = None,
queued_prompt_generation: int | None = None, display_kind: str | None = None) -> dict:
cfg = _load_dashboard_process_isolation_config()
frame = _compute_host_turn_frame(
rid, sid, session, text, image_paths=image_paths,
queued_prompt_generation=queued_prompt_generation, display_kind=display_kind)
frame = _compute_host_turn_frame(rid, sid, session, text, image_paths=image_paths,
queued_prompt_generation=queued_prompt_generation,
display_kind=display_kind)
def _complete(done: dict) -> None:
# submit_turn reports a synchronous pipe failure via the callback before
# re-raising; leave the session untouched so prompt.submit can fail open
# to the in-process path without a duplicate terminal error.
if done.get("reason") == "send_failed":
return
_on_compute_host_turn_done(rid, sid, session, done)
# submit_turn reports a synchronous pipe failure via the callback before re-raising;
# leave the session untouched so prompt.submit can fail open to the in-process path.
if done.get("reason") != "send_failed":
_on_compute_host_turn_done(rid, sid, session, done)
try:
_get_compute_host_supervisor(cfg).submit_turn(frame, on_complete=_complete)
except Exception as exc:
@@ -267,29 +254,22 @@ def _compute_host_compress_wait_seconds(cfg: dict | None = None) -> float:
return float(min(max(ceiling + 30.0, 120.0), _COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS))
def _announce_compute_host_compress_done(sid: str, session: dict, ack: dict) -> None:
"""Mirror a compress ack and push the ``session.info`` + ``compacted`` edges the
in-process /compress path emits, so a client whose RPC wait expired still learns."""
_apply_compute_host_metadata_mirror(session, ack)
_emit("session.info", sid, _compute_host_session_info(session))
_status_update(sid, "compacted", "✓ Context compression complete")
def _adopt_late_compute_host_compress_ack(
sid: str, session: dict, ack: dict, *, route_name: str) -> None:
"""Adopt a compress ack that arrived after its RPC waiter answered ``pending``: the
only place the rotated session_key / history_version / mirror can land and the
client's only signal. A late ``control.error`` goes out via ``error``."""
def _adopt_late_compute_host_compress_ack(sid: str, session: dict, ack: dict, *, route_name: str) -> None:
"""Adopt a compress ack that arrived after its RPC waiter answered ``pending``: the only place
the rotated session_key / history_version / mirror can land and the client's only signal
(the same ``session.info`` + ``compacted`` edges the in-process /compress path emits). A late
``control.error`` goes out via ``error``."""
with _sessions_lock:
live = _sessions.get(sid)
if live is not session:
return
if _sessions.get(sid) is not session:
return
if not isinstance(ack, dict) or ack.get("type") in {"control.error", "error"}:
message = str((ack or {}).get("message") or f"compute-host {route_name} failed")
_emit("error", sid, {"message": f"compression failed: {message}"})
_status_update(sid, "ready")
return
_announce_compute_host_compress_done(sid, session, ack)
_apply_compute_host_metadata_mirror(session, ack)
_emit("session.info", sid, _compute_host_session_info(session))
_status_update(sid, "compacted", "✓ Context compression complete")
def register(server) -> None:

57
tui_gateway/git_probe.py
View File

@@ -1,11 +1,8 @@
"""Git working-tree probing for the gateway: run git, resolve repo roots, fold linked worktrees.
Probing runs where the gateway runs (covers remote backends). Roots go through a thread-safe
single-flight cache so concurrent identical probes share one ``git`` spawn. Positives are cached
for the process lifetime; negatives (not a repo / deleted dir) only for ``_NEG_TTL`` —
``build_tree`` resolves a cwd once *per session*, so hundreds of non-git cwds would otherwise
re-spawn ``git`` on every sidebar open, while the TTL keeps a fresh ``git init`` re-probable.
"""
single-flight cache so concurrent identical probes share one ``git`` spawn: positives live for the
process, negatives (not a repo / deleted dir) for ``_NEG_TTL`` — hundreds of non-git session cwds
would otherwise re-spawn ``git`` on every sidebar open, while the TTL keeps ``git init`` re-probable."""
from __future__ import annotations
@@ -19,19 +16,14 @@ from hermes_cli._subprocess_compat import bounded_git_probe
_GIT_TIMEOUT = 1.5
_WARM_WORKERS = 8
# "Not a git repo" TTL: short enough that a fresh `git init` shows within seconds,
# long enough to collapse a tree build's hundreds of redundant probes.
_NEG_TTL = 30.0
_NEG_TTL = 30.0 # "not a git repo" TTL: a fresh `git init` shows within seconds
def run_git(cwd: str, *args: str) -> str:
"""``git -C <cwd> <args>`` → stripped stdout, or ``""`` on any failure.
``bounded_git_probe`` bounds post-kill cleanup on Windows — a plain ``subprocess.run(timeout)``
deadlocked Desktop readiness when a killed git left a suspended descendant holding the pipes.
"""
# `git -C` on a missing dir can only fail, at the price of a fork; deleted worktrees
# dominate a long session history's cwds, so the stat pays off.
"""``git -C <cwd> <args>`` → stripped stdout, or ``""`` on any failure. ``bounded_git_probe``
bounds post-kill cleanup on Windows (a killed git's suspended descendant held the pipes)."""
# A missing dir can only fail at the price of a fork; deleted worktrees dominate a long
# session history's cwds, so the stat pays off.
if not cwd or not os.path.isdir(cwd):
return ""
return bounded_git_probe(["git", "-C", cwd, *args], timeout=_GIT_TIMEOUT)
@@ -60,8 +52,7 @@ class _RootCache:
def resolve(self, key: str, probe) -> str:
while True:
with self._lock:
hit = self._roots.get(key)
if hit:
if hit := self._roots.get(key):
return hit
expiry = self._neg.get(key)
if expiry is not None:
@@ -72,8 +63,7 @@ class _RootCache:
leader = gate is None
if leader:
gate = self._inflight[key] = threading.Event()
if not leader:
# Another thread is probing this key — wait, then re-read.
if not leader: # another thread is probing this key — wait, then re-read
gate.wait(timeout=_GIT_TIMEOUT + 0.5)
continue
value = ""
@@ -100,21 +90,16 @@ def invalidate() -> None:
def repo_root(cwd: str) -> str:
"""Top-level git repo root for ``cwd`` (``""`` when not a repo)."""
if not cwd:
return ""
return _cache.resolve(cwd, lambda: run_git(cwd, "rev-parse", "--show-toplevel"))
return _cache.resolve(cwd, lambda: run_git(cwd, "rev-parse", "--show-toplevel")) if cwd else ""
def common_repo_root(cwd: str) -> str:
"""The MAIN (common) repo root for ``cwd``, folding linked worktrees.
``--show-toplevel`` returns a linked worktree's OWN root; the parent of the shared
``--git-common-dir`` is the one true root (fallback: the toplevel root). Normalized to git's
forward-slash spelling so it compares equal to :func:`repo_root` — with native ``\\`` on
Windows the main checkout was misread as a linked worktree and the sidebar rendered it twice.
"""
# Not a repo: nothing to fold. Checking the (warmed, negative-cached) toplevel first spares
# every non-repo cwd a second `git` spawn the parallel warm can't absorb.
"""The MAIN (common) repo root for ``cwd``, folding linked worktrees: ``--show-toplevel`` is a
linked worktree's OWN root; the parent of the shared ``--git-common-dir`` is the one true root
(fallback: toplevel). Normalized to git's forward-slash spelling so it compares equal to
:func:`repo_root` (native ``\\`` on Windows made the main checkout look like a worktree)."""
# Checking the (warmed, negative-cached) toplevel first spares every non-repo cwd a second
# `git` spawn the parallel warm can't absorb.
if not cwd or not repo_root(cwd):
return ""
@@ -142,10 +127,8 @@ def warm_roots(cwds: Iterable[str], max_workers: int = _WARM_WORKERS) -> None:
"""Pre-resolve many cwds' roots in parallel (bounded) so a cold first paint
doesn't serialize one git spawn per session cwd; results land in the cache."""
pending = sorted({(cwd or "").strip() for cwd in cwds} - {""})
if not pending:
return
if len(pending) == 1:
resolve(pending[0])
return
with ThreadPoolExecutor(max_workers=min(max_workers, len(pending))) as pool:
list(pool.map(resolve, pending))
elif pending:
with ThreadPoolExecutor(max_workers=min(max_workers, len(pending))) as pool:
list(pool.map(resolve, pending))

199
tui_gateway/host_supervisor.py
View File

@@ -23,7 +23,6 @@ from hermes_constants import get_hermes_home
from tools.environments.local import hermes_subprocess_env
logger = logging.getLogger(__name__)
_Thread = threading.Thread
MUTATOR_ROUTE_TABLE: dict[str, str] = {
"prompt.submit": "turn-path", "session.interrupt": "turn-path", "reload.mcp": "run-concurrent",
@@ -36,9 +35,8 @@ MUTATOR_ROUTE_TABLE: dict[str, str] = {
_REGISTRY_NAME = "dashboard-compute-host.json"
_RESPAWN_WINDOW_SECS = 300.0
_SHUTDOWN_TIMEOUT_SECS = 10.0
# Late control-ack handlers: a compress that outlives its RPC waiter can run for the
# full compression ceiling plus a stall-fallback retry, so keep registrations well
# past that — but bounded.
# Late control-ack handlers: a compress that outlives its RPC waiter can run for the full
# compression ceiling plus a stall-fallback retry, so keep registrations past that — bounded.
_LATE_CONTROL_TTL_SECS = 1800.0
_LATE_CONTROL_MAX = 64
# Host frames whose ``request_id`` resolves a pending/late control waiter.
@@ -64,17 +62,15 @@ def _repo_root() -> Path:
def _check_output(argv: list[str], **kwargs: Any) -> str:
"""Stripped stdout of a short subprocess, or ``""`` on any failure."""
try:
with contextlib.suppress(Exception):
return subprocess.check_output(
argv, text=True, encoding="utf-8", errors="replace", stderr=subprocess.DEVNULL,
timeout=2, **kwargs).strip()
except Exception:
return ""
return ""
def _build_sha() -> str:
"""Current checkout's HEAD sha, or ``"unknown"``. Shared with ``compute_host`` so
the hello handshake and the supervisor's expectation agree byte-for-byte."""
"""HEAD sha or ``"unknown"``; shared with ``compute_host`` so the hello handshake agrees."""
return _check_output(["git", "rev-parse", "HEAD"], cwd=str(_repo_root())) or "unknown"
@@ -91,21 +87,21 @@ def _pid_alive(pid: int) -> bool:
return False
try:
os.kill(pid, 0)
return True
except Exception as exc:
return isinstance(exc, PermissionError)
return True
def _signal_pid(pid: int, sig: int, label: str) -> bool:
"""Send ``sig``; False when the pid is gone or the signal failed (logged)."""
try:
os.kill(pid, sig)
return True
except ProcessLookupError:
return False
except Exception:
logger.debug("failed to %s compute host pid=%s", label, pid, exc_info=True)
return False
return True
def _pid_command(pid: int) -> str:
@@ -140,10 +136,9 @@ class HostSupervisor:
self.rpc_sink = rpc_sink or (lambda _obj: None)
self.respawn_max = max(0, int(respawn_max))
self.heartbeat_secs = max(1, int(heartbeat_secs))
self.expected_build_sha = (
expected_build_sha if expected_build_sha is not None else _build_sha())
self.expected_build_sha = _build_sha() if expected_build_sha is None else expected_build_sha
self.expected_hermes_home = (
expected_hermes_home if expected_hermes_home is not None else str(get_hermes_home()))
str(get_hermes_home()) if expected_hermes_home is None else expected_hermes_home)
self._lock = threading.RLock()
self._proc: subprocess.Popen[str] | None = None
self._hello_event = threading.Event()
@@ -153,9 +148,8 @@ class HostSupervisor:
self._restart_times: list[float] = []
self._pending_turns: dict[str, tuple[str, Callable[[dict], None] | None]] = {}
self._pending_controls: dict[str, queue.Queue[dict]] = {}
# request_id -> (registered_at, handler) for control waiters that timed out
# while their host work still runs; without it the eventual control.ack
# matched no queue and was silently dropped.
# request_id -> (registered_at, handler) for control waiters that timed out while their
# host work still runs, so the eventual control.ack is not silently dropped.
self._late_control_handlers: dict[str, tuple[float, Callable[[dict], None]]] = {}
self._stderr_tail: list[str] = []
self._last_progress_counter = 0
@@ -201,25 +195,24 @@ class HostSupervisor:
except FileNotFoundError:
return "none"
except Exception:
self._remove_registry()
return "invalid-registry"
data = None
try:
pid = int(data.get("host_pid") or 0)
pid = int((data or {}).get("host_pid") or 0)
except Exception:
pid = 0
if pid <= 0 or not _pid_alive(pid):
self._remove_registry()
return "not-running"
if not self._pid_matches_compute_host(pid):
# PID was reused by another process. Never signal it.
self._remove_registry()
return "pid-reuse-ignored"
self._terminate_pid(pid, timeout=_SHUTDOWN_TIMEOUT_SECS)
if data is None:
outcome = "invalid-registry"
elif pid <= 0 or not _pid_alive(pid):
outcome = "not-running"
elif not self._pid_matches_compute_host(pid):
outcome = "pid-reuse-ignored" # PID reused by another process: never signal it
else:
self._terminate_pid(pid, timeout=_SHUTDOWN_TIMEOUT_SECS)
outcome = "terminated"
self._remove_registry()
return "terminated"
return outcome
def submit_turn(
self, frame: dict[str, Any], *, on_complete: Callable[[dict], None] | None = None) -> str:
def submit_turn(self, frame: dict[str, Any], *, on_complete: Callable[[dict], None] | None = None) -> str:
self.start()
request_id = str(frame.get("request_id") or uuid.uuid4().hex)
sid = str(frame.get("sid") or "")
@@ -232,16 +225,15 @@ class HostSupervisor:
with self._lock:
self._pending_turns.pop(request_id, None)
if on_complete is not None:
on_complete({
"type": "turn.error", "sid": sid, "request_id": request_id,
"reason": "send_failed", "message": str(exc)})
on_complete({"type": "turn.error", "sid": sid, "request_id": request_id,
"reason": "send_failed", "message": str(exc)})
raise
return request_id
def interrupt(self, sid: str, *, request_id: str | None = None) -> None:
self.start()
self._send_frame({
"type": "interrupt", "sid": sid, "request_id": request_id or uuid.uuid4().hex})
self._send_frame(
{"type": "interrupt", "sid": sid, "request_id": request_id or uuid.uuid4().hex})
def _await_reply(self, frame: dict[str, Any], request_id: str, timeout: float) -> dict:
"""Send ``frame`` and block for the host reply carrying ``request_id``."""
@@ -263,28 +255,24 @@ class HostSupervisor:
return self._await_reply(frame, request_id, timeout)
def reload_mcp(self, sid: str, *, request_id: str | None = None) -> dict:
return self.control(
sid, route_name="reload.mcp", wait=True,
payload={
"type": "reload_mcp", "sid": sid, "request_id": request_id or uuid.uuid4().hex})
payload = {"type": "reload_mcp", "sid": sid, "request_id": request_id or uuid.uuid4().hex}
return self.control(sid, route_name="reload.mcp", wait=True, payload=payload)
def control(
self, sid: str, *, route_name: str, payload: dict[str, Any] | None = None,
wait: bool = True, timeout: float = 30.0, on_late_ack: Callable[[dict], None] | None = None,
) -> dict:
"""Send a control frame; with ``wait`` block up to ``timeout`` for its ack.
``on_late_ack`` (only with ``wait``) keeps the request adoptable after the waiter
gives up: the host's eventual ``control.ack``/``control.error``/``error`` fires it
once instead of being dropped (bounded by ``_LATE_CONTROL_TTL_SECS``/``_MAX``).
"""
"""Send a control frame; with ``wait`` block up to ``timeout`` for its ack. ``on_late_ack``
(only with ``wait``) keeps the request adoptable after the waiter gives up: the host's
eventual ``control.ack``/``control.error``/``error`` fires it once (bounded by
``_LATE_CONTROL_TTL_SECS``/``_MAX``) instead of being dropped."""
if route_name not in MUTATOR_ROUTE_TABLE:
raise ValueError(f"unclassified host mutator route: {route_name}")
self.start()
request_id = str((payload or {}).get("request_id") or uuid.uuid4().hex)
frame = {
"type": "control", **(payload or {}), "sid": sid, "route_name": route_name,
"request_id": request_id}
payload = payload or {}
request_id = str(payload.get("request_id") or uuid.uuid4().hex)
frame = {"type": "control", **payload, "sid": sid, "route_name": route_name,
"request_id": request_id}
if not wait:
self._send_frame(frame)
return {"status": "sent", "request_id": request_id}
@@ -295,13 +283,11 @@ class HostSupervisor:
self._register_late_control_handler(request_id, on_late_ack)
raise
def _register_late_control_handler(
self, request_id: str, handler: Callable[[dict], None]) -> None:
def _register_late_control_handler(self, request_id: str, handler: Callable[[dict], None]) -> None:
now = time.monotonic()
with self._lock:
handlers = self._late_control_handlers
expired = [r for r, (at, _cb) in handlers.items() if now - at > _LATE_CONTROL_TTL_SECS]
for rid in expired:
for rid in [r for r, (at, _cb) in handlers.items() if now - at > _LATE_CONTROL_TTL_SECS]:
handlers.pop(rid, None)
while len(handlers) >= _LATE_CONTROL_MAX:
handlers.pop(min(handlers, key=lambda rid: handlers[rid][0]), None)
@@ -314,11 +300,8 @@ class HostSupervisor:
if q is not None:
with contextlib.suppress(queue.Full):
q.put_nowait(frame)
return
if late is not None:
_call_logged(
late[1], frame,
f"compute host late control ack handler failed (request_id={request_id})")
elif late is not None:
_call_logged(late[1], frame, f"compute host late control ack handler failed (request_id={request_id})")
def _spawn_locked(self, *, reason: str) -> None:
if self._stopped_respawning:
@@ -331,18 +314,16 @@ class HostSupervisor:
env.setdefault("PYTHONPATH", root)
if root not in env["PYTHONPATH"].split(os.pathsep):
env["PYTHONPATH"] = root + os.pathsep + env["PYTHONPATH"]
# Lossy UTF-8 decode: a locale-mismatched byte must not raise inside the drain threads.
proc = subprocess.Popen(
self.argv, cwd=str(self.cwd), env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True,
# Lossy UTF-8 decode: a locale-mismatched byte must not raise inside the
# drain threads and kill the supervisor.
encoding="utf-8", errors="replace", bufsize=1, start_new_session=True)
stderr=subprocess.PIPE, text=True, encoding="utf-8", errors="replace", bufsize=1,
start_new_session=True)
self._proc = proc
for target, name in (
(self._drain_stdout, "compute-host-stdout"),
(self._drain_stderr, "compute-host-stderr"), (self._wait_for_exit, "compute-host-wait"),
):
_Thread(target=target, args=(proc,), name=name, daemon=True).start()
for target, name in ((self._drain_stdout, "compute-host-stdout"),
(self._drain_stderr, "compute-host-stderr"),
(self._wait_for_exit, "compute-host-wait")):
threading.Thread(target=target, args=(proc,), name=name, daemon=True).start()
if not self._hello_event.wait(timeout=10.0):
self._terminate_process(proc)
raise RuntimeError(f"compute host did not send hello; stderr={self._stderr_tail[-5:]}")
@@ -366,10 +347,9 @@ class HostSupervisor:
def _persist_registry(self) -> None:
self.registry_path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.registry_path.with_suffix(self.registry_path.suffix + ".tmp")
payload = {
"host_pid": self.pid, "boot_id": self._hello.get("boot_id") or "",
"build_sha": self._hello.get("build_sha") or "", "started_at": time.time(),
"argv": self.argv}
payload = {"host_pid": self.pid, "boot_id": self._hello.get("boot_id") or "",
"build_sha": self._hello.get("build_sha") or "", "started_at": time.time(),
"argv": self.argv}
tmp.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8")
tmp.replace(self.registry_path)
@@ -401,45 +381,29 @@ class HostSupervisor:
def _drain_stderr(self, proc: subprocess.Popen[str]) -> None:
assert proc.stderr is not None
for raw in proc.stderr:
text = raw.rstrip("\n")
if text:
if text := raw.rstrip("\n"):
self._stderr_tail = (self._stderr_tail + [text])[-80:]
logger.warning("compute host stderr: %s", text)
def _handle_host_frame(self, frame: dict[str, Any]) -> None:
ftype = str(frame.get("type") or "")
if ftype in _CONTROL_REPLY_TYPES or (ftype == "error" and frame.get("request_id")):
self._deliver_control_frame(str(frame.get("request_id") or ""), frame)
return
handler = self._HOST_FRAME_HANDLERS.get(ftype)
if handler is not None:
getattr(self, handler)(frame)
# host frame ``type`` -> handler method name (see also _CONTROL_REPLY_TYPES).
_HOST_FRAME_HANDLERS: dict[str, str] = {
"hello": "_on_hello", "hb": "_on_heartbeat", "rpc": "_on_rpc", "turn.end": "_complete_turn",
"turn.error": "_complete_turn"}
def _on_hello(self, frame: dict[str, Any]) -> None:
self._hello = dict(frame)
self._hello_event.set()
def _on_heartbeat(self, frame: dict[str, Any]) -> None:
self._last_progress_counter = int(
frame.get("progress_counter") or self._last_progress_counter)
logger.debug("compute host heartbeat: %s", frame)
def _on_rpc(self, frame: dict[str, Any]) -> None:
message = frame.get("message")
if isinstance(message, dict):
self.rpc_sink(message)
def _complete_turn(self, frame: dict[str, Any]) -> None:
request_id = str(frame.get("request_id") or "")
with self._lock:
pending = self._pending_turns.pop(request_id, None)
if pending is not None and pending[1] is not None:
_call_logged(pending[1], frame, "compute host turn completion callback failed")
if ftype in _CONTROL_REPLY_TYPES or (ftype == "error" and request_id):
self._deliver_control_frame(request_id, frame)
elif ftype == "hello":
self._hello = dict(frame)
self._hello_event.set()
elif ftype == "hb":
self._last_progress_counter = int(frame.get("progress_counter") or self._last_progress_counter)
logger.debug("compute host heartbeat: %s", frame)
elif ftype == "rpc":
if isinstance(frame.get("message"), dict):
self.rpc_sink(frame["message"])
elif ftype in ("turn.end", "turn.error"):
with self._lock:
pending = self._pending_turns.pop(request_id, None)
if pending is not None and pending[1] is not None:
_call_logged(pending[1], frame, "compute host turn completion callback failed")
def _wait_for_exit(self, proc: subprocess.Popen[str]) -> None:
code = proc.wait()
@@ -459,14 +423,13 @@ class HostSupervisor:
self._pending_turns = {}
failure = {"reason": reason, "message": message}
for request_id, (sid, cb) in pending.items():
self.rpc_sink({
"jsonrpc": "2.0", "method": "event",
"params": {"type": "error", "session_id": sid, "payload": dict(failure)}})
self.rpc_sink({"jsonrpc": "2.0", "method": "event",
"params": {"type": "error", "session_id": sid, "payload": dict(failure)}})
if cb is not None:
frame = {"type": "turn.error", "sid": sid, "request_id": request_id, **failure}
_call_logged(cb, frame, "compute host error callback failed")
# A crashed host never emits the late acks timed-out control waiters still
# expect; fail them too so the client's "still running" notice can't hang.
# A crashed host never emits the late acks timed-out control waiters still expect; fail
# them too so the client's "still running" notice can't hang.
with self._lock:
late = self._late_control_handlers
self._late_control_handlers = {}
@@ -496,18 +459,19 @@ class HostSupervisor:
self._spawn_locked(reason="crash")
except Exception:
logger.exception("compute host respawn failed")
_Thread(target=_respawn, name="compute-host-respawn", daemon=True).start()
threading.Thread(target=_respawn, name="compute-host-respawn", daemon=True).start()
_pid_matches_compute_host = staticmethod(is_compute_host_identity)
def _terminate_pid(self, pid: int, *, timeout: float = _SHUTDOWN_TIMEOUT_SECS) -> None:
if not _signal_pid(pid, signal.SIGTERM, "SIGTERM"):
return
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if not _pid_alive(pid):
while _pid_alive(pid):
if time.monotonic() >= deadline:
_signal_pid(pid, signal.SIGKILL, "SIGKILL")
return
time.sleep(0.05)
_signal_pid(pid, signal.SIGKILL, "SIGKILL")
def _terminate_process(self, proc: subprocess.Popen[str]) -> None:
if proc.poll() is not None:
@@ -516,10 +480,9 @@ class HostSupervisor:
proc.terminate()
proc.wait(timeout=_SHUTDOWN_TIMEOUT_SECS)
return
with contextlib.suppress(Exception):
proc.kill()
with contextlib.suppress(Exception):
proc.wait(timeout=2)
for step in (proc.kill, lambda: proc.wait(timeout=2)):
with contextlib.suppress(Exception):
step()
__all__ = ["MUTATOR_ROUTE_TABLE", "HostSupervisor", "append_log_record", "is_compute_host_identity"]

137
tui_gateway/hosted_room_peer_transport.py
View File

@@ -1,9 +1,6 @@
"""Peer-backed session transport for one hosted-room member task.
This adapter implements :class:`InternalSessionRPC` without using canonical
Bot Chat. The remote client must resolve a hidden ``Group: <room_id>`` session
with ``source=bot_room`` and verify the scoped grant at admission.
"""
"""Peer-backed session transport for one hosted-room member task: implements
:class:`InternalSessionRPC` without canonical Bot Chat. The remote client must resolve a hidden
``Group: <room_id>`` session with ``source=bot_room`` and verify the scoped grant at admission."""
from __future__ import annotations
@@ -24,27 +21,16 @@ class HostedRoomPeerClient(Protocol):
"""Authenticated client for a target gateway's narrow room-member API."""
def bind_room_scope(self, **scope: Any) -> None: ...
def prepare(
self, *, room_id: str, profile: str, source: str, grant: str, create: bool,
expected_session_id: str | None = None,
) -> Mapping[str, Any] | None: ...
def prepare(self, *, room_id: str, profile: str, source: str, grant: str, create: bool,
expected_session_id: str | None = None) -> Mapping[str, Any] | None: ...
def dispatch(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any]: ...
def history(
self, *, room_id: str, profile: str, session_id: str, grant: str
) -> Sequence[Mapping[str, Any]]: ...
def status(
self, *, room_id: str, profile: str, session_id: str, grant: str
) -> Mapping[str, Any]: ...
def history(self, *, room_id: str, profile: str, session_id: str, grant: str
) -> Sequence[Mapping[str, Any]]: ...
def status(self, *, room_id: str, profile: str, session_id: str, grant: str
) -> Mapping[str, Any]: ...
def stop(self, *, dispatch: Mapping[str, Any], grant: str) -> Mapping[str, Any] | None: ...
def stop_receipt(
self, *, task_id: str, execution_generation: int, grant: str
) -> Mapping[str, Any] | None: ...
def stop_receipt(self, *, task_id: str, execution_generation: int, grant: str
) -> Mapping[str, Any] | None: ...
@dataclass(frozen=True)
@@ -69,10 +55,8 @@ class FailoverHostedRoomPeerClient:
raise ValueError("RoomLink candidates must target one installation")
if reprobe_interval_seconds <= 0:
raise ValueError("reprobe_interval_seconds must be positive")
self.candidates = tuple(candidates)
self._active = 0
self.candidates, self._active, self.clock = tuple(candidates), 0, clock
self.reprobe_interval_seconds = float(reprobe_interval_seconds)
self.clock = clock
self._last_primary_probe = 0.0
@property
@@ -80,11 +64,9 @@ class FailoverHostedRoomPeerClient:
return self.candidates[self._active]
def _call(self, method: str, **kwargs):
"""Try the active link (re-probing the primary after a cooldown), then the rest.
Ambiguous or non-retryable failures propagate immediately: failing over
after an ambiguous dispatch could run the same task twice.
"""
"""Try the active link (re-probing the primary after a cooldown), then the rest. Ambiguous
or non-retryable failures propagate: failing over after an ambiguous dispatch could run
the same task twice."""
now = self.clock()
order = [self._active]
if self._active != 0 and now - self._last_primary_probe >= self.reprobe_interval_seconds:
@@ -102,29 +84,20 @@ class FailoverHostedRoomPeerClient:
continue
self._active = index
return result
if last_error is not None:
raise last_error
raise RuntimeError("no RoomLink candidate was attempted")
raise last_error if last_error is not None else RuntimeError("no RoomLink candidate was attempted")
def prepare(self, **kwargs):
return self._call("prepare", **kwargs)
def _delegate(method: str):
def call(self, **kwargs):
return self._call(method, **kwargs)
call.__name__ = method
return call
def dispatch(self, **kwargs):
return self._call("dispatch", **kwargs)
def history(self, **kwargs):
return self._call("history", **kwargs)
def status(self, **kwargs):
return self._call("status", **kwargs)
def stop(self, **kwargs):
return self._call("stop", **kwargs)
prepare, dispatch, history, status, stop = map(_delegate, ("prepare", "dispatch", "history", "status", "stop"))
del _delegate
def bind_room_scope(self, **kwargs):
for candidate in self.candidates:
bind = getattr(candidate.client, "bind_room_scope", None)
if callable(bind):
if callable(bind := getattr(candidate.client, "bind_room_scope", None)):
bind(**kwargs)
@@ -149,23 +122,15 @@ def build_member_dispatch(
trace_id: str) -> HostedMemberDispatch:
"""Build the fully fenced member dispatch shared by submit and recovery."""
return HostedMemberDispatch.from_mapping({
"protocol_version": PROTOCOL_VERSION,
"room_id": room_id,
"home_install_id": route.home_install_id,
"authority_gateway_id": binding.gateway_id,
"authority_epoch": binding.authority_epoch,
"member_id": route.member_id,
"target_install_id": route.target_install_id,
"target_profile": target_profile,
"task_id": task_id,
"execution_generation": execution_generation,
"source_event_seq": source_event_seq,
"cancellation_scope_id": route.cancellation_scope_id,
"prompt": prompt,
"prompt_digest": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
"protocol_version": PROTOCOL_VERSION, "room_id": room_id,
"home_install_id": route.home_install_id, "authority_gateway_id": binding.gateway_id,
"authority_epoch": binding.authority_epoch, "member_id": route.member_id,
"target_install_id": route.target_install_id, "target_profile": target_profile,
"task_id": task_id, "execution_generation": execution_generation,
"source_event_seq": source_event_seq, "cancellation_scope_id": route.cancellation_scope_id,
"prompt": prompt, "prompt_digest": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
"capability_digest": route.capability_digest,
"execution_policy_digest": route.execution_policy_digest,
"trace_id": trace_id})
"execution_policy_digest": route.execution_policy_digest, "trace_id": trace_id})
class PeerHostedRoomTransport(InternalSessionRPC):
@@ -185,8 +150,7 @@ class PeerHostedRoomTransport(InternalSessionRPC):
self.execution_generation = execution_generation
self._session_id: str | None = None
self._dispatch: HostedMemberDispatch | None = None
bind_scope = getattr(self.client, "bind_room_scope", None)
if callable(bind_scope):
if callable(bind_scope := getattr(self.client, "bind_room_scope", None)):
bind_scope(
room_id=binding.room_id, home_install_id=route.home_install_id,
authority_gateway_id=binding.gateway_id, authority_epoch=binding.authority_epoch,
@@ -205,26 +169,23 @@ class PeerHostedRoomTransport(InternalSessionRPC):
"""Room id + grant keyword arguments shared by every scoped client call."""
return {"room_id": self.binding.room_id, "grant": self.route.grant, **extra}
def _prepare(self, *, profile: str, source: str, create: bool, **extra):
return self.client.prepare(
**self._scoped(profile=profile, source=source, create=create, **extra))
def _prepare(self, *, profile: str, source: str, create: bool, title: str | None = None, **extra):
"""Validate coordinates, then the scoped ``prepare`` call."""
self._validate_coordinates(profile=profile, source=source, title=title)
return self.client.prepare(**self._scoped(profile=profile, source=source, create=create, **extra))
def resolve_exact(self, *, profile: str, title: str, source: str) -> Mapping[str, Any] | None:
self._validate_coordinates(profile=profile, source=source, title=title)
return self._prepare(profile=profile, source=source, create=False)
return self._prepare(profile=profile, source=source, create=False, title=title)
def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]:
self._validate_coordinates(profile=profile, source=source, title=title)
session = self._prepare(profile=profile, source=source, create=True)
session = self._prepare(profile=profile, source=source, create=True, title=title)
if session is None:
raise RuntimeError("peer did not create the room session")
self._session_id = str(session.get("session_id") or session.get("id") or "")
return session
def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
self._validate_coordinates(profile=profile, source=source)
session = self._prepare(
profile=profile, source=source, create=False, expected_session_id=session_id)
session = self._prepare(profile=profile, source=source, create=False, expected_session_id=session_id)
if session is None:
raise RuntimeError("peer room session is unavailable")
self._session_id = session_id
@@ -262,15 +223,13 @@ class PeerHostedRoomTransport(InternalSessionRPC):
) -> Mapping[str, Any] | None:
self._validate_coordinates(profile=profile, source=source)
dispatch = self._dispatch
if dispatch is None:
if (
self.task_id != expected_task_id
or not self.execution_generation
or not hasattr(self.client, "stop_receipt")):
if dispatch is not None:
if dispatch.task_id != expected_task_id:
return None
return self.client.stop_receipt(
task_id=expected_task_id, execution_generation=self.execution_generation,
grant=self.route.grant)
if dispatch.task_id != expected_task_id:
return self.client.stop(dispatch=dispatch.as_mapping(), grant=self.route.grant)
if (self.task_id != expected_task_id or not self.execution_generation
or not hasattr(self.client, "stop_receipt")):
return None
return self.client.stop(dispatch=dispatch.as_mapping(), grant=self.route.grant)
return self.client.stop_receipt(
task_id=expected_task_id, execution_generation=self.execution_generation,
grant=self.route.grant)

89
tui_gateway/hosted_room_server_rpc.py
View File

@@ -1,10 +1,6 @@
"""In-process session adapter for the hosted room driver.
The room worker must not depend on a Desktop/WebSocket transport, but it should
still use the same session handlers as every other TUI/Desktop turn. This
adapter calls the installed handler registry directly and keeps the extra
task proof as an in-process-only Python object that JSON clients cannot forge.
"""
"""In-process session adapter for the hosted room driver: the room worker uses the same
installed session handlers as every TUI/Desktop turn (no WebSocket transport), passing the
task proof as an in-process-only Python object that JSON clients cannot forge."""
from __future__ import annotations
@@ -57,50 +53,32 @@ class HostedRoomServerRPC:
if not isinstance(rows, list) or not rows or not isinstance(rows[0], dict):
return None
row = rows[0]
return {
"session_id": row.get("resolved_id") or row.get("id"),
"title": row.get("title") or title}
return {"session_id": row.get("resolved_id") or row.get("id"),
"title": row.get("title") or title}
def create(self, *, profile: str, title: str, source: str) -> Mapping[str, Any]:
return self._call(
"session.create",
{
"profile": profile,
"title": title,
"source": source,
"hidden": True,
"room_plumbing": True,
"follow_profile_config": True,
"close_on_disconnect": False})
return self._call("session.create", {
"profile": profile, "title": title, "source": source, "hidden": True,
"room_plumbing": True, "follow_profile_config": True, "close_on_disconnect": False})
def resume(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
return self._call(
"session.resume",
{"profile": profile, "session_id": session_id, "omit_messages": True, "source": source})
return self._call("session.resume", {
"profile": profile, "session_id": session_id, "omit_messages": True, "source": source})
def submit(
self, *, profile: str, session_id: str, prompt: str, source: str, task: state.TaskIdentity,
execution_generation: int, on_terminal: Callable[[Mapping[str, Any]], None],
) -> Mapping[str, Any]:
try:
return self._call(
"prompt.submit",
{
"profile": profile,
"session_id": session_id,
"text": prompt,
"source": source,
"_hosted_task": {
"room_id": task.room_id,
"task_id": task.task_id,
"thread_id": task.thread_id,
"turn_id": task.turn_id,
"execution_generation": execution_generation},
"_hosted_terminal_callback": on_terminal})
return self._call("prompt.submit", {
"profile": profile, "session_id": session_id, "text": prompt, "source": source,
"_hosted_task": {
"room_id": task.room_id, "task_id": task.task_id, "thread_id": task.thread_id,
"turn_id": task.turn_id, "execution_generation": execution_generation},
"_hosted_terminal_callback": on_terminal})
except HostedRoomSessionError as exc:
# In-process prompt.submit error envelopes are returned before the
# background turn is admitted. Preserve that proof so the driver
# can defer or requeue without waiting out an ambiguity lease.
# In-process prompt.submit error envelopes come back before the background turn is
# admitted; keep that proof so the driver can defer/requeue without an ambiguity lease.
exc.not_admitted = True
raise
@@ -115,10 +93,8 @@ class HostedRoomServerRPC:
record = self.server._sessions.get(session_id)
if record is not None:
return record
for candidate in self.server._sessions.values():
if str(candidate.get("session_key") or "") == session_id:
return candidate
return None
return next((c for c in self.server._sessions.values()
if str(c.get("session_key") or "") == session_id), None)
def info(self, *, profile: str, session_id: str, source: str) -> Mapping[str, Any]:
del profile, source
@@ -130,32 +106,23 @@ class HostedRoomServerRPC:
return {"active": bool(record.get("running")), "task_id": None}
with lock:
task = record.get("_hosted_room_task")
result = {
"active": bool(record.get("running")),
"task_id": task.get("task_id") if isinstance(task, dict) else None}
result = {"active": bool(record.get("running")),
"task_id": task.get("task_id") if isinstance(task, dict) else None}
pending_reader = getattr(self.server, "_pending_approval_request_payload", None)
pending = (
pending_reader(str(record.get("session_key") or ""))
if callable(pending_reader)
else None)
if pending:
if callable(pending_reader) and (pending := pending_reader(str(record.get("session_key") or ""))):
result["status"] = "waiting_for_approval"
result["pending_approval"] = pending
return result
def approve(self, *, session_id: str, request_id: str, choice: str) -> Mapping[str, Any]:
"""Resolve one exact local room approval without broad policy changes."""
return self._call(
"approval.respond",
{"session_id": session_id, "request_id": request_id, "choice": choice, "all": False})
return self._call("approval.respond", {
"session_id": session_id, "request_id": request_id, "choice": choice, "all": False})
def interrupt(
self, *, profile: str, session_id: str, source: str, expected_task_id: str
) -> Mapping[str, Any] | None:
del source
return self._call(
"session.interrupt",
{
"profile": profile,
"session_id": session_id,
"expected_hosted_task_id": expected_task_id})
return self._call("session.interrupt", {
"profile": profile, "session_id": session_id,
"expected_hosted_task_id": expected_task_id})

54
tui_gateway/method_ctx.py
View File

@@ -1,12 +1,9 @@
"""Seam for the server.py handler/helper split.
server.py's JSON-RPC handlers and helpers close over its module globals (``_sessions``,
``_ok``, ``_err``, ...). Split modules define their code normally and server.py calls
:func:`bind_module` at the end of its own import, once every global exists: bodies are
re-created with ``types.FunctionType`` against server.py's namespace, so they stay
byte-identical and ``global X`` statements keep mutating server.py state. No import
cycle: split modules never import server at module level — server passes itself in.
"""
"""Seam for the server.py handler/helper split. server.py's JSON-RPC handlers and helpers close
over its module globals (``_sessions``, ``_ok``, ``_err``, ...). Split modules define their code
normally and server.py calls :func:`bind_module` at the end of its own import, once every global
exists: bodies are re-created with ``types.FunctionType`` against server.py's namespace, so they
stay byte-identical and ``global X`` keeps mutating server.py state. No import cycle: split
modules never import server at module level — server passes itself in."""
import contextlib
import types
@@ -27,18 +24,15 @@ def rebind(fn, g: dict, _seen=None):
return contextlib.contextmanager(rebind(wrapped, g, _seen))
closure = fn.__closure__
if closure:
cells = []
for cell in closure:
def _cell(cell):
try:
val = cell.cell_contents
except ValueError: # empty cell
cells.append(cell)
continue
return cell
if isinstance(val, types.FunctionType) and val.__module__ == fn.__module__:
cells.append(types.CellType(rebind(val, g, _seen)))
else:
cells.append(cell)
closure = tuple(cells)
return types.CellType(rebind(val, g, _seen))
return cell
closure = tuple(_cell(c) for c in closure)
real = types.FunctionType(fn.__code__, g, fn.__name__, fn.__defaults__, closure)
real.__kwdefaults__ = fn.__kwdefaults__
real.__doc__ = fn.__doc__
@@ -55,11 +49,9 @@ class HandlerRegistry:
def method(self, name: str):
"""Drop-in for server.py's ``@method`` decorator (defers registration)."""
def dec(fn):
self._pending.append((name, fn))
return fn
return dec
def profile_scoped(self, fn):
@@ -82,16 +74,11 @@ _PLUMBING = {"HandlerRegistry", "method", "_profile_scoped", "register", "rebind
def bind_module(module_globals: dict, server, *, skip=()) -> None:
"""Publish everything a split module defines onto ``server``, rebound to its globals.
``module_globals`` is the caller's ``globals()`` (not ``sys.modules[__name__]``: tests
that ``patch.dict(sys.modules)`` around the server import drop the submodule entries
while the package attribute survives, so a re-import would KeyError). Functions are
rebound; classes get their methods rebound in place; dispatch tables (dict/tuple/list
holding this module's functions) get their values rebound; other values (constants,
``global``-mutated state seeds) are copied as-is. Imported modules/functions, dunders
and registry plumbing are skipped, so no hand-maintained export list is needed.
Finally the module's ``_registry`` (if any) installs its @method handlers.
"""
``module_globals`` is the caller's ``globals()`` (not ``sys.modules[__name__]``: tests that
``patch.dict(sys.modules)`` around the server import drop the submodule entries). Functions
are rebound; classes get their methods rebound in place; dispatch tables (dict/tuple/list of
this module's functions) get their values rebound; other values are copied as-is. Imported
modules/functions, dunders and registry plumbing are skipped; finally ``_registry`` installs."""
g = vars(server)
mod_name = module_globals["__name__"]
seen: dict = {}
@@ -104,18 +91,15 @@ def bind_module(module_globals: dict, server, *, skip=()) -> None:
return rebind(v, g, seen)
if isinstance(v, dict):
return {k: _rebind_in(x) for k, x in v.items()}
if isinstance(v, (tuple, list)):
return type(v)(_rebind_in(x) for x in v)
return v
return type(v)(_rebind_in(x) for x in v) if isinstance(v, (tuple, list)) else v
def _has_own_fn(v):
items = v.values() if isinstance(v, dict) else v if isinstance(v, (tuple, list)) else None
return _own_fn(v) if items is None else any(_has_own_fn(x) for x in items)
for name, obj in list(module_globals.items()):
if name.startswith("__") or name in _PLUMBING or name in skip:
continue
if isinstance(obj, (types.ModuleType, HandlerRegistry)):
if (name.startswith("__") or name in _PLUMBING or name in skip
or isinstance(obj, (types.ModuleType, HandlerRegistry))):
continue
if isinstance(obj, types.FunctionType):
if obj.__module__ == mod_name:

93
tui_gateway/methods_bot_relay.py
View File

@@ -1,12 +1,10 @@
"""Bot-relay JSON-RPC handlers — the gateway side of cross-connection A2A.
Connections ARE the peer set: the Desktop owns every gateway socket and relays between them via
four doors on EACH gateway: ``roster.sync`` (push OTHER connections' agents so ``message_agent``
resolves them), ``outbox.drain`` (collect envelopes queued here for other connections), ``deliver``
(one-turn Bot Chat delivery on the TARGET gateway, returns the reply), ``reply`` (write the
reply/error back on the SENDER gateway for its waiter). Plumbing: ``tools/bot_relay.py``.
Handlers are rebound onto server.py's globals (method_ctx.py) and reference ``_ok``/``_err`` bare.
"""
"""Bot-relay JSON-RPC handlers — the gateway side of cross-connection A2A. Connections ARE the
peer set: the Desktop owns every gateway socket and relays between them via four doors on EACH
gateway: ``roster.sync`` (push OTHER connections' agents so ``message_agent`` resolves them),
``outbox.drain`` (collect envelopes queued here for other connections), ``deliver`` (one-turn Bot
Chat delivery on the TARGET gateway, returns the reply), ``reply`` (write the reply/error back on
the SENDER gateway for its waiter). Plumbing: ``tools/bot_relay.py``; handlers are rebound onto
server.py's globals (method_ctx.py) and reference ``_ok``/``_err`` bare."""
import os
import subprocess
@@ -26,7 +24,6 @@ def _relay_root() -> Path:
def _run_delivery(profile: str, tmp: str) -> subprocess.CompletedProcess:
from tools.bot_relay import local_delivery_command
return subprocess.run(
local_delivery_command(profile, tmp), capture_output=True, text=True, encoding="utf-8",
errors="replace", timeout=600)
@@ -34,14 +31,10 @@ def _run_delivery(profile: str, tmp: str) -> subprocess.CompletedProcess:
@method("bot_relay.roster.sync")
def _(rid, params: dict, _root=_relay_root) -> dict:
"""Replace this gateway's view of agents on OTHER connections → ``{count}`` accepted rows.
``agents``: rows ``{profile, handle, connection_id, connection_label?, title?, description?}``;
rows failing validation are dropped, not fatal.
"""
"""Replace this gateway's view of agents on OTHER connections → ``{count}`` accepted rows
(``agents`` rows ``{profile, handle, connection_id, ...}``; invalid rows are dropped)."""
try:
from tools.bot_relay import write_remote_roster
return _ok(rid, {"count": write_remote_roster(_root(), params.get("agents"))})
except Exception as e:
return _err(rid, 5090, str(e))
@@ -49,13 +42,10 @@ def _(rid, params: dict, _root=_relay_root) -> dict:
@method("bot_relay.outbox.drain")
def _(rid, params: dict, _root=_relay_root) -> dict:
"""Claim every pending cross-connection envelope queued on this gateway → ``{envelopes}``.
Claimed envelopes move to ``claimed/`` atomically, so concurrent drains can't double-deliver.
"""
"""Claim every pending cross-connection envelope queued here → ``{envelopes}``; claimed
envelopes move to ``claimed/`` atomically so concurrent drains can't double-deliver."""
try:
from tools.bot_relay import claim_pending_envelopes
return _ok(rid, {"envelopes": claim_pending_envelopes(_root())})
except Exception as e:
return _err(rid, 5091, str(e))
@@ -66,10 +56,7 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
"""Deliver a relayed DM (``profile``, attribution-prefixed ``message``) into a Bot Chat ON THIS
GATEWAY via the one-turn ``hermes -p <profile> chat -c "Bot Chat"`` transport local DMs use →
``{reply}``. Blocking by design (Desktop relay worker; the RPC pool keeps it off the reader)."""
import os
import subprocess
import tempfile
profile = str(params.get("profile") or "").strip()
message = str(params.get("message") or "").strip()
if not profile or not message:
@@ -77,14 +64,12 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
try:
from tools.bot_mode_dm import MESSAGE_MAX_CHARS
from tools.bot_relay import acquire_turn_lock
if len(message) > MESSAGE_MAX_CHARS + 200: # + attribution headroom
return _err(rid, 4091, "message too long")
root = _root()
known = {"default"}
profiles_dir = root / "profiles"
if profiles_dir.is_dir():
known.update(c.name for c in profiles_dir.iterdir() if c.is_dir())
if (root / "profiles").is_dir():
known.update(c.name for c in (root / "profiles").iterdir() if c.is_dir())
resolved = "default" if profile.lower() == "hermes" else profile
if resolved not in known:
return _err(rid, 4092, f"no profile '{profile}' on this gateway")
@@ -92,23 +77,15 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
# When THIS gateway already hosts the target's Bot Chat live, the subprocess transport is
# fenced out by the single-owner lease and the payload dropped. Land the DM in the live
# session via prompt.submit — the composer's choke point, so role alternation, persistence
# and streaming behave as a typed message would. (Nested: needs server globals via rebind.)
def _live_bot_chat_sid(profile_name: str) -> str:
from tools.bot_mode_probe import BOT_CHAT_TITLE
live_home = _profile_home(profile_name)
want_home = str(live_home) if live_home is not None else None
for live_sid, record in list(_sessions.items()):
if not isinstance(record, dict):
continue
if (record.get("profile_home") or None) != want_home:
continue
key = _session_lookup_key(record, fallback=live_sid)
if _session_live_title(record, key) == BOT_CHAT_TITLE:
return live_sid
return ""
live_sid = _live_bot_chat_sid(resolved)
# and streaming behave as a typed message would.
from tools.bot_mode_probe import BOT_CHAT_TITLE
live_home = _profile_home(resolved)
want_home = str(live_home) if live_home is not None else None
live_sid = next((
live_sid for live_sid, record in list(_sessions.items())
if isinstance(record, dict) and (record.get("profile_home") or None) == want_home
and _session_live_title(
record, _session_lookup_key(record, fallback=live_sid)) == BOT_CHAT_TITLE), "")
if live_sid:
# queued=True: a teammate's DM runs as the NEXT turn and never interrupts or steers a
# turn in flight (the default busy mode does); arrivals queue in order.
@@ -118,6 +95,9 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
reply = f"Delivered into @{resolved}'s open Bot Chat; the reply will appear there."
return _ok(rid, {"reply": reply})
def _detail(p) -> str:
return (p.stderr or p.stdout or "").strip()[-500:]
fd, tmp = tempfile.mkstemp(prefix="hermes-relay-dm-", suffix=".txt", text=True)
try:
with os.fdopen(fd, "w", encoding="utf-8") as f:
@@ -133,28 +113,22 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
# transcript first (no fresh session is minted). Auth/quota/config never retry.
from tools.bot_failure_reasons import (
RETRY_NONE, classify_agent_error, retry_action)
first_detail = (proc.stderr or proc.stdout or "").strip()[-500:]
if retry_action(classify_agent_error(first_detail)) != RETRY_NONE:
if retry_action(classify_agent_error(_detail(proc))) != RETRY_NONE:
proc = _run(resolved, tmp)
finally:
with contextlib.suppress(OSError):
os.unlink(tmp)
if proc.returncode != 0:
from tools.bot_failure_reasons import classify_agent_error
detail = (proc.stderr or proc.stdout or "").strip()[-500:]
return _err(
rid, 5092, f"delivery turn failed: {detail or proc.returncode}",
data={"reason": classify_agent_error(detail)})
detail = _detail(proc)
return _err(rid, 5092, f"delivery turn failed: {detail or proc.returncode}",
data={"reason": classify_agent_error(detail)})
return _ok(rid, {"reply": (proc.stdout or "").strip()})
except subprocess.TimeoutExpired:
return _err(rid, 5093, "delivery turn timed out")
except Exception as e:
# 'target_busy' extends the structured refusal enum.
if getattr(e, "reason", "") == "target_busy":
return _err(rid, 5096, str(e))
return _err(rid, 5094, str(e))
return _err(rid, 5096 if getattr(e, "reason", "") == "target_busy" else 5094, str(e))
@method("bot_relay.reply")
@@ -166,10 +140,8 @@ def _(rid, params: dict, _root=_relay_root) -> dict:
return _err(rid, 4093, "id required")
try:
from tools.bot_relay import write_reply
write_reply(
_root(), envelope_id, reply=str(params.get("reply") or ""),
error=str(params.get("error") or ""), reason=str(params.get("reason") or ""))
write_reply(_root(), envelope_id, reply=str(params.get("reply") or ""),
error=str(params.get("error") or ""), reason=str(params.get("reason") or ""))
return _ok(rid, {"ok": True})
except ValueError as e:
return _err(rid, 4094, str(e))
@@ -180,7 +152,6 @@ def _(rid, params: dict, _root=_relay_root) -> dict:
def register(server) -> None:
_registry.install(server)
from . import methods_groups
server._LONG_HANDLERS = server._LONG_HANDLERS | methods_groups.LONG_HANDLERS
for name in (
"get_hosted_room_service", "_WORKER_UNAVAILABLE", "_profile_name", "_requested_profile",

113
tui_gateway/methods_browser.py
View File

@@ -1,8 +1,5 @@
"""Browser connect/disconnect helpers for the browser.* RPCs (CDP probing, no network I/O on status).
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
"""Browser connect/disconnect helpers for the browser.* RPCs (CDP probing, no network I/O on
status). Bodies are rebound onto server.py's globals at install time (method_ctx.bind_module)."""
from __future__ import annotations
@@ -14,25 +11,17 @@ _CDP_SCHEMES = {"http", "https", "ws", "wss"}
def _resolve_browser_cdp_url() -> str:
"""Configured browser CDP override without network I/O.
``/browser status`` must be fast: ``tools.browser_tool._get_cdp_override`` runs an HTTP
probe with a multi-second timeout for discovery-style URLs. Mirrors its precedence (env var,
then ``browser.cdp_url``) minus the WS-resolution step, so the answer reflects user intent
even when the host is unreachable; ``browser_navigate`` normalizes on the next tool call.
"""
env_url = os.environ.get("BROWSER_CDP_URL", "").strip()
if env_url:
"""Configured browser CDP override without network I/O (``/browser status`` must be fast;
``tools.browser_tool._get_cdp_override`` HTTP-probes discovery URLs). Same precedence (env,
then ``browser.cdp_url``) minus WS resolution; ``browser_navigate`` normalizes on the next call."""
if env_url := os.environ.get("BROWSER_CDP_URL", "").strip():
return env_url
try:
with contextlib.suppress(Exception):
from hermes_cli.config import read_raw_config
cfg = read_raw_config()
browser_cfg = cfg.get("browser", {}) if isinstance(cfg, dict) else {}
if isinstance(browser_cfg, dict):
return str(browser_cfg.get("cdp_url", "") or "").strip()
except Exception:
pass
return ""
@@ -40,59 +29,29 @@ def _is_default_local_cdp(parsed) -> bool:
"""Match the discovery-style local default; never the concrete WS form — a
``ws://127.0.0.1:9222/devtools/browser/<id>`` is connectable as-is and collapsing it to bare
``http://...:9222`` would break the connect."""
try:
port = parsed.port or 80
except ValueError:
return False
return (parsed.scheme in {"http", "ws"} and parsed.hostname in {"127.0.0.1", "localhost"}
and port == 9222 and parsed.path in {"", "/", "/json", "/json/version"})
with contextlib.suppress(ValueError):
return (parsed.scheme in {"http", "ws"} and parsed.hostname in {"127.0.0.1", "localhost"}
and (parsed.port or 80) == 9222 and parsed.path in {"", "/", "/json", "/json/version"})
return False
def _cdp_http_reachable(parsed, timeout: float = 2.0) -> bool:
"""True when ``/json/version`` or ``/json`` on the CDP host answers 2xx."""
import urllib.request
scheme = {"ws": "http", "wss": "https"}.get(parsed.scheme, parsed.scheme)
root = f"{scheme}://{parsed.netloc}".rstrip("/")
for url in (f"{root}/json/version", f"{root}/json"):
try:
with urllib.request.urlopen(url, timeout=timeout) as resp:
if 200 <= getattr(resp, "status", 200) < 300:
return True
except Exception:
pass
with contextlib.suppress(Exception), urllib.request.urlopen(url, timeout=timeout) as resp:
if 200 <= getattr(resp, "status", 200) < 300:
return True
return False
def _normalize_cdp_url(parsed) -> str:
# Concrete ``/devtools/browser/<id>`` endpoints stay as-is; discovery-style inputs
# collapse to ``scheme://host:port`` so ``_resolve_cdp_override`` can append ``/json/version``.
if parsed.path.startswith("/devtools/browser/"):
return parsed.geturl()
return parsed._replace(path="", params="", query="", fragment="").geturl()
def _launch_failure_hints(port: int, system: str) -> list[str]:
from hermes_cli.browser_connect import manual_chrome_debug_command
command = manual_chrome_debug_command(port, system)
hint = (
["Start a Chromium-family browser with remote debugging, then retry /browser connect:", command]
if command
else [
"No supported Chromium-family browser executable was found in this environment.",
f"Install one or start a Chromium-family browser with --remote-debugging-port={port}, then retry /browser connect.",
])
return [
*hint,
"Browser not connected — start a Chromium-family browser with remote debugging and retry /browser connect",
]
def _connect_local_default(port: int, system: str, announce) -> str | None:
"""Discover (or launch) the default local debug browser → CDP URL, or None after announcing."""
from hermes_cli.browser_connect import (
discover_local_cdp_url, find_free_debug_port, launch_chrome_debug, local_port_in_use)
discover_local_cdp_url, find_free_debug_port, launch_chrome_debug, local_port_in_use,
manual_chrome_debug_command)
# Dual-stack discovery: when another app squats the IPv4 loopback on the debug port, a
# browser bound there comes up on [::1] only; an IPv4-only probe misses it AND hangs
@@ -104,10 +63,9 @@ def _connect_local_default(port: int, system: str, announce) -> str | None:
launch_port = port
if local_port_in_use(port):
launch_port = find_free_debug_port(port)
announce(
f"Port {port} is occupied by another application that isn't a CDP browser "
"(an IDE debugger or dev server may be using it) — launching a debug browser "
f"on port {launch_port} instead...")
announce(f"Port {port} is occupied by another application that isn't a CDP browser "
"(an IDE debugger or dev server may be using it) — launching a debug browser "
f"on port {launch_port} instead...")
else:
announce("Chromium-family browser isn't running with remote debugging — attempting to launch...")
launch = launch_chrome_debug(launch_port, system)
@@ -115,8 +73,7 @@ def _connect_local_default(port: int, system: str, announce) -> str | None:
# Bounded wait: the whole connect must finish inside the client RPC timeout.
deadline = time.monotonic() + 10.0
while time.monotonic() < deadline:
discovered = discover_local_cdp_url(launch_port, timeout=1.0)
if discovered:
if discovered := discover_local_cdp_url(launch_port, timeout=1.0):
break
time.sleep(0.5)
if discovered:
@@ -124,18 +81,23 @@ def _connect_local_default(port: int, system: str, announce) -> str | None:
return discovered
if launch.hint:
announce(launch.hint, level="error")
for line in _launch_failure_hints(launch_port, system):
command = manual_chrome_debug_command(launch_port, system)
hints = (
["Start a Chromium-family browser with remote debugging, then retry /browser connect:", command]
if command else [
"No supported Chromium-family browser executable was found in this environment.",
f"Install one or start a Chromium-family browser with --remote-debugging-port={launch_port}, then retry /browser connect."])
hints.append("Browser not connected — start a Chromium-family browser with remote debugging and retry /browser connect")
for line in hints:
announce(line, level="error")
return None
def _browser_connect(rid, params: dict) -> dict:
import platform
from hermes_cli.browser_connect import DEFAULT_BROWSER_CDP_URL
from tools.browser_tool import cleanup_all_browsers
from urllib.parse import urlparse
raw_url = params.get("url")
if raw_url is not None and not isinstance(raw_url, str):
return _err(rid, 4015, f"browser url must be a string, got {type(raw_url).__name__}")
@@ -144,10 +106,9 @@ def _browser_connect(rid, params: dict) -> dict:
def announce(message: str, *, level: str = "info") -> None:
messages.append(message)
# Without a session id the TUI prints `messages` from the response; an event would double-render.
# Without a session id the TUI prints `messages` from the response (an event would double-render).
if sid:
_emit("browser.progress", sid, {"message": message, "level": level})
parsed = urlparse(url if "://" in url else f"http://{url}")
if parsed.scheme not in _CDP_SCHEMES:
return _err(rid, 4015, f"unsupported browser url: {url}")
@@ -167,10 +128,8 @@ def _browser_connect(rid, params: dict) -> dict:
# check TCP reachability only and let browser_navigate handshake.
if parsed.scheme in {"ws", "wss"} and parsed.path.startswith("/devtools/browser/"):
import socket
try:
with socket.create_connection((parsed.hostname, port), timeout=2.0):
pass
socket.create_connection((parsed.hostname, port), timeout=2.0).close()
except OSError as e:
return _err(rid, 5031, f"could not reach browser CDP at {url}: {e}")
elif _is_default_local_cdp(parsed):
@@ -182,7 +141,10 @@ def _browser_connect(rid, params: dict) -> dict:
parsed = urlparse(url)
elif not _cdp_http_reachable(parsed):
return _err(rid, 5031, f"could not reach browser CDP at {url}")
normalized = _normalize_cdp_url(parsed)
# Concrete ``/devtools/browser/<id>`` endpoints stay as-is; discovery-style inputs collapse
# to ``scheme://host:port`` so ``_resolve_cdp_override`` can append ``/json/version``.
normalized = (parsed.geturl() if parsed.path.startswith("/devtools/browser/")
else parsed._replace(path="", params="", query="", fragment="").geturl())
# Reap BEFORE publishing the new env (an in-flight tool call sees the old supervisor closed)
# and AFTER (the default task's cached supervisor drains against the new URL).
cleanup_all_browsers()
@@ -190,10 +152,8 @@ def _browser_connect(rid, params: dict) -> dict:
cleanup_all_browsers()
except Exception as e:
return _err(rid, 5031, str(e))
payload: dict[str, object] = {"connected": True, "url": normalized}
if messages:
payload["messages"] = messages
return _ok(rid, payload)
return _ok(rid, {"connected": True, "url": normalized,
**({"messages": messages} if messages else {})})
def _browser_disconnect(rid) -> dict:
@@ -201,7 +161,6 @@ def _browser_disconnect(rid) -> dict:
def reap() -> None:
with contextlib.suppress(Exception):
from tools.browser_tool import cleanup_all_browsers
cleanup_all_browsers()
reap()

150
tui_gateway/model_switch.py
View File

@@ -1,8 +1,6 @@
"""Model switching for a live session: persist, snapshot/restore runtime, /model apply with guards, bot-capability + config sync.
Bodies are rebound onto server.py's globals at install time (see
method_ctx.bind_module), so they reference server.py globals bare.
"""
"""Model switching for a live session: persist, snapshot/restore runtime, /model apply with
guards, bot-capability + config sync. Bodies are rebound onto server.py's globals at install
time (method_ctx.bind_module), so they reference server.py globals bare."""
from __future__ import annotations
@@ -17,7 +15,6 @@ def _persist_model_switch(result) -> None:
# Targeted key writes: a full `model:` block rewrite via save_config() would destroy
# sibling keys the user set there (`model_slots`, `model_fallback`, ...).
from cli import save_config_value
save_config_value("model.default", result.new_model)
save_config_value("model.provider", result.target_provider)
# A provider without a base_url must clear the stale one (custom endpoint -> native)
@@ -25,11 +22,13 @@ def _persist_model_switch(result) -> None:
save_config_value("model.base_url", result.base_url or None)
_RUNTIME_KEYS = ("model", "provider", "api_key", "base_url", "api_mode")
def _snapshot_agent_model_runtime(agent) -> dict:
"""Capture the current agent model runtime for a one-turn restore."""
snap = {k: getattr(agent, k, "") for k in ("model", "provider", "api_key", "base_url", "api_mode")}
snap["primary_runtime"] = copy.deepcopy(getattr(agent, "_primary_runtime", None))
return snap
return {**{k: getattr(agent, k, "") for k in _RUNTIME_KEYS},
"primary_runtime": copy.deepcopy(getattr(agent, "_primary_runtime", None))}
def _restore_agent_model_runtime(agent, snapshot: dict | None) -> None:
@@ -47,10 +46,10 @@ def _restore_agent_model_runtime(agent, snapshot: dict | None) -> None:
except Exception:
logger.debug("TUI one-turn model restore via primary runtime failed", exc_info=True)
if hasattr(agent, "switch_model"):
model, provider, api_key, base_url, api_mode = (snapshot.get(k, "") for k in _RUNTIME_KEYS)
agent.switch_model(
new_model=snapshot.get("model", ""), new_provider=snapshot.get("provider", ""),
api_key=snapshot.get("api_key", ""), base_url=snapshot.get("base_url", ""),
api_mode=snapshot.get("api_mode", ""), capabilities=snapshot.get("capabilities"))
new_model=model, new_provider=provider, api_key=api_key, base_url=base_url,
api_mode=api_mode, capabilities=snapshot.get("capabilities"))
@contextlib.contextmanager
@@ -65,7 +64,6 @@ def _session_profile_runtime_scope(session: dict):
# Same terminal policy the gateway binds per turn: a docker-configured profile
# must never resolve the launch process's pinned env. Failure → refusal scope.
from tools.terminal_scope import install_profile_terminal_scope, reset_terminal_scope
terminal_token = install_profile_terminal_scope(Path(profile_home))
try:
yield
@@ -79,17 +77,14 @@ def _restart_completed_failed_agent_build(sid: str, session: dict, failed_ready:
"""Replace one completed failed build generation and start its retry."""
if failed_ready is None:
return False
build_lock = session.setdefault("agent_build_lock", threading.Lock())
with build_lock:
if (
session.get("agent") is not None or session.get("agent_error") is None
or session.get("agent_ready") is not failed_ready or not failed_ready.is_set()):
with session.setdefault("agent_build_lock", threading.Lock()):
if (session.get("agent") is not None or session.get("agent_error") is None
or session.get("agent_ready") is not failed_ready or not failed_ready.is_set()):
return False
model_override = session.get("model_override")
resume_overrides = session.get("resume_runtime_overrides")
if isinstance(model_override, dict) and isinstance(resume_overrides, dict):
resume_overrides = dict(resume_overrides)
resume_overrides["model_override"] = model_override
resume_overrides = {**resume_overrides, "model_override": model_override}
if provider := model_override.get("provider"):
resume_overrides["provider_override"] = provider
else:
@@ -107,18 +102,11 @@ def _switch_request(raw_input: str, parsed_flags, persist_override) -> tuple[str
"""Normalize /model flags → (model_input, explicit_provider, one_turn, persist_global)."""
from hermes_cli.model_switch import (
MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL, MODEL_SWITCH_ERROR_TEXT, parse_model_switch_args,
resolve_persist_behavior,
)
resolve_persist_behavior)
if parsed_flags is None:
parsed_flags = parse_model_switch_args(raw_input)
if hasattr(parsed_flags, "model_input"):
model_input, explicit_provider = parsed_flags.model_input, parsed_flags.explicit_provider
is_global_flag, is_session = parsed_flags.is_global, parsed_flags.is_session
one_turn = parsed_flags.is_once
else:
model_input, explicit_provider, is_global_flag, _force_refresh, is_session = parsed_flags
one_turn = False
f = parse_model_switch_args(raw_input) if parsed_flags is None else parsed_flags
model_input, explicit_provider, is_global_flag, is_session, one_turn = (
f.model_input, f.explicit_provider, f.is_global, f.is_session, f.is_once)
# Conflict validation is the shared parser's; surface it with the canonical copy.
if is_global_flag and one_turn:
raise ValueError(MODEL_SWITCH_ERROR_TEXT[MODEL_SWITCH_ERR_ONCE_WITH_GLOBAL])
@@ -134,13 +122,11 @@ def _current_model_runtime(agent, explicit_provider: str) -> tuple:
"""(provider, model, base_url, api_key) to switch from: live agent, else configured runtime."""
if agent:
return tuple(
getattr(agent, k, "") or "" for k in ("provider", "model", "base_url", "api_key")
)
getattr(agent, k, "") or "" for k in ("provider", "model", "base_url", "api_key"))
current_model = _resolve_model()
if explicit_provider:
return explicit_provider.strip(), current_model, "", ""
from hermes_cli.runtime_provider import resolve_runtime_provider
runtime = resolve_runtime_provider(requested=None)
# Keep a callable api_key (Azure Entra bearer) unchanged: ``str()`` would
# yield "<function ...>" and poison switch_model validation.
@@ -151,30 +137,14 @@ def _current_model_runtime(agent, explicit_provider: str) -> tuple:
return provider, current_model, str(runtime.get("base_url", "") or ""), key
def _provider_context() -> tuple:
"""(user providers, compatible custom providers, cfg) from config; all None on load failure."""
user_provs = custom_provs = cfg = None
try:
from hermes_cli.config import get_compatible_custom_providers, load_config
cfg = load_config()
user_provs = cfg.get("providers")
custom_provs = get_compatible_custom_providers(cfg)
except Exception:
pass
return user_provs, custom_provs, cfg
def _merge_preflight_warning(result, agent, session: dict, cfg, custom_provs) -> None:
"""Fold the context-compression preflight warning into ``result`` (best-effort)."""
try:
from hermes_cli.context_switch_guard import merge_preflight_compression_warning
cfg_ctx = None
if isinstance(cfg, dict):
mc = cfg.get("model", {})
if isinstance(mc, dict) and mc.get("context_length") is not None:
cfg_ctx = int(mc["context_length"])
mc = cfg.get("model", {}) if isinstance(cfg, dict) else None
if isinstance(mc, dict) and mc.get("context_length") is not None:
cfg_ctx = int(mc["context_length"])
merge_preflight_compression_warning(
result, agent=agent, messages=list(session.get("history", [])),
custom_providers=custom_provs, config_context_length=cfg_ctx)
@@ -186,7 +156,6 @@ def _expensive_model_confirm(result, current_base_url: str, current_api_key) ->
"""Deferred-confirm response when the selection guards flag the target model, else None."""
try:
from hermes_cli.model_selection_guards import combined_selection_warning
warning = combined_selection_warning(
result.new_model, provider=result.target_provider, base_url=result.base_url or current_base_url,
api_key=result.api_key or current_api_key, model_info=result.model_info)
@@ -194,12 +163,9 @@ def _expensive_model_confirm(result, current_base_url: str, current_api_key) ->
warning = None
if warning is None:
return None
confirm_msg = warning.message
if result.warning_message:
confirm_msg = f"{confirm_msg}\n\n{result.warning_message}"
# Same contract as _set_model's deferred branch: confirm_message is
# canonical, warning is the legacy alias — keep identical.
return {"value": result.new_model, "warning": confirm_msg, "confirm_required": True, "confirm_message": confirm_msg}
msg = f"{warning.message}\n\n{result.warning_message}" if result.warning_message else warning.message
# Same contract as _set_model's deferred branch: confirm_message is canonical, warning legacy.
return {"value": result.new_model, "warning": msg, "confirm_required": True, "confirm_message": msg}
def _commit_agent_switch(sid: str, session: dict, agent, result, current_model: str, snapshot):
@@ -213,10 +179,8 @@ def _commit_agent_switch(sid: str, session: dict, agent, result, current_model:
# The in-place swap rolled the agent back and re-raised. Abort the whole commit (worker
# restart, persist, marker, override, config write) or the session pins a broken model.
logger.warning("In-place model switch failed for TUI agent: %s", exc)
raise ValueError(
f"Model switch to {result.new_model} failed ({exc}); "
f"staying on {getattr(agent, 'model', current_model)}."
) from exc
raise ValueError(f"Model switch to {result.new_model} failed ({exc}); "
f"staying on {getattr(agent, 'model', current_model)}.") from exc
_restart_slash_worker(sid, session)
_persist_live_session_runtime(session)
_persist_live_session_system_prompt(session)
@@ -233,7 +197,6 @@ def _apply_model_switch(
pin_session_override: bool = True, parsed_flags: Any | None = None,
persist_override: bool | None = None) -> dict:
from hermes_cli.model_switch import switch_model
model_input, explicit_provider, one_turn, persist_global = _switch_request(
raw_input, parsed_flags, persist_override)
agent = session.get("agent")
@@ -243,12 +206,17 @@ def _apply_model_switch(
agent, explicit_provider)
# User-defined providers let switch_model resolve named custom endpoints
# (e.g. "ollama-launch") and validate against saved model lists.
user_provs, custom_provs, cfg = _provider_context()
user_provs = custom_provs = cfg = None
with contextlib.suppress(Exception):
from hermes_cli.config import get_compatible_custom_providers, load_config
cfg = load_config()
user_provs = cfg.get("providers")
custom_provs = get_compatible_custom_providers(cfg)
result = switch_model(
raw_input=model_input, current_provider=current_provider, current_model=current_model,
current_base_url=current_base_url, current_api_key=current_api_key, is_global=persist_global,
explicit_provider=explicit_provider, user_providers=user_provs, custom_providers=custom_provs,
)
explicit_provider=explicit_provider, user_providers=user_provs,
custom_providers=custom_provs)
if not result.success:
raise ValueError(result.error_message or "model switch failed")
restore_snapshot = _snapshot_agent_model_runtime(agent) if (one_turn and agent) else None
@@ -276,12 +244,10 @@ def _apply_model_switch(
def _sync_bot_capabilities(sid: str, session: dict) -> None:
"""Rebuild a Bot Chat session's agent when its capability surface changed.
Bot Chats are eternal sessions with toolsets/MCP baked in at construction, so a capability
edit would otherwise wait for /new. At turn start, fingerprint the profile's capabilities
and on change swap in a fresh agent for the SAME session (history is DB-backed).
"""
"""Rebuild a Bot Chat session's agent when its capability surface changed. Bot Chats are
eternal sessions with toolsets/MCP baked in at construction, so a capability edit would
otherwise wait for /new: fingerprint at turn start and on change swap in a fresh agent for
the SAME session (history is DB-backed)."""
agent = session.get("agent")
if agent is None:
return
@@ -293,7 +259,6 @@ def _sync_bot_capabilities(sid: str, session: dict) -> None:
if title != "Bot Chat":
return
from tools.bot_mode_probe import capability_fingerprint
current = capability_fingerprint(session.get("profile_home") or None)
if current == "unavailable":
return
@@ -303,29 +268,23 @@ def _sync_bot_capabilities(sid: str, session: dict) -> None:
return
except Exception:
return
try:
tokens = _set_session_context(sid, cwd=_session_cwd(session))
try:
new_agent = _make_agent(
sid, session["session_key"], session_id=session["session_key"], platform_override=_session_source(session)
)
new_agent = _make_agent(sid, session["session_key"], session_id=session["session_key"],
platform_override=_session_source(session))
finally:
_clear_session_context(tokens)
new_agent._session_title_hint = "Bot Chat"
session["agent"] = new_agent
session["config_model_seen"] = _config_model_target()
session.update(agent=new_agent, config_model_seen=_config_model_target())
_emit("notice", sid, {"message": "Capabilities updated — this bot's tools and prompt were refreshed."})
except Exception as e:
logger.warning("Bot capability sync failed for %s: %s", sid, e)
def _sync_agent_model_with_config(sid: str, session: dict) -> None:
"""Adopt a config.yaml model change at turn start (like gateways do per message).
Sessions pinned with /model keep their choice; a failed switch keeps the current
model and never blocks the turn.
"""
"""Adopt a config.yaml model change at turn start (like gateways do per message). Sessions
pinned with /model keep their choice; a failed switch keeps the current model."""
agent = session.get("agent")
if agent is None or session.get("model_override"):
return
@@ -335,36 +294,31 @@ def _sync_agent_model_with_config(sid: str, session: dict) -> None:
seen = session.get("config_model_seen")
# Record first so a broken config gets one attempt per edit, not per turn.
session["config_model_seen"] = target
if target == seen:
return
model, provider = target
# Already on the configured model (resumed before first sync, or a config revert after
# a failed switch): adopt without switching.
if model == getattr(agent, "model", "") and (not provider or provider == getattr(agent, "provider", "")):
if target == seen or (
model == getattr(agent, "model", "") and (not provider or provider == getattr(agent, "provider", ""))):
return
raw = f"{model} --provider {provider}" if provider else model
try:
# This sync ADOPTS a config.yaml change; it must never write config back (that is
# how `hermes --tui -m` once leaked into config.yaml).
_apply_model_switch(
sid, session, raw, confirm_expensive_model=True, pin_session_override=False, persist_override=False
)
sid, session, raw, confirm_expensive_model=True, pin_session_override=False,
persist_override=False)
except Exception as e:
_emit("error", sid, {"message": f"Could not switch to configured model {model}: {e}"})
def _pending_switch_selection_warning(model: str, provider: str) -> str | None:
"""Selection-guard message for a model queued mid-turn, or ``None``.
Runs BEFORE the pick is stashed, while the client can still turn the response into a
confirm prompt. Only pre-resolution inputs exist, so this can only under-fire;
``_apply_model_switch`` is the backstop. Exceptions mean "no warning".
"""
"""Selection-guard message for a model queued mid-turn, or ``None``. Runs BEFORE the pick is
stashed (the client can still turn the response into a confirm prompt); only pre-resolution
inputs exist so it can only under-fire — ``_apply_model_switch`` is the backstop."""
if not model:
return None
try:
from hermes_cli.model_selection_guards import combined_selection_warning
warning = combined_selection_warning(model, provider=provider or None)
except Exception:
return None