重构: aichat agent 能力系统化(L1元能力+L2/L3后端+list去重)
L1 agent 元能力层(治痛①死循环零交付): - env_profile 环境姿势注入 + shell 默认 PowerShell(防引号地狱) - 断路器:同类工具失败≥3熔断 + guard.reset - detect_environment 主动探测工具(python/node/shell) - 求助协议 AiHelpRequired 事件 + 前端求助卡 L2 统一状态机后端(治痛②,前端批2): - ConvState enum 5态 + 合法转换守卫(conv_state.rs) - GeneratingGuard 接入视图层(guard.rs) L3 事件总线后端骨架(治痛③④⑤,接入批2): - EventBus pub-sub + AiBusEvent 8变体(event_bus.rs) list 工具调用重复治理第一步: - build_system_prompt_with_excluded 去重被@实体 + 清单注明语
This commit is contained in:
@@ -0,0 +1,375 @@
|
||||
//! 全局事件数据总线 — pub-sub 骨架(L3 阶段1,渐进第一步)
|
||||
//!
|
||||
//! 关联设计:docs/02-架构设计/专项设计/全局事件数据总线-2026-06-21.md
|
||||
//!
|
||||
//! 本模块是 **统一事件数据总线(Unified Event Bus)** 的后端骨架,聚焦 **pub-sub**(发布订阅)
|
||||
//! 语义。背景:当前 AI 事件经 `app_handle.emit("ai-chat-event", ...)` 散布在各处,无统一总线,
|
||||
//! 跨模块直接耦合,阻碍 ai-working 可插拔 + 跨端透传。
|
||||
//!
|
||||
//! ## 本批范围(L3 骨架,纯新增无行为变化)
|
||||
//! - `EventBus` 结构(基于 tokio::sync::broadcast,容量 256)
|
||||
//! - `AiBusEvent` 强类型 enum(覆盖关键事件:Stream/Tool/Round/Help/Heartbeat/Completed/Error,
|
||||
//! 对齐现有 `AiChatEvent` 子集)
|
||||
//! - `subscribe()` / `publish()` 基础接口
|
||||
//! - `EVENT_BUS_ENABLED` 开关(预留,默认 on —— 总线骨架自带,但接入实际事件源时此开关控制是否经总线)
|
||||
//! - pub/sub 基础单元测试
|
||||
//!
|
||||
//! ## 不做(留后续批次)
|
||||
//! - 不接入 `agentic/mod.rs` 各 emit 点(B 域,留批2接入)
|
||||
//! - 不入 `AppState`(接入留批2)
|
||||
//! - 不碰前端订阅渲染(留批2 Tauri bridge)
|
||||
//! - 不实现 request-reply / 流式 reply(设计阶段2/5,本批仅 pub-sub)
|
||||
//! - 不做跨端透传(df-tunnel adapter,阶段6)
|
||||
//!
|
||||
//! ## 开关与兜底(渐进可回退)
|
||||
//! - `EVENT_BUS_ENABLED: bool`(默认 true):接入事件源后,关闭则 publish 静默丢弃(不影响原 emit 路径)。
|
||||
//! 骨架阶段未接入,无实际效果,预留作接入期的快速回退开关。
|
||||
//! - publish 返回 `usize`(接收者数量),调用方可忽略;无接收者时不报错(broadcast 语义)。
|
||||
//! - 慢消费者(broadcast 满):`send` 返 `Err(SendError)`,本骨架 `publish` 静默丢弃并返回 0,
|
||||
//! 不 panic(对齐设计待决策6「背压/限流」的保守默认,正式背压策略留后续)。
|
||||
//!
|
||||
//! ## 与 df-workflow EventBus 关系
|
||||
//! df-workflow 的 `EventBus`(`crates/df-workflow/src/eventbus.rs`)是工作流节点间专用总线
|
||||
//! (仅 `WorkflowEvent`),本总线是面向 AI 域的通用 pub-sub(对齐 `AiChatEvent` 子集)。
|
||||
//! 设计阶段1 计划「df-workflow EventBus 包装为通用 DomainEvent」,本骨架先建 AI 域总线,
|
||||
//! 通用 DomainEvent 泛化留后续(待决策1 bus crate 位置 + 待决策2 事件 schema)。
|
||||
|
||||
use tokio::sync::broadcast;
|
||||
|
||||
// ============================================================
|
||||
// 开关(EVENT_BUS_ENABLED,预留,默认 on)
|
||||
//
|
||||
// 机制:接入事件源后,关闭则 EventBus::publish 静默丢弃事件(原 app.emit 路径不受影响,
|
||||
// 双写桥接由接入批负责,关闭总线仅丢总线副本)。骨架阶段未接入,开关预留作接入期快速回退。
|
||||
//
|
||||
// 兜底:即便开关误关,总线静默丢事件不影响原 emit 路径(AiChatEvent 经 app.emit 仍正常推送前端)。
|
||||
// ============================================================
|
||||
|
||||
/// 事件总线开关(预留,默认 on)。
|
||||
///
|
||||
/// 接入实际事件源后,`EventBus::publish` 在 `false` 时静默丢弃(返回 0),原 `app.emit`
|
||||
/// 路径不受影响。骨架阶段未接入,无实际效果。
|
||||
///
|
||||
/// dead_code 说明:本批为骨架,EVENT_BUS_ENABLED 暂无消费方(未接入 emit 点);
|
||||
/// 标 allow 保留作批2接入期的快速回退开关(零调用方≠垃圾,预留保留)。
|
||||
#[allow(dead_code)]
|
||||
pub const EVENT_BUS_ENABLED: bool = true;
|
||||
|
||||
/// 默认事件通道容量(对齐 df-workflow EventBus 默认 256)。
|
||||
///
|
||||
/// 容量权衡:太小 → 慢消费者丢事件(订阅者消费不及时);太大 → 内存占用。
|
||||
/// 256 是 broadcast 常见默认值,覆盖典型 AI 流场景(文本片段 + 工具调用 + 心跳混合)。
|
||||
///
|
||||
/// dead_code 说明:骨架阶段 EventBus::new() 内联用字面量路径常量,但常量本身供
|
||||
/// 外部自定义容量场景引用,标 allow 保留(预留)。
|
||||
#[allow(dead_code)]
|
||||
pub const DEFAULT_BUS_CAPACITY: usize = 256;
|
||||
|
||||
// ============================================================
|
||||
// AiBusEvent — 总线事件类型(对齐 AiChatEvent 子集)
|
||||
//
|
||||
// 设计取舍(对齐设计文档待决策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:
|
||||
// - pub-sub 多订阅者(mpsc 单消费者不满足「多模块订阅同一事件」场景)
|
||||
// - broadcast 容量固定,慢消费者丢老事件而非阻塞发布者(对齐流式场景:宁可丢老 chunk 不阻塞 LLM 流)
|
||||
//
|
||||
// Clone 语义:broadcast::Sender 内部 Arc,clone 共享同一通道(多持有者 publish 到同一总线)。
|
||||
// 对齐设计 §架构「模块订阅 + 发布,无相互 import」—— 各模块持 clone 的 EventBus 即可 pub/sub。
|
||||
// ============================================================
|
||||
|
||||
/// AI 事件总线(pub-sub 域,L3 阶段1 骨架)。
|
||||
///
|
||||
/// 基于 `tokio::sync::broadcast`,多订阅者多发布者共享同一通道。
|
||||
/// - `subscribe()`:订阅,返回 [`broadcast::Receiver`] recv [`AiBusEvent`]
|
||||
/// - `publish(event)`:发布,所有活跃订阅者收到(broadcast 语义)
|
||||
///
|
||||
/// 容量默认 256(`DEFAULT_BUS_CAPACITY`),可经 [`EventBus::with_capacity`] 自定义。
|
||||
///
|
||||
/// 兜底:
|
||||
/// - 无订阅者时 publish 不报错(broadcast 语义,事件丢弃)
|
||||
/// - 慢消费者(容量满)publish 返 0(静默丢老事件,不 panic)
|
||||
/// - `EVENT_BUS_ENABLED=false` 时 publish 静默丢弃(预留开关,接入期回退用)
|
||||
///
|
||||
/// dead_code 说明:骨架阶段未入 AppState、未在 emit 点构造(批2接入),标 allow 保留
|
||||
/// (总线核心结构,批2接入 AppState 即消费)。
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct EventBus {
|
||||
sender: broadcast::Sender<AiBusEvent>,
|
||||
}
|
||||
|
||||
/// 事件订阅者(broadcast::Receiver,AiBusEvent 接收端)。
|
||||
///
|
||||
/// dead_code 说明:骨架阶段无外部消费方(批2接入订阅),标 allow 保留。
|
||||
#[allow(dead_code)]
|
||||
pub type EventSubscriber = broadcast::Receiver<AiBusEvent>;
|
||||
|
||||
// dead_code 说明(impl 块):骨架阶段 EventBus 未入 AppState、无外部调用方
|
||||
// (批2接入 AppState + emit 点后立即消费)。标 allow 覆盖 new/with_capacity/
|
||||
// subscribe/publish/subscriber_count 全部关联项的 never-used 警告。
|
||||
// 对齐零调用方原则:预留保留不盲删(批2接入即消除)。
|
||||
#[allow(dead_code)]
|
||||
impl EventBus {
|
||||
/// 创建默认容量(256)的事件总线。
|
||||
pub fn new() -> Self {
|
||||
Self::with_capacity(DEFAULT_BUS_CAPACITY)
|
||||
}
|
||||
|
||||
/// 创建指定容量的事件总线。
|
||||
///
|
||||
/// capacity 为 0 会 panic(broadcast::channel 要求 capacity ≥ 1)。
|
||||
pub fn with_capacity(capacity: usize) -> Self {
|
||||
let (sender, _) = broadcast::channel(capacity);
|
||||
Self { sender }
|
||||
}
|
||||
|
||||
/// 订阅事件总线,返回 Receiver。
|
||||
///
|
||||
/// 订阅后仅收到订阅时刻之后的 publish(历史事件不补发)。可多次 subscribe 获多个独立 Receiver,
|
||||
/// 每个 Receiver 各自维护消费进度(broadcast 多消费者语义)。
|
||||
pub fn subscribe(&self) -> EventSubscriber {
|
||||
self.sender.subscribe()
|
||||
}
|
||||
|
||||
/// 发布事件到总线。
|
||||
///
|
||||
/// 返回值:成功送达的活跃订阅者数量(0 = 无订阅者或容量满丢老事件)。
|
||||
/// `EVENT_BUS_ENABLED=false` 时静默丢弃返回 0(预留开关,骨架阶段未接入)。
|
||||
///
|
||||
/// 注:broadcast::send 是同步方法(非 async),与 df-workflow EventBus::send 的 async 签名
|
||||
/// 不同 —— 本骨架 publish 同步返回更直接(broadcast::send 内部无 await 点),df-workflow 的
|
||||
/// async 签名是为对齐 trait 抽象,本总线无此约束故同步。
|
||||
pub fn publish(&self, event: AiBusEvent) -> usize {
|
||||
if !EVENT_BUS_ENABLED {
|
||||
return 0;
|
||||
}
|
||||
// broadcast::send 返 Result<usize, SendError>:
|
||||
// Ok(n) = 送达 n 个活跃订阅者
|
||||
// Err(SendError(_)) = 无订阅者(事件丢弃)
|
||||
// 两种情况都不 panic,Err 视为 0(对齐「无订阅者静默丢弃」兜底)。
|
||||
self.sender.send(event).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// 当前活跃订阅者数量(broadcast::receiver_count)。
|
||||
///
|
||||
/// 诊断/监控用:接入期可观测订阅者是否到位。骨架阶段无实际调用,预留作可观测性扩展。
|
||||
pub fn subscriber_count(&self) -> usize {
|
||||
self.sender.receiver_count()
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for EventBus {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================
|
||||
// 单元测试 — pub/sub 基础语义
|
||||
//
|
||||
// 覆盖:
|
||||
// 1. subscribe 后 publish,receiver 收到事件(基础 pub-sub)
|
||||
// 2. 多订阅者都收到(broadcast 多消费者)
|
||||
// 3. 无订阅者时 publish 不 panic(兜底)
|
||||
// 4. publish 返回值 = 活跃订阅者数量
|
||||
//
|
||||
// 不覆盖(留后续批次接入时):
|
||||
// - 慢消费者容量满丢老事件(需灌满容量场景,接入期补)
|
||||
// - 跨模块解耦实际效果(需接入真实 emit 点,批2)
|
||||
// ============================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// 基础 pub-sub:subscribe 后 publish,receiver 收到事件。
|
||||
#[tokio::test]
|
||||
async fn test_basic_pub_sub() {
|
||||
let bus = EventBus::new();
|
||||
let mut rx = bus.subscribe();
|
||||
|
||||
let event = AiBusEvent::Stream {
|
||||
delta: "hello".to_string(),
|
||||
conversation_id: Some("conv-1".to_string()),
|
||||
};
|
||||
|
||||
let delivered = bus.publish(event.clone());
|
||||
assert_eq!(delivered, 1, "publish 应送达 1 个订阅者");
|
||||
|
||||
let received = rx.recv().await.expect("receiver 应收到事件");
|
||||
match received {
|
||||
AiBusEvent::Stream { delta, conversation_id } => {
|
||||
assert_eq!(delta, "hello");
|
||||
assert_eq!(conversation_id.as_deref(), Some("conv-1"));
|
||||
}
|
||||
other => panic!("期望 Stream 事件,收到 {:?}", other),
|
||||
}
|
||||
}
|
||||
|
||||
/// 多订阅者都收到(broadcast 多消费者语义)。
|
||||
#[tokio::test]
|
||||
async fn test_multiple_subscribers() {
|
||||
let bus = EventBus::new();
|
||||
let mut rx1 = bus.subscribe();
|
||||
let mut rx2 = bus.subscribe();
|
||||
|
||||
let event = AiBusEvent::Heartbeat {
|
||||
conversation_id: None,
|
||||
};
|
||||
let delivered = bus.publish(event.clone());
|
||||
assert_eq!(delivered, 2, "publish 应送达 2 个订阅者");
|
||||
|
||||
// 两个 receiver 各自收到(独立消费进度)
|
||||
let r1 = rx1.recv().await.expect("rx1 应收到");
|
||||
let r2 = rx2.recv().await.expect("rx2 应收到");
|
||||
assert!(matches!(r1, AiBusEvent::Heartbeat { .. }));
|
||||
assert!(matches!(r2, AiBusEvent::Heartbeat { .. }));
|
||||
}
|
||||
|
||||
/// 无订阅者时 publish 不 panic,返回 0(兜底)。
|
||||
#[test]
|
||||
fn test_publish_no_subscribers() {
|
||||
let bus = EventBus::new();
|
||||
// 无 subscribe 直接 publish
|
||||
let delivered = bus.publish(AiBusEvent::Error {
|
||||
error: "no one listening".to_string(),
|
||||
conversation_id: None,
|
||||
});
|
||||
assert_eq!(delivered, 0, "无订阅者 publish 应返回 0");
|
||||
}
|
||||
|
||||
/// publish 返回值 = 活跃订阅者数量(subscriber_count 对齐)。
|
||||
#[test]
|
||||
fn test_delivered_count_matches_subscribers() {
|
||||
let bus = EventBus::new();
|
||||
assert_eq!(bus.subscriber_count(), 0);
|
||||
|
||||
let _rx1 = bus.subscribe();
|
||||
assert_eq!(bus.subscriber_count(), 1);
|
||||
|
||||
let _rx2 = bus.subscribe();
|
||||
assert_eq!(bus.subscriber_count(), 2);
|
||||
|
||||
// delivered 应等于 subscriber_count
|
||||
let delivered = bus.publish(AiBusEvent::Round {
|
||||
round: 1,
|
||||
conversation_id: None,
|
||||
});
|
||||
assert_eq!(delivered, bus.subscriber_count());
|
||||
}
|
||||
|
||||
/// EventBus Clone 后共享同一通道(同一总线多持有者)。
|
||||
#[tokio::test]
|
||||
async fn test_clone_shares_channel() {
|
||||
let bus = EventBus::new();
|
||||
let mut rx = bus.subscribe();
|
||||
let bus_clone = bus.clone();
|
||||
|
||||
// 经 clone 的实例 publish,原实例订阅的 receiver 也应收到(共享通道)
|
||||
bus_clone.publish(AiBusEvent::Completed {
|
||||
total_tokens: 100,
|
||||
prompt_tokens: 50,
|
||||
completion_tokens: 50,
|
||||
conversation_id: None,
|
||||
});
|
||||
|
||||
let received = rx.recv().await.expect("clone 共享通道,receiver 应收到");
|
||||
assert!(matches!(received, AiBusEvent::Completed { .. }));
|
||||
}
|
||||
|
||||
/// 全事件类型枚举可构造 + 可序列化(serde 备跨端透传)。
|
||||
#[test]
|
||||
fn test_event_variants_serde() {
|
||||
let events = vec![
|
||||
AiBusEvent::Stream { delta: "x".into(), conversation_id: None },
|
||||
AiBusEvent::ToolStarted { id: "t1".into(), name: "write_file".into(), args: serde_json::json!({}), conversation_id: None },
|
||||
AiBusEvent::ToolCompleted { id: "t1".into(), result: serde_json::json!({"ok": true}), conversation_id: None },
|
||||
AiBusEvent::Round { round: 2, conversation_id: None },
|
||||
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 {
|
||||
let json = serde_json::to_string(event).expect("事件应可序列化");
|
||||
assert!(json.contains("\"type\""), "序列化后应含 type tag: {}", json);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user