a9429a5657
将单一全局 isLoading 布尔值重构为按 topic_id 跟踪的 processingTopicIds 集合，isLoading 派生自当前选中话题是否在集合中。导航响应不再清空处理 状态，切换话题后切回原话题仍能正确禁用输入。重连时通过新增的 /api/executions 端点对账后端权威执行状态，修正断连期间丢失的 execution_completed 信号导致的状态漂移。
85 lines
2.9 KiB
Rust
85 lines
2.9 KiB
Rust
use std::collections::HashMap;
|
||
use std::sync::Arc;
|
||
use tokio::sync::{Mutex, watch};
|
||
|
||
/// 共享的 Agent 取消注册表。
|
||
///
|
||
/// 每个正在执行的 Agent 在启动前按 topic_id 注册一个 watch::Sender，
|
||
/// 外部（如 /stop 命令）通过 cancel_by_topic() 发送取消信号。
|
||
/// Agent 循环内部通过 watch::Receiver::has_changed() 检测取消。
|
||
///
|
||
/// key 使用 topic_id（UUID），全局唯一，精确到话题级别。
|
||
#[derive(Clone)]
|
||
pub struct CancelManager {
|
||
tokens: Arc<Mutex<HashMap<String, watch::Sender<()>>>>,
|
||
}
|
||
|
||
impl CancelManager {
|
||
pub fn new() -> Self {
|
||
Self {
|
||
tokens: Arc::new(Mutex::new(HashMap::new())),
|
||
}
|
||
}
|
||
|
||
/// 按 topic_id 注册一个取消通道，返回 receiver 供 Agent 持有。
|
||
///
|
||
/// 如果同 topic_id 已有注册，旧 sender 被覆盖并 drop，
|
||
/// 旧 receiver 将收到通道关闭信号。
|
||
pub async fn register(&self, topic_id: &str) -> watch::Receiver<()> {
|
||
let (tx, rx) = watch::channel(());
|
||
self.tokens.lock().await.insert(topic_id.to_string(), tx);
|
||
rx
|
||
}
|
||
|
||
/// 按 topic_id 发送取消信号并移除注册条目。
|
||
///
|
||
/// 返回 `true` 表示找到了对应的任务并发送了取消信号，
|
||
/// 返回 `false` 表示没有找到对应的任务（可能已经完成或从未注册）。
|
||
pub async fn cancel_by_topic(&self, topic_id: &str) -> bool {
|
||
if let Some(tx) = self.tokens.lock().await.remove(topic_id) {
|
||
// send 可能失败（receiver 已被 drop），这不影响语义
|
||
let _ = tx.send(());
|
||
true
|
||
} else {
|
||
false
|
||
}
|
||
}
|
||
|
||
/// 按 topic_id 正常完成后清理注册条目（幂等）。
|
||
///
|
||
/// 与 cancel_by_topic() 不同，此方法不发送取消信号，仅移除条目。
|
||
/// 如果条目已被 cancel_by_topic() 移除，此调用为 no-op。
|
||
pub async fn remove_by_topic(&self, topic_id: &str) {
|
||
self.tokens.lock().await.remove(topic_id);
|
||
}
|
||
|
||
/// 返回当前正在运行的 Agent 数量。
|
||
pub async fn active_count(&self) -> usize {
|
||
self.tokens.lock().await.len()
|
||
}
|
||
|
||
/// 返回当前正在执行的 Agent 的 topic_id 列表。
|
||
///
|
||
/// 用于前端重连时对账执行状态：前端通过此 API 判断断连期间
|
||
/// 哪些话题的智能体仍在运行、哪些已完成。
|
||
pub async fn list_active_topic_ids(&self) -> Vec<String> {
|
||
self.tokens.lock().await.keys().cloned().collect()
|
||
}
|
||
|
||
/// 取消所有正在运行的 Agent 并清空注册表。
|
||
///
|
||
/// 用于 graceful shutdown / restart 场景。
|
||
pub async fn cancel_all(&self) {
|
||
let mut tokens = self.tokens.lock().await;
|
||
for (_, tx) in tokens.drain() {
|
||
let _ = tx.send(());
|
||
}
|
||
}
|
||
}
|
||
|
||
impl Default for CancelManager {
|
||
fn default() -> Self {
|
||
Self::new()
|
||
}
|
||
}
|