修复: B-14 stop 即时打断 Notify 跨文件接入(去 30s 心跳延迟)

AiSession 加 notify:Arc<Notify> 字段 + stop_notify() 取引用;stream_llm 加 notify 参数 + select! notify.notified() 分支(唤醒后 load stop_flag 判退出,防误唤醒继续跑——Notify 仅承载即时唤醒,停止真值仍由 stop_flag 决定);agentic.rs session.notify.clone() 透传 &notify;commands.rs ai_chat_stop 流式分支 stop_flag.store 后立即 notify_one() 唤醒阻塞的 stream.next()。批1 脚手架(本地无用实例)已 revert,本批完整接入。批3 wkitz0twz,cargo workspace 0 err
This commit is contained in:
2026-06-15 05:34:31 +08:00
parent af8f74e29e
commit 4665e910ee
4 changed files with 42 additions and 2 deletions

View File

@@ -140,7 +140,11 @@ pub(crate) async fn run_agentic_loop(
); );
let tool_defs = tools_arc.tool_definitions(); let tool_defs = tools_arc.tool_definitions();
// 停止信号副本stream_llm 与每轮迭代共享读取,避免重复加锁 // 停止信号副本stream_llm 与每轮迭代共享读取,避免重复加锁
let stop_flag = session_arc.lock().await.stop_flag.clone(); // notify 同取一份 Arc 引用B-260615-14stream_llm select! 监听 notified() 即时唤醒
let (stop_flag, notify) = {
let session = session_arc.lock().await;
(session.stop_flag.clone(), session.notify.clone())
};
// token 累加器:loop 生命周期内各轮叠加,退出时传 save_conversation(累加模式落库) // token 累加器:loop 生命周期内各轮叠加,退出时传 save_conversation(累加模式落库)
let mut tokens = TokenAccumulator::default(); let mut tokens = TokenAccumulator::default();
@@ -221,7 +225,7 @@ pub(crate) async fn run_agentic_loop(
let _global_permit = llm_concurrency.acquire_global().await; let _global_permit = llm_concurrency.acquire_global().await;
let _per_conv_permit = llm_concurrency.acquire_per_conv().await; let _per_conv_permit = llm_concurrency.acquire_per_conv().await;
// 流式接收(内部处理 idle timeout / 断连检测 / 停止信号) // 流式接收(内部处理 idle timeout / 断连检测 / 停止信号)
let (full_text, tool_calls_acc, round_usage) = match stream_llm(&*provider, request, &app_handle, &stop_flag, &conv_id).await { let (full_text, tool_calls_acc, round_usage) = match stream_llm(&*provider, request, &app_handle, &stop_flag, &notify, &conv_id).await {
Some(result) => result, Some(result) => result,
None => { None => {
// 错误已在 stream_llm 中 emit直接结束 // 错误已在 stream_llm 中 emit直接结束

View File

@@ -259,6 +259,10 @@ pub async fn ai_chat_stop(state: State<'_, AppState>, app: AppHandle) -> Result<
} }
// 流式生成中:置位让 loop 自行收尾 // 流式生成中:置位让 loop 自行收尾
session.stop_flag.store(true, Ordering::SeqCst); session.stop_flag.store(true, Ordering::SeqCst);
// B-260615-14:置位后立即 notify_one 唤醒阻塞在 stream.next() 的 select!
// 不再等 30s 心跳 tick 或 120s idle timeout 才轮到 stop_flag 检查。
// Notify 仅承载「即时唤醒」,停止真值仍由 stop_flag 决定stream_llm 唤醒后再判 flag
session.stop_notify().notify_one();
let conv_id = session.active_conversation_id.clone(); let conv_id = session.active_conversation_id.clone();
drop(session); drop(session);

View File

@@ -139,6 +139,17 @@ pub struct AiSession {
pub agent_language: Option<String>, pub agent_language: Option<String>,
/// 停止信号ai_chat_stop 置位agentic loop / stream_llm 检测后尽快退出 /// 停止信号ai_chat_stop 置位agentic loop / stream_llm 检测后尽快退出
pub stop_flag: Arc<AtomicBool>, pub stop_flag: Arc<AtomicBool>,
/// 即时停止唤醒B-260615-14stream_llm 阻塞在 `stream.next()` 等 chunk 时,
/// `notify_one()` 立即唤醒跳出 select!,不再等 30s 心跳 tick 轮询。
///
/// 语义边界Notify 仅承载「即时唤醒」这一职责——停止的真值仍由 `stop_flag`
/// AtomicBool决定。`ai_chat_stop` 置 `stop_flag=true` 后调 `notify_one()`
/// 被唤醒的 select! 分支再判 `stop_flag` 决定退出或继续,避免 Notify 的通知
/// 与「真的该停」耦合(防误唤醒继续跑)。
///
/// `Arc<Notify>` 可 CloneNotify 内部用 Atomic 计数Clone 等价于 Arc 引用 +1
/// 不影响 AiSession 其他字段语义。
pub notify: Arc<tokio::sync::Notify>,
} }
impl AiSession { impl AiSession {
@@ -152,9 +163,19 @@ impl AiSession {
generating: false, generating: false,
agent_language: None, agent_language: None,
stop_flag: Arc::new(AtomicBool::new(false)), stop_flag: Arc::new(AtomicBool::new(false)),
notify: Arc::new(tokio::sync::Notify::new()),
} }
} }
/// 取即时停止唤醒器引用B-260615-14
///
/// 供 `stream_llm`(接 `&Notify` 进 select! 监听 `notified()`)与 `ai_chat_stop`
/// (置 `stop_flag` 后调 `notify_one()`)等外部持有方获取同一 Notify 实例。
/// 返回 `&Notify` 而非 clone避免不必要的 Arc 引用计数调整。
pub fn stop_notify(&self) -> &tokio::sync::Notify {
&self.notify
}
/// 推导会话读侧状态视图(只读,不写任何字段)。 /// 推导会话读侧状态视图(只读,不写任何字段)。
/// ///
/// 优先级:`pending_approvals` 非空 → [`SessionState::AwaitingApproval`] /// 优先级:`pending_approvals` 非空 → [`SessionState::AwaitingApproval`]

View File

@@ -109,6 +109,7 @@ pub(crate) async fn stream_llm(
request: CompletionRequest, request: CompletionRequest,
app_handle: &AppHandle, app_handle: &AppHandle,
stop_flag: &AtomicBool, stop_flag: &AtomicBool,
notify: &tokio::sync::Notify,
conv_id: &str, conv_id: &str,
) -> Option<(String, HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage)> { ) -> Option<(String, HashMap<u32, ToolCallDraft>, df_ai::provider::TokenUsage)> {
/// 流式读取空闲超时:超过此时长无任何 chunk 即判定连接已断 /// 流式读取空闲超时:超过此时长无任何 chunk 即判定连接已断
@@ -239,6 +240,16 @@ pub(crate) async fn stream_llm(
conversation_id: Some(conv_id.to_string()), conversation_id: Some(conv_id.to_string()),
}); });
} }
// 3) 即时停止唤醒(B-260615-14):用户点 stop → ai_chat_stop 置 stop_flag 后 notify_one()。
// stream.next() 正阻塞等 chunk 时立即被唤醒,不再等 30s 心跳 tick 或 120s idle timeout。
// 唤醒后判 stop_flag:真值仍由 AtomicBool 决定,Notify 仅承载「即时唤醒」职责
// (防误唤醒继续跑——若非 stop 触发的 notify,flag 仍 false 则照常进下一轮循环)。
_ = notify.notified() => {
if stop_flag.load(std::sync::atomic::Ordering::SeqCst) {
stopped = true;
break;
}
}
} }
} }