From 452c5ad0effe3e53e6589edbe8bfc8fd987fedbe Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Tue, 18 Aug 2026 09:33:07 +0800 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=E4=BF=AE=E5=A4=8D=E9=95=BF?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D=E5=90=8E=E6=96=B0=E6=B6=88=E6=81=AF=E6=97=A0?= =?UTF-8?q?=E5=93=8D=E5=BA=94=E2=80=94=E2=80=94=E5=83=B5=E5=B0=B8=E5=AD=90?= =?UTF-8?q?=E4=BB=A3=E7=90=86=E6=B8=85=E7=90=86=E3=80=81panic=20=E5=85=9C?= =?UTF-8?q?=E5=BA=95=E4=B8=8E=E8=BF=90=E8=A1=8C=E9=A2=84=E7=AE=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 发送 ExecutionCompleted 前惰性清理僵尸 running 子代理（DB running 但执行任务已消失），解除 pending 阻塞 - processor panic 路径补发 error 通知 + ExecutionCompleted，防止前端永久 loading - agent 单轮增加墙钟预算 max_run_secs，超时优雅退出释放 topic 串行锁 - topic 串行锁等待增加 info/warn 日志，长等待可观测 - sanitize 清理结果回写 DB（delete_messages_by_ids），消除每次加载的重复修复 - storage 新增 keyset 分页与按 ID 批量删除的单元测试 --- src/agent/agent_loop.rs | 46 ++++++++++++ src/agent/runtime_config.rs | 9 +++ src/bus/message.rs | 13 +++- src/gateway/execution.rs | 41 ++++++++++- src/gateway/processor.rs | 129 ++++++++++++++++++++++++++++++++- src/gateway/session_history.rs | 32 +++++++- src/storage/mod.rs | 29 ++++++++ src/storage/ports.rs | 15 ++++ src/storage/tests.rs | 102 ++++++++++++++++++++++++++ src/tools/task/runtime.rs | 75 +++++++++++++++++++ 10 files changed, 485 insertions(+), 6 deletions(-) diff --git a/src/agent/agent_loop.rs b/src/agent/agent_loop.rs index ad743a4..72feac5 100644 --- a/src/agent/agent_loop.rs +++ b/src/agent/agent_loop.rs @@ -1180,6 +1180,7 @@ impl AgentLoop { .map(estimate_tokens_from_serialized_json) .unwrap_or_default(); + let run_started = Instant::now(); for iteration in 0..self.max_iterations { #[cfg(debug_assertions)] tracing::debug!(iteration, "Agent iteration started"); @@ -1195,6 +1196,28 @@ impl AgentLoop { return Ok(cancel); } + // 墙钟预算：max_tool_iterations 很大（默认 1000）时，单轮合法长跑 + // 可能占住 topic serial lock 数小时，期间所有新用户消息无限排队。 + // 超预算优雅退出并说明，让用户可以继续交互。 + if self.runtime_config.max_run_secs > 0 { + let elapsed = run_started.elapsed(); + let budget = Duration::from_secs(self.runtime_config.max_run_secs); + if elapsed >= budget { + tracing::warn!( + iteration, + elapsed_secs = elapsed.as_secs(), + max_run_secs = self.runtime_config.max_run_secs, + emitted_count = emitted_messages.len(), + "Agent run wall-clock budget exhausted, exiting gracefully" + ); + let exhausted = + Self::build_budget_exhausted_result(iteration, elapsed, emitted_messages); + self.emit_live_tool_call_message(exhausted.final_response.clone()) + .await; + return Ok(exhausted); + } + } + // Defense-in-depth: sanitize incomplete tool call sequences // before EVERY LLM request, not just once at process() entry. // This catches edge cases where compression, persistence races, @@ -2031,6 +2054,29 @@ impl AgentLoop { } } + /// 墙钟预算耗尽的优雅退出结果。 + fn build_budget_exhausted_result( + iteration: usize, + elapsed: std::time::Duration, + mut emitted_messages: Vec, + ) -> AgentProcessResult { + let emitted_count = emitted_messages.len(); + let message = format!( + "\n\n[本轮运行已达单次执行时间上限（{} 分钟，实际迭代 {} 次，生成 {} 条消息），已自动停止以释放会话。任务如有未完成部分，请发送新消息继续。]", + elapsed.as_secs() / 60, + iteration, + emitted_count + ); + let assistant_message = ChatMessage::assistant(message); + emitted_messages.push(assistant_message.clone()); + AgentProcessResult { + final_response: assistant_message, + emitted_messages, + compaction_performed: false, + engineering_compaction_applied: false, + } + } + async fn emit_live_tool_call_message(&self, message: ChatMessage) { if let Some(handler) = &self.emitted_message_handler { handler.handle(message).await; diff --git a/src/agent/runtime_config.rs b/src/agent/runtime_config.rs index 63ccf01..7200056 100644 --- a/src/agent/runtime_config.rs +++ b/src/agent/runtime_config.rs @@ -15,8 +15,16 @@ pub struct AgentRuntimeConfig { /// LLM 请求瞬态失败的最大重试次数（仅对 timeout/502/503/504/429 等可恢复错误重试）。 /// 0 表示不重试。归属 agent 行为层，不进 ProviderRuntimeConfig（保持 provider 构造包纯净）。 pub max_retries: u32, + /// 单次 process() 的墙钟预算（秒）。超时后 agent 优雅退出并给出说明。 + /// 防止 max_tool_iterations 很大（默认 1000）时，单轮合法长跑占住 + /// topic serial lock 数小时，期间所有新用户消息无限排队。 + /// 0 表示不限制。默认见 DEFAULT_MAX_RUN_SECS。 + pub max_run_secs: u64, } +/// 单次 agent run 默认墙钟预算。 +pub const DEFAULT_MAX_RUN_SECS: u64 = 3600; + impl From for AgentRuntimeConfig { fn from(config: LLMProviderConfig) -> Self { let context_window_tokens = config.context_window_tokens(); @@ -43,6 +51,7 @@ impl From for AgentRuntimeConfig { max_images_in_context: config.max_images_in_context, max_image_age_rounds: config.max_image_age_rounds, max_retries: config.max_retries, + max_run_secs: DEFAULT_MAX_RUN_SECS, } } } diff --git a/src/bus/message.rs b/src/bus/message.rs index bbe01fc..f0c22e2 100644 --- a/src/bus/message.rs +++ b/src/bus/message.rs @@ -310,7 +310,16 @@ impl ChatMessage { /// /// Returns the number of messages removed. pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec) -> usize { + sanitize_incomplete_tool_call_sequences_with_ids(messages).0 +} + +/// 同 [sanitize_incomplete_tool_call_sequences]，但额外返回被移除消息的 id 列表。 +/// 调用方（历史加载路径）可据此把修复结果回写 DB，避免每次加载重复修复。 +pub(crate) fn sanitize_incomplete_tool_call_sequences_with_ids( + messages: &mut Vec, +) -> (usize, Vec) { let mut removed = 0; + let mut removed_ids: Vec = Vec::new(); // Phase 1: Single reverse pass to find ALL assistant messages with // incomplete tool_calls, regardless of position. @@ -461,6 +470,7 @@ pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec，需 clone 保留 serial_lock 供 coordinator 使用 - let serial_guard = serial_lock.clone().lock_owned().await; + // + // 可观测性：此前等待完全静默（无日志无反馈），前序 run 若长时间不结束， + // 用户视角即"发消息无任何响应"。此处每 30s 打一条 warn 标记等待进展， + // 拿到锁后打 info 收尾，便于从日志定位"卡在等锁"还是"卡在 run 内部"。 + let serial_guard = { + let lock_for_wait = serial_lock.clone(); + let mut waited = false; + loop { + match tokio::time::timeout( + std::time::Duration::from_secs(30), + lock_for_wait.clone().lock_owned(), + ) + .await + { + Ok(guard) => { + if waited { + tracing::info!( + topic_id = %lock_key, + "Topic serial lock acquired after waiting" + ); + } + break guard; + } + Err(_) => { + if !waited { + tracing::info!( + topic_id = %lock_key, + "Waiting for topic serial lock (previous run still in progress)" + ); + waited = true; + } else { + tracing::warn!( + topic_id = %lock_key, + "Still waiting for topic serial lock (previous run still in progress)" + ); + } + } + } + } + }; // guard_slot：wait_coordinator 通过此 slot 释放/重获取 serial_lock。 // 正常执行时 guard 留在 slot 中（锁持有）；wait 工具调用时 take guard 释放锁， diff --git a/src/gateway/processor.rs b/src/gateway/processor.rs index e31f0a2..388a6a3 100644 --- a/src/gateway/processor.rs +++ b/src/gateway/processor.rs @@ -40,6 +40,8 @@ pub struct InboundProcessor { command_router: Arc, cancel_manager: CancelManager, description_generation_in_flight: Arc>>, + /// 子代理运行时（用于 ExecutionCompleted 兜底前的僵尸清理） + subagent_executor: Option>, } impl InboundProcessor { @@ -120,7 +122,7 @@ impl InboundProcessor { command_router.register(Box::new(StopExecutionCommandHandler::new( cancel_manager.clone(), session_manager.clone(), - subagent_executor, + subagent_executor.clone(), ))); Self { @@ -131,6 +133,7 @@ impl InboundProcessor { command_router: Arc::new(command_router), cancel_manager, description_generation_in_flight: Arc::new(Mutex::new(HashSet::new())), + subagent_executor, } } @@ -180,6 +183,12 @@ impl InboundProcessor { let chat_id_for_span = inbound.chat_id.clone(); let session_id_for_span = crate::storage::persistent_session_id(&inbound.channel, &inbound.chat_id); + // panic 兜底需用：panic 时 inbound 已被 process_one 消费，提前克隆路由字段 + let panic_channel = inbound.channel.clone(); + let panic_chat_id = inbound.chat_id.clone(); + let panic_trace_id = inbound.trace_id.clone(); + let panic_session_id = session_id_for_span.clone(); + let panic_forwarded_metadata = inbound.forwarded_metadata.clone(); tokio::spawn(crate::observability::tracing_ctx::traced( &trace_id, &chat_id_for_span, @@ -202,11 +211,82 @@ impl InboundProcessor { crate::observability::metrics::record_message_processing_error(); } Err(payload) => { + let panic_msg = crate::utils::panic_payload_message(&payload); tracing::error!( - error = %crate::utils::panic_payload_message(&payload), + error = %panic_msg, "Message processing panicked" ); crate::observability::metrics::record_message_processing_error(); + + // panic 兜底：process_one 中途夭折，既不会发错误提示、 + // 也不会发 ExecutionCompleted，前端将永久停在 loading + // 且输入框被禁用（用户视角"卡死"）。此处补发两者， + // 并尽力清理该 topic 的取消信号注册。 + let current_topic = processor + .session_manager + .get_current_topic(&panic_channel, &panic_chat_id) + .await + .ok() + .flatten(); + if let Some(ref topic_id) = current_topic { + processor.cancel_manager.remove_by_topic(topic_id).await; + } + + let mut error_metadata = panic_forwarded_metadata.clone(); + error_metadata + .insert("error_kind".to_string(), "panic".to_string()); + if let Err(publish_error) = processor + .bus + .publish_outbound( + OutboundMessage::error_notification( + panic_channel.clone(), + panic_chat_id.clone(), + None, + format!("内部处理错误（panic）：{panic_msg}"), + None, + error_metadata, + ) + .with_trace_id(&panic_trace_id), + ) + .await + { + match publish_error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %publish_error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %publish_error, "Failed to publish panic error outbound"); + } + } + } + + let mut completion_metadata = panic_forwarded_metadata; + if let Some(ref topic_id) = current_topic { + completion_metadata + .insert("topic_id".to_string(), topic_id.clone()); + } + if let Err(publish_error) = processor + .bus + .publish_outbound( + OutboundMessage::execution_completed( + panic_channel, + panic_chat_id, + Some(panic_session_id), + completion_metadata, + ) + .with_trace_id(&panic_trace_id), + ) + .await + { + match publish_error { + crate::bus::BusError::Dropped => { + tracing::warn!(error = %publish_error, "Outbound dropped (bus full)"); + } + crate::bus::BusError::Closed => { + tracing::error!(error = %publish_error, "Failed to publish panic execution_completed"); + } + } + } } } }, @@ -338,7 +418,43 @@ impl InboundProcessor { // 阻塞获取 serial_lock // - agent 正常运行：阻塞至其完成（天然串行化） // - agent 在 wait 中：wait 已释放锁，可立即获取 - let _inject_guard = serial_lock.clone().lock_owned().await; + // 每 30s 打 warn 标记等待进展（与 execution.rs 主路径一致的可观测性） + let _inject_guard = { + let lock_for_wait = serial_lock.clone(); + let mut waited = false; + loop { + match tokio::time::timeout( + std::time::Duration::from_secs(30), + lock_for_wait.clone().lock_owned(), + ) + .await + { + Ok(guard) => { + if waited { + tracing::info!( + topic_id = %lock_key, + "Injection path serial lock acquired after waiting" + ); + } + break guard; + } + Err(_) => { + if !waited { + tracing::info!( + topic_id = %lock_key, + "Injection path waiting for topic serial lock" + ); + waited = true; + } else { + tracing::warn!( + topic_id = %lock_key, + "Injection path still waiting for topic serial lock" + ); + } + } + } + } + }; // 检查 is_waiting（持锁状态下安全） let is_waiting = { @@ -598,6 +714,13 @@ impl InboundProcessor { // 前端过早停止 loading 导致子代理结果"丢失"的观感。 // 恢复路径：下一条用户消息触发新的 process_one → 加载 history → // LLM 看到 "running" 占位 → 调用 wait_for_subagents → 消费 sub_done_q 结果。 + // + // 僵尸清理：先剔除"DB=running 但执行任务已消失"的僵尸记录， + // 否则它们会让 pending 永远非空 → ExecutionCompleted 永远被跳过 + // → 前端 loading 永不停止（用户视角"卡死"）。 + if let (Some(ref topic_id), Some(ref runtime)) = (current_topic.as_ref(), self.subagent_executor.as_ref()) { + runtime.reap_orphan_subagents(topic_id).await; + } let has_pending_subagents = if let Some(ref topic_id) = current_topic { // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker let store = self.session_manager.store(); diff --git a/src/gateway/session_history.rs b/src/gateway/session_history.rs index 0930991..fea6746 100644 --- a/src/gateway/session_history.rs +++ b/src/gateway/session_history.rs @@ -256,13 +256,43 @@ impl SessionHistory { .load_messages_for_topic(tid, Some(&sid)) .map_err(|err| AgentError::Other(format!("session history load error: {}", err)))?; - let removed = crate::bus::message::sanitize_incomplete_tool_call_sequences(&mut history); + let (removed, removed_ids) = + crate::bus::message::sanitize_incomplete_tool_call_sequences_with_ids(&mut history); if removed > 0 { tracing::warn!( topic_id = %tid, removed_count = removed, "Sanitized incomplete tool_call sequences on history load" ); + // 修复回写 DB：否则损坏序列在 DB 中永久残留，每次加载都重复 + // sanitize（日志噪音 + 内存/DB 漂移）。失败不阻断加载，下次加载重试。 + match self + .conversations + .delete_messages_by_ids(&sid, &removed_ids) + { + Ok(deleted) if deleted == removed => { + tracing::info!( + topic_id = %tid, + deleted_count = deleted, + "Persisted sanitize repair to DB (deleted broken message rows)" + ); + } + Ok(deleted) => { + tracing::warn!( + topic_id = %tid, + deleted_count = deleted, + expected = removed, + "Partial sanitize repair persisted to DB" + ); + } + Err(e) => { + tracing::warn!( + error = %e, + topic_id = %tid, + "Failed to persist sanitize repair to DB; will retry on next load" + ); + } + } } self.topic_histories.insert(tid.to_string(), history); diff --git a/src/storage/mod.rs b/src/storage/mod.rs index f19beb5..e666697 100644 --- a/src/storage/mod.rs +++ b/src/storage/mod.rs @@ -630,6 +630,35 @@ impl SessionStore { Ok(()) } + /// 按 id 批量删除指定 session 的消息行（历史加载时 sanitize 修复回写用）。 + /// 返回实际删除的行数。 + pub fn delete_messages_by_ids( + &self, + session_id: &str, + ids: &[String], + ) -> Result { + if ids.is_empty() { + return Ok(0); + } + let conn = self.pool.get()?; + let mut total = 0; + // SQLite 绑定变量上限（SQLITE_MAX_VARIABLE_NUMBER 默认 999）， + // 分批构造 IN 子句，避免超限。 + for chunk in ids.chunks(500) { + let placeholders = vec!["?"; chunk.len()].join(","); + let sql = format!( + "DELETE FROM messages WHERE session_id = ? AND id IN ({placeholders})" + ); + let mut params_vec: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(chunk.len() + 1); + params_vec.push(&session_id); + for id in chunk { + params_vec.push(id); + } + total += conn.execute(&sql, params_vec.as_slice())?; + } + Ok(total) + } + pub fn clear_messages(&self, session_id: &str) -> Result<(), StorageError> { let now = current_timestamp(); let conn = self.pool.get()?; diff --git a/src/storage/ports.rs b/src/storage/ports.rs index edf8dee..0d6efa9 100644 --- a/src/storage/ports.rs +++ b/src/storage/ports.rs @@ -58,6 +58,13 @@ pub trait ConversationRepository: Send + Sync + 'static { fn clear_messages(&self, session_id: &str) -> Result<(), StorageError>; + /// 按 id 批量删除指定 session 的消息行（历史加载时 sanitize 修复回写用）。 + fn delete_messages_by_ids( + &self, + session_id: &str, + ids: &[String], + ) -> Result; + fn compact_active_history( &self, session_id: &str, @@ -259,6 +266,14 @@ impl ConversationRepository for super::SessionStore { super::SessionStore::clear_messages(self, session_id) } + fn delete_messages_by_ids( + &self, + session_id: &str, + ids: &[String], + ) -> Result { + super::SessionStore::delete_messages_by_ids(self, session_id, ids) + } + fn compact_active_history( &self, session_id: &str, diff --git a/src/storage/tests.rs b/src/storage/tests.rs index 5fd8bc4..2d86038 100644 --- a/src/storage/tests.rs +++ b/src/storage/tests.rs @@ -882,3 +882,105 @@ fn test_cleanup_legacy_empty_cli_sessions() { assert!(store.get_session(&with_data.id).unwrap().is_some()); assert!(store.get_session(&ws.id).unwrap().is_some()); } + +#[test] +fn test_load_messages_for_topic_page_keyset_pagination() { + let store = SessionStore::in_memory().unwrap(); + let session = store.create_cli_session(Some("paged")).unwrap(); + let topic = store.create_topic(&session.id, "topic-page", None).unwrap(); + + for i in 1..=10 { + store + .append_message_with_topic( + &session.id, + Some(&topic.id), + &ChatMessage::user(format!("m{i}")), + ) + .unwrap(); + } + + // 首页：无游标，取最新 4 条（正序），且标记还有更早消息 + let (page1, has_more1) = store + .load_messages_for_topic_page(&topic.id, Some(&session.id), None, 4) + .unwrap(); + assert!(has_more1); + let contents: Vec<_> = page1.iter().map(|m| m.content.as_str()).collect(); + assert_eq!(contents, vec!["m7", "m8", "m9", "m10"]); + // 历史加载路径必须填充 seq 作为下一页游标 + let oldest1 = page1.first().and_then(|m| m.seq).expect("seq populated"); + + // 第二页：before_seq 游标，取更早的 4 条 + let (page2, has_more2) = store + .load_messages_for_topic_page(&topic.id, Some(&session.id), Some(oldest1), 4) + .unwrap(); + assert!(has_more2); + let contents: Vec<_> = page2.iter().map(|m| m.content.as_str()).collect(); + assert_eq!(contents, vec!["m3", "m4", "m5", "m6"]); + let oldest2 = page2.first().and_then(|m| m.seq).expect("seq populated"); + + // 末页：剩余 2 条，has_more=false + let (page3, has_more3) = store + .load_messages_for_topic_page(&topic.id, Some(&session.id), Some(oldest2), 4) + .unwrap(); + assert!(!has_more3); + let contents: Vec<_> = page3.iter().map(|m| m.content.as_str()).collect(); + assert_eq!(contents, vec!["m1", "m2"]); + + // 游标越过最老消息：空页且 has_more=false + let oldest3 = page3.first().and_then(|m| m.seq).unwrap(); + let (page4, has_more4) = store + .load_messages_for_topic_page(&topic.id, Some(&session.id), Some(oldest3), 4) + .unwrap(); + assert!(!has_more4); + assert!(page4.is_empty()); + + // 拼接所有页 == 全量加载（顺序与内容一致） + let mut combined = page3; + combined.extend(page2); + combined.extend(page1); + let full = store.load_messages_for_topic(&topic.id, None).unwrap(); + let combined_contents: Vec<_> = combined.iter().map(|m| m.content.as_str()).collect(); + let full_contents: Vec<_> = full.iter().map(|m| m.content.as_str()).collect(); + assert_eq!(combined_contents, full_contents); +} + +#[test] +fn test_delete_messages_by_ids_removes_only_target_rows() { + let store = SessionStore::in_memory().unwrap(); + let session = store.create_cli_session(Some("del")).unwrap(); + let topic = store.create_topic(&session.id, "topic-del", None).unwrap(); + + for i in 1..=5 { + store + .append_message_with_topic( + &session.id, + Some(&topic.id), + &ChatMessage::user(format!("m{i}")), + ) + .unwrap(); + } + + let all = store.load_messages_for_topic(&topic.id, None).unwrap(); + assert_eq!(all.len(), 5); + + // 删除中间 2 条（sanitize 回写场景：只删被清理的消息） + let to_delete: Vec = all[1..3].iter().map(|m| m.id.clone()).collect(); + let deleted = store.delete_messages_by_ids(&session.id, &to_delete).unwrap(); + assert_eq!(deleted, 2); + + let remaining = store.load_messages_for_topic(&topic.id, None).unwrap(); + assert_eq!(remaining.len(), 3); + let contents: Vec<_> = remaining.iter().map(|m| m.content.as_str()).collect(); + assert_eq!(contents, vec!["m1", "m4", "m5"]); + + // 空列表 no-op；不存在的 id 返回 0 + assert_eq!(store.delete_messages_by_ids(&session.id, &[]).unwrap(), 0); + assert_eq!( + store + .delete_messages_by_ids(&session.id, &["nonexistent".to_string()]) + .unwrap(), + 0 + ); + // 消息计数同步修正 + assert_eq!(store.get_topic_message_count(&topic.id).unwrap(), 3); +} diff --git a/src/tools/task/runtime.rs b/src/tools/task/runtime.rs index 8f9652f..7c13f5d 100644 --- a/src/tools/task/runtime.rs +++ b/src/tools/task/runtime.rs @@ -148,6 +148,19 @@ pub trait SubAgentRuntime: Send + Sync + 'static { /// 用于 /stop 命令传播：用户取消主 agent 时，同步取消其后台子代理。 /// 返回被触发取消的子代理数量。 async fn cancel_pending_for_topic(&self, topic_id: &str) -> usize; + + /// 清理指定 topic 下的"僵尸"子代理：DB 状态为 running， + /// 但其执行任务已不存在（不在 cancel_registry 中——执行器 panic/被杀/ + /// 写库失败后残留）。将这些记录标记为 interrupted。 + /// + /// 用于 ExecutionCompleted 兜底判定前的惰性对账： + /// 僵尸记录会让"pending running 非空"永远成立，从而永远跳过 + /// ExecutionCompleted，前端 loading 永不停止（用户视角即"卡死"）。 + /// 返回被清理的记录数。 + async fn reap_orphan_subagents(&self, topic_id: &str) -> usize { + let _ = topic_id; + 0 + } } /// 静态系统提示词提供者（用于子代理） @@ -1445,6 +1458,68 @@ impl SubAgentRuntime for DefaultSubAgentRuntime { count } + + async fn reap_orphan_subagents(&self, topic_id: &str) -> usize { + let running = match self.store.list_pending_subagents(topic_id, Some("running")) { + Ok(records) => records, + Err(e) => { + tracing::warn!( + error = %e, + topic_id = %topic_id, + "Failed to list pending subagents for zombie reaping" + ); + return 0; + } + }; + if running.is_empty() { + return 0; + } + + let mut reaped = 0; + let registry = self.cancel_registry.lock(); + for record in &running { + // 执行任务仍在（registry 有 token）→ 真正在跑，保留 + if registry.contains_key(&record.task_id) { + continue; + } + // 僵尸：DB=running 但执行任务已消失。 + // 条件 UPDATE（仅 running→interrupted），不覆盖已终态。 + match self.store.try_update_pending_subagent_status( + &record.task_id, + "running", + "interrupted", + ) { + Ok(true) => { + reaped += 1; + tracing::warn!( + task_id = %record.task_id, + def_name = ?record.def_name, + spawned_at = ?record.spawned_at, + "Reaped zombie subagent (DB running but executor gone); marked interrupted" + ); + } + Ok(false) => {} + Err(e) => { + tracing::warn!( + error = %e, + task_id = %record.task_id, + "Failed to reap zombie subagent" + ); + } + } + } + drop(registry); + + if reaped > 0 { + tracing::info!( + topic_id = %topic_id, + reaped, + remaining_running = running.len() - reaped, + "Zombie subagent reaping finished" + ); + } + reaped + } } /// 子代理定义目录