From 612de4cd7c9e81077e2c6ebb0e0c4cc48f3d8406 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com> Date: Mon, 22 Jun 2026 03:31:58 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E:=20Phase3=E9=98=B6=E6=AE=B52?= =?UTF-8?q?=20emit=E5=8F=8C=E5=86=99=E5=85=A8=E6=8E=A5=E5=85=A5(D2?= =?UTF-8?q?=E5=85=A819=E5=8F=98=E4=BD=93)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit chat.rs 19 + audit/mod.rs 12 + guard.rs 3 + stream_recv.rs 2 + agentic/mod.rs 7(emit-only 补全,原11已落地)。每处 app.emit 前端 + publish_event tunnel 双路独立(emit 不删)。D2 全 19 变体透传 EventBus 供 tunnel subscriber 透传 miniapp。cargo EXIT=0。 --- src-tauri/src/commands/ai/agentic/guard.rs | 23 +++- src-tauri/src/commands/ai/agentic/mod.rs | 49 +++++-- src-tauri/src/commands/ai/audit/mod.rs | 85 ++++++++---- src-tauri/src/commands/ai/commands/chat.rs | 149 ++++++++++++++------- src-tauri/src/commands/ai/stream_recv.rs | 16 ++- 5 files changed, 225 insertions(+), 97 deletions(-) diff --git a/src-tauri/src/commands/ai/agentic/guard.rs b/src-tauri/src/commands/ai/agentic/guard.rs index 9dfd12f..917d27a 100644 --- a/src-tauri/src/commands/ai/agentic/guard.rs +++ b/src-tauri/src/commands/ai/agentic/guard.rs @@ -5,7 +5,7 @@ use std::sync::Arc; -use tauri::{AppHandle, Emitter}; +use tauri::{AppHandle, Emitter, Manager}; use tokio::sync::Mutex; use crate::commands::ai::{AiChatEvent, AiSession}; @@ -83,10 +83,13 @@ impl GeneratingGuard { } }; // 持久化层迁移成功后 emit 前端(同步,失败忽略)。锁已随作用域 drop,可安全 emit。 - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiConvStateChanged { + // L3 emit 双写:提变量避免构造两次,publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 + let ev = AiChatEvent::AiConvStateChanged { conv_state: new_state, conversation_id: Some(cid), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); }); } Self { session, conv_id, done: false, app_handle } @@ -107,10 +110,13 @@ impl GeneratingGuard { Ok(ns) => { conv.conv_state = ns; // L2 批2 1b:持久化层 Idle 收敛成功后 emit 前端(同步,失败忽略)。 - let _ = self.app_handle.emit("ai-chat-event", AiChatEvent::AiConvStateChanged { + // L3 emit 双写:提变量避免构造两次,publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 + let ev = AiChatEvent::AiConvStateChanged { conv_state: ns, conversation_id: Some(self.conv_id.clone()), - }); + }; + let _ = self.app_handle.emit("ai-chat-event", ev.clone()); + let _ = self.app_handle.state::().ai_event_bus.publish_event(ev); } Err(e) => tracing::warn!( conv_id = %self.conv_id, @@ -154,10 +160,13 @@ impl Drop for GeneratingGuard { if let Ok(ns) = conv.conv_state.transition_to(ConvState::Idle) { conv.conv_state = ns; // 1b:持久化 Idle 收敛后 emit 前端(异常退出兜底也需通知前端退出生成态)。 - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiConvStateChanged { + // L3 emit 双写:提变量避免构造两次,publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 + let ev = AiChatEvent::AiConvStateChanged { conv_state: ns, conversation_id: Some(conv_id.clone()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } } }); diff --git a/src-tauri/src/commands/ai/agentic/mod.rs b/src-tauri/src/commands/ai/agentic/mod.rs index dca59e5..69f142d 100644 --- a/src-tauri/src/commands/ai/agentic/mod.rs +++ b/src-tauri/src/commands/ai/agentic/mod.rs @@ -354,11 +354,14 @@ async fn stream_one_provider( candidate.name, delay.as_millis(), total_retry, max_retries + 1, ); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiStreamRetry { + let ev = AiChatEvent::AiStreamRetry { attempt: total_retry as u32, max_attempts: (max_retries + 1) as u32, conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app_handle.state::().ai_event_bus.publish_event(ev); tokio::time::sleep(delay).await; continue; @@ -861,9 +864,12 @@ pub(crate) async fn run_agentic_loop( if should_compress { // emit 压缩开始 + 置位防重入 + 读 active 克隆(不改 status)。 - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompressing { + let ev = AiChatEvent::AiCompressing { conversation_id: Some(conv_id.clone()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app_handle.state::().ai_event_bus.publish_event(ev); let (active_msgs, lang) = { let mut session = session_arc.lock().await; let conv = session.conv(&conv_id); @@ -923,10 +929,13 @@ pub(crate) async fn run_agentic_loop( pre_tokens = pre_compress_tokens, "[ai] 自动压缩成功,摘要已插首位" ); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiCompressed { + let ev = AiChatEvent::AiCompressed { conversation_id: Some(conv_id.clone()), summary, - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } Ok(None) => { // 保护区外无 active 可压缩(已全 compressed/archived)→ noop,仅复位 is_compressing。 @@ -973,11 +982,14 @@ pub(crate) async fn run_agentic_loop( ); } } - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiError { + let ev = AiChatEvent::AiError { error: format!("自动上下文压缩失败,已降级为普通裁剪: {}", e), error_type: Some(ErrorType::Unknown), conversation_id: Some(conv_id.clone()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } } } @@ -1488,9 +1500,12 @@ pub(crate) async fn run_agentic_loop( save_conversation(&session_arc, &db, &conv_id, Some(&usage), Some(&resolved_model), true).await; // 暂停态保持 generating=true(防其他 send 抢占,仿审批),disarm guard 跳过 Drop 兜底复位 guard.disarm(); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiMaxRoundsReached { + let ev = AiChatEvent::AiMaxRoundsReached { conversation_id: Some(conv_id.clone()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app_handle.state::().ai_event_bus.publish_event(ev); return; // generating 保持 true,等 ai_continue_loop / ai_stop_loop } @@ -1629,13 +1644,16 @@ pub(crate) async fn try_continue_agent_loop( Some(cid) => cid, None => conv_id.to_string(), }; - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + let ev = AiChatEvent::AiCompleted { total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, incomplete: None, conversation_id: Some(emit_conv_id), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app.state::().ai_event_bus.publish_event(ev); } return; } @@ -1713,13 +1731,16 @@ pub(crate) async fn try_continue_agent_loop( }; if !still_generating { tracing::info!(conv_id = %conv_id, "[ai] try_continue 终止:spawn 前重检 generating 已被 stop 复位,补发 AiCompleted"); - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + let ev = AiChatEvent::AiCompleted { total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, incomplete: None, conversation_id: Some(conv_id_owned.clone()), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2)透传 miniapp + let _ = app.state::().ai_event_bus.publish_event(ev); return; } diff --git a/src-tauri/src/commands/ai/audit/mod.rs b/src-tauri/src/commands/ai/audit/mod.rs index 91dac82..04060c2 100644 --- a/src-tauri/src/commands/ai/audit/mod.rs +++ b/src-tauri/src/commands/ai/audit/mod.rs @@ -282,12 +282,16 @@ pub(crate) async fn process_tool_calls( let drafts: Vec<(u32, ToolCallDraft, serde_json::Value)> = tc_list.into_iter() .map(|(idx, draft)| { let args = serde_json::from_str(&draft.args).unwrap_or(serde_json::Value::Object(Default::default())); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallStarted { + // L3 emit 双写:AiToolCallStarted publish 到事件总线(tunnel subscriber 透传 miniapp)。 + // AiTextDelta/AiToolCall* 高频事件不双写原则不适用此处(工具调用生命周期事件属关键状态变更,需透传)。 + let ev = AiChatEvent::AiToolCallStarted { id: draft.id.clone(), name: draft.name.clone(), args: args.clone(), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); (idx, draft, args) }) .collect(); @@ -325,11 +329,14 @@ pub(crate) async fn process_tool_calls( for (draft, reason) in path_denied { let err_msg = format!("路径授权拒绝: {}", reason); session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &err_msg)); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:路径黑名单拒绝 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(err_msg.clone()), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); // 审计:路径黑名单拒绝(decided_by=auto_blacklist),留痕可追溯 let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High); audit_tool_call( @@ -362,11 +369,14 @@ pub(crate) async fn process_tool_calls( draft.id ); session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &skip_msg)); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:重试跳过 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(skip_msg.clone()), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High); audit_tool_call( &audit_repo, conv_id, &draft.id, &draft.name, &draft.args, @@ -393,13 +403,16 @@ pub(crate) async fn process_tool_calls( // 检测豁免保留 + 出口断言自愈补头(防占位头被裁后 orphan result 触发 provider 400)。 session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &pending_placeholder_for(&draft.id))); tracing::debug!(target: "ai_dirauth", conv = %conv_id, tool = %draft.name, tc_id = %draft.id, "emit AiDirAuthRequired(路径授权挂起,等 ai_authorize_dir 恢复)"); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiDirAuthRequired { + // L3 emit 双写:路径授权挂起 AiDirAuthRequired 双路发布(tunnel 透传 miniapp 弹窗)。 + let ev = AiChatEvent::AiDirAuthRequired { id: draft.id.clone(), tool: draft.name.clone(), path: path_str, dir: dir_str, conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); let risk_level = tools_arc.get(&draft.name).map(|t| t.risk_level).unwrap_or(RiskLevel::High); audit_tool_call( &audit_repo, conv_id, &draft.id, &draft.name, &draft.args, @@ -450,12 +463,15 @@ pub(crate) async fn process_tool_calls( ); // emit 轻量 toast 事件(前端 AiChat.vue 显示"🔓 自动放行: tool(dir)") // toast 即时反馈,锁内 emit 不阻塞 - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolAutoApproved { + // L3 emit 双写:会话信任自动放行 toast 双路发布(tunnel 透传 miniapp 即时反馈)。 + let ev = AiChatEvent::AiToolAutoApproved { id: draft.id.clone(), tool: draft.name.clone(), dir: dir_label.clone(), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); // 收集后循环外 spawn 执行,对齐 Low risk 不持锁(原注释 L593 声称"锁外"但代码持锁, // run_command 慢命令会阻塞同会话所有触 state.ai_session 的 IPC,CR-51 修此) trust_hits.push((draft, args, dir_label, risk_level)); @@ -472,11 +488,14 @@ pub(crate) async fn process_tool_calls( ); // F-260616-09 B 批2:写 per_conv.messages(conv_id 来源:本函数入参)。 session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &cached)); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:高危去重命中复用缓存 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(cached.clone()), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); // 审计:去重命中记一条(status 透传缓存来源 completed/rejected/failed,SW-260618-16;decided_by=auto_dedup),不进 pending audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, &status, risk_level, Some(cached), Some("auto_dedup"), current_message_id).await; continue; @@ -504,11 +523,14 @@ pub(crate) async fn process_tool_calls( draft.id ); session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &skip_msg)); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:重试 guard 跳过 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(skip_msg.clone()), conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, "skipped_retry", risk_level, Some(skip_msg), Some("auto_retry_guard"), current_message_id).await; continue; } @@ -525,14 +547,17 @@ pub(crate) async fn process_tool_calls( // 阶段2:占位带 __PENDING__:tc_id 标记,供 sanitize 豁免保留 + 出口断言自愈(防 400 orphan) session.conv(conv_id).messages.push(ChatMessage::tool_result(&draft.id, &pending_placeholder_for(&draft.id))); let reason = build_approval_reason(&draft.name, &args, risk_level, db).await; - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiApprovalRequired { + // L3 emit 双写:Med/High 风险审批挂起 AiApprovalRequired 双路发布(tunnel 透传 miniapp 弹审批窗)。 + let ev = AiChatEvent::AiApprovalRequired { id: draft.id.clone(), name: draft.name.clone(), args: args.clone(), reason, diff: approval_diff, conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); audit_tool_call(&audit_repo, conv_id, &draft.id, &draft.name, &draft.args, "pending", risk_level, None, None, current_message_id).await; } } @@ -555,11 +580,14 @@ pub(crate) async fn process_tool_calls( match exec_result { Ok(val) => { let content = val.to_string(); - let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:trust 放行工具执行成功 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(content.clone()), conversation_id: Some(conv_clone), - }); + }; + let _ = app_clone.emit("ai-chat-event", ev.clone()); + let _ = app_clone.state::().ai_event_bus.publish_event(ev); // AR-11：数据变更类工具执行成功后 emit df-data-changed 联动刷新 // (write_file/run_command 不在 data_change_for_tool 映射内,emit_data_changed 内部 None 即 noop) emit_data_changed(&app_clone, &draft.name); @@ -567,11 +595,14 @@ pub(crate) async fn process_tool_calls( } Err(e) => { let content = e.to_string(); - let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:trust 放行工具执行失败 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(content.clone()), conversation_id: Some(conv_clone), - }); + }; + let _ = app_clone.emit("ai-chat-event", ev.clone()); + let _ = app_clone.state::().ai_event_bus.publish_event(ev); (draft, risk_level, Err(content)) } } @@ -605,11 +636,14 @@ pub(crate) async fn process_tool_calls( let result = tools.execute(&draft.name, args).await; match result { Ok(val) => { - let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:Low 风险工具执行成功 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: val.clone(), conversation_id: Some(conv_clone), - }); + }; + let _ = app_clone.emit("ai-chat-event", ev.clone()); + let _ = app_clone.state::().ai_event_bus.publish_event(ev); // AR-11（方案A）：数据变更工具执行成功后 emit df-data-changed, // 前端 store listen 刷新列表（仅命中映射的工具 emit，见 emit_data_changed）。 emit_data_changed(&app_clone, &draft.name); @@ -622,11 +656,14 @@ pub(crate) async fn process_tool_calls( // 现改为正常 emit AiToolCallCompleted（result=错误信息），错误包进 tool_result // 让 LLM 看到工具失败自行决定下一步，loop 正常续，前端状态一致。 let err_msg = format!("工具 {} 执行失败: {}", draft.name, e); - let _ = app_clone.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + // L3 emit 双写:Low 风险工具执行失败 emit Completed 双路发布。 + let ev = AiChatEvent::AiToolCallCompleted { id: draft.id.clone(), result: serde_json::Value::String(err_msg.clone()), conversation_id: Some(conv_clone), - }); + }; + let _ = app_clone.emit("ai-chat-event", ev.clone()); + let _ = app_clone.state::().ai_event_bus.publish_event(ev); (draft, serde_json::Value::Null, Err(err_msg)) } } diff --git a/src-tauri/src/commands/ai/commands/chat.rs b/src-tauri/src/commands/ai/commands/chat.rs index 38f7bb0..85a8150 100644 --- a/src-tauri/src/commands/ai/commands/chat.rs +++ b/src-tauri/src/commands/ai/commands/chat.rs @@ -13,7 +13,7 @@ use std::sync::atomic::Ordering; use serde::Serialize; -use tauri::{AppHandle, Emitter, State}; +use tauri::{AppHandle, Emitter, Manager, State}; use df_ai::provider::{ChatMessage, ContentPart}; use df_storage::models::AiProviderRecord; @@ -492,11 +492,15 @@ pub async fn ai_approve( session.conv(cid).messages.replace_tool_result_content(&tool_call_id, "用户拒绝了此操作"); } let conv_id = approval.conversation_id.clone(); - let _ = app.emit("ai-chat-event", AiChatEvent::AiApprovalResult { + let ev = AiChatEvent::AiApprovalResult { id: tool_call_id.clone(), approved: false, conversation_id: conv_id.clone(), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp;publish 内 EVENT_BUS_ENABLED 门控, + // 无消费者空转。app.emit 前端 + publish_event tunnel 两路独立,emit 保留不删。 + let _ = app.state::().ai_event_bus.publish_event(ev); drop(session); // 拒绝结果立即落库(含 recovered 积压审批)——switch 时已 restore_from_messages 载完整历史, // messages 非空,save 不会污染老对话;原 if !recovered 守卫前提不成立已移除。 @@ -603,16 +607,22 @@ pub async fn ai_approve( session.conv(cid).messages.replace_tool_result_content(&id, &result_val.to_string()); } - let _ = app.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + let ev_completed = AiChatEvent::AiToolCallCompleted { id: id.clone(), result: result_val.clone(), conversation_id: conv_id.clone(), - }); - let _ = app.emit("ai-chat-event", AiChatEvent::AiApprovalResult { + }; + let _ = app.emit("ai-chat-event", ev_completed.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_completed); + let ev_approval = AiChatEvent::AiApprovalResult { id, approved: true, conversation_id: conv_id.clone(), - }); + }; + let _ = app.emit("ai-chat-event", ev_approval.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_approval); drop(session); // F-260619-04 P2(方案 B): create_idea source 消息级溯源补全。 @@ -804,16 +814,22 @@ pub async fn ai_authorize_dir( session.conv(cid).messages.replace_tool_result_content(&tool_call_id, &err_msg); } } - let _ = app.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + let ev_completed = AiChatEvent::AiToolCallCompleted { id: tool_call_id.clone(), result: serde_json::Value::String(err_msg.clone()), conversation_id: conv_id.clone(), - }); - let _ = app.emit("ai-chat-event", AiChatEvent::AiApprovalResult { + }; + let _ = app.emit("ai-chat-event", ev_completed.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_completed); + let ev_approval = AiChatEvent::AiApprovalResult { id: tool_call_id.clone(), approved: false, conversation_id: conv_id.clone(), - }); + }; + let _ = app.emit("ai-chat-event", ev_approval.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_approval); audit_finalize(&state, &tool_call_id, "rejected", Some(err_msg)).await; if let Some(ref cid) = conv_id { save_conversation(&state.ai_session, &state.db, cid, None, None, true).await; @@ -899,16 +915,22 @@ pub async fn ai_authorize_dir( session.conv(cid).messages.replace_tool_result_content(&tool_call_id, &result_val.to_string()); } } - let _ = app.emit("ai-chat-event", AiChatEvent::AiToolCallCompleted { + let ev_completed = AiChatEvent::AiToolCallCompleted { id: tool_call_id.clone(), result: result_val.clone(), conversation_id: conv_id.clone(), - }); - let _ = app.emit("ai-chat-event", AiChatEvent::AiApprovalResult { + }; + let _ = app.emit("ai-chat-event", ev_completed.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_completed); + let ev_approval = AiChatEvent::AiApprovalResult { id: tool_call_id.clone(), approved: true, conversation_id: conv_id.clone(), - }); + }; + let _ = app.emit("ai-chat-event", ev_approval.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev_approval); if let Some(ref cid) = conv_id { save_conversation(&state.ai_session, &state.db, cid, None, None, true).await; } @@ -1023,9 +1045,12 @@ pub async fn ai_chat_clear_context( // 空会话/全在保护区(消息 ≤ N 条) → 无可分段消息,emit 后 noop if protect_start == 0 { drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiContextCleared { + let ev = AiChatEvent::AiContextCleared { conversation_id: Some(conversation_id.clone()), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Ok(()); } // 对保护区外 active 消息标 archived_segment(messages_mut 直接改 status)。 @@ -1039,9 +1064,12 @@ pub async fn ai_chat_clear_context( }; // 落库持久化新 status(照 save_conversation 模式,DB 持久化新 status) save_conversation(&state.ai_session, &state.db, &conv_id, None, None, true).await; - let _ = app.emit("ai-chat-event", AiChatEvent::AiContextCleared { + let ev = AiChatEvent::AiContextCleared { conversation_id: Some(conv_id), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); Ok(()) } @@ -1076,9 +1104,12 @@ pub async fn ai_chat_compress_context( return Err("压缩正在进行中".to_string()); } // emit 开始 + 置位防重入(成对释放,见下方所有出口) - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompressing { + let ev = AiChatEvent::AiCompressing { conversation_id: Some(conversation_id.clone()), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); conv.messages.set_compressing(true); let protect_start = conv.messages.len().saturating_sub(PROTECT_COUNT); @@ -1087,10 +1118,13 @@ pub async fn ai_chat_compress_context( conv.messages.set_compressing(false); let cid = session.active_conversation_id.clone(); drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompressed { + let ev = AiChatEvent::AiCompressed { conversation_id: cid, summary: String::new(), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Ok(()); } // 读 active 克隆(不改 status):compress_old_messages 留到 LLM 成功后才调 @@ -1110,10 +1144,13 @@ pub async fn ai_chat_compress_context( let mut session = state.ai_session.lock().await; session.conv(&conv_id).messages.set_compressing(false); drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompressed { + let ev = AiChatEvent::AiCompressed { conversation_id: Some(conv_id), summary: String::new(), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Ok(()); } @@ -1125,11 +1162,14 @@ pub async fn ai_chat_compress_context( let mut session = state.ai_session.lock().await; session.conv(&conv_id).messages.set_compressing(false); drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiError { + let ev = AiChatEvent::AiError { error: format!("压缩失败: {}", e), error_type: Some(super::super::ErrorType::Auth), conversation_id: Some(conv_id), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Err(format!("压缩失败: {}", e)); } }; @@ -1148,11 +1188,14 @@ pub async fn ai_chat_compress_context( let mut session = state.ai_session.lock().await; session.conv(&conv_id).messages.set_compressing(false); drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiError { + let ev = AiChatEvent::AiError { error: format!("压缩失败: {}", e), error_type: Some(super::super::ErrorType::Unknown), conversation_id: Some(conv_id), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Err(format!("压缩失败: {}", e)); } }; @@ -1166,10 +1209,13 @@ pub async fn ai_chat_compress_context( conv.messages.set_compressing(false); } save_conversation(&state.ai_session, &state.db, &conv_id, None, None, false).await; - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompressed { + let ev = AiChatEvent::AiCompressed { conversation_id: Some(conv_id), summary, - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); Ok(()) } @@ -1376,13 +1422,16 @@ pub async fn ai_chat_force_send( // 通知前端目标 conv 旧生成已结束(若先前在生成;emit 在锁外,避免持锁调 runtime emit) if let Some(ref cid) = old_conv_id { - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + let ev = AiChatEvent::AiCompleted { total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, incomplete: None, conversation_id: Some(cid.clone()), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); } // system prompt 构建(照 ai_chat_send):augmentation 注入 + 知识注入顺序一致。 @@ -1476,10 +1525,13 @@ pub async fn ai_chat_stop( conv.generating = false; conv.stop_flag.store(true, Ordering::SeqCst); // 双保险：防 try_continue 误判重启 drop(session); - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + let ev = AiChatEvent::AiCompleted { total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, incomplete: None, conversation_id: Some(target), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); return Ok(()); } // 流式生成中：置位让目标 conv 的 loop 自行收尾 @@ -1513,16 +1565,16 @@ pub async fn ai_chat_stop( if still_gen { session.conv(&target_owned).generating = false; drop(session); // 释放锁后再 emit，避免持锁调 runtime emit - let _ = app_handle.emit( - "ai-chat-event", - AiChatEvent::AiCompleted { - total_tokens: 0, - prompt_tokens: 0, - completion_tokens: 0, - incomplete: None, - conversation_id: conv_id, - }, - ); + let ev = AiChatEvent::AiCompleted { + total_tokens: 0, + prompt_tokens: 0, + completion_tokens: 0, + incomplete: None, + conversation_id: conv_id, + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。闭包内 app_handle 仍可访问 AppState。 + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } }); @@ -1595,12 +1647,15 @@ pub async fn ai_stop_loop( conv.generating = false; } // 暂停态进入前已 save_conversation，此处零 token 上报仅作收敛信号（与 try_continue 补发 AiCompleted 一致） - let _ = app.emit("ai-chat-event", AiChatEvent::AiCompleted { + let ev = AiChatEvent::AiCompleted { total_tokens: 0, prompt_tokens: 0, completion_tokens: 0, incomplete: None, conversation_id: Some(conversation_id), - }); + }; + let _ = app.emit("ai-chat-event", ev.clone()); + // L3 emit 双写:tunnel subscriber(阶段2 后续)透传 miniapp。 + let _ = app.state::().ai_event_bus.publish_event(ev); Ok("ok".to_string()) } diff --git a/src-tauri/src/commands/ai/stream_recv.rs b/src-tauri/src/commands/ai/stream_recv.rs index db64923..58b7ed8 100644 --- a/src-tauri/src/commands/ai/stream_recv.rs +++ b/src-tauri/src/commands/ai/stream_recv.rs @@ -4,7 +4,7 @@ use std::collections::HashMap; use std::sync::atomic::AtomicBool; use std::time::Duration; -use tauri::{AppHandle, Emitter}; +use tauri::{AppHandle, Emitter, Manager}; use futures::StreamExt; use tracing::warn; @@ -226,10 +226,13 @@ pub(crate) async fn stream_llm( Ok(chunk) => { if !chunk.delta.is_empty() { full_text.push_str(&chunk.delta); - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiTextDelta { + // L3 emit 双写:提变量避免构造两次,publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 + let ev = AiChatEvent::AiTextDelta { delta: chunk.delta, conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } if let Some(tc_deltas) = &chunk.tool_calls { for tc_delta in tc_deltas { @@ -326,9 +329,12 @@ pub(crate) async fn stream_llm( // 2) 心跳:静默期 30s 发 AiHeartbeat,前端 watchdog reset(B-260615-02) _ = heartbeat.tick() => { // 心跳只在「仍在等下一 chunk」时有意义——若 stop_flag 已置,顶部快检会 break,无需发心跳 - let _ = app_handle.emit("ai-chat-event", AiChatEvent::AiHeartbeat { + // L3 emit 双写:提变量避免构造两次,publish 到事件总线(EVENT_BUS_ENABLED 门控在 publish 内)。 + let ev = AiChatEvent::AiHeartbeat { conversation_id: Some(conv_id.to_string()), - }); + }; + let _ = app_handle.emit("ai-chat-event", ev.clone()); + let _ = app_handle.state::().ai_event_bus.publish_event(ev); } // 3) 即时停止唤醒(B-260615-14):用户点 stop → ai_chat_stop 置 stop_flag 后 notify_one()。 // stream.next() 正阻塞等 chunk 时立即被唤醒,不再等 30s 心跳 tick 或 120s idle timeout。