zclaw_openfang/docs/superpowers/specs/2026-04-02-chatstore-refactor-design.md

ChatStore 结构化重构设计

日期: 2026-04-02 状态: Complete (Phase 0-8, 三轮审计通过) 范围: desktop/src/store/chatStore.ts 及关联文件

1. 背景

ChatStore（908 行）是 ZCLAW 桌面端聊天的核心状态管理模块，承担了消息管理、流式处理、对话管理、Artifact 面板、离线队列、ChatMode 切换等职责。经过多轮功能迭代，存在以下问题：

1.1 功能断裂

问题	影响
cancelStream() 是 no-op	用户无法取消长时间运行的流式响应
GatewayClient 路径不支持 thinking delta	Web 版/远程连接用户无法使用推理模式
双路径（Kernel/Gateway）逻辑重复且不一致	维护成本高，行为不可预测

1.2 数据可靠性

问题	影响
流式过程中刷新页面丢失未持久化内容	用户丢失已生成的回复
无消息重试机制	网络波动后用户需手动重新输入
对话持久化仅在 onComplete 时触发	长对话中断后数据丢失

1.3 架构债务

问题	影响
ChatStore 908 行、职责过多	难以理解和修改
Message vs SessionMessage 两套类型体系	类型转换混乱
未纳入 Store Coordinator	不符合项目 store 注入模式
Conversation.tsx 有未使用的 Context	死代码增加认知负担

1.4 性能风险

问题	影响
所有消息全量在内存	长对话占用过多内存
每次 onDelta 触发全量 set() 映射	频繁渲染影响性能

2. 设计决策

2.1 方案选择

选择 方案 B: 结构化多 Store 重构，理由：

• 每个拆分后的 Store 可在一个上下文窗口内完整理解
• 统一流式抽象层消除双路径重复
• 定时批量持久化平衡性能与可靠性
• 逐步迁移保证每步可验证

2.2 双路径统一策略

统一 KernelClient（Tauri 事件）和 GatewayClient（WebSocket）的流式体验，使两条路径具备相同能力：thinking delta 支持、5 分钟超时、取消机制。

2.3 持久化策略

采用定时批量保存（每 3 秒或每 50 条 delta），使用 requestIdleCallback（不可用时降级为 setTimeout）降低对 UI 的性能影响。存储目标为 IndexedDB（通过 idb-keyval 库），避免 localStorage 的 5-10 MB 大小限制。localStorage 仅保留对话元数据（id 列表、当前对话 ID、当前 agent）。

3. 架构设计

3.1 Store 拆分

desktop/src/store/
├── chat/                      # 新建目录
│   ├── conversationStore.ts   # 对话列表管理（~200行）
│   ├── messageStore.ts        # 消息管理 + 检索（~250行）
│   ├── streamStore.ts         # 统一流式处理（~200行）
│   └── artifactStore.ts       # Artifact 面板（~80行）
├── chatStore.ts               # 保留为 facade，re-export 统一接口

3.2 conversationStore

职责: 对话生命周期管理

状态:

• conversations: Conversation[]
• currentConversationId: string | null
• agents: Agent[]（agent 列表，从现有 chatStore 迁入）
• currentAgent: Agent | null
• sessionKey: string | null
• currentModel: string（当前模型名称，持久化）

Actions:

• newConversation() — 保存当前对话，创建新的空对话
• switchConversation(id: string) — 保存当前，加载目标对话
• deleteConversation(id: string) — 删除对话
• upsertActiveConversation() — 批量保存当前对话的 messages/sessionKey 到 conversations 数组
• getCurrentConversation() — 获取当前活跃对话
• setCurrentAgent(agent: Agent) — 切换 agent，保存/恢复对话
• syncAgents(profiles: AgentProfileLike[]) — 同步 agent 列表
• setCurrentModel(model: string) — 切换模型

Agent 绑定: 每个 Conversation 关联一个 agentId，切换对话时恢复对应 agent。

存储: 对话列表和 agent 信息持久化到 IndexedDB（通过 zustand persist 的自定义 storage），localStorage 仅存元数据。包含存储配额检查：写入前估算数据大小，超过 4 MB 时自动清理最旧的已归档对话。

3.3 messageStore

职责: 当前对话的消息数据管理

状态:

• messages: ChatMessage[]
• totalInputTokens: number
• totalOutputTokens: number

Actions:

• addMessage(message: ChatMessage) — 追加消息
• updateMessage(id: string, updates: Partial<ChatMessage>) — 合并更新
• getStreamingMessage() — 获取当前流式消息（role=assistant 且 streaming=true）
• updateStreamingContent(id: string, delta: string) — 高性能增量更新
• appendThinking(id: string, delta: string) — 追加 thinking 内容
• addToolStep(id: string, step: ToolStep) — 追加工具调用步骤
• completeMessage(id: string, tokens: TokenUsage) — 标记消息完成，记录 token
• failMessage(id: string, error: string) — 标记消息失败，保存原始内容用于重试
• retryMessage(id: string) — 使用 originalContent 创建重试
• addTokenUsage(input: number, output: number) — 累计 token
• resetMessages(messages: ChatMessage[]) — 切换对话时重载消息
• searchMessages(query: string) — 消息内文本搜索

3.4 streamStore

职责: 统一流式处理、离线队列、完成后副作用

状态:

• isStreaming: boolean
• isLoading: boolean
• streamHandle: StreamHandle | null
• chatMode: ChatModeType
• suggestions: string[]

Actions:

• sendMessage(content: string, context?: SendMessageContext) — 核心发送逻辑
  1. 离线检查：调用 offlineStore.isOffline()；若离线，委托 offlineStore.queueMessage() 并显示系统消息后返回
  2. 流式守卫：若 isStreaming === true，拒绝发送（前端防重复）
  3. 选择活跃的 StreamingAdapter
  4. 原子消息创建：一次性创建 optimistic 用户消息 + 流式 assistant 占位消息，通过单次 set() 写入 messageStore（避免部分状态被批量保存）
  5. 启动流式请求，注册 callbacks
  6. 管理 dirty 标志触发批量保存
  7. 完成后副作用（onComplete 回调中）：
     • conversationStore.upsertActiveConversation() — 立即保存
     • memoryExtractor.extractFromConversation() — 异步记忆提取（.catch 静默处理）
     • intelligenceClient.reflection.recordConversation() — 对话记录（.catch 静默处理）
     • intelligenceClient.reflection.shouldReflect() — 反射触发检查
     • generateFollowUpSuggestions(content) — 关键词建议生成 → setSuggestions()
     • 浏览器 TTS（如已启用）
• cancelStream() — 取消当前流式响应
• setChatMode(mode: ChatModeType) — 切换聊天模式
• getChatModeConfig() — 获取当前模式配置
• setSuggestions(suggestions: string[]) — 设置建议列表

批量保存机制:

流式开始（用户消息 + assistant 占位已原子写入 messageStore）
  │
  ├── dirty 标志管理
  │   └── 每次 delta/thinking/tool 更新后设置 dirty = true
  │
  ├── 每 3 秒检查 dirty 标志
  │   └── dirty → conversationStore.upsertActiveConversation()
  │
  ├── 每累积 50 条 delta 强制保存
  │
  ├── onComplete → 立即保存 + 触发副作用
  │
  └── onError → 立即保存（保留已接收的部分内容）

跨 Store 同步契约: streamStore 调用 messageStore 和 conversationStore 的方法均为同步 Zustand set() 调用。批量保存计时器（setInterval）在 Zustand 事务外运行，读取的 messages 始终是上一帧的完整快照——不存在部分写入的中间态。

依赖: streamStore → messageStore（更新流式消息）、conversationStore（保存对话）、offlineStore（离线队列）、connectionStore（选择 adapter）

3.5 artifactStore

职责: Artifact 面板管理（从现有 ChatStore 直接提取，无逻辑变更）

状态:

• artifacts: ArtifactFile[]
• selectedArtifactId: string | null
• artifactPanelOpen: boolean

Actions:

• addArtifact(artifact), selectArtifact(id), setArtifactPanelOpen(open), clearArtifacts()

3.6 Facade 兼容层

保留 chatStore.ts 作为 re-export facade，确保渐进迁移：

// chatStore.ts (facade)
export { useConversationStore } from './chat/conversationStore'
export { useMessageStore } from './chat/messageStore'
export { useStreamStore } from './chat/streamStore'
export { useArtifactStore } from './chat/artifactStore'

// 兼容层 — 逐步迁移后删除
export { useChatStore } from './chat/chatStoreCompat'

chatStoreCompat 聚合所有子 Store 的状态和 actions 为统一的 useChatStore 接口，使现有组件无需修改即可继续工作。

4. 统一流式抽象层

4.1 StreamingAdapter 接口

文件: desktop/src/lib/streaming-adapter.ts

interface StreamCallbacks {
  onDelta: (delta: string) => void
  onThinkingDelta?: (delta: string) => void
  onToolStart?: (name: string, input: unknown) => void
  onToolEnd?: (name: string, output: unknown) => void
  onHandStart?: (name: string) => void
  onHandEnd?: (name: string, result: unknown) => void
  onWorkflowStart?: (workflowId: string) => void
  onWorkflowEnd?: (workflowId: string, result: unknown) => void
  onComplete: (tokens: TokenUsage) => void
  onError: (error: string) => void
}

interface TokenUsage {
  inputTokens: number
  outputTokens: number
}

interface StreamHandle {
  cancel(): void
  readonly active: boolean
}

interface StreamingAdapter {
  start(
    agentId: string,
    message: string,
    sessionId: string,
    mode: ChatModeType,
    callbacks: StreamCallbacks
  ): StreamHandle

  isAvailable(): boolean
}

4.2 KernelStreamAdapter

封装现有 kernel-chat.ts 的 chatStream 方法:

• 监听 Tauri stream:chunk 事件
• TokenUsage 适配: 现有 kernel-types.ts 的 onComplete 使用位置参数 (inputTokens?: number, outputTokens?: number)，KernelStreamAdapter 内部将其转换为 TokenUsage 对象 { inputTokens, outputTokens }
• 映射 StreamChatEvent 到 StreamCallbacks:

  StreamChatEvent	StreamCallbacks
  delta	onDelta(text)
  thinkingDelta	onThinkingDelta(thinking)
  tool_start	onToolStart(name, input)
  tool_end	onToolEnd(name, output)
  handStart	onHandStart(name)
  handEnd	onHandEnd(name, result)
  complete	onComplete({ inputTokens, outputTokens })
  error	onError(message)

• 5 分钟超时（保持现有行为）
• cancel() 调用新增的 Tauri command cancel_stream(session_id)
• iteration_start 事件内部日志记录，不暴露到 callbacks

4.3 GatewayStreamAdapter

封装 GatewayClient 的 WebSocket 流式:

• 使用 GatewayClient 的 chatStream 方法
• 映射 AgentStreamDelta 事件到 StreamCallbacks:

  AgentStreamDelta	StreamCallbacks
  stream === 'assistant'	onDelta(content)
  stream === 'thinking'	onThinkingDelta(content)
  stream === 'tool' + step === 'start'	onToolStart(name, input)
  stream === 'tool' + step === 'end'	onToolEnd(name, output)
  stream === 'hand' + step === 'start'	onHandStart(name)
  stream === 'hand' + step === 'end'	onHandEnd(name, result)
  stream === 'workflow' + step === 'start'	onWorkflowStart(workflowId)
  stream === 'workflow' + step === 'end'	onWorkflowEnd(workflowId, result)
  stream === 'lifecycle' + phase === 'end'	onComplete(tokens)
  stream === 'error'	onError(message)

• 新增 5 分钟超时（与 Kernel 统一）
• cancel() 关闭 WebSocket 连接并发送取消消息
• 统一 onComplete 签名，包含 TokenUsage 参数

4.4 适配器选择

streamStore 通过 connectionStore.getClient() 获取当前客户端实例，判断类型选择 adapter:

const client = getClient()
const adapter = client instanceof KernelClient
  ? kernelStreamAdapter
  : gatewayStreamAdapter

5. 类型统一

5.1 统一 ChatMessage 类型

文件: desktop/src/types/chat.ts

interface ChatMessage {
  id: string
  role: 'user' | 'assistant' | 'tool' | 'hand' | 'workflow' | 'system'
  content: string
  timestamp: Date
  streaming?: boolean
  optimistic?: boolean
  // thinking
  thinkingContent?: string
  // error & retry
  error?: string
  originalContent?: string
  // tool/hand context
  toolSteps?: ToolStep[]
  handName?: string
  handStatus?: string
  handResult?: unknown
  // workflow
  workflowId?: string
  workflowStep?: number
  workflowStatus?: string
  workflowResult?: unknown
  // subtasks
  subtasks?: Subtask[]
  // attachments
  files?: MessageFile[]
  codeBlocks?: CodeBlock[]
  // metadata
  metadata?: {
    inputTokens?: number
    outputTokens?: number
    model?: string
    runId?: string
  }
}

5.2 辅助类型

// ToolStep 替代现有 ToolCallStep，统一命名
interface ToolStep {
  id: string
  name: string
  status: 'running' | 'completed' | 'error'
  input?: unknown
  output?: unknown
  startTime: Date
  endTime?: Date
}

interface Subtask {
  id: string
  title: string
  status: 'pending' | 'in_progress' | 'completed' | 'failed'
  result?: unknown
}

interface SendMessageContext {
  files?: MessageFile[]
  parentMessageId?: string
}

5.3 SessionMessage → ChatMessage 映射

Gateway 路径的会话历史使用 SessionMessage（API 响应格式，字符串日期），需要映射函数：

// desktop/src/types/chat.ts
function sessionToChatMessage(sm: SessionMessage): ChatMessage {
  return {
    id: sm.id,
    role: mapSessionRole(sm.role),  // 'user'/'assistant'/'system' 直接映射
    content: sm.content,
    timestamp: new Date(sm.timestamp),
    metadata: {
      model: sm.metadata?.model,
      inputTokens: sm.metadata?.tokens?.input,
      outputTokens: sm.metadata?.tokens?.output,
    },
  }
}

sessionStore 内部继续使用 API 响应的 SessionMessage 类型（它是 API 契约），仅在展示层转换为 ChatMessage。不需要修改 sessionStore 的内部类型。

5.4 类型清理

类型	文件	动作
ChatStore.Message	store/chatStore.ts	迁移到 types/chat.ts 的 ChatMessage
ConversationContext	components/ai/Conversation.tsx	仅删除未使用的 Provider/Context/hook，保留滚动容器组件
initStreamListener	store/chatStore.ts	被 streamStore + StreamingAdapter 替代

6. Cancel 机制

6.1 前端（Phase 5a — 纯前端取消）

前端 cancel 分两步实现，先纯前端方案（Phase 5a），后端配合后完善（Phase 5b）。

// streamStore
cancelStream() {
  if (this.streamHandle?.active) {
    this.streamHandle.cancel()
  }
  // 标记当前流式消息为已完成
  const streamingMsg = messageStore.getStreamingMessage()
  if (streamingMsg) {
    messageStore.updateMessage(streamingMsg.id, {
      streaming: false,
      content: streamingMsg.content + '\n\n_（响应已取消）_'
    })
  }
  set({ isStreaming: false, streamHandle: null })
  conversationStore.upsertActiveConversation() // 立即保存
}

KernelStreamAdapter.cancel()（Phase 5a 纯前端）：

• 停止监听 Tauri stream:chunk 事件（移除 listener）
• 不通知后端停止，后端继续运行直到自然完成或 5 分钟超时
• 前端标记消息为已取消，用户可继续操作

GatewayStreamAdapter.cancel()（Phase 5a 纯前端）：

• 发送 WebSocket 取消消息 { type: "cancel", sessionId }（如果 Gateway 服务端已支持则生效）
• 关闭事件监听器

6.2 后端配合（Phase 5b — 需新基础设施）

Phase 5b 在后端基础设施就绪后实施。需要在 Tauri 端新增：

1. SessionStreamGuards 状态: 在 lib.rs 中注册 DashMap<String, Arc<AtomicBool>> 作为 Tauri managed state
2. cancel_stream command: 读取 guards map，设置取消标志
3. 流式循环检查: tokio::spawn 内每轮迭代检查 cancel_flag

// chat.rs — 取消标志写入
#[tauri::command]
async fn cancel_stream(
    session_id: String,
    guards: State<'_, SessionStreamGuards>,
) -> Result<(), String> {
    if let Some(pair) = guards.0.get(&session_id) {
        pair.value().store(true, Ordering::SeqCst);
    }
    Ok(())
}

// agent_chat_stream — 在每轮接收循环中检查
let cancelled = cancel_flag.load(Ordering::Relaxed);
if cancelled {
    tx.send(LoopEvent::Complete(AgentLoopResult {
        response: "...".into(),
        input_tokens: 0,
        output_tokens: 0,
        iterations,
    })).ok();
    break;
}

Phase 5a 和 5b 可独立交付，Phase 5a 即可满足用户需求。

7. 迁移计划

7.1 迁移顺序

Phase	内容	影响文件数	风险
0	创建 types/chat.ts + store/chat/ 目录	新建	无
1	提取 artifactStore	~5	极低
2	提取 conversationStore（含 agents、sessionKey）	~8	低
3	提取 messageStore（含 token 统计、search）	~10	中
4	提取 streamStore + StreamingAdapter（含离线检查、副作用、suggestions）	~12	中高
5a	前端 cancel（纯前端，停止监听+标记已取消）	~3	低
5b	后端 cancel（Rust SessionStreamGuards + cancel_stream command）	~4	中
6	实现定时批量持久化（IndexedDB + 配额检查）	~4	中
7	删除旧代码 + 清理 facade	~6	低
8	删除死代码（ConversationContext、旧类型映射）	~4	低

7.2 每阶段验证

每个 Phase 完成后执行:

1. pnpm tsc --noEmit — 类型检查通过
2. pnpm vitest run — 现有测试通过
3. 手动验证: 发送消息 → 流式响应 → 切换对话 → 刷新页面数据保持
4. 手动验证（Phase 5 后）: cancel 流式响应 → 状态正确恢复

7.3 关键文件清单

文件	角色
desktop/src/store/chatStore.ts	重构对象，最终保留为 facade
desktop/src/store/chat/conversationStore.ts	新建
desktop/src/store/chat/messageStore.ts	新建
desktop/src/store/chat/streamStore.ts	新建
desktop/src/store/chat/artifactStore.ts	新建
desktop/src/store/chat/chatStoreCompat.ts	新建（兼容层，最终删除）
desktop/src/lib/streaming-adapter.ts	新建（StreamingAdapter 接口 + 双实现）
desktop/src/lib/kernel-chat.ts	修改（KernelStreamAdapter 封装）
desktop/src/types/chat.ts	新建（ChatMessage + ToolStep + Subtask）
desktop/src/types/session.ts	保留（API 契约类型），添加映射函数
desktop/src/components/ChatArea.tsx	逐步迁移 import
desktop/src/components/ai/Conversation.tsx	清理死代码（仅 Context/Provider）
desktop/src/store/index.ts	注册新 Store
desktop/src/store/offlineStore.ts	不修改，streamStore 调用其 API
desktop/src-tauri/src/kernel_commands/chat.rs	Phase 5b: 新增 cancel_stream + SessionStreamGuards
desktop/src-tauri/src/lib.rs	Phase 5b: 注册 cancel_stream + guards state

8. streamStore 完成后副作用

streamStore.sendMessage 的 onComplete 回调在流式响应完成后触发以下副作用。这些副作用在 streamStore 内部处理，不属于 StreamingAdapter 的职责。

// streamStore 内部 onComplete 处理
async function handleComplete(tokens: TokenUsage) {
  // 1. 更新消息状态
  messageStore.completeMessage(streamingMsgId, tokens)

  // 2. 立即持久化
  conversationStore.upsertActiveConversation()

  // 3. 记忆提取（非阻塞，失败静默）
  try {
    const extractor = getMemoryExtractor()
    if (extractor) {
      await extractor.extractFromConversation(
        conversationStore.getCurrentConversation()
      )
    }
  } catch (e) {
    logger.warn('Memory extraction failed', e)
  }

  // 4. 对话反思跟踪（非阻塞）
  try {
    const client = getIntelligenceClient()
    if (client?.reflection) {
      await client.reflection.recordConversation(...)
      const shouldReflect = await client.reflection.shouldReflect(...)
      if (shouldReflect) {
        // 触发反思流程
      }
    }
  } catch (e) {
    logger.warn('Reflection tracking failed', e)
  }

  // 5. 后续建议生成
  const suggestions = generateFollowUpSuggestions(lastAssistantContent)
  set({ suggestions })

  // 6. 语音朗读（如果用户开启）
  if (speechSettings.autoSpeak) {
    speechSynth.speak(lastAssistantContent)
  }
}

9. 离线队列集成

streamStore.sendMessage 在发起流式请求之前检查离线状态：

async sendMessage(content: string, context?: SendMessageContext) {
  // 1. 离线检查（优先级最高）
  // 注意：isOffline 是 boolean 属性（不是函数），queueMessage 使用位置参数
  const { isOffline, queueMessage } = useOfflineStore.getState()
  if (isOffline) {
    const userMsg = createUserMessage(content, context?.files)
    messageStore.addMessage(userMsg)
    messageStore.addMessage(createSystemMessage('消息已加入离线队列，网络恢复后将自动发送'))
    queueMessage(content, conversationStore.currentAgent?.id, conversationStore.sessionKey ?? undefined)
    conversationStore.upsertActiveConversation()
    return // 不继续流式请求
  }

  // 2. 正常流式流程...
}

10. streamStore 状态补充

suggestions 和 chatMode 归属 streamStore：

状态:

• suggestions: string[]
• chatMode: ChatModeType

Actions:

• setSuggestions(suggestions: string[]) — 设置后续建议
• setChatMode(mode: ChatModeType) — 切换聊天模式
• getChatModeConfig() — 获取当前模式配置
• searchSkills(query: string) — 委托给 getSkillDiscovery().searchSkills()

totalInputTokens/totalOutputTokens 为仅会话内累计（不持久化），刷新后重置为 0。

11. 兼容层迁移指南

chatStoreCompat.ts 聚合子 Store 为统一的 useChatStore 接口，确保现有 19 个消费者文件无需修改。

// chatStoreCompat.ts — 使用 Zustand subscribe 保持响应式
// 注意：不能在 create() 中直接 .getState()，那样只会读取初始值不会响应变化
import { subscribe } from 'zustand'

// 方案：直接 re-export 子 Store，组件按需导入
export { useConversationStore as useConversationStore } from './chat/conversationStore'
export { useMessageStore as useMessageStore } from './chat/messageStore'
export { useStreamStore as useStreamStore } from './chat/streamStore'
export { useArtifactStore as useArtifactStore } from './chat/artifactStore'

// 兼容 hook：聚合所有子 store 状态供旧组件使用
// 使用 useSyncExternalStore 或每个子 store 的独立 hook 组合
export function useChatStore<T>(selector: (state: ChatCompatState) => T): T {
  // 方案 A（推荐）：组件直接从子 store 导入
  // 方案 B（过渡期）：聚合 hook，内部使用多个 useSelector
  const conv = useConversationStore(selector)
  const msg = useMessageStore(selector)
  const stream = useStreamStore(selector)
  const art = useArtifactStore(selector)
  return selector({ ...conv, ...msg, ...stream, ...art })
}

重要：兼容层是过渡性代码，仅保证旧组件可编译运行。新代码必须直接使用子 Store。每个 Phase 迁移一部分组件后，兼容层逐步缩小。最终删除。

迁移方式：每 Phase 完成后，逐步将组件的 import { useChatStore } from './chatStore' 改为直接从子 Store 导入。最终删除 chatStoreCompat.ts。

12. 不在本次范围内

以下项目明确排除，作为后续迭代考虑:

• 消息分页/懒加载（当前所有消息全量在内存，Phase 2 考虑）
• 文件真实上传（当前附件是伪文本标记，需后端配合）
• TitleMiddleware 实现（后端 placeholder，需 LLM driver 接入）
• 消息导出增强（当前仅 Markdown 导出）