Compare commits

...

6 Commits

Author SHA1 Message Date
0c356e7ac4 重构: 添加斜杠命令解析和执行功能 2026-04-26 21:51:24 +08:00
38425e23f6 重构: 优化终端初始化和清理流程 2026-04-26 21:40:34 +08:00
bcee62713f 重构: 移除调试日志以简化消息处理流程 2026-04-26 21:26:27 +08:00
cf6d57c568 重构: 添加 TUI 组件以支持聊天界面和输入处理 2026-04-26 21:00:17 +08:00
86dea0f874 Refactor session management to support dialog-based architecture
- Removed InputHandler and related input event handling code.
- Updated GatewayState to handle new session commands for dialogs.
- Introduced UnifiedSessionId for managing session identifiers across channels and chats.
- Refactored Session and SessionManager to manage dialogs instead of sessions.
- Added methods for creating, listing, switching, renaming, archiving, and deleting dialogs.
- Updated storage functions to accommodate dialog IDs in persistent session management.
- Enhanced tests to cover new dialog functionalities and ensure stability.
2026-04-26 20:59:54 +08:00
5ce5502c56 refactor: 抽取独立 session 模块
将 Session/SessionManager 从 gateway 抽取到独立的 session 模块:

- 新建 src/session/ 目录
  - mod.rs: 模块导出
  - error.rs: SessionError 类型
  - commands.rs: SessionCommand 枚举
  - events.rs: SessionEvent 枚举
  - session.rs: Session 和 SessionManager 实现

- 更新 src/gateway/mod.rs
  - 移除 pub mod session
  - 改用 crate::session::SessionManager

- 更新 src/lib.rs
  - 添加 pub mod session

- 删除 src/gateway/session.rs (内容已移动)

注意: ControlInbound/ControlOutbound/ControlMessage 保留在 bus/message.rs,
cli_chat 仍通过消息总线与 SessionManager 通信,待后续简化。
2026-04-26 17:48:23 +08:00
35 changed files with 1915 additions and 1735 deletions

View File

@ -28,3 +28,7 @@ base64 = "0.22"
tempfile = "3"
meval = "0.2"
rusqlite = { version = "0.32", features = ["bundled"] }
ratatui = "0.27"
crossterm = { version = "0.28", features = ["event-stream"] }
termimad = "0.34"
textwrap = "0.16"

View File

@ -1,346 +0,0 @@
# Picobot 工具机制增强实现日志
## 实现记录
### 1. SchemaCleanr - 跨 Provider Schema 归一化
**日期**: 2026-04-07
**Commit**: `d5b6cd2`
#### 背景
不同 LLM provider 对 JSON Schema 支持差异很大:
- **Gemini**: 最严格,不支持 `minLength`, `maxLength`, `pattern`, `minimum`, `maximum`
- **Anthropic**: 中等,只要求解决 `$ref`
- **OpenAI**: 最宽松,支持大部分关键词
#### 实现方案
创建 `src/tools/schema.rs`,提供:
1. **`CleaningStrategy` enum**
```rust
pub enum CleaningStrategy {
Gemini, // 最严格
Anthropic, // 中等
OpenAI, // 最宽松
Conservative,
}
```
2. **`SchemaCleanr::clean()`** - 核心清洗函数
- 移除 provider 不支持的关键词
- 解析 `$ref``$defs`/`definitions`
- 将 `anyOf`/`oneOf` 合并为 `enum`
- 将 `const` 转换为 `enum`
- 移除 `type` 数组中的 `null`
3. **`SchemaCleanr::validate()`** - Schema 验证
#### 使用方法
```rust
use picobot::tools::{SchemaCleanr, CleaningStrategy};
// Gemini 兼容清洗(最严格)
let cleaned = SchemaCleanr::clean_for_gemini(schema);
// Anthropic 兼容清洗
let cleaned = SchemaCleanr::clean_for_anthropic(schema);
// OpenAI 兼容清洗(最宽松)
let cleaned = SchemaCleanr::clean_for_openai(schema);
// 自定义策略
let cleaned = SchemaCleanr::clean(schema, CleaningStrategy::Conservative);
```
#### 工具 Trait 增强
`src/tools/traits.rs``Tool` trait 中新增:
```rust
pub trait Tool: Send + Sync + 'static {
// ... 原有方法 ...
/// 是否只读(无副作用)
fn read_only(&self) -> bool { false }
/// 是否可以与其他工具并行执行
fn concurrency_safe(&self) -> bool {
self.read_only() && !self.exclusive()
}
/// 是否需要独占执行
fn exclusive(&self) -> bool { false }
}
```
这些属性为后续的并行工具执行提供基础。
#### 测试
- 12 个单元测试覆盖所有清洗逻辑
- 运行 `cargo test --lib tools::schema` 验证
### 2. file_read 工具
**日期**: 2026-04-07
**Commit**: `a9e7aab`
#### 功能
- 读取文件内容(支持 offset/limit 分页)
- 返回带行号的内容,便于引用
- 自动处理二进制文件base64 编码)
- 可选的目录限制(安全隔离)
#### Schema
```json
{
"type": "object",
"properties": {
"path": { "type": "string", "description": "文件路径" },
"offset": { "type": "integer", "description": "起始行号(1-indexed)" },
"limit": { "type": "integer", "description": "最大行数" }
},
"required": ["path"]
}
```
#### 使用方法
```rust
use picobot::tools::FileReadTool;
// 基本用法
let tool = FileReadTool::new();
let result = tool.execute(json!({
"path": "/some/file.txt",
"offset": 1,
"limit": 100
})).await;
```
#### 测试
- 4 个单元测试
- `cargo test --lib tools::file_read`
### 3. file_write 工具
**日期**: 2026-04-07
**Commit**: `16b052b`
#### 功能
- 写入内容到文件
- 自动创建父目录
- 覆盖已存在文件
#### Schema
```json
{
"type": "object",
"properties": {
"path": { "type": "string", "description": "文件路径" },
"content": { "type": "string", "description": "写入内容" }
},
"required": ["path", "content"]
}
```
#### 测试
- 5 个单元测试
- `cargo test --lib tools::file_write`
### 4. file_edit 工具
**日期**: 2026-04-07
**Commit**: `f3187ce`
#### 功能
- 编辑文件,替换 old_text 为 new_text
- 支持多行编辑
- 模糊匹配处理微小差异
- replace_all 选项批量替换
#### Schema
```json
{
"type": "object",
"properties": {
"path": { "type": "string", "description": "文件路径" },
"old_text": { "type": "string", "description": "要替换的文本" },
"new_text": { "type": "string", "description": "替换后的文本" },
"replace_all": { "type": "boolean", "description": "替换所有匹配", "default": false }
},
"required": ["path", "old_text", "new_text"]
}
```
#### 测试
- 5 个单元测试
- `cargo test --lib tools::file_edit`
### 5. bash 工具
**日期**: 2026-04-07
**Commit**: `68e3663`
#### 功能
- 执行 shell 命令
- 超时控制
- 危险命令检测rm -rf, fork bombs
- 输出截断
- 工作目录支持
#### Schema
```json
{
"type": "object",
"properties": {
"command": { "type": "string", "description": "Shell 命令" },
"timeout": { "type": "integer", "description": "超时秒数", "minimum": 1, "maximum": 600 }
},
"required": ["command"]
}
```
#### 测试
- 7 个单元测试
- `cargo test --lib tools::bash`
### 6. http_request 工具
**日期**: 2026-04-07
**Commit**: `1581732`
#### 功能
- HTTP 客户端支持 GET/POST/PUT/DELETE/PATCH
- 域名白名单
- SSRF 保护阻止私有IP、localhost
- 响应大小限制和截断
- 超时控制
#### Schema
```json
{
"type": "object",
"properties": {
"url": { "type": "string", "description": "请求 URL" },
"method": { "type": "string", "description": "HTTP 方法", "enum": ["GET", "POST", "PUT", "DELETE", "PATCH"] },
"headers": { "type": "object", "description": "请求头" },
"body": { "type": "string", "description": "请求体" }
},
"required": ["url"]
}
```
#### 测试
- 8 个单元测试
- `cargo test --lib tools::http_request`
### 7. web_fetch 工具
**日期**: 2026-04-07
**Commit**: `8936e70`
#### 功能
- 获取 URL 并提取可读文本
- HTML 转纯文本
- 移除 scripts, styles, HTML 标签
- 解码 HTML 实体
- JSON 格式化输出
- SSRF 保护
#### Schema
```json
{
"type": "object",
"properties": {
"url": { "type": "string", "description": "要获取的 URL" }
},
"required": ["url"]
}
```
#### 测试
- 6 个单元测试
- `cargo test --lib tools::web_fetch`
---
## 工具清单
| 工具 | 名称 | 文件 | 功能 |
|------|------|------|------|
| calculator | 计算器 | `src/tools/calculator.rs` | 25+ 数学和统计函数 |
| file_read | 文件读取 | `src/tools/file_read.rs` | 带分页的文件读取 |
| file_write | 文件写入 | `src/tools/file_write.rs` | 创建/覆盖文件 |
| file_edit | 文件编辑 | `src/tools/file_edit.rs` | 文本替换编辑 |
| bash | Shell 执行 | `src/tools/bash.rs` | 带安全保护的命令执行 |
| http_request | HTTP 请求 | `src/tools/http_request.rs` | API 请求 |
| web_fetch | 网页获取 | `src/tools/web_fetch.rs` | HTML 内容提取 |
## 工具机制增强
### SchemaCleanr
跨 LLM Provider 的 JSON Schema 归一化,支持:
- Gemini (最严格)
- Anthropic (中等)
- OpenAI (最宽松)
- Conservative (保守)
### 工具属性
```rust
fn read_only(&self) -> bool { false } // 是否只读
fn concurrency_safe(&self) -> bool { true } // 是否可并行
fn exclusive(&self) -> bool { false } // 是否独占
```
## 运行测试
```bash
cargo test --lib # 所有测试
cargo test --lib tools::schema # SchemaCleanr
```
## 工具注册
工具在 `src/gateway/session.rs``default_tools()` 函数中注册:
```rust
fn default_tools() -> ToolRegistry {
let mut registry = ToolRegistry::new();
registry.register(CalculatorTool::new());
registry.register(FileReadTool::new());
registry.register(FileWriteTool::new());
registry.register(FileEditTool::new());
registry.register(BashTool::new());
registry.register(HttpRequestTool::new(
vec!["*".to_string()], // 允许所有域名
1_000_000, // max_response_size
30, // timeout_secs
false, // allow_private_hosts
));
registry.register(WebFetchTool::new(50_000, 30));
registry
}
```
SessionManager 使用这些工具创建 AgentLoop 实例,所有工具自动对 LLM 可用。
## 更新日志
| 日期 | Commit | 变更 |
|------|--------|------|
| 2026-04-07 | `d5b6cd2` | feat: add SchemaCleanr |
| 2026-04-07 | `a9e7aab` | feat: add file_read tool |
| 2026-04-07 | `16b052b` | feat: add file_write tool |
| 2026-04-07 | `f3187ce` | feat: add file_edit tool |
| 2026-04-07 | `68e3663` | feat: add bash tool |
| 2026-04-07 | `1581732` | feat: add http_request tool |
| 2026-04-07 | `8936e70` | feat: add web_fetch tool |
| 2026-04-07 | `b13bb8c` | docs: add implementation log |
| 2026-04-08 | `98bc973` | feat: register all tools in SessionManager |
cargo test --lib tools::file_read # file_read
cargo test --lib tools::file_write # file_write
cargo test --lib tools::file_edit # file_edit
cargo test --lib tools::bash # bash
cargo test --lib tools::http_request # http_request
cargo test --lib tools::web_fetch # web_fetch
```

View File

@ -1,322 +0,0 @@
# PicoBot 持久化设计说明
本文档介绍 PicoBot 当前的会话持久化实现,目标读者是需要维护或集成该模块的技术人员。
## 1. 总览
PicoBot 使用 SQLite 持久化会话和消息历史,当前只有一份数据库文件:
- 默认路径:`~/.picobot/storage/sessions.db`
- 初始化入口:`SessionStore::new()`
- 核心实现:`src/storage/mod.rs`
数据库启动时会完成以下初始化:
- 打开 SQLite 连接
- 创建父目录
- 打开 WAL 模式
- 打开外键约束
- 自动建表和建索引
当前持久化只覆盖两类核心数据:
- `sessions`:会话元数据
- `messages`:会话内的消息流水
内存中的 `Session` 负责运行态处理SQLite 负责跨进程、跨重启保留历史。整体设计是“内存缓存 + SQLite 事实来源”。
## 2. 持久化在系统中的位置
相关模块职责如下:
- `src/gateway/session.rs`
- 管理运行时 `Session`
- 在收到消息时确保持久化会话存在
- 首次访问某个 `chat_id` 时从数据库加载历史
- 在新消息产生后同时写入数据库和内存历史
- `src/storage/mod.rs`
- 封装 SQLite 访问
- 提供会话和消息的增删改查
- `src/bus/message.rs`
- 定义持久化消息结构 `ChatMessage`
- `src/providers/*`
- 将历史消息转换为不同 LLM provider 需要的格式
典型关系如下:
1. 网关收到用户消息。
2. `SessionManager` 定位到对应 channel 的运行时 `Session`
3. `Session::ensure_persistent_session(chat_id)` 确保数据库里有对应会话记录。
4. `Session::ensure_chat_loaded(chat_id)` 在内存中没有历史时,从 `messages` 表加载该会话全部历史。
5. 新的用户消息先写入 `messages`,再放入内存历史。
6. Agent 执行后产生的 assistant/tool 消息按实际顺序继续写入 `messages`
7. 下次进程重启或 session 过期后,可从数据库完整恢复上下文。
## 3. 会话标识规则
数据库中的会话主键并不总是随机 UUID而是依据 channel 类型区分:
- CLI 会话:`session_id == chat_id`
- 非 CLI 会话:`session_id = "{channel_name}:{chat_id}"`
这套规则由 `persistent_session_id(channel_name, chat_id)` 统一生成,目的是:
- 对 CLI 支持显式创建、切换和管理多个会话
- 对外部渠道(例如飞书)让同一个 chat 稳定映射到同一条持久化会话
## 4. 表结构
### 4.1 sessions
保存会话级元数据,每条记录代表一个可被恢复的历史会话。
字段说明:
| 字段 | 类型 | 含义 | 当前用途 |
| --- | --- | --- | --- |
| `id` | `TEXT PRIMARY KEY` | 会话主键 | 作为会话唯一标识,被 `messages.session_id` 引用 |
| `title` | `TEXT NOT NULL` | 会话标题 | CLI 展示、重命名 |
| `channel_name` | `TEXT NOT NULL` | 来源渠道名 | 例如 `cli``feishu` |
| `chat_id` | `TEXT NOT NULL` | 渠道侧会话标识 | 用于恢复和路由到同一聊天 |
| `summary` | `TEXT` | 会话摘要 | 预留字段,当前 schema 中存在,但当前代码未写入实际摘要 |
| `created_at` | `INTEGER NOT NULL` | 创建时间 | 毫秒级 Unix 时间戳 |
| `updated_at` | `INTEGER NOT NULL` | 最近元数据更新时间 | 重命名、归档、追加消息时更新 |
| `last_active_at` | `INTEGER NOT NULL` | 最近活跃时间 | 追加消息、清空历史时更新,用于排序 |
| `archived_at` | `INTEGER` | 归档时间 | 非空表示会话已归档 |
| `deleted_at` | `INTEGER` | 删除时间 | 预留字段,当前读取逻辑会过滤该字段,但当前删除实现是物理删除 |
| `message_count` | `INTEGER NOT NULL DEFAULT 0` | 消息数 | 追加消息时自增,清空历史时重置 |
| `reset_cutoff_seq` | `INTEGER NOT NULL DEFAULT 0` | 逻辑重置切点 | `/reset` 后默认只恢复 `seq > reset_cutoff_seq` 的活动段 |
索引:
- `idx_sessions_channel_archived(channel_name, archived_at, last_active_at DESC)`
- 用于按渠道列出会话,并支持过滤归档态和按最近活跃排序
- `idx_sessions_updated_at(updated_at DESC)`
- 用于最近更新时间维度的查询优化
### 4.2 messages
保存会话中的消息流水。这里的“消息”不仅包括用户和助手文本,还包括工具调用结果。
字段说明:
| 字段 | 类型 | 含义 | 当前用途 |
| --- | --- | --- | --- |
| `id` | `TEXT PRIMARY KEY` | 消息唯一标识 | 对应 `ChatMessage.id` |
| `session_id` | `TEXT NOT NULL` | 所属会话 | 外键指向 `sessions.id` |
| `seq` | `INTEGER NOT NULL` | 会话内顺序号 | 保证同一会话消息顺序稳定 |
| `role` | `TEXT NOT NULL` | 消息角色 | 典型值为 `user``assistant``system``tool` |
| `content` | `TEXT NOT NULL` | 消息正文 | 文本内容或工具结果文本 |
| `media_refs_json` | `TEXT NOT NULL` | 媒体引用列表 JSON | 存储附件、本地文件路径等上下文引用 |
| `tool_call_id` | `TEXT` | 工具调用 ID | 仅 `role=tool` 时通常有值,用来关联某次工具结果对应哪一个 tool call |
| `tool_name` | `TEXT` | 工具名称 | 例如 `calculator``file_write` |
| `tool_calls_json` | `TEXT` | assistant 发起的工具调用列表 JSON | 仅 assistant 发出工具调用时有值 |
| `created_at` | `INTEGER NOT NULL` | 消息创建时间 | 毫秒级 Unix 时间戳 |
约束和索引:
- 外键:`FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE`
- 唯一约束:`UNIQUE(session_id, seq)`,确保同一会话内顺序号不重复
- 索引:
- `idx_messages_session_seq(session_id, seq)`,按顺序读取历史
- `idx_messages_session_created(session_id, created_at)`,按时间维度检索
## 5. 字段与运行时结构的映射
持久化层存储的消息对象是 `ChatMessage`,关键映射关系如下:
| `ChatMessage` 字段 | 对应数据库字段 | 说明 |
| --- | --- | --- |
| `id` | `messages.id` | 消息唯一 ID |
| `role` | `messages.role` | 消息角色 |
| `content` | `messages.content` | 文本主体 |
| `media_refs` | `messages.media_refs_json` | 序列化为 JSON 数组 |
| `timestamp` | `messages.created_at` | 时间戳 |
| `tool_call_id` | `messages.tool_call_id` | 工具结果与调用的关联 ID |
| `tool_name` | `messages.tool_name` | 工具名 |
| `tool_calls` | `messages.tool_calls_json` | assistant 发起的工具调用列表 |
设计上分成 `tool_call_id``tool_calls_json` 两种字段,是因为两者表达的是不同方向的信息:
- `tool_calls_json` 表示“assistant 想调用哪些工具”
- `tool_call_id` 表示“这一条 tool 结果是在回应哪一次工具调用”
## 6. 数据写入流程
### 6.1 创建会话
有两种进入方式:
- CLI 模式调用 `create_cli_session()` 显式创建会话
- 渠道消息进入时调用 `ensure_channel_session()` 自动创建或复用会话
创建时会写入 `sessions` 表,初始状态:
- `summary = NULL`
- `archived_at = NULL`
- `deleted_at = NULL`
- `message_count = 0`
### 6.2 追加消息
消息持久化统一走 `append_message()`,写入过程是一个 SQLite 事务:
1. 查询当前会话 `MAX(seq) + 1` 作为下一条消息顺序。
2. 将 `media_refs` 序列化为 `media_refs_json`
3. 将 `tool_calls` 序列化为 `tool_calls_json`
4. 插入一条 `messages` 记录。
5. 更新 `sessions.message_count``updated_at``last_active_at`
6. 将 `sessions.archived_at` 置空。
7. 提交事务。
其中第 6 步很重要:归档会话一旦收到新消息,会自动恢复为活跃态。
### 6.3 读取历史
`load_messages(session_id)` 会按 `seq ASC` 读取当前活动段历史,并把 JSON 字段反序列化回 `ChatMessage`。活动段的定义是:
- 只返回 `seq > sessions.reset_cutoff_seq` 的消息
- 因此 `/reset` 之后,旧消息仍然保留在数据库中,但不会默认回灌到运行时上下文
如果需要审计、导出或查看完整历史,应使用全量读取接口 `load_all_messages(session_id)`
因此运行态恢复的是“当前活动段的逻辑顺序”,而不是简单按创建时间排序。只要 `seq` 连续,重放顺序就稳定。
## 7. 典型时序
### 7.1 普通问答
1. 用户消息进入网关。
2. 如果数据库中没有对应会话,先插入一条 `sessions`
3. 用户消息写入 `messages``role = user`
4. Agent 基于历史生成回复。
5. assistant 回复写入 `messages``role = assistant`
6. 会话的 `message_count` 增加 2`last_active_at` 更新时间。
### 7.2 带工具调用的问答
1. assistant 先生成一条带 `tool_calls_json` 的消息,`role = assistant`
2. 系统执行对应工具。
3. 每个工具结果作为独立消息写入 `messages``role = tool`
4. 这些 `tool` 消息会带 `tool_call_id``tool_name`
5. assistant 最终整理工具结果后再写入一条普通回复。
这样保存后,即使进程重启,后续仍能完整恢复:
- assistant 当时发起了哪些工具调用
- 每个工具调用返回了什么
- 最终 assistant 给了什么结论
## 8. 会话生命周期操作
### 8.1 重命名
`rename_session(session_id, title)`
- 更新 `sessions.title`
- 更新 `sessions.updated_at`
### 8.2 归档
`archive_session(session_id)`
- 将 `sessions.archived_at` 设为当前时间
- 更新 `sessions.updated_at`
- 不删除消息数据
列出会话时:
- `include_archived = false` 只返回 `archived_at IS NULL` 的会话
- `include_archived = true` 返回全部未删除会话
### 8.3 清空消息
`clear_messages(session_id)`
- 删除该会话在 `messages` 中的所有记录
- 将 `sessions.message_count` 重置为 0
- 将 `sessions.reset_cutoff_seq` 重置为 0
- 更新 `updated_at``last_active_at`
- 保留会话本身
这适合“保留会话入口,但丢弃聊天内容”的场景。
### 8.4 逻辑重置
`reset_session(session_id)`
- 不删除 `messages` 中的任何记录
- 将当前会话的 `MAX(seq)` 写入 `sessions.reset_cutoff_seq`
- 更新 `updated_at``last_active_at`
- 后续默认恢复和发给模型的历史,只包含这次重置之后新增的消息
这适合“开始新对话,但保留完整历史以便审计或未来检索”的场景。
### 8.5 删除会话
`delete_session(session_id)`
- 显式删除 `messages`
- 再删除 `sessions`
虽然表结构中存在 `deleted_at` 字段,并且查询时也会过滤 `deleted_at IS NULL`,但当前实现并没有做软删除,而是直接物理删除。换句话说:
- `deleted_at` 当前是保留字段
- 如果后续需要回收站或审计恢复,可以基于它演进成软删除
## 9. 并发与一致性
当前 `SessionStore` 的一致性策略比较直接:
- 进程内使用 `Arc<Mutex<Connection>>` 保护单连接访问
- 追加消息时使用 SQLite 事务
- 单条消息的写入与会话计数更新在同一事务中完成
这意味着:
- 对单进程场景,消息顺序和 `message_count` 是一致的
- `seq` 通过事务内 `MAX(seq) + 1` 分配,避免同一连接并发下的顺序错乱
- WAL 模式提升读取和写入并存时的稳定性
需要注意的是,当前设计主要面向单进程本地运行。如果未来要扩展到多进程或多实例共享同一数据库,需要重新评估:
- 单连接模型
- `MAX(seq) + 1` 的扩展性
- 会话加载缓存和跨实例同步
## 10. 当前实现中的保留点
下面这些字段或能力已经在 schema 中出现,但还没有完整业务闭环:
- `sessions.summary`
- 当前代码没有把 `ContextCompressor` 产出的摘要写回数据库
- 目前摘要只参与运行时上下文压缩,不参与持久化
- `sessions.deleted_at`
- 当前查询逻辑兼容软删除
- 当前删除实现仍然是物理删除
- `sessions.reset_cutoff_seq`
- 当前已用于实现 `/reset` 的非破坏性逻辑重置
- 只影响默认恢复的活动段,不影响数据库中的全量历史
这说明当前 schema 已经为“会话摘要”和“软删除”预留了演进空间,但并未完全落地。
## 11. 给维护者的快速判断指南
如果你要排查持久化问题,可以先按下面的思路判断:
- 会话查不到:先看 `persistent_session_id` 是否和实际 `channel_name/chat_id` 一致
- 重启后没历史:检查 `ensure_chat_loaded()` 调用链,以及数据库文件路径是否正确
- `/reset` 后重启又带回旧上下文:检查 `sessions.reset_cutoff_seq` 是否已写入,以及恢复路径是否走了活动段读取而不是全量读取
- 消息顺序不对:检查 `messages.seq`
- 工具调用上下文异常:同时检查 `tool_calls_json``tool_call_id`
- 会话列表里看不到记录:检查 `archived_at``include_archived` 参数
- 清空后仍有上下文:确认是内存历史没清掉,还是数据库 `messages` 没删掉
## 12. 总结
PicoBot 当前的持久化设计比较克制,核心目标只有两个:
- 让同一会话在重启后可以恢复上下文
- 让工具调用链可以被完整回放
从实现上看,它不是通用 ORM也不是复杂事件存储而是一层针对聊天历史的轻量 SQLite 封装。对于当前单机、单进程、聊天驱动的运行方式,这个设计足够直接,也便于继续演进。

View File

@ -25,13 +25,6 @@ impl OutboundDispatcher {
loop {
let msg = self.bus.consume_outbound().await;
#[cfg(debug_assertions)]
tracing::debug!(
channel = %msg.channel,
chat_id = %msg.chat_id,
content_len = msg.content.len(),
"OutboundDispatcher received message"
);
let channel_name = msg.channel.clone();
let channel = self.channel_manager.get_channel(&channel_name).await;

View File

@ -164,6 +164,7 @@ pub struct InboundMessage {
pub channel: String,
pub sender_id: String,
pub chat_id: String,
pub dialog_id: Option<String>,
pub content: String,
pub timestamp: i64,
pub media: Vec<MediaItem>,
@ -199,52 +200,20 @@ impl OutboundMessage {
}
}
// ============================================================================
// ControlInbound - Session management operations (CLI channel only)
// ============================================================================
/// Session management operations that flow through the control channel
#[derive(Debug, Clone)]
pub enum ControlInbound {
CreateSession { title: Option<String> },
ListSessions { include_archived: bool },
LoadSession { session_id: String },
RenameSession { session_id: String, title: String },
ArchiveSession { session_id: String },
DeleteSession { session_id: String },
ClearHistory { session_id: String },
}
// ============================================================================
// ControlOutbound - Responses for control operations
// ============================================================================
/// Responses for session management operations
#[derive(Debug, Clone)]
pub enum ControlOutbound {
SessionCreated { session_id: String, title: String },
SessionList { sessions: Vec<crate::protocol::SessionSummary> },
SessionLoaded { session_id: String, title: String, message_count: i64 },
SessionRenamed { session_id: String, title: String },
SessionArchived { session_id: String },
SessionDeleted { session_id: String },
HistoryCleared { session_id: String },
Pong,
Error { code: String, message: String },
}
// ============================================================================
// ControlMessage - Message for control channel (session management)
// Uses SessionCommand from session module
// ============================================================================
use crate::channels::base::ChannelError;
use crate::session::{SessionCommand, SessionEvent};
use tokio::sync::mpsc;
/// Control message containing a session operation and reply channel
#[derive(Debug, Clone)]
pub struct ControlMessage {
pub op: ControlInbound,
pub reply_tx: mpsc::Sender<Result<ControlOutbound, ChannelError>>,
pub op: SessionCommand,
pub reply_tx: mpsc::Sender<Result<SessionEvent, ChannelError>>,
}
// ============================================================================

View File

@ -2,7 +2,7 @@ pub mod dispatcher;
pub mod message;
pub use dispatcher::OutboundDispatcher;
pub use message::{ChatMessage, ContentBlock, ControlInbound, ControlMessage, ControlOutbound, InboundMessage, MediaItem, OutboundMessage};
pub use message::{ChatMessage, ContentBlock, ControlMessage, InboundMessage, MediaItem, OutboundMessage};
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex};

View File

@ -3,10 +3,17 @@ use async_trait::async_trait;
use tokio::sync::{mpsc, Mutex};
use uuid::Uuid;
use crate::bus::{ControlInbound, ControlMessage, ControlOutbound, InboundMessage, MessageBus, OutboundMessage};
use crate::bus::{ControlMessage, InboundMessage, MessageBus, OutboundMessage};
use crate::session::{SessionCommand, SessionEvent, UnifiedSessionId};
use crate::protocol::{parse_inbound, WsInbound, WsOutbound};
use super::base::{Channel, ChannelError};
use super::slash_command::parse_slash_command;
/// Generate a short ID (8 characters) from a UUID
fn short_id() -> String {
Uuid::new_v4().to_string()[..8].to_string()
}
// ============================================================================
// Client - Connected CLI client
@ -36,6 +43,9 @@ impl CliChatChannel {
/// Register a new client connection, returns (session_id, client)
pub(crate) async fn register_client(&self, sender: mpsc::Sender<WsOutbound>) -> (String, Arc<Client>) {
// Generate connection ID (used as chat_id) - use short ID
let connection_id = short_id();
let client = Arc::new(Client {
sender,
current_session_id: Mutex::new(None),
@ -43,11 +53,12 @@ impl CliChatChannel {
self.clients.lock().await.push(client.clone());
// Create initial session via control message
let session_id = match self.create_session_via_control(None).await {
let session_id = match self.create_session_via_control(&connection_id, None).await {
Ok(id) => id,
Err(e) => {
tracing::error!(error = %e, "Failed to create initial session");
Uuid::new_v4().to_string()
// Fall back to old format for backward compatibility
connection_id.clone()
}
};
@ -101,21 +112,64 @@ impl CliChatChannel {
match inbound {
WsInbound::UserInput { content, chat_id, .. } => {
let chat_id = chat_id.or(current_session_guard.clone()).unwrap_or_else(|| Uuid::new_v4().to_string());
let chat_id = chat_id.or(current_session_guard.clone()).unwrap_or_else(short_id);
// If no session, create one first
if current_session_guard.is_none() {
let new_id = self.create_session_via_control(None).await?;
let new_id = self.create_session_via_control(&chat_id, None).await?;
*current_session_guard = Some(new_id);
}
let session_id = current_session_guard.clone().unwrap();
// Check for slash command
if let Some((cmd_name, _)) = parse_slash_command(&content) {
// Send ExecuteSlashCommand via control plane
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let unified_id = UnifiedSessionId::parse(&session_id);
bus.publish_control(ControlMessage {
op: SessionCommand::ExecuteSlashCommand {
command: cmd_name.to_string(),
channel: self.name().to_string(),
chat_id: chat_id.clone(),
current_session_id: unified_id,
},
reply_tx,
}).await?;
// Handle response
if let Some(result) = reply_rx.recv().await {
match result {
Ok(SessionEvent::SlashCommandExecuted { new_session_id, message }) => {
// Update current session if new one was created
if let Some(new_id) = new_session_id {
*current_session_guard = Some(new_id.to_string());
}
let _ = client.sender.send(WsOutbound::CommandExecuted { message }).await;
}
Ok(SessionEvent::Error { code, message }) => {
let _ = client.sender.send(WsOutbound::Error { code, message }).await;
}
Err(e) => {
let _ = client.sender.send(WsOutbound::Error { code: "EXECUTION_ERROR".to_string(), message: e.to_string() }).await;
}
_ => {}
}
}
return Ok(());
}
// Parse UnifiedSessionId to get chat_id and dialog_id
let (channel_name, chat_id_part, dialog_id_part) = UnifiedSessionId::parse(&session_id)
.map(|sid| (sid.channel, sid.chat_id, Some(sid.dialog_id.clone())))
.unwrap_or_else(|| (self.name().to_string(), session_id.clone(), None));
// Publish to bus for AI processing
let msg = InboundMessage {
channel: self.name().to_string(),
channel: channel_name,
sender_id: "cli".to_string(),
chat_id: session_id.clone(),
chat_id: chat_id_part,
dialog_id: dialog_id_part,
content,
timestamp: crate::bus::message::current_timestamp(),
media: Vec::new(),
@ -131,13 +185,15 @@ impl CliChatChannel {
.ok_or_else(|| ChannelError::Other("No active session".to_string()))?;
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let session_id = UnifiedSessionId::parse(&target)
.ok_or_else(|| ChannelError::Other("Invalid session ID format".to_string()))?;
bus.publish_control(ControlMessage {
op: ControlInbound::ClearHistory { session_id: target.clone() },
op: SessionCommand::ClearHistory { session_id },
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::HistoryCleared { .. })) => {
Some(Ok(SessionEvent::HistoryCleared { .. })) => {
let _ = client
.sender
.send(WsOutbound::HistoryCleared { session_id: target })
@ -155,7 +211,10 @@ impl CliChatChannel {
}
}
WsInbound::CreateSession { title } => {
let new_id = self.create_session_via_control(title.as_deref()).await?;
// Use current session's chat_id if available, otherwise generate new one
let chat_id = current_session_guard.clone()
.unwrap_or_else(short_id);
let new_id = self.create_session_via_control(&chat_id, title.as_deref()).await?;
*current_session_guard = Some(new_id.clone());
let _ = client
.sender
@ -166,19 +225,42 @@ impl CliChatChannel {
.await;
}
WsInbound::ListSessions { include_archived } => {
// List dialogs for the current chat
let chat_id = current_session_guard.clone()
.unwrap_or_else(|| "".to_string());
let chat_id_for_response = chat_id.clone();
let (reply_tx, mut reply_rx) = mpsc::channel(1);
bus.publish_control(ControlMessage {
op: ControlInbound::ListSessions { include_archived },
op: SessionCommand::ListDialogs {
channel: "cli_chat".to_string(),
chat_id,
include_archived,
},
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionList { sessions })) => {
Some(Ok(SessionEvent::DialogList { dialogs, current_dialog_id })) => {
// Convert DialogInfo to SessionSummary for backward compatibility
let sessions: Vec<crate::protocol::SessionSummary> = dialogs.into_iter().map(|d| {
crate::protocol::SessionSummary {
session_id: d.session_id.to_string(),
title: d.title,
channel_name: d.session_id.channel.clone(),
chat_id: d.session_id.chat_id,
message_count: d.message_count,
last_active_at: d.last_active_at,
archived_at: d.archived_at,
}
}).collect();
let current_session_id = current_dialog_id.map(|did| {
UnifiedSessionId::new("cli_chat", chat_id_for_response.clone(), did).to_string()
});
let _ = client
.sender
.send(WsOutbound::SessionList {
sessions,
current_session_id: current_session_guard.clone(),
current_session_id,
})
.await;
}
@ -194,28 +276,44 @@ impl CliChatChannel {
}
}
WsInbound::LoadSession { session_id } => {
// LoadSession: parse the session_id and get current dialog info
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let unified_id = UnifiedSessionId::parse(&session_id)
.ok_or_else(|| ChannelError::Other("Invalid session ID format".to_string()))?;
bus.publish_control(ControlMessage {
op: ControlInbound::LoadSession { session_id: session_id.clone() },
op: SessionCommand::GetCurrentDialog {
channel: unified_id.channel.clone(),
chat_id: unified_id.chat_id.clone(),
},
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionLoaded { session_id, title, message_count })) => {
*current_session_guard = Some(session_id.clone());
let _ = client
.sender
.send(WsOutbound::SessionLoaded {
session_id,
title,
message_count,
})
.await;
Some(Ok(SessionEvent::CurrentDialog { session_id: current_session_id_opt })) => {
if let Some(current_session_id) = current_session_id_opt {
*current_session_guard = Some(current_session_id.to_string());
let _ = client
.sender
.send(WsOutbound::SessionLoaded {
session_id: current_session_id.to_string(),
title: "Session".to_string(), // TODO: get actual title
message_count: 0, // TODO: get actual count
})
.await;
} else {
let _ = client
.sender
.send(WsOutbound::Error {
code: "NO_CURRENT_DIALOG".to_string(),
message: "No current dialog".to_string(),
})
.await;
}
}
Some(Ok(_)) => {
// Unexpected response type
}
Some(Err(e)) => {
Some(Err(_e)) => {
let _ = client
.sender
.send(WsOutbound::Error {
@ -235,16 +333,18 @@ impl CliChatChannel {
})?;
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let unified_id = UnifiedSessionId::parse(&target)
.ok_or_else(|| ChannelError::Other("Invalid session ID format".to_string()))?;
bus.publish_control(ControlMessage {
op: ControlInbound::RenameSession { session_id: target.clone(), title: title.clone() },
op: SessionCommand::RenameDialog { session_id: unified_id, title: title.clone() },
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionRenamed { session_id, title })) => {
Some(Ok(SessionEvent::DialogRenamed { session_id, title })) => {
let _ = client
.sender
.send(WsOutbound::SessionRenamed { session_id, title })
.send(WsOutbound::SessionRenamed { session_id: session_id.to_string(), title })
.await;
}
Some(Ok(_)) => {
@ -264,16 +364,18 @@ impl CliChatChannel {
})?;
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let unified_id = UnifiedSessionId::parse(&target)
.ok_or_else(|| ChannelError::Other("Invalid session ID format".to_string()))?;
bus.publish_control(ControlMessage {
op: ControlInbound::ArchiveSession { session_id: target.clone() },
op: SessionCommand::ArchiveDialog { session_id: unified_id },
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionArchived { session_id })) => {
Some(Ok(SessionEvent::DialogArchived { session_id })) => {
let _ = client
.sender
.send(WsOutbound::SessionArchived { session_id })
.send(WsOutbound::SessionArchived { session_id: session_id.to_string() })
.await;
}
Some(Ok(_)) => {
@ -293,22 +395,24 @@ impl CliChatChannel {
})?;
let (reply_tx, mut reply_rx) = mpsc::channel(1);
let unified_id = UnifiedSessionId::parse(&target)
.ok_or_else(|| ChannelError::Other("Invalid session ID format".to_string()))?;
bus.publish_control(ControlMessage {
op: ControlInbound::DeleteSession { session_id: target.clone() },
op: SessionCommand::DeleteDialog { session_id: unified_id },
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionDeleted { session_id })) => {
Some(Ok(SessionEvent::DialogDeleted { session_id })) => {
let _ = client
.sender
.send(WsOutbound::SessionDeleted { session_id: session_id.clone() })
.send(WsOutbound::SessionDeleted { session_id: session_id.to_string() })
.await;
// If deleting current session, create a new one
if current_session_guard.as_deref() == Some(&target) {
drop(reply_rx);
if let Ok(new_id) = self.create_session_via_control(None).await {
if let Ok(new_id) = self.create_session_via_control(&target, None).await {
*current_session_guard = Some(new_id.clone());
let _ = client
.sender
@ -339,7 +443,7 @@ impl CliChatChannel {
}
/// Create a session via control message and return the session_id
async fn create_session_via_control(&self, title: Option<&str>) -> Result<String, ChannelError> {
async fn create_session_via_control(&self, connection_id: &str, title: Option<&str>) -> Result<String, ChannelError> {
let bus = {
let guard = self.bus.lock().unwrap();
guard.clone().ok_or_else(|| ChannelError::Other("Channel not started".to_string()))?
@ -347,13 +451,17 @@ impl CliChatChannel {
let (reply_tx, mut reply_rx) = mpsc::channel(1);
bus.publish_control(ControlMessage {
op: ControlInbound::CreateSession { title: title.map(String::from) },
op: SessionCommand::CreateDialog {
channel: "cli_chat".to_string(),
chat_id: connection_id.to_string(),
title: title.map(String::from),
},
reply_tx,
}).await?;
match reply_rx.recv().await {
Some(Ok(ControlOutbound::SessionCreated { session_id, .. })) => {
Ok(session_id)
Some(Ok(SessionEvent::DialogCreated { session_id, .. })) => {
Ok(session_id.to_string())
}
Some(Ok(_)) => {
Err(ChannelError::Other("Unexpected response type".to_string()))
@ -388,7 +496,7 @@ impl Channel for CliChatChannel {
let clients = self.clients.lock().await.clone();
for client in clients {
let outbound = WsOutbound::AssistantResponse {
id: Uuid::new_v4().to_string(),
id: short_id(),
content: msg.content.clone(),
role: "assistant".to_string(),
};

View File

@ -1106,6 +1106,7 @@ impl FeishuChannel {
channel: "feishu".to_string(),
sender_id: parsed.open_id.clone(),
chat_id: parsed.chat_id.clone(),
dialog_id: None, // Use default/current dialog
content: parsed.content.clone(),
timestamp: crate::bus::message::current_timestamp(),
media: parsed.media.map(|m| vec![m]).unwrap_or_default(),

View File

@ -2,8 +2,10 @@ pub mod base;
pub mod feishu;
pub mod cli_chat;
pub mod manager;
pub mod slash_command;
pub use base::{Channel, ChannelError};
pub use manager::ChannelManager;
pub use feishu::FeishuChannel;
pub use cli_chat::CliChatChannel;
pub use slash_command::{parse_slash_command, command_matches};

View File

@ -0,0 +1,44 @@
/// 解析斜杠命令
/// 返回 (command_name, args) 或 None
pub fn parse_slash_command(content: &str) -> Option<(&str, &str)> {
let trimmed = content.trim();
if !trimmed.starts_with('/') {
return None;
}
let rest = &trimmed[1..];
if let Some((name, args)) = rest.split_once(' ') {
Some((name, args.trim()))
} else {
Some((rest, ""))
}
}
/// 检查内容是否匹配指定命令
pub fn command_matches(content: &str, aliases: &[&str]) -> bool {
let trimmed = content.trim();
aliases.iter().any(|&alias| trimmed == alias || trimmed.starts_with(&format!("{} ", alias)))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_slash_command() {
assert_eq!(parse_slash_command("/reset"), Some(("reset", "")));
assert_eq!(parse_slash_command("/reset arg"), Some(("reset", "arg")));
assert_eq!(parse_slash_command("/new hello world"), Some(("new", "hello world")));
assert_eq!(parse_slash_command("hello"), None);
assert_eq!(parse_slash_command("/"), Some(("", "")));
}
#[test]
fn test_command_matches() {
let aliases = &["/reset", "/new"];
assert!(command_matches("/reset", aliases));
assert!(command_matches("/new", aliases));
assert!(command_matches("/reset arg", aliases));
assert!(!command_matches("/help", aliases));
assert!(!command_matches("reset", aliases));
}
}

View File

@ -1,50 +0,0 @@
use tokio::io::{AsyncBufReadExt, BufReader, AsyncWriteExt};
pub struct CliChannel {
read: BufReader<tokio::io::Stdin>,
write: tokio::io::Stdout,
}
impl CliChannel {
pub fn new() -> Self {
Self {
read: BufReader::new(tokio::io::stdin()),
write: tokio::io::stdout(),
}
}
pub async fn read_line(&mut self, prompt: &str) -> Result<Option<String>, std::io::Error> {
print!("{}", prompt);
self.write.flush().await?;
let mut line = String::new();
let bytes_read = self.read.read_line(&mut line).await?;
if bytes_read == 0 {
return Ok(None);
}
Ok(Some(line.trim_end().to_string()))
}
pub async fn write_line(&mut self, content: &str) -> Result<(), std::io::Error> {
self.write.write_all(content.as_bytes()).await?;
self.write.write_all(b"\n").await?;
self.write.flush().await
}
pub async fn write_response(&mut self, content: &str) -> Result<(), std::io::Error> {
for line in content.lines() {
self.write.write_all(b" ").await?;
self.write.write_all(line.as_bytes()).await?;
self.write.write_all(b"\n").await?;
}
self.write.flush().await
}
}
impl Default for CliChannel {
fn default() -> Self {
Self::new()
}
}

View File

@ -1,127 +0,0 @@
use super::channel::CliChannel;
pub enum InputEvent {
Message(String),
Command(InputCommand),
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InputCommand {
Exit,
Clear,
New(Option<String>),
Sessions,
Use(String),
Rename(String),
Archive,
Delete,
}
pub struct InputHandler {
channel: CliChannel,
}
impl InputHandler {
pub fn new() -> Self {
Self {
channel: CliChannel::new(),
}
}
pub async fn read_input(&mut self, prompt: &str) -> Result<Option<InputEvent>, InputError> {
match self.channel.read_line(prompt).await {
Ok(Some(line)) => {
if line.trim().is_empty() {
return Ok(None);
}
if let Some(cmd) = self.handle_special_commands(&line) {
return Ok(Some(InputEvent::Command(cmd)));
}
Ok(Some(InputEvent::Message(line)))
}
Ok(None) => Ok(None),
Err(e) => Err(InputError::IoError(e)),
}
}
pub async fn write_output(&mut self, content: &str) -> Result<(), InputError> {
self.channel.write_line(content).await.map_err(InputError::IoError)
}
pub async fn write_response(&mut self, content: &str) -> Result<(), InputError> {
self.channel.write_response(content).await.map_err(InputError::IoError)
}
fn handle_special_commands(&self, line: &str) -> Option<InputCommand> {
let trimmed = line.trim();
let mut parts = trimmed.splitn(2, char::is_whitespace);
let command = parts.next()?;
let arg = parts.next().map(str::trim).filter(|value| !value.is_empty());
match command {
"/quit" | "/exit" | "/q" => Some(InputCommand::Exit),
"/clear" => Some(InputCommand::Clear),
"/new" => Some(InputCommand::New(arg.map(ToOwned::to_owned))),
"/sessions" => Some(InputCommand::Sessions),
"/use" => arg.map(|value| InputCommand::Use(value.to_string())),
"/rename" => arg.map(|value| InputCommand::Rename(value.to_string())),
"/archive" => Some(InputCommand::Archive),
"/delete" => Some(InputCommand::Delete),
_ => None,
}
}
}
impl Default for InputHandler {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug)]
pub enum InputError {
IoError(std::io::Error),
}
impl std::fmt::Display for InputError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
InputError::IoError(e) => write!(f, "IO error: {}", e),
}
}
}
impl std::error::Error for InputError {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_special_command_parsing() {
let handler = InputHandler::new();
assert_eq!(handler.handle_special_commands("/quit"), Some(InputCommand::Exit));
assert_eq!(handler.handle_special_commands("/clear"), Some(InputCommand::Clear));
assert_eq!(handler.handle_special_commands("/new"), Some(InputCommand::New(None)));
assert_eq!(
handler.handle_special_commands("/new planning"),
Some(InputCommand::New(Some("planning".to_string())))
);
assert_eq!(handler.handle_special_commands("/sessions"), Some(InputCommand::Sessions));
assert_eq!(
handler.handle_special_commands("/use abc123"),
Some(InputCommand::Use("abc123".to_string()))
);
assert_eq!(
handler.handle_special_commands("/rename project alpha"),
Some(InputCommand::Rename("project alpha".to_string()))
);
assert_eq!(handler.handle_special_commands("/archive"), Some(InputCommand::Archive));
assert_eq!(handler.handle_special_commands("/delete"), Some(InputCommand::Delete));
assert_eq!(handler.handle_special_commands("/unknown"), None);
assert_eq!(handler.handle_special_commands("/use"), None);
}
}

View File

@ -1,214 +1,121 @@
pub use crate::protocol::{WsInbound, WsOutbound, serialize_inbound, serialize_outbound};
mod channel;
mod input;
mod tui;
use futures_util::{SinkExt, StreamExt};
use crate::client::tui::app::{App, MessageRole};
use crate::client::tui::event::handle_key_event;
use crate::client::tui::ui::render_ui;
use crossterm::{
event::{self, Event},
execute,
terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
};
use futures_util::StreamExt;
use ratatui::{prelude::CrosstermBackend, Terminal};
use std::io;
use tokio_tungstenite::{connect_async, tungstenite::Message};
use input::{InputCommand, InputEvent, InputHandler};
fn format_session_list(sessions: &[crate::protocol::SessionSummary], current_session_id: Option<&str>) -> String {
if sessions.is_empty() {
return "No sessions found.".to_string();
}
let mut lines = Vec::with_capacity(sessions.len() + 1);
lines.push("Sessions:".to_string());
for session in sessions {
let marker = if current_session_id == Some(session.session_id.as_str()) {
"*"
} else {
"-"
};
let archived = if session.archived_at.is_some() {
" [archived]"
} else {
""
};
lines.push(format!(
"{} {} | {} | {} messages{}",
marker,
session.session_id,
session.title,
session.message_count,
archived,
));
}
lines.join("\n")
}
fn parse_message(raw: &str) -> Result<WsOutbound, serde_json::Error> {
serde_json::from_str(raw)
}
pub async fn run(gateway_url: &str) -> Result<(), Box<dyn std::error::Error>> {
let (ws_stream, _) = connect_async(gateway_url).await?;
tracing::info!(url = %gateway_url, "Connected to gateway");
let (mut sender, mut receiver) = ws_stream.split();
let (ws_sender, ws_receiver) = ws_stream.split();
let mut input = InputHandler::new();
let mut current_session_id: Option<String> = None;
input.write_output("picobot CLI - Commands: /new [title], /reset, /sessions, /use <session>, /rename <title>, /archive, /delete, /clear, /quit\n").await?;
let mut app = App::new(gateway_url.to_string());
app.ws_sender = Some(ws_sender);
app.ws_receiver = Some(ws_receiver);
// 初始化终端
enable_raw_mode()?;
let mut stdout = io::stdout();
execute!(stdout, EnterAlternateScreen)?;
let backend = CrosstermBackend::new(stdout);
let mut terminal = Terminal::new(backend)?;
terminal.clear()?;
let result = run_app(&mut terminal, app).await;
// 清理终端 - 确保正确的顺序,忽略错误
let _ = execute!(terminal.backend_mut(), LeaveAlternateScreen);
let _ = disable_raw_mode();
let _ = terminal.show_cursor();
result
}
async fn run_app(
terminal: &mut Terminal<CrosstermBackend<io::Stdout>>,
mut app: App,
) -> Result<(), Box<dyn std::error::Error>> {
let mut ws_receiver = app.ws_receiver.take().unwrap();
let mut event_reader = event::EventStream::new();
// Main loop: poll both stdin and WebSocket
loop {
terminal.draw(|f| render_ui(f, &app))?;
tokio::select! {
// Handle WebSocket messages
msg = receiver.next() => {
msg = ws_receiver.next() => {
match msg {
Some(Ok(Message::Text(text))) => {
let text = text.to_string();
if let Ok(outbound) = parse_message(&text) {
match outbound {
WsOutbound::AssistantResponse { content, .. } => {
input.write_response(&content).await?;
}
WsOutbound::Error { message, .. } => {
input.write_output(&format!("Error: {}", message)).await?;
}
WsOutbound::SessionEstablished { session_id } => {
current_session_id = Some(session_id.clone());
#[cfg(debug_assertions)]
tracing::debug!(session_id = %session_id, "Session established");
input.write_output(&format!("Session: {}\n", session_id)).await?;
}
WsOutbound::SessionCreated { session_id, title } => {
current_session_id = Some(session_id.clone());
input.write_output(&format!("Created session: {} ({})\n", session_id, title)).await?;
}
WsOutbound::SessionList { sessions, current_session_id: listed_current } => {
let display = format_session_list(&sessions, listed_current.as_deref());
input.write_output(&format!("{}\n", display)).await?;
}
WsOutbound::SessionLoaded { session_id, title, message_count } => {
current_session_id = Some(session_id.clone());
input.write_output(&format!("Loaded session: {} ({}, {} messages)\n", session_id, title, message_count)).await?;
}
WsOutbound::SessionRenamed { session_id, title } => {
input.write_output(&format!("Renamed session: {} -> {}\n", session_id, title)).await?;
}
WsOutbound::SessionArchived { session_id } => {
input.write_output(&format!("Archived session: {}\n", session_id)).await?;
}
WsOutbound::SessionDeleted { session_id } => {
if current_session_id.as_deref() == Some(session_id.as_str()) {
current_session_id = None;
}
input.write_output(&format!("Deleted session: {}\n", session_id)).await?;
}
WsOutbound::HistoryCleared { session_id } => {
input.write_output(&format!("Cleared history for session: {}\n", session_id)).await?;
}
_ => {}
}
if let Ok(outbound) = serde_json::from_str::<WsOutbound>(&text) {
handle_ws_message(&mut app, outbound);
}
}
Some(Ok(Message::Close(_))) | None => {
tracing::info!("Gateway disconnected");
input.write_output("Gateway disconnected").await?;
break;
app.quit();
}
_ => {}
}
}
// Handle stdin input
result = input.read_input("> ") => {
match result {
Ok(Some(event)) => {
match event {
InputEvent::Command(InputCommand::Exit) => {
input.write_output("Goodbye!").await?;
break;
}
InputEvent::Command(InputCommand::Clear) => {
let inbound = WsInbound::ClearHistory {
chat_id: None,
session_id: current_session_id.clone(),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::New(title)) => {
let inbound = WsInbound::CreateSession { title };
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::Sessions) => {
let inbound = WsInbound::ListSessions {
include_archived: true,
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::Use(session_id)) => {
let inbound = WsInbound::LoadSession { session_id };
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::Rename(title)) => {
let inbound = WsInbound::RenameSession {
session_id: current_session_id.clone(),
title,
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::Archive) => {
let inbound = WsInbound::ArchiveSession {
session_id: current_session_id.clone(),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Command(InputCommand::Delete) => {
let inbound = WsInbound::DeleteSession {
session_id: current_session_id.clone(),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(Message::Text(text.into())).await;
}
continue;
}
InputEvent::Message(content) => {
let inbound = WsInbound::UserInput {
content,
channel: None,
chat_id: current_session_id.clone(),
sender_id: None,
};
if let Ok(text) = serialize_inbound(&inbound) {
if sender.send(Message::Text(text.into())).await.is_err() {
tracing::error!("Failed to send message to gateway");
break;
}
}
}
}
}
Ok(None) => break,
Err(e) => {
tracing::error!(error = %e, "Input error");
break;
}
event_result = event_reader.next() => {
if let Some(Ok(Event::Key(key))) = event_result {
handle_key_event(&mut app, key).await;
}
}
}
if app.should_quit {
break;
}
}
Ok(())
}
fn handle_ws_message(app: &mut App, outbound: WsOutbound) {
match outbound {
WsOutbound::AssistantResponse { content, .. } => {
app.add_message(MessageRole::Assistant, content);
}
WsOutbound::Error { message, .. } => {
app.add_message(MessageRole::System, format!("Error: {}", message));
}
WsOutbound::SessionEstablished { session_id } => {
app.set_current_session(Some(session_id));
}
WsOutbound::SessionCreated { session_id, .. } => {
app.set_current_session(Some(session_id));
}
WsOutbound::SessionList { sessions, current_session_id } => {
app.set_sessions(sessions);
if let Some(id) = current_session_id {
app.set_current_session(Some(id));
}
}
WsOutbound::SessionLoaded { session_id, .. } => {
app.set_current_session(Some(session_id));
}
WsOutbound::SessionRenamed { .. } => {}
WsOutbound::SessionArchived { .. } => {}
WsOutbound::SessionDeleted { session_id } => {
if app.current_session_id.as_ref() == Some(&session_id) {
app.set_current_session(None);
}
}
WsOutbound::HistoryCleared { .. } => {
app.messages.clear();
}
_ => {}
}
}

155
src/client/tui/app.rs Normal file
View File

@ -0,0 +1,155 @@
use crate::protocol::SessionSummary;
use std::collections::VecDeque;
use tokio_tungstenite::tungstenite::Message;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum FocusArea {
TitleBar,
SessionList,
ChatHistory,
InputArea,
}
#[derive(Debug, Clone)]
pub enum MessageRole {
User,
Assistant,
System,
}
#[derive(Debug, Clone)]
pub struct ChatMessage {
pub role: MessageRole,
pub content: String,
}
pub struct App {
pub gateway_url: String,
pub ws_sender: Option<
futures_util::stream::SplitSink<
tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>,
Message,
>,
>,
pub ws_receiver: Option<
futures_util::stream::SplitStream<
tokio_tungstenite::WebSocketStream<
tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
>,
>,
>,
pub current_session_id: Option<String>,
pub sessions: Vec<SessionSummary>,
pub messages: VecDeque<ChatMessage>,
pub focus: FocusArea,
pub input: String,
pub input_cursor_pos: usize,
pub show_help: bool,
pub chat_scroll_offset: u16,
pub session_scroll_offset: u16,
pub should_quit: bool,
}
impl App {
pub fn new(gateway_url: String) -> Self {
Self {
gateway_url,
ws_sender: None,
ws_receiver: None,
current_session_id: None,
sessions: Vec::new(),
messages: VecDeque::new(),
focus: FocusArea::InputArea,
input: String::new(),
input_cursor_pos: 0,
show_help: false,
chat_scroll_offset: 0,
session_scroll_offset: 0,
should_quit: false,
}
}
pub fn add_message(&mut self, role: MessageRole, content: String) {
self.messages.push_back(ChatMessage { role, content });
self.chat_scroll_offset = 0;
}
pub fn set_sessions(&mut self, sessions: Vec<SessionSummary>) {
self.sessions = sessions;
}
pub fn set_current_session(&mut self, session_id: Option<String>) {
self.current_session_id = session_id;
self.messages.clear();
}
pub fn scroll_chat_up(&mut self) {
self.chat_scroll_offset = self.chat_scroll_offset.saturating_add(1);
}
pub fn scroll_chat_down(&mut self) {
self.chat_scroll_offset = self.chat_scroll_offset.saturating_sub(1);
}
pub fn scroll_session_up(&mut self) {
self.session_scroll_offset = self.session_scroll_offset.saturating_add(1);
}
pub fn scroll_session_down(&mut self) {
self.session_scroll_offset = self.session_scroll_offset.saturating_sub(1);
}
pub fn input_insert_char(&mut self, c: char) {
self.input.insert(self.input_cursor_pos, c);
self.input_cursor_pos += 1;
}
pub fn input_delete_char(&mut self) {
if self.input_cursor_pos > 0 {
self.input.remove(self.input_cursor_pos - 1);
self.input_cursor_pos -= 1;
}
}
pub fn input_move_cursor_left(&mut self) {
self.input_cursor_pos = self.input_cursor_pos.saturating_sub(1);
}
pub fn input_move_cursor_right(&mut self) {
if self.input_cursor_pos < self.input.len() {
self.input_cursor_pos += 1;
}
}
pub fn input_move_cursor_to_start(&mut self) {
self.input_cursor_pos = 0;
}
pub fn input_move_cursor_to_end(&mut self) {
self.input_cursor_pos = self.input.len();
}
pub fn input_clear(&mut self) {
self.input.clear();
self.input_cursor_pos = 0;
}
pub fn take_input(&mut self) -> String {
let input = std::mem::take(&mut self.input);
self.input_cursor_pos = 0;
input
}
pub fn toggle_help(&mut self) {
self.show_help = !self.show_help;
}
pub fn quit(&mut self) {
self.should_quit = true;
}
}

View File

@ -0,0 +1,37 @@
use crate::client::tui::app::{App, MessageRole};
use ratatui::{
layout::Rect,
style::{Color, Modifier, Style},
text::Line,
widgets::{Block, Borders, List, ListItem},
Frame,
};
pub fn render(f: &mut Frame, area: Rect, app: &App) {
let items: Vec<ListItem> = app
.messages
.iter()
.map(|msg| {
let (prefix, color) = match msg.role {
MessageRole::User => ("[User] ", Color::Blue),
MessageRole::Assistant => ("[Assistant] ", Color::Green),
MessageRole::System => ("[System] ", Color::Red),
};
let content = vec![
Line::from(vec![ratatui::text::Span::styled(
prefix,
Style::default().fg(color).add_modifier(Modifier::BOLD),
)]),
Line::from(msg.content.as_str()),
Line::from(""),
];
ListItem::new(content)
})
.collect();
let list = List::new(items).block(Block::default().title("Conversation").borders(Borders::ALL));
f.render_widget(list, area);
}

View File

@ -0,0 +1,42 @@
use ratatui::{
layout::Rect,
style::{Color, Modifier, Style},
widgets::{Block, Borders, Clear, List, ListItem},
Frame,
};
pub fn render(f: &mut Frame, area: Rect) {
f.render_widget(Clear, area);
let help_text = vec![
ListItem::new("Commands:"),
ListItem::new(" /new [title] - Create new session"),
ListItem::new(" /sessions - List all sessions"),
ListItem::new(" /use <id> - Load session"),
ListItem::new(" /rename <title> - Rename session"),
ListItem::new(" /archive - Archive session"),
ListItem::new(" /delete - Delete session"),
ListItem::new(" /clear - Clear history"),
ListItem::new(" /help, /? - Show this help"),
ListItem::new(" /quit, /q - Quit"),
ListItem::new(""),
ListItem::new("Keyboard:"),
ListItem::new(" Enter - Send message"),
ListItem::new(" Esc, q - Quit"),
ListItem::new(" ? - Show help"),
ListItem::new(" Arrow keys - Navigate"),
];
let list = List::new(help_text).block(
Block::default()
.title("Help")
.title_style(
Style::default()
.fg(Color::Cyan)
.add_modifier(Modifier::BOLD),
)
.borders(Borders::ALL),
);
f.render_widget(list, area);
}

View File

@ -0,0 +1,21 @@
use crate::client::tui::app::App;
use ratatui::{
layout::Rect,
style::{Color, Style},
widgets::{Block, Borders, Paragraph},
Frame,
};
pub fn render(f: &mut Frame, area: Rect, app: &App) {
let input = Paragraph::new(app.input.as_str())
.style(Style::default().fg(Color::White))
.block(Block::default().title("Input").borders(Borders::ALL));
f.render_widget(input, area);
let cursor_x = area.x + 1 + app.input_cursor_pos as u16;
let cursor_y = area.y + 1;
if cursor_x < area.right() && cursor_y < area.bottom() {
f.set_cursor(cursor_x, cursor_y);
}
}

View File

@ -0,0 +1,5 @@
pub mod chat_history;
pub mod help_popup;
pub mod input_area;
pub mod session_list;
pub mod title_bar;

View File

@ -0,0 +1,45 @@
use crate::client::tui::app::App;
use ratatui::{
layout::Rect,
style::{Color, Modifier, Style},
widgets::{Block, Borders, List, ListItem},
Frame,
};
pub fn render(f: &mut Frame, area: Rect, app: &App) {
let items: Vec<ListItem> = app
.sessions
.iter()
.map(|session| {
let is_current = app
.current_session_id
.as_ref()
.map_or(false, |id| id == &session.session_id);
let archived = session.archived_at.is_some();
let mut content = if is_current {
format!("{}", session.title)
} else {
format!(" {}", session.title)
};
if archived {
content.push_str(" [archived]");
}
let style = if is_current {
Style::default()
.fg(Color::Yellow)
.add_modifier(Modifier::BOLD)
} else {
Style::default().fg(Color::White)
};
ListItem::new(content).style(style)
})
.collect();
let list = List::new(items).block(Block::default().title("Sessions").borders(Borders::ALL));
f.render_widget(list, area);
}

View File

@ -0,0 +1,25 @@
use crate::client::tui::app::App;
use ratatui::{
layout::Rect,
style::{Color, Modifier, Style},
widgets::{Block, Borders, Paragraph},
Frame,
};
pub fn render(f: &mut Frame, area: Rect, app: &App) {
let title = if let Some(session_id) = &app.current_session_id {
format!("PicoBot | Session: {}", session_id)
} else {
"PicoBot".to_string()
};
let paragraph = Paragraph::new(title)
.style(
Style::default()
.fg(Color::Cyan)
.add_modifier(Modifier::BOLD),
)
.block(Block::default().borders(Borders::ALL));
f.render_widget(paragraph, area);
}

196
src/client/tui/event.rs Normal file
View File

@ -0,0 +1,196 @@
use crate::client::tui::app::{App, MessageRole};
use crate::protocol::serialize_inbound;
use crate::protocol::WsInbound;
use crossterm::event::{KeyCode, KeyEvent};
use futures_util::SinkExt;
pub async fn handle_key_event(app: &mut App, key: KeyEvent) {
if app.show_help {
match key.code {
KeyCode::Esc | KeyCode::Char('q') => {
app.toggle_help();
}
_ => {}
}
return;
}
match key.code {
KeyCode::Esc | KeyCode::Char('q') => {
app.quit();
}
KeyCode::Char('?') => {
app.toggle_help();
}
KeyCode::Char(c) => {
app.input_insert_char(c);
}
KeyCode::Backspace => {
app.input_delete_char();
}
KeyCode::Left => {
app.input_move_cursor_left();
}
KeyCode::Right => {
app.input_move_cursor_right();
}
KeyCode::Home => {
app.input_move_cursor_to_start();
}
KeyCode::End => {
app.input_move_cursor_to_end();
}
KeyCode::Up => {
app.scroll_chat_up();
}
KeyCode::Down => {
app.scroll_chat_down();
}
KeyCode::Enter => {
let input = app.take_input();
if !input.is_empty() {
process_input(app, input).await;
}
}
_ => {}
}
}
async fn process_input(app: &mut App, input: String) {
let trimmed = input.trim();
if let Some(cmd) = parse_command(trimmed) {
match cmd {
InputCommand::Quit => {
app.quit();
}
InputCommand::Help => {
app.toggle_help();
}
InputCommand::Clear => {
if let Some(session_id) = &app.current_session_id {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::ClearHistory {
chat_id: None,
session_id: Some(session_id.clone()),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
}
InputCommand::New(title) => {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::CreateSession { title };
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
InputCommand::Sessions => {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::ListSessions {
include_archived: true,
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
InputCommand::Use(session_id) => {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::LoadSession { session_id };
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
InputCommand::Rename(title) => {
if let Some(session_id) = &app.current_session_id {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::RenameSession {
session_id: Some(session_id.clone()),
title,
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
}
InputCommand::Archive => {
if let Some(session_id) = &app.current_session_id {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::ArchiveSession {
session_id: Some(session_id.clone()),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
}
InputCommand::Delete => {
if let Some(session_id) = &app.current_session_id {
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::DeleteSession {
session_id: Some(session_id.clone()),
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
}
}
} else {
app.add_message(MessageRole::User, input.clone());
if let Some(sender) = &mut app.ws_sender {
let inbound = WsInbound::UserInput {
content: input,
chat_id: app.current_session_id.clone(),
channel: None,
sender_id: None,
};
if let Ok(text) = serialize_inbound(&inbound) {
let _ = sender.send(tokio_tungstenite::tungstenite::Message::Text(text.into())).await;
}
}
}
}
enum InputCommand {
Quit,
Help,
Clear,
New(Option<String>),
Sessions,
Use(String),
Rename(String),
Archive,
Delete,
}
fn parse_command(input: &str) -> Option<InputCommand> {
if !input.starts_with('/') {
return None;
}
let parts: Vec<&str> = input.splitn(2, char::is_whitespace).collect();
let cmd = parts[0];
let arg = parts.get(1).map(|s| s.trim().to_string()).filter(|s| !s.is_empty());
match cmd {
"/quit" | "/exit" | "/q" => Some(InputCommand::Quit),
"/help" | "/?" => Some(InputCommand::Help),
"/clear" | "/reset" => Some(InputCommand::Clear),
"/new" => Some(InputCommand::New(arg)),
"/sessions" => Some(InputCommand::Sessions),
"/use" => arg.map(InputCommand::Use),
"/rename" => arg.map(InputCommand::Rename),
"/archive" => Some(InputCommand::Archive),
"/delete" => Some(InputCommand::Delete),
_ => None,
}
}

View File

@ -0,0 +1,3 @@
pub fn render_markdown(content: &str) -> String {
content.to_string()
}

5
src/client/tui/mod.rs Normal file
View File

@ -0,0 +1,5 @@
pub mod app;
pub mod components;
pub mod event;
pub mod markdown;
pub mod ui;

57
src/client/tui/ui.rs Normal file
View File

@ -0,0 +1,57 @@
use crate::client::tui::app::App;
use ratatui::{
layout::{Constraint, Direction, Layout, Rect},
widgets::{Block, Borders, Clear, List, ListItem, Paragraph, Wrap},
Frame,
};
use super::components::*;
pub fn render_ui(f: &mut Frame, app: &App) {
let size = f.size();
let chunks = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Length(3),
Constraint::Min(0),
Constraint::Length(5),
])
.split(size);
title_bar::render(f, chunks[0], app);
let middle_chunks = Layout::default()
.direction(Direction::Horizontal)
.constraints([Constraint::Percentage(25), Constraint::Percentage(75)])
.split(chunks[1]);
session_list::render(f, middle_chunks[0], app);
chat_history::render(f, middle_chunks[1], app);
input_area::render(f, chunks[2], app);
if app.show_help {
let area = centered_rect(60, 60, size);
help_popup::render(f, area);
}
}
fn centered_rect(percent_x: u16, percent_y: u16, r: Rect) -> Rect {
let popup_layout = Layout::default()
.direction(Direction::Vertical)
.constraints([
Constraint::Percentage((100 - percent_y) / 2),
Constraint::Percentage(percent_y),
Constraint::Percentage((100 - percent_y) / 2),
])
.split(r);
Layout::default()
.direction(Direction::Horizontal)
.constraints([
Constraint::Percentage((100 - percent_x) / 2),
Constraint::Percentage(percent_x),
Constraint::Percentage((100 - percent_x) / 2),
])
.split(popup_layout[1])[1]
}

View File

@ -1,18 +1,16 @@
pub mod http;
pub mod session;
pub mod ws;
use std::sync::Arc;
use axum::{routing, Router};
use tokio::net::TcpListener;
use crate::bus::{ControlInbound, ControlMessage, ControlOutbound, OutboundDispatcher};
use crate::bus::{ControlMessage, OutboundDispatcher};
use crate::channels::{ChannelManager, CliChatChannel};
use crate::channels::base::{Channel, ChannelError};
use crate::config::Config;
use crate::logging;
use crate::protocol::SessionSummary;
use session::SessionManager;
use crate::session::SessionManager;
pub struct GatewayState {
pub config: Config,
@ -74,27 +72,11 @@ impl GatewayState {
tokio::select! {
// Inbound: AI message flow
inbound = bus.consume_inbound() => {
#[cfg(debug_assertions)]
{
tracing::debug!(
channel = %inbound.channel,
chat_id = %inbound.chat_id,
sender = %inbound.sender_id,
content = %inbound.content,
media_count = %inbound.media.len(),
"Processing inbound message"
);
if !inbound.media.is_empty() {
for (i, m) in inbound.media.iter().enumerate() {
tracing::debug!(media_index = i, media_type = %m.media_type, path = %m.path, "Media item");
}
}
}
match session_manager.handle_message(
&inbound.channel,
&inbound.sender_id,
&inbound.chat_id,
inbound.dialog_id.as_deref(),
&inbound.content,
inbound.media,
).await {
@ -139,59 +121,62 @@ impl GatewayState {
session_manager: &SessionManager,
msg: ControlMessage,
) {
use crate::session::{SessionCommand::*, SessionEvent};
let reply_tx = msg.reply_tx;
let result = match msg.op {
ControlInbound::CreateSession { title } => {
session_manager.create_cli_session(title.as_deref())
.map(|record| ControlOutbound::SessionCreated {
session_id: record.id,
title: record.title,
})
let result: Result<SessionEvent, ChannelError> = match msg.op {
CreateDialog { channel, chat_id, title } => {
session_manager.create_dialog(&channel, &chat_id, title.as_deref()).await
.map(|(session_id, title)| SessionEvent::DialogCreated { session_id, title })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::ListSessions { include_archived } => {
session_manager.list_cli_sessions(include_archived)
.map(|records| ControlOutbound::SessionList {
sessions: records.into_iter().map(|r| SessionSummary {
session_id: r.id,
title: r.title,
channel_name: r.channel_name,
chat_id: r.chat_id,
message_count: r.message_count,
last_active_at: r.last_active_at,
archived_at: r.archived_at,
}).collect()
})
ListDialogs { channel, chat_id, include_archived } => {
session_manager.list_dialogs(&channel, &chat_id, include_archived).await
.map(|(dialogs, current_dialog_id)| SessionEvent::DialogList { dialogs, current_dialog_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::LoadSession { session_id } => {
session_manager.get_session_record(&session_id)
.map(|opt| opt.map(|r| ControlOutbound::SessionLoaded {
session_id: r.id,
title: r.title,
message_count: r.message_count,
}).unwrap_or_else(|| ControlOutbound::Error {
code: "SESSION_NOT_FOUND".to_string(),
message: format!("Session not found: {}", session_id),
}))
GetCurrentDialog { channel, chat_id } => {
session_manager.get_current_dialog(&channel, &chat_id).await
.map(|session_id| SessionEvent::CurrentDialog { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::RenameSession { session_id, title } => {
session_manager.rename_session(&session_id, &title)
.map(|()| ControlOutbound::SessionRenamed { session_id, title })
SwitchDialog { channel, chat_id, dialog_id } => {
session_manager.switch_dialog(&channel, &chat_id, &dialog_id).await
.map(|session_id| SessionEvent::DialogSwitched { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::ArchiveSession { session_id } => {
session_manager.archive_session(&session_id)
.map(|()| ControlOutbound::SessionArchived { session_id })
RenameDialog { session_id, title } => {
session_manager.rename_dialog(&session_id, &title)
.map(|()| SessionEvent::DialogRenamed { session_id, title })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::DeleteSession { session_id } => {
session_manager.delete_session(&session_id)
.map(|()| ControlOutbound::SessionDeleted { session_id })
ArchiveDialog { session_id } => {
session_manager.archive_dialog(&session_id)
.map(|()| SessionEvent::DialogArchived { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ControlInbound::ClearHistory { session_id } => {
session_manager.clear_session_messages(&session_id)
.map(|()| ControlOutbound::HistoryCleared { session_id })
DeleteDialog { session_id } => {
session_manager.delete_dialog(&session_id)
.map(|()| SessionEvent::DialogDeleted { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
ClearHistory { session_id } => {
session_manager.clear_dialog_history(&session_id)
.map(|()| SessionEvent::HistoryCleared { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
GetSlashCommands { channel: _, chat_id: _ } => {
let commands = session_manager.get_slash_commands().to_vec();
Ok(SessionEvent::SlashCommandsList { commands })
}
ExecuteSlashCommand { command, channel, chat_id, current_session_id } => {
session_manager.execute_slash_command(&command, &channel, &chat_id, current_session_id.as_ref())
.await
.map(|(new_id, msg)| SessionEvent::SlashCommandExecuted { new_session_id: new_id, message: msg })
.map_err(|e| ChannelError::Other(e.to_string()))
}
};
let result = result.map_err(|e| ChannelError::Other(e.to_string()));
let _ = reply_tx.send(result).await;
}
}

View File

@ -1,546 +0,0 @@
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, mpsc};
use uuid::Uuid;
use crate::bus::ChatMessage;
use crate::config::LLMProviderConfig;
use crate::agent::{AgentLoop, AgentError, ContextCompressor};
use crate::protocol::WsOutbound;
use crate::providers::{create_provider, LLMProvider};
use crate::storage::{SessionRecord, SessionStore, persistent_session_id};
use crate::tools::{
BashTool, CalculatorTool, FileEditTool, FileReadTool, FileWriteTool,
HttpRequestTool, ToolRegistry, WebFetchTool,
};
/// Session 按 channel 隔离,每个 channel 一个 Session
/// History 按 chat_id 隔离,由 Session 统一管理
pub struct Session {
pub id: Uuid,
pub channel_name: String,
/// 按 chat_id 路由到不同会话历史,支持多用户多会话
chat_histories: HashMap<String, Vec<ChatMessage>>,
pub user_tx: mpsc::Sender<WsOutbound>,
provider_config: LLMProviderConfig,
provider: Arc<dyn LLMProvider>,
tools: Arc<ToolRegistry>,
compressor: ContextCompressor,
store: Arc<SessionStore>,
}
impl Session {
pub async fn new(
channel_name: String,
provider_config: LLMProviderConfig,
user_tx: mpsc::Sender<WsOutbound>,
tools: Arc<ToolRegistry>,
store: Arc<SessionStore>,
) -> Result<Self, AgentError> {
let provider_box = create_provider(provider_config.clone())
.map_err(|e| AgentError::Other(format!("provider creation error: {}", e)))?;
let provider: Arc<dyn LLMProvider> = Arc::from(provider_box);
Ok(Self {
id: Uuid::new_v4(),
channel_name,
chat_histories: HashMap::new(),
user_tx,
provider_config: provider_config.clone(),
provider: provider.clone(),
tools,
compressor: ContextCompressor::new(provider.clone(), provider_config.token_limit),
store,
})
}
pub fn persistent_session_id(&self, chat_id: &str) -> String {
persistent_session_id(&self.channel_name, chat_id)
}
pub fn ensure_persistent_session(&self, chat_id: &str) -> Result<SessionRecord, AgentError> {
self.store
.ensure_channel_session(&self.channel_name, chat_id)
.map_err(|err| AgentError::Other(format!("session persistence error: {}", err)))
}
pub fn ensure_chat_loaded(&mut self, chat_id: &str) -> Result<(), AgentError> {
if self.chat_histories.contains_key(chat_id) {
return Ok(());
}
let history = self
.store
.load_messages(&self.persistent_session_id(chat_id))
.map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?;
self.chat_histories.insert(chat_id.to_string(), history);
Ok(())
}
/// 获取或创建指定 chat_id 的会话历史
pub fn get_or_create_history(&mut self, chat_id: &str) -> &mut Vec<ChatMessage> {
self.chat_histories
.entry(chat_id.to_string())
.or_insert_with(Vec::new)
}
/// 获取指定 chat_id 的会话历史(不创建)
pub fn get_history(&self, chat_id: &str) -> Option<&Vec<ChatMessage>> {
self.chat_histories.get(chat_id)
}
/// 使用完整消息追加到历史
pub fn add_message(&mut self, chat_id: &str, message: ChatMessage) {
let history = self.get_or_create_history(chat_id);
history.push(message);
}
pub fn remove_history(&mut self, chat_id: &str) {
self.chat_histories.remove(chat_id);
}
pub fn clear_chat_history(&mut self, chat_id: &str) -> Result<(), AgentError> {
if let Some(history) = self.chat_histories.get_mut(chat_id) {
let len = history.len();
history.clear();
#[cfg(debug_assertions)]
tracing::debug!(chat_id = %chat_id, previous_len = len, "Chat history cleared");
}
self.store
.clear_messages(&self.persistent_session_id(chat_id))
.map_err(|err| AgentError::Other(format!("clear history persistence error: {}", err)))
}
pub fn reset_chat_context(&mut self, chat_id: &str) -> Result<(), AgentError> {
if let Some(history) = self.chat_histories.get_mut(chat_id) {
let len = history.len();
history.clear();
#[cfg(debug_assertions)]
tracing::debug!(chat_id = %chat_id, previous_len = len, "Chat history reset in memory");
}
self.store
.reset_session(&self.persistent_session_id(chat_id))
.map_err(|err| AgentError::Other(format!("reset history persistence error: {}", err)))
}
/// 将消息写入内存与持久化层
pub fn append_persisted_message(&mut self, chat_id: &str, message: ChatMessage) -> Result<(), AgentError> {
let session_id = self.persistent_session_id(chat_id);
self.store
.append_message(&session_id, &message)
.map_err(|err| AgentError::Other(format!("append message persistence error: {}", err)))?;
self.add_message(chat_id, message);
Ok(())
}
pub fn append_persisted_messages<I>(&mut self, chat_id: &str, messages: I) -> Result<(), AgentError>
where
I: IntoIterator<Item = ChatMessage>,
{
for message in messages {
self.append_persisted_message(chat_id, message)?;
}
Ok(())
}
pub fn create_user_message(&self, content: &str, media_refs: Vec<String>) -> ChatMessage {
if media_refs.is_empty() {
ChatMessage::user(content)
} else {
ChatMessage::user_with_media(content, media_refs)
}
}
/// 清除所有历史
pub fn clear_all_history(&mut self) -> Result<(), AgentError> {
let chat_ids: Vec<String> = self.chat_histories.keys().cloned().collect();
let total: usize = self.chat_histories.values().map(|h| h.len()).sum();
self.chat_histories.clear();
#[cfg(debug_assertions)]
tracing::debug!(previous_total = total, "All chat histories cleared");
for chat_id in chat_ids {
self.store
.clear_messages(&self.persistent_session_id(&chat_id))
.map_err(|err| AgentError::Other(format!("clear history persistence error: {}", err)))?;
}
Ok(())
}
pub async fn send(&self, msg: WsOutbound) {
let _ = self.user_tx.send(msg).await;
}
/// 获取 provider_config 引用
pub fn provider_config(&self) -> &LLMProviderConfig {
&self.provider_config
}
/// 获取 compressor 引用
pub fn compressor(&self) -> &ContextCompressor {
&self.compressor
}
/// 创建一个临时的 AgentLoop 实例来处理消息
pub fn create_agent(&self) -> Result<AgentLoop, AgentError> {
Ok(AgentLoop::with_provider_and_tools(
self.provider.clone(),
self.tools.clone(),
self.provider_config.max_tool_iterations,
))
}
}
/// SessionManager 管理所有 Session按 channel_name 路由
#[derive(Clone)]
pub struct SessionManager {
inner: Arc<Mutex<SessionManagerInner>>,
provider_config: LLMProviderConfig,
tools: Arc<ToolRegistry>,
store: Arc<SessionStore>,
}
struct SessionManagerInner {
sessions: HashMap<String, Arc<Mutex<Session>>>,
session_timestamps: HashMap<String, Instant>,
session_ttl: Duration,
}
fn default_tools() -> ToolRegistry {
let mut registry = ToolRegistry::new();
registry.register(CalculatorTool::new());
registry.register(FileReadTool::new());
registry.register(FileWriteTool::new());
registry.register(FileEditTool::new());
registry.register(BashTool::new());
registry.register(HttpRequestTool::new(
vec!["*".to_string()], // 允许所有域名,实际使用时建议限制
1_000_000, // max_response_size
30, // timeout_secs
false, // allow_private_hosts
));
registry.register(WebFetchTool::new(50_000, 30)); // max_chars, timeout_secs
registry
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum InChatCommand {
FreshConversation,
}
fn parse_in_chat_command(content: &str) -> Option<InChatCommand> {
match content.trim() {
"/new" | "/reset" => Some(InChatCommand::FreshConversation),
_ => None,
}
}
pub(crate) fn handle_in_chat_command(
session: &mut Session,
chat_id: &str,
content: &str,
) -> Result<Option<String>, AgentError> {
match parse_in_chat_command(content) {
Some(InChatCommand::FreshConversation) => {
session.reset_chat_context(chat_id)?;
Ok(Some("Started a fresh conversation.".to_string()))
}
None => Ok(None),
}
}
impl SessionManager {
pub fn new(session_ttl_hours: u64, provider_config: LLMProviderConfig) -> Result<Self, AgentError> {
let store = Arc::new(
SessionStore::new()
.map_err(|err| AgentError::Other(format!("session store init error: {}", err)))?,
);
Ok(Self {
inner: Arc::new(Mutex::new(SessionManagerInner {
sessions: HashMap::new(),
session_timestamps: HashMap::new(),
session_ttl: Duration::from_secs(session_ttl_hours * 3600),
})),
provider_config,
tools: Arc::new(default_tools()),
store,
})
}
pub fn tools(&self) -> Arc<ToolRegistry> {
self.tools.clone()
}
pub fn store(&self) -> Arc<SessionStore> {
self.store.clone()
}
pub fn create_cli_session(&self, title: Option<&str>) -> Result<SessionRecord, AgentError> {
self.store
.create_cli_session(title)
.map_err(|err| AgentError::Other(format!("create session error: {}", err)))
}
pub fn get_session_record(&self, session_id: &str) -> Result<Option<SessionRecord>, AgentError> {
self.store
.get_session(session_id)
.map_err(|err| AgentError::Other(format!("get session error: {}", err)))
}
pub fn list_cli_sessions(&self, include_archived: bool) -> Result<Vec<SessionRecord>, AgentError> {
self.store
.list_sessions("cli", include_archived)
.map_err(|err| AgentError::Other(format!("list sessions error: {}", err)))
}
pub fn rename_session(&self, session_id: &str, title: &str) -> Result<(), AgentError> {
self.store
.rename_session(session_id, title)
.map_err(|err| AgentError::Other(format!("rename session error: {}", err)))
}
pub fn archive_session(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.archive_session(session_id)
.map_err(|err| AgentError::Other(format!("archive session error: {}", err)))
}
pub fn delete_session(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.delete_session(session_id)
.map_err(|err| AgentError::Other(format!("delete session error: {}", err)))
}
pub fn clear_session_messages(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.clear_messages(session_id)
.map_err(|err| AgentError::Other(format!("clear session error: {}", err)))
}
pub fn load_session_messages(&self, session_id: &str) -> Result<Vec<ChatMessage>, AgentError> {
self.store
.load_messages(session_id)
.map_err(|err| AgentError::Other(format!("load messages error: {}", err)))
}
/// 确保 session 存在且未超时,超时则重建
pub async fn ensure_session(&self, channel_name: &str) -> Result<(), AgentError> {
let mut inner = self.inner.lock().await;
let should_recreate = if let Some(last_active) = inner.session_timestamps.get(channel_name) {
let elapsed = last_active.elapsed();
if elapsed > inner.session_ttl {
tracing::info!(channel = %channel_name, elapsed_hours = elapsed.as_secs() / 3600, "Session expired, recreating");
true
} else {
false
}
} else {
#[cfg(debug_assertions)]
tracing::debug!(channel = %channel_name, "Creating new session");
true
};
if should_recreate {
// 移除旧 session
inner.sessions.remove(channel_name);
// 创建新 session使用临时 user_tx因为 Feishu 不通过 WS
let (user_tx, _rx) = mpsc::channel::<WsOutbound>(100);
let session = Session::new(
channel_name.to_string(),
self.provider_config.clone(),
user_tx,
self.tools.clone(),
self.store.clone(),
)
.await?;
let arc = Arc::new(Mutex::new(session));
inner.sessions.insert(channel_name.to_string(), arc.clone());
inner.session_timestamps.insert(channel_name.to_string(), Instant::now());
}
Ok(())
}
/// 获取 session不检查超时
pub async fn get(&self, channel_name: &str) -> Option<Arc<Mutex<Session>>> {
let inner = self.inner.lock().await;
inner.sessions.get(channel_name).cloned()
}
/// 更新最后活跃时间
pub async fn touch(&self, channel_name: &str) {
let mut inner = self.inner.lock().await;
inner.session_timestamps.insert(channel_name.to_string(), Instant::now());
}
/// 处理消息:路由到对应 session 的 agent
pub async fn handle_message(
&self,
channel_name: &str,
_sender_id: &str,
chat_id: &str,
content: &str,
media: Vec<crate::bus::MediaItem>,
) -> Result<String, AgentError> {
#[cfg(debug_assertions)]
{
tracing::debug!(
channel = %channel_name,
chat_id = %chat_id,
content_len = content.len(),
media_count = %media.len(),
"Routing message to agent"
);
for (i, m) in media.iter().enumerate() {
tracing::debug!(media_index = i, media_type = %m.media_type, path = %m.path, "Media in handle_message");
}
}
// 确保 session 存在(可能需要重建)
self.ensure_session(channel_name).await?;
// 更新活跃时间
self.touch(channel_name).await;
// 获取 session
let session = self
.get(channel_name)
.await
.ok_or_else(|| AgentError::Other("Session not found".to_string()))?;
// 处理消息
let response = {
let mut session_guard = session.lock().await;
session_guard.ensure_persistent_session(chat_id)?;
session_guard.ensure_chat_loaded(chat_id)?;
if let Some(command_response) = handle_in_chat_command(&mut session_guard, chat_id, content)? {
return Ok(command_response);
}
// 添加用户消息到历史
let media_refs: Vec<String> = media.iter().map(|m| m.path.clone()).collect();
#[cfg(debug_assertions)]
if !media_refs.is_empty() {
tracing::debug!(media_count = %media.len(), media_refs = ?media_refs, "Adding user message with media");
}
let user_message = session_guard.create_user_message(content, media_refs);
session_guard.append_persisted_message(chat_id, user_message)?;
// 获取完整历史
let history = session_guard.get_or_create_history(chat_id).clone();
// 压缩历史(如果需要)
let history = session_guard.compressor
.compress_if_needed(history)
.await?;
// 创建 agent 并处理
let agent = session_guard.create_agent()?;
let result = agent.process(history).await?;
// 按真实顺序持久化 assistant tool_calls、tool 结果和最终 assistant 回复
session_guard.append_persisted_messages(chat_id, result.emitted_messages.clone())?;
result.final_response
};
#[cfg(debug_assertions)]
tracing::debug!(
channel = %channel_name,
chat_id = %chat_id,
response_len = response.content.len(),
"Agent response received"
);
Ok(response.content)
}
/// 清除指定 session 的所有历史
pub async fn clear_session_history(&self, channel_name: &str) -> Result<(), AgentError> {
if let Some(session) = self.get(channel_name).await {
let mut session_guard = session.lock().await;
session_guard.clear_all_history()?;
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
use tokio::sync::mpsc;
fn test_provider_config() -> LLMProviderConfig {
LLMProviderConfig {
provider_type: "openai".to_string(),
name: "test".to_string(),
base_url: "http://localhost".to_string(),
api_key: "test-key".to_string(),
extra_headers: HashMap::new(),
model_id: "test-model".to_string(),
temperature: Some(0.0),
max_tokens: Some(32),
model_extra: HashMap::new(),
max_tool_iterations: 1,
token_limit: 4096,
}
}
#[test]
fn test_parse_in_chat_command_aliases() {
assert_eq!(parse_in_chat_command("/new"), Some(InChatCommand::FreshConversation));
assert_eq!(parse_in_chat_command(" /reset \n"), Some(InChatCommand::FreshConversation));
assert_eq!(parse_in_chat_command("/new planning"), None);
assert_eq!(parse_in_chat_command("please /reset"), None);
}
#[tokio::test]
async fn test_handle_in_chat_command_resets_active_history_only() {
let store = Arc::new(SessionStore::in_memory().unwrap());
let (user_tx, _user_rx) = mpsc::channel(4);
let tools = Arc::new(default_tools());
let mut session = Session::new(
"feishu".to_string(),
test_provider_config(),
user_tx,
tools,
store.clone(),
)
.await
.unwrap();
session.ensure_persistent_session("chat-1").unwrap();
session.ensure_chat_loaded("chat-1").unwrap();
session
.append_persisted_message("chat-1", ChatMessage::user("hello"))
.unwrap();
let response = handle_in_chat_command(&mut session, "chat-1", "/reset")
.unwrap()
.unwrap();
assert_eq!(response, "Started a fresh conversation.");
assert!(session.get_history("chat-1").unwrap().is_empty());
assert!(store
.load_messages(&session.persistent_session_id("chat-1"))
.unwrap()
.is_empty());
assert_eq!(
store
.load_all_messages(&session.persistent_session_id("chat-1"))
.unwrap()
.len(),
1,
);
}
}

View File

@ -3,6 +3,7 @@ pub mod providers;
pub mod bus;
pub mod agent;
pub mod gateway;
pub mod session;
pub mod client;
pub mod protocol;
pub mod channels;

View File

@ -99,6 +99,8 @@ pub enum WsOutbound {
HistoryCleared { session_id: String },
#[serde(rename = "pong")]
Pong,
#[serde(rename = "command_executed")]
CommandExecuted { message: String },
}
pub fn parse_inbound(raw: &str) -> Result<WsInbound, serde_json::Error> {

78
src/session/commands.rs Normal file
View File

@ -0,0 +1,78 @@
use super::session_id::UnifiedSessionId;
/// Session management commands issued by Channel to SessionManager
#[derive(Debug, Clone)]
pub enum SessionCommand {
/// Create a new dialog in the given chat
CreateDialog {
channel: String,
chat_id: String,
title: Option<String>,
},
/// List all dialogs in a chat
ListDialogs {
channel: String,
chat_id: String,
include_archived: bool,
},
/// Switch to a specific dialog (set as current)
SwitchDialog {
channel: String,
chat_id: String,
dialog_id: String,
},
/// Get the current dialog for a chat
GetCurrentDialog {
channel: String,
chat_id: String,
},
/// Rename a dialog
RenameDialog {
session_id: UnifiedSessionId,
title: String,
},
/// Archive a dialog
ArchiveDialog {
session_id: UnifiedSessionId,
},
/// Delete a dialog
DeleteDialog {
session_id: UnifiedSessionId,
},
/// Clear dialog history
ClearHistory {
session_id: UnifiedSessionId,
},
/// Get list of available slash commands
GetSlashCommands {
channel: String,
chat_id: String,
},
/// Execute a slash command
ExecuteSlashCommand {
command: String,
channel: String,
chat_id: String,
current_session_id: Option<UnifiedSessionId>,
},
}
impl SessionCommand {
/// Create a CreateDialog command
pub fn create_dialog(channel: impl Into<String>, chat_id: impl Into<String>, title: Option<String>) -> Self {
Self::CreateDialog {
channel: channel.into(),
chat_id: chat_id.into(),
title,
}
}
/// Create a ListDialogs command
pub fn list_dialogs(channel: impl Into<String>, chat_id: impl Into<String>, include_archived: bool) -> Self {
Self::ListDialogs {
channel: channel.into(),
chat_id: chat_id.into(),
include_archived,
}
}
}

22
src/session/error.rs Normal file
View File

@ -0,0 +1,22 @@
use thiserror::Error;
#[derive(Debug, Error)]
pub enum SessionError {
#[error("session not found: {0}")]
NotFound(String),
#[error("session error: {0}")]
Other(String),
}
impl From<crate::channels::base::ChannelError> for SessionError {
fn from(e: crate::channels::base::ChannelError) -> Self {
SessionError::Other(e.to_string())
}
}
impl From<crate::storage::StorageError> for SessionError {
fn from(e: crate::storage::StorageError) -> Self {
SessionError::Other(e.to_string())
}
}

67
src/session/events.rs Normal file
View File

@ -0,0 +1,67 @@
use super::session_id::UnifiedSessionId;
use super::session::SlashCommand;
/// Dialog information returned by SessionManager
#[derive(Debug, Clone)]
pub struct DialogInfo {
pub session_id: UnifiedSessionId,
pub title: String,
pub created_at: i64,
pub last_active_at: i64,
pub message_count: i64,
pub archived_at: Option<i64>,
}
/// Session events emitted by SessionManager to Channel
#[derive(Debug, Clone)]
pub enum SessionEvent {
/// A new dialog was created
DialogCreated {
session_id: UnifiedSessionId,
title: String,
},
/// List of dialogs returned
DialogList {
dialogs: Vec<DialogInfo>,
current_dialog_id: Option<String>,
},
/// Current dialog info returned
CurrentDialog {
session_id: Option<UnifiedSessionId>,
},
/// Dialog switched successfully
DialogSwitched {
session_id: UnifiedSessionId,
},
/// Dialog renamed
DialogRenamed {
session_id: UnifiedSessionId,
title: String,
},
/// Dialog archived
DialogArchived {
session_id: UnifiedSessionId,
},
/// Dialog deleted
DialogDeleted {
session_id: UnifiedSessionId,
},
/// Dialog history cleared
HistoryCleared {
session_id: UnifiedSessionId,
},
/// List of available slash commands
SlashCommandsList {
commands: Vec<SlashCommand>,
},
/// Slash command executed successfully
SlashCommandExecuted {
new_session_id: Option<UnifiedSessionId>,
message: String,
},
/// Error occurred
Error {
code: String,
message: String,
},
}

11
src/session/mod.rs Normal file
View File

@ -0,0 +1,11 @@
pub mod error;
pub mod commands;
pub mod events;
pub mod session;
pub mod session_id;
pub use error::SessionError;
pub use commands::SessionCommand;
pub use events::{SessionEvent, DialogInfo};
pub use session::{Session, SessionManager, SlashCommand, SLASH_COMMANDS};
pub use session_id::UnifiedSessionId;

669
src/session/session.rs Normal file
View File

@ -0,0 +1,669 @@
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, mpsc};
use uuid::Uuid;
use crate::bus::ChatMessage;
use crate::config::LLMProviderConfig;
use crate::agent::{AgentLoop, AgentError, ContextCompressor};
use crate::protocol::WsOutbound;
use crate::providers::{create_provider, LLMProvider};
use crate::session::session_id::{UnifiedSessionId, DEFAULT_DIALOG_ID};
use crate::session::events::DialogInfo;
use crate::storage::{SessionRecord, SessionStore};
use crate::tools::{
BashTool, CalculatorTool, FileEditTool, FileReadTool, FileWriteTool,
HttpRequestTool, ToolRegistry, WebFetchTool,
};
/// Generate a short ID (8 characters) from a UUID
fn short_id() -> String {
Uuid::new_v4().to_string()[..8].to_string()
}
/// Session = 一个 dialog
/// 每个 Session 对应一个 UnifiedSessionId有独立的 messages history
pub struct Session {
pub id: UnifiedSessionId,
messages: Vec<ChatMessage>,
pub user_tx: mpsc::Sender<WsOutbound>,
provider_config: LLMProviderConfig,
provider: Arc<dyn LLMProvider>,
tools: Arc<ToolRegistry>,
compressor: ContextCompressor,
store: Arc<SessionStore>,
}
impl Session {
pub async fn new(
id: UnifiedSessionId,
provider_config: LLMProviderConfig,
user_tx: mpsc::Sender<WsOutbound>,
tools: Arc<ToolRegistry>,
store: Arc<SessionStore>,
) -> Result<Self, AgentError> {
let provider_box = create_provider(provider_config.clone())
.map_err(|e| AgentError::Other(format!("provider creation error: {}", e)))?;
let provider: Arc<dyn LLMProvider> = Arc::from(provider_box);
Ok(Self {
id,
messages: Vec::new(),
user_tx,
provider_config: provider_config.clone(),
provider: provider.clone(),
tools,
compressor: ContextCompressor::new(provider.clone(), provider_config.token_limit),
store,
})
}
/// 获取持久化 session ID
pub fn persistent_session_id(&self) -> String {
self.id.to_string()
}
/// 确保存储中有此 session
pub fn ensure_persistent_session(&self) -> Result<SessionRecord, AgentError> {
self.store
.ensure_channel_session(&self.id.channel, &self.id.chat_id, &self.id.dialog_id)
.map_err(|err| AgentError::Other(format!("session persistence error: {}", err)))
}
/// 加载历史消息到内存
pub fn load_history(&mut self) -> Result<(), AgentError> {
if !self.messages.is_empty() {
return Ok(());
}
let history = self.store
.load_messages(&self.persistent_session_id())
.map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?;
self.messages = history;
Ok(())
}
/// 添加消息到历史
pub fn add_message(&mut self, message: ChatMessage) {
self.messages.push(message);
}
/// 获取消息历史
pub fn get_history(&self) -> &[ChatMessage] {
&self.messages
}
/// 清除历史消息
pub fn clear_history(&mut self) -> Result<(), AgentError> {
let len = self.messages.len();
self.messages.clear();
#[cfg(debug_assertions)]
tracing::debug!(session_id = %self.id, previous_len = len, "Chat history cleared");
self.store
.clear_messages(&self.persistent_session_id())
.map_err(|err| AgentError::Other(format!("clear history error: {}", err)))
}
/// 重置对话上下文
pub fn reset_context(&mut self) -> Result<(), AgentError> {
let len = self.messages.len();
self.messages.clear();
#[cfg(debug_assertions)]
tracing::debug!(session_id = %self.id, previous_len = len, "Chat context reset in memory");
self.store
.reset_session(&self.persistent_session_id())
.map_err(|err| AgentError::Other(format!("reset context error: {}", err)))
}
/// Archive 此 session
pub fn archive(&self) -> Result<(), AgentError> {
self.store
.archive_session(&self.persistent_session_id())
.map_err(|err| AgentError::Other(format!("archive session error: {}", err)))
}
/// 持久化消息
pub fn append_message(&self, message: &ChatMessage) -> Result<(), AgentError> {
self.store
.append_message(&self.persistent_session_id(), message)
.map_err(|err| AgentError::Other(format!("append message error: {}", err)))
}
/// 持久化多条消息
pub fn append_messages<I>(&self, messages: I) -> Result<(), AgentError>
where
I: IntoIterator<Item = ChatMessage>,
{
for message in messages {
self.append_message(&message)?;
}
Ok(())
}
pub fn create_user_message(&self, content: &str, media_refs: Vec<String>) -> ChatMessage {
if media_refs.is_empty() {
ChatMessage::user(content)
} else {
ChatMessage::user_with_media(content, media_refs)
}
}
pub async fn send(&self, msg: WsOutbound) {
let _ = self.user_tx.send(msg).await;
}
/// 获取 provider_config 引用
pub fn provider_config(&self) -> &LLMProviderConfig {
&self.provider_config
}
/// 获取 compressor 引用
pub fn compressor(&self) -> &ContextCompressor {
&self.compressor
}
/// 创建一个临时的 AgentLoop 实例来处理消息
pub fn create_agent(&self) -> Result<AgentLoop, AgentError> {
Ok(AgentLoop::with_provider_and_tools(
self.provider.clone(),
self.tools.clone(),
self.provider_config.max_tool_iterations,
))
}
}
/// SessionManager 管理所有 Session按 channel_name 路由
#[derive(Clone)]
pub struct SessionManager {
inner: Arc<Mutex<SessionManagerInner>>,
provider_config: LLMProviderConfig,
tools: Arc<ToolRegistry>,
store: Arc<SessionStore>,
}
struct SessionManagerInner {
/// Sessions keyed by UnifiedSessionId.to_string()
sessions: HashMap<String, Arc<Mutex<Session>>>,
session_timestamps: HashMap<String, Instant>,
session_ttl: Duration,
}
fn default_tools() -> ToolRegistry {
let mut registry = ToolRegistry::new();
registry.register(CalculatorTool::new());
registry.register(FileReadTool::new());
registry.register(FileWriteTool::new());
registry.register(FileEditTool::new());
registry.register(BashTool::new());
registry.register(HttpRequestTool::new(
vec!["*".to_string()], // 允许所有域名,实际使用时建议限制
1_000_000, // max_response_size
30, // timeout_secs
false, // allow_private_hosts
));
registry.register(WebFetchTool::new(50_000, 30)); // max_chars, timeout_secs
registry
}
/// 斜杠命令定义
#[derive(Debug, Clone)]
pub struct SlashCommand {
/// 命令名称
pub name: &'static str,
/// 命令描述
pub description: &'static str,
/// 命令别名(触发词)
pub aliases: &'static [&'static str],
}
impl SlashCommand {
/// 检查给定内容是否匹配此命令
pub fn matches(&self, content: &str) -> bool {
let trimmed = content.trim();
self.aliases.iter().any(|&alias| trimmed == alias || trimmed.starts_with(&format!("{} ", alias)))
}
}
/// Session 支持的斜杠命令列表
pub static SLASH_COMMANDS: &[SlashCommand] = &[
SlashCommand {
name: "reset",
description: "Start a fresh conversation (archives current dialog)",
aliases: &["/reset", "/new"],
},
];
impl SessionManager {
pub fn new(session_ttl_hours: u64, provider_config: LLMProviderConfig) -> Result<Self, AgentError> {
let store = Arc::new(
SessionStore::new()
.map_err(|err| AgentError::Other(format!("session store init error: {}", err)))?,
);
Ok(Self {
inner: Arc::new(Mutex::new(SessionManagerInner {
sessions: HashMap::new(),
session_timestamps: HashMap::new(),
session_ttl: Duration::from_secs(session_ttl_hours * 3600),
})),
provider_config,
tools: Arc::new(default_tools()),
store,
})
}
pub fn tools(&self) -> Arc<ToolRegistry> {
self.tools.clone()
}
/// 获取所有可用的斜杠命令
pub fn get_slash_commands(&self) -> &[SlashCommand] {
SLASH_COMMANDS
}
/// 执行斜杠命令
/// 返回 (新session_id, 响应消息)
pub async fn execute_slash_command(
&self,
command: &str,
channel: &str,
chat_id: &str,
current_session_id: Option<&UnifiedSessionId>,
) -> Result<(Option<UnifiedSessionId>, String), AgentError> {
// 查找匹配的 command
let cmd = SLASH_COMMANDS
.iter()
.find(|c| c.name == command)
.ok_or_else(|| AgentError::Other(format!("Unknown command: {}", command)))?;
match cmd.name {
"reset" => {
// Archive current session if exists
if let Some(sid) = current_session_id {
let unified_str = sid.to_string();
self.store
.archive_session(&unified_str)
.map_err(|e| AgentError::Other(format!("archive session error: {}", e)))?;
}
// Create new dialog
let (new_id, _title) = self.create_session(channel, chat_id, None).await?;
Ok((Some(new_id), "Starting a fresh conversation...".to_string()))
}
_ => Err(AgentError::Other(format!("Command not implemented: {}", cmd.name))),
}
}
pub fn store(&self) -> Arc<SessionStore> {
self.store.clone()
}
pub fn create_cli_session(&self, title: Option<&str>) -> Result<SessionRecord, AgentError> {
self.store
.create_cli_session(title)
.map_err(|err| AgentError::Other(format!("create session error: {}", err)))
}
pub fn get_session_record(&self, session_id: &str) -> Result<Option<SessionRecord>, AgentError> {
self.store
.get_session(session_id)
.map_err(|err| AgentError::Other(format!("get session error: {}", err)))
}
pub fn list_cli_sessions(&self, include_archived: bool) -> Result<Vec<SessionRecord>, AgentError> {
self.store
.list_sessions("cli", include_archived)
.map_err(|err| AgentError::Other(format!("list sessions error: {}", err)))
}
pub fn rename_session(&self, session_id: &str, title: &str) -> Result<(), AgentError> {
self.store
.rename_session(session_id, title)
.map_err(|err| AgentError::Other(format!("rename session error: {}", err)))
}
pub fn archive_session(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.archive_session(session_id)
.map_err(|err| AgentError::Other(format!("archive session error: {}", err)))
}
pub fn delete_session(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.delete_session(session_id)
.map_err(|err| AgentError::Other(format!("delete session error: {}", err)))
}
pub fn clear_session_messages(&self, session_id: &str) -> Result<(), AgentError> {
self.store
.clear_messages(session_id)
.map_err(|err| AgentError::Other(format!("clear session error: {}", err)))
}
pub fn load_session_messages(&self, session_id: &str) -> Result<Vec<ChatMessage>, AgentError> {
self.store
.load_messages(session_id)
.map_err(|err| AgentError::Other(format!("load messages error: {}", err)))
}
// =========================================================================
// Dialog management methods (UnifiedSessionId based)
// =========================================================================
/// Create a new session (dialog) and return (session_id, title)
pub async fn create_session(
&self,
channel: &str,
chat_id: &str,
title: Option<&str>,
) -> Result<(UnifiedSessionId, String), AgentError> {
let dialog_id = short_id();
let unified_id = UnifiedSessionId::new(channel, chat_id, &dialog_id);
let session_id_str = unified_id.to_string();
let title = title
.map(str::trim)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned)
.unwrap_or_else(|| format!("Dialog {}", &dialog_id));
// Ensure storage record exists
self.store
.ensure_channel_session(channel, chat_id, &dialog_id)
.map_err(|err| AgentError::Other(format!("create session error: {}", err)))?;
// Create session instance
let (user_tx, _rx) = mpsc::channel::<WsOutbound>(100);
let session = Session::new(
unified_id.clone(),
self.provider_config.clone(),
user_tx,
self.tools.clone(),
self.store.clone(),
)
.await?;
let arc = Arc::new(Mutex::new(session));
let inner = &mut *self.inner.lock().await;
inner.sessions.insert(session_id_str.clone(), arc.clone());
inner.session_timestamps.insert(session_id_str, Instant::now());
Ok((unified_id, title))
}
/// Get or create a session by UnifiedSessionId
pub async fn get_or_create_session(&self, unified_id: &UnifiedSessionId) -> Result<Arc<Mutex<Session>>, AgentError> {
let session_id_str = unified_id.to_string();
let inner = &mut *self.inner.lock().await;
// Check if session exists
if let Some(session) = inner.sessions.get(&session_id_str) {
// Update timestamp
inner.session_timestamps.insert(session_id_str, Instant::now());
return Ok(session.clone());
}
// Check if session exists in storage
if let Ok(Some(_)) = self.store.get_session(&session_id_str) {
// Create session instance from storage
let (user_tx, _rx) = mpsc::channel::<WsOutbound>(100);
let session = Session::new(
unified_id.clone(),
self.provider_config.clone(),
user_tx,
self.tools.clone(),
self.store.clone(),
)
.await?;
let arc = Arc::new(Mutex::new(session));
inner.sessions.insert(session_id_str.clone(), arc.clone());
inner.session_timestamps.insert(session_id_str, Instant::now());
return Ok(arc);
}
// Session doesn't exist - create new directly
let (user_tx, _rx) = mpsc::channel::<WsOutbound>(100);
let session = Session::new(
unified_id.clone(),
self.provider_config.clone(),
user_tx,
self.tools.clone(),
self.store.clone(),
)
.await?;
let arc = Arc::new(Mutex::new(session));
inner.sessions.insert(session_id_str.clone(), arc.clone());
inner.session_timestamps.insert(session_id_str, Instant::now());
Ok(arc)
}
/// List all dialogs for a chat scope (internal)
async fn list_dialogs_for_chat(
&self,
channel: &str,
chat_id: &str,
include_archived: bool,
) -> Result<Vec<DialogInfo>, AgentError> {
let records = self.store
.list_sessions(channel, include_archived)
.map_err(|err| AgentError::Other(format!("list dialogs error: {}", err)))?;
let dialogs: Vec<DialogInfo> = records
.into_iter()
.filter(|r| {
// Filter to only dialogs for this chat_id
if let Some(sid) = UnifiedSessionId::parse(&r.id) {
sid.chat_id == chat_id
} else {
false
}
})
.map(|r| {
let sid = UnifiedSessionId::parse(&r.id).unwrap();
DialogInfo {
session_id: sid,
title: r.title,
created_at: r.created_at,
last_active_at: r.last_active_at,
message_count: r.message_count,
archived_at: r.archived_at,
}
})
.collect();
Ok(dialogs)
}
/// Get the most recent dialog for a chat scope (from storage)
pub async fn get_most_recent_dialog(
&self,
channel: &str,
chat_id: &str,
) -> Result<Option<UnifiedSessionId>, AgentError> {
let records = self.store
.list_sessions(channel, false)
.map_err(|err| AgentError::Other(format!("get recent dialog error: {}", err)))?;
let most_recent = records
.into_iter()
.filter(|r| {
if let Some(sid) = UnifiedSessionId::parse(&r.id) {
sid.chat_id == chat_id
} else {
false
}
})
.max_by_key(|r| r.last_active_at);
Ok(most_recent.map(|r| UnifiedSessionId::parse(&r.id).unwrap()))
}
/// Rename a dialog
pub fn rename_dialog(&self, session_id: &UnifiedSessionId, title: &str) -> Result<(), AgentError> {
self.store
.rename_session(&session_id.to_string(), title)
.map_err(|err| AgentError::Other(format!("rename dialog error: {}", err)))
}
/// Create a new dialog (wrapper for create_session to match gateway interface)
pub async fn create_dialog(
&self,
channel: &str,
chat_id: &str,
title: Option<&str>,
) -> Result<(UnifiedSessionId, String), AgentError> {
self.create_session(channel, chat_id, title).await
}
/// Get current dialog for a chat (wrapper for get_most_recent_dialog)
pub async fn get_current_dialog(
&self,
channel: &str,
chat_id: &str,
) -> Result<Option<UnifiedSessionId>, AgentError> {
self.get_most_recent_dialog(channel, chat_id).await
}
/// Switch to a different dialog - not applicable in new architecture
/// Each Session IS a dialog, so switching is just loading that session
pub async fn switch_dialog(
&self,
_channel: &str,
_chat_id: &str,
_dialog_id: &str,
) -> Result<UnifiedSessionId, AgentError> {
Err(AgentError::Other("switch_dialog not applicable in new architecture".to_string()))
}
/// List all dialogs for a chat scope (returns tuple for gateway compatibility)
pub async fn list_dialogs(
&self,
channel: &str,
chat_id: &str,
include_archived: bool,
) -> Result<(Vec<DialogInfo>, Option<String>), AgentError> {
let dialogs = self.list_dialogs_for_chat(channel, chat_id, include_archived).await?;
let current = self.get_most_recent_dialog(channel, chat_id).await?;
Ok((dialogs, current.map(|id| id.to_string())))
}
/// Archive a dialog
pub fn archive_dialog(&self, session_id: &UnifiedSessionId) -> Result<(), AgentError> {
self.store
.archive_session(&session_id.to_string())
.map_err(|err| AgentError::Other(format!("archive dialog error: {}", err)))
}
/// Delete a dialog
pub fn delete_dialog(&self, session_id: &UnifiedSessionId) -> Result<(), AgentError> {
self.store
.delete_session(&session_id.to_string())
.map_err(|err| AgentError::Other(format!("delete dialog error: {}", err)))
}
/// Clear dialog history
pub fn clear_dialog_history(&self, session_id: &UnifiedSessionId) -> Result<(), AgentError> {
self.store
.clear_messages(&session_id.to_string())
.map_err(|err| AgentError::Other(format!("clear dialog history error: {}", err)))
}
/// 处理消息:路由到对应 session 的 agent
pub async fn handle_message(
&self,
channel: &str,
_sender_id: &str,
chat_id: &str,
dialog_id: Option<&str>,
content: &str,
media: Vec<crate::bus::MediaItem>,
) -> Result<String, AgentError> {
// 确定 dialog_id
let dialog_id = dialog_id.unwrap_or(DEFAULT_DIALOG_ID);
// 获取或创建 session
let unified_id = UnifiedSessionId::new(channel, chat_id, dialog_id);
let session = self.get_or_create_session(&unified_id).await?;
// 处理消息
let response: String = {
let mut session_guard = session.lock().await;
// 确保 session 持久化记录存在
session_guard.ensure_persistent_session()?;
// 添加用户消息到历史
let media_refs: Vec<String> = media.iter().map(|m| m.path.clone()).collect();
#[cfg(debug_assertions)]
if !media_refs.is_empty() {
tracing::debug!(media_count = %media.len(), media_refs = ?media_refs, "Adding user message with media");
}
let user_message = session_guard.create_user_message(content, media_refs);
session_guard.add_message(user_message.clone());
session_guard.append_message(&user_message)?;
// 加载历史
session_guard.load_history()?;
// 压缩历史(如果需要)
let history = session_guard.get_history().to_vec();
let history = session_guard.compressor
.compress_if_needed(history)
.await?;
// 创建 agent 并处理
let agent = session_guard.create_agent()?;
let result = agent.process(history).await?;
// 持久化 assistant 消息
for msg in &result.emitted_messages {
session_guard.append_message(msg)?;
}
result.final_response.content
};
#[cfg(debug_assertions)]
tracing::debug!(
channel = %channel,
chat_id = %chat_id,
response_len = %response.len(),
"Agent response received"
);
Ok(response)
}
/// 清除指定 session 的所有历史
pub async fn clear_session_history(&self, unified_id: &UnifiedSessionId) -> Result<(), AgentError> {
let session = self.get_or_create_session(unified_id).await?;
let mut session_guard = session.lock().await;
session_guard.clear_history()?;
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
use tokio::sync::mpsc;
fn test_provider_config() -> LLMProviderConfig {
LLMProviderConfig {
provider_type: "openai".to_string(),
name: "test".to_string(),
base_url: "http://localhost".to_string(),
api_key: "test-key".to_string(),
extra_headers: HashMap::new(),
model_id: "test-model".to_string(),
temperature: Some(0.0),
max_tokens: Some(32),
model_extra: HashMap::new(),
max_tool_iterations: 1,
token_limit: 4096,
}
}
}

120
src/session/session_id.rs Normal file
View File

@ -0,0 +1,120 @@
/// Unified session identifier composed of channel, chat_id, and dialog_id
///
/// Format: `channel:chat_id:dialog_id`
///
/// Examples:
/// - CLI: `"cli_chat:sid_abc123:dialog_xyz"`
/// - Feishu: `"feishu:oc_123456:dialog_xyz"`
///
/// For simple cases where only one dialog exists per chat:
/// - `dialog_id` defaults to `"default"`
use serde::{Deserialize, Serialize};
pub const DEFAULT_DIALOG_ID: &str = "default";
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct UnifiedSessionId {
pub channel: String,
pub chat_id: String,
pub dialog_id: String,
}
impl UnifiedSessionId {
/// Create a new UnifiedSessionId
pub fn new(channel: impl Into<String>, chat_id: impl Into<String>, dialog_id: impl Into<String>) -> Self {
Self {
channel: channel.into(),
chat_id: chat_id.into(),
dialog_id: dialog_id.into(),
}
}
/// Create with default dialog_id ("default")
pub fn with_default_dialog(channel: impl Into<String>, chat_id: impl Into<String>) -> Self {
Self {
channel: channel.into(),
chat_id: chat_id.into(),
dialog_id: DEFAULT_DIALOG_ID.to_string(),
}
}
/// Parse from string format "channel:chat_id:dialog_id"
pub fn parse(s: &str) -> Option<Self> {
let parts: Vec<&str> = s.split(':').collect();
if parts.len() != 3 {
return None;
}
Some(Self {
channel: parts[0].to_string(),
chat_id: parts[1].to_string(),
dialog_id: parts[2].to_string(),
})
}
/// Convert to string format "channel:chat_id:dialog_id"
pub fn to_string(&self) -> String {
format!("{}:{}:{}", self.channel, self.chat_id, self.dialog_id)
}
/// Get the session key without dialog_id (channel:chat_id)
/// This is used to group all dialogs within a chat
pub fn chat_scope(&self) -> String {
format!("{}:{}", self.channel, self.chat_id)
}
}
impl std::fmt::Display for UnifiedSessionId {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.to_string())
}
}
// Note: No Deref implementation to avoid confusion between String and UnifiedSessionId
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_new() {
let id = UnifiedSessionId::new("cli_chat", "sid123", "dialog456");
assert_eq!(id.channel, "cli_chat");
assert_eq!(id.chat_id, "sid123");
assert_eq!(id.dialog_id, "dialog456");
}
#[test]
fn test_with_default_dialog() {
let id = UnifiedSessionId::with_default_dialog("feishu", "oc123");
assert_eq!(id.channel, "feishu");
assert_eq!(id.chat_id, "oc123");
assert_eq!(id.dialog_id, "default");
}
#[test]
fn test_parse() {
let id = UnifiedSessionId::parse("cli_chat:sid123:dialog456").unwrap();
assert_eq!(id.channel, "cli_chat");
assert_eq!(id.chat_id, "sid123");
assert_eq!(id.dialog_id, "dialog456");
}
#[test]
fn test_parse_invalid() {
assert!(UnifiedSessionId::parse("invalid").is_none());
assert!(UnifiedSessionId::parse("only:two").is_none());
}
#[test]
fn test_to_string() {
let id = UnifiedSessionId::new("feishu", "oc123", "dialog789");
assert_eq!(id.to_string(), "feishu:oc123:dialog789");
}
#[test]
fn test_chat_scope() {
let id = UnifiedSessionId::new("feishu", "oc123", "dialog789");
assert_eq!(id.chat_scope(), "feishu:oc123");
}
}

View File

@ -140,8 +140,9 @@ impl SessionStore {
&self,
channel_name: &str,
chat_id: &str,
dialog_id: &str,
) -> Result<SessionRecord, StorageError> {
let session_id = persistent_session_id(channel_name, chat_id);
let session_id = persistent_session_id(channel_name, chat_id, dialog_id);
if let Some(record) = self.get_session(&session_id)? {
return Ok(record);
}
@ -343,12 +344,8 @@ impl SessionStore {
}
}
pub fn persistent_session_id(channel_name: &str, chat_id: &str) -> String {
if channel_name == "cli" || channel_name == "cli_chat" {
chat_id.to_string()
} else {
format!("{}:{}", channel_name, chat_id)
}
pub fn persistent_session_id(channel_name: &str, chat_id: &str, dialog_id: &str) -> String {
format!("{}:{}:{}", channel_name, chat_id, dialog_id)
}
fn default_session_db_path() -> Result<PathBuf, std::io::Error> {
@ -474,9 +471,9 @@ mod tests {
#[test]
fn test_persistent_session_id_for_cli_and_channel() {
assert_eq!(persistent_session_id("cli", "abc"), "abc");
assert_eq!(persistent_session_id("cli_chat", "abc"), "abc");
assert_eq!(persistent_session_id("feishu", "abc"), "feishu:abc");
assert_eq!(persistent_session_id("cli", "abc", "default"), "cli:abc:default");
assert_eq!(persistent_session_id("cli_chat", "abc", "default"), "cli_chat:abc:default");
assert_eq!(persistent_session_id("feishu", "abc", "default"), "feishu:abc:default");
}
#[test]
@ -535,8 +532,8 @@ mod tests {
fn test_ensure_channel_session_is_stable() {
let store = SessionStore::in_memory().unwrap();
let first = store.ensure_channel_session("feishu", "chat-1").unwrap();
let second = store.ensure_channel_session("feishu", "chat-1").unwrap();
let first = store.ensure_channel_session("feishu", "chat-1", "default").unwrap();
let second = store.ensure_channel_session("feishu", "chat-1", "default").unwrap();
assert_eq!(first.id, second.id);
assert_eq!(first.chat_id, "chat-1");