xiaoski/PicoBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: unify turn progress delivery

Browse Source
This commit is contained in:
xiaoxixi 2026-07-19 13:36:42 +08:00
parent 42fe650785
commit 355244a3d6
3 changed files with 55 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
docs/ARCHITECTURE.md
View File

@ -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 后台任务都应通过它注册。
两种注册方式：

16
src/agent/agent_loop.rs
View File

@ -305,7 +305,6 @@ pub struct AgentLoop {
workspace_dir: PathBuf,
model_name: String,
context_window: usize,
notify_tx: Option<tokio::sync::mpsc::UnboundedSender<String>>,
input_types: Vec<String>,
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<String>) -> 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<String> = 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();

105
src/session/session.rs
View File

@ -376,6 +376,8 @@ pub struct Session {
active_turn_emitter: Option<ActiveTurnEmitter>,
/// 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<String>,
) -> Result<AgentLoop, AgentError> {
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<Mutex<Session>>) -> Result<(), AgentError> {
async fn generate_title(
session: Arc<Mutex<Session>>,
provider: Arc<dyn LLMProvider>,
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<Mutex<Session>>) -> Resu
Ok(())
}
async fn schedule_title_generation(
session: Arc<Mutex<Session>>,
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<AgentTask>,
session: Arc<Mutex<Session>>,
@ -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 {