Merge pull request #4692 from NousResearch/feat/ink-refactor

Feat/ink refactor

tests/cli/test_cli_copy_command.py

@@ -0,0 +1,71 @@
+"""Tests for CLI /copy command."""
+
+from unittest.mock import MagicMock, patch
+
+from cli import HermesCLI
+
+
+def _make_cli() -> HermesCLI:
+    cli_obj = HermesCLI.__new__(HermesCLI)
+    cli_obj.config = {}
+    cli_obj.console = MagicMock()
+    cli_obj.agent = None
+    cli_obj.conversation_history = []
+    cli_obj.session_id = "sess-copy-test"
+    cli_obj._pending_input = MagicMock()
+    cli_obj._app = None
+    return cli_obj
+
+
+def test_copy_copies_latest_assistant_message():
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [
+        {"role": "user", "content": "hi"},
+        {"role": "assistant", "content": "first"},
+        {"role": "assistant", "content": "latest"},
+    ]
+
+    with patch.object(cli_obj, "_write_osc52_clipboard") as mock_copy:
+        result = cli_obj.process_command("/copy")
+
+    assert result is True
+    mock_copy.assert_called_once_with("latest")
+
+
+def test_copy_with_index_uses_requested_assistant_message():
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [
+        {"role": "assistant", "content": "one"},
+        {"role": "assistant", "content": "two"},
+    ]
+
+    with patch.object(cli_obj, "_write_osc52_clipboard") as mock_copy:
+        cli_obj.process_command("/copy 1")
+
+    mock_copy.assert_called_once_with("one")
+
+
+def test_copy_strips_reasoning_blocks_before_copy():
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [
+        {
+            "role": "assistant",
+            "content": "<REASONING_SCRATCHPAD>internal</REASONING_SCRATCHPAD>\nVisible answer",
+        }
+    ]
+
+    with patch.object(cli_obj, "_write_osc52_clipboard") as mock_copy:
+        cli_obj.process_command("/copy")
+
+    mock_copy.assert_called_once_with("Visible answer")
+
+
+def test_copy_invalid_index_does_not_copy():
+    cli_obj = _make_cli()
+    cli_obj.conversation_history = [{"role": "assistant", "content": "only"}]
+
+    with patch.object(cli_obj, "_write_osc52_clipboard") as mock_copy, patch("cli._cprint") as mock_print:
+        cli_obj.process_command("/copy 99")
+
+    mock_copy.assert_not_called()
+    assert any("Invalid response number" in str(call) for call in mock_print.call_args_list)

tests/gateway/test_command_bypass_active_session.py

@@ -160,6 +160,30 @@ class TestCommandBypassActiveSession:
         assert sk not in adapter._pending_messages
         assert any("handled:status" in r for r in adapter.sent_responses)
 
+    @pytest.mark.asyncio
+    async def test_agents_bypasses_guard(self):
+        """/agents must bypass so active-task queries don't interrupt runs."""
+        adapter = _make_adapter()
+        sk = _session_key()
+        adapter._active_sessions[sk] = asyncio.Event()
+
+        await adapter.handle_message(_make_event("/agents"))
+
+        assert sk not in adapter._pending_messages
+        assert any("handled:agents" in r for r in adapter.sent_responses)
+
+    @pytest.mark.asyncio
+    async def test_tasks_alias_bypasses_guard(self):
+        """/tasks alias must bypass active-session guard too."""
+        adapter = _make_adapter()
+        sk = _session_key()
+        adapter._active_sessions[sk] = asyncio.Event()
+
+        await adapter.handle_message(_make_event("/tasks"))
+
+        assert sk not in adapter._pending_messages
+        assert any("handled:tasks" in r for r in adapter.sent_responses)
+
     @pytest.mark.asyncio
     async def test_background_bypasses_guard(self):
         """/background must bypass so it spawns a parallel task, not an interrupt."""

tests/gateway/test_status_command.py

@@ -1,6 +1,7 @@
 """Tests for gateway /status behavior and token persistence."""
 
 from datetime import datetime
+import time
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock
 
@@ -111,6 +112,75 @@ async def test_status_command_includes_session_title_when_present():
     assert "**Title:** My titled session" in result
 
 
+@pytest.mark.asyncio
+async def test_agents_command_reports_active_agents_and_processes(monkeypatch):
+    session_key = build_session_key(_make_source())
+    session_entry = SessionEntry(
+        session_key=session_key,
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        total_tokens=0,
+    )
+    runner = _make_runner(session_entry)
+    running_agent = SimpleNamespace(
+        session_id="sess-running",
+        model="openrouter/test-model",
+        interrupt=MagicMock(),
+        get_activity_summary=lambda: {"seconds_since_activity": 0},
+    )
+    runner._running_agents[session_key] = running_agent
+    runner._running_agents_ts = {session_key: time.time() - 8}
+    runner._background_tasks = set()
+
+    class _FakeRegistry:
+        def list_sessions(self):
+            return [
+                {
+                    "session_id": "proc-1",
+                    "status": "running",
+                    "uptime_seconds": 17,
+                    "command": "sleep 30",
+                }
+            ]
+
+    monkeypatch.setattr("tools.process_registry.process_registry", _FakeRegistry())
+
+    result = await runner._handle_message(_make_event("/agents"))
+
+    assert "**Active agents:** 1" in result
+    assert "**Running background processes:** 1" in result
+    assert "proc-1" in result
+    running_agent.interrupt.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_tasks_alias_routes_to_agents_command(monkeypatch):
+    session_entry = SessionEntry(
+        session_key=build_session_key(_make_source()),
+        session_id="sess-1",
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        platform=Platform.TELEGRAM,
+        chat_type="dm",
+        total_tokens=0,
+    )
+    runner = _make_runner(session_entry)
+    runner._background_tasks = set()
+
+    class _FakeRegistry:
+        def list_sessions(self):
+            return []
+
+    monkeypatch.setattr("tools.process_registry.process_registry", _FakeRegistry())
+
+    result = await runner._handle_message(_make_event("/tasks"))
+
+    assert "Active Agents & Tasks" in result
+
+
 @pytest.mark.asyncio
 async def test_handle_message_persists_agent_token_counts(monkeypatch):
     import gateway.run as gateway_run

tests/hermes_cli/test_cmd_update.py

@@ -106,6 +106,49 @@ class TestCmdUpdateBranchFallback:
         pull_cmds = [c for c in commands if "pull" in c]
         assert len(pull_cmds) == 0
 
+    @patch("shutil.which")
+    @patch("subprocess.run")
+    def test_update_refreshes_repo_and_tui_node_dependencies(
+        self, mock_run, mock_which, mock_args
+    ):
+        mock_which.side_effect = {"uv": "/usr/bin/uv", "npm": "/usr/bin/npm"}.get
+        mock_run.side_effect = _make_run_side_effect(
+            branch="main", verify_ok=True, commit_count="1"
+        )
+
+        cmd_update(mock_args)
+
+        npm_calls = [
+            (call.args[0], call.kwargs.get("cwd"))
+            for call in mock_run.call_args_list
+            if call.args and call.args[0][0] == "/usr/bin/npm"
+        ]
+
+        assert npm_calls == [
+            (
+                [
+                    "/usr/bin/npm",
+                    "install",
+                    "--silent",
+                    "--no-fund",
+                    "--no-audit",
+                    "--progress=false",
+                ],
+                PROJECT_ROOT,
+            ),
+            (
+                [
+                    "/usr/bin/npm",
+                    "install",
+                    "--silent",
+                    "--no-fund",
+                    "--no-audit",
+                    "--progress=false",
+                ],
+                PROJECT_ROOT / "ui-tui",
+            ),
+        ]
+
     def test_update_non_interactive_skips_migration_prompt(self, mock_args, capsys):
         """When stdin/stdout aren't TTYs, config migration prompt is skipped."""
         with patch("shutil.which", return_value=None), patch(

tests/hermes_cli/test_commands.py

@@ -93,6 +93,8 @@ class TestResolveCommand:
     def test_canonical_name_resolves(self):
         assert resolve_command("help").name == "help"
         assert resolve_command("background").name == "background"
+        assert resolve_command("copy").name == "copy"
+        assert resolve_command("agents").name == "agents"
 
     def test_alias_resolves_to_canonical(self):
         assert resolve_command("bg").name == "background"
@@ -102,6 +104,7 @@ class TestResolveCommand:
         assert resolve_command("gateway").name == "platforms"
         assert resolve_command("set-home").name == "sethome"
         assert resolve_command("reload_mcp").name == "reload-mcp"
+        assert resolve_command("tasks").name == "agents"
 
     def test_leading_slash_stripped(self):
         assert resolve_command("/help").name == "help"

tests/hermes_cli/test_model_validation.py

@@ -403,7 +403,8 @@ class TestValidateFormatChecks:
 
     def test_no_slash_model_rejected_if_not_in_api(self):
         result = _validate("gpt-5.4", api_models=["openai/gpt-5.4"])
-        assert result["accepted"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
         assert "not found" in result["message"]
 
 
@@ -429,10 +430,10 @@ class TestValidateApiFound:
 # -- validate — API not found ------------------------------------------------
 
 class TestValidateApiNotFound:
-    def test_model_not_in_api_accepted_with_warning(self):
+    def test_model_not_in_api_rejected_with_guidance(self):
         result = _validate("anthropic/claude-nonexistent")
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
         assert "not found" in result["message"]
 
     def test_warning_includes_suggestions(self):
@@ -456,30 +457,29 @@ class TestValidateApiNotFound:
         assert "not found" in result["message"]
 
 
-# -- validate — API unreachable — accept and persist everything ----------------
+# -- validate — API unreachable — reject with guidance ----------------
 
 class TestValidateApiFallback:
-    def test_any_model_accepted_when_api_down(self):
+    def test_any_model_rejected_when_api_down(self):
         result = _validate("anthropic/claude-opus-4.6", api_models=None)
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
 
-    def test_unknown_model_also_accepted_when_api_down(self):
-        """No hardcoded catalog gatekeeping — accept, persist, and warn."""
+    def test_unknown_model_also_rejected_when_api_down(self):
         result = _validate("anthropic/claude-next-gen", api_models=None)
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
         assert "could not reach" in result["message"].lower()
 
-    def test_zai_model_accepted_when_api_down(self):
+    def test_zai_model_rejected_when_api_down(self):
         result = _validate("glm-5", provider="zai", api_models=None)
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
 
-    def test_unknown_provider_accepted_when_api_down(self):
+    def test_unknown_provider_rejected_when_api_down(self):
         result = _validate("some-model", provider="totally-unknown", api_models=None)
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
 
     def test_custom_endpoint_warns_with_probed_url_and_v1_hint(self):
         with patch(
@@ -499,8 +499,8 @@ class TestValidateApiFallback:
                 base_url="http://localhost:8000",
             )
 
-        assert result["accepted"] is True
-        assert result["persist"] is True
+        assert result["accepted"] is False
+        assert result["persist"] is False
         assert "http://localhost:8000/v1/models" in result["message"]
         assert "http://localhost:8000/v1" in result["message"]
 

tests/hermes_cli/test_tui_npm_install.py

@@ -0,0 +1,53 @@
+"""_tui_need_npm_install: auto npm when lockfile ahead of node_modules."""
+
+import os
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def main_mod():
+    import hermes_cli.main as m
+
+    return m
+
+
+def _touch_ink(root: Path) -> None:
+    ink = root / "node_modules" / "@hermes" / "ink" / "package.json"
+    ink.parent.mkdir(parents=True, exist_ok=True)
+    ink.write_text("{}")
+
+
+def test_need_install_when_ink_missing(tmp_path: Path, main_mod) -> None:
+    (tmp_path / "package-lock.json").write_text("{}")
+    assert main_mod._tui_need_npm_install(tmp_path) is True
+
+
+def test_need_install_when_lock_newer_than_marker(tmp_path: Path, main_mod) -> None:
+    _touch_ink(tmp_path)
+    (tmp_path / "package-lock.json").write_text("{}")
+    (tmp_path / "node_modules" / ".package-lock.json").write_text("{}")
+    os.utime(tmp_path / "package-lock.json", (200, 200))
+    os.utime(tmp_path / "node_modules" / ".package-lock.json", (100, 100))
+    assert main_mod._tui_need_npm_install(tmp_path) is True
+
+
+def test_no_install_when_lock_older_than_marker(tmp_path: Path, main_mod) -> None:
+    _touch_ink(tmp_path)
+    (tmp_path / "package-lock.json").write_text("{}")
+    (tmp_path / "node_modules" / ".package-lock.json").write_text("{}")
+    os.utime(tmp_path / "package-lock.json", (100, 100))
+    os.utime(tmp_path / "node_modules" / ".package-lock.json", (200, 200))
+    assert main_mod._tui_need_npm_install(tmp_path) is False
+
+
+def test_need_install_when_marker_missing(tmp_path: Path, main_mod) -> None:
+    _touch_ink(tmp_path)
+    (tmp_path / "package-lock.json").write_text("{}")
+    assert main_mod._tui_need_npm_install(tmp_path) is True
+
+
+def test_no_install_without_lockfile_when_ink_present(tmp_path: Path, main_mod) -> None:
+    _touch_ink(tmp_path)
+    assert main_mod._tui_need_npm_install(tmp_path) is False

tests/hermes_cli/test_tui_resume_flow.py

@@ -0,0 +1,121 @@
+from argparse import Namespace
+import sys
+import types
+
+import pytest
+
+
+def _args(**overrides):
+    base = {
+        "continue_last": None,
+        "resume": None,
+        "tui": True,
+    }
+    base.update(overrides)
+    return Namespace(**base)
+
+
+@pytest.fixture
+def main_mod(monkeypatch):
+    import hermes_cli.main as mod
+
+    monkeypatch.setattr(mod, "_has_any_provider_configured", lambda: True)
+    return mod
+
+
+def test_cmd_chat_tui_continue_uses_latest_tui_session(monkeypatch, main_mod):
+    calls = []
+    captured = {}
+
+    def fake_resolve_last(source="cli"):
+        calls.append(source)
+        return "20260408_235959_a1b2c3" if source == "tui" else None
+
+    def fake_launch(resume_session_id=None, tui_dev=False):
+        captured["resume"] = resume_session_id
+        raise SystemExit(0)
+
+    monkeypatch.setattr(main_mod, "_resolve_last_session", fake_resolve_last)
+    monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", lambda val: val)
+    monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
+
+    with pytest.raises(SystemExit):
+        main_mod.cmd_chat(_args(continue_last=True))
+
+    assert calls == ["tui"]
+    assert captured["resume"] == "20260408_235959_a1b2c3"
+
+
+def test_cmd_chat_tui_continue_falls_back_to_latest_cli_session(monkeypatch, main_mod):
+    calls = []
+    captured = {}
+
+    def fake_resolve_last(source="cli"):
+        calls.append(source)
+        if source == "tui":
+            return None
+        if source == "cli":
+            return "20260408_235959_d4e5f6"
+        return None
+
+    def fake_launch(resume_session_id=None, tui_dev=False):
+        captured["resume"] = resume_session_id
+        raise SystemExit(0)
+
+    monkeypatch.setattr(main_mod, "_resolve_last_session", fake_resolve_last)
+    monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", lambda val: val)
+    monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
+
+    with pytest.raises(SystemExit):
+        main_mod.cmd_chat(_args(continue_last=True))
+
+    assert calls == ["tui", "cli"]
+    assert captured["resume"] == "20260408_235959_d4e5f6"
+
+
+def test_cmd_chat_tui_resume_resolves_title_before_launch(monkeypatch, main_mod):
+    captured = {}
+
+    def fake_launch(resume_session_id=None, tui_dev=False):
+        captured["resume"] = resume_session_id
+        raise SystemExit(0)
+
+    monkeypatch.setattr(main_mod, "_resolve_session_by_name_or_id", lambda val: "20260409_000000_aa11bb")
+    monkeypatch.setattr(main_mod, "_launch_tui", fake_launch)
+
+    with pytest.raises(SystemExit):
+        main_mod.cmd_chat(_args(resume="my t0p session"))
+
+    assert captured["resume"] == "20260409_000000_aa11bb"
+
+
+def test_print_tui_exit_summary_includes_resume_and_token_totals(monkeypatch, capsys):
+    import hermes_cli.main as main_mod
+
+    class _FakeDB:
+        def get_session(self, session_id):
+            assert session_id == "20260409_000001_abc123"
+            return {
+                "message_count": 2,
+                "input_tokens": 10,
+                "output_tokens": 6,
+                "cache_read_tokens": 2,
+                "cache_write_tokens": 2,
+                "reasoning_tokens": 1,
+            }
+
+        def get_session_title(self, _session_id):
+            return "demo title"
+
+        def close(self):
+            return None
+
+    monkeypatch.setitem(sys.modules, "hermes_state", types.SimpleNamespace(SessionDB=lambda: _FakeDB()))
+
+    main_mod._print_tui_exit_summary("20260409_000001_abc123")
+    out = capsys.readouterr().out
+
+    assert "Resume this session with:" in out
+    assert "hermes --tui --resume 20260409_000001_abc123" in out
+    assert 'hermes --tui -c "demo title"' in out
+    assert "Tokens:         21 (in 10, out 6, cache 4, reasoning 1)" in out

tests/test_tui_gateway_server.py

@@ -0,0 +1,440 @@
+import json
+import sys
+import threading
+import time
+import types
+from pathlib import Path
+from unittest.mock import patch
+
+from tui_gateway import server
+
+
+class _ChunkyStdout:
+    def __init__(self):
+        self.parts: list[str] = []
+
+    def write(self, text: str) -> int:
+        for ch in text:
+            self.parts.append(ch)
+            time.sleep(0.0001)
+        return len(text)
+
+    def flush(self) -> None:
+        return None
+
+
+class _BrokenStdout:
+    def write(self, text: str) -> int:
+        raise BrokenPipeError
+
+    def flush(self) -> None:
+        return None
+
+
+def test_write_json_serializes_concurrent_writes(monkeypatch):
+    out = _ChunkyStdout()
+    monkeypatch.setattr(server, "_real_stdout", out)
+
+    threads = [
+        threading.Thread(target=server.write_json, args=({"seq": i, "text": "x" * 24},))
+        for i in range(8)
+    ]
+
+    for t in threads:
+        t.start()
+
+    for t in threads:
+        t.join()
+
+    lines = "".join(out.parts).splitlines()
+
+    assert len(lines) == 8
+    assert {json.loads(line)["seq"] for line in lines} == set(range(8))
+
+
+def test_write_json_returns_false_on_broken_pipe(monkeypatch):
+    monkeypatch.setattr(server, "_real_stdout", _BrokenStdout())
+
+    assert server.write_json({"ok": True}) is False
+
+
+def test_status_callback_emits_kind_and_text():
+    with patch("tui_gateway.server._emit") as emit:
+        cb = server._agent_cbs("sid")["status_callback"]
+        cb("context_pressure", "85% to compaction")
+
+    emit.assert_called_once_with(
+        "status.update",
+        "sid",
+        {"kind": "context_pressure", "text": "85% to compaction"},
+    )
+
+
+def test_status_callback_accepts_single_message_argument():
+    with patch("tui_gateway.server._emit") as emit:
+        cb = server._agent_cbs("sid")["status_callback"]
+        cb("thinking...")
+
+    emit.assert_called_once_with(
+        "status.update",
+        "sid",
+        {"kind": "status", "text": "thinking..."},
+    )
+
+
+def _session(agent=None, **extra):
+    return {
+        "agent": agent if agent is not None else types.SimpleNamespace(),
+        "session_key": "session-key",
+        "history": [],
+        "history_lock": threading.Lock(),
+        "history_version": 0,
+        "running": False,
+        "attached_images": [],
+        "image_counter": 0,
+        "cols": 80,
+        "slash_worker": None,
+        "show_reasoning": False,
+        "tool_progress_mode": "all",
+        **extra,
+    }
+
+
+def test_config_set_yolo_toggles_session_scope():
+    from tools.approval import clear_session, is_session_yolo_enabled
+
+    server._sessions["sid"] = _session()
+    try:
+        resp_on = server.handle_request({"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "yolo"}})
+        assert resp_on["result"]["value"] == "1"
+        assert is_session_yolo_enabled("session-key") is True
+
+        resp_off = server.handle_request({"id": "2", "method": "config.set", "params": {"session_id": "sid", "key": "yolo"}})
+        assert resp_off["result"]["value"] == "0"
+        assert is_session_yolo_enabled("session-key") is False
+    finally:
+        clear_session("session-key")
+        server._sessions.clear()
+
+
+def test_enable_gateway_prompts_sets_gateway_env(monkeypatch):
+    monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
+    monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
+    monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
+
+    server._enable_gateway_prompts()
+
+    assert server.os.environ["HERMES_GATEWAY_SESSION"] == "1"
+    assert server.os.environ["HERMES_EXEC_ASK"] == "1"
+    assert server.os.environ["HERMES_INTERACTIVE"] == "1"
+
+
+def test_setup_status_reports_provider_config(monkeypatch):
+    monkeypatch.setattr("hermes_cli.main._has_any_provider_configured", lambda: False)
+
+    resp = server.handle_request({"id": "1", "method": "setup.status", "params": {}})
+
+    assert resp["result"]["provider_configured"] is False
+
+
+def test_config_set_reasoning_updates_live_session_and_agent(tmp_path, monkeypatch):
+    monkeypatch.setattr(server, "_hermes_home", tmp_path)
+    agent = types.SimpleNamespace(reasoning_config=None)
+    server._sessions["sid"] = _session(agent=agent)
+
+    resp_effort = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "reasoning", "value": "low"}}
+    )
+    assert resp_effort["result"]["value"] == "low"
+    assert agent.reasoning_config == {"enabled": True, "effort": "low"}
+
+    resp_show = server.handle_request(
+        {"id": "2", "method": "config.set", "params": {"session_id": "sid", "key": "reasoning", "value": "show"}}
+    )
+    assert resp_show["result"]["value"] == "show"
+    assert server._sessions["sid"]["show_reasoning"] is True
+
+
+def test_config_set_verbose_updates_session_mode_and_agent(tmp_path, monkeypatch):
+    monkeypatch.setattr(server, "_hermes_home", tmp_path)
+    agent = types.SimpleNamespace(verbose_logging=False)
+    server._sessions["sid"] = _session(agent=agent)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "verbose", "value": "cycle"}}
+    )
+
+    assert resp["result"]["value"] == "verbose"
+    assert server._sessions["sid"]["tool_progress_mode"] == "verbose"
+    assert agent.verbose_logging is True
+
+
+def test_config_set_model_uses_live_switch_path(monkeypatch):
+    server._sessions["sid"] = _session()
+    seen = {}
+
+    def _fake_apply(sid, session, raw):
+        seen["args"] = (sid, session["session_key"], raw)
+        return {"value": "new/model", "warning": "catalog unreachable"}
+
+    monkeypatch.setattr(server, "_apply_model_switch", _fake_apply)
+    resp = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "model", "value": "new/model"}}
+    )
+
+    assert resp["result"]["value"] == "new/model"
+    assert resp["result"]["warning"] == "catalog unreachable"
+    assert seen["args"] == ("sid", "session-key", "new/model")
+
+
+def test_config_set_model_global_persists(monkeypatch):
+    class _Agent:
+        provider = "openrouter"
+        model = "old/model"
+        base_url = ""
+        api_key = "sk-old"
+
+        def switch_model(self, **kwargs):
+            return None
+
+    result = types.SimpleNamespace(
+        success=True,
+        new_model="anthropic/claude-sonnet-4.6",
+        target_provider="anthropic",
+        api_key="sk-new",
+        base_url="https://api.anthropic.com",
+        api_mode="anthropic_messages",
+        warning_message="",
+    )
+    seen = {}
+    saved = {}
+
+    def _switch_model(**kwargs):
+        seen.update(kwargs)
+        return result
+
+    server._sessions["sid"] = _session(agent=_Agent())
+    monkeypatch.setattr("hermes_cli.model_switch.switch_model", _switch_model)
+    monkeypatch.setattr(server, "_restart_slash_worker", lambda session: None)
+    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
+    monkeypatch.setattr("hermes_cli.config.save_config", lambda cfg: saved.update(cfg))
+
+    resp = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "model", "value": "anthropic/claude-sonnet-4.6 --global"}}
+    )
+
+    assert resp["result"]["value"] == "anthropic/claude-sonnet-4.6"
+    assert seen["is_global"] is True
+    assert saved["model"]["default"] == "anthropic/claude-sonnet-4.6"
+    assert saved["model"]["provider"] == "anthropic"
+    assert saved["model"]["base_url"] == "https://api.anthropic.com"
+
+
+def test_config_set_personality_rejects_unknown_name(monkeypatch):
+    monkeypatch.setattr(server, "_available_personalities", lambda cfg=None: {"helpful": "You are helpful."})
+    resp = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"key": "personality", "value": "bogus"}}
+    )
+
+    assert "error" in resp
+    assert "Unknown personality" in resp["error"]["message"]
+
+
+def test_config_set_personality_resets_history_and_returns_info(monkeypatch):
+    session = _session(agent=types.SimpleNamespace(), history=[{"role": "user", "text": "hi"}], history_version=4)
+    new_agent = types.SimpleNamespace(model="x")
+    emits = []
+
+    server._sessions["sid"] = session
+    monkeypatch.setattr(server, "_available_personalities", lambda cfg=None: {"helpful": "You are helpful."})
+    monkeypatch.setattr(server, "_make_agent", lambda sid, key, session_id=None: new_agent)
+    monkeypatch.setattr(server, "_session_info", lambda agent: {"model": getattr(agent, "model", "?")})
+    monkeypatch.setattr(server, "_restart_slash_worker", lambda session: None)
+    monkeypatch.setattr(server, "_emit", lambda *args: emits.append(args))
+    monkeypatch.setattr(server, "_write_config_key", lambda path, value: None)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "config.set", "params": {"session_id": "sid", "key": "personality", "value": "helpful"}}
+    )
+
+    assert resp["result"]["history_reset"] is True
+    assert resp["result"]["info"] == {"model": "x"}
+    assert session["history"] == []
+    assert session["history_version"] == 5
+    assert ("session.info", "sid", {"model": "x"}) in emits
+
+
+def test_session_compress_uses_compress_helper(monkeypatch):
+    agent = types.SimpleNamespace()
+    server._sessions["sid"] = _session(agent=agent)
+
+    monkeypatch.setattr(server, "_compress_session_history", lambda session, focus_topic=None: (2, {"total": 42}))
+    monkeypatch.setattr(server, "_session_info", lambda _agent: {"model": "x"})
+
+    with patch("tui_gateway.server._emit") as emit:
+        resp = server.handle_request({"id": "1", "method": "session.compress", "params": {"session_id": "sid"}})
+
+    assert resp["result"]["removed"] == 2
+    assert resp["result"]["usage"]["total"] == 42
+    emit.assert_called_once_with("session.info", "sid", {"model": "x"})
+
+
+def test_prompt_submit_sets_approval_session_key(monkeypatch):
+    from tools.approval import get_current_session_key
+
+    captured = {}
+
+    class _Agent:
+        def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
+            captured["session_key"] = get_current_session_key(default="")
+            return {"final_response": "ok", "messages": [{"role": "assistant", "content": "ok"}]}
+
+    class _ImmediateThread:
+        def __init__(self, target=None, daemon=None):
+            self._target = target
+
+        def start(self):
+            self._target()
+
+    server._sessions["sid"] = _session(agent=_Agent())
+    monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
+    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
+    monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
+    monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
+
+    resp = server.handle_request({"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "ping"}})
+
+    assert resp["result"]["status"] == "streaming"
+    assert captured["session_key"] == "session-key"
+
+
+def test_prompt_submit_expands_context_refs(monkeypatch):
+    captured = {}
+
+    class _Agent:
+        model = "test/model"
+        base_url = ""
+        api_key = ""
+
+        def run_conversation(self, prompt, conversation_history=None, stream_callback=None):
+            captured["prompt"] = prompt
+            return {"final_response": "ok", "messages": [{"role": "assistant", "content": "ok"}]}
+
+    class _ImmediateThread:
+        def __init__(self, target=None, daemon=None):
+            self._target = target
+
+        def start(self):
+            self._target()
+
+    fake_ctx = types.ModuleType("agent.context_references")
+    fake_ctx.preprocess_context_references = lambda message, **kwargs: types.SimpleNamespace(
+        blocked=False, message="expanded prompt", warnings=[], references=[], injected_tokens=0
+    )
+    fake_meta = types.ModuleType("agent.model_metadata")
+    fake_meta.get_model_context_length = lambda *args, **kwargs: 100000
+
+    server._sessions["sid"] = _session(agent=_Agent())
+    monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
+    monkeypatch.setattr(server, "_emit", lambda *args, **kwargs: None)
+    monkeypatch.setattr(server, "make_stream_renderer", lambda cols: None)
+    monkeypatch.setattr(server, "render_message", lambda raw, cols: None)
+    monkeypatch.setitem(sys.modules, "agent.context_references", fake_ctx)
+    monkeypatch.setitem(sys.modules, "agent.model_metadata", fake_meta)
+
+    server.handle_request({"id": "1", "method": "prompt.submit", "params": {"session_id": "sid", "text": "@diff"}})
+
+    assert captured["prompt"] == "expanded prompt"
+
+
+def test_image_attach_appends_local_image(monkeypatch):
+    fake_cli = types.ModuleType("cli")
+    fake_cli._IMAGE_EXTENSIONS = {".png"}
+    fake_cli._split_path_input = lambda raw: (raw, "")
+    fake_cli._resolve_attachment_path = lambda raw: Path("/tmp/cat.png")
+
+    server._sessions["sid"] = _session()
+    monkeypatch.setitem(sys.modules, "cli", fake_cli)
+
+    resp = server.handle_request({"id": "1", "method": "image.attach", "params": {"session_id": "sid", "path": "/tmp/cat.png"}})
+
+    assert resp["result"]["attached"] is True
+    assert resp["result"]["name"] == "cat.png"
+    assert len(server._sessions["sid"]["attached_images"]) == 1
+
+
+def test_command_dispatch_exec_nonzero_surfaces_error(monkeypatch):
+    monkeypatch.setattr(server, "_load_cfg", lambda: {"quick_commands": {"boom": {"type": "exec", "command": "boom"}}})
+    monkeypatch.setattr(
+        server.subprocess,
+        "run",
+        lambda *args, **kwargs: types.SimpleNamespace(returncode=1, stdout="", stderr="failed"),
+    )
+
+    resp = server.handle_request({"id": "1", "method": "command.dispatch", "params": {"name": "boom"}})
+
+    assert "error" in resp
+    assert "failed" in resp["error"]["message"]
+
+
+def test_plugins_list_surfaces_loader_error(monkeypatch):
+    with patch("hermes_cli.plugins.get_plugin_manager", side_effect=Exception("boom")):
+        resp = server.handle_request({"id": "1", "method": "plugins.list", "params": {}})
+
+    assert "error" in resp
+    assert "boom" in resp["error"]["message"]
+
+
+def test_complete_slash_surfaces_completer_error(monkeypatch):
+    with patch("hermes_cli.commands.SlashCommandCompleter", side_effect=Exception("no completer")):
+        resp = server.handle_request({"id": "1", "method": "complete.slash", "params": {"text": "/mo"}})
+
+    assert "error" in resp
+    assert "no completer" in resp["error"]["message"]
+
+
+def test_input_detect_drop_attaches_image(monkeypatch):
+    fake_cli = types.ModuleType("cli")
+    fake_cli._detect_file_drop = lambda raw: {
+        "path": Path("/tmp/cat.png"),
+        "is_image": True,
+        "remainder": "",
+    }
+
+    server._sessions["sid"] = _session()
+    monkeypatch.setitem(sys.modules, "cli", fake_cli)
+
+    resp = server.handle_request(
+        {"id": "1", "method": "input.detect_drop", "params": {"session_id": "sid", "text": "/tmp/cat.png"}}
+    )
+
+    assert resp["result"]["matched"] is True
+    assert resp["result"]["is_image"] is True
+    assert resp["result"]["text"] == "[User attached image: cat.png]"
+
+
+def test_rollback_restore_resolves_number_and_file_path():
+    calls = {}
+
+    class _Mgr:
+        enabled = True
+
+        def list_checkpoints(self, cwd):
+            return [{"hash": "aaa111"}, {"hash": "bbb222"}]
+
+        def restore(self, cwd, target, file_path=None):
+            calls["args"] = (cwd, target, file_path)
+            return {"success": True, "message": "done"}
+
+    server._sessions["sid"] = _session(agent=types.SimpleNamespace(_checkpoint_mgr=_Mgr()), history=[])
+    resp = server.handle_request(
+        {
+            "id": "1",
+            "method": "rollback.restore",
+            "params": {"session_id": "sid", "hash": "2", "file_path": "src/app.tsx"},
+        }
+    )
+
+    assert resp["result"]["success"] is True
+    assert calls["args"][1] == "bbb222"
+    assert calls["args"][2] == "src/app.tsx"

tests/tools/test_clipboard.py

@@ -250,6 +250,15 @@ class TestWslHasImage:
             mock_run.return_value = MagicMock(stdout="False\n", returncode=0)
             assert _wsl_has_image() is False
 
+    def test_falls_back_to_get_clipboard_image(self):
+        with patch("hermes_cli.clipboard.subprocess.run") as mock_run:
+            mock_run.side_effect = [
+                MagicMock(stdout="False\n", returncode=0),
+                MagicMock(stdout="True\n", returncode=0),
+            ]
+            assert _wsl_has_image() is True
+            assert mock_run.call_count == 2
+
     def test_powershell_not_found(self):
         with patch("hermes_cli.clipboard.subprocess.run", side_effect=FileNotFoundError):
             assert _wsl_has_image() is False
@@ -269,6 +278,18 @@ class TestWslSave:
             assert _wsl_save(dest) is True
         assert dest.read_bytes() == FAKE_PNG
 
+    def test_falls_back_to_get_clipboard_extraction(self, tmp_path):
+        dest = tmp_path / "out.png"
+        b64_png = base64.b64encode(FAKE_PNG).decode()
+        with patch("hermes_cli.clipboard.subprocess.run") as mock_run:
+            mock_run.side_effect = [
+                MagicMock(stdout="", returncode=1),
+                MagicMock(stdout=b64_png + "\n", returncode=0),
+            ]
+            assert _wsl_save(dest) is True
+            assert mock_run.call_count == 2
+        assert dest.read_bytes() == FAKE_PNG
+
     def test_no_image_returns_false(self, tmp_path):
         dest = tmp_path / "out.png"
         with patch("hermes_cli.clipboard.subprocess.run") as mock_run:
@@ -528,6 +549,16 @@ class TestWindowsHasImage:
                 mock_run.return_value = MagicMock(stdout="False\n", returncode=0)
                 assert _windows_has_image() is False
 
+    def test_falls_back_to_get_clipboard_image(self):
+        with patch("hermes_cli.clipboard._get_ps_exe", return_value="powershell"):
+            with patch("hermes_cli.clipboard.subprocess.run") as mock_run:
+                mock_run.side_effect = [
+                    MagicMock(stdout="False\n", returncode=0),
+                    MagicMock(stdout="True\n", returncode=0),
+                ]
+                assert _windows_has_image() is True
+                assert mock_run.call_count == 2
+
     def test_no_powershell_available(self):
         with patch("hermes_cli.clipboard._get_ps_exe", return_value=None):
             assert _windows_has_image() is False
@@ -559,6 +590,20 @@ class TestWindowsSave:
                 assert _windows_save(dest) is True
         assert dest.read_bytes() == FAKE_PNG
 
+    def test_falls_back_to_filedrop_image(self, tmp_path):
+        dest = tmp_path / "out.png"
+        b64_png = base64.b64encode(FAKE_PNG).decode()
+        with patch("hermes_cli.clipboard._get_ps_exe", return_value="powershell"):
+            with patch("hermes_cli.clipboard.subprocess.run") as mock_run:
+                mock_run.side_effect = [
+                    MagicMock(stdout="", returncode=1),
+                    MagicMock(stdout="", returncode=1),
+                    MagicMock(stdout=b64_png + "\n", returncode=0),
+                ]
+                assert _windows_save(dest) is True
+                assert mock_run.call_count == 3
+        assert dest.read_bytes() == FAKE_PNG
+
     def test_no_image_returns_false(self, tmp_path):
         dest = tmp_path / "out.png"
         with patch("hermes_cli.clipboard._get_ps_exe", return_value="powershell"):
@@ -734,6 +779,18 @@ class TestHasClipboardImage:
                     assert has_clipboard_image() is True
                     m.assert_called_once()
 
+    def test_wsl_falls_through_to_wayland_when_windows_path_empty(self):
+        """WSLg often bridges images to wl-paste even when powershell.exe check fails."""
+        with patch("hermes_cli.clipboard.sys") as mock_sys:
+            mock_sys.platform = "linux"
+            with patch("hermes_cli.clipboard._is_wsl", return_value=True):
+                with patch("hermes_cli.clipboard._wsl_has_image", return_value=False) as wsl:
+                    with patch.dict(os.environ, {"WAYLAND_DISPLAY": "wayland-0"}):
+                        with patch("hermes_cli.clipboard._wayland_has_image", return_value=True) as wl:
+                            assert has_clipboard_image() is True
+                            wsl.assert_called_once()
+                            wl.assert_called_once()
+
     def test_linux_wayland_dispatch(self):
         with patch("hermes_cli.clipboard.sys") as mock_sys:
             mock_sys.platform = "linux"

tests/tui_gateway/test_protocol.py

@@ -0,0 +1,233 @@
+"""Tests for tui_gateway JSON-RPC protocol plumbing."""
+
+import io
+import json
+import sys
+import threading
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+_original_stdout = sys.stdout
+
+
+@pytest.fixture(autouse=True)
+def _restore_stdout():
+    yield
+    sys.stdout = _original_stdout
+
+
+@pytest.fixture()
+def server():
+    with patch.dict("sys.modules", {
+        "hermes_constants": MagicMock(get_hermes_home=MagicMock(return_value="/tmp/hermes_test")),
+        "hermes_cli.env_loader": MagicMock(),
+        "hermes_cli.banner": MagicMock(),
+        "hermes_state": MagicMock(),
+    }):
+        import importlib
+        mod = importlib.import_module("tui_gateway.server")
+        yield mod
+        mod._sessions.clear()
+        mod._pending.clear()
+        mod._answers.clear()
+        mod._methods.clear()
+        importlib.reload(mod)
+
+
+@pytest.fixture()
+def capture(server):
+    """Redirect server's real stdout to a StringIO and return (server, buf)."""
+    buf = io.StringIO()
+    server._real_stdout = buf
+    return server, buf
+
+
+# ── JSON-RPC envelope ────────────────────────────────────────────────
+
+
+def test_unknown_method(server):
+    resp = server.handle_request({"id": "1", "method": "bogus"})
+    assert resp["error"]["code"] == -32601
+
+
+def test_ok_envelope(server):
+    assert server._ok("r1", {"x": 1}) == {
+        "jsonrpc": "2.0", "id": "r1", "result": {"x": 1},
+    }
+
+
+def test_err_envelope(server):
+    assert server._err("r2", 4001, "nope") == {
+        "jsonrpc": "2.0", "id": "r2", "error": {"code": 4001, "message": "nope"},
+    }
+
+
+# ── write_json ───────────────────────────────────────────────────────
+
+
+def test_write_json(capture):
+    server, buf = capture
+    assert server.write_json({"test": True})
+    assert json.loads(buf.getvalue()) == {"test": True}
+
+
+def test_write_json_broken_pipe(server):
+    class _Broken:
+        def write(self, _): raise BrokenPipeError
+        def flush(self): raise BrokenPipeError
+
+    server._real_stdout = _Broken()
+    assert server.write_json({"x": 1}) is False
+
+
+# ── _emit ────────────────────────────────────────────────────────────
+
+
+def test_emit_with_payload(capture):
+    server, buf = capture
+    server._emit("test.event", "s1", {"key": "val"})
+    msg = json.loads(buf.getvalue())
+
+    assert msg["method"] == "event"
+    assert msg["params"]["type"] == "test.event"
+    assert msg["params"]["session_id"] == "s1"
+    assert msg["params"]["payload"]["key"] == "val"
+
+
+def test_emit_without_payload(capture):
+    server, buf = capture
+    server._emit("ping", "s2")
+
+    assert "payload" not in json.loads(buf.getvalue())["params"]
+
+
+# ── Blocking prompt round-trip ───────────────────────────────────────
+
+
+def test_block_and_respond(capture):
+    server, _ = capture
+    result = [None]
+
+    threading.Thread(
+        target=lambda: result.__setitem__(0, server._block("test.prompt", "s1", {"q": "?"}, timeout=5)),
+    ).start()
+
+    for _ in range(100):
+        if server._pending:
+            break
+        threading.Event().wait(0.01)
+
+    rid = next(iter(server._pending))
+    server._answers[rid] = "my_answer"
+    server._pending[rid].set()
+
+    threading.Event().wait(0.1)
+    assert result[0] == "my_answer"
+
+
+def test_clear_pending(server):
+    ev = threading.Event()
+    server._pending["r1"] = ev
+    server._clear_pending()
+
+    assert ev.is_set()
+    assert server._answers["r1"] == ""
+
+
+# ── Session lookup ───────────────────────────────────────────────────
+
+
+def test_sess_missing(server):
+    _, err = server._sess({"session_id": "nope"}, "r1")
+    assert err["error"]["code"] == 4001
+
+
+def test_sess_found(server):
+    server._sessions["abc"] = {"agent": MagicMock()}
+    s, err = server._sess({"session_id": "abc"}, "r1")
+
+    assert s is not None
+    assert err is None
+
+
+# ── session.resume payload ────────────────────────────────────────────
+
+
+def test_session_resume_returns_hydrated_messages(server, monkeypatch):
+    class _DB:
+        def get_session(self, _sid):
+            return {"id": "20260409_010101_abc123"}
+
+        def get_session_by_title(self, _title):
+            return None
+
+        def reopen_session(self, _sid):
+            return None
+
+        def get_messages_as_conversation(self, _sid):
+            return [
+                {"role": "user", "content": "hello"},
+                {"role": "assistant", "content": "yo"},
+                {"role": "tool", "content": "searched"},
+                {"role": "assistant", "content": "   "},
+                {"role": "assistant", "content": None},
+                {"role": "narrator", "content": "skip"},
+            ]
+
+    monkeypatch.setattr(server, "_get_db", lambda: _DB())
+    monkeypatch.setattr(server, "_make_agent", lambda sid, key, session_id=None: object())
+    monkeypatch.setattr(server, "_init_session", lambda sid, key, agent, history, cols=80: None)
+    monkeypatch.setattr(server, "_session_info", lambda _agent: {"model": "test/model"})
+
+    resp = server.handle_request(
+        {
+            "id": "r1",
+            "method": "session.resume",
+            "params": {"session_id": "20260409_010101_abc123", "cols": 100},
+        }
+    )
+
+    assert "error" not in resp
+    assert resp["result"]["message_count"] == 3
+    assert resp["result"]["messages"] == [
+        {"role": "user", "text": "hello"},
+        {"role": "assistant", "text": "yo"},
+        {"role": "tool", "name": "tool", "context": ""},
+    ]
+
+
+# ── Config I/O ───────────────────────────────────────────────────────
+
+
+def test_config_load_missing(server, tmp_path):
+    server._hermes_home = tmp_path
+    assert server._load_cfg() == {}
+
+
+def test_config_roundtrip(server, tmp_path):
+    server._hermes_home = tmp_path
+    server._save_cfg({"model": "test/model"})
+    assert server._load_cfg()["model"] == "test/model"
+
+
+# ── _cli_exec_blocked ────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("argv", [
+    [],
+    ["setup"],
+    ["gateway"],
+    ["sessions", "browse"],
+    ["config", "edit"],
+])
+def test_cli_exec_blocked(server, argv):
+    assert server._cli_exec_blocked(argv) is not None
+
+
+@pytest.mark.parametrize("argv", [
+    ["version"],
+    ["sessions", "list"],
+])
+def test_cli_exec_allowed(server, argv):
+    assert server._cli_exec_blocked(argv) is None

tests/tui_gateway/test_render.py

@@ -0,0 +1,67 @@
+"""Tests for tui_gateway.render — rendering bridge fallback behavior."""
+
+from unittest.mock import MagicMock, patch
+
+from tui_gateway.render import make_stream_renderer, render_diff, render_message
+
+
+def _stub_rich(mock_mod):
+    return patch.dict("sys.modules", {"agent.rich_output": mock_mod})
+
+
+def _no_rich():
+    return patch.dict("sys.modules", {"agent.rich_output": None})
+
+
+# ── render_message ───────────────────────────────────────────────────
+
+
+def test_render_message_none_without_module():
+    with _no_rich():
+        assert render_message("hello") is None
+
+
+def test_render_message_formatted():
+    mod = MagicMock()
+    mod.format_response.return_value = "<b>hi</b>"
+
+    with _stub_rich(mod):
+        assert render_message("hi", 100) == "<b>hi</b>"
+
+
+def test_render_message_type_error_fallback():
+    mod = MagicMock()
+    mod.format_response.side_effect = [TypeError, "fallback"]
+
+    with _stub_rich(mod):
+        assert render_message("hi") == "fallback"
+
+
+def test_render_message_exception_returns_none():
+    mod = MagicMock()
+    mod.format_response.side_effect = RuntimeError
+
+    with _stub_rich(mod):
+        assert render_message("hi") is None
+
+
+# ── render_diff / make_stream_renderer ───────────────────────────────
+
+
+def test_render_diff_none_without_module():
+    with _no_rich():
+        assert render_diff("+line") is None
+
+
+def test_stream_renderer_none_without_module():
+    with _no_rich():
+        assert make_stream_renderer() is None
+
+
+def test_stream_renderer_returns_instance():
+    renderer = MagicMock()
+    mod = MagicMock()
+    mod.StreamingRenderer.return_value = renderer
+
+    with _stub_rich(mod):
+        assert make_stream_renderer(120) is renderer