From 3f200b907ef3e6a5d2350309a677e96a1b8f0098 Mon Sep 17 00:00:00 2001 From: andrexibiza <84248988+andrexibiza@users.noreply.github.com> Date: Wed, 5 Aug 2026 03:05:46 -0500 Subject: [PATCH] refactor(tui): move config.set handler into methods_config (server.py god-file slice R4) Signed-off-by: andrexibiza <84248988+andrexibiza@users.noreply.github.com> --- tests/tui_gateway/test_methods_config_seam.py | 203 +++++ tui_gateway/methods_config.py | 697 ++++++++++++++++- tui_gateway/server.py | 700 ------------------ 3 files changed, 897 insertions(+), 703 deletions(-) create mode 100644 tests/tui_gateway/test_methods_config_seam.py diff --git a/tests/tui_gateway/test_methods_config_seam.py b/tests/tui_gateway/test_methods_config_seam.py new file mode 100644 index 000000000000..7bf9f19f7f77 --- /dev/null +++ b/tests/tui_gateway/test_methods_config_seam.py @@ -0,0 +1,203 @@ +"""R4-S1 seam-identity regression: config.set lives in methods_config.py. + +God-file slice (epic #78647 / target #78630): the anonymous +``@method("config.set")`` handler moved byte-verbatim from +``tui_gateway/server.py`` (lines 10471-11162 at the pre-splice base) into +``tui_gateway/methods_config.py``. The HandlerRegistry.install seam +(method_ctx.py) rebinds the moved handler's ``__globals__`` onto server.py's +namespace, so every ``server.`` monkeypatch keeps landing and every +module-global read keeps resolving at call time exactly as before the split. + +These tests pin that identity: + +* T1 (source seam): the handler source now in methods_config.py is + byte-identical to the golden window from the pre-splice base. +* T2 (registry seam): ``server._methods["config.set"]`` is the live dispatch + target and its ``__globals__`` IS server.py's namespace. +* T3 (patch-liveness): monkeypatching ``server._load_cfg`` / + ``server._write_config_key`` is seen by the moved handler through the + dispatch path (the re-export/rebind trap regression). +* T4 (aggressive dispatch): representative branches (battery toggle, + indicator validation, unknown-key 4002, skin broadcast) behave identically + through the moved handler. +""" + +from __future__ import annotations + +import inspect +import subprocess +from pathlib import Path + +import pytest + +from tui_gateway import server +from tui_gateway import methods_config + +REPO_ROOT = Path(__file__).resolve().parents[2] + +# Golden byte window of the moved handler at the pre-splice base +# (sed -n '10471,11162p' of tui_gateway/server.py, pinned 2026-08-05). +GOLDEN_SHA = "0788df804f864b1007562c70abbe4b39666af971bf1811d693e8d33cf0dbe3fa" + + +def _pre_splice_window() -> str: + """The config.set window from the pre-splice base (git HEAD~ or HEAD).""" + out = subprocess.run( + ["git", "show", "HEAD:tui_gateway/server.py"], + capture_output=True, + text=True, + check=True, + ).stdout + lines = out.split("\n") + return "\n".join(lines[10470:11162]) # 0-idx: lines 10471-11162 + + +def _moved_handler_source() -> str: + """config.set handler source as it now lives in methods_config.py.""" + src = methods_config.__file__ + text = Path(src).read_text(encoding="utf-8").split("\n") + start = next(i for i, l in enumerate(text) if l == '@method("config.set")') + # The anonymous handler is the immediate FunctionDef after the decorator; + # capture through the end of the file body before `def register`. + end = next(i for i in range(start + 1, len(text)) if text[i].startswith("def register(")) + return "\n".join(text[start:end]).rstrip("\n") + + +# --------------------------------------------------------------------------- +# T1 — source seam: moved handler is byte-identical to the golden window +# --------------------------------------------------------------------------- + + +def test_t1_moved_handler_source_matches_golden_window(): + moved = _moved_handler_source() + assert moved, "config.set handler not found in methods_config.py" + import hashlib + + sha = hashlib.sha256((moved + "\n").encode("utf-8")).hexdigest() + assert sha == GOLDEN_SHA, ( + "moved config.set source drifted from the golden window " + f"(got {sha}, want {GOLDEN_SHA})" + ) + + +def test_t1b_config_set_no_longer_registered_in_server_py(): + """server.py must not define @method('config.set') anymore.""" + text = Path(server.__file__).read_text(encoding="utf-8") + assert '@method("config.set")' not in text + + +def test_t1c_stale_note_deleted_from_both_files(): + """Both stale NOTE copies (server.py + methods_config.py docstring) are gone.""" + needle = "config.set intentionally stays in server.py" + assert needle not in Path(server.__file__).read_text(encoding="utf-8") + assert needle not in Path(methods_config.__file__).read_text(encoding="utf-8") + + +# --------------------------------------------------------------------------- +# T2 — registry seam: dispatch target identity +# --------------------------------------------------------------------------- + + +def test_t2_config_set_registered_in_server_methods(): + handler = server._methods.get("config.set") + assert handler is not None, "config.set missing from server._methods" + assert callable(handler) + + +def test_t2b_handler_globals_rebound_onto_server_namespace(): + """install() rebinds __globals__ to server.py — patch-liveness prerequisite.""" + handler = server._methods["config.set"] + assert handler.__globals__ is vars(server) + + +def test_t2c_source_has_single_definition_in_methods_config(): + """The handler is defined exactly once, in methods_config.py.""" + assert '@method("config.set")' in Path(methods_config.__file__).read_text(encoding="utf-8") + + +# --------------------------------------------------------------------------- +# T3 — patch-liveness through the rebind seam (the re-export trap regression) +# --------------------------------------------------------------------------- + + +def test_t3_patched_server_globals_are_seen_by_moved_handler(monkeypatch): + """Patching server._load_cfg/_write_config_key must affect the moved handler.""" + writes: dict[str, object] = {} + monkeypatch.setattr(server, "_load_cfg", lambda: {"display": {"battery": False}}) + monkeypatch.setattr( + server, "_write_config_key", lambda k, v: writes.__setitem__(k, v) + ) + + resp = server.dispatch( + {"id": "c1", "method": "config.set", "params": {"key": "battery", "value": ""}} + ) + + assert resp["result"] == {"key": "battery", "value": "on"} + assert writes == {"display.battery": True} + + +def test_t3b_patched_emit_is_seen_by_moved_handler(monkeypatch): + """config.set emits session.info through server._emit — patch must land.""" + emitted: list[tuple] = [] + monkeypatch.setattr(server, "_emit", lambda *a: emitted.append(a)) + monkeypatch.setattr(server, "_load_cfg", lambda: {"display": {"battery": False}}) + monkeypatch.setattr( + server, "_write_config_key", lambda k, v: None + ) + + resp = server.dispatch( + {"id": "c1", "method": "config.set", "params": {"key": "battery", "value": "on"}} + ) + assert resp["result"] == {"key": "battery", "value": "on"} + # battery branch emits session.info only when a live session exists; with + # none, no emit is expected — this pins that the handler runs without + # error and resolves _emit through server.py (no AttributeError). + assert resp["result"]["value"] == "on" + + +# --------------------------------------------------------------------------- +# T4 — aggressive dispatch branches through the moved handler +# --------------------------------------------------------------------------- + + +def test_t4_unknown_key_returns_4002(): + resp = server.dispatch( + {"id": "u1", "method": "config.set", "params": {"key": "no_such_key", "value": "x"}} + ) + assert resp.get("error", {}).get("code") == 4002 + + +def test_t4_indicator_validates_against_INDICATOR_STYLES(monkeypatch): + monkeypatch.setattr( + server, "_write_config_key", lambda k, v: None + ) + resp = server.dispatch( + {"id": "i1", "method": "config.set", "params": {"key": "indicator", "value": "bogus"}} + ) + err = resp.get("error", {}) + assert err.get("code") == 4002 + assert "unknown indicator" in err.get("message", "") + + +def test_t4_skin_broadcast_path(monkeypatch): + """skin branch broadcasts via server._broadcast_global_event + resolve_skin.""" + events: list[tuple] = [] + monkeypatch.setattr(server, "_broadcast_global_event", lambda *a: events.append(a)) + monkeypatch.setattr(server, "resolve_skin", lambda: {"name": "default"}) + monkeypatch.setattr( + server, "_write_config_key", lambda k, v: None + ) + + resp = server.dispatch( + {"id": "s1", "method": "config.set", "params": {"key": "skin", "value": "default"}} + ) + # Accept ok result; the broadcast may or may not fire depending on + # whether a live session exists — the contract is no crash + ok path. + assert resp.get("result") is not None or resp.get("error") is None + + +def test_t4_handler_is_dispatch_pool_safe(): + """#60654 guard: handler body does not acquire _stdout_lock or history_lock.""" + src = inspect.getsource(server._methods["config.set"]) + assert "_stdout_lock" not in src + assert "history_lock" not in src diff --git a/tui_gateway/methods_config.py b/tui_gateway/methods_config.py index 9501237989c6..5ca855ba2c40 100644 --- a/tui_gateway/methods_config.py +++ b/tui_gateway/methods_config.py @@ -1,8 +1,5 @@ """Config / projects / setup JSON-RPC handlers (moved verbatim from server.py). -NOTE: ``config.set`` stays in server.py for now — the in-flight -opt/model-resolution-core PR touches it; move it in a follow-up once merged. - Handler bodies are byte-identical to their pre-split server.py form; they are rebound onto server.py's globals at install time — see method_ctx.py. """ @@ -417,6 +414,700 @@ def _(rid, params: dict) -> dict: return _ok(rid, {"ok": False, "error": str(e)}) +@method("config.set") +def _(rid, params: dict) -> dict: + key, value = params.get("key", ""), params.get("value", "") + session = _sessions.get(params.get("session_id", "")) + + if key == "model": + try: + if not value: + return _err(rid, 4002, "model value required") + if session: + from hermes_cli.model_switch import parse_model_switch_args + + # A live swap can't run in-place while a turn streams: + # agent.switch_model() mutates self.model / self.provider / + # self.base_url / self.client, and the worker thread running + # agent.run_conversation reads those every iteration — a + # mid-turn swap can fire an HTTP request with the new base_url + # but old model (400/404s). So instead of rejecting the pick + # (the old 4009), stash it and apply it at the NEXT turn start + # (_apply_pending_model_switch), where nothing is in flight. + # The user gets to pick, keep typing, and send the next turn on + # the new model without waiting for the swap or interrupting. + if session.get("running"): + parsed = parse_model_switch_args(value) + try: + pending_model = parsed.model_input + except Exception: + pending_model = str(value) + session["pending_model_switch"] = { + "raw": value, + "confirm_expensive_model": bool( + params.get("confirm_expensive_model", False) + ), + # The resolved model/provider the next turn will run on. + # _session_info reports these while the switch is pending + # so the end-of-turn settle keeps showing the user's pick + # instead of blipping back to the still-live old model. + "display_model": pending_model, + "display_provider": ( + getattr(parsed, "explicit_provider", "") or "" + ).strip(), + } + return _ok( + rid, + { + "key": key, + "value": pending_model, + "warning": "", + "confirm_required": False, + "confirm_message": "", + "scope": "session", + "deferred": True, + }, + ) + parsed_flags = parse_model_switch_args(value) + explicit_provider = parsed_flags.explicit_provider + if session.get("agent") is None and not explicit_provider.strip(): + session_id = params.get("session_id", "") + _start_agent_build(session_id, session) + init_err = _wait_agent(session, rid) + if init_err: + return init_err + if session.get("agent") is None: + return _err(rid, 5032, "agent initialization failed") + result = _apply_model_switch( + params.get("session_id", ""), + session, + value, + confirm_expensive_model=bool( + params.get("confirm_expensive_model", False) + ), + parsed_flags=parsed_flags, + ) + else: + result = _apply_model_switch( + "", + {"agent": None}, + value, + confirm_expensive_model=bool( + params.get("confirm_expensive_model", False) + ), + ) + return _ok( + rid, + { + "key": key, + "value": result["value"], + "warning": result["warning"], + "confirm_required": result.get("confirm_required", False), + "confirm_message": result.get("confirm_message", ""), + "scope": result.get("scope", "session"), + }, + ) + except Exception as e: + return _err(rid, 5001, str(e)) + + if key == "fast": + raw = str(value or "").strip().lower() + agent = session.get("agent") if session else None + if agent is not None: + current_fast = getattr(agent, "service_tier", None) == "priority" + elif session is not None and session.get("create_service_tier_override") is not None: + # Pre-build session with a pinned tier (desktop draft pick or an + # earlier session-scoped toggle) — report/toggle from the pin, not + # the global default. + current_fast = session["create_service_tier_override"] == "priority" + else: + current_fast = _load_service_tier() == "priority" + + if raw in {"status"}: + return _ok( + rid, + {"key": key, "value": "fast" if current_fast else "normal"}, + ) + + if raw in {"", "toggle"}: + nv = "normal" if current_fast else "fast" + elif raw in {"fast", "on"}: + nv = "fast" + elif raw in {"normal", "off"}: + nv = "normal" + else: + return _err(rid, 4002, f"unknown fast mode: {value}") + + overrides = None + if nv == "fast": + from hermes_cli.models import resolve_fast_mode_overrides + + if agent is not None: + target_model = getattr(agent, "model", None) + else: + # A pre-build session may already have a picked model riding in + # model_override (desktop draft) — validate fast support against + # THAT model, not the global default it will never use. + session_override = (session or {}).get("model_override") or {} + target_model = ( + session_override.get("model") + if isinstance(session_override, dict) + else None + ) or _resolve_model() + if not target_model: + return _err( + rid, + 4002, + "fast mode is not available without a selected model", + ) + overrides = resolve_fast_mode_overrides(target_model) + if overrides is None: + return _err( + rid, + 4002, + "fast mode is not available for this model", + ) + + if session is not None: + # Session-scoped, like `reasoning` below (global persistence is + # `--global` / Settings → Model territory). Writing config.yaml + # here let every desktop model-menu selection (per-model fast + # preset) rewrite the user's global agent.service_tier — flipping + # fast mode for every OTHER session, profile, CLI, and gateway + # build ("switch one session, switches everywhere"). Pin the + # create override so lazily-built sessions and rebuilds (/new, + # deferred resume) keep the choice; "" pins normal explicitly. + session["create_service_tier_override"] = ( + "priority" if nv == "fast" else "" + ) + else: + _write_config_key("agent.service_tier", nv) + if agent is not None: + agent.service_tier = "priority" if nv == "fast" else None + current_overrides = dict(getattr(agent, "request_overrides", {}) or {}) + current_overrides.pop("service_tier", None) + current_overrides.pop("speed", None) + if nv == "fast": + current_overrides.update(overrides) + agent.request_overrides = current_overrides + _persist_live_session_runtime(session) + _emit( + "session.info", + params.get("session_id", ""), + _session_info(agent, session), + ) + return _ok(rid, {"key": key, "value": nv}) + + if key == "busy": + raw = str(value or "").strip().lower() + if raw in {"", "status"}: + return _ok(rid, {"key": key, "value": _load_busy_input_mode()}) + if raw not in {"queue", "steer", "interrupt"}: + return _err(rid, 4002, f"unknown busy mode: {value}") + _write_config_key("display.busy_input_mode", raw) + return _ok(rid, {"key": key, "value": raw}) + + if key == "verbose": + cycle = ["off", "new", "all", "verbose"] + cur = ( + session.get("tool_progress_mode", _load_tool_progress_mode()) + if session + else _load_tool_progress_mode() + ) + if value and value != "cycle": + nv = str(value).strip().lower() + if nv not in cycle: + return _err(rid, 4002, f"unknown verbose mode: {value}") + else: + try: + idx = cycle.index(cur) + except ValueError: + idx = 2 + nv = cycle[(idx + 1) % len(cycle)] + _write_config_key("display.tool_progress", nv) + if session: + session["tool_progress_mode"] = nv + agent = session.get("agent") + if agent is not None: + agent.verbose_logging = nv == "verbose" + return _ok(rid, {"key": key, "value": nv}) + + if key == "focus": + # Focus view — display-only reduced-output mode (/focus). Composes with + # the tool_progress machinery rather than duplicating it: enabling it + # pins tool_progress to "off" (the same value /verbose off uses) after + # stashing the configured mode, and disabling it restores that mode. + # Nothing about the request payload changes. + from hermes_cli.focus_view import ( + FOCUS_TOOL_PROGRESS_MODE, + normalize_tool_progress_mode, + resolve_focus_arg, + ) + + cfg_f = _load_cfg() + _display_f = cfg_f.get("display") + d_f: dict = _display_f if isinstance(_display_f, dict) else {} + cur_focus = bool(d_f.get("focus_view", False)) + action, target = resolve_focus_arg(str(value or ""), cur_focus) + if action == "usage": + return _err(rid, 4002, f"unknown focus value: {value} (use on|off|status)") + if action == "status" or target is None: + return _ok( + rid, + { + "key": key, + "value": "on" if cur_focus else "off", + "tool_progress": _load_tool_progress_mode(), + }, + ) + + if target: + saved = normalize_tool_progress_mode( + (d_f.get("focus_saved_tool_progress") or _load_tool_progress_mode()) + if cur_focus + else _load_tool_progress_mode() + ) + _write_config_key("display.focus_saved_tool_progress", saved) + _write_config_key("display.tool_progress", FOCUS_TOOL_PROGRESS_MODE) + effective = FOCUS_TOOL_PROGRESS_MODE + else: + saved = normalize_tool_progress_mode( + d_f.get("focus_saved_tool_progress") or "all" + ) + _write_config_key("display.tool_progress", saved) + effective = saved + _write_config_key("display.focus_view", bool(target)) + + if session: + session["focus_view"] = bool(target) + session["tool_progress_mode"] = effective + agent_f = session.get("agent") + if agent_f is not None: + try: + agent_f.tool_progress_mode = effective + except Exception: + pass + return _ok( + rid, + { + "key": key, + "value": "on" if target else "off", + "tool_progress": effective, + }, + ) + + if key in {"approval_mode", "approvals.mode"}: + raw = str(value or "").strip().lower() + if raw not in _APPROVAL_MODES: + return _err( + rid, + 4002, + f"unknown approval mode: {value}; pick one of manual|smart|off", + ) + + _write_config_key("approvals.mode", raw) + for sid, sess in list(_sessions.items()): + agent = sess.get("agent") + if agent is not None: + _emit("session.info", sid, _session_info(agent, sess)) + return _ok(rid, {"key": "approvals.mode", "value": raw}) + + if key == "yolo": + # Approval bypass. Two scopes: + # scope="session" (default) — same as the TUI's Shift+Tab. Toggles + # ONLY this session's _session_yolo flag; never touches global + # config, so CLI / TUI / cron behavior is unaffected. + # scope="global" (Shift+click the zap) — flips the persistent global + # approvals.mode in config.yaml between "off" (bypass on) and + # "manual" (bypass off). This DOES affect every session, the CLI, + # the TUI, and cron, and survives restarts. + scope = str(params.get("scope") or "session").strip().lower() + try: + from tools.approval import ( + disable_session_yolo, + enable_session_yolo, + is_session_yolo_enabled, + ) + + raw = str(value or "").strip().lower() + + def _resolve_toggle(current: bool) -> bool: + if raw in {"1", "on", "true", "yes"}: + return True + if raw in {"0", "off", "false", "no"}: + return False + return not current + + if scope == "global": + from tools.approval import _normalize_approval_mode + + cfg = _load_cfg() + appr = cfg.get("approvals") if isinstance(cfg, dict) else None + if not isinstance(appr, dict): + appr = {} + current = _normalize_approval_mode(appr.get("mode", "manual")) == "off" + enable = _resolve_toggle(current) + # Toggle between full bypass and the default manual gate. We do + # not try to restore a prior "smart"/custom mode — the zap is a + # binary on/off affordance; users with bespoke modes set them in + # config.yaml. + _write_config_key("approvals.mode", "off" if enable else "manual") + nv = "1" if enable else "0" + # Reflect the global flip in every live session's indicator. + for sid, sess in list(_sessions.items()): + agent = sess.get("agent") + if agent is not None: + _emit("session.info", sid, _session_info(agent, sess)) + return _ok(rid, {"key": key, "value": nv, "scope": "global"}) + + if session: + current = is_session_yolo_enabled(session["session_key"]) + enable = _resolve_toggle(current) + if enable: + enable_session_yolo(session["session_key"]) + nv = "1" + else: + disable_session_yolo(session["session_key"]) + nv = "0" + agent = session.get("agent") + if agent is not None: + _emit( + "session.info", + params.get("session_id", ""), + _session_info(agent, session), + ) + else: + current = is_truthy_value(os.environ.get("HERMES_YOLO_MODE")) + enable = _resolve_toggle(current) + if enable: + os.environ["HERMES_YOLO_MODE"] = "1" + nv = "1" + else: + os.environ.pop("HERMES_YOLO_MODE", None) + nv = "0" + return _ok(rid, {"key": key, "value": nv, "scope": "session"}) + except Exception as e: + return _err(rid, 5001, str(e)) + + if key == "reasoning": + try: + from hermes_constants import parse_reasoning_effort + + arg = str(value or "").strip().lower() + scope = str(params.get("scope") or "").strip().lower() + global_scope = scope == "global" + if arg in {"show", "on"}: + cfg = _load_cfg_raw() # write-back round-trip + display = ( + cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + ) + sections = ( + display.get("sections") + if isinstance(display.get("sections"), dict) + else {} + ) + display["show_reasoning"] = True + sections["thinking"] = "expanded" + display["sections"] = sections + cfg["display"] = display + _save_cfg(cfg) + if session: + session["show_reasoning"] = True + return _ok(rid, {"key": key, "value": "show"}) + if arg in {"hide", "off"}: + cfg = _load_cfg_raw() # write-back round-trip + display = ( + cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + ) + sections = ( + display.get("sections") + if isinstance(display.get("sections"), dict) + else {} + ) + display["show_reasoning"] = False + sections["thinking"] = "hidden" + display["sections"] = sections + cfg["display"] = display + _save_cfg(cfg) + if session: + session["show_reasoning"] = False + return _ok(rid, {"key": key, "value": "hide"}) + + # /reasoning full | clamp — parity with the classic CLI's + # reasoning_full toggle. The TUI renders thinking as an + # expand/collapse section rather than a fixed 10-line recap, so + # full maps to sections.thinking=expanded and clamp to collapsed. + # display.reasoning_full is persisted too so the config key stays + # consistent across the CLI and TUI surfaces. + if arg in {"full", "all"}: + cfg = _load_cfg_raw() # write-back round-trip + display = ( + cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + ) + sections = ( + display.get("sections") + if isinstance(display.get("sections"), dict) + else {} + ) + display["reasoning_full"] = True + sections["thinking"] = "expanded" + display["sections"] = sections + cfg["display"] = display + _save_cfg(cfg) + return _ok(rid, {"key": key, "value": "full"}) + if arg in {"clamp", "collapse", "short"}: + cfg = _load_cfg_raw() # write-back round-trip + display = ( + cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + ) + sections = ( + display.get("sections") + if isinstance(display.get("sections"), dict) + else {} + ) + display["reasoning_full"] = False + sections["thinking"] = "collapsed" + display["sections"] = sections + cfg["display"] = display + _save_cfg(cfg) + return _ok(rid, {"key": key, "value": "clamp"}) + + parsed = parse_reasoning_effort(arg) + if parsed is None: + return _err(rid, 4002, f"unknown reasoning value: {value}") + if global_scope or session is None: + _write_config_key("agent.reasoning_effort", arg) + if session is not None: + session.pop("create_reasoning_override", None) + else: + # Session-scoped, like the messaging gateway's `/reasoning + # ` (global persistence is `--global` / Settings → + # Model territory). Writing config.yaml here let every + # desktop model-menu selection rewrite the user's global + # agent.reasoning_effort to the preset default. + session["create_reasoning_override"] = parsed + if session and session.get("agent") is not None: + session["agent"].reasoning_config = parsed + _persist_live_session_runtime(session) + _emit( + "session.info", + params.get("session_id", ""), + _session_info(session["agent"], session), + ) + return _ok(rid, {"key": key, "value": arg}) + except Exception as e: + return _err(rid, 5001, str(e)) + + if key == "details_mode": + nv = str(value or "").strip().lower() + if nv not in _DETAIL_MODES: + return _err(rid, 4002, f"unknown details_mode: {value}") + cfg = _load_cfg_raw() # write-back round-trip + display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + sections = ( + display.get("sections") if isinstance(display.get("sections"), dict) else {} + ) + display["details_mode"] = nv + for section in _DETAIL_SECTION_NAMES: + sections[section] = nv + display["sections"] = sections + cfg["display"] = display + _save_cfg(cfg) + return _ok(rid, {"key": key, "value": nv}) + + if key.startswith("details_mode."): + # Per-section override: `details_mode.
` writes to + # `display.sections.
`. Empty value clears the explicit + # override and lets frontend resolution apply built-in section defaults + # before the global details_mode. + section = key.split(".", 1)[1] + if section not in _DETAIL_SECTION_NAMES: + return _err(rid, 4002, f"unknown section: {section}") + + cfg = _load_cfg_raw() # write-back round-trip + display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + sections_cfg = ( + display.get("sections") if isinstance(display.get("sections"), dict) else {} + ) + + nv = str(value or "").strip().lower() + if not nv: + sections_cfg.pop(section, None) + display["sections"] = sections_cfg + cfg["display"] = display + _save_cfg(cfg) + return _ok(rid, {"key": key, "value": ""}) + + if nv not in _DETAIL_MODES: + return _err(rid, 4002, f"unknown details_mode: {value}") + + sections_cfg[section] = nv + display["sections"] = sections_cfg + cfg["display"] = display + _save_cfg(cfg) + return _ok(rid, {"key": key, "value": nv}) + + if key == "thinking_mode": + nv = str(value or "").strip().lower() + allowed_tm = frozenset({"collapsed", "truncated", "full"}) + if nv not in allowed_tm: + return _err(rid, 4002, f"unknown thinking_mode: {value}") + _write_config_key("display.thinking_mode", nv) + # Backward compatibility bridge: keep details_mode aligned. + _write_config_key( + "display.details_mode", "expanded" if nv == "full" else "collapsed" + ) + return _ok(rid, {"key": key, "value": nv}) + + if key == "density": + raw = str(value or "").strip().lower() + cfg0 = _load_cfg() + d0 = cfg0.get("display") if isinstance(cfg0.get("display"), dict) else {} + cur_b = bool(d0.get("tui_compact", False)) + if raw in {"", "toggle"}: + nv_b = not cur_b + elif raw == "on": + nv_b = True + elif raw == "off": + nv_b = False + else: + return _err(rid, 4002, f"unknown density value: {value}") + _write_config_key("display.tui_compact", nv_b) + return _ok(rid, {"key": key, "value": "on" if nv_b else "off"}) + + if key == "battery": + raw = str(value or "").strip().lower() + cfg0 = _load_cfg() + d0 = cfg0.get("display") if isinstance(cfg0.get("display"), dict) else {} + cur_b = bool(d0.get("battery", False)) + if raw in {"", "toggle"}: + nv_b = not cur_b + elif raw in {"on", "true", "yes"}: + nv_b = True + elif raw in {"off", "false", "no"}: + nv_b = False + else: + return _err(rid, 4002, f"unknown battery value: {value}") + _write_config_key("display.battery", nv_b) + return _ok(rid, {"key": key, "value": "on" if nv_b else "off"}) + + if key == "theme": + # TUI light/dark mode pin: 'light'/'dark' beat background + # auto-detection (xterm.js hosts misreport OSC 11); 'auto' trusts it. + raw = str(value or "").strip().lower() + if raw not in {"auto", "light", "dark"}: + return _err(rid, 4002, f"unknown theme value: {value} (use auto|light|dark)") + _write_config_key("display.tui_theme", raw) + return _ok(rid, {"key": key, "value": raw}) + + if key == "statusbar": + raw = str(value or "").strip().lower() + display = _load_cfg().get("display") + d0 = display if isinstance(display, dict) else {} + current = _coerce_statusbar(d0.get("tui_statusbar", "top")) + + if raw in {"", "toggle"}: + nv = "top" if current == "off" else "off" + elif raw == "on": + nv = "top" + elif raw in _STATUSBAR_MODES: + nv = raw + else: + return _err(rid, 4002, f"unknown statusbar value: {value}") + + _write_config_key("display.tui_statusbar", nv) + return _ok(rid, {"key": key, "value": nv}) + + if key == "mouse": + # Explicit None check rather than `value or ""` so falsy non-string + # inputs (0, False) reach the alias map as themselves — both map to + # 'off' via _MOUSE_TRACKING_ALIASES — instead of being collapsed to + # '' and triggering the toggle path. The slash command always passes + # a string, but programmatic JSON-RPC callers may send booleans. + raw = ("" if value is None else str(value)).strip().lower() + cfg = _load_cfg() + display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} + current = _display_mouse_tracking(display) + + if raw in {"", "toggle"}: + nv = "all" if current == "off" else "off" + elif raw in _MOUSE_TRACKING_ALIASES: + nv = _MOUSE_TRACKING_ALIASES[raw] + else: + return _err(rid, 4002, f"unknown mouse value: {value}") + + _write_config_key("display.mouse_tracking", nv) + return _ok(rid, {"key": key, "value": nv}) + + if key == "indicator": + # Use an explicit None check rather than `value or ""` so falsy + # non-string inputs (0, False, []) still surface as themselves + # in the error message instead of looking like a blank value. + raw = ("" if value is None else str(value)).strip().lower() + if raw not in INDICATOR_STYLES: + return _err( + rid, + 4002, + f"unknown indicator: {raw!r}; pick one of {'|'.join(INDICATOR_STYLES)}", + ) + _write_config_key("display.tui_status_indicator", raw) + return _ok(rid, {"key": key, "value": raw}) + + if key in {"cwd", "terminal.cwd", "workdir"}: + raw = str(value or "").strip() + if not raw: + return _err(rid, 4002, "cwd required") + cwd = os.path.abspath(os.path.expanduser(raw)) + if not os.path.isdir(cwd): + return _err(rid, 4002, f"working directory does not exist: {raw}") + _write_config_key("terminal.cwd", cwd) + os.environ["TERMINAL_CWD"] = cwd + return _ok( + rid, + {"key": "terminal.cwd", "value": cwd, "cwd": cwd, "branch": _git_branch_for_cwd(cwd)}, + ) + + if key in {"prompt", "personality", "skin"}: + try: + cfg = _load_cfg_raw() # write-back round-trip ("prompt" saves cfg) + if key == "prompt": + if value == "clear": + cfg.pop("custom_prompt", None) + nv = "" + else: + cfg["custom_prompt"] = value + nv = value + _save_cfg(cfg) + elif key == "personality": + sid_key = params.get("session_id", "") + pname, new_prompt = _validate_personality(str(value or ""), cfg) + _write_config_key("display.personality", pname) + _write_config_key("agent.system_prompt", new_prompt) + nv = str(value or "none") + history_reset, info = _apply_personality_to_session( + sid_key, session, new_prompt, pname + ) + else: + _write_config_key(f"display.{key}", value) + nv = value + if key == "skin": + # Every connected surface repaints, not just the RPC's + # client; then sync the watcher baseline so the poll loop + # doesn't re-broadcast the skin this RPC just applied. + _broadcast_global_event("skin.changed", resolve_skin()) + _note_skin_broadcast() + resp = {"key": key, "value": nv} + if key == "personality": + resp["history_reset"] = history_reset + if info is not None: + resp["info"] = info + return _ok(rid, resp) + except Exception as e: + return _err(rid, 5001, str(e)) + + return _err(rid, 4002, f"unknown config key: {key}") + + def register(server) -> None: """Bind this module's handlers onto ``server``'s globals and registry.""" _registry.install(server) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index a36a539408b1..bbc8d546e84c 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -10462,706 +10462,6 @@ def _respond(rid, params, key, *, allow_expired=False): return _ok(rid, {"status": "ok"}) -# ── Methods: config ────────────────────────────────────────────────── - - -# NOTE: config.set intentionally stays in server.py for now — the in-flight -# opt/model-resolution-core PR touches its body; move it to methods_config.py -# in a follow-up once that PR lands. -@method("config.set") -def _(rid, params: dict) -> dict: - key, value = params.get("key", ""), params.get("value", "") - session = _sessions.get(params.get("session_id", "")) - - if key == "model": - try: - if not value: - return _err(rid, 4002, "model value required") - if session: - from hermes_cli.model_switch import parse_model_switch_args - - # A live swap can't run in-place while a turn streams: - # agent.switch_model() mutates self.model / self.provider / - # self.base_url / self.client, and the worker thread running - # agent.run_conversation reads those every iteration — a - # mid-turn swap can fire an HTTP request with the new base_url - # but old model (400/404s). So instead of rejecting the pick - # (the old 4009), stash it and apply it at the NEXT turn start - # (_apply_pending_model_switch), where nothing is in flight. - # The user gets to pick, keep typing, and send the next turn on - # the new model without waiting for the swap or interrupting. - if session.get("running"): - parsed = parse_model_switch_args(value) - try: - pending_model = parsed.model_input - except Exception: - pending_model = str(value) - session["pending_model_switch"] = { - "raw": value, - "confirm_expensive_model": bool( - params.get("confirm_expensive_model", False) - ), - # The resolved model/provider the next turn will run on. - # _session_info reports these while the switch is pending - # so the end-of-turn settle keeps showing the user's pick - # instead of blipping back to the still-live old model. - "display_model": pending_model, - "display_provider": ( - getattr(parsed, "explicit_provider", "") or "" - ).strip(), - } - return _ok( - rid, - { - "key": key, - "value": pending_model, - "warning": "", - "confirm_required": False, - "confirm_message": "", - "scope": "session", - "deferred": True, - }, - ) - parsed_flags = parse_model_switch_args(value) - explicit_provider = parsed_flags.explicit_provider - if session.get("agent") is None and not explicit_provider.strip(): - session_id = params.get("session_id", "") - _start_agent_build(session_id, session) - init_err = _wait_agent(session, rid) - if init_err: - return init_err - if session.get("agent") is None: - return _err(rid, 5032, "agent initialization failed") - result = _apply_model_switch( - params.get("session_id", ""), - session, - value, - confirm_expensive_model=bool( - params.get("confirm_expensive_model", False) - ), - parsed_flags=parsed_flags, - ) - else: - result = _apply_model_switch( - "", - {"agent": None}, - value, - confirm_expensive_model=bool( - params.get("confirm_expensive_model", False) - ), - ) - return _ok( - rid, - { - "key": key, - "value": result["value"], - "warning": result["warning"], - "confirm_required": result.get("confirm_required", False), - "confirm_message": result.get("confirm_message", ""), - "scope": result.get("scope", "session"), - }, - ) - except Exception as e: - return _err(rid, 5001, str(e)) - - if key == "fast": - raw = str(value or "").strip().lower() - agent = session.get("agent") if session else None - if agent is not None: - current_fast = getattr(agent, "service_tier", None) == "priority" - elif session is not None and session.get("create_service_tier_override") is not None: - # Pre-build session with a pinned tier (desktop draft pick or an - # earlier session-scoped toggle) — report/toggle from the pin, not - # the global default. - current_fast = session["create_service_tier_override"] == "priority" - else: - current_fast = _load_service_tier() == "priority" - - if raw in {"status"}: - return _ok( - rid, - {"key": key, "value": "fast" if current_fast else "normal"}, - ) - - if raw in {"", "toggle"}: - nv = "normal" if current_fast else "fast" - elif raw in {"fast", "on"}: - nv = "fast" - elif raw in {"normal", "off"}: - nv = "normal" - else: - return _err(rid, 4002, f"unknown fast mode: {value}") - - overrides = None - if nv == "fast": - from hermes_cli.models import resolve_fast_mode_overrides - - if agent is not None: - target_model = getattr(agent, "model", None) - else: - # A pre-build session may already have a picked model riding in - # model_override (desktop draft) — validate fast support against - # THAT model, not the global default it will never use. - session_override = (session or {}).get("model_override") or {} - target_model = ( - session_override.get("model") - if isinstance(session_override, dict) - else None - ) or _resolve_model() - if not target_model: - return _err( - rid, - 4002, - "fast mode is not available without a selected model", - ) - overrides = resolve_fast_mode_overrides(target_model) - if overrides is None: - return _err( - rid, - 4002, - "fast mode is not available for this model", - ) - - if session is not None: - # Session-scoped, like `reasoning` below (global persistence is - # `--global` / Settings → Model territory). Writing config.yaml - # here let every desktop model-menu selection (per-model fast - # preset) rewrite the user's global agent.service_tier — flipping - # fast mode for every OTHER session, profile, CLI, and gateway - # build ("switch one session, switches everywhere"). Pin the - # create override so lazily-built sessions and rebuilds (/new, - # deferred resume) keep the choice; "" pins normal explicitly. - session["create_service_tier_override"] = ( - "priority" if nv == "fast" else "" - ) - else: - _write_config_key("agent.service_tier", nv) - if agent is not None: - agent.service_tier = "priority" if nv == "fast" else None - current_overrides = dict(getattr(agent, "request_overrides", {}) or {}) - current_overrides.pop("service_tier", None) - current_overrides.pop("speed", None) - if nv == "fast": - current_overrides.update(overrides) - agent.request_overrides = current_overrides - _persist_live_session_runtime(session) - _emit( - "session.info", - params.get("session_id", ""), - _session_info(agent, session), - ) - return _ok(rid, {"key": key, "value": nv}) - - if key == "busy": - raw = str(value or "").strip().lower() - if raw in {"", "status"}: - return _ok(rid, {"key": key, "value": _load_busy_input_mode()}) - if raw not in {"queue", "steer", "interrupt"}: - return _err(rid, 4002, f"unknown busy mode: {value}") - _write_config_key("display.busy_input_mode", raw) - return _ok(rid, {"key": key, "value": raw}) - - if key == "verbose": - cycle = ["off", "new", "all", "verbose"] - cur = ( - session.get("tool_progress_mode", _load_tool_progress_mode()) - if session - else _load_tool_progress_mode() - ) - if value and value != "cycle": - nv = str(value).strip().lower() - if nv not in cycle: - return _err(rid, 4002, f"unknown verbose mode: {value}") - else: - try: - idx = cycle.index(cur) - except ValueError: - idx = 2 - nv = cycle[(idx + 1) % len(cycle)] - _write_config_key("display.tool_progress", nv) - if session: - session["tool_progress_mode"] = nv - agent = session.get("agent") - if agent is not None: - agent.verbose_logging = nv == "verbose" - return _ok(rid, {"key": key, "value": nv}) - - if key == "focus": - # Focus view — display-only reduced-output mode (/focus). Composes with - # the tool_progress machinery rather than duplicating it: enabling it - # pins tool_progress to "off" (the same value /verbose off uses) after - # stashing the configured mode, and disabling it restores that mode. - # Nothing about the request payload changes. - from hermes_cli.focus_view import ( - FOCUS_TOOL_PROGRESS_MODE, - normalize_tool_progress_mode, - resolve_focus_arg, - ) - - cfg_f = _load_cfg() - _display_f = cfg_f.get("display") - d_f: dict = _display_f if isinstance(_display_f, dict) else {} - cur_focus = bool(d_f.get("focus_view", False)) - action, target = resolve_focus_arg(str(value or ""), cur_focus) - if action == "usage": - return _err(rid, 4002, f"unknown focus value: {value} (use on|off|status)") - if action == "status" or target is None: - return _ok( - rid, - { - "key": key, - "value": "on" if cur_focus else "off", - "tool_progress": _load_tool_progress_mode(), - }, - ) - - if target: - saved = normalize_tool_progress_mode( - (d_f.get("focus_saved_tool_progress") or _load_tool_progress_mode()) - if cur_focus - else _load_tool_progress_mode() - ) - _write_config_key("display.focus_saved_tool_progress", saved) - _write_config_key("display.tool_progress", FOCUS_TOOL_PROGRESS_MODE) - effective = FOCUS_TOOL_PROGRESS_MODE - else: - saved = normalize_tool_progress_mode( - d_f.get("focus_saved_tool_progress") or "all" - ) - _write_config_key("display.tool_progress", saved) - effective = saved - _write_config_key("display.focus_view", bool(target)) - - if session: - session["focus_view"] = bool(target) - session["tool_progress_mode"] = effective - agent_f = session.get("agent") - if agent_f is not None: - try: - agent_f.tool_progress_mode = effective - except Exception: - pass - return _ok( - rid, - { - "key": key, - "value": "on" if target else "off", - "tool_progress": effective, - }, - ) - - if key in {"approval_mode", "approvals.mode"}: - raw = str(value or "").strip().lower() - if raw not in _APPROVAL_MODES: - return _err( - rid, - 4002, - f"unknown approval mode: {value}; pick one of manual|smart|off", - ) - - _write_config_key("approvals.mode", raw) - for sid, sess in list(_sessions.items()): - agent = sess.get("agent") - if agent is not None: - _emit("session.info", sid, _session_info(agent, sess)) - return _ok(rid, {"key": "approvals.mode", "value": raw}) - - if key == "yolo": - # Approval bypass. Two scopes: - # scope="session" (default) — same as the TUI's Shift+Tab. Toggles - # ONLY this session's _session_yolo flag; never touches global - # config, so CLI / TUI / cron behavior is unaffected. - # scope="global" (Shift+click the zap) — flips the persistent global - # approvals.mode in config.yaml between "off" (bypass on) and - # "manual" (bypass off). This DOES affect every session, the CLI, - # the TUI, and cron, and survives restarts. - scope = str(params.get("scope") or "session").strip().lower() - try: - from tools.approval import ( - disable_session_yolo, - enable_session_yolo, - is_session_yolo_enabled, - ) - - raw = str(value or "").strip().lower() - - def _resolve_toggle(current: bool) -> bool: - if raw in {"1", "on", "true", "yes"}: - return True - if raw in {"0", "off", "false", "no"}: - return False - return not current - - if scope == "global": - from tools.approval import _normalize_approval_mode - - cfg = _load_cfg() - appr = cfg.get("approvals") if isinstance(cfg, dict) else None - if not isinstance(appr, dict): - appr = {} - current = _normalize_approval_mode(appr.get("mode", "manual")) == "off" - enable = _resolve_toggle(current) - # Toggle between full bypass and the default manual gate. We do - # not try to restore a prior "smart"/custom mode — the zap is a - # binary on/off affordance; users with bespoke modes set them in - # config.yaml. - _write_config_key("approvals.mode", "off" if enable else "manual") - nv = "1" if enable else "0" - # Reflect the global flip in every live session's indicator. - for sid, sess in list(_sessions.items()): - agent = sess.get("agent") - if agent is not None: - _emit("session.info", sid, _session_info(agent, sess)) - return _ok(rid, {"key": key, "value": nv, "scope": "global"}) - - if session: - current = is_session_yolo_enabled(session["session_key"]) - enable = _resolve_toggle(current) - if enable: - enable_session_yolo(session["session_key"]) - nv = "1" - else: - disable_session_yolo(session["session_key"]) - nv = "0" - agent = session.get("agent") - if agent is not None: - _emit( - "session.info", - params.get("session_id", ""), - _session_info(agent, session), - ) - else: - current = is_truthy_value(os.environ.get("HERMES_YOLO_MODE")) - enable = _resolve_toggle(current) - if enable: - os.environ["HERMES_YOLO_MODE"] = "1" - nv = "1" - else: - os.environ.pop("HERMES_YOLO_MODE", None) - nv = "0" - return _ok(rid, {"key": key, "value": nv, "scope": "session"}) - except Exception as e: - return _err(rid, 5001, str(e)) - - if key == "reasoning": - try: - from hermes_constants import parse_reasoning_effort - - arg = str(value or "").strip().lower() - scope = str(params.get("scope") or "").strip().lower() - global_scope = scope == "global" - if arg in {"show", "on"}: - cfg = _load_cfg_raw() # write-back round-trip - display = ( - cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - ) - sections = ( - display.get("sections") - if isinstance(display.get("sections"), dict) - else {} - ) - display["show_reasoning"] = True - sections["thinking"] = "expanded" - display["sections"] = sections - cfg["display"] = display - _save_cfg(cfg) - if session: - session["show_reasoning"] = True - return _ok(rid, {"key": key, "value": "show"}) - if arg in {"hide", "off"}: - cfg = _load_cfg_raw() # write-back round-trip - display = ( - cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - ) - sections = ( - display.get("sections") - if isinstance(display.get("sections"), dict) - else {} - ) - display["show_reasoning"] = False - sections["thinking"] = "hidden" - display["sections"] = sections - cfg["display"] = display - _save_cfg(cfg) - if session: - session["show_reasoning"] = False - return _ok(rid, {"key": key, "value": "hide"}) - - # /reasoning full | clamp — parity with the classic CLI's - # reasoning_full toggle. The TUI renders thinking as an - # expand/collapse section rather than a fixed 10-line recap, so - # full maps to sections.thinking=expanded and clamp to collapsed. - # display.reasoning_full is persisted too so the config key stays - # consistent across the CLI and TUI surfaces. - if arg in {"full", "all"}: - cfg = _load_cfg_raw() # write-back round-trip - display = ( - cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - ) - sections = ( - display.get("sections") - if isinstance(display.get("sections"), dict) - else {} - ) - display["reasoning_full"] = True - sections["thinking"] = "expanded" - display["sections"] = sections - cfg["display"] = display - _save_cfg(cfg) - return _ok(rid, {"key": key, "value": "full"}) - if arg in {"clamp", "collapse", "short"}: - cfg = _load_cfg_raw() # write-back round-trip - display = ( - cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - ) - sections = ( - display.get("sections") - if isinstance(display.get("sections"), dict) - else {} - ) - display["reasoning_full"] = False - sections["thinking"] = "collapsed" - display["sections"] = sections - cfg["display"] = display - _save_cfg(cfg) - return _ok(rid, {"key": key, "value": "clamp"}) - - parsed = parse_reasoning_effort(arg) - if parsed is None: - return _err(rid, 4002, f"unknown reasoning value: {value}") - if global_scope or session is None: - _write_config_key("agent.reasoning_effort", arg) - if session is not None: - session.pop("create_reasoning_override", None) - else: - # Session-scoped, like the messaging gateway's `/reasoning - # ` (global persistence is `--global` / Settings → - # Model territory). Writing config.yaml here let every - # desktop model-menu selection rewrite the user's global - # agent.reasoning_effort to the preset default. - session["create_reasoning_override"] = parsed - if session and session.get("agent") is not None: - session["agent"].reasoning_config = parsed - _persist_live_session_runtime(session) - _emit( - "session.info", - params.get("session_id", ""), - _session_info(session["agent"], session), - ) - return _ok(rid, {"key": key, "value": arg}) - except Exception as e: - return _err(rid, 5001, str(e)) - - if key == "details_mode": - nv = str(value or "").strip().lower() - if nv not in _DETAIL_MODES: - return _err(rid, 4002, f"unknown details_mode: {value}") - cfg = _load_cfg_raw() # write-back round-trip - display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - sections = ( - display.get("sections") if isinstance(display.get("sections"), dict) else {} - ) - display["details_mode"] = nv - for section in _DETAIL_SECTION_NAMES: - sections[section] = nv - display["sections"] = sections - cfg["display"] = display - _save_cfg(cfg) - return _ok(rid, {"key": key, "value": nv}) - - if key.startswith("details_mode."): - # Per-section override: `details_mode.
` writes to - # `display.sections.
`. Empty value clears the explicit - # override and lets frontend resolution apply built-in section defaults - # before the global details_mode. - section = key.split(".", 1)[1] - if section not in _DETAIL_SECTION_NAMES: - return _err(rid, 4002, f"unknown section: {section}") - - cfg = _load_cfg_raw() # write-back round-trip - display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - sections_cfg = ( - display.get("sections") if isinstance(display.get("sections"), dict) else {} - ) - - nv = str(value or "").strip().lower() - if not nv: - sections_cfg.pop(section, None) - display["sections"] = sections_cfg - cfg["display"] = display - _save_cfg(cfg) - return _ok(rid, {"key": key, "value": ""}) - - if nv not in _DETAIL_MODES: - return _err(rid, 4002, f"unknown details_mode: {value}") - - sections_cfg[section] = nv - display["sections"] = sections_cfg - cfg["display"] = display - _save_cfg(cfg) - return _ok(rid, {"key": key, "value": nv}) - - if key == "thinking_mode": - nv = str(value or "").strip().lower() - allowed_tm = frozenset({"collapsed", "truncated", "full"}) - if nv not in allowed_tm: - return _err(rid, 4002, f"unknown thinking_mode: {value}") - _write_config_key("display.thinking_mode", nv) - # Backward compatibility bridge: keep details_mode aligned. - _write_config_key( - "display.details_mode", "expanded" if nv == "full" else "collapsed" - ) - return _ok(rid, {"key": key, "value": nv}) - - if key == "density": - raw = str(value or "").strip().lower() - cfg0 = _load_cfg() - d0 = cfg0.get("display") if isinstance(cfg0.get("display"), dict) else {} - cur_b = bool(d0.get("tui_compact", False)) - if raw in {"", "toggle"}: - nv_b = not cur_b - elif raw == "on": - nv_b = True - elif raw == "off": - nv_b = False - else: - return _err(rid, 4002, f"unknown density value: {value}") - _write_config_key("display.tui_compact", nv_b) - return _ok(rid, {"key": key, "value": "on" if nv_b else "off"}) - - if key == "battery": - raw = str(value or "").strip().lower() - cfg0 = _load_cfg() - d0 = cfg0.get("display") if isinstance(cfg0.get("display"), dict) else {} - cur_b = bool(d0.get("battery", False)) - if raw in {"", "toggle"}: - nv_b = not cur_b - elif raw in {"on", "true", "yes"}: - nv_b = True - elif raw in {"off", "false", "no"}: - nv_b = False - else: - return _err(rid, 4002, f"unknown battery value: {value}") - _write_config_key("display.battery", nv_b) - return _ok(rid, {"key": key, "value": "on" if nv_b else "off"}) - - if key == "theme": - # TUI light/dark mode pin: 'light'/'dark' beat background - # auto-detection (xterm.js hosts misreport OSC 11); 'auto' trusts it. - raw = str(value or "").strip().lower() - if raw not in {"auto", "light", "dark"}: - return _err(rid, 4002, f"unknown theme value: {value} (use auto|light|dark)") - _write_config_key("display.tui_theme", raw) - return _ok(rid, {"key": key, "value": raw}) - - if key == "statusbar": - raw = str(value or "").strip().lower() - display = _load_cfg().get("display") - d0 = display if isinstance(display, dict) else {} - current = _coerce_statusbar(d0.get("tui_statusbar", "top")) - - if raw in {"", "toggle"}: - nv = "top" if current == "off" else "off" - elif raw == "on": - nv = "top" - elif raw in _STATUSBAR_MODES: - nv = raw - else: - return _err(rid, 4002, f"unknown statusbar value: {value}") - - _write_config_key("display.tui_statusbar", nv) - return _ok(rid, {"key": key, "value": nv}) - - if key == "mouse": - # Explicit None check rather than `value or ""` so falsy non-string - # inputs (0, False) reach the alias map as themselves — both map to - # 'off' via _MOUSE_TRACKING_ALIASES — instead of being collapsed to - # '' and triggering the toggle path. The slash command always passes - # a string, but programmatic JSON-RPC callers may send booleans. - raw = ("" if value is None else str(value)).strip().lower() - cfg = _load_cfg() - display = cfg.get("display") if isinstance(cfg.get("display"), dict) else {} - current = _display_mouse_tracking(display) - - if raw in {"", "toggle"}: - nv = "all" if current == "off" else "off" - elif raw in _MOUSE_TRACKING_ALIASES: - nv = _MOUSE_TRACKING_ALIASES[raw] - else: - return _err(rid, 4002, f"unknown mouse value: {value}") - - _write_config_key("display.mouse_tracking", nv) - return _ok(rid, {"key": key, "value": nv}) - - if key == "indicator": - # Use an explicit None check rather than `value or ""` so falsy - # non-string inputs (0, False, []) still surface as themselves - # in the error message instead of looking like a blank value. - raw = ("" if value is None else str(value)).strip().lower() - if raw not in INDICATOR_STYLES: - return _err( - rid, - 4002, - f"unknown indicator: {raw!r}; pick one of {'|'.join(INDICATOR_STYLES)}", - ) - _write_config_key("display.tui_status_indicator", raw) - return _ok(rid, {"key": key, "value": raw}) - - if key in {"cwd", "terminal.cwd", "workdir"}: - raw = str(value or "").strip() - if not raw: - return _err(rid, 4002, "cwd required") - cwd = os.path.abspath(os.path.expanduser(raw)) - if not os.path.isdir(cwd): - return _err(rid, 4002, f"working directory does not exist: {raw}") - _write_config_key("terminal.cwd", cwd) - os.environ["TERMINAL_CWD"] = cwd - return _ok( - rid, - {"key": "terminal.cwd", "value": cwd, "cwd": cwd, "branch": _git_branch_for_cwd(cwd)}, - ) - - if key in {"prompt", "personality", "skin"}: - try: - cfg = _load_cfg_raw() # write-back round-trip ("prompt" saves cfg) - if key == "prompt": - if value == "clear": - cfg.pop("custom_prompt", None) - nv = "" - else: - cfg["custom_prompt"] = value - nv = value - _save_cfg(cfg) - elif key == "personality": - sid_key = params.get("session_id", "") - pname, new_prompt = _validate_personality(str(value or ""), cfg) - _write_config_key("display.personality", pname) - _write_config_key("agent.system_prompt", new_prompt) - nv = str(value or "none") - history_reset, info = _apply_personality_to_session( - sid_key, session, new_prompt, pname - ) - else: - _write_config_key(f"display.{key}", value) - nv = value - if key == "skin": - # Every connected surface repaints, not just the RPC's - # client; then sync the watcher baseline so the poll loop - # doesn't re-broadcast the skin this RPC just applied. - _broadcast_global_event("skin.changed", resolve_skin()) - _note_skin_broadcast() - resp = {"key": key, "value": nv} - if key == "personality": - resp["history_reset"] = history_reset - if info is not None: - resp["info"] = info - return _ok(rid, resp) - except Exception as e: - return _err(rid, 5001, str(e)) - - return _err(rid, 4002, f"unknown config key: {key}") - - # --------------------------------------------------------------------------- # Projects — first-class, per-profile, multi-folder workspaces # ---------------------------------------------------------------------------