fix(gateway): 修复长会话后新消息无响应——僵尸子代理清理、panic 兜底与运行预算
- 发送 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 批量删除的单元测试
This commit is contained in:
oudecheng
parent
a629486ad3
commit
452c5ad0ef
@ -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<ChatMessage>,
|
||||
) -> 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;
|
||||
|
||||
@ -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<LLMProviderConfig> for AgentRuntimeConfig {
|
||||
fn from(config: LLMProviderConfig) -> Self {
|
||||
let context_window_tokens = config.context_window_tokens();
|
||||
@ -43,6 +51,7 @@ impl From<LLMProviderConfig> 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,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -310,7 +310,16 @@ impl ChatMessage {
|
||||
///
|
||||
/// Returns the number of messages removed.
|
||||
pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMessage>) -> 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<ChatMessage>,
|
||||
) -> (usize, Vec<String>) {
|
||||
let mut removed = 0;
|
||||
let mut removed_ids: Vec<String> = 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<ChatMes
|
||||
remove_indices.sort_unstable_by(|a, b| b.cmp(a));
|
||||
remove_indices.dedup();
|
||||
for &idx in &remove_indices {
|
||||
removed_ids.push(messages[idx].id.clone());
|
||||
messages.remove(idx);
|
||||
removed += 1;
|
||||
}
|
||||
@ -490,6 +500,7 @@ pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMes
|
||||
"Removing orphaned tool result message — its parent assistant \
|
||||
tool_calls message was removed or never persisted"
|
||||
);
|
||||
removed_ids.push(msg.id.clone());
|
||||
messages.remove(i);
|
||||
removed += 1;
|
||||
continue;
|
||||
@ -499,7 +510,7 @@ pub(crate) fn sanitize_incomplete_tool_call_sequences(messages: &mut Vec<ChatMes
|
||||
}
|
||||
}
|
||||
|
||||
removed
|
||||
(removed, removed_ids)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
|
||||
@ -311,7 +311,46 @@ impl AgentExecutionService {
|
||||
// await 串行锁时不持有 session 锁，其他 topic 的消息可以正常处理
|
||||
// 使用 lock_owned 获取 OwnedMutexGuard，存入 guard_slot 供 wait_coordinator 释放/重获取
|
||||
// 注意：lock_owned 消费 Arc<Self>，需 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 释放锁，
|
||||
|
||||
@ -40,6 +40,8 @@ pub struct InboundProcessor {
|
||||
command_router: Arc<CommandRouter>,
|
||||
cancel_manager: CancelManager,
|
||||
description_generation_in_flight: Arc<Mutex<HashSet<String>>>,
|
||||
/// 子代理运行时（用于 ExecutionCompleted 兜底前的僵尸清理）
|
||||
subagent_executor: Option<Arc<dyn crate::tools::SubAgentRuntime>>,
|
||||
}
|
||||
|
||||
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();
|
||||
|
||||
@ -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);
|
||||
|
||||
@ -630,6 +630,35 @@ impl SessionStore {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 按 id 批量删除指定 session 的消息行（历史加载时 sanitize 修复回写用）。
|
||||
/// 返回实际删除的行数。
|
||||
pub fn delete_messages_by_ids(
|
||||
&self,
|
||||
session_id: &str,
|
||||
ids: &[String],
|
||||
) -> Result<usize, StorageError> {
|
||||
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()?;
|
||||
|
||||
@ -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<usize, StorageError>;
|
||||
|
||||
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<usize, StorageError> {
|
||||
super::SessionStore::delete_messages_by_ids(self, session_id, ids)
|
||||
}
|
||||
|
||||
fn compact_active_history(
|
||||
&self,
|
||||
session_id: &str,
|
||||
|
||||
@ -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<String> = 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);
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// 子代理定义目录
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user