Compare commits

...

27 Commits

Author SHA1 Message Date
e2fd836794 fix(session): 消息持久化和恢复
问题:消息只写内存,重启后丢失。from_storage 恢复了 meta 但没加载消息。

修复:
- add_message 改为 async,同时写内存和 Storage
- compact 场景用 persist=false 避免重复持久化
- clear_session_history 同时清内存和 Storage 消息
- 所有调用点添加 .await 和错误转换
2026-04-28 23:10:53 +08:00
c48ed83a23 refactor(session): 移除 InboundMessage.dialog_id,dialog_id 完全由 SessionManager 管理
Channel 消息不再携带 dialog_id,完全由 SessionManager 通过
current_sessions 内部管理。/new 后下一条消息会自动路由到新 session。

改动:
- InboundMessage 移除 dialog_id 字段
- handle_message 移除 dialog_id 参数
- Feishu/CLI channel 创建 InboundMessage 时不再设置 dialog_id
- 路由逻辑简化为:current_sessions → find_active_session → 创建新 session
2026-04-28 23:05:06 +08:00
e235268133 fix(session): /new 后仍停留在旧对话的问题
问题原因:/new 创建新 session 后,客户端下次发消息仍带着旧的
dialog_id,导致服务端找到旧 session。

解决方案:在 SessionManager 中新增 current_sessions 跟踪
每个 channel:chat_id 的当前活跃 session:
- create_session / get_or_create_session 时更新 current_sessions
- switch_dialog / delete_dialog 时同步更新 current_sessions
- handle_message 无 dialog_id 时优先使用 current_sessions
2026-04-28 22:53:37 +08:00
7fac68b51c feat(session): Phase 4 - 斜杠命令完善
新增命令:
- /sessions — 列出 channel:chat_id 下最近 10 条对话
- /switch <dialog_id> — 切换到指定对话
- /rename <新标题> — 重命名当前对话

更新命令:
- /delete — 软删除后创建新对话
- /info — 显示 session_id + message_count

所有命令描述改为中文。
2026-04-28 22:30:22 +08:00
ac7576bb4b feat(session): Phase 3 - SessionManager 完善
- SessionManager::new 接收 Arc<Storage> 参数
- create_session / get_or_create_session 写入/恢复 Storage
- handle_message 支持无 dialog_id 时自动查找最近活跃 session 或创建新 session
- 实现 list_dialogs() — 从 Storage 读取最近 10 条
- 实现 switch_dialog() — 从 Storage 恢复 session
- 实现 delete_dialog() — 软删除 Storage + 内存移除
- 实现 rename_dialog() — 更新 Storage 和内存 title
- 实现 archive_dialog()(空实现,archive 概念已删除)
- 新增 start_cleanup_task() 后台 TTL 清理任务
- GatewayConfig 新增 session_db_path 和 cleanup_interval_minutes 配置
- Gateway::new 改为 async,创建 Storage 并启动清理任务
- rename_dialog / delete_dialog 改为 async(需 .await)
- WsOutbound::SystemNotification 已在 Phase 2 添加
2026-04-28 22:29:24 +08:00
5f7ffd28ef feat(session): Phase 2 - 扩展 Session 结构
- Session 新增字段: title, created_at, last_active_at,
  message_count, total_message_count, seq_counter,
  storage, routing_info
- Session::new 新增 storage/routing_info/title 参数
- 新增 Session::from_storage() 从 Storage 恢复 Session
- 新增 Session::add_message_and_persist() 持久化版本
- 新增 Session::send_system_notification() 不记历史的通知
- 新增 Session::persist_session_meta() 写回 Storage
- 新增 Session::should_generate_title() / generate_title()
- 新增 LLM title 自动生成(10 条消息后触发)
- SessionManager::create_session 新增 routing_info 参数
- WsOutbound 新增 SystemNotification variant
- Client mod.rs 处理 SystemNotification
2026-04-28 22:25:10 +08:00
c17e286db1 test(storage): 编写 Storage 单元测试 2026-04-28 22:17:02 +08:00
3ffa8c41f7 feat(storage): 添加写入重试逻辑 2026-04-28 22:13:49 +08:00
69a8ec2775 feat(storage): 实现 Message CRUD 操作 2026-04-28 22:13:29 +08:00
eeb8a77ac8 feat(storage): 实现 Session CRUD 操作 2026-04-28 22:13:10 +08:00
c51ba900fa feat(storage): 定义 SessionMeta 和 MessageMeta 数据结构 2026-04-28 22:12:44 +08:00
88f8a86b99 feat(storage): 实现 Storage 主结构和初始化 2026-04-28 22:12:30 +08:00
c4c87aac95 feat(storage): 创建 storage 模块骨架 2026-04-28 22:09:10 +08:00
22546d3ad0 feat(storage): 添加 StorageError 类型 2026-04-28 22:08:47 +08:00
3ed7ac7cbd deps: 添加 sqlx + sqlite 依赖 2026-04-28 22:07:55 +08:00
575f264773 docs: 添加 Phase 1 Storage 基础实现计划
9 个 Task:
1. 添加 sqlx + sqlite 依赖
2. 创建 StorageError 类型
3. 创建 storage 模块骨架
4. 实现 Storage 主结构 + init_schema
5. 定义 SessionMeta / MessageMeta
6. 实现 Session CRUD
7. 实现 Message CRUD
8. 添加写入重试逻辑
9. 编写 7 个单元测试
2026-04-28 21:46:32 +08:00
2cbd959ac4 docs: 添加 Session 持久化设计方案
设计目标:
- SQLite 持久化,消息实时落盘
- Session = Dialog 概念等价,不再分层
- handle_message 自动恢复或创建 session
- Dialog 完整生命周期管理(/new /sessions /switch /rename /delete)
- Title 自动生成(10 条用户消息后)
- TTL 自动内存清理,Storage 保留所有数据
2026-04-28 21:45:05 +08:00
1ac6de118a 修复 /dump 命令:保存到文件并注入系统提示词
- /dump 现在保存到 {workspace}/dumps/ 目录而非仅返回文本
- dump 文件包含注入模型的完整系统提示词(AgentLoop + Skills)
- 修复 cli_chat 中 current_session_guard 变量声明问题
- gateway 正确处理 HandleResult::AgentResponse 和 CommandOutput
- 移除 cli_chat 中未使用的 parse_slash_command 导入
2026-04-28 21:27:36 +08:00
e61b78eaff 重构:斜杠命令通过 handle_message 处理,Channel 不再管理 Session
架构优化:
- Feishu 渠道的 handle_and_publish 简化为只发布消息到 bus
- Session 创建/复用由 SessionManager 在 handle_message 内部处理
- 斜杠命令检测移到 handle_message,execute_slash_command 直接调用
- CLI 和 Feishu 统一通过消息总线处理斜杠命令
2026-04-28 21:10:28 +08:00
e787203e94 修复飞书渠道会话创建时机:执行斜杠命令前先确保会话存在
在 Feishu 收到斜杠命令时,先通过 CreateDialog 创建会话,
再执行实际的命令。这样 /dump 等命令就能正确获取对话历史。
2026-04-28 21:06:25 +08:00
c52461055d 修复飞书渠道 /dump 和 /info 命令:支持无会话时自动创建
当 Feishu 用户首次发送斜杠命令时,如果没有会话则自动创建。
这使得 /dump 和 /info 命令可以在对话开始前使用。
2026-04-28 21:03:58 +08:00
30b68599a4 为飞书渠道添加斜杠命令支持
重写 FeishuChannel 的 handle_and_publish 方法,在发布消息前
检查是否为斜杠命令。如果是,则通过控制平面执行命令并返回结果。
2026-04-28 20:58:31 +08:00
a3d8ebb534 添加 /dump 命令,导出 session 为 markdown 文档
/dump 命令会输出当前 session 的完整信息:
- Session 元信息 (ID, channel, chat_id, model 等)
- 所有对话历史 (system, user, assistant, tool)
- 每条消息包含角色、时间戳、内容、工具调用等
2026-04-28 20:54:54 +08:00
84bb06cc76 优化日志输出:斜杠命令执行时打印,过滤冗余的压缩日志
- 添加斜杠命令执行的 info 日志
- Context compression 日志只在实际触发压缩时输出 (debug级别)
2026-04-28 20:52:20 +08:00
7220e89a22 简化斜杠命令,移除不可用的命令
删除:sessions, switch, rename, archive
保留:new, delete, compact, info
2026-04-28 20:50:01 +08:00
8219e7c928 优化系统提示词:移除冗余的工具列表,自动创建skills目录
- 移除 ToolsSection:工具定义已通过 API 的 tools 参数传递,无需在提示词中重复
- SkillsLoader 启动时自动创建 ~/.picobot/skills 目录
2026-04-28 20:41:36 +08:00
6c50f433d1 删除旧的session持久化机制 2026-04-28 20:34:44 +08:00
21 changed files with 2921 additions and 999 deletions

View File

@ -27,10 +27,10 @@ mime_guess = "2.0"
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"
chrono = "0.4"
hostname = "0.3"
sqlx = { version = "0.8", features = ["sqlite", "macros", "chrono", "runtime-tokio"] }

View File

@ -0,0 +1,877 @@
# Phase 1: Storage 基础 实现计划
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
**Goal:** 创建 `src/storage/` 模块,实现 SQLite 持久化层,为后续 Session 扩展提供 Storage 基础设施。
**Architecture:** 使用 `sqlx` + `sqlite`,通过 `SqlitePool` 实现异步连接池,所有 Storage 操作均为 async`Storage` 内部管理连接池的生命周期。
**Tech Stack:** `sqlx` (sqlite, tokio), `serde`, `chrono` (时间戳), `tokio::time::sleep` (重试退避)
---
## Task 1: 添加依赖
**Files:**
- Modify: `Cargo.toml:36` (在 `[dependencies]` 末尾添加)
**Step 1: 添加 sqlx + sqlite 依赖**
`Cargo.toml` 末尾添加:
```toml
sqlx = { version = "0.8", features = ["sqlite", "tokio", "macros", "chrono"] }
```
**Step 2: 运行 cargo check 验证依赖**
Run: `cargo check 2>&1`
Expected: 无报错,依赖解析成功
**Step 3: Commit**
```bash
git add Cargo.toml
git commit -m "deps: 添加 sqlx + sqlite 依赖"
```
---
## Task 2: 创建 Storage Error 类型
**Files:**
- Create: `src/storage/error.rs`
**Step 1: 编写 StorageError 枚举和测试**
```rust
use thiserror::Error;
#[derive(Error, Debug)]
pub enum StorageError {
#[error("session not found: {0}")]
NotFound(String),
#[error("session already exists: {0}")]
AlreadyExists(String),
#[error("database error: {0}")]
Database(#[from] sqlx::Error),
#[error("serialization error: {0}")]
Serialization(String),
}
```
**Step 2: 验证编译**
Run: `cargo build --lib 2>&1 | head -30`
Expected: 报错 "cannot find module `storage`"(因为模块未创建),这是预期的
**Step 3: Commit**
```bash
git add src/storage/error.rs
git commit -m "feat(storage): 添加 StorageError 类型"
```
---
## Task 3: 创建 Storage 模块骨架
**Files:**
- Create: `src/storage/mod.rs`
- Create: `src/storage/session.rs`
- Create: `src/storage/message.rs`
**Step 1: 创建 `src/storage/mod.rs`**
```rust
pub mod error;
pub mod session;
pub mod message;
pub use error::StorageError;
```
**Step 2: 创建 `src/storage/session.rs`(空壳)**
```rust
// Session CRUD 操作占位符
```
**Step 3: 创建 `src/storage/message.rs`(空壳)**
```rust
// Message CRUD 操作占位符
```
**Step 4: 在 `src/lib.rs` 中添加 storage 模块**
`src/lib.rs` 末尾添加:
```rust
pub mod storage;
```
**Step 5: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功(空壳模块)
**Step 6: Commit**
```bash
git add src/storage/ src/lib.rs
git commit -m "feat(storage): 创建 storage 模块骨架"
```
---
## Task 4: 实现 Storage 主结构
**Files:**
- Modify: `src/storage/mod.rs`
**Step 1: 编写 Storage 结构和初始化逻辑**
```rust
use sqlx::{Pool, Sqlite, SqlitePool};
use std::path::Path;
pub struct Storage {
pool: Pool<Sqlite>,
}
impl Storage {
/// 打开或创建数据库
pub async fn new(db_path: &Path) -> Result<Self, StorageError> {
let database_url = format!("sqlite:{}?mode=rwc", db_path.display());
let pool = SqlitePool::connect(&database_url).await?;
let storage = Self { pool };
storage.init_schema().await?;
Ok(storage)
}
/// 初始化数据库 schema
async fn init_schema(&self) -> Result<(), StorageError> {
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS sessions (
id TEXT PRIMARY KEY,
channel TEXT NOT NULL,
chat_id TEXT NOT NULL,
dialog_id TEXT NOT NULL,
title TEXT NOT NULL DEFAULT '新对话',
created_at INTEGER NOT NULL,
last_active_at INTEGER NOT NULL,
message_count INTEGER DEFAULT 0,
routing_info TEXT,
deleted_at INTEGER,
UNIQUE(channel, chat_id, dialog_id)
)
"#,
)
.execute(&self.pool)
.await?;
sqlx::query(
r#"
CREATE INDEX IF NOT EXISTS idx_sessions_chat
ON sessions(channel, chat_id, deleted_at)
"#,
)
.execute(&self.pool)
.await?;
sqlx::query(
r#"
CREATE TABLE IF NOT EXISTS messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
media_refs TEXT,
tool_call_id TEXT,
tool_name TEXT,
tool_calls TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
)
"#,
)
.execute(&self.pool)
.await?;
sqlx::query(
r#"
CREATE INDEX IF NOT EXISTS idx_messages_session_seq
ON messages(session_id, seq)
"#,
)
.execute(&self.pool)
.await?;
Ok(())
}
/// 获取连接池引用(供内部 CRUD 使用)
pub(crate) fn pool(&self) -> &Pool<Sqlite> {
&self.pool
}
}
```
**Step 2: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功
**Step 3: Commit**
```bash
git add src/storage/mod.rs
git commit -m "feat(storage): 实现 Storage 主结构和初始化"
```
---
## Task 5: 定义 SessionMeta 和 MessageMeta 数据结构
**Files:**
- Modify: `src/storage/session.rs`
- Modify: `src/storage/message.rs`
**Step 1: 在 `session.rs` 中定义 SessionMeta**
```rust
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionMeta {
pub id: String,
pub channel: String,
pub chat_id: String,
pub dialog_id: String,
pub title: String,
pub created_at: i64,
pub last_active_at: i64,
pub message_count: i64,
pub routing_info: Option<String>,
pub deleted_at: Option<i64>,
}
```
**Step 2: 在 `message.rs` 中定义 MessageMeta**
```rust
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageMeta {
pub id: String,
pub session_id: String,
pub seq: i64,
pub role: String,
pub content: String,
pub media_refs: Option<String>,
pub tool_call_id: Option<String>,
pub tool_name: Option<String>,
pub tool_calls: Option<String>,
pub created_at: i64,
}
```
**Step 3: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功
**Step 4: Commit**
```bash
git add src/storage/session.rs src/storage/message.rs
git commit -m "feat(storage): 定义 SessionMeta 和 MessageMeta 数据结构"
```
---
## Task 6: 实现 Session CRUD 操作
**Files:**
- Modify: `src/storage/session.rs`
**Step 1: 编写 upsert_session**
```rust
use sqlx::Row;
use super::SessionMeta;
impl Storage {
pub async fn upsert_session(&self, meta: &SessionMeta) -> Result<(), StorageError> {
sqlx::query(
r#"
INSERT INTO sessions (id, channel, chat_id, dialog_id, title, created_at, last_active_at, message_count, routing_info, deleted_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
title = excluded.title,
last_active_at = excluded.last_active_at,
message_count = excluded.message_count,
routing_info = excluded.routing_info,
deleted_at = excluded.deleted_at
"#,
)
.bind(&meta.id)
.bind(&meta.channel)
.bind(&meta.chat_id)
.bind(&meta.dialog_id)
.bind(&meta.title)
.bind(meta.created_at)
.bind(meta.last_active_at)
.bind(meta.message_count)
.bind(&meta.routing_info)
.bind(meta.deleted_at)
.execute(self.pool())
.await?;
Ok(())
}
pub async fn get_session(&self, id: &str) -> Result<SessionMeta, StorageError> {
let row = sqlx::query(
r#"
SELECT id, channel, chat_id, dialog_id, title, created_at, last_active_at, message_count, routing_info, deleted_at
FROM sessions WHERE id = ? AND deleted_at IS NULL
"#,
)
.bind(id)
.fetch_optional(self.pool())
.await?
.ok_or_else(|| StorageError::NotFound(id.to_string()))?;
Ok(SessionMeta {
id: row.get("id"),
channel: row.get("channel"),
chat_id: row.get("chat_id"),
dialog_id: row.get("dialog_id"),
title: row.get("title"),
created_at: row.get("created_at"),
last_active_at: row.get("last_active_at"),
message_count: row.get("message_count"),
routing_info: row.get("routing_info"),
deleted_at: row.get("deleted_at"),
})
}
pub async fn list_sessions(
&self,
channel: &str,
chat_id: &str,
limit: i64,
) -> Result<Vec<SessionMeta>, StorageError> {
let rows = sqlx::query(
r#"
SELECT id, channel, chat_id, dialog_id, title, created_at, last_active_at, message_count, routing_info, deleted_at
FROM sessions
WHERE channel = ? AND chat_id = ? AND deleted_at IS NULL
ORDER BY last_active_at DESC
LIMIT ?
"#,
)
.bind(channel)
.bind(chat_id)
.bind(limit)
.fetch_all(self.pool())
.await?;
Ok(rows
.into_iter()
.map(|row| SessionMeta {
id: row.get("id"),
channel: row.get("channel"),
chat_id: row.get("chat_id"),
dialog_id: row.get("dialog_id"),
title: row.get("title"),
created_at: row.get("created_at"),
last_active_at: row.get("last_active_at"),
message_count: row.get("message_count"),
routing_info: row.get("routing_info"),
deleted_at: row.get("deleted_at"),
})
.collect())
}
pub async fn touch_session(
&self,
id: &str,
message_count: i64,
last_active_at: i64,
) -> Result<(), StorageError> {
sqlx::query(
r#"
UPDATE sessions SET message_count = ?, last_active_at = ?
WHERE id = ?
"#,
)
.bind(message_count)
.bind(last_active_at)
.bind(id)
.execute(self.pool())
.await?;
Ok(())
}
pub async fn soft_delete_session(&self, id: &str) -> Result<(), StorageError> {
let now = chrono::Utc::now().timestamp_millis();
sqlx::query(
r#"UPDATE sessions SET deleted_at = ? WHERE id = ?"#,
)
.bind(now)
.bind(id)
.execute(self.pool())
.await?;
Ok(())
}
/// 查找 channel:chat_id 下最近活跃且未过期的 session
pub async fn find_active_session(
&self,
channel: &str,
chat_id: &str,
ttl_millis: i64,
) -> Result<Option<SessionMeta>, StorageError> {
let cutoff = chrono::Utc::now().timestamp_millis() - ttl_millis;
let row = sqlx::query(
r#"
SELECT id, channel, chat_id, dialog_id, title, created_at, last_active_at, message_count, routing_info, deleted_at
FROM sessions
WHERE channel = ? AND chat_id = ? AND deleted_at IS NULL AND last_active_at > ?
ORDER BY last_active_at DESC
LIMIT 1
"#,
)
.bind(channel)
.bind(chat_id)
.bind(cutoff)
.fetch_optional(self.pool())
.await?;
match row {
Some(row) => Ok(Some(SessionMeta {
id: row.get("id"),
channel: row.get("channel"),
chat_id: row.get("chat_id"),
dialog_id: row.get("dialog_id"),
title: row.get("title"),
created_at: row.get("created_at"),
last_active_at: row.get("last_active_at"),
message_count: row.get("message_count"),
routing_info: row.get("routing_info"),
deleted_at: row.get("deleted_at"),
})),
None => Ok(None),
}
}
}
```
> 注意:`Storage` 的 CRUD 方法需要能访问 `pool()`,但目前 `pool()``pub(crate)`。在 `mod.rs` 中为 `session.rs` 实现 `Storage` 的 CRUD所以同模块内可访问。
**Step 2: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功
**Step 3: Commit**
```bash
git add src/storage/session.rs
git commit -m "feat(storage): 实现 Session CRUD 操作"
```
---
## Task 7: 实现 Message CRUD 操作
**Files:**
- Modify: `src/storage/message.rs`
**Step 1: 编写 Message CRUD**
```rust
use sqlx::Row;
use super::MessageMeta;
impl Storage {
pub async fn append_message(&self, session_id: &str, msg: &MessageMeta) -> Result<i64, StorageError> {
sqlx::query(
r#"
INSERT INTO messages (id, session_id, seq, role, content, media_refs, tool_call_id, tool_name, tool_calls, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"#,
)
.bind(&msg.id)
.bind(session_id)
.bind(msg.seq)
.bind(&msg.role)
.bind(&msg.content)
.bind(&msg.media_refs)
.bind(&msg.tool_call_id)
.bind(&msg.tool_name)
.bind(&msg.tool_calls)
.bind(msg.created_at)
.execute(self.pool())
.await?;
Ok(msg.seq)
}
pub async fn append_messages(
&self,
session_id: &str,
msgs: &[MessageMeta],
) -> Result<Vec<i64>, StorageError> {
let mut seqs = Vec::with_capacity(msgs.len());
for msg in msgs {
let seq = self.append_message(session_id, msg).await?;
seqs.push(seq);
}
Ok(seqs)
}
pub async fn load_messages(
&self,
session_id: &str,
from_seq: i64,
) -> Result<Vec<MessageMeta>, StorageError> {
let rows = sqlx::query(
r#"
SELECT id, session_id, seq, role, content, media_refs, tool_call_id, tool_name, tool_calls, created_at
FROM messages
WHERE session_id = ? AND seq >= ?
ORDER BY seq ASC
"#,
)
.bind(session_id)
.bind(from_seq)
.fetch_all(self.pool())
.await?;
Ok(rows
.into_iter()
.map(|row| MessageMeta {
id: row.get("id"),
session_id: row.get("session_id"),
seq: row.get("seq"),
role: row.get("role"),
content: row.get("content"),
media_refs: row.get("media_refs"),
tool_call_id: row.get("tool_call_id"),
tool_name: row.get("tool_name"),
tool_calls: row.get("tool_calls"),
created_at: row.get("created_at"),
})
.collect())
}
pub async fn clear_messages(&self, session_id: &str) -> Result<(), StorageError> {
sqlx::query(r#"DELETE FROM messages WHERE session_id = ?"#)
.bind(session_id)
.execute(self.pool())
.await?;
Ok(())
}
}
```
> 注意:同样在 `mod.rs` 中实现,这样 `Storage` 的方法对 `message.rs` 中的 impl 可见。
**Step 2: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功
**Step 3: Commit**
```bash
git add src/storage/message.rs
git commit -m "feat(storage): 实现 Message CRUD 操作"
```
---
## Task 8: 实现写入重试逻辑
**Files:**
- Modify: `src/storage/mod.rs`
**Step 1: 在 Storage 中添加带重试的 append_message**
`mod.rs``Storage` impl 块中添加:
```rust
use tokio::time::{sleep, Duration};
impl Storage {
/// 追加消息,带重试逻辑
/// 重试 3 次100/200/300ms 退避),仍失败返回错误
pub async fn append_message_with_retry(
&self,
session_id: &str,
msg: &MessageMeta,
) -> Result<i64, StorageError> {
let delays = [100, 200, 300];
for (i, delay) in delays.iter().enumerate() {
match self.append_message(session_id, msg).await {
Ok(seq) => return Ok(seq),
Err(e) if i < delays.len() - 1 => {
sleep(Duration::from_millis(*delay)).await;
tracing::warn!("Storage write failed, retrying: {}", e);
}
Err(e) => {
tracing::error!("Storage write failed after retries: {}", e);
return Err(e);
}
}
}
unreachable!()
}
}
```
> 注意:需要 `use sqlx::Row;``mod.rs` 中。
**Step 2: 验证编译**
Run: `cargo build --lib 2>&1`
Expected: 编译成功
**Step 3: Commit**
```bash
git add src/storage/mod.rs
git commit -m "feat(storage): 添加写入重试逻辑"
```
---
## Task 9: 编写 Storage 单元测试
**Files:**
- Modify: `src/storage/mod.rs`(添加测试模块)
**Step 1: 编写 Storage 集成测试**
`src/storage/mod.rs` 末尾添加:
```rust
#[cfg(test)]
mod tests {
use super::*;
use tempfile::tempdir;
use std::path::Path;
async fn create_test_storage() -> (Storage, impl Fn()) {
let dir = tempdir().unwrap();
let db_path = dir.path().join("test.db");
let storage = Storage::new(&db_path).await.unwrap();
let cleanup = || {
drop(dir);
};
(storage, cleanup)
}
#[tokio::test]
async fn test_upsert_and_get_session() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
let meta = SessionMeta {
id: "cli_chat:sid123:dialog1".to_string(),
channel: "cli_chat".to_string(),
chat_id: "sid123".to_string(),
dialog_id: "dialog1".to_string(),
title: "测试会话".to_string(),
created_at: 1000,
last_active_at: 1000,
message_count: 0,
routing_info: Some(r#"{"type":"cli"}"#.to_string()),
deleted_at: None,
};
storage.upsert_session(&meta).await.unwrap();
let loaded = storage.get_session(&meta.id).await.unwrap();
assert_eq!(loaded.title, "测试会话");
assert_eq!(loaded.channel, "cli_chat");
}
#[tokio::test]
async fn test_get_nonexistent_session() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
let result = storage.get_session("nonexistent").await;
assert!(result.is_err());
matches!(result.unwrap_err(), StorageError::NotFound(_));
}
#[tokio::test]
async fn test_list_sessions() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
for i in 0..5 {
let meta = SessionMeta {
id: format!("cli_chat:sid123:dialog{}", i),
channel: "cli_chat".to_string(),
chat_id: "sid123".to_string(),
dialog_id: format!("dialog{}", i),
title: format!("会话{}", i),
created_at: i as i64 * 1000,
last_active_at: i as i64 * 1000,
message_count: i,
routing_info: None,
deleted_at: None,
};
storage.upsert_session(&meta).await.unwrap();
}
let sessions = storage.list_sessions("cli_chat", "sid123", 10).await.unwrap();
assert_eq!(sessions.len(), 5);
// 按 last_active_at DESC 排序
assert_eq!(sessions[0].dialog_id, "dialog4");
}
#[tokio::test]
async fn test_soft_delete() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
let meta = SessionMeta {
id: "cli_chat:sid123:dialog1".to_string(),
channel: "cli_chat".to_string(),
chat_id: "sid123".to_string(),
dialog_id: "dialog1".to_string(),
title: "测试".to_string(),
created_at: 1000,
last_active_at: 1000,
message_count: 0,
routing_info: None,
deleted_at: None,
};
storage.upsert_session(&meta).await.unwrap();
storage.soft_delete_session(&meta.id).await.unwrap();
let result = storage.get_session(&meta.id).await;
assert!(result.is_err());
matches!(result.unwrap_err(), StorageError::NotFound(_));
}
#[tokio::test]
async fn test_append_and_load_messages() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
let session_meta = SessionMeta {
id: "cli_chat:sid123:dialog1".to_string(),
channel: "cli_chat".to_string(),
chat_id: "sid123".to_string(),
dialog_id: "dialog1".to_string(),
title: "测试".to_string(),
created_at: 1000,
last_active_at: 1000,
message_count: 0,
routing_info: None,
deleted_at: None,
};
storage.upsert_session(&session_meta).await.unwrap();
let msg = MessageMeta {
id: "msg1".to_string(),
session_id: session_meta.id.clone(),
seq: 1,
role: "user".to_string(),
content: "你好".to_string(),
media_refs: None,
tool_call_id: None,
tool_name: None,
tool_calls: None,
created_at: 1000,
};
let seq = storage.append_message(&session_meta.id, &msg).await.unwrap();
assert_eq!(seq, 1);
let loaded = storage.load_messages(&session_meta.id, 0).await.unwrap();
assert_eq!(loaded.len(), 1);
assert_eq!(loaded[0].content, "你好");
}
#[tokio::test]
async fn test_touch_session() {
let (storage, cleanup) = create_test_storage().await;
defer { cleanup(); }
let meta = SessionMeta {
id: "cli_chat:sid123:dialog1".to_string(),
channel: "cli_chat".to_string(),
chat_id: "sid123".to_string(),
dialog_id: "dialog1".to_string(),
title: "测试".to_string(),
created_at: 1000,
last_active_at: 1000,
message_count: 0,
routing_info: None,
deleted_at: None,
};
storage.upsert_session(&meta).await.unwrap();
storage.touch_session(&meta.id, 5, 2000).await.unwrap();
let loaded = storage.get_session(&meta.id).await.unwrap();
assert_eq!(loaded.message_count, 5);
assert_eq!(loaded.last_active_at, 2000);
}
}
```
> 需要在 `Cargo.toml` 中添加 `tempfile` 依赖(已存在)。`defer` 宏可自己实现一个简单的:`fn defer<F: FnOnce()>(f: F) { f() }`
**Step 2: 运行测试**
Run: `cargo test storage::tests --lib 2>&1`
Expected: 所有 7 个测试 PASS
**Step 3: Commit**
```bash
git add src/storage/mod.rs
git commit -m "test(storage): 编写 Storage 单元测试"
```
---
## 汇总
| Task | 改动文件 | 关键交付物 |
|------|----------|-----------|
| 1 | `Cargo.toml` | sqlx 依赖 |
| 2 | `src/storage/error.rs` | StorageError |
| 3 | `src/storage/{mod.rs,session.rs,message.rs}`, `src/lib.rs` | 模块骨架 |
| 4 | `src/storage/mod.rs` | Storage 结构 + init_schema |
| 5 | `src/storage/session.rs`, `message.rs` | SessionMeta, MessageMeta |
| 6 | `src/storage/session.rs` | Session CRUD |
| 7 | `src/storage/message.rs` | Message CRUD |
| 8 | `src/storage/mod.rs` | append_message_with_retry |
| 9 | `src/storage/mod.rs` | 7 个单元测试 |
**Phase 1 完成后:** Storage 模块可独立使用,具备完整的持久化能力,可安全地集成到 Session 和 SessionManager 中。

View File

@ -0,0 +1,278 @@
# Session 持久化设计方案
## 概述
为 PicoBot 添加 SQLite 持久化层,实现 Session 数据的持久化、完整 Dialog 生命周期管理、消息实时落盘、以及基于 TTL 的自动内存清理。
## 核心概念
```
UnifiedSessionId = {channel}:{chat_id}:{dialog_id}
Session = Dialog两者等价不再分层
```
每个 Session 独立管理自己的消息历史、LLM 配置和路由信息。
## 数据库 Schema
### sessions 表
```sql
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
channel TEXT NOT NULL,
chat_id TEXT NOT NULL,
dialog_id TEXT NOT NULL,
title TEXT NOT NULL DEFAULT '新对话',
created_at INTEGER NOT NULL,
last_active_at INTEGER NOT NULL,
message_count INTEGER DEFAULT 0,
routing_info TEXT,
deleted_at INTEGER,
UNIQUE(channel, chat_id, dialog_id)
);
CREATE INDEX idx_sessions_chat ON sessions(channel, chat_id, deleted_at);
```
### messages 表
```sql
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
media_refs TEXT,
tool_call_id TEXT,
tool_name TEXT,
tool_calls TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
CREATE INDEX idx_messages_session_seq ON messages(session_id, seq);
```
## Storage API
### Session 操作
| 方法 | 说明 |
|------|------|
| `new(db_path) -> Storage` | 打开/创建数据库 |
| `upsert_session(meta) -> Result<(), StorageError>` | 插入或更新 session 元数据 |
| `get_session(id) -> Result<SessionMeta, StorageError>` | 获取单个 session |
| `list_sessions(channel, chat_id, limit) -> Result<Vec<SessionMeta>>` | 最近 N 条 |
| `touch_session(id, message_count, last_active_at)` | 更新计数和最后活跃时间 |
| `soft_delete_session(id) -> Result<(), StorageError>` | 软删除 |
### Message 操作
| 方法 | 说明 |
|------|------|
| `append_message(session_id, msg) -> Result<i64, StorageError>` | 追加单条消息,返回 seq |
| `append_messages(session_id, msgs) -> Result<Vec<i64>, StorageError>` | 批量追加 |
| `load_messages(session_id, from_seq) -> Result<Vec<MessageMeta>>` | 从指定 seq 加载 |
| `clear_messages(session_id) -> Result<(), StorageError>` | 清除消息(保留 session |
### 写入失败处理
重试 3 次100/200/300ms 退避),仍失败则发送系统通知告警。
## Session 结构
```rust
pub struct Session {
pub id: UnifiedSessionId,
pub title: String,
pub created_at: i64,
pub last_active_at: i64,
pub message_count: i64, // 用户消息计数
pub total_message_count: i64, // 含系统消息
messages: Vec<ChatMessage>, // 内存消息历史
seq_counter: i64, // 下一个消息的 seq
provider_config: LLMProviderConfig,
provider: Arc<dyn LLMProvider>,
tools: Arc<ToolRegistry>,
compressor: ContextCompressor,
user_tx: mpsc::Sender<WsOutbound>,
storage: Arc<Storage>, // 持久化 sink
routing_info: String, // JSON 路由信息
}
```
### 初始化流程
```
new() 或 from_storage()
注入 storage 引用
创建 provider, tools, compressor
从 Storage 加载 messagesfrom_seq = 0
设置 seq_counter = messages.len() + 1
返回 Session 实例
```
## handle_message 流程
```
handle_message(channel, chat_id, sender_id, content, media)
├── 1. 确定 dialog_id
│ │
│ ├── 显式传入 dialog_id → 使用
│ └── 无 dialog_id
│ ├── 查找 channel:chat_id 下最近活跃且未过期的 session
│ ├── 找到 → 使用该 session
│ └── 未找到 → 创建新 sessiondialog_id = 新随机 ID
├── 2. 获取或创建 Session
│ 有 → 更新 session_timestamps
│ 无 → 从 Storage 恢复 或 创建新 Session
├── 3. 追加用户消息并持久化
│ seq = seq_counter; seq_counter += 1
│ Storage.append_message()(失败重试 → 告警)
│ messages.push(user_msg)
│ message_count += 1
├── 4. 检查 title 自动生成
│ message_count == 10 且 title == 默认值 → LLM 生成 → 更新 title → 写回 Storage
├── 5. 注入 skills_prompt
├── 6. 新 session 注入欢迎消息(系统消息,不计入 message_count
├── 7. 上下文压缩(如需要)
├── 8. 调用 AgentLoop
├── 9. 持久化 Agent 响应
└── 10. 返回响应
```
## Dialog 生命周期命令
| 命令 | 行为 |
|------|------|
| `/new [标题]` | 创建新 dialog新随机 dialog_id新建 Session |
| `/sessions` | 列出 channel:chat_id 下最近 10 条 session按 last_active_at 倒序) |
| `/switch <dialog_id>` | 切换到指定 session从 Storage 恢复或内存命中) |
| `/rename <新标题>` | 重命名当前 session |
| `/delete` | 软删除当前 session内存移除 + Storage 标记 deleted_at |
| `/info` | 显示当前 session 信息 |
| `/compact` | 手动触发上下文压缩 |
## 路由信息
每种 Channel 在创建 Session 时注入路由信息:
```rust
// CLI
routing_info = json!({"type": "cli", "ws_sender_id": "xxx"})
// Feishu
routing_info = json!({"type": "feishu", "open_conversation_id": "oc_xxx", "tenant_key": "xxx"})
```
## Title 自动生成
调用时机:
1. Session 首次创建时(初始 title = "新对话"
2. `message_count` 达到 10 且 title 仍为默认值时,自动更新
生成 Prompt
```
给定以下对话历史生成一个简短的会话标题5-15 个中文字符),
概括这个对话的核心内容或用户的主要需求。只返回一个标题,不要解释。
历史:
{messages}
```
## TTL 清理
- 内存 session 超时 → 释放内存Storage 记录保留
- 用户切换回该 session → 从 Storage 重新加载到内存
- Storage 中的 session 记录通过 `deleted_at` 软删除,不会物理删除
## 文件结构
```
src/
├── storage/
│ ├── mod.rs # Storage 主模块
│ ├── session.rs # Session CRUD
│ ├── message.rs # Message CRUD
│ └── error.rs # StorageError
└── session/
├── mod.rs # 导出 Session, SessionManager
├── session.rs # Session, SessionManager 实现
├── session_id.rs # UnifiedSessionId
├── commands.rs # SessionCommand
├── events.rs # SessionEvent, DialogInfo
└── error.rs # SessionError
```
## 实现顺序
### Phase 1: Storage 基础
1. 添加 `sqlx` + `sqlite` 依赖
2. 实现 `Storage` 结构(连接池、初始化)
3. Session CRUD + Message CRUD
4. 写入重试逻辑
5. 单元测试
### Phase 2: Session 扩展
1. 扩展 `Session` 结构(添加 storage、routing_info、计数字段、seq_counter
2. `from_storage()` 恢复逻辑
3. `add_message` 持久化集成
4. `send_system_notification` 接口
5. Title 自动生成
### Phase 3: SessionManager 完善
1. 注入 `Arc<Storage>`
2. 实现 `list_dialogs()`
3. 实现 `switch_dialog()`
4. 实现 `delete_dialog()` / `rename_dialog()`
5. 后台 TTL 清理任务
6. 集成测试
### Phase 4: 斜杠命令
1. 实现 `/sessions`
2. 实现 `/switch`
3. 实现 `/rename`
4. 实现 `/delete`
5. 端到端测试
## 配置项
```json
{
"session": {
"ttl_hours": 24,
"cleanup_interval_minutes": 60,
"auto_title_after_n_messages": 10,
"storage_retry_delays_ms": [100, 200, 300]
}
}
```
## 与现有代码的冲突点
| 冲突 | 处理方式 |
|------|----------|
| `DialogInfo``archived_at` | 删除该字段,改用 `deleted_at` |
| `SessionCommand::ArchiveDialog` | 删除 |
| `/new` 现有行为 | 改为创建新 session新 dialog_id |
| 现有 `Session` 无 storage/routing_info | 扩展结构,新增 `from_storage()` |
| `SessionManager` 需注入 `Arc<Storage>` | 扩展构造方法 |
| stub 方法 | 实现 |

433
session_plan.md Normal file
View File

@ -0,0 +1,433 @@
# Session 管理详细设计
## 一、设计目标
1. Session 数据持久化到 SQLite系统重启后可恢复
2. 支持 Dialog 的完整生命周期管理(创建/列表/切换/重命名/删除)
3. 基于 TTL 的自动内存清理DB 保留所有数据)
4. LLM 自动生成会话标题title帮助用户和 AI 理解对话上下文
5. 每条消息实时写入 DB失败后重试 + 告警
---
## 二、概念定义
### 2.1 层级结构
```
Channel (渠道)
└── Chat (聊天)
└── Dialog (对话)
└── Session (会话实例)
```
- **Channel**:消息来源渠道(如 `cli_chat``feishu`
- **Chat**:同一渠道下的一个聊天会话(如某个 CLI session ID 或飞书 open_conversation_id
- **Dialog**:聊天内的多个独立对话线程(如 `/new` 创建的新对话)
- **Session**:一个 Dialog 的运行时实例包含消息历史、LLM 配置、工具等
### 2.2 UnifiedSessionId
```
格式:{channel}:{chat_id}:{dialog_id}
示例cli_chat:sid_abc123:dialog_xyz
feishu:oc_123456:default
```
| 字段 | 说明 |
|------|------|
| channel | 渠道标识 |
| chat_id | 聊天标识 |
| dialog_id | 对话标识(默认 `default` |
### 2.3 Session 与 Dialog 的关系
- 每个 Dialog 在运行时对应一个 `Session` 实例
- `Session` 存在于内存中,可通过 `UnifiedSessionId` 访问
- `Dialog` 是 Storage 中的持久化记录,`Session` 是其运行时投影
---
## 三、数据库 Schema
### 3.1 sessions 表
```sql
CREATE TABLE sessions (
id TEXT PRIMARY KEY,
channel TEXT NOT NULL,
chat_id TEXT NOT NULL,
dialog_id TEXT NOT NULL,
title TEXT NOT NULL,
created_at INTEGER NOT NULL,
last_active_at INTEGER NOT NULL,
message_count INTEGER DEFAULT 0,
deleted_at INTEGER,
UNIQUE(channel, chat_id, dialog_id)
);
```
> 注意:已删除 `archived_at` 字段,不保留归档概念。
### 3.2 messages 表
```sql
CREATE TABLE messages (
id TEXT PRIMARY KEY,
session_id TEXT NOT NULL,
seq INTEGER NOT NULL,
role TEXT NOT NULL,
content TEXT NOT NULL,
media_refs TEXT,
tool_call_id TEXT,
tool_name TEXT,
tool_calls TEXT,
created_at INTEGER NOT NULL,
FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE
);
CREATE INDEX idx_messages_session_seq ON messages(session_id, seq);
CREATE INDEX idx_sessions_chat ON sessions(channel, chat_id, deleted_at);
```
---
## 四、Storage API
### 4.1 Storage 职责
- 唯一的持久化 source of truth
- 由调用方GatewayState构造通过 `Arc<Storage>` 注入 `SessionManager`
- 所有写操作失败后重试 3 次100ms, 200ms, 300ms 退避),仍失败则触发系统通知告警
### 4.2 Session 操作
| 方法 | 说明 |
|------|------|
| `new(db_path) -> Storage` | 打开/创建数据库 |
| `upsert_session(meta) -> Result<(), StorageError>` | 插入或更新 session 元数据 |
| `get_session(id) -> Result<SessionMeta, StorageError>` | 获取单个 session |
| `list_sessions(channel, chat_id, limit) -> Result<Vec<SessionMeta>>` | 最近 N 条(供 `/sessions` |
| `delete_session(id) -> Result<(), StorageError>` | 物理删除 session 及关联消息 |
| `touch_session(id, message_count, last_active_at)` | 更新计数和最后活跃时间 |
### 4.3 Message 操作
| 方法 | 说明 |
|------|------|
| `append_message(session_id, msg) -> Result<i64, StorageError>` | 追加单条消息,返回 seq |
| `append_messages(session_id, msgs) -> Result<Vec<i64>, StorageError>` | 批量追加 |
| `load_messages(session_id, from_seq) -> Result<Vec<MessageMeta>>` | 从指定 seq 加载消息 |
| `clear_messages(session_id) -> Result<(), StorageError>` | 清除消息(保留 session |
### 4.4 写入失败处理
```rust
async fn append_message_with_retry(&self, session_id: &str, msg: &MessageMeta) -> Result<i64, StorageError> {
let delays = [100, 200, 300];
for (i, delay) in delays.iter().enumerate() {
match self.append_message(session_id, msg) {
Ok(seq) => return Ok(seq),
Err(e) if i < delays.len() - 1 => {
sleep(Duration::from_millis(*delay)).await;
tracing::warn!("Storage write failed, retrying: {}", e);
}
Err(e) => {
// 全部重试失败后,通过 Session 发送系统通知
return Err(e);
}
}
}
unreachable!()
}
```
---
## 五、Session 结构
```rust
pub struct Session {
pub id: UnifiedSessionId,
pub title: String, // 会话标题(用户指定或 LLM 自动生成)
pub created_at: i64, // 创建时间ms
pub last_active_at: i64, // 最后活跃时间ms
pub message_count: i64, // 用户消息计数(触发 title 自动生成)
pub total_message_count: i64, // 含系统消息的总数
messages: Vec<ChatMessage>, // 内存中的消息历史(压缩后)
seq_counter: i64, // 下一个消息的 seq
provider_config: LLMProviderConfig,
provider: Arc<dyn LLMProvider>,
tools: Arc<ToolRegistry>,
compressor: ContextCompressor,
user_tx: mpsc::Sender<WsOutbound>,
storage: Arc<Storage>, // 持久化 sink
}
```
### 5.1 初始化流程
```
new() 或 from_storage()
注入 storage 引用
创建 provider, tools, compressor
从 Storage 加载 messagesfrom_seq = 0
设置 seq_counter = messages.len() + 1
返回 Session 实例
```
### 5.2 消息管理
```rust
pub async fn add_message(&mut self, msg: ChatMessage) -> Result<(), StorageError> {
// 1. 分配序号: seq = seq_counter; seq_counter += 1
// 2. 转换为 MessageMeta
// 3. 写入 Storage重试 + 告警)
// 4. 更新内存: messages.push(msg)
// 5. 更新计数: message_count / total_message_count += 1
// 6. 更新 last_active_at
}
```
### 5.3 系统通知接口(不记历史)
```rust
impl Session {
/// 发送系统通知(不记录进 session 历史)
pub async fn send_system_notification(&self, content: &str) {
let msg = WsOutbound::SystemNotification {
content: content.to_string(),
};
let _ = self.user_tx.send(msg).await;
}
}
```
### 5.4 Title 自动生成
调用时机:
1. Session 首次创建时(初始 title = "Dialog {dialog_id}"
2. `message_count` 达到阈值10 条)且 title 仍为默认值时,自动更新为 LLM 生成
3. 用户执行 `/rename` 命令手动更新
生成 Prompt
```
给定以下对话历史生成一个简短的会话标题5-15 个中文字符),
概括这个对话的核心内容或用户的主要需求。只返回一个标题,不要解释。
历史:
{messages}
```
---
## 六、SessionManager 设计
### 6.1 数据结构
```rust
pub struct SessionManager {
inner: Arc<Mutex<SessionManagerInner>>,
provider_config: LLMProviderConfig,
tools: Arc<ToolRegistry>,
skills_loader: Arc<SkillsLoader>,
storage: Arc<Storage>, // 由调用方注入
cleanup_interval: Duration,
}
struct SessionManagerInner {
sessions: HashMap<String, Arc<Mutex<Session>>>,
session_timestamps: HashMap<String, Instant>,
session_ttl: Duration,
}
```
### 6.2 handle_message 完整流程
```
handle_message(channel, sender_id, chat_id, dialog_id, content, media)
├── 1. 确定 UnifiedSessionId
│ │
│ ├── dialog_id 有值 → 直接使用
│ │
│ └── dialog_id 无值 → 查找 channel:chat_id 下最近活跃的 session
│ ├── 找到且未过期 → 使用该 session
│ └── 未找到或已过期 → 创建新 session
├── 2. 获取或创建 Session
│ 有 → 更新 session_timestamps
│ 无 → 从 Storage 恢复 或 创建新 Session
├── 3. 添加用户消息并持久化
│ seq = seq_counter; seq_counter += 1
│ Storage.append_message()(失败重试 → 仍失败则 send_system_notification
│ messages.push(user_msg)
│ message_count += 1
├── 4. 检查 title 自动生成条件
│ message_count == 10 → 调用 LLM 生成 → 更新 title → 写回 Storage
├── 5. 注入 skills_promptindex 0 之后)
├── 6. 新 session 注入欢迎消息(作为系统消息,不计入 message_count
├── 7. 上下文压缩(如需要)
├── 8. 调用 AgentLoop
├── 9. 保存 Agent 响应消息并持久化(同样流程)
└── 10. 返回最终响应
```
**欢迎消息**(仅新 session 创建时注入历史):
```
新对话已创建!会话 ID: {dialog_id}
使用 /sessions 查看所有对话,/switch <id> 切换对话。
```
### 6.3 Dialog 生命周期
| 操作 | 方法 | 说明 |
|------|------|------|
| 创建 | `create_dialog()` | 生成 dialog_id创建 Session写入 Storage |
| 列表 | `list_dialogs()` | 从 Storage 读取limit=10 |
| 切换 | `switch_dialog()` | 从 Storage 加载 session激活到内存 |
| 重命名 | `rename_dialog()` | 更新 Storage 和内存 title |
| 删除 | `delete_dialog()` | 删除内存 session + 删除 Storage 记录 |
| 软重置 | **已删除** | 用户直接 `/new` 开新 session |
### 6.4 TTL 清理
```rust
fn start_cleanup_task(&self) {
tokio::spawn(async move {
loop {
sleep(cleanup_interval).await;
self.run_cleanup().await;
}
});
}
async fn run_cleanup(&self) {
// 扫描 session_timestamps
// 超时的 session → 从内存 HashMap 移除
// Storage 中的 session 记录保留(用户切回可重新加载)
}
```
清理策略:
- 内存 session 超时 → 仅释放内存Storage 记录保留
- 用户切换回该 session → 从 Storage 重新加载到内存
---
## 七、斜杠命令
| 命令 | 触发词 | 说明 |
|------|--------|------|
| new | `/new [标题]` | 创建新 dialog |
| sessions | `/sessions` | 列出当前 chat 最近 10 条 dialog |
| switch | `/switch <dialog_id>` | 切换到指定 dialog |
| rename | `/rename <新标题>` | 重命名当前 dialog |
| delete | `/delete` | 删除当前 dialog内存 + Storage |
| compact | `/compact` | 手动触发上下文压缩 |
| info | `/info` | 显示当前 dialog 信息 |
---
## 八、错误处理
```rust
pub enum StorageError {
NotFound(String),
AlreadyExists(String),
Database(String),
Serialization(String),
}
```
| 场景 | 处理 |
|------|------|
| Storage 写入失败 | 重试 3 次 → 发送系统通知告警 |
| Storage 读取失败 | 若 session 在内存中,继续使用内存数据 |
| Session 不存在 | 创建新 session |
| 并发冲突 | SQLite transaction 保护 |
---
## 九、配置项
```json
{
"session": {
"ttl_hours": 24,
"cleanup_interval_minutes": 60,
"auto_title_after_n_messages": 10,
"storage_retry_delays_ms": [100, 200, 300]
}
}
```
---
## 十、文件结构
```
src/
├── storage/
│ ├── mod.rs # Storage 主模块
│ ├── session.rs # Session CRUD
│ ├── message.rs # Message CRUD
│ └── error.rs # StorageError
└── session/
├── mod.rs # 导出 Session, SessionManager
├── session.rs # Session, SessionManager 实现
├── session_id.rs # UnifiedSessionId已有
├── commands.rs # SessionCommand已有
├── events.rs # SessionEvent, DialogInfo已有
└── error.rs # SessionError已有
```
---
## 十一、实现顺序
### Phase 1: Storage 基础
1. `Storage` 结构和数据库初始化
2. Session CRUDupsert_session, get_session, list_sessions, delete_session
3. Message CRUDappend_message, load_messages
4. 写入失败重试逻辑
5. 单元测试
### Phase 2: Session 扩展
1. 扩展 `Session` 结构体(添加 storage 引用、计数字段、seq_counter
2. `add_message` 持久化集成
3. `send_system_notification` 接口
4. `from_storage()` 恢复逻辑
5. Title 自动生成 LLM 调用
### Phase 3: SessionManager 完善
1. 将 `Arc<Storage>` 集成到 `SessionManager`
2. 实现 `list_dialogs()`limit=10
3. 实现 `switch_dialog()`(从 Storage 加载)
4. 实现 `delete_dialog()`(内存 + Storage
5. 实现 `rename_dialog()`
6. 后台 TTL 清理任务
7. 集成测试
### Phase 4: 斜杠命令
1. 实现 `/sessions`(列出最近 10 条)
2. 实现 `/switch`
3. 实现 `/rename`
4. 实现 `/delete`
5. 端到端测试

View File

@ -110,17 +110,11 @@ impl ContextCompressor {
// Check if compression is needed
let tokens = estimate_tokens(&history);
if tokens <= self.threshold() {
#[cfg(debug_assertions)]
tracing::info!(
tokens = tokens,
threshold = self.threshold(),
msg_count = history.len(),
"Context compression not needed"
);
return Ok(history);
}
tracing::info!(
#[cfg(debug_assertions)]
tracing::debug!(
tokens = tokens,
threshold = self.threshold(),
msg_count = history.len(),
@ -172,7 +166,8 @@ impl ContextCompressor {
}
}
tracing::info!(
#[cfg(debug_assertions)]
tracing::debug!(
final_tokens = estimate_tokens(&current_history),
final_msg_count = current_history.len(),
"Context compression completed"

View File

@ -39,7 +39,6 @@ impl SystemPromptBuilder {
sections: vec![
Box::new(ToolHonestySection),
Box::new(NoToolNarrationSection),
Box::new(ToolsSection),
Box::new(YourTaskSection),
Box::new(SafetySection),
Box::new(WorkspaceSection),

View File

@ -164,7 +164,6 @@ 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>,

View File

@ -8,7 +8,6 @@ use crate::session::{SessionCommand, SessionEvent, UnifiedSessionId};
use crate::protocol::{parse_inbound, WsInbound, WsOutbound, SlashCommandInfo};
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 {
@ -112,65 +111,12 @@ impl CliChatChannel {
match inbound {
WsInbound::UserInput { content, chat_id, .. } => {
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(&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, cmd_args)) = 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(),
args: if cmd_args.is_empty() { None } else { Some(cmd_args.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
// All messages (including slash commands) go through the normal inbound flow
// SessionManager handles session creation/reuse internally
let msg = InboundMessage {
channel: channel_name,
channel: self.name().to_string(),
sender_id: "cli".to_string(),
chat_id: chat_id_part,
dialog_id: dialog_id_part,
chat_id: chat_id.unwrap_or_else(short_id),
content,
timestamp: crate::bus::message::current_timestamp(),
media: Vec::new(),

View File

@ -1110,7 +1110,6 @@ 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(),
@ -1995,6 +1994,18 @@ impl Channel for FeishuChannel {
"feishu"
}
/// Handle an inbound message: check for slash commands first, then publish to bus
async fn handle_and_publish(
&self,
bus: &Arc<MessageBus>,
msg: &crate::bus::InboundMessage,
) -> Result<(), ChannelError> {
// All messages (including slash commands) go through the normal inbound flow
// SessionManager handles session creation/reuse internally
bus.publish_inbound(msg.clone()).await?;
Ok(())
}
async fn start(&self, bus: Arc<MessageBus>) -> Result<(), ChannelError> {
if self.config.app_id.is_empty() || self.config.app_secret.is_empty() {
return Err(ChannelError::ConfigError(

View File

@ -130,5 +130,8 @@ async fn handle_ws_message(app: &mut App, outbound: WsOutbound) {
WsOutbound::CommandExecuted { message } => {
app.add_message(MessageRole::System, message);
}
WsOutbound::SystemNotification { content } => {
app.add_message(MessageRole::System, content);
}
}
}

View File

@ -136,6 +136,10 @@ pub struct GatewayConfig {
pub port: u16,
#[serde(default, rename = "session_ttl_hours")]
pub session_ttl_hours: Option<u64>,
#[serde(default, rename = "cleanup_interval_minutes")]
pub cleanup_interval_minutes: Option<u64>,
#[serde(default, rename = "session_db_path")]
pub session_db_path: Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
@ -162,6 +166,8 @@ impl Default for GatewayConfig {
host: default_gateway_host(),
port: default_gateway_port(),
session_ttl_hours: None,
cleanup_interval_minutes: None,
session_db_path: None,
}
}
}

View File

@ -20,7 +20,7 @@ pub struct GatewayState {
}
impl GatewayState {
pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
pub async fn new() -> Result<Self, Box<dyn std::error::Error>> {
let config = Config::load_default()?;
// Initialize workspace directory: expand path and ensure it exists
@ -41,7 +41,24 @@ impl GatewayState {
// Session TTL from config (default 4 hours)
let session_ttl_hours = config.gateway.session_ttl_hours.unwrap_or(4);
let session_manager = SessionManager::new(session_ttl_hours, provider_config.clone())?;
// Initialize Storage
let db_path = if let Some(ref path) = config.gateway.session_db_path {
std::path::PathBuf::from(path)
} else {
workspace_path.join(".picobot_sessions.db")
};
let storage = Arc::new(
crate::storage::Storage::new(&db_path).await
.map_err(|e| format!("failed to initialize session storage: {}", e))?
);
tracing::info!("Session storage: {}", db_path.display());
let session_manager = SessionManager::new(session_ttl_hours, provider_config.clone(), storage.clone())?;
// Start background cleanup task (default 60 minutes)
let cleanup_interval = config.gateway.cleanup_interval_minutes.unwrap_or(60);
Arc::new(session_manager.clone()).start_cleanup_task(cleanup_interval);
tracing::info!("Session cleanup task started (interval: {} min)", cleanup_interval);
// Create CLI Chat Channel first (needed for ChannelManager)
let cli_chat_channel = Arc::new(CliChatChannel::new());
@ -90,15 +107,27 @@ impl GatewayState {
&inbound.channel,
&inbound.sender_id,
&inbound.chat_id,
inbound.dialog_id.as_deref(),
&inbound.content,
inbound.media,
).await {
Ok(response_content) => {
Ok(crate::session::session::HandleResult::AgentResponse(content)) => {
let outbound = crate::bus::OutboundMessage {
channel: inbound.channel.clone(),
chat_id: inbound.chat_id.clone(),
content: response_content,
content,
reply_to: None,
media: vec![],
metadata: inbound.forwarded_metadata,
};
if let Err(e) = bus.publish_outbound(outbound).await {
tracing::error!(error = %e, "Failed to publish outbound");
}
}
Ok(crate::session::session::HandleResult::CommandOutput(content)) => {
let outbound = crate::bus::OutboundMessage {
channel: inbound.channel.clone(),
chat_id: inbound.chat_id.clone(),
content,
reply_to: None,
media: vec![],
metadata: inbound.forwarded_metadata,
@ -160,7 +189,7 @@ impl GatewayState {
.map_err(|e| ChannelError::Other(e.to_string()))
}
RenameDialog { session_id, title } => {
session_manager.rename_dialog(&session_id, &title)
session_manager.rename_dialog(&session_id, &title).await
.map(|()| SessionEvent::DialogRenamed { session_id, title })
.map_err(|e| ChannelError::Other(e.to_string()))
}
@ -170,7 +199,7 @@ impl GatewayState {
.map_err(|e| ChannelError::Other(e.to_string()))
}
DeleteDialog { session_id } => {
session_manager.delete_dialog(&session_id)
session_manager.delete_dialog(&session_id).await
.map(|()| SessionEvent::DialogDeleted { session_id })
.map_err(|e| ChannelError::Other(e.to_string()))
}
@ -200,7 +229,7 @@ pub async fn run(host: Option<String>, port: Option<u16>) -> Result<(), Box<dyn
logging::init_logging();
tracing::info!("Starting PicoBot Gateway");
let state = Arc::new(GatewayState::new()?);
let state = Arc::new(GatewayState::new().await?);
// Initialize and start channels with workspace directory
state.channel_manager.init(

View File

@ -9,6 +9,6 @@ pub mod protocol;
pub mod channels;
pub mod logging;
pub mod observability;
pub mod storage;
pub mod skills;
pub mod storage;
pub mod tools;

View File

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

View File

@ -14,9 +14,3 @@ impl From<crate::channels::base::ChannelError> for SessionError {
SessionError::Other(e.to_string())
}
}
impl From<crate::storage::StorageError> for SessionError {
fn from(e: crate::storage::StorageError) -> Self {
SessionError::Other(e.to_string())
}
}

File diff suppressed because it is too large Load Diff

View File

@ -70,6 +70,15 @@ impl SkillsLoader {
let mut state = self.state.lock().unwrap();
state.loaded_skills.clear();
// Ensure ~/.picobot/skills directory exists
if !self.picobot_skills_dir.exists() {
if let Err(e) = std::fs::create_dir_all(&self.picobot_skills_dir) {
tracing::warn!(dir = %self.picobot_skills_dir.display(), error = %e, "Failed to create skills directory");
} else {
tracing::info!(dir = %self.picobot_skills_dir.display(), "Created skills directory");
}
}
// Load from ~/.picobot/skills
if self.picobot_skills_dir.exists() {
let loaded = self.load_skills_from_dir(&self.picobot_skills_dir);

16
src/storage/error.rs Normal file
View File

@ -0,0 +1,16 @@
use thiserror::Error;
#[derive(Error, Debug)]
pub enum StorageError {
#[error("session not found: {0}")]
NotFound(String),
#[error("session already exists: {0}")]
AlreadyExists(String),
#[error("database error: {0}")]
Database(#[from] sqlx::Error),
#[error("serialization error: {0}")]
Serialization(String),
}

15
src/storage/message.rs Normal file
View File

@ -0,0 +1,15 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageMeta {
pub id: String,
pub session_id: String,
pub seq: i64,
pub role: String,
pub content: String,
pub media_refs: Option<String>,
pub tool_call_id: Option<String>,
pub tool_name: Option<String>,
pub tool_calls: Option<String>,
pub created_at: i64,
}

File diff suppressed because it is too large Load Diff

15
src/storage/session.rs Normal file
View File

@ -0,0 +1,15 @@
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionMeta {
pub id: String,
pub channel: String,
pub chat_id: String,
pub dialog_id: String,
pub title: String,
pub created_at: i64,
pub last_active_at: i64,
pub message_count: i64,
pub routing_info: Option<String>,
pub deleted_at: Option<i64>,
}