feat: Enhance session management with topic support
- Added topic management capabilities, allowing users to create, switch, and query topics within sessions. - Updated command structure to include new commands: SwitchSession and GetCurrentSession. - Introduced TopicRecord for managing topic data in the storage layer. - Modified session handlers to accommodate topic operations, including listing and loading topics. - Enhanced database schema to support topics, including new tables and relationships. - Updated input adapters to recognize new commands and handle topic-related actions. - Improved logging for session and topic operations to aid in debugging and monitoring.
This commit is contained in:
oudecheng
parent
025c0b5d7f
commit
2e13f6932c
272
README.md
272
README.md
@ -4,14 +4,15 @@ PicoBot 是一个用 Rust 构建的多通道 Agent 网关。它把消息接入
|
||||
|
||||
当前代码库已经实现以下核心能力：
|
||||
|
||||
- 基于 Gateway 的统一消息入口，支持 WebSocket CLI 与飞书通道
|
||||
- 基于 Gateway 的统一消息入口，支持 WebSocket CLI、飞书通道和微信通道
|
||||
- 面向工具调用的 Agent 循环，支持多轮 tool calling
|
||||
- SQLite 持久化会话、消息、长期记忆、技能事件和调度任务
|
||||
- 基于用户维度的长期记忆检索与写入机制
|
||||
- 基于 SKILL.md 的项目级 / 用户级技能加载与运行时管理
|
||||
- 基于 SKILL.md 的项目级 / 用户级 / OpenClaw 级技能加载与运行时管理
|
||||
- 定时任务系统，支持延迟、周期、绝对时间和 cron 调度
|
||||
- 超长上下文压缩与历史摘要
|
||||
- 持久化 Agent 配置文件注入与周期性重注入
|
||||
- 会话管理支持按通道查询和切换
|
||||
|
||||
## 1. 项目定位
|
||||
|
||||
@ -37,110 +38,116 @@ PicoBot 的设计目标不是“只会聊天”的单进程 Bot，而是一个
|
||||
|
||||
### 2.1 消息流转图
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
autonumber
|
||||
participant U as User / External Chat
|
||||
participant C as Channel
|
||||
participant B as MessageBus
|
||||
participant P as InboundProcessor
|
||||
participant SM as SessionManager
|
||||
participant SES as Session
|
||||
participant AES as AgentExecutionService
|
||||
participant AL as AgentLoop
|
||||
participant T as ToolRegistry
|
||||
participant DB as SQLite
|
||||
participant OD as OutboundDispatcher
|
||||
```text
|
||||
用户消息输入
|
||||
│
|
||||
▼
|
||||
┌─────────────┐
|
||||
│ Channel │ ◄── WebSocket CLI / 飞书等
|
||||
└──────┬──────┘
|
||||
│ publish_inbound
|
||||
▼
|
||||
┌─────────────┐
|
||||
│ MessageBus │ ◄── 统一消息总线
|
||||
└──────┬──────┘
|
||||
│ consume_inbound
|
||||
▼
|
||||
┌──────────────────┐
|
||||
│ InboundProcessor │ ◄── 入站处理器
|
||||
└────────┬─────────┘
|
||||
│ handle_message
|
||||
▼
|
||||
┌────────────────┐
|
||||
│ SessionManager │ ◄── 会话管理器
|
||||
└────────┬───────┘
|
||||
│ 定位/创建 Session
|
||||
▼
|
||||
┌────────────────────────┐
|
||||
│ AgentExecutionService │ ◄── 准备上下文、执行 Agent
|
||||
└────────┬───────────────┘
|
||||
│
|
||||
┌────┴────┐
|
||||
▼ ▼
|
||||
┌───────┐ ┌──────────┐
|
||||
│SQLite │ │ AgentLoop│ ◄── 调用 LLM、工具执行
|
||||
└───────┘ └────┬─────┘
|
||||
▲ │
|
||||
│ ▼
|
||||
│ ┌──────────┐
|
||||
└────┤ToolRegistry│ ◄── 工具注册表
|
||||
└──────────┘
|
||||
│
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ OutboundDispatcher │ ◄── 出站分发
|
||||
└────────┬────────┘
|
||||
│ dispatch
|
||||
▼
|
||||
┌─────────────────┐
|
||||
│ Channel │
|
||||
└────────┬────────┘
|
||||
│ 最终回复
|
||||
▼
|
||||
用户
|
||||
|
||||
U->>C: 输入消息
|
||||
C->>B: publish_inbound
|
||||
B->>P: consume_inbound
|
||||
P->>SM: handle_message(channel, sender, chat, content)
|
||||
SM->>SES: active_session(channel)
|
||||
AES->>SES: ensure_persistent_session / ensure_chat_loaded
|
||||
AES->>DB: 追加 user / system 消息
|
||||
AES->>AL: process(history)
|
||||
AL->>T: 调用工具
|
||||
T-->>AL: 返回 tool result
|
||||
AL-->>AES: emitted_messages
|
||||
AES->>DB: 按真实顺序持久化 assistant / tool / assistant
|
||||
AES->>SES: 安排后台历史压缩
|
||||
AES-->>SM: outbound messages
|
||||
SM-->>P: outbound messages
|
||||
P->>B: publish_outbound
|
||||
B->>OD: consume_outbound
|
||||
OD->>C: dispatch
|
||||
C-->>U: 最终回复
|
||||
关键步骤说明:
|
||||
1. Channel 接收外部消息 → 2. MessageBus 统一路由 → 3. InboundProcessor 处理
|
||||
4. SessionManager 定位 Session → 5. AgentExecutionService 执行
|
||||
6. 消息持久化到 SQLite → 7. AgentLoop 推理/工具调用
|
||||
8. 结果经 OutboundDispatcher 返回 Channel
|
||||
```
|
||||
|
||||
### 2.2 项目架构图
|
||||
|
||||
```mermaid
|
||||
flowchart TB
|
||||
subgraph Edge[接入层]
|
||||
CLI[CLI Client / WebSocket]
|
||||
FEI[Feishu Channel]
|
||||
HTTP[HTTP Health / WS Gateway]
|
||||
end
|
||||
```text
|
||||
接入层 (Edge)
|
||||
│
|
||||
├── CLI Client / WebSocket ──┐
|
||||
├── Feishu Channel ─────────┼──► 网关与运行时编排 (Gateway)
|
||||
└── HTTP Health / WS ────────┘ │
|
||||
├── ChannelManager ◄──┐
|
||||
├── MessageBus ────────┼── 双向通信
|
||||
├── InboundProcessor ──┘
|
||||
├── OutboundDispatcher
|
||||
├── SessionManager
|
||||
├── SessionLifecycle / Message / ScheduledTask Services
|
||||
└── AgentExecutionService ◄── 调用 Agent 执行层
|
||||
│
|
||||
Agent 执行层 (Agent) ◄─────────────────────────────────────────┘
|
||||
│
|
||||
├── AgentLoop ──► ToolRegistry (工具调用)
|
||||
├── ContextCompressor (上下文压缩)
|
||||
├── SkillRuntime (技能系统)
|
||||
└── LLM Providers (OpenAI / Anthropic)
|
||||
|
||||
subgraph Gateway[网关与运行时编排]
|
||||
CM[ChannelManager]
|
||||
BUS[MessageBus]
|
||||
IP[InboundProcessor]
|
||||
OD[OutboundDispatcher]
|
||||
SSM[SessionManager]
|
||||
SVC[SessionLifecycle / Message / ScheduledTask Services]
|
||||
AES[AgentExecutionService]
|
||||
end
|
||||
持久化与后台能力 (Runtime)
|
||||
│
|
||||
├── SessionStore / SQLite (会话/消息/记忆存储)
|
||||
├── Scheduler (定时任务调度)
|
||||
└── Memory Maintenance (记忆维护)
|
||||
|
||||
subgraph Agent[Agent 执行层]
|
||||
LOOP[AgentLoop]
|
||||
COMP[ContextCompressor]
|
||||
SK[SkillRuntime]
|
||||
TOOLS[ToolRegistry]
|
||||
PROV[LLM Providers]
|
||||
end
|
||||
|
||||
subgraph Runtime[持久化与后台能力]
|
||||
STORE[SessionStore / SQLite]
|
||||
SCH[Scheduler]
|
||||
MEM[Memory Maintenance]
|
||||
end
|
||||
|
||||
CLI --> HTTP
|
||||
FEI --> CM
|
||||
HTTP --> CM
|
||||
CM --> BUS
|
||||
BUS --> IP
|
||||
IP --> SSM
|
||||
SSM --> SVC
|
||||
SVC --> AES
|
||||
AES --> LOOP
|
||||
LOOP --> TOOLS
|
||||
LOOP --> SK
|
||||
LOOP --> PROV
|
||||
AES --> COMP
|
||||
SSM --> STORE
|
||||
AES --> STORE
|
||||
SCH --> BUS
|
||||
SCH --> SSM
|
||||
MEM --> STORE
|
||||
BUS --> OD
|
||||
OD --> CM
|
||||
数据流向:
|
||||
- 接入层 ◄──► 网关 ◄──► Agent 执行层
|
||||
- 网关 ◄──► 持久化层 (SQLite)
|
||||
- Scheduler ◄──► 总线 ◄──► SessionManager
|
||||
```
|
||||
|
||||
主要模块如下：
|
||||
|
||||
- src/gateway：网关生命周期、InboundProcessor、OutboundDispatcher、SessionManager，以及消息执行、调度任务执行、Prompt 注入、历史压缩和记忆维护编排
|
||||
- src/agent：AgentLoop、上下文压缩器、运行时配置、系统提示构建
|
||||
- src/bus：消息总线队列与消息结构定义，不包含渠道投递逻辑
|
||||
- src/agent：AgentLoop 与上下文压缩器
|
||||
- src/channels：渠道适配层，当前已有 CLI、飞书、微信通道
|
||||
- src/cli：本地 CLI 客户端、输入命令解析
|
||||
- src/client：WebSocket CLI 客户端实现
|
||||
- src/command：命令系统，包括处理器、适配器、上下文和响应处理
|
||||
- src/config：配置解析与默认值定义
|
||||
- src/domain：领域模型，包含消息和工具定义
|
||||
- src/gateway：网关生命周期、InboundProcessor、OutboundDispatcher、SessionManager，以及消息执行、调度任务执行、Prompt 注入、历史压缩和记忆维护编排
|
||||
- src/providers：不同 LLM Provider 的统一抽象，当前支持 openai 和 anthropic
|
||||
- src/tools：内置工具集合与 ToolRegistry
|
||||
- src/storage：SQLite 持久化实现
|
||||
- src/channels：渠道适配层，当前已有 CLI 与飞书通道
|
||||
- src/scheduler：数据库驱动的计划任务调度器
|
||||
- src/skills：技能发现、加载与运行时管理
|
||||
- src/client / src/cli：本地 CLI 客户端、输入命令解析与会话交互
|
||||
- src/protocol：WebSocket 入站 / 出站协议结构
|
||||
|
||||
## 3. 消息机制
|
||||
@ -379,8 +386,10 @@ PicoBot 支持基于文件系统的技能系统，用来给 Agent 注入某一
|
||||
|
||||
- 用户级技能：~/.picobot/skills/*/SKILL.md
|
||||
- 用户 Agent 级技能：~/.agents/skills/*/SKILL.md
|
||||
- 用户 OpenClaw 级技能：~/.openclaw/skills/*/SKILL.md
|
||||
- 项目级技能：.picobot/skills/*/SKILL.md
|
||||
- 项目 Agent 级技能：.agents/skills/*/SKILL.md
|
||||
- 项目 OpenClaw 级技能：.openclaw/skills/*/SKILL.md
|
||||
|
||||
### 7.2 最小 SKILL.md 格式
|
||||
|
||||
@ -459,7 +468,7 @@ skills 配置示例：
|
||||
{
|
||||
"skills": {
|
||||
"enabled": true,
|
||||
"sources": ["user", "user_agent", "project", "project_agent"],
|
||||
"sources": ["user", "user_agent", "user_openclaw", "project", "project_agent", "project_openclaw"],
|
||||
"max_index_chars": 4000,
|
||||
"max_listed_skills": 32
|
||||
}
|
||||
@ -654,6 +663,7 @@ silent_agent_task 和 agent_task 使用同一套 Agent 执行能力，但路由
|
||||
|
||||
- WebSocket CLI 客户端
|
||||
- 飞书通道
|
||||
- 微信通道
|
||||
|
||||
### 10.2 Gateway 接口
|
||||
|
||||
@ -678,13 +688,15 @@ cargo run -- agent
|
||||
|
||||
CLI 中已实现的交互命令包括：
|
||||
|
||||
- /new [title]
|
||||
- /reset
|
||||
- /sessions
|
||||
- /use <session>
|
||||
- /rename <title>
|
||||
- /archive
|
||||
- /delete
|
||||
- /new [title] - 创建新会话
|
||||
- /reset - 重置当前会话上下文
|
||||
- /sessions - 列出当前通道的所有会话（支持跨通道隔离）
|
||||
- /use <session> - 切换到指定会话
|
||||
- /rename <title> - 重命名当前会话
|
||||
- /archive - 归档当前会话
|
||||
- /delete - 删除指定会话
|
||||
- /clear - 清屏
|
||||
- /quit - 退出 CLI
|
||||
- /clear
|
||||
- /quit
|
||||
|
||||
@ -707,7 +719,8 @@ CLI 中已实现的交互命令包括：
|
||||
"base_url": "<OPENAI_BASE_URL>",
|
||||
"api_key": "<OPENAI_API_KEY>",
|
||||
"extra_headers": {},
|
||||
"llm_timeout_secs": 120
|
||||
"llm_timeout_secs": 120,
|
||||
"memory_maintenance_timeout_secs": 600
|
||||
}
|
||||
},
|
||||
"models": {
|
||||
@ -744,12 +757,12 @@ CLI 中已实现的交互命令包括：
|
||||
|
||||
常用配置项：
|
||||
|
||||
- providers：Provider 连接信息
|
||||
- providers：Provider 连接信息，包含 llm_timeout_secs（LLM 调用超时，默认 120 秒）和 memory_maintenance_timeout_secs（记忆维护超时，默认 600 秒）
|
||||
- models：模型参数，包括上下文窗口估算所用的 context_window_tokens
|
||||
- agents：Agent 级别的工具轮次、工具结果裁剪与上下文裁剪参数
|
||||
- gateway：监听地址、端口、工具结果展示、会话 TTL、Prompt 重注入策略
|
||||
- scheduler：调度器开关、worker 队列容量、误触发策略和任务列表
|
||||
- channels：飞书等通道配置
|
||||
- channels：飞书、微信等通道配置
|
||||
- skills：技能来源与索引限制
|
||||
- tools：工具启用/禁用配置（通过 disabled 列表指定禁用的工具）
|
||||
- time.timezone：时区，默认应使用 IANA 时区名，例如 Asia/Shanghai
|
||||
@ -761,6 +774,46 @@ CLI 中已实现的交互命令包括：
|
||||
1. 复制并修改 config.json，或把配置放到 ~/.picobot/config.json
|
||||
2. 配置好 Provider 的 base_url、api_key、model_id
|
||||
3. 如果要接飞书，再补充 channels.feishu 配置
|
||||
4. 如果要接微信，再补充 channels.wechat 配置
|
||||
|
||||
飞书通道配置示例：
|
||||
|
||||
```json
|
||||
{
|
||||
"channels": {
|
||||
"feishu": {
|
||||
"type": "feishu",
|
||||
"app_id": "<FEISHU_APP_ID>",
|
||||
"app_secret": "<FEISHU_APP_SECRET>",
|
||||
"enabled": true,
|
||||
"allow_from": ["*"],
|
||||
"agent": "default",
|
||||
"media_dir": "~/.picobot/media/feishu",
|
||||
"reaction_emoji": "Typing",
|
||||
"max_message_chars": 20000,
|
||||
"reply_context_max_chars": 20000
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
微信通道配置示例：
|
||||
|
||||
```json
|
||||
{
|
||||
"channels": {
|
||||
"wechat": {
|
||||
"type": "wechat",
|
||||
"enabled": true,
|
||||
"allow_from": ["*"],
|
||||
"agent": "default",
|
||||
"base_url": "https://ilinkai.weixin.qq.com",
|
||||
"cred_path": "~/.picobot/wechat/credentials.json",
|
||||
"force_login": false
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### 12.2 启动网关
|
||||
|
||||
@ -797,17 +850,23 @@ curl http://127.0.0.1:19876/health
|
||||
```text
|
||||
PicoBot/
|
||||
├── src/
|
||||
│ ├── agent/ # AgentLoop、上下文压缩
|
||||
│ ├── agent/ # AgentLoop、上下文压缩、运行时配置、系统提示
|
||||
│ ├── bus/ # 消息总线与消息结构
|
||||
│ ├── channels/ # CLI / 飞书等渠道适配
|
||||
│ ├── cli/ # CLI 输入命令
|
||||
│ ├── channels/ # CLI / 飞书 / 微信等渠道适配
|
||||
│ ├── cli/ # CLI 输入命令与通道实现
|
||||
│ ├── client/ # WebSocket CLI 客户端
|
||||
│ ├── command/ # 命令系统（处理器、适配器、上下文）
|
||||
│ ├── config/ # 配置解析
|
||||
│ ├── gateway/ # Gateway、Session 编排、WS/HTTP 控制面
|
||||
│ ├── domain/ # 领域模型（消息、工具定义）
|
||||
│ ├── gateway/ # Gateway、Session 编排、WS/HTTP 控制面、执行服务
|
||||
│ ├── logging/ # 日志配置
|
||||
│ ├── observability/ # 可观测性支持
|
||||
│ ├── platform/ # 平台抽象
|
||||
│ ├── providers/ # OpenAI / Anthropic Provider
|
||||
│ ├── scheduler/ # 定时任务系统
|
||||
│ ├── skills/ # 技能运行时
|
||||
│ ├── storage/ # SQLite 持久化
|
||||
│ ├── storage/ # SQLite 持久化（存储、端口、记录、错误）
|
||||
│ ├── text/ # 文本处理工具
|
||||
│ └── tools/ # 内置工具集合
|
||||
├── docs/
|
||||
│ ├── IMPLEMENTATION_LOG.md
|
||||
@ -824,8 +883,11 @@ PicoBot/
|
||||
|
||||
- docs/PERSISTENCE.md：持久化结构是否与代码一致
|
||||
- src/gateway/session.rs：会话状态、会话路由和运行时服务编排
|
||||
- src/gateway/execution.rs：Agent 执行服务
|
||||
- src/storage/mod.rs：SQLite schema 变更
|
||||
- src/config/mod.rs：配置项变更是否同步到 README
|
||||
- src/bus/message.rs：消息结构变更（如 OutboundMessage 新增 session_id）
|
||||
- src/command/handlers/：命令处理器实现
|
||||
|
||||
## 15. 总结
|
||||
|
||||
|
||||
@ -78,14 +78,19 @@ impl InputAdapter for ChannelInputAdapter {
|
||||
}));
|
||||
}
|
||||
|
||||
// 解析 /use 命令
|
||||
// 解析 /use 命令 - 切换会话（支持 session_id 或序号）
|
||||
if let Some(session_id) = trimmed.strip_prefix("/use ") {
|
||||
let session_id = session_id.trim();
|
||||
return Ok(Some(Command::LoadSession {
|
||||
return Ok(Some(Command::SwitchSession {
|
||||
session_id: session_id.to_string(),
|
||||
}));
|
||||
}
|
||||
|
||||
// 解析 /current 命令 - 获取当前会话信息
|
||||
if trimmed == "/current" {
|
||||
return Ok(Some(Command::GetCurrentSession));
|
||||
}
|
||||
|
||||
// 不是命令，返回 None
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
@ -79,14 +79,19 @@ impl InputAdapter for CliInputAdapter {
|
||||
}));
|
||||
}
|
||||
|
||||
// 解析 /use 命令
|
||||
// 解析 /use 命令 - 切换会话（支持 session_id 或序号）
|
||||
if let Some(session_id) = trimmed.strip_prefix("/use ") {
|
||||
let session_id = session_id.trim();
|
||||
return Ok(Some(Command::LoadSession {
|
||||
return Ok(Some(Command::SwitchSession {
|
||||
session_id: session_id.to_string(),
|
||||
}));
|
||||
}
|
||||
|
||||
// 解析 /current 命令 - 获取当前会话信息
|
||||
if trimmed == "/current" {
|
||||
return Ok(Some(Command::GetCurrentSession));
|
||||
}
|
||||
|
||||
// 不是命令，返回 None
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
@ -8,6 +8,8 @@ pub struct CommandContext {
|
||||
pub request_id: Uuid,
|
||||
/// 当前会话ID
|
||||
pub session_id: Option<String>,
|
||||
/// 当前话题ID
|
||||
pub topic_id: Option<String>,
|
||||
/// 当前聊天ID
|
||||
pub chat_id: Option<String>,
|
||||
/// 发送者ID
|
||||
@ -24,6 +26,7 @@ impl CommandContext {
|
||||
Self {
|
||||
request_id: Uuid::new_v4(),
|
||||
session_id: None,
|
||||
topic_id: None,
|
||||
chat_id: None,
|
||||
sender_id: sender_id.into(),
|
||||
channel_name: channel_name.into(),
|
||||
@ -37,6 +40,12 @@ impl CommandContext {
|
||||
self
|
||||
}
|
||||
|
||||
/// 设置话题ID
|
||||
pub fn with_topic_id(mut self, topic_id: impl Into<String>) -> Self {
|
||||
self.topic_id = Some(topic_id.into());
|
||||
self
|
||||
}
|
||||
|
||||
/// 设置聊天ID
|
||||
pub fn with_chat_id(mut self, chat_id: impl Into<String>) -> Self {
|
||||
self.chat_id = Some(chat_id.into());
|
||||
@ -55,6 +64,8 @@ impl CommandContext {
|
||||
pub struct AdapterContext {
|
||||
/// 当前会话ID
|
||||
pub session_id: Option<String>,
|
||||
/// 当前话题ID
|
||||
pub topic_id: Option<String>,
|
||||
/// 当前聊天ID
|
||||
pub chat_id: Option<String>,
|
||||
/// 发送者ID
|
||||
@ -66,6 +77,7 @@ impl AdapterContext {
|
||||
pub fn new(sender_id: impl Into<String>) -> Self {
|
||||
Self {
|
||||
session_id: None,
|
||||
topic_id: None,
|
||||
chat_id: None,
|
||||
sender_id: sender_id.into(),
|
||||
}
|
||||
@ -77,6 +89,12 @@ impl AdapterContext {
|
||||
self
|
||||
}
|
||||
|
||||
/// 设置话题ID
|
||||
pub fn with_topic_id(mut self, topic_id: impl Into<String>) -> Self {
|
||||
self.topic_id = Some(topic_id.into());
|
||||
self
|
||||
}
|
||||
|
||||
/// 设置聊天ID
|
||||
pub fn with_chat_id(mut self, chat_id: impl Into<String>) -> Self {
|
||||
self.chat_id = Some(chat_id.into());
|
||||
|
||||
@ -128,11 +128,39 @@ async fn handle_save_session(
|
||||
include_all: bool,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
tracing::debug!(
|
||||
ctx_session_id = ?ctx.session_id,
|
||||
ctx_chat_id = ?ctx.chat_id,
|
||||
channel = %ctx.channel_name,
|
||||
"SaveSession command received"
|
||||
);
|
||||
|
||||
let session_id = ctx
|
||||
.session_id
|
||||
.as_deref()
|
||||
.ok_or_else(|| CommandError::new("NO_SESSION", "No active session".to_string()))?;
|
||||
|
||||
tracing::debug!(session_id = %session_id, "Attempting to save session");
|
||||
|
||||
// 先检查会话是否存在
|
||||
match handler.store.get_session(session_id) {
|
||||
Ok(Some(record)) => {
|
||||
tracing::debug!(
|
||||
session_id = %session_id,
|
||||
title = %record.title,
|
||||
chat_id = %record.chat_id,
|
||||
message_count = record.message_count,
|
||||
"Session found for saving"
|
||||
);
|
||||
}
|
||||
Ok(None) => {
|
||||
tracing::warn!(session_id = %session_id, "Session not found in store");
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::error!(session_id = %session_id, error = %e, "Error querying session");
|
||||
}
|
||||
}
|
||||
|
||||
// 调用公共函数
|
||||
let output_path = save_session_to_file(
|
||||
session_id,
|
||||
|
||||
@ -2,23 +2,24 @@ use crate::command::context::CommandContext;
|
||||
use crate::command::handler::CommandHandler;
|
||||
use crate::command::response::{CommandError, CommandResponse, MessageKind};
|
||||
use crate::command::Command;
|
||||
use crate::gateway::cli_session::CliSessionService;
|
||||
use crate::storage::SessionStore;
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// 会话命令处理器
|
||||
///
|
||||
/// 处理与会话管理相关的命令
|
||||
pub struct SessionCommandHandler {
|
||||
cli_sessions: CliSessionService,
|
||||
store: Arc<SessionStore>,
|
||||
}
|
||||
|
||||
impl SessionCommandHandler {
|
||||
/// 创建新的会话命令处理器
|
||||
///
|
||||
/// # Arguments
|
||||
/// * `cli_sessions` - CLI 会话服务
|
||||
pub(crate) fn new(cli_sessions: CliSessionService) -> Self {
|
||||
Self { cli_sessions }
|
||||
/// * `store` - Session 存储
|
||||
pub(crate) fn new(store: Arc<SessionStore>) -> Self {
|
||||
Self { store }
|
||||
}
|
||||
}
|
||||
|
||||
@ -47,37 +48,49 @@ async fn handle_create_session(
|
||||
title: Option<String>,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
let record = handler
|
||||
.cli_sessions
|
||||
.create_with_channel(&ctx.channel_name, title.as_deref())
|
||||
.map_err(|e| CommandError::new("CREATE_SESSION_ERROR", e.to_string()))?;
|
||||
// 获取当前 session_id，如果没有则报错
|
||||
let session_id = ctx.session_id.as_deref()
|
||||
.ok_or_else(|| CommandError::new("NO_SESSION", "No active session. Please ensure a session exists first."))?;
|
||||
|
||||
// 创建新话题（在同一个 Session 内）
|
||||
let topic_title = title.unwrap_or_else(|| {
|
||||
format!("Topic {}", &uuid::Uuid::new_v4().to_string()[..8])
|
||||
});
|
||||
|
||||
let topic = handler
|
||||
.store
|
||||
.create_topic(session_id, &topic_title, None)
|
||||
.map_err(|e| CommandError::new("CREATE_TOPIC_ERROR", e.to_string()))?;
|
||||
|
||||
Ok(CommandResponse::success(ctx.request_id)
|
||||
.with_message(MessageKind::Notification, &record.title)
|
||||
.with_metadata("session_id", &record.id)
|
||||
.with_metadata("channel_name", &record.channel_name)
|
||||
.with_metadata("message_count", &record.message_count.to_string()))
|
||||
.with_message(MessageKind::Notification, &topic.title)
|
||||
.with_metadata("topic_id", &topic.id)
|
||||
.with_metadata("session_id", &topic.session_id)
|
||||
.with_metadata("message_count", &topic.message_count.to_string()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::command::response::MessageKind;
|
||||
use crate::storage::{SessionRecord, SessionStore};
|
||||
use crate::storage::SessionStore;
|
||||
use std::sync::Arc;
|
||||
|
||||
fn create_test_service() -> CliSessionService {
|
||||
fn create_test_handler() -> SessionCommandHandler {
|
||||
let store = Arc::new(SessionStore::in_memory().unwrap());
|
||||
CliSessionService::new(store)
|
||||
SessionCommandHandler::new(store)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_session_with_title() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionCommandHandler::new(service);
|
||||
let ctx = CommandContext::new("test", "test");
|
||||
let handler = create_test_handler();
|
||||
// 需要先创建一个 session
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test session")).unwrap();
|
||||
|
||||
let ctx = CommandContext::new("test", "cli")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::CreateSession {
|
||||
title: Some("my session".to_string()),
|
||||
title: Some("my topic".to_string()),
|
||||
};
|
||||
|
||||
let result = handler.handle(cmd, ctx).await;
|
||||
@ -85,16 +98,17 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
let resp = result.unwrap();
|
||||
assert!(resp.success);
|
||||
assert_eq!(resp.messages.len(), 1);
|
||||
assert_eq!(resp.messages[0].content, "my session");
|
||||
assert!(resp.metadata.contains_key("session_id"));
|
||||
assert!(resp.metadata.contains_key("topic_id"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_create_session_without_title() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionCommandHandler::new(service);
|
||||
let ctx = CommandContext::new("test", "test");
|
||||
let handler = create_test_handler();
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test session")).unwrap();
|
||||
|
||||
let ctx = CommandContext::new("test", "cli")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::CreateSession { title: None };
|
||||
|
||||
let result = handler.handle(cmd, ctx).await;
|
||||
@ -102,15 +116,11 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
let resp = result.unwrap();
|
||||
assert!(resp.success);
|
||||
assert_eq!(resp.messages.len(), 1);
|
||||
// 自动生成的标题
|
||||
assert!(!resp.messages[0].content.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_can_handle() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionCommandHandler::new(service);
|
||||
let handler = create_test_handler();
|
||||
|
||||
assert!(handler.can_handle(&Command::CreateSession { title: None }));
|
||||
}
|
||||
|
||||
@ -2,28 +2,39 @@ use crate::command::context::CommandContext;
|
||||
use crate::command::handler::CommandHandler;
|
||||
use crate::command::response::{CommandError, CommandResponse, MessageKind};
|
||||
use crate::command::Command;
|
||||
use crate::gateway::cli_session::CliSessionService;
|
||||
use crate::protocol::SessionSummary;
|
||||
use crate::gateway::session::SessionManager;
|
||||
use crate::storage::{SessionStore, TopicRecord};
|
||||
use async_trait::async_trait;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// 会话查询命令处理器
|
||||
///
|
||||
/// 处理 ListSessions 和 LoadSession 命令
|
||||
/// 处理 ListSessions、LoadSession 和 SwitchSession 命令（现在操作 Topic）
|
||||
pub struct SessionQueryCommandHandler {
|
||||
cli_sessions: CliSessionService,
|
||||
store: Arc<SessionStore>,
|
||||
session_manager: Option<SessionManager>,
|
||||
}
|
||||
|
||||
impl SessionQueryCommandHandler {
|
||||
/// 创建新的会话查询命令处理器
|
||||
pub fn new(cli_sessions: CliSessionService) -> Self {
|
||||
Self { cli_sessions }
|
||||
pub fn new(store: Arc<SessionStore>) -> Self {
|
||||
Self {
|
||||
store,
|
||||
session_manager: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// 设置 SessionManager（用于 SwitchSession 命令）
|
||||
pub fn with_session_manager(mut self, session_manager: SessionManager) -> Self {
|
||||
self.session_manager = Some(session_manager);
|
||||
self
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl CommandHandler for SessionQueryCommandHandler {
|
||||
fn can_handle(&self, cmd: &Command) -> bool {
|
||||
matches!(cmd, Command::ListSessions { .. } | Command::LoadSession { .. })
|
||||
matches!(cmd, Command::ListSessions { .. } | Command::LoadSession { .. } | Command::SwitchSession { .. } | Command::GetCurrentSession)
|
||||
}
|
||||
|
||||
async fn handle(
|
||||
@ -38,84 +49,217 @@ impl CommandHandler for SessionQueryCommandHandler {
|
||||
Command::LoadSession { session_id } => {
|
||||
handle_load_session(self, session_id, ctx).await
|
||||
}
|
||||
Command::SwitchSession { session_id } => {
|
||||
handle_switch_session(self, session_id, ctx).await
|
||||
}
|
||||
Command::GetCurrentSession => {
|
||||
handle_get_current_session(self, ctx).await
|
||||
}
|
||||
_ => unreachable!(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 处理列出会话命令
|
||||
/// 处理列出话题命令
|
||||
async fn handle_list_sessions(
|
||||
handler: &SessionQueryCommandHandler,
|
||||
include_archived: bool,
|
||||
_include_archived: bool,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
// 使用当前通道名称查询会话，而不是硬编码 "cli"
|
||||
let channel_name = &ctx.channel_name;
|
||||
let records = handler
|
||||
.cli_sessions
|
||||
.list_by_channel(channel_name, include_archived)
|
||||
.map_err(|e| CommandError::new("LIST_SESSIONS_ERROR", e.to_string()))?;
|
||||
// 获取当前 session_id
|
||||
let session_id = ctx.session_id.as_deref()
|
||||
.ok_or_else(|| CommandError::new("NO_SESSION", "No active session"))?;
|
||||
|
||||
let summaries: Vec<SessionSummary> = records
|
||||
.into_iter()
|
||||
.map(|r| SessionSummary {
|
||||
session_id: r.id,
|
||||
title: r.title,
|
||||
channel_name: r.channel_name,
|
||||
chat_id: r.chat_id,
|
||||
message_count: r.message_count,
|
||||
last_active_at: r.last_active_at,
|
||||
archived_at: r.archived_at,
|
||||
})
|
||||
.collect();
|
||||
// 查询该 session 的所有 topic
|
||||
let topics = handler
|
||||
.store
|
||||
.list_topics(session_id)
|
||||
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
|
||||
|
||||
// 将会话列表序列化为 JSON 存储在 metadata 中
|
||||
let sessions_json =
|
||||
serde_json::to_string(&summaries).map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;
|
||||
// 获取当前 topic ID
|
||||
let current_topic_id = ctx.topic_id.as_deref().unwrap_or("");
|
||||
|
||||
// 构建可读的会话列表消息
|
||||
let message = if summaries.is_empty() {
|
||||
"No sessions found.".to_string()
|
||||
// 构建表格格式的话题列表消息
|
||||
let message = if topics.is_empty() {
|
||||
"No topics found. Use /new <title> to create a topic.".to_string()
|
||||
} else {
|
||||
let mut lines = vec![format!("Found {} session(s):", summaries.len())];
|
||||
for summary in &summaries {
|
||||
let archived_info = summary
|
||||
.archived_at
|
||||
.map(|_| " [archived]")
|
||||
.unwrap_or("");
|
||||
let mut lines = vec![format!("Found {} topic(s):", topics.len())];
|
||||
lines.push(String::new());
|
||||
|
||||
// 表格头部
|
||||
lines.push("┌────┬─────────────────┬──────────────────────┬──────────┬─────────────────┐".to_string());
|
||||
lines.push("│ No │ Topic ID │ Title │ Messages │ Last Active │".to_string());
|
||||
lines.push("├────┼─────────────────┼──────────────────────┼──────────┼─────────────────┤".to_string());
|
||||
|
||||
// 表格内容
|
||||
for (idx, topic) in topics.iter().enumerate() {
|
||||
let row_num = idx + 1;
|
||||
let is_current = topic.id == current_topic_id;
|
||||
let num_marker = if is_current { " * ".to_string() } else { format!(" {:<2}", row_num) };
|
||||
|
||||
// 截断过长的字段
|
||||
let topic_id_display = if topic.id.len() > 15 {
|
||||
format!("{}...", &topic.id[..12])
|
||||
} else {
|
||||
topic.id.clone()
|
||||
};
|
||||
let title_display = if topic.title.len() > 20 {
|
||||
format!("{}...", &topic.title[..17])
|
||||
} else {
|
||||
topic.title.clone()
|
||||
};
|
||||
|
||||
let last_active = format_time_ago(topic.last_active_at);
|
||||
|
||||
lines.push(format!(
|
||||
" - {}: {}{}",
|
||||
summary.session_id, summary.title, archived_info
|
||||
"│{}│ {:<15} │ {:<20} │ {:<8} │ {:<15} │",
|
||||
num_marker,
|
||||
topic_id_display,
|
||||
title_display,
|
||||
topic.message_count,
|
||||
last_active
|
||||
));
|
||||
}
|
||||
lines.push("".to_string());
|
||||
lines.push("Use /use <session_id> to switch to a session".to_string());
|
||||
|
||||
// 表格底部
|
||||
lines.push("└────┴─────────────────┴──────────────────────┴──────────┴─────────────────┘".to_string());
|
||||
lines.push(String::new());
|
||||
lines.push("* = current topic".to_string());
|
||||
lines.push("Use /use <number> or /use <topic_id> to switch".to_string());
|
||||
|
||||
lines.join("\n")
|
||||
};
|
||||
|
||||
let topics_json = serde_json::to_string(&topics)
|
||||
.map_err(|e| CommandError::new("SERIALIZE_ERROR", e.to_string()))?;
|
||||
|
||||
Ok(CommandResponse::success(ctx.request_id)
|
||||
.with_message(MessageKind::Notification, &message)
|
||||
.with_metadata("sessions", &sessions_json)
|
||||
.with_metadata("count", &summaries.len().to_string()))
|
||||
.with_metadata("topics", &topics_json)
|
||||
.with_metadata("count", &topics.len().to_string())
|
||||
.with_metadata("current_topic_id", current_topic_id))
|
||||
}
|
||||
|
||||
/// 处理加载会话命令
|
||||
/// 处理加载话题命令
|
||||
async fn handle_load_session(
|
||||
handler: &SessionQueryCommandHandler,
|
||||
session_id: String,
|
||||
topic_id: String,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
let record = handler
|
||||
.cli_sessions
|
||||
.get(&session_id)
|
||||
.map_err(|e| CommandError::new("LOAD_SESSION_ERROR", e.to_string()))?
|
||||
.ok_or_else(|| CommandError::new("SESSION_NOT_FOUND", format!("Session not found: {}", session_id)))?;
|
||||
let topic = handler
|
||||
.store
|
||||
.get_topic(&topic_id)
|
||||
.map_err(|e| CommandError::new("LOAD_TOPIC_ERROR", e.to_string()))?
|
||||
.ok_or_else(|| CommandError::new("TOPIC_NOT_FOUND", format!("Topic not found: {}", topic_id)))?;
|
||||
|
||||
Ok(CommandResponse::success(ctx.request_id)
|
||||
.with_message(MessageKind::Notification, &record.title)
|
||||
.with_metadata("session_id", &record.id)
|
||||
.with_metadata("title", &record.title)
|
||||
.with_metadata("message_count", &record.message_count.to_string()))
|
||||
.with_message(MessageKind::Notification, &topic.title)
|
||||
.with_metadata("topic_id", &topic.id)
|
||||
.with_metadata("title", &topic.title)
|
||||
.with_metadata("message_count", &topic.message_count.to_string()))
|
||||
}
|
||||
|
||||
/// 格式化时间为相对时间（如 "2 mins ago"）
|
||||
fn format_time_ago(timestamp_ms: i64) -> String {
|
||||
let now = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.unwrap()
|
||||
.as_millis() as i64;
|
||||
|
||||
let diff_ms = now - timestamp_ms;
|
||||
let diff_secs = diff_ms / 1000;
|
||||
|
||||
if diff_secs < 60 {
|
||||
"just now".to_string()
|
||||
} else if diff_secs < 3600 {
|
||||
format!("{} mins ago", diff_secs / 60)
|
||||
} else if diff_secs < 86400 {
|
||||
format!("{} hours ago", diff_secs / 3600)
|
||||
} else {
|
||||
format!("{} days ago", diff_secs / 86400)
|
||||
}
|
||||
}
|
||||
|
||||
/// 处理获取当前话题命令
|
||||
async fn handle_get_current_session(
|
||||
handler: &SessionQueryCommandHandler,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
let topic_id = ctx.topic_id.as_deref()
|
||||
.ok_or_else(|| CommandError::new("NO_CURRENT_TOPIC", "No current topic"))?;
|
||||
|
||||
let topic = handler
|
||||
.store
|
||||
.get_topic(topic_id)
|
||||
.map_err(|e| CommandError::new("GET_TOPIC_ERROR", e.to_string()))?
|
||||
.ok_or_else(|| CommandError::new("TOPIC_NOT_FOUND", format!("Topic not found: {}", topic_id)))?;
|
||||
|
||||
let last_active = format_time_ago(topic.last_active_at);
|
||||
let created_at = format_time_ago(topic.created_at);
|
||||
|
||||
let message = format!(
|
||||
"Current Topic:\n\n Topic ID: {}\n Title: {}\n Messages: {}\n Created: {}\n Last Active: {}",
|
||||
topic.id,
|
||||
topic.title,
|
||||
topic.message_count,
|
||||
created_at,
|
||||
last_active
|
||||
);
|
||||
|
||||
Ok(CommandResponse::success(ctx.request_id)
|
||||
.with_message(MessageKind::Notification, &message)
|
||||
.with_metadata("topic_id", &topic.id)
|
||||
.with_metadata("title", &topic.title)
|
||||
.with_metadata("message_count", &topic.message_count.to_string()))
|
||||
}
|
||||
|
||||
/// 处理切换话题命令
|
||||
async fn handle_switch_session(
|
||||
handler: &SessionQueryCommandHandler,
|
||||
topic_id: String,
|
||||
ctx: CommandContext,
|
||||
) -> Result<CommandResponse, CommandError> {
|
||||
// 获取当前 session_id
|
||||
let session_id = ctx.session_id.as_deref()
|
||||
.ok_or_else(|| CommandError::new("NO_SESSION", "No active session"))?;
|
||||
|
||||
// 尝试解析为序号
|
||||
let target_topic_id = if let Ok(index) = topic_id.parse::<usize>() {
|
||||
let topics = handler
|
||||
.store
|
||||
.list_topics(session_id)
|
||||
.map_err(|e| CommandError::new("LIST_TOPICS_ERROR", e.to_string()))?;
|
||||
|
||||
let index = index.saturating_sub(1);
|
||||
if index >= topics.len() {
|
||||
return Err(CommandError::new(
|
||||
"INVALID_TOPIC_INDEX",
|
||||
format!("Topic index {} is out of range (1-{})", index + 1, topics.len())
|
||||
));
|
||||
}
|
||||
topics[index].id.clone()
|
||||
} else {
|
||||
topic_id
|
||||
};
|
||||
|
||||
// 验证目标话题存在
|
||||
let topic = handler
|
||||
.store
|
||||
.get_topic(&target_topic_id)
|
||||
.map_err(|e| CommandError::new("SWITCH_TOPIC_ERROR", e.to_string()))?
|
||||
.ok_or_else(|| CommandError::new("TOPIC_NOT_FOUND", format!("Topic not found: {}", target_topic_id)))?;
|
||||
|
||||
// 返回切换成功响应
|
||||
let message = format!(
|
||||
"✓ Switched to topic: {} ({} messages)",
|
||||
topic.title, topic.message_count
|
||||
);
|
||||
|
||||
Ok(CommandResponse::success(ctx.request_id)
|
||||
.with_message(MessageKind::Notification, &message)
|
||||
.with_metadata("topic_id", &topic.id)
|
||||
.with_metadata("title", &topic.title)
|
||||
.with_metadata("message_count", &topic.message_count.to_string()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@ -124,17 +268,20 @@ mod tests {
|
||||
use crate::storage::SessionStore;
|
||||
use std::sync::Arc;
|
||||
|
||||
fn create_test_service() -> CliSessionService {
|
||||
fn create_test_handler() -> SessionQueryCommandHandler {
|
||||
let store = Arc::new(SessionStore::in_memory().unwrap());
|
||||
CliSessionService::new(store)
|
||||
SessionQueryCommandHandler::new(store)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_sessions_empty() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionQueryCommandHandler::new(service);
|
||||
// 使用 "cli" 通道，与 CliSessionService::create 一致
|
||||
let ctx = CommandContext::new("test", "cli");
|
||||
let handler = create_test_handler();
|
||||
// 需要先创建一个 session 和 topic
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test")).unwrap();
|
||||
|
||||
let ctx = CommandContext::new("test", "cli")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::ListSessions {
|
||||
include_archived: false,
|
||||
};
|
||||
@ -144,19 +291,20 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
let resp = result.unwrap();
|
||||
assert!(resp.success);
|
||||
assert!(resp.messages[0].content.contains("No sessions"));
|
||||
assert!(resp.messages[0].content.contains("No topics"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_list_sessions_with_items() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionQueryCommandHandler::new(service.clone());
|
||||
let handler = create_test_handler();
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test")).unwrap();
|
||||
|
||||
// 创建一些会话（使用 cli 通道）
|
||||
service.create(Some("test session")).unwrap();
|
||||
// 创建一个 topic
|
||||
store.create_topic(&session.id, "Test Topic", None).unwrap();
|
||||
|
||||
// 使用 "cli" 通道查询，与创建会话的通道一致
|
||||
let ctx = CommandContext::new("test", "cli");
|
||||
let ctx = CommandContext::new("test", "cli")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::ListSessions {
|
||||
include_archived: false,
|
||||
};
|
||||
@ -166,14 +314,17 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
let resp = result.unwrap();
|
||||
assert!(resp.success);
|
||||
assert!(resp.metadata.contains_key("sessions"));
|
||||
assert!(resp.metadata.contains_key("topics"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_load_session_not_found() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionQueryCommandHandler::new(service);
|
||||
let ctx = CommandContext::new("test", "test");
|
||||
let handler = create_test_handler();
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test")).unwrap();
|
||||
|
||||
let ctx = CommandContext::new("test", "test")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::LoadSession {
|
||||
session_id: "nonexistent".to_string(),
|
||||
};
|
||||
@ -185,15 +336,15 @@ mod tests {
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_load_session_success() {
|
||||
let service = create_test_service();
|
||||
let handler = SessionQueryCommandHandler::new(service.clone());
|
||||
let handler = create_test_handler();
|
||||
let store = handler.store.clone();
|
||||
let session = store.create_session("cli", Some("test")).unwrap();
|
||||
let topic = store.create_topic(&session.id, "Test Topic", None).unwrap();
|
||||
|
||||
// 创建会话
|
||||
let record = service.create(Some("test session")).unwrap();
|
||||
|
||||
let ctx = CommandContext::new("test", "test");
|
||||
let ctx = CommandContext::new("test", "test")
|
||||
.with_session_id(&session.id);
|
||||
let cmd = Command::LoadSession {
|
||||
session_id: record.id.clone(),
|
||||
session_id: topic.id.clone(),
|
||||
};
|
||||
|
||||
let result = handler.handle(cmd, ctx).await;
|
||||
@ -201,6 +352,6 @@ mod tests {
|
||||
assert!(result.is_ok());
|
||||
let resp = result.unwrap();
|
||||
assert!(resp.success);
|
||||
assert_eq!(resp.metadata.get("session_id").unwrap(), &record.id);
|
||||
assert_eq!(resp.metadata.get("topic_id").unwrap(), &topic.id);
|
||||
}
|
||||
}
|
||||
|
||||
@ -11,17 +11,21 @@ use serde::{Deserialize, Serialize};
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
#[serde(tag = "type", rename_all = "snake_case")]
|
||||
pub enum Command {
|
||||
/// 创建新会话
|
||||
/// 创建新话题（在同一个 Session 内）
|
||||
CreateSession { title: Option<String> },
|
||||
/// 保存会话内容到 Markdown 文件
|
||||
/// 保存话题内容到 Markdown 文件
|
||||
SaveSession {
|
||||
filepath: Option<String>,
|
||||
include_all: bool,
|
||||
},
|
||||
/// 列出会话
|
||||
/// 列出当前 Session 的所有话题
|
||||
ListSessions { include_archived: bool },
|
||||
/// 加载指定会话
|
||||
/// 加载指定话题
|
||||
LoadSession { session_id: String },
|
||||
/// 切换到指定话题（清理当前历史并加载新话题）
|
||||
SwitchSession { session_id: String },
|
||||
/// 获取当前话题信息
|
||||
GetCurrentSession,
|
||||
}
|
||||
|
||||
impl Command {
|
||||
@ -32,6 +36,8 @@ impl Command {
|
||||
Command::SaveSession { .. } => "save_session",
|
||||
Command::ListSessions { .. } => "list_sessions",
|
||||
Command::LoadSession { .. } => "load_session",
|
||||
Command::SwitchSession { .. } => "switch_session",
|
||||
Command::GetCurrentSession => "get_current_session",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -14,6 +14,7 @@ use crate::command::Command;
|
||||
use crate::config::LLMProviderConfig;
|
||||
use crate::gateway::agent_prompt_provider::AgentPromptProvider;
|
||||
use crate::skills::SkillPromptProvider;
|
||||
use crate::storage::persistent_session_id;
|
||||
|
||||
use super::session::{BusToolCallEmitter, SessionManager};
|
||||
|
||||
@ -35,13 +36,15 @@ impl InboundProcessor {
|
||||
) -> Self {
|
||||
// 创建命令路由器并注册处理器
|
||||
let mut command_router = CommandRouter::new();
|
||||
let store = session_manager.store();
|
||||
|
||||
// 注册 Session 处理器
|
||||
let cli_sessions = session_manager.cli_sessions();
|
||||
command_router.register(Box::new(SessionCommandHandler::new(cli_sessions.clone())));
|
||||
command_router.register(Box::new(SessionCommandHandler::new(store.clone())));
|
||||
|
||||
// 注册 session_query 处理器
|
||||
command_router.register(Box::new(SessionQueryCommandHandler::new(cli_sessions)));
|
||||
let session_query_handler = SessionQueryCommandHandler::new(store)
|
||||
.with_session_manager(session_manager.clone());
|
||||
command_router.register(Box::new(session_query_handler));
|
||||
|
||||
// 注册 save_session 处理器
|
||||
let store = session_manager.store();
|
||||
@ -115,15 +118,19 @@ impl InboundProcessor {
|
||||
}
|
||||
|
||||
async fn process_one(&self, inbound: InboundMessage) -> Result<(), AgentError> {
|
||||
// 计算正确的 session_id（根据 channel_name 和 chat_id）
|
||||
let session_id = persistent_session_id(&inbound.channel, &inbound.chat_id);
|
||||
|
||||
// 使用 ChannelInputAdapter 尝试解析命令
|
||||
let adapter = ChannelInputAdapter::new();
|
||||
let ctx = crate::command::context::AdapterContext::new(&inbound.channel)
|
||||
.with_session_id(&inbound.chat_id);
|
||||
.with_session_id(&session_id);
|
||||
|
||||
if let Ok(Some(cmd)) = adapter.try_parse(&inbound.content, ctx) {
|
||||
// 使用命令路由器处理
|
||||
let cmd_ctx = crate::command::context::CommandContext::new(&inbound.channel, &inbound.channel)
|
||||
.with_session_id(&inbound.chat_id);
|
||||
.with_session_id(&session_id)
|
||||
.with_chat_id(&inbound.chat_id);
|
||||
|
||||
// 记录是否是创建会话命令（用于后续处理）
|
||||
let _is_create_session = matches!(cmd, Command::CreateSession { .. });
|
||||
|
||||
@ -221,16 +221,24 @@ async fn handle_inbound(
|
||||
]));
|
||||
|
||||
let mut router = CommandRouter::new();
|
||||
router.register(Box::new(SessionCommandHandler::new(cli_sessions.clone())));
|
||||
router.register(Box::new(SessionQueryCommandHandler::new(cli_sessions)));
|
||||
router.register(Box::new(SessionCommandHandler::new(store.clone())));
|
||||
// 修复：添加 SessionManager 到 SessionQueryCommandHandler
|
||||
let session_query_handler = SessionQueryCommandHandler::new(store.clone())
|
||||
.with_session_manager(state.session_manager.clone());
|
||||
router.register(Box::new(session_query_handler));
|
||||
router.register(Box::new(SaveSessionCommandHandler::new(
|
||||
store,
|
||||
system_prompt_provider,
|
||||
)));
|
||||
|
||||
// 构建命令上下文
|
||||
tracing::debug!(
|
||||
current_session_id = %current_session_id,
|
||||
"Building CommandContext for WebSocket command"
|
||||
);
|
||||
let cmd_ctx = CommandContext::new("websocket", "cli")
|
||||
.with_session_id(current_session_id.as_str());
|
||||
.with_session_id(current_session_id.as_str())
|
||||
.with_chat_id(current_session_id.as_str());
|
||||
|
||||
// 执行命令
|
||||
let response = router.dispatch_with_response(cmd, cmd_ctx).await;
|
||||
@ -239,6 +247,11 @@ async fn handle_inbound(
|
||||
if response.success {
|
||||
// 更新当前会话 ID（如果是创建会话）
|
||||
if let Some(session_id) = response.metadata.get("session_id") {
|
||||
tracing::info!(
|
||||
old_session_id = %current_session_id,
|
||||
new_session_id = %session_id,
|
||||
"Updating current_session_id"
|
||||
);
|
||||
*current_session_id = session_id.clone();
|
||||
state
|
||||
.channel_manager
|
||||
@ -250,6 +263,12 @@ async fn handle_inbound(
|
||||
)
|
||||
.await;
|
||||
}
|
||||
} else if let Some(ref error) = response.error {
|
||||
tracing::warn!(
|
||||
error_code = %error.code,
|
||||
error_message = %error.message,
|
||||
"Command failed"
|
||||
);
|
||||
}
|
||||
|
||||
// 适配并发送响应
|
||||
|
||||
@ -17,7 +17,7 @@ pub use ports::{
|
||||
};
|
||||
pub use records::{
|
||||
MemoryRecord, MemoryUpsert, SchedulerJobRecord, SchedulerJobState, SchedulerJobStatus,
|
||||
SchedulerJobUpsert, SessionRecord, SkillEventRecord,
|
||||
SchedulerJobUpsert, SessionRecord, SkillEventRecord, TopicRecord,
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
@ -79,6 +79,7 @@ impl SessionStore {
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
topic_id TEXT,
|
||||
seq INTEGER NOT NULL,
|
||||
role TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
@ -90,6 +91,7 @@ impl SessionStore {
|
||||
tool_calls_json TEXT,
|
||||
created_at INTEGER NOT NULL,
|
||||
FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE,
|
||||
FOREIGN KEY(topic_id) REFERENCES topics(id) ON DELETE SET NULL,
|
||||
UNIQUE(session_id, seq)
|
||||
);
|
||||
|
||||
@ -98,6 +100,21 @@ impl SessionStore {
|
||||
CREATE INDEX IF NOT EXISTS idx_messages_session_created
|
||||
ON messages(session_id, created_at);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS topics (
|
||||
id TEXT PRIMARY KEY,
|
||||
session_id TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
description TEXT,
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL,
|
||||
last_active_at INTEGER NOT NULL,
|
||||
message_count INTEGER NOT NULL DEFAULT 0,
|
||||
FOREIGN KEY(session_id) REFERENCES sessions(id) ON DELETE CASCADE
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_topics_session
|
||||
ON topics(session_id, last_active_at DESC);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS skill_events (
|
||||
id TEXT PRIMARY KEY,
|
||||
session_id TEXT,
|
||||
@ -211,6 +228,8 @@ impl SessionStore {
|
||||
) -> Result<SessionRecord, StorageError> {
|
||||
let now = current_timestamp();
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
// 统一使用 persistent_session_id 格式
|
||||
let session_id = persistent_session_id(channel_name, &id);
|
||||
let title = title
|
||||
.map(str::trim)
|
||||
.filter(|value| !value.is_empty())
|
||||
@ -232,11 +251,11 @@ impl SessionStore {
|
||||
reset_cutoff_seq, user_turn_count, agent_prompt_reinjection_count
|
||||
) VALUES (?1, ?2, ?3, ?4, NULL, ?5, ?5, ?5, NULL, NULL, 0, 0, 0, 0)
|
||||
",
|
||||
params![id, title, channel_name, id, now],
|
||||
params![&session_id, title, channel_name, id, now],
|
||||
)?;
|
||||
|
||||
drop(conn);
|
||||
self.get_session(&id)?
|
||||
self.get_session(&session_id)?
|
||||
.ok_or_else(|| rusqlite::Error::QueryReturnedNoRows.into())
|
||||
}
|
||||
|
||||
@ -354,6 +373,103 @@ impl SessionStore {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ==================== Topic Methods ====================
|
||||
|
||||
pub fn create_topic(
|
||||
&self,
|
||||
session_id: &str,
|
||||
title: &str,
|
||||
description: Option<&str>,
|
||||
) -> Result<TopicRecord, StorageError> {
|
||||
let now = current_timestamp();
|
||||
let id = format!("topic:{}", uuid::Uuid::new_v4());
|
||||
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
conn.execute(
|
||||
"INSERT INTO topics (id, session_id, title, description, created_at, updated_at, last_active_at, message_count) VALUES (?1, ?2, ?3, ?4, ?5, ?5, ?5, 0)",
|
||||
params![&id, session_id, title, description.unwrap_or(""), now],
|
||||
)?;
|
||||
drop(conn);
|
||||
|
||||
self.get_topic(&id)?
|
||||
.ok_or_else(|| rusqlite::Error::QueryReturnedNoRows.into())
|
||||
}
|
||||
|
||||
pub fn get_topic(&self, topic_id: &str) -> Result<Option<TopicRecord>, StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count FROM topics WHERE id = ?1",
|
||||
)?;
|
||||
|
||||
stmt.query_row(params![topic_id], |row| {
|
||||
Ok(TopicRecord {
|
||||
id: row.get(0)?,
|
||||
session_id: row.get(1)?,
|
||||
title: row.get(2)?,
|
||||
description: row.get(3)?,
|
||||
created_at: row.get(4)?,
|
||||
updated_at: row.get(5)?,
|
||||
last_active_at: row.get(6)?,
|
||||
message_count: row.get(7)?,
|
||||
})
|
||||
})
|
||||
.optional()
|
||||
.map_err(StorageError::from)
|
||||
}
|
||||
|
||||
pub fn list_topics(&self, session_id: &str) -> Result<Vec<TopicRecord>, StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
let mut stmt = conn.prepare(
|
||||
"SELECT id, session_id, title, description, created_at, updated_at, last_active_at, message_count FROM topics WHERE session_id = ?1 ORDER BY last_active_at DESC"
|
||||
)?;
|
||||
|
||||
let rows = stmt.query_map(params![session_id], |row| {
|
||||
Ok(TopicRecord {
|
||||
id: row.get(0)?,
|
||||
session_id: row.get(1)?,
|
||||
title: row.get(2)?,
|
||||
description: row.get(3)?,
|
||||
created_at: row.get(4)?,
|
||||
updated_at: row.get(5)?,
|
||||
last_active_at: row.get(6)?,
|
||||
message_count: row.get(7)?,
|
||||
})
|
||||
})?;
|
||||
|
||||
let mut topics = Vec::new();
|
||||
for row in rows {
|
||||
topics.push(row?);
|
||||
}
|
||||
Ok(topics)
|
||||
}
|
||||
|
||||
pub fn update_topic_title(&self, topic_id: &str, title: &str) -> Result<(), StorageError> {
|
||||
let now = current_timestamp();
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
conn.execute(
|
||||
"UPDATE topics SET title = ?2, updated_at = ?3 WHERE id = ?1",
|
||||
params![topic_id, title.trim(), now],
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn delete_topic(&self, topic_id: &str) -> Result<(), StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
// Messages 的 topic_id 会被设为 NULL（ON DELETE SET NULL）
|
||||
conn.execute("DELETE FROM topics WHERE id = ?1", params![topic_id])?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn touch_topic(&self, topic_id: &str) -> Result<(), StorageError> {
|
||||
let now = current_timestamp();
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
conn.execute(
|
||||
"UPDATE topics SET last_active_at = ?2 WHERE id = ?1",
|
||||
params![topic_id, now],
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn clear_messages(&self, session_id: &str) -> Result<(), StorageError> {
|
||||
let now = current_timestamp();
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
@ -410,6 +526,15 @@ impl SessionStore {
|
||||
&self,
|
||||
session_id: &str,
|
||||
message: &ChatMessage,
|
||||
) -> Result<(), StorageError> {
|
||||
self.append_message_with_topic(session_id, None, message)
|
||||
}
|
||||
|
||||
pub fn append_message_with_topic(
|
||||
&self,
|
||||
session_id: &str,
|
||||
topic_id: Option<&str>,
|
||||
message: &ChatMessage,
|
||||
) -> Result<(), StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
let tx = conn.unchecked_transaction()?;
|
||||
@ -429,13 +554,14 @@ impl SessionStore {
|
||||
tx.execute(
|
||||
"
|
||||
INSERT INTO messages (
|
||||
id, session_id, seq, role, content,
|
||||
id, session_id, topic_id, seq, role, content,
|
||||
system_context, reasoning_content, media_refs_json, tool_call_id, tool_name, tool_calls_json, created_at
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12)
|
||||
) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13)
|
||||
",
|
||||
params![
|
||||
message.id,
|
||||
session_id,
|
||||
topic_id,
|
||||
seq,
|
||||
message.role,
|
||||
message.content,
|
||||
@ -1177,6 +1303,62 @@ impl SessionStore {
|
||||
load_messages_after(&conn, session_id, cutoff_seq)
|
||||
}
|
||||
|
||||
pub fn load_messages_for_topic(&self, topic_id: &str) -> Result<Vec<ChatMessage>, StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
let mut stmt = conn.prepare(
|
||||
"
|
||||
SELECT id, role, content, system_context, reasoning_content, media_refs_json, created_at, tool_call_id, tool_name, tool_calls_json
|
||||
FROM messages
|
||||
WHERE topic_id = ?1
|
||||
ORDER BY seq ASC
|
||||
",
|
||||
)?;
|
||||
|
||||
let rows = stmt.query_map(params![topic_id], |row| {
|
||||
let media_refs_json: String = row.get(5)?;
|
||||
let media_refs: Vec<String> = serde_json::from_str(&media_refs_json).map_err(|err| {
|
||||
rusqlite::Error::FromSqlConversionFailure(
|
||||
media_refs_json.len(),
|
||||
rusqlite::types::Type::Text,
|
||||
Box::new(err),
|
||||
)
|
||||
})?;
|
||||
|
||||
let tool_calls_json: Option<String> = row.get(9)?;
|
||||
let tool_calls = tool_calls_json
|
||||
.as_deref()
|
||||
.map(serde_json::from_str)
|
||||
.transpose()
|
||||
.map_err(|err| {
|
||||
rusqlite::Error::FromSqlConversionFailure(
|
||||
9,
|
||||
rusqlite::types::Type::Text,
|
||||
Box::new(err),
|
||||
)
|
||||
})?;
|
||||
|
||||
Ok(ChatMessage {
|
||||
id: row.get(0)?,
|
||||
role: row.get(1)?,
|
||||
content: row.get(2)?,
|
||||
system_context: row.get(3)?,
|
||||
reasoning_content: row.get(4)?,
|
||||
media_refs,
|
||||
timestamp: row.get(6)?,
|
||||
tool_call_id: row.get(7)?,
|
||||
tool_name: row.get(8)?,
|
||||
tool_state: None,
|
||||
tool_calls,
|
||||
})
|
||||
})?;
|
||||
|
||||
let mut messages = Vec::new();
|
||||
for row in rows {
|
||||
messages.push(row?);
|
||||
}
|
||||
Ok(messages)
|
||||
}
|
||||
|
||||
pub fn load_all_messages(&self, session_id: &str) -> Result<Vec<ChatMessage>, StorageError> {
|
||||
let conn = self.conn.lock().expect("session db mutex poisoned");
|
||||
load_messages_after(&conn, session_id, 0)
|
||||
@ -1349,6 +1531,18 @@ fn ensure_messages_schema(conn: &Connection) -> Result<(), StorageError> {
|
||||
)?;
|
||||
}
|
||||
|
||||
if !has_column(conn, "messages", "topic_id")? {
|
||||
add_column_if_missing(conn, "ALTER TABLE messages ADD COLUMN topic_id TEXT")?;
|
||||
// 添加外键约束（SQLite 不支持 ALTER TABLE ADD FOREIGN KEY，需要重建表）
|
||||
// 这里只添加列，外键约束由应用层保证
|
||||
}
|
||||
|
||||
// 创建 topic_id 索引（如果不存在）
|
||||
conn.execute(
|
||||
"CREATE INDEX IF NOT EXISTS idx_messages_topic_seq ON messages(topic_id, seq) WHERE topic_id IS NOT NULL",
|
||||
[],
|
||||
)?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@ -28,6 +28,18 @@ pub struct SessionRecord {
|
||||
pub agent_prompt_reinjection_count: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct TopicRecord {
|
||||
pub id: String,
|
||||
pub session_id: String,
|
||||
pub title: String,
|
||||
pub description: Option<String>,
|
||||
pub created_at: i64,
|
||||
pub updated_at: i64,
|
||||
pub last_active_at: i64,
|
||||
pub message_count: i64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MemoryRecord {
|
||||
pub id: String,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user