From 355244a3d66c62a41b96912d92bc3dca3e1b2a01 Mon Sep 17 00:00:00 2001 From: xiaoxixi Date: Sun, 19 Jul 2026 13:36:42 +0800 Subject: [PATCH] refactor: unify turn progress delivery --- docs/ARCHITECTURE.md | 4 +- src/agent/agent_loop.rs | 16 +----- src/session/session.rs | 105 +++++++++++++++++++--------------------- 3 files changed, 55 insertions(+), 70 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 98956f6..8fe77e7 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -185,6 +185,8 @@ Session ID 格式为： SessionManager 负责组装会话上下文：系统提示、Skills、召回的 Knowledge、压缩后的 Timeline、可选的 active plan 摘要和当前消息历史。`session::turn_input` 在 Session 锁外并行读取 Knowledge、active plan 并压缩历史，然后通过同一个 assembly 路径生成首次请求和 context-overflow 重试输入；重试不得复制系统提示或 runtime context 拼接逻辑。普通闲聊 session 没有 plan 摘要；计划状态由 `WorkManager` 从 SQLite 读取，因此不以自然语言摘要作为权威来源。`AgentLoop` 接收完整输入执行一次模型/工具循环，本身不拥有会话状态。 +当前 Turn 的工具进度只从 `AgentLoop` 的结构化 `TurnEvent` 进入 `TurnController`，不能另建字符串 notification 通道重复投递。后台子 Agent 的 `TaskNotification` 表达跨 Turn 的任务完成，仍由独立的受监督消费者投递。自动标题属于非关键派生工作：Turn 持久化完成后由 `TaskSupervisor` 调度，Session worker 不等待模型生成；同一 Session 同时最多有一个标题任务，提交时仍校验标题保持默认值，避免覆盖用户改名。 + 每个 session 最多有一个 active plan，但 plan 内多个 item 可以分别绑定不同子 Agent 并行执行。主 Agent 通过 `todo` 创建和维护计划，通过 `delegate.plan_item_id` 委托；子 Agent 不获得 `todo` 或 `delegate`。领取 item 使用条件更新防止重复执行，迟到结果只有在 plan 仍 active 且 execution ID 匹配时才允许提交。 ## 6. 持久化 @@ -213,7 +215,7 @@ SessionManager 负责组装会话上下文：系统提示、Skills、召回的 K ## 7. 后台任务与生命周期 -`TaskSupervisor` 是 Gateway 内部后台任务的统一所有者。inbound/control routers、inbound lanes、outbound dispatcher、scheduler、session workers、Turn delivery、outbound lanes、通知消费者和子 Agent 后台任务都应通过它注册。 +`TaskSupervisor` 是 Gateway 内部后台任务的统一所有者。inbound/control routers、inbound lanes、outbound dispatcher、scheduler、session workers、Turn delivery、outbound lanes、后台任务通知消费者、自动标题和子 Agent 后台任务都应通过它注册。 两种注册方式： diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index 4519115..62d0240 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -305,7 +305,6 @@ pub struct AgentLoop { workspace_dir: PathBuf, model_name: String, context_window: usize, - notify_tx: Option>, input_types: Vec, media_registry: MediaHandlerRegistry, } @@ -356,7 +355,6 @@ impl AgentLoop { provider: Arc::from(provider), tools: Arc::new(ToolRegistry::new()), observer: None, - notify_tx: None, context_window: 0, max_iterations, workspace_dir, @@ -382,7 +380,6 @@ impl AgentLoop { provider: Arc::from(provider), tools, observer: None, - notify_tx: None, context_window: 0, max_iterations, workspace_dir, @@ -404,7 +401,6 @@ impl AgentLoop { provider, tools: Arc::new(ToolRegistry::new()), observer: None, - notify_tx: None, context_window: 0, max_iterations, workspace_dir, @@ -427,7 +423,6 @@ impl AgentLoop { provider, tools, observer: None, - notify_tx: None, context_window: 0, max_iterations, workspace_dir, @@ -455,11 +450,6 @@ impl AgentLoop { self } - pub fn with_notify(mut self, tx: tokio::sync::mpsc::UnboundedSender) -> Self { - self.notify_tx = Some(tx); - self - } - /// Preemptive trim: truncate old tool results in-place when history is /// approaching the context window limit. Old results (outside of `keep_recent` /// zone) are replaced with a short placeholder; recent results are truncated @@ -718,7 +708,8 @@ impl AgentLoop { .map_err(|error| AgentError::Other(format!("turn event rejected: {error}")))?; } - // Execute tool calls — log and notify immediately + // Execute tool calls. User-visible progress is emitted through the + // structured TurnEvent stream, not a second notification channel. { let tools_info: Vec = response .tool_calls @@ -726,9 +717,6 @@ impl AgentLoop { .map(|tc| { let args = serde_json::to_string(&tc.arguments).unwrap_or_default(); let s = format!("{}:{}", tc.name, args); - if let Some(ref tx) = self.notify_tx { - let _ = tx.send(format!("调用工具 {}", s)); - } s }) .collect(); diff --git a/src/session/session.rs b/src/session/session.rs index ac70fc1..4d851d8 100644 --- a/src/session/session.rs +++ b/src/session/session.rs @@ -376,6 +376,8 @@ pub struct Session { active_turn_emitter: Option, /// Monotonic counter to detect stale workers worker_generation: u64, + /// Prevents duplicate background title requests while the title is still default. + title_generation_in_flight: bool, /// Monotonic counter for in-memory session mutations. /// /// Slow work such as memory recall, compression, and title generation runs @@ -468,6 +470,7 @@ impl Session { current_cancel: None, active_turn_emitter: None, worker_generation: 0, + title_generation_in_flight: false, state_version: 0, persistence_lock: Arc::new(Mutex::new(())), }) @@ -657,6 +660,7 @@ impl Session { current_cancel: None, active_turn_emitter: None, worker_generation: 0, + title_generation_in_flight: false, state_version: 0, persistence_lock: Arc::new(Mutex::new(())), }) @@ -916,14 +920,6 @@ impl Session { .with_context_window(self.provider_config.token_limit)) } - /// 创建一个附通知通道的 AgentLoop 实例 - pub fn create_agent_with_notify( - &self, - notify_tx: tokio::sync::mpsc::UnboundedSender, - ) -> Result { - Ok(self.create_agent()?.with_notify(notify_tx)) - } - /// 构建系统提示词（包含 AgentLoop 的基础提示词 + skills + memory） pub fn build_system_prompt(&self, skills_prompt: &str) -> String { let base_prompt = build_system_prompt( @@ -2442,17 +2438,13 @@ impl SessionManager { } } -async fn maybe_generate_title_outside_lock(session: Arc>) -> Result<(), AgentError> { +async fn generate_title( + session: Arc>, + provider: Arc, + prompt: String, +) -> Result<(), AgentError> { use crate::providers::{ChatCompletionRequest, ChatCompletionResponse, Message}; - let (provider, prompt) = { - let guard = session.lock().await; - let Some(prompt) = guard.title_prompt_snapshot() else { - return Ok(()); - }; - (guard.provider.clone(), prompt) - }; - let request = ChatCompletionRequest { messages: vec![Message::user(prompt)], temperature: Some(0.3), @@ -2487,6 +2479,39 @@ async fn maybe_generate_title_outside_lock(session: Arc>) -> Resu Ok(()) } +async fn schedule_title_generation( + session: Arc>, + supervisor: crate::task_supervisor::TaskSupervisor, + session_id: &str, +) { + let title_job = { + let mut guard = session.lock().await; + if guard.title_generation_in_flight { + None + } else { + guard.title_prompt_snapshot().map(|prompt| { + guard.title_generation_in_flight = true; + (guard.provider.clone(), prompt) + }) + } + }; + let Some((provider, prompt)) = title_job else { + return; + }; + + let title_session = session.clone(); + let task_session = session.clone(); + let spawned = supervisor.spawn(format!("session-title:{session_id}"), async move { + if let Err(error) = generate_title(title_session, provider, prompt).await { + tracing::warn!(error = %error, "Failed to generate session title"); + } + task_session.lock().await.title_generation_in_flight = false; + }); + if !spawned { + session.lock().await.title_generation_in_flight = false; + } +} + fn spawn_agent_worker( mut task_rx: mpsc::Receiver, session: Arc>, @@ -2510,40 +2535,6 @@ fn spawn_agent_worker( let task_chan = task.channel.clone(); let task_cid = task.chat_id.clone(); let task_metadata = task.forwarded_metadata.clone(); - let notification_session_id = unified_str.clone(); - - let (notify_tx, mut notify_rx) = mpsc::unbounded_channel(); - - // Spawn notification publisher - { - let bus = bus.clone(); - let ch = task_chan.clone(); - let cid = task_cid.clone(); - worker_supervisor.spawn( - format!("session-notifications:{ch}:{cid}"), - async move { - while let Some(notif) = notify_rx.recv().await { - let mut metadata = HashMap::new(); - metadata.insert("_type".to_string(), "notification".to_string()); - metadata.insert( - "_session_id".to_string(), - notification_session_id.clone(), - ); - let outbound = OutboundMessage { - channel: ch.clone(), - chat_id: cid.clone(), - content: notif, - reply_to: None, - media: vec![], - metadata, - delivery: None, - }; - let _ = bus.publish_outbound(outbound).await; - } - }, - ); - } - // Phase 1: capture a stable session snapshot under lock. // Memory recall and compression happen outside this block so // /stop and other commands are not blocked behind slow I/O or @@ -2589,7 +2580,7 @@ fn spawn_agent_worker( let history_raw = guard.get_history().to_vec(); - let agent = match guard.create_agent_with_notify(notify_tx) { + let agent = match guard.create_agent() { Ok(a) => a, Err(e) => { tracing::error!(error = %e, "Failed to create agent"); @@ -2721,6 +2712,7 @@ fn spawn_agent_worker( let cid2 = task_cid.clone(); let unified_str2 = unified_str.clone(); let task_metadata2 = task_metadata.clone(); + let title_supervisor = worker_supervisor.clone(); let turn_lifecycle = &turn_controller; let process_future = async move { let response_session_id = unified_str2.clone(); @@ -2929,9 +2921,12 @@ fn spawn_agent_worker( return; }; - if let Err(e) = maybe_generate_title_outside_lock(session2.clone()).await { - tracing::warn!("failed to generate title: {}", e); - } + schedule_title_generation( + session2.clone(), + title_supervisor, + &response_session_id, + ) + .await; if !live_delivery_started { let outbound = OutboundMessage {