xiaoski/PicoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: xiaoski:e2fd8367946aa513c83bb1187034e4e48e2c0142
Branches Tags
xiaoski:main
..
compare: xiaoski:61eea62bfc2c8f5b79893e53164f0ac77bd8b705
Branches Tags
xiaoski:main

No commits in common. "e2fd8367946aa513c83bb1187034e4e48e2c0142" and "61eea62bfc2c8f5b79893e53164f0ac77bd8b705" have entirely different histories.
e2fd836794 ... 61eea62bfc

21 changed files with 975 additions and 2897 deletions
Show Stats Expand all files Collapse all files

2
Cargo.toml
View File

@ -27,10 +27,10 @@ mime_guess = "2.0"
base64 = "0.22" base64 = "0.22"
tempfile = "3" tempfile = "3"
meval = "0.2" meval = "0.2"
rusqlite = { version = "0.32", features = ["bundled"] }
ratatui = "0.27" ratatui = "0.27"
crossterm = { version = "0.28", features = ["event-stream"] } crossterm = { version = "0.28", features = ["event-stream"] }
termimad = "0.34" termimad = "0.34"
textwrap = "0.16" textwrap = "0.16"
chrono = "0.4" chrono = "0.4"
hostname = "0.3" hostname = "0.3"
sqlx = { version = "0.8", features = ["sqlite", "macros", "chrono", "runtime-tokio"] }

877
docs/plans/2026-04-28-phase1-storage-implementation.md
View File

@ -1,877 +0,0 @@
# 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 中。

278
docs/plans/2026-04-28-session-persistence-design.md
View File

@ -1,278 +0,0 @@
# 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 加载 messages（from_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
│ └── 未找到 → 创建新 session（dialog_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
View File

@ -1,433 +0,0 @@
# 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 加载 messages（from_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_prompt（index 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 CRUD（upsert_session, get_session, list_sessions, delete_session）
3. Message CRUD（append_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. 端到端测试

13
src/agent/context_compressor.rs
View File

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

1
src/agent/system_prompt.rs
View File

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

1
src/bus/message.rs
View File

@ -164,6 +164,7 @@ pub struct InboundMessage {
pub channel: String, pub channel: String,
pub sender_id: String, pub sender_id: String,
pub chat_id: String, pub chat_id: String,
pub dialog_id: Option<String>,
pub content: String, pub content: String,
pub timestamp: i64, pub timestamp: i64,
pub media: Vec<MediaItem>, pub media: Vec<MediaItem>,

62
src/channels/cli_chat.rs
View File

@ -8,6 +8,7 @@ use crate::session::{SessionCommand, SessionEvent, UnifiedSessionId};
use crate::protocol::{parse_inbound, WsInbound, WsOutbound, SlashCommandInfo}; use crate::protocol::{parse_inbound, WsInbound, WsOutbound, SlashCommandInfo};
use super::base::{Channel, ChannelError}; use super::base::{Channel, ChannelError};
use super::slash_command::parse_slash_command;
/// Generate a short ID (8 characters) from a UUID /// Generate a short ID (8 characters) from a UUID
fn short_id() -> String { fn short_id() -> String {
@ -111,12 +112,65 @@ impl CliChatChannel {
match inbound { match inbound {
WsInbound::UserInput { content, chat_id, .. } => { WsInbound::UserInput { content, chat_id, .. } => {
// All messages (including slash commands) go through the normal inbound flow let chat_id = chat_id.or(current_session_guard.clone()).unwrap_or_else(short_id);
// SessionManager handles session creation/reuse internally
// 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
let msg = InboundMessage { let msg = InboundMessage {
channel: self.name().to_string(), channel: channel_name,
sender_id: "cli".to_string(), sender_id: "cli".to_string(),
chat_id: chat_id.unwrap_or_else(short_id), chat_id: chat_id_part,
dialog_id: dialog_id_part,
content, content,
timestamp: crate::bus::message::current_timestamp(), timestamp: crate::bus::message::current_timestamp(),
media: Vec::new(), media: Vec::new(),

13
src/channels/feishu.rs
View File

@ -1110,6 +1110,7 @@ impl FeishuChannel {
channel: "feishu".to_string(), channel: "feishu".to_string(),
sender_id: parsed.open_id.clone(), sender_id: parsed.open_id.clone(),
chat_id: parsed.chat_id.clone(), chat_id: parsed.chat_id.clone(),
dialog_id: None, // Use default/current dialog
content: parsed.content.clone(), content: parsed.content.clone(),
timestamp: crate::bus::message::current_timestamp(), timestamp: crate::bus::message::current_timestamp(),
media: parsed.media.map(|m| vec![m]).unwrap_or_default(), media: parsed.media.map(|m| vec![m]).unwrap_or_default(),
@ -1994,18 +1995,6 @@ impl Channel for FeishuChannel {
"feishu" "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> { async fn start(&self, bus: Arc<MessageBus>) -> Result<(), ChannelError> {
if self.config.app_id.is_empty() || self.config.app_secret.is_empty() { if self.config.app_id.is_empty() || self.config.app_secret.is_empty() {
return Err(ChannelError::ConfigError( return Err(ChannelError::ConfigError(

3
src/client/mod.rs
View File

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

6
src/config/mod.rs
View File

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

45
src/gateway/mod.rs
View File

@ -20,7 +20,7 @@ pub struct GatewayState {
} }
impl GatewayState { impl GatewayState {
pub async fn new() -> Result<Self, Box<dyn std::error::Error>> { pub fn new() -> Result<Self, Box<dyn std::error::Error>> {
let config = Config::load_default()?; let config = Config::load_default()?;
// Initialize workspace directory: expand path and ensure it exists // Initialize workspace directory: expand path and ensure it exists
@ -41,24 +41,7 @@ impl GatewayState {
// Session TTL from config (default 4 hours) // Session TTL from config (default 4 hours)
let session_ttl_hours = config.gateway.session_ttl_hours.unwrap_or(4); let session_ttl_hours = config.gateway.session_ttl_hours.unwrap_or(4);
// Initialize Storage let session_manager = SessionManager::new(session_ttl_hours, provider_config.clone())?;
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) // Create CLI Chat Channel first (needed for ChannelManager)
let cli_chat_channel = Arc::new(CliChatChannel::new()); let cli_chat_channel = Arc::new(CliChatChannel::new());
@ -107,27 +90,15 @@ impl GatewayState {
&inbound.channel, &inbound.channel,
&inbound.sender_id, &inbound.sender_id,
&inbound.chat_id, &inbound.chat_id,
inbound.dialog_id.as_deref(),
&inbound.content, &inbound.content,
inbound.media, inbound.media,
).await { ).await {
Ok(crate::session::session::HandleResult::AgentResponse(content)) => { Ok(response_content) => {
let outbound = crate::bus::OutboundMessage { let outbound = crate::bus::OutboundMessage {
channel: inbound.channel.clone(), channel: inbound.channel.clone(),
chat_id: inbound.chat_id.clone(), chat_id: inbound.chat_id.clone(),
content, content: response_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, reply_to: None,
media: vec![], media: vec![],
metadata: inbound.forwarded_metadata, metadata: inbound.forwarded_metadata,
@ -189,7 +160,7 @@ impl GatewayState {
.map_err(|e| ChannelError::Other(e.to_string())) .map_err(|e| ChannelError::Other(e.to_string()))
} }
RenameDialog { session_id, title } => { RenameDialog { session_id, title } => {
session_manager.rename_dialog(&session_id, &title).await session_manager.rename_dialog(&session_id, &title)
.map(|()| SessionEvent::DialogRenamed { session_id, title }) .map(|()| SessionEvent::DialogRenamed { session_id, title })
.map_err(|e| ChannelError::Other(e.to_string())) .map_err(|e| ChannelError::Other(e.to_string()))
} }
@ -199,7 +170,7 @@ impl GatewayState {
.map_err(|e| ChannelError::Other(e.to_string())) .map_err(|e| ChannelError::Other(e.to_string()))
} }
DeleteDialog { session_id } => { DeleteDialog { session_id } => {
session_manager.delete_dialog(&session_id).await session_manager.delete_dialog(&session_id)
.map(|()| SessionEvent::DialogDeleted { session_id }) .map(|()| SessionEvent::DialogDeleted { session_id })
.map_err(|e| ChannelError::Other(e.to_string())) .map_err(|e| ChannelError::Other(e.to_string()))
} }
@ -229,7 +200,7 @@ pub async fn run(host: Option<String>, port: Option<u16>) -> Result<(), Box<dyn
logging::init_logging(); logging::init_logging();
tracing::info!("Starting PicoBot Gateway"); tracing::info!("Starting PicoBot Gateway");
let state = Arc::new(GatewayState::new().await?); let state = Arc::new(GatewayState::new()?);
// Initialize and start channels with workspace directory // Initialize and start channels with workspace directory
state.channel_manager.init( state.channel_manager.init(

2
src/lib.rs
View File

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

2
src/protocol.rs
View File

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

6
src/session/error.rs
View File

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

1042
src/session/session.rs
View File

File diff suppressed because it is too large Load Diff

9
src/skills/mod.rs
View File

@ -70,15 +70,6 @@ impl SkillsLoader {
let mut state = self.state.lock().unwrap(); let mut state = self.state.lock().unwrap();
state.loaded_skills.clear(); 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 // Load from ~/.picobot/skills
if self.picobot_skills_dir.exists() { if self.picobot_skills_dir.exists() {
let loaded = self.load_skills_from_dir(&self.picobot_skills_dir); let loaded = self.load_skills_from_dir(&self.picobot_skills_dir);

16
src/storage/error.rs
View File

@ -1,16 +0,0 @@
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
View File

@ -1,15 +0,0 @@
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,
}

1031
src/storage/mod.rs
View File

File diff suppressed because it is too large Load Diff

15
src/storage/session.rs
View File

@ -1,15 +0,0 @@
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>,
}