> ## Documentation Index
> Fetch the complete documentation index at: https://docs.zeus.agentspro.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# 消息队列

> Zeus 消息队列 — 用户消息排队、对话回退、事件持久化与并发控制

Zeus 的消息队列包含两个层面：**用户侧消息队列**（允许用户在 Agent 处理时排队发送消息）和**后端事件持久化**系统（确保消息可靠投递）。

***

## 用户消息队列

当 Agent 正在处理回复时，用户可以继续输入并排队发送消息，无需等待。

### 行为规则

| 场景                  | 行为                    |
| ------------------- | --------------------- |
| Agent 处理中 + 用户发送    | 消息加入队列（不立即发送）         |
| Agent 正常完成          | 100 ms 后自动发送队列中的下一条   |
| Agent 被中断（Stop）     | 队列**不自动消费**，由用户决定发送哪条 |
| 用户点击队列中的**立即发送**（↑） | 停止当前生成（如有），然后发送该消息    |

### 队列 UI

当有排队消息时，队列面板显示在聊天输入框上方：

* **编辑**（铅笔图标）— 行内编辑排队消息内容
* **立即发送**（↑ 箭头图标）— 停止当前生成并立即发送此消息
* **移除**（垃圾桶图标）— 从队列中删除

### 截断上下文

当回复被中途停止时，AI 的部分内容会**保留**在对话历史中，作为后续消息的上下文。同时**持久化到数据库**，刷新页面后不丢失。

***

## 对话与文件回退

用户可以回退到任意一轮对话，类似 Cursor 的分支模型。

### 对话回退

1. 鼠标悬停在用户消息上 → 点击**回退**按钮（↺）
2. 回退点之后的消息**变灰**（40% 透明度），不会删除
3. 出现分叉分隔线，带有**取消回退**选项
4. 新消息添加在分隔线下方
5. 发送给后端的聊天历史只包含**回退点之前**的消息

### 文件回退

当用户回退时，回退点之后被修改的沙盒文件会恢复到之前的状态：

```
handleRollback(forkMessageId)
  → 保存当前文件为 preRollbackFiles
  → 收集 forkMessageId 之后的 FileSnapshot
  → POST /api/sandbox/restore-files（写回旧版本）
  → 更新 trajectoryStore.sandboxFiles
```

取消回退会逆向操作，从 `preRollbackFiles` 恢复最新文件版本。

***

## 刷新后文件恢复

沙盒文件从**数据库事件**中恢复，而非仅依赖 E2B 沙盒存活状态，因此即使沙盒过期文件也不会丢失。

```mermaid theme={null}
flowchart TD
    refresh["页面刷新"]
    events["await loadEventsFromDb"]
    rebuild["rebuildStateFromEvents 从<br/>tool_call_result 事件提取 sandboxFiles"]
    sandbox{"沙盒仍存活?"}
    merge["合并沙盒新文件<br/>（不覆盖事件中的文件）"]
    done["文件树就绪"]
    skip["仅使用事件中的文件"]

    refresh --> events --> rebuild --> sandbox
    sandbox -->|"是"| merge --> done
    sandbox -->|"否"| skip --> done
```

文件内容**按需加载**：当用户点击内容为空的文件标签时，前端调用 `GET /api/sandbox/read-file` 从沙盒获取内容。

***

## Event Persistence

### RealtimeEventSaver

`RealtimeEventSaver` 负责将实时 SSE 事件批量持久化到数据库：

```mermaid theme={null}
flowchart TD
    events["SSE Events 到达"]
    buffer["事件缓冲区"]
    batch{"达到批量阈值?"}
    write["批量写入 PostgreSQL"]
    retry{"写入成功?"}
    backoff["指数退避重试"]
    fallback["降级: LocalStorage"]
    recover["下次加载时回写"]

    events --> buffer --> batch
    batch -->|"3 个事件 or 100ms"| write
    batch -->|"未达到"| buffer
    write --> retry
    retry -->|"成功"| buffer
    retry -->|"失败"| backoff
    backoff -->|"≤3 次"| write
    backoff -->|">3 次"| fallback
    fallback --> recover
```

### 配置

| 配置项  | 值            | 说明            |
| ---- | ------------ | ------------- |
| 批量大小 | 3 个事件        | 缓冲区达到此数量时触发写入 |
| 批量间隔 | 100ms        | 即使未满也定时刷新     |
| 重试次数 | 3 次          | 最大重试次数        |
| 重试策略 | 指数退避         | 1s → 2s → 4s  |
| 降级方案 | LocalStorage | 所有重试失败后的备份存储  |

***

## Message Batching

### 前端消息合并

前端合并快速连续的文本更新，避免过度渲染：

| 策略      | 说明                                       |
| ------- | ---------------------------------------- |
| UI 批量刷新 | 约 60fps 频率批量更新，合并快速连续的 token             |
| 虚拟滚动    | 只渲染可视区域的消息卡片                             |
| 选择性持久化  | 只持久化 `sessionId` 和 `messageIds`，消息内容按需加载 |

### Chat History 构建

前端从 Zustand Store 加载当前会话的消息历史，排除当前正在发送的消息，传递给后端。后端将其转换为 LangChain 消息类型，限制最多 **30 条**。

***

## Concurrency & Isolation

```mermaid theme={null}
graph TD
    subgraph Session1["Session A"]
        s1_agent["Agent"]
        s1_checkpoint["Checkpoint"]
        s1_cache["Context Cache"]
    end

    subgraph Session2["Session B"]
        s2_agent["Agent"]
        s2_checkpoint["Checkpoint"]
        s2_cache["Context Cache"]
    end

    subgraph UserSpace["User Workspace"]
        workspace["CloudDriveBackend<br/>按 user_id 隔离"]
    end

    Session1 --> UserSpace
    Session2 --> UserSpace
```

| 隔离维度    | 机制           | 说明                                            |
| ------- | ------------ | --------------------------------------------- |
| Session | `thread_id`  | 每个 Session 拥有独立的 Checkpointer 状态              |
| Context | `session_id` | Context Cache 按 session\_id 隔离，resume 只恢复对应会话 |
| 工具执行    | LangGraph    | 同一 Session 内工具串行执行，不会并发                       |
| 工作空间    | `user_id`    | 用户文件系统按 user\_id 完全隔离                         |

***

## Session Recovery

系统支持会话中断后的恢复：

### 保存时机

* 工具调用完成后
* 消息发送后
* HITL 中断时（通过 Checkpointer）

### 恢复流程

```mermaid theme={null}
flowchart TD
    load["页面加载"]
    check{"有保存的会话状态?"}
    restore["恢复消息列表 + 待处理工具调用"]
    local{"有 LocalStorage 备份?"}
    sync["回写事件到服务器"]
    ready["会话就绪"]
    fresh["新会话"]

    load --> check
    check -->|"有"| restore --> local
    check -->|"无"| fresh
    local -->|"有"| sync --> ready
    local -->|"无"| ready
```

***

## Related Docs

<CardGroup cols={2}>
  <Card title="Messages" icon="comments" href="/en/ai-backend/messages/Messages">
    完整的消息流程、请求参数和状态管理
  </Card>

  <Card title="Retry Policy" icon="rotate" href="/en/ai-backend/messages/Retry-Policy">
    错误处理、重试策略和降级方案
  </Card>
</CardGroup>