TUI & JSON-RPC

Ink Terminal UI 与双进程 RPC 桥接服务

全面透视 ui-tui 终端画布绘制与 tui_gateway 二进制双通道异步桥接逻辑
TUI 与 RPC 桥接
📊 图 6-1：React Ink TUI 与 Python 智能体核心的双进程通信架构图

📺 1. Ink (React) 终端画布渲染原理

项目下的 ui-tui/ 提供了极其华丽的 ANSI 终端仪表盘（Dashboard）。它彻底颠覆了传统的字符行流式打印，在控制台中实现了动态布局：

  • React 终端化（React-to-Terminal）：React Ink 在 Node.js 进程中执行，将 React 组件的状态与生命周期映射为 ANSI 逃逸控制字符，实现终端界面的实时局部刷新，避免传统 CLI 输出时的严重闪烁。
  • 微型 NanoStores 状态总线：TUI 内部不推荐使用旁系 React Context 传递状态，而是使用轻量化的 NanoStores 管理当前活跃会话、输入提示符状态和折叠的任务轨迹树，实现零渲染延迟。

🔌 2. 双进程 WebSocket JSON-RPC 桥接与数据帧

为保证推理和终端界面更新不互相卡死死锁，Node TUI 前端与 Python 核心独立为两个不同的进程运转，利用本地 WebSocket 进行高频 JSON-RPC 协议桥接。这套通信机理被集中在项目 tui_gateway/ 中，其具体交互数据流如下图时序图所示：

JSON-RPC 数据帧时序图
📊 图 6-2：双进程 WebSocket JSON-RPC 上下行数据帧时序图
  • tui_gateway/ws.py：启动 WebSocket 服务。
  • tui_gateway/transport.py：下行事件广播中心。后端智能体输出的 token fragment、思考步骤，会被其格式化为 JSON 帧推送到 WebSocket 连接。
  • tui_gateway/server.py：上行指令响应中心。定义了对下述核心 JSON-RPC 方法的拦截处理：

📄 核心 JSON-RPC 上行方法映射表

RPC 方法名 (Method) 调用参数 (Params) Python 后端接收后的核心处理器
send_message {"message": "user question", "session_id": "api-..."} server.py 中的对应方法接收后，将指令抛给后台线程去启动 AIAgent.run_conversation()。
interrupt_conversation (空) 直接将当前活跃 AIAgent 实例的 _interrupt_requested 属性置为 True，令大模型流式循环强行提前回滚中断。
get_session_info {"session_id": "api-..."} 直接调用 SessionDB 执行 SQL 查询并返回格式化的 Turn 列表。
process_slash_command {"command": "/compress"} 将快捷斜杠动作提取并异步丢给 SlashWorker 执行队列。

⚙️ 3. 为什么需要 SlashWorker 异步队列？ (tui_gateway/slash_worker.py)

在 TUI 界面中，当用户发出诸如 /compress 强制进行对话摘要、或者 /branch 进行分支等斜杠快捷动作时，由于这些任务涉及高开销的数据库锁定、辅助大模型长文本推理，如果直接在 RPC 主服务线程中同步执行，会发生严重阻塞：WebSocket 卡死与 SQLite 并发锁冲突。为了保障这一并发场景的高可用，在后台维护了一个单线程管道队列：

# SlashWorker 排队执行实现
class SlashWorker(threading.Thread):
    def __init__(self):
        super().__init__()
        self.queue = queue.Queue()
        self.daemon = True

    def run(self):
        while True:
            # 串行排队拉取，完全避免了高 IO 命令间的锁冲突
            task = self.queue.get()
            try:
                # 调用核心 cli 逻辑处理快捷指令并向 TUI 广播反馈结果
                result = self.execute_cmd(task.command)
                self.publish_result_to_tui(result)
            except Exception as e:
                self.publish_error(e)
            finally:
                self.queue.task_done()

🔗 本章子任务深入剖析专栏

为深度掌握 TUI 控制台，推荐阅读以下子技术专题：