From e7e287cdd84ae88969bb18ec6fa365bd7a3d68f8 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 3 Sep 2026 02:28:12 -0700 Subject: [PATCH] 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 --- tui_gateway/agent_callbacks.py | 237 +++++++++------------- tui_gateway/change_watcher.py | 110 ++++------ tui_gateway/compute_host_bridge.py | 104 ++++------ tui_gateway/git_probe.py | 57 ++---- tui_gateway/host_supervisor.py | 199 ++++++++---------- tui_gateway/hosted_room_peer_transport.py | 137 +++++-------- tui_gateway/hosted_room_server_rpc.py | 89 +++----- tui_gateway/method_ctx.py | 54 ++--- tui_gateway/methods_bot_relay.py | 93 +++------ tui_gateway/methods_browser.py | 113 ++++------- tui_gateway/model_switch.py | 150 +++++--------- 11 files changed, 493 insertions(+), 850 deletions(-) diff --git a/tui_gateway/agent_callbacks.py b/tui_gateway/agent_callbacks.py index 21592409a2..ea30f137c8 100644 --- a/tui_gateway/agent_callbacks.py +++ b/tui_gateway/agent_callbacks.py @@ -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"] = [] diff --git a/tui_gateway/change_watcher.py b/tui_gateway/change_watcher.py index 395cee5a31..9cb3dfa698 100644 --- a/tui_gateway/change_watcher.py +++ b/tui_gateway/change_watcher.py @@ -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() diff --git a/tui_gateway/compute_host_bridge.py b/tui_gateway/compute_host_bridge.py index f5155f57a5..5a5a0398d9 100644 --- a/tui_gateway/compute_host_bridge.py +++ b/tui_gateway/compute_host_bridge.py @@ -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: diff --git a/tui_gateway/git_probe.py b/tui_gateway/git_probe.py index 5930e90374..875385c7d7 100644 --- a/tui_gateway/git_probe.py +++ b/tui_gateway/git_probe.py @@ -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 `` → 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 `` → 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)) diff --git a/tui_gateway/host_supervisor.py b/tui_gateway/host_supervisor.py index 5cec67c807..cd19373672 100644 --- a/tui_gateway/host_supervisor.py +++ b/tui_gateway/host_supervisor.py @@ -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"] diff --git a/tui_gateway/hosted_room_peer_transport.py b/tui_gateway/hosted_room_peer_transport.py index 170c98a28b..0bda289022 100644 --- a/tui_gateway/hosted_room_peer_transport.py +++ b/tui_gateway/hosted_room_peer_transport.py @@ -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: `` 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: `` 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) diff --git a/tui_gateway/hosted_room_server_rpc.py b/tui_gateway/hosted_room_server_rpc.py index edbe5b8eaa..e470c91a09 100644 --- a/tui_gateway/hosted_room_server_rpc.py +++ b/tui_gateway/hosted_room_server_rpc.py @@ -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}) diff --git a/tui_gateway/method_ctx.py b/tui_gateway/method_ctx.py index a35e1c6103..445f02c274 100644 --- a/tui_gateway/method_ctx.py +++ b/tui_gateway/method_ctx.py @@ -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: diff --git a/tui_gateway/methods_bot_relay.py b/tui_gateway/methods_bot_relay.py index 31b46a2264..59fae4ecb6 100644 --- a/tui_gateway/methods_bot_relay.py +++ b/tui_gateway/methods_bot_relay.py @@ -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 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", diff --git a/tui_gateway/methods_browser.py b/tui_gateway/methods_browser.py index ac9a303718..e4e4ed7b05 100644 --- a/tui_gateway/methods_browser.py +++ b/tui_gateway/methods_browser.py @@ -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/`` 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/`` 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/`` 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() diff --git a/tui_gateway/model_switch.py b/tui_gateway/model_switch.py index 86cd88aa6d..02c458105c 100644 --- a/tui_gateway/model_switch.py +++ b/tui_gateway/model_switch.py @@ -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 "" 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