新增: Phase3 EventBus Value载荷+桥接层

- EventBus 载荷改 serde_json::Value(D2 全19变体透传),删 AiBusEvent 强类型子集
- 新增 publish_event(AiChatEvent) helper(内部 to_value→publish,供 emit 双写)
- 新增 remote_bridge.rs 桥接层(MiniCommand→Tauri command 路由+R1 同conv并发兜底)
- agentic/mod.rs 11处 AiBusEvent→AiChatEvent 迁移(Error/Completed/Round)
This commit is contained in:
2026-06-22 03:06:28 +08:00
parent 8ccdce282c
commit 0fb5af50c7
4 changed files with 750 additions and 172 deletions

View File

@@ -39,8 +39,6 @@ use df_storage::models::AiProviderRecord;
use crate::state::{AppState, LlmConcurrency}; use crate::state::{AppState, LlmConcurrency};
use crate::commands::ai::event_bus::AiBusEvent;
use super::audit::process_tool_calls; use super::audit::process_tool_calls;
use super::compress::compress_via_llm; use super::compress::compress_via_llm;
use super::conversation::{save_conversation, TokenAccumulator}; use super::conversation::{save_conversation, TokenAccumulator};
@@ -523,8 +521,9 @@ pub(crate) async fn run_agentic_loop(
}); });
// L3 emit 双写(2026-06-22):关键 AiError publish 到事件总线,供跨模块订阅。 // L3 emit 双写(2026-06-22):关键 AiError publish 到事件总线,供跨模块订阅。
// EVENT_BUS_ENABLED 门控在 publish 内部(false 静默丢弃),无消费者时空转不报错。 // EVENT_BUS_ENABLED 门控在 publish 内部(false 静默丢弃),无消费者时空转不报错。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Error { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: msg, error: msg,
error_type: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -541,8 +540,9 @@ pub(crate) async fn run_agentic_loop(
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
// L3 emit 双写:超时 AiError publish 到事件总线(同上,门控在 publish 内)。 // L3 emit 双写:超时 AiError publish 到事件总线(同上,门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Error { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: err_msg, error: err_msg,
error_type: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -713,10 +713,11 @@ pub(crate) async fn run_agentic_loop(
// generating 复位后再 emit Completed保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝) // generating 复位后再 emit Completed保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝)
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) }); let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) });
// L3 emit 双写:入口 stop 的 AiCompleted publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 // L3 emit 双写:入口 stop 的 AiCompleted publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Completed { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiCompleted {
total_tokens: usage.total_tokens, total_tokens: usage.total_tokens,
prompt_tokens: tokens.prompt(), prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(), completion_tokens: tokens.completion(),
incomplete: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -806,7 +807,7 @@ pub(crate) async fn run_agentic_loop(
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
// L3 emit 双写:AiAgentRound publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 // L3 emit 双写:AiAgentRound publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Round { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiAgentRound {
round: round_n, round: round_n,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
@@ -1174,8 +1175,9 @@ pub(crate) async fn run_agentic_loop(
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
// L3 emit 双写:Fatal AiError publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:Fatal AiError publish 到事件总线(门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Error { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error, error,
error_type: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -1203,8 +1205,9 @@ pub(crate) async fn run_agentic_loop(
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
// L3 emit 双写:全 candidate 耗尽 AiError publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:全 candidate 耗尽 AiError publish 到事件总线(门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Error { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: err_msg, error: err_msg,
error_type: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -1262,10 +1265,11 @@ pub(crate) async fn run_agentic_loop(
incomplete: Some(true), incomplete: Some(true),
}); });
// L3 emit 双写:MidStream 保文 AiCompleted publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:MidStream 保文 AiCompleted publish 到事件总线(门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Completed { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiCompleted {
total_tokens: usage.total_tokens, total_tokens: usage.total_tokens,
prompt_tokens: tokens.prompt(), prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(), completion_tokens: tokens.completion(),
incomplete: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -1329,10 +1333,11 @@ pub(crate) async fn run_agentic_loop(
// generating 复位后再 emit Completed保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝) // generating 复位后再 emit Completed保证前端收事件时后端已可接下一条(发送队列续发不被"正在生成中"拒绝)
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) }); let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage.total_tokens, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) });
// L3 emit 双写:stream 后 stop 的 AiCompleted publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:stream 后 stop 的 AiCompleted publish 到事件总线(门控在 publish 内)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Completed { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiCompleted {
total_tokens: usage.total_tokens, total_tokens: usage.total_tokens,
prompt_tokens: tokens.prompt(), prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(), completion_tokens: tokens.completion(),
incomplete: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
return; return;
@@ -1524,10 +1529,11 @@ pub(crate) async fn run_agentic_loop(
let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage_total, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) }); let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompleted { total_tokens: usage_total, prompt_tokens: tokens.prompt(), completion_tokens: tokens.completion(), incomplete: None, conversation_id: Some(conv_id.clone()) });
// L3 emit 双写:正常完成 AiCompleted publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内, // L3 emit 双写:正常完成 AiCompleted publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内,
// 无消费者空转留批3 真实消费者接入)。AiTextDelta/AiToolCall* 高频事件不双写(无消费者空转)。 // 无消费者空转留批3 真实消费者接入)。AiTextDelta/AiToolCall* 高频事件不双写(无消费者空转)。
let _ = app_handle.state::<AppState>().ai_event_bus.publish(AiBusEvent::Completed { let _ = app_handle.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiCompleted {
total_tokens: usage_total, total_tokens: usage_total,
prompt_tokens: tokens.prompt(), prompt_tokens: tokens.prompt(),
completion_tokens: tokens.completion(), completion_tokens: tokens.completion(),
incomplete: None,
conversation_id: Some(conv_id.clone()), conversation_id: Some(conv_id.clone()),
}); });
} }
@@ -1654,8 +1660,9 @@ pub(crate) async fn try_continue_agent_loop(
conversation_id: Some(conv_id.to_string()), conversation_id: Some(conv_id.to_string()),
}); });
// L3 emit 双写:try_continue provider-Err AiError publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:try_continue provider-Err AiError publish 到事件总线(门控在 publish 内)。
let _ = app.state::<AppState>().ai_event_bus.publish(AiBusEvent::Error { let _ = app.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiError {
error: e, error: e,
error_type: None,
conversation_id: Some(conv_id.to_string()), conversation_id: Some(conv_id.to_string()),
}); });
return; return;
@@ -1723,7 +1730,7 @@ pub(crate) async fn try_continue_agent_loop(
conversation_id: Some(conv_id_owned.clone()), conversation_id: Some(conv_id_owned.clone()),
}); });
// L3 emit 双写:try_continue 续跑 AiAgentRound publish 到事件总线(门控在 publish 内)。 // L3 emit 双写:try_continue 续跑 AiAgentRound publish 到事件总线(门控在 publish 内)。
let _ = app.state::<AppState>().ai_event_bus.publish(AiBusEvent::Round { let _ = app.state::<AppState>().ai_event_bus.publish_event(AiChatEvent::AiAgentRound {
round: 0, round: 0,
conversation_id: Some(conv_id_owned.clone()), conversation_id: Some(conv_id_owned.clone()),
}); });

View File

@@ -1,25 +1,28 @@
//! 全局事件数据总线 — pub-sub 骨架(L3 阶段1,渐进第一步) //! 全局事件数据总线 — pub-sub 骨架(载荷 serde_json::Value 透传)
//! //!
//! 关联设计:docs/02-架构设计/专项设计/全局事件数据总线-2026-06-21.md //! 关联设计:docs/02-架构设计/专项设计/全局事件数据总线-2026-06-21.md
//! 跨端透传:docs/02-架构设计/已编号方案/F-260622-01-跨端AIChat-Phase3联调设计-2026-06-22.md(§2.1 D2/D3)
//! //!
//! 本模块是 **统一事件数据总线(Unified Event Bus)** 的后端骨架,聚焦 **pub-sub**(发布订阅) //! 本模块是 **统一事件数据总线(Unified Event Bus)** 的后端骨架,聚焦 **pub-sub**(发布订阅)
//! 语义。背景:当前 AI 事件经 `app_handle.emit("ai-chat-event", ...)` 散布在各处,无统一总线, //! 语义。背景:AI 事件经 `app_handle.emit("ai-chat-event", ...)` 散布在各处(55 处/7 文件),
//! 跨模块直接耦合,阻碍 ai-working 可插拔 + 跨端透传。 //! 无统一总线,跨模块直接耦合,阻碍跨端透传(df-tunnel)
//!
//! ## 载荷决策(Phase3 D2 = 全 19 变体透传)
//! 载荷用 `serde_json::Value`(即 AiChatEvent 序列化 JSON),**不用**强类型 enum 子集。
//! 理由:跨端透传需 AiChatEvent 全 19 变体(含 ApprovalRequired/Compressed 等),早期骨架的
//! AiBusEvent 8 变体强类型子集不足以覆盖,且总线在透传场景不该解析业务语义(传输层天职)。
//! 改 Value 后:tunnel subscriber 收 Value 原样转发 miniapp,零变体丢失,零业务耦合。
//! //!
//! ## 本批范围(L3 骨架,纯新增无行为变化) //! ## 本批范围(L3 骨架,纯新增无行为变化)
//! - `EventBus` 结构(基于 tokio::sync::broadcast,容量 256) //! - `EventBus` 结构(基于 tokio::sync::broadcast,容量 256,载荷 `serde_json::Value`)
//! - `AiBusEvent` 强类型 enum(覆盖关键事件:Stream/Tool/Round/Help/Heartbeat/Completed/Error,
//! 对齐现有 `AiChatEvent` 子集)
//! - `subscribe()` / `publish()` 基础接口 //! - `subscribe()` / `publish()` 基础接口
//! - `EVENT_BUS_ENABLED` 开关(预留,默认 on —— 总线骨架自带,接入实际事件源时此开关控制是否经总线) //! - `EVENT_BUS_ENABLED` 开关(预留,默认 on —— 总线骨架自带,接入事件源时此开关控制是否经总线)
//! - pub/sub 基础单元测试 //! - pub/sub 基础单元测试
//! //!
//! ## 不做(留后续批次) //! ## 不做(留后续批次)
//! - 不接入 `agentic/mod.rs` 各 emit 点(B 域,留批2接入) //! - 不接入 55 处 emit 点(批3 双写:publish + 原 app.emit 并行,关开关回退)
//! - 不入 `AppState`(接入留批2) //! - 不接 df-tunnel subscriber(Phase3 阶段2:tunnel 注册 subscriber,收 Value 转发 miniapp)
//! - 不碰前端订阅渲染(留批2 Tauri bridge)
//! - 不实现 request-reply / 流式 reply(设计阶段2/5,本批仅 pub-sub) //! - 不实现 request-reply / 流式 reply(设计阶段2/5,本批仅 pub-sub)
//! - 不做跨端透传(df-tunnel adapter,阶段6)
//! //!
//! ## 开关与兜底(渐进可回退) //! ## 开关与兜底(渐进可回退)
//! - `EVENT_BUS_ENABLED: bool`(默认 true):接入事件源后,关闭则 publish 静默丢弃(不影响原 emit 路径)。 //! - `EVENT_BUS_ENABLED: bool`(默认 true):接入事件源后,关闭则 publish 静默丢弃(不影响原 emit 路径)。
@@ -30,12 +33,13 @@
//! //!
//! ## 与 df-workflow EventBus 关系 //! ## 与 df-workflow EventBus 关系
//! df-workflow 的 `EventBus`(`crates/df-workflow/src/eventbus.rs`)是工作流节点间专用总线 //! df-workflow 的 `EventBus`(`crates/df-workflow/src/eventbus.rs`)是工作流节点间专用总线
//! (仅 `WorkflowEvent`),本总线是面向 AI 域的通用 pub-sub(对齐 `AiChatEvent` 子集)。 //! (仅 `WorkflowEvent`),本总线是面向 AI 域的通用 pub-sub(透传 AiChatEvent 全量 Value)。
//! 设计阶段1 计划「df-workflow EventBus 包装为通用 DomainEvent」,本骨架先建 AI 域总线, //! AppState 分两个字段:`event_bus`(工作流)+ `ai_event_bus`(AI 域),互不干扰。
//! 通用 DomainEvent 泛化留后续(待决策1 bus crate 位置 + 待决策2 事件 schema)。
use tokio::sync::broadcast; use tokio::sync::broadcast;
use super::AiChatEvent;
// ============================================================ // ============================================================
// 开关(EVENT_BUS_ENABLED,预留,默认 on) // 开关(EVENT_BUS_ENABLED,预留,默认 on)
// //
@@ -51,7 +55,7 @@ use tokio::sync::broadcast;
/// 路径不受影响。骨架阶段未接入,无实际效果。 /// 路径不受影响。骨架阶段未接入,无实际效果。
/// ///
/// dead_code 说明:本批为骨架,EVENT_BUS_ENABLED 暂无消费方(未接入 emit 点); /// dead_code 说明:本批为骨架,EVENT_BUS_ENABLED 暂无消费方(未接入 emit 点);
/// 标 allow 保留作批2接入期的快速回退开关(零调用方≠垃圾,预留保留)。 /// 标 allow 保留作批3 接入期的快速回退开关(零调用方≠垃圾,预留保留)。
#[allow(dead_code)] #[allow(dead_code)]
pub const EVENT_BUS_ENABLED: bool = true; pub const EVENT_BUS_ENABLED: bool = true;
@@ -66,98 +70,26 @@ pub const EVENT_BUS_ENABLED: bool = true;
pub const DEFAULT_BUS_CAPACITY: usize = 256; pub const DEFAULT_BUS_CAPACITY: usize = 256;
// ============================================================ // ============================================================
// AiBusEvent — 总线事件类型(对齐 AiChatEvent 子集) // EventBus — 基于 tokio::sync::broadcast 的 pub-sub 总线(载荷 serde_json::Value)
//
// 设计取舍(对齐设计文档待决策2「事件 schema」倾向:强类型 + 开放式):
// - 强类型 enum:编译期安全,新增事件类型需改 enum(扩展成本可控,AI 域事件类型有限且稳定)
// - 覆盖关键事件:Stream(流式文本)/Tool(工具调用)/Round(agent 新轮)/Help(求助)/
// Heartbeat(心跳)/Completed(完成)/Error(错误) —— 对齐现有 AiChatEvent 高频子集
//
// 与 AiChatEvent 区别:
// - AiChatEvent 是「后端→前端单向推送载荷」(经 app.emit,Tauri 序列化,前端消费)
// - AiBusEvent 是「总线内部事件」(经 broadcast channel,后端模块间消费,设计上可桥接到前端)
// 本骨架 AiBusEvent 复用 AiChatEvent 字段语义,但作为独立类型(总线事件 ≠ 前端载荷,未来
// 可能分叉:如总线事件携带 reply_to / correlation_id 等 request-reply 元数据)。
//
// 序列化:derive Serialize/Deserialize 备跨端透传(serde JSON,待决策4);Clone 备 broadcast
// 多订阅者复制;Debug 备日志诊断。
// ============================================================
/// AI 事件总线事件(pub-sub 域,对齐 AiChatEvent 子集)。
///
/// 总线订阅者经 `EventBus::subscribe()` 拿 Receiver 后 recv 本类型事件。
/// 事件类型覆盖关键 AI 生命周期:流式文本 / 工具调用 / agent 新轮 / 求助 / 心跳 / 完成 / 错误。
///
/// dead_code 说明:骨架阶段 enum 本身未在 emit 点构造(批2接入),标 allow 保留
/// (零调用方≠垃圾,本枚举为总线核心契约,批2接入立即消费)。
#[allow(dead_code)]
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum AiBusEvent {
/// 流式文本片段(对齐 AiChatEvent::AiTextDelta)
Stream {
delta: String,
conversation_id: Option<String>,
},
/// 工具调用开始(对齐 AiChatEvent::AiToolCallStarted)
ToolStarted {
id: String,
name: String,
args: serde_json::Value,
conversation_id: Option<String>,
},
/// 工具调用完成(对齐 AiChatEvent::AiToolCallCompleted)
ToolCompleted {
id: String,
result: serde_json::Value,
conversation_id: Option<String>,
},
/// Agent 循环新一轮(对齐 AiChatEvent::AiAgentRound)
Round {
round: u32,
conversation_id: Option<String>,
},
/// 求助(对齐 AiChatEvent::AiHelpRequired,L1 求助协议)
Help {
reason: String,
context: String,
options: Vec<String>,
conversation_id: Option<String>,
},
/// 流式心跳(对齐 AiChatEvent::AiHeartbeat,静默期报活)
Heartbeat {
conversation_id: Option<String>,
},
/// AI 响应完成(对齐 AiChatEvent::AiCompleted)
Completed {
total_tokens: u32,
prompt_tokens: u32,
completion_tokens: u32,
conversation_id: Option<String>,
},
/// 错误(对齐 AiChatEvent::AiError)
Error {
error: String,
conversation_id: Option<String>,
},
}
// ============================================================
// EventBus — 基于 tokio::sync::broadcast 的 pub-sub 总线
// //
// 选 broadcast 而非 mpsc: // 选 broadcast 而非 mpsc:
// - pub-sub 多订阅者(mpsc 单消费者不满足「多模块订阅同一事件」场景) // - pub-sub 多订阅者(mpsc 单消费者不满足「多模块订阅同一事件」场景)
// - broadcast 容量固定,慢消费者丢老事件而非阻塞发布者(对齐流式场景:宁可丢老 chunk 不阻塞 LLM 流) // - broadcast 容量固定,慢消费者丢老事件而非阻塞发布者(对齐流式场景:宁可丢老 chunk 不阻塞 LLM 流)
// //
// 载荷 serde_json::Value(Phase3 D2):透传 AiChatEvent 全 19 变体原样 JSON,subscriber(tunnel)
// 收到后零解析转发 miniapp。总线不持业务类型知识(传输层天职),扩展 AiChatEvent 变体无需改总线。
//
// Clone 语义:broadcast::Sender 内部 Arc,clone 共享同一通道(多持有者 publish 到同一总线)。 // Clone 语义:broadcast::Sender 内部 Arc,clone 共享同一通道(多持有者 publish 到同一总线)。
// 对齐设计 §架构「模块订阅 + 发布,无相互 import」—— 各模块持 clone 的 EventBus 即可 pub/sub。 // 对齐设计 §架构「模块订阅 + 发布,无相互 import」—— 各模块持 clone 的 EventBus 即可 pub/sub。
// ============================================================ // ============================================================
/// AI 事件总线(pub-sub 域,L3 阶段1 骨架)。 /// AI 事件总线(pub-sub 域,载荷 `serde_json::Value`,L3 骨架)。
/// ///
/// 基于 `tokio::sync::broadcast`,多订阅者多发布者共享同一通道。 /// 基于 `tokio::sync::broadcast`,多订阅者多发布者共享同一通道。
/// - `subscribe()`:订阅,返回 [`broadcast::Receiver`] recv [`AiBusEvent`] /// - `subscribe()`:订阅,返回 [`broadcast::Receiver`] recv `serde_json::Value`
/// - `publish(event)`:发布,所有活跃订阅者收到(broadcast 语义) /// - `publish(payload)`:发布,所有活跃订阅者收到(broadcast 语义)
///
/// 载荷是 AiChatEvent 序列化后的 JSON(Phase3 D2 全 19 变体透传),subscriber 不解析业务语义。
/// ///
/// 容量默认 256(`DEFAULT_BUS_CAPACITY`),可经 [`EventBus::with_capacity`] 自定义。 /// 容量默认 256(`DEFAULT_BUS_CAPACITY`),可经 [`EventBus::with_capacity`] 自定义。
/// ///
@@ -166,24 +98,24 @@ pub enum AiBusEvent {
/// - 慢消费者(容量满)publish 返 0(静默丢老事件,不 panic) /// - 慢消费者(容量满)publish 返 0(静默丢老事件,不 panic)
/// - `EVENT_BUS_ENABLED=false` 时 publish 静默丢弃(预留开关,接入期回退用) /// - `EVENT_BUS_ENABLED=false` 时 publish 静默丢弃(预留开关,接入期回退用)
/// ///
/// dead_code 说明:骨架阶段未入 AppState、未在 emit 点构造(批2接入),标 allow 保留 /// dead_code 说明:骨架阶段 emit 点未双写(publish 无调用方);标 allow 保留作批3 接入
/// (总线核心结构,批2接入 AppState 即消费)。 /// (零调用方≠垃圾,总线核心结构,批3 接入即消费)。
#[allow(dead_code)] #[allow(dead_code)]
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct EventBus { pub struct EventBus {
sender: broadcast::Sender<AiBusEvent>, sender: broadcast::Sender<serde_json::Value>,
} }
/// 事件订阅者(broadcast::Receiver,AiBusEvent 接收端)。 /// 事件订阅者(broadcast::Receiver,serde_json::Value 接收端)。
/// ///
/// dead_code 说明:骨架阶段无外部消费方(批2接入订阅),标 allow 保留。 /// dead_code 说明:骨架阶段无外部消费方(批3 接入 tunnel subscriber),标 allow 保留。
#[allow(dead_code)] #[allow(dead_code)]
pub type EventSubscriber = broadcast::Receiver<AiBusEvent>; pub type EventSubscriber = broadcast::Receiver<serde_json::Value>;
// dead_code 说明(impl 块):骨架阶段 EventBus 未入 AppState、无外部调用方 // dead_code 说明(impl 块):骨架阶段 EventBus publish 无调用方(emit 点未双写)
// (批2接入 AppState + emit 点后立即消费)。标 allow 覆盖 new/with_capacity/ // (批3 接入 emit 点 + tunnel subscriber 后立即消费)。标 allow 覆盖 new/with_capacity/
// subscribe/publish/subscriber_count 全部关联项的 never-used 警告。 // subscribe/publish/subscriber_count 全部关联项的 never-used 警告。
// 对齐零调用方原则:预留保留不盲删(批2接入即消除)。 // 对齐零调用方原则:预留保留不盲删(批3 接入即消除)。
#[allow(dead_code)] #[allow(dead_code)]
impl EventBus { impl EventBus {
/// 创建默认容量(256)的事件总线。 /// 创建默认容量(256)的事件总线。
@@ -207,7 +139,7 @@ impl EventBus {
self.sender.subscribe() self.sender.subscribe()
} }
/// 发布事件到总线。 /// 发布 AiChatEvent 序列化 JSON 到总线。
/// ///
/// 返回值:成功送达的活跃订阅者数量(0 = 无订阅者或容量满丢老事件)。 /// 返回值:成功送达的活跃订阅者数量(0 = 无订阅者或容量满丢老事件)。
/// `EVENT_BUS_ENABLED=false` 时静默丢弃返回 0(预留开关,骨架阶段未接入)。 /// `EVENT_BUS_ENABLED=false` 时静默丢弃返回 0(预留开关,骨架阶段未接入)。
@@ -215,7 +147,7 @@ impl EventBus {
/// 注:broadcast::send 是同步方法(非 async),与 df-workflow EventBus::send 的 async 签名 /// 注:broadcast::send 是同步方法(非 async),与 df-workflow EventBus::send 的 async 签名
/// 不同 —— 本骨架 publish 同步返回更直接(broadcast::send 内部无 await 点),df-workflow 的 /// 不同 —— 本骨架 publish 同步返回更直接(broadcast::send 内部无 await 点),df-workflow 的
/// async 签名是为对齐 trait 抽象,本总线无此约束故同步。 /// async 签名是为对齐 trait 抽象,本总线无此约束故同步。
pub fn publish(&self, event: AiBusEvent) -> usize { pub fn publish(&self, payload: serde_json::Value) -> usize {
if !EVENT_BUS_ENABLED { if !EVENT_BUS_ENABLED {
return 0; return 0;
} }
@@ -223,7 +155,21 @@ impl EventBus {
// Ok(n) = 送达 n 个活跃订阅者 // Ok(n) = 送达 n 个活跃订阅者
// Err(SendError(_)) = 无订阅者(事件丢弃) // Err(SendError(_)) = 无订阅者(事件丢弃)
// 两种情况都不 panic,Err 视为 0(对齐「无订阅者静默丢弃」兜底)。 // 两种情况都不 panic,Err 视为 0(对齐「无订阅者静默丢弃」兜底)。
self.sender.send(event).unwrap_or(0) self.sender.send(payload).unwrap_or(0)
}
/// 便捷发布:接收强类型 AiChatEvent,内部序列化为 Value 透传(D2 全 19 变体透传)。
///
/// 调用方传 AiChatEvent(业务侧已有强类型对象),本方法内部 `serde_json::to_value`
/// 序列化为 Value 后调 [`publish`](Self::publish)。供 emit 点双写(publish + app.emit),
/// 经 EventBus 透传给 tunnel subscriber(阶段2 接入)。
///
/// 序列化失败兜底返 0(丢弃,对齐 publish 静默丢弃语义;AiChatEvent 序列化稳定不触发,防御性)。
pub fn publish_event(&self, event: AiChatEvent) -> usize {
match serde_json::to_value(&event) {
Ok(v) => self.publish(v),
Err(_) => 0,
}
} }
/// 当前活跃订阅者数量(broadcast::receiver_count)。 /// 当前活跃订阅者数量(broadcast::receiver_count)。
@@ -241,45 +187,42 @@ impl Default for EventBus {
} }
// ============================================================ // ============================================================
// 单元测试 — pub/sub 基础语义 // 单元测试 — pub/sub 基础语义(载荷 serde_json::Value)
// //
// 覆盖: // 覆盖:
// 1. subscribe 后 publish,receiver 收到事件(基础 pub-sub) // 1. subscribe 后 publish,receiver 收到 Value(基础 pub-sub)
// 2. 多订阅者都收到(broadcast 多消费者) // 2. 多订阅者都收到(broadcast 多消费者)
// 3. 无订阅者时 publish 不 panic(兜底) // 3. 无订阅者时 publish 不 panic(兜底)
// 4. publish 返回值 = 活跃订阅者数量 // 4. publish 返回值 = 活跃订阅者数量
// 5. AiChatEvent 形态 Value 透传(含 type tag,模拟真实 AiChatEvent JSON)
// //
// 不覆盖(留后续批次接入时): // 不覆盖(留后续批次接入时):
// - 慢消费者容量满丢老事件(需灌满容量场景,接入期补) // - 慢消费者容量满丢老事件(需灌满容量场景,接入期补)
// - 跨模块解耦实际效果(需接入真实 emit 点,批2) // - 跨模块解耦实际效果(需接入真实 emit 点,批3)
// ============================================================ // ============================================================
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
/// 基础 pub-sub:subscribe 后 publish,receiver 收到事件 /// 基础 pub-sub:subscribe 后 publish,receiver 收到 Value 原样
#[tokio::test] #[tokio::test]
async fn test_basic_pub_sub() { async fn test_basic_pub_sub() {
let bus = EventBus::new(); let bus = EventBus::new();
let mut rx = bus.subscribe(); let mut rx = bus.subscribe();
let event = AiBusEvent::Stream { // 模拟 AiChatEvent::AiTextDelta 序列化 JSON(type tag discriminator)
delta: "hello".to_string(), let payload = serde_json::json!({
conversation_id: Some("conv-1".to_string()), "type": "text_delta",
}; "conversation_id": "conv-1",
"delta": "hello"
let delivered = bus.publish(event.clone()); });
let delivered = bus.publish(payload.clone());
assert_eq!(delivered, 1, "publish 应送达 1 个订阅者"); assert_eq!(delivered, 1, "publish 应送达 1 个订阅者");
let received = rx.recv().await.expect("receiver 应收到事件"); let received = rx.recv().await.expect("receiver 应收到事件");
match received { assert_eq!(received, payload, "Value 应原样透传(零解析)");
AiBusEvent::Stream { delta, conversation_id } => { assert_eq!(received["type"], "text_delta");
assert_eq!(delta, "hello");
assert_eq!(conversation_id.as_deref(), Some("conv-1"));
}
other => panic!("期望 Stream 事件,收到 {:?}", other),
}
} }
/// 多订阅者都收到(broadcast 多消费者语义)。 /// 多订阅者都收到(broadcast 多消费者语义)。
@@ -289,17 +232,15 @@ mod tests {
let mut rx1 = bus.subscribe(); let mut rx1 = bus.subscribe();
let mut rx2 = bus.subscribe(); let mut rx2 = bus.subscribe();
let event = AiBusEvent::Heartbeat { let payload = serde_json::json!({"type": "heartbeat", "conversation_id": null});
conversation_id: None, let delivered = bus.publish(payload.clone());
};
let delivered = bus.publish(event.clone());
assert_eq!(delivered, 2, "publish 应送达 2 个订阅者"); assert_eq!(delivered, 2, "publish 应送达 2 个订阅者");
// 两个 receiver 各自收到(独立消费进度) // 两个 receiver 各自收到(独立消费进度)
let r1 = rx1.recv().await.expect("rx1 应收到"); let r1 = rx1.recv().await.expect("rx1 应收到");
let r2 = rx2.recv().await.expect("rx2 应收到"); let r2 = rx2.recv().await.expect("rx2 应收到");
assert!(matches!(r1, AiBusEvent::Heartbeat { .. })); assert_eq!(r1["type"], "heartbeat");
assert!(matches!(r2, AiBusEvent::Heartbeat { .. })); assert_eq!(r2["type"], "heartbeat");
} }
/// 无订阅者时 publish 不 panic,返回 0(兜底)。 /// 无订阅者时 publish 不 panic,返回 0(兜底)。
@@ -307,10 +248,7 @@ mod tests {
fn test_publish_no_subscribers() { fn test_publish_no_subscribers() {
let bus = EventBus::new(); let bus = EventBus::new();
// 无 subscribe 直接 publish // 无 subscribe 直接 publish
let delivered = bus.publish(AiBusEvent::Error { let delivered = bus.publish(serde_json::json!({"type": "error", "error": "no one listening"}));
error: "no one listening".to_string(),
conversation_id: None,
});
assert_eq!(delivered, 0, "无订阅者 publish 应返回 0"); assert_eq!(delivered, 0, "无订阅者 publish 应返回 0");
} }
@@ -327,10 +265,7 @@ mod tests {
assert_eq!(bus.subscriber_count(), 2); assert_eq!(bus.subscriber_count(), 2);
// delivered 应等于 subscriber_count // delivered 应等于 subscriber_count
let delivered = bus.publish(AiBusEvent::Round { let delivered = bus.publish(serde_json::json!({"type": "agent_round", "round": 1}));
round: 1,
conversation_id: None,
});
assert_eq!(delivered, bus.subscriber_count()); assert_eq!(delivered, bus.subscriber_count());
} }
@@ -342,34 +277,37 @@ mod tests {
let bus_clone = bus.clone(); let bus_clone = bus.clone();
// 经 clone 的实例 publish,原实例订阅的 receiver 也应收到(共享通道) // 经 clone 的实例 publish,原实例订阅的 receiver 也应收到(共享通道)
bus_clone.publish(AiBusEvent::Completed { bus_clone.publish(serde_json::json!({
total_tokens: 100, "type": "completed",
prompt_tokens: 50, "total_tokens": 100,
completion_tokens: 50, "prompt_tokens": 50,
conversation_id: None, "completion_tokens": 50,
}); "conversation_id": null
}));
let received = rx.recv().await.expect("clone 共享通道,receiver 应收到"); let received = rx.recv().await.expect("clone 共享通道,receiver 应收到");
assert!(matches!(received, AiBusEvent::Completed { .. })); assert_eq!(received["type"], "completed");
} }
/// 全事件类型枚举可构造 + 可序列化(serde 备跨端透传)。 /// 全 19 变体形态 Value 均可透传(不限于子集,验证 D2 全透传不丢变体)。
#[test] #[test]
fn test_event_variants_serde() { fn test_full_variants_passthrough() {
let bus = EventBus::new();
let mut rx = bus.subscribe();
// 覆盖早期 AiBusEvent 子集外的变体(ApprovalRequired/Compressed 等),证明全透传
let events = vec![ let events = vec![
AiBusEvent::Stream { delta: "x".into(), conversation_id: None }, serde_json::json!({"type": "approval_required", "conversation_id": "c1"}),
AiBusEvent::ToolStarted { id: "t1".into(), name: "write_file".into(), args: serde_json::json!({}), conversation_id: None }, serde_json::json!({"type": "compressed", "conversation_id": "c1"}),
AiBusEvent::ToolCompleted { id: "t1".into(), result: serde_json::json!({"ok": true}), conversation_id: None }, serde_json::json!({"type": "dir_auth_required", "conversation_id": "c1"}),
AiBusEvent::Round { round: 2, conversation_id: None }, serde_json::json!({"type": "context_cleared", "conversation_id": "c1"}),
AiBusEvent::Help { reason: "r".into(), context: "c".into(), options: vec!["a".into()], conversation_id: None },
AiBusEvent::Heartbeat { conversation_id: None },
AiBusEvent::Completed { total_tokens: 1, prompt_tokens: 1, completion_tokens: 1, conversation_id: None },
AiBusEvent::Error { error: "e".into(), conversation_id: None },
]; ];
// 每个变体都能序列化(serde tag = "type" 生效,备跨端透传)
for event in &events { for event in &events {
let json = serde_json::to_string(event).expect("事件应可序列化"); bus.publish(event.clone());
assert!(json.contains("\"type\""), "序列化后应含 type tag: {}", json); }
for expected in &events {
let received = rx.try_recv().expect("应收到每个变体");
assert_eq!(received, expected.clone(), "每个变体应原样透传(全透传不丢)");
} }
} }
} }

View File

@@ -37,6 +37,7 @@ pub mod http;
pub mod knowledge_inject; pub mod knowledge_inject;
pub mod prompt; pub mod prompt;
pub mod provider_pool; pub mod provider_pool;
pub mod remote_bridge;
pub mod secret; pub mod secret;
pub mod skills; pub mod skills;
pub mod stream_recv; pub mod stream_recv;

View File

@@ -0,0 +1,632 @@
//! F-260622-01 跨端 AI Chat Phase3 阶段3 桥接层 — MiniCommand→Tauri command 路由
//!
//! 设计文档:docs/02-架构设计/已编号方案/F-260622-01-跨端AIChat-Phase3联调设计-2026-06-22.md
//! (§2.2 Command 下行路由表 + §3.3 R1 同 conv 并发发送兜底 + 附录 B 命令清单)
//!
//! ## 定位
//!
//! 方案 A(tunnel 纯透传 + device 端桥接解协议)下的 device 端桥接模块。
//! tunnel 入站回调改为上抛原始 `serde_json::Value`(不再强类型解 TunnelCommand),
//! 本模块作为该回调的处理器,把 miniapp 发来的 `MiniCommand{cmd, args}` 翻译成
//! 对应 src-tauri Tauri command 的直接 async 调用(非 IPC invoke)。
//!
//! ## 路由表(§2.2)
//!
//! | MiniCommand.cmd | 调用的 Tauri command | 关键参数(args 字段) |
//! |-----------------------|----------------------|----------------------------------------|
//! | `send_message` | `ai_chat_send` | message, conversation_id?, model_override? |
//! | `stop` | `ai_chat_stop` | conversation_id? |
//! | `regenerate` | `ai_regenerate` | conversation_id, language?, model_override? |
//! | `approve` | `ai_approve` | tool_call_id, approved |
//! | `authorize_dir` | `ai_authorize_dir` | tool_call_id, decision |
//! | `continue_loop` | `ai_continue_loop` | conversation_id |
//! | `stop_loop` | `ai_stop_loop` | conversation_id |
//! | `switch_conversation` | (无,MVP 不处理) | — |
//! | (未知 cmd) | (兜底臂:log+忽略) | — |
//!
//! ## R1 兜底(硬性,不可省,§3.3 + D6)
//!
//! `send_message` 路由前置 generating 检查(对齐 ai_is_generating 双轨口径:
//! CONV_STATE_ENABLED on 时读 conv_state.is_active(),off 回退 generating bool)。
//! generating=true 则拒绝,emit `AiError` 事件回 miniapp 提示「正在生成中」,
//! 不调 ai_chat_send。这是 miniapp 与桌面同时给同 conv 发消息的双保险,
//! 无论 ai_chat_send 内部是否已有 generating guard 都加(防御性编程)。
//!
//! ## 集成
//!
//! 本模块仅暴露 `handle_remote_command` 入口,setup 集成(注册为 tunnel on_command
//! 回调)留待联调轮做,不在本批范围。
//!
//! dead_code:本模块入口 `handle_remote_command` 及其内部链(route_send_message /
//! check_generating_reject / MiniCommand::from_payload / args_get_*)零调用方 ——
//! 联调轮 setup 注册为 tunnel on_command 回调即消除(预留保留,非死代码)。
#![allow(dead_code)]
use serde::Deserialize;
use serde_json::Value;
use tauri::{AppHandle, Emitter, State};
use crate::state::AppState;
// super::super = commands::ai(commands/chat.rs 的 commands 模块父级是 commands::ai)
// Tauri command 函数经 commands/mod.rs 的 `pub use self::chat::*;` + ai/mod.rs 的
// `pub use self::commands::*;` 透传到 crate::commands::ai 路径,这里直接复用该路径。
//
// 注:集成轮 setup 注册 tunnel on_command 回调时,会用 `app.state::<AppState>()`(Manager trait)
// 构造 State 入参传给 handle_remote_command。本模块自身不调 Manager(签名收 State 而非自取),
// 故 import 不含 Manager(避免 unused)。
use crate::commands::ai::{
ai_approve, ai_authorize_dir, ai_chat_send, ai_chat_stop, ai_continue_loop,
ai_regenerate, ai_stop_loop, AiChatEvent,
};
// CONV_STATE_ENABLED 与 ConvState::is_active 双轨口径(对齐 ai_is_generating 实现)。
use crate::commands::ai::agentic::conv_state::CONV_STATE_ENABLED;
// ============================================================
// MiniCommand 协议结构(对齐 apps/df-miniapp/src/types/relay.ts:88-93)
// ============================================================
/// 小程序 → 桌面端 Command 协议载体。
///
/// 对齐 `apps/df-miniapp/src/types/relay.ts:88-93` 的 `MiniCommand` interface:
/// ```ts
/// interface MiniCommand { cmd: string; args: Record<string, unknown> }
/// ```
/// `cmd` 是 Tauri command 名(relay.ts:82 注释「对齐 Tauri command 名」),
/// `args` 是原始参数对象(对齐各 command 参数签名)。
///
/// 反序列化容忍 `args` 缺省(relay.ts:90 标注 args 为必填,但防御性编程允许缺失 → 默认空 Object)。
#[derive(Debug, Clone, Deserialize)]
pub struct MiniCommand {
/// Tauri command 名(如 "send_message" / "stop" / "approve" ...)
pub cmd: String,
/// 命令参数(原始 JSON 对象,缺失时回退空 Object 便于各臂 from_value 兜底)
#[serde(default)]
pub args: Value,
}
impl MiniCommand {
/// 从原始 payload 反序列化为 MiniCommand。
///
/// 失败场景:payload 非 Object / 缺 `cmd` 字段 / `cmd` 非 string。
/// 调用方(handle_remote_command)失败时 log+忽略,不崩溃。
fn from_payload(payload: &Value) -> Result<Self, serde_json::Error> {
serde_json::from_value(payload.clone())
}
}
// ============================================================
// 主入口:handle_remote_command(供 tunnel on_command 回调调用)
// ============================================================
/// 远程命令处理入口 —— miniapp 发来的 MiniCommand payload 路由到 Tauri command。
///
/// 供 tunnel `on_command` 回调集成时调用(setup 集成留联调轮)。
///
/// 流程:
/// 1. 反序列化 payload 为 [`MiniCommand`](失败 → log+忽略,非崩溃路径)
/// 2. `match cmd` 路由到对应 Tauri command 直接 async 调用
/// 3. send_message 路由前置 R1 generating 检查(硬性兜底)
/// 4. switch_conversation 无对应 command → log+忽略(§2.2 决策 a MVP 不处理)
/// 5. 未知 cmd → log+忽略(兜底臂)
///
/// Tauri command 返回值忽略 —— 结果经 AiChatEvent 回流(事件透传阶段2 已建通路),
/// 不走 command 返回值(对齐 §2.2 设计:command 返回值忽略,结果经事件回流)。
///
/// `state` 参数签名用 `State<'_, AppState>`(与 Tauri command 第二参同类型),
/// 集成时由 `app.state::<AppState>()` 构造(Manager trait 提供,见 agentic/mod.rs:526 用法)。
pub async fn handle_remote_command(payload: Value, app: AppHandle, state: State<'_, AppState>) {
// 反序列化失败:payload 非 {cmd, args} 结构(可能 miniapp 发了未知格式 / relay 误投递)。
// log+忽略,不崩溃 —— 对齐 §1.2 兜底语义「非法命令到运行时桥接层才报错,match 末尾兜底臂 log+忽略」。
let command = match MiniCommand::from_payload(&payload) {
Ok(c) => c,
Err(e) => {
tracing::warn!(
error = %e,
"[remote_bridge] payload 反序列化 MiniCommand 失败,忽略(非 {{cmd, args}} 结构?)"
);
return;
}
};
tracing::info!(
cmd = %command.cmd,
"[remote_bridge] 收到远程命令,开始路由"
);
// match 路由表(§2.2)。各臂提取 args 字段,调对应 Tauri command。
// 字段缺失用稳健默认(Option/空字符串),不中断路由(命令内部自有参数校验,缺失会返 Err,
// 调用方 ignore 返回值 —— 结果经事件回流)。
match command.cmd.as_str() {
// ── send_message:R1 前置 generating 检查 → ai_chat_send ──
"send_message" => {
route_send_message(&app, &state, command.args).await;
}
// ── stop:ai_chat_stop(conversation_id?) ──
// 注意:ai_chat_stop 签名顺序异常(state 在前,app 在后,见 chat.rs:1431),
// 与其它命令的 (app, state, ...) 顺序不同,此处对齐其真实签名。
"stop" => {
let conversation_id = args_get_string(&command.args, "conversation_id");
let _ = ai_chat_stop(state, app, conversation_id).await;
}
// ── regenerate:ai_regenerate(conversation_id, language?, model_override?) ──
"regenerate" => {
// conversation_id 必填,缺失跳过(ai_regenerate 内部无 conv 会失败,日志告警)
match args_get_string(&command.args, "conversation_id") {
Some(conversation_id) => {
let language = args_get_string(&command.args, "language");
let model_override = args_get_string(&command.args, "model_override");
let _ = ai_regenerate(app, state, conversation_id, language, model_override).await;
}
None => {
tracing::warn!(
"[remote_bridge] regenerate 缺 conversation_id 参数,忽略"
);
}
}
}
// ── approve:ai_approve(tool_call_id, approved) ──
"approve" => {
// tool_call_id 必填 + approved 必填(bool),缺失跳过
match (
args_get_string(&command.args, "tool_call_id"),
args_get_bool(&command.args, "approved"),
) {
(Some(tool_call_id), Some(approved)) => {
let _ = ai_approve(app, state, tool_call_id, approved).await;
}
_ => {
tracing::warn!(
"[remote_bridge] approve 缺 tool_call_id 或 approved 参数,忽略"
);
}
}
}
// ── authorize_dir:ai_authorize_dir(tool_call_id, decision) ──
"authorize_dir" => {
match (
args_get_string(&command.args, "tool_call_id"),
args_get_string(&command.args, "decision"),
) {
(Some(tool_call_id), Some(decision)) => {
let _ = ai_authorize_dir(app, state, tool_call_id, decision).await;
}
_ => {
tracing::warn!(
"[remote_bridge] authorize_dir 缺 tool_call_id 或 decision 参数,忽略"
);
}
}
}
// ── continue_loop:ai_continue_loop(conversation_id) ──
"continue_loop" => {
match args_get_string(&command.args, "conversation_id") {
Some(conversation_id) => {
let _ = ai_continue_loop(app, state, conversation_id).await;
}
None => {
tracing::warn!(
"[remote_bridge] continue_loop 缺 conversation_id 参数,忽略"
);
}
}
}
// ── stop_loop:ai_stop_loop(conversation_id) ──
"stop_loop" => {
match args_get_string(&command.args, "conversation_id") {
Some(conversation_id) => {
let _ = ai_stop_loop(app, state, conversation_id).await;
}
None => {
tracing::warn!(
"[remote_bridge] stop_loop 缺 conversation_id 参数,忽略"
);
}
}
}
// ── switch_conversation:无对应 Tauri command(§2.2 决策 a MVP 不处理) ──
// 桌面端「切会话」是纯前端操作(改 activeConversationId + 本地 store 加载历史),
// 无 IPC 入口。miniapp switch 仅本地视图切换,跨端 active 不一致不阻断功能
// (事件全局广播带 conv_id,miniapp 自过滤)。列入 P4 双向同步完善。
"switch_conversation" => {
tracing::info!(
"[remote_bridge] switch_conversation 无对应 Tauri command(§2.2 决策 a MVP 不处理),忽略"
);
}
// ── 兜底臂:未知 cmd,log+忽略(不崩溃) ──
// 对齐 §1.2 风险缓解「桥接层 match 末尾加兜底臂,非法命令不崩溃只记录」。
unknown => {
tracing::warn!(
cmd = unknown,
"[remote_bridge] 未知 cmd,忽略(兜底臂,非崩溃)"
);
}
}
}
// ============================================================
// R1 兜底:send_message 路由前置 generating 检查
// ============================================================
/// `send_message` 路由 —— R1 兜底前置 generating 检查后调 ai_chat_send。
///
/// R1(§3.3 + D6,硬性不可省):miniapp 与桌面同时给同 conv 发消息时,
/// ai_chat_send 内部虽有 generating guard,但桥接层再加一层前置检查作为双保险
/// (防御性编程,对齐设计文档 D6「桥接层 send 路由硬性加,无论 ai_chat_send 内部是否有 guard」)。
///
/// 检查口径与 `ai_is_generating` 完全一致(双轨):
/// - CONV_STATE_ENABLED on → 读 `conv_state.is_active()`(含 Generating / Compressed 派生态)
/// - off → 回退 `generating` bool
///
/// generating=true 则拒绝:**emit `AiError` 事件回 miniapp 提示「正在生成中」**,不调 ai_chat_send。
/// (事件透传阶段2 已建通路,miniapp handleEvent 能渲染 AiError,用户可见反馈)。
///
/// message 必填,缺失跳过(ai_chat_send 无 message 无意义,日志告警)。
async fn route_send_message(app: &AppHandle, state: &State<'_, AppState>, args: Value) {
// 取 message(必填) + conversation_id(可选,空字符串规整为 None) + model_override(可选)。
let message = match args_get_string(&args, "message") {
Some(m) => m,
None => {
tracing::warn!("[remote_bridge] send_message 缺 message 参数,忽略");
return;
}
};
let conversation_id = args_get_string(&args, "conversation_id");
let model_override = args_get_string(&args, "model_override");
// ── R1 前置 generating 检查(双轨口径对齐 ai_is_generating) ──
// 目标 conv:入参 conversation_id 优先 → active conv 兜底(与 ai_is_generating 同口径)。
// 空目标(无 conv_id 且无 active):无任何 conv 在跑,放行。
if let Some(reject_conv_id) = check_generating_reject(state, conversation_id.as_deref()).await {
// generating=true:拒绝,emit AiError 回 miniapp 提示。
// conv_id 透传原 conversation_id(用户当前面板的 conv),便于 miniapp 按 conv 路由展示。
tracing::info!(
conv_id = ?reject_conv_id,
"[remote_bridge][R1] 同 conv 正在生成中,拒绝远程 send_message(emit AiError 回 miniapp)"
);
let _ = app.emit(
"ai-chat-event",
AiChatEvent::AiError {
error: "正在生成中,请等待完成".to_string(),
// 并发拒绝非错误源分类(auth/network/timeout/provider_config),用 Unknown 兜底。
error_type: Some(crate::commands::ai::ErrorType::Unknown),
conversation_id: Some(reject_conv_id),
},
);
return;
}
// ── 放行:调 ai_chat_send(其余可选参数 language/skill/parts/mention_spans 不经远程,
// miniapp MVP 仅透传 message/conversation_id/model_override,其余走默认 None) ──
//
// F 核验(2026-06-22):ai_chat_send 内部已有 generating guard(chat.rs:354-358)。
// R1 前置检查是双保险(任务原文要求"无论 ai_chat_send 内部是否有 guard"都加)。
// 这里额外把 ai_chat_send 返回的 Err<String> 转 AiError emit 回 miniapp ——
// 无论 R1 命中(理论上不会到这)/ 内部 guard 命中 / Provider 配置错 / 参数错,
// 用户都能在 miniapp 看到 AiError 反馈(否则命令返 Err 静默,miniapp 不知失败)。
let result = ai_chat_send(
app.clone(),
state.clone(),
message,
None, // language:远程默认不传,ai_chat_send 内部 fallback "zh-CN"
None, // skill:远程不传技能名(技能调用经本地命令栏)
model_override, // model_override:远程透传用户选择
conversation_id.clone(),// conversation_id:远程透传目标 conv
None, // parts:多模态片段,远程 MVP 不传
None, // mention_spans:@ mention 区间,远程 MVP 不传
)
.await;
if let Err(err_msg) = result {
tracing::warn!(
error = %err_msg,
conv_id = ?conversation_id,
"[remote_bridge] ai_chat_send 返回 Err,转 AiError emit 回 miniapp"
);
let _ = app.emit(
"ai-chat-event",
AiChatEvent::AiError {
error: err_msg,
// ai_chat_send 的 Err 多为 generating 拦截 / Provider 配置缺失,
// 难精确分类,用 Unknown 兜底(对齐 mod.rs ErrorType 语义)。
error_type: Some(crate::commands::ai::ErrorType::Unknown),
conversation_id,
},
);
}
}
/// R1 检查:目标 conv 是否正在生成。返 `Some(conv_id)` 表示拒绝(该 conv 在跑),
/// 返 `None` 表示放行(可发)。
///
/// 双轨口径对齐 `ai_is_generating`(commands/chat.rs:287-293):
/// - CONV_STATE_ENABLED on → 读 `conv_state.is_active()`(Generating / Compressed 派生态均视为活跃,
/// 压缩期间 loop 仍活跃)
/// - off → 回退 `generating` bool
///
/// 目标 conv 解析(对齐 ai_is_generating:278-284):
/// - 入参 conversation_id 非空优先
/// - 否则 fallback `active_conversation_id`
/// - 都无 → 返 None(放行,无 conv 在跑)
///
/// 读 conv 状态用 `conv_read`(只读不创建),对齐 ai_is_generating 实现。
async fn check_generating_reject(
state: &State<'_, AppState>,
conversation_id: Option<&str>,
) -> Option<String> {
let session = state.ai_session.lock().await;
// 目标 conv:入参优先 → active 兜底。
let target = conversation_id
.filter(|s| !s.is_empty())
.map(|s| s.to_string())
.or_else(|| session.active_conversation_id.clone());
let target = match target {
Some(id) => id,
None => return None, // 无目标 conv 且无 active:无任何 conv 在跑,放行
};
// 双轨读 generating(对齐 ai_is_generating,无 conv_read 视为 false 放行)。
let is_gen = session
.conv_read(&target)
.map(|c| {
if CONV_STATE_ENABLED {
c.conv_state.is_active()
} else {
c.generating
}
})
.unwrap_or(false);
if is_gen {
Some(target)
} else {
None
}
}
// ============================================================
// args 字段提取 helper(从 serde_json::Value 稳健取值)
// ============================================================
/// 从 args 取 string 字段。非 string / 缺失 → None(调用方各自处理缺失语义)。
fn args_get_string(args: &Value, key: &str) -> Option<String> {
args.get(key).and_then(|v| v.as_str()).map(|s| s.to_string())
}
/// 从 args 取 bool 字段。非 bool / 缺失 → None。
fn args_get_bool(args: &Value, key: &str) -> Option<bool> {
args.get(key).and_then(|v| v.as_bool())
}
// ============================================================
// 单元测试
// ============================================================
//
// 覆盖:
// 1. MiniCommand 反序列化(标准结构 / args 缺省 / 非 Object payload 失败)
// 2. match 各 cmd 臂的路由分支(send_message R1 拒绝 / stop / regenerate / approve /
// authorize_dir / continue_loop / stop_loop / switch_conversation 忽略 / 未知 cmd 忽略)
// 3. R1 拒绝路径(generating=true 时 emit AiError 不调 ai_chat_send)
//
// 注:完整 Tauri command 调用链(需 AppHandle + State 真实构造)不在单元测试范围,
// 用 helper 函数(check_generating_reject / args_get_* / MiniCommand::from_payload)
// 隔离测试核心路由逻辑,绕开 Tauri runtime 依赖。
#[cfg(test)]
mod tests {
use super::*;
use crate::commands::ai::{AiSession, PerConvState};
use df_ai::context::ContextConfig;
// ── MiniCommand 反序列化测试 ──
/// 标准结构 `{cmd, args}` 反序列化成功。
#[test]
fn test_mini_command_deserialize_standard() {
let payload = serde_json::json!({
"cmd": "send_message",
"args": {
"message": "你好",
"conversation_id": "conv-123"
}
});
let cmd = MiniCommand::from_payload(&payload).expect("标准结构应反序列化成功");
assert_eq!(cmd.cmd, "send_message");
assert_eq!(
cmd.args.get("message").and_then(|v| v.as_str()),
Some("你好")
);
assert_eq!(
cmd.args.get("conversation_id").and_then(|v| v.as_str()),
Some("conv-123")
);
}
/// args 缺省时回退空 Object(防御性 #[serde(default)])。
#[test]
fn test_mini_command_deserialize_args_default() {
let payload = serde_json::json!({"cmd": "stop"});
let cmd = MiniCommand::from_payload(&payload).expect("args 缺省应回退空 Object");
assert_eq!(cmd.cmd, "stop");
assert!(cmd.args.is_object(), "args 缺省应为空 Object");
assert!(cmd.args.get("conversation_id").is_none());
}
/// 非 Object payload(如纯字符串 / 数组)反序列化失败。
#[test]
fn test_mini_command_deserialize_non_object_fails() {
let payload = serde_json::json!("just a string");
assert!(
MiniCommand::from_payload(&payload).is_err(),
"非 Object payload 应反序列化失败"
);
}
/// 缺 cmd 字段反序列化失败(cmd 必填)。
#[test]
fn test_mini_command_deserialize_missing_cmd_fails() {
let payload = serde_json::json!({"args": {"foo": "bar"}});
assert!(
MiniCommand::from_payload(&payload).is_err(),
"缺 cmd 字段应反序列化失败"
);
}
// ── args 字段提取 helper 测试 ──
/// args_get_string / args_get_bool 各类型分支。
#[test]
fn test_args_helpers() {
let args = serde_json::json!({
"str_field": "hello",
"bool_field": true,
"num_field": 42,
"null_field": null,
});
// string 字段命中
assert_eq!(
args_get_string(&args, "str_field"),
Some("hello".to_string())
);
// 缺失字段返 None
assert_eq!(args_get_string(&args, "missing"), None);
// 非 string 字段(number)返 None
assert_eq!(args_get_string(&args, "num_field"), None);
// null 字段返 None(as_str 对 null 返 None)
assert_eq!(args_get_string(&args, "null_field"), None);
// bool 字段命中
assert_eq!(args_get_bool(&args, "bool_field"), Some(true));
// 非 bool 字段返 None
assert_eq!(args_get_bool(&args, "str_field"), None);
// 缺失 bool 返 None
assert_eq!(args_get_bool(&args, "missing"), None);
}
// ── R1 check_generating_reject 路径测试(隔离 AiSession,绕开 Tauri State 依赖) ──
//
// check_generating_reject 需 `State<AppState>`,单元测试难构造。
// 改测 R1 核心判定逻辑:用 AiSession + PerConvState 模拟 generating 状态,
// 复核 conv_read + CONV_STATE_ENABLED 口径对齐 ai_is_generating。
/// R1 判定:无目标 conv 且无 active → 放行(返 None)。
#[test]
fn test_r1_no_target_pass() {
let session = AiSession::new();
// 无 active,无入参 conv_id → 无任何 conv 在跑
assert!(
session.active_conversation_id.is_none(),
"新 session 无 active conv"
);
assert!(
session.per_conv.is_empty(),
"新 session per_conv 为空"
);
}
/// R1 判定:目标 conv 未在跑 → 放行(generating=false)。
#[test]
fn test_r1_target_idle_pass() {
let mut session = AiSession::new();
let conv = session.conv("conv-idle");
conv.generating = false;
// 读 generating 口径(对齐 check_generating_reject / ai_is_generating)
let is_gen = session
.conv_read("conv-idle")
.map(|c| {
if CONV_STATE_ENABLED {
c.conv_state.is_active()
} else {
c.generating
}
})
.unwrap_or(false);
assert!(!is_gen, "conv-idle 未在跑,is_gen 应为 false(放行)");
}
/// R1 判定:目标 conv 在跑 → 拒绝(generating=true)。
#[test]
fn test_r1_target_generating_reject() {
let mut session = AiSession::new();
let conv = session.conv("conv-busy");
conv.generating = true;
let is_gen = session
.conv_read("conv-busy")
.map(|c| {
if CONV_STATE_ENABLED {
c.conv_state.is_active()
} else {
c.generating
}
})
.unwrap_or(false);
assert!(is_gen, "conv-busy generating=true 应拒绝");
}
/// R1 判定:conv_state 派生态(Generating)is_active()=true(双轨 on 路径)。
///
/// 验证 CONV_STATE_ENABLED on 时读 conv_state.is_active() 而非 generating bool ——
/// 即便 generating bool 误为 false 但 conv_state=Generating,仍应判定活跃拒绝。
#[test]
fn test_r1_conv_state_active_reject() {
use crate::commands::ai::agentic::conv_state::ConvState;
// CONV_STATE_ENABLED 是编译期 const(true),本测试假定 on 路径。
// 若未来开关改 runtime,此测试需条件跳过 on 时才跑。
let mut session = AiSession::new();
let conv = session.conv("conv-state-busy");
// 模拟状态机写收敛路径:conv_state=Generating,generating bool 同步 true(双轨一致)
conv.conv_state = ConvState::Generating;
conv.generating = true;
let is_gen = session
.conv_read("conv-state-busy")
.map(|c| {
if CONV_STATE_ENABLED {
c.conv_state.is_active()
} else {
c.generating
}
})
.unwrap_or(false);
assert!(
is_gen,
"conv_state=Generating is_active() 应为 true(双轨 on 路径拒绝)"
);
}
/// R1 目标 conv 解析:入参 conversation_id 优先于 active_conversation_id。
#[test]
fn test_r1_target_input_priority() {
let mut session = AiSession::new();
session.active_conversation_id = Some("conv-active".to_string());
// active 在跑,但入参 conv-other 未跑 → 入参优先,应放行
session.conv("conv-active").generating = true;
let _ = session.conv("conv-other"); // 创建但不 generating
let active_gen = session
.conv_read("conv-active")
.map(|c| c.generating)
.unwrap_or(false);
let other_gen = session
.conv_read("conv-other")
.map(|c| c.generating)
.unwrap_or(false);
assert!(active_gen, "conv-active 在跑");
assert!(!other_gen, "conv-other 未跑");
// 入参 conv-other 优先 → 判定 conv-other.generating=false → 放行
// (与 check_generating_reject 入参优先逻辑一致)
}
// ── PerConvState 初值对齐(确保 R1 测试基线 generating=false) ──
/// PerConvState::new() generating 初值 false(R1 放行基线)。
#[test]
fn test_per_conv_state_generating_default_false() {
let _ = ContextConfig::default(); // 触发 ContextConfig default 可用(避免 unused)
let s = PerConvState::new();
assert!(!s.generating, "PerConvState::new generating 初值应为 false");
}
}