use futures_util::FutureExt;
use parking_lot::Mutex;
use std::collections::HashSet;
use std::sync::Arc;

use tokio::sync::Semaphore;

use crate::agent::{AgentError, PersistingEmittedMessageHandler};
use crate::bus::{InboundMessage, MessageBus, OutboundMessage};
use crate::command::adapter::InputAdapter;
use crate::command::adapters::channel::ChannelInputAdapter;
use crate::command::handler::CommandRouter;
use crate::command::handlers::delete_topic::DeleteTopicCommandHandler;
use crate::command::handlers::get_current::GetCurrentSessionCommandHandler;
use crate::command::handlers::help::HelpCommandHandler;
use crate::command::handlers::list_sessions::ListSessionsCommandHandler;
use crate::command::handlers::load_topic::LoadTopicCommandHandler;
use crate::command::handlers::rename_topic::RenameTopicCommandHandler;
use crate::command::handlers::save_session::SaveSessionCommandHandler;
use crate::command::handlers::save_topic::SaveTopicCommandHandler;
use crate::command::handlers::session::SessionCommandHandler;
use crate::command::handlers::stop_execution::StopExecutionCommandHandler;
use crate::command::handlers::switch_topic::SwitchTopicCommandHandler;
use crate::config::LLMProviderConfig;
use crate::gateway::agent_factory::build_system_prompt_provider;
use crate::gateway::cancel_manager::CancelManager;
use crate::providers::{ProviderRuntimeConfig, create_provider};
use crate::storage::persistent_session_id;
use crate::topic_description::generate_topic_description;

use super::message_prepare::enrich_user_content_with_media_refs;
use super::session::{BusToolCallEmitter, SessionManager};

#[derive(Clone)]
pub struct InboundProcessor {
    bus: Arc<MessageBus>,
    session_manager: SessionManager,
    semaphore: Arc<Semaphore>,
    provider_config: LLMProviderConfig,
    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 {
    pub fn new(
        bus: Arc<MessageBus>,
        session_manager: SessionManager,
        semaphore: Arc<Semaphore>,
        provider_config: LLMProviderConfig,
        cancel_manager: CancelManager,
        subagent_executor: Option<Arc<dyn crate::tools::SubAgentRuntime>>,
    ) -> Self {
        // 创建命令路由器并注册处理器
        let mut command_router = CommandRouter::new();
        let store = session_manager.store();

        // 注册 Session 处理器
        let session_handler =
            SessionCommandHandler::new(store.clone()).with_session_manager(session_manager.clone());
        command_router.register(Box::new(session_handler));

        // 注册 list_sessions 处理器
        command_router.register(Box::new(ListSessionsCommandHandler::new(store.clone())));

        // 注册 switch_topic 处理器
        let switch_handler = SwitchTopicCommandHandler::new(store.clone())
            .with_session_manager(session_manager.clone());
        command_router.register(Box::new(switch_handler));

        // 创建 system_prompt_provider（用于 save_session, save_topic, get_current）
        // 与 AgentFactory::create 共享同一构建逻辑，确保保存到文件的系统提示词
        // 与 LLM 实际接收的提示词完全一致（含 Expert/Subagent/Todo）
        let system_prompt_provider = build_system_prompt_provider(
            0, // 命令侧不需要 reinject 逻辑
            provider_config.clone(),
            session_manager.store().clone(),
            session_manager.skills(),
            session_manager.experts(),
            session_manager.subagent_runtime(),
        );

        // 注册 get_current 处理器
        command_router.register(Box::new(
            GetCurrentSessionCommandHandler::new(store.clone())
                .with_system_prompt_provider(system_prompt_provider.clone()),
        ));

        // 注册 load_topic 处理器
        command_router.register(Box::new(LoadTopicCommandHandler::new(store.clone())));

        // 注册 save_session 处理器
        command_router.register(Box::new(SaveSessionCommandHandler::new(
            store.clone(),
            session_manager.task_repository(),
            system_prompt_provider.clone(),
        )));

        // 注册 save_topic 处理器
        command_router.register(Box::new(SaveTopicCommandHandler::new(
            store.clone(),
            session_manager.task_repository(),
            system_prompt_provider,
        )));

        // 注册 delete_topic 处理器
        command_router.register(Box::new(
            DeleteTopicCommandHandler::new(store.clone())
                .with_session_manager(session_manager.clone()),
        ));

        // 注册 rename_topic 处理器
        command_router.register(Box::new(RenameTopicCommandHandler::new(store.clone())));

        // 注册 help 处理器（最后注册，获取所有已注册命令的元数据）
        let metadata = command_router.metadata_arc();
        command_router.register(Box::new(HelpCommandHandler::new(metadata)));

        // 注册 stop_execution 处理器
        command_router.register(Box::new(StopExecutionCommandHandler::new(
            cancel_manager.clone(),
            session_manager.clone(),
            subagent_executor.clone(),
        )));

        Self {
            bus,
            session_manager,
            semaphore,
            provider_config,
            command_router: Arc::new(command_router),
            cancel_manager,
            description_generation_in_flight: Arc::new(Mutex::new(HashSet::new())),
            subagent_executor,
        }
    }

    pub async fn run(self) {
        let max_concurrent = self.semaphore.available_permits();
        tracing::info!(
            max_concurrent_requests = max_concurrent,
            "Inbound processor started"
        );

        loop {
            // 1. 消费消息 (channel 关闭时返回 None，优雅退出)
            let inbound = match self.bus.consume_inbound().await {
                Some(msg) => msg,
                None => {
                    tracing::info!("Inbound bus closed, stopping inbound processor");
                    break;
                }
            };

            tracing::debug!(
                channel = %inbound.channel,
                chat_id = %inbound.chat_id,
                trace_id = %inbound.trace_id,
                sender = %inbound.sender_id,
                content_len = %inbound.content.len(),
                media_count = %inbound.media.len(),
                "Processing inbound message"
            );

            // 2. 获取 semaphore permit（控制并发）
            let permit = match self.semaphore.clone().acquire_owned().await {
                Ok(permit) => permit,
                Err(_) => {
                    tracing::error!("Semaphore closed, stopping inbound processor");
                    break;
                }
            };

            // 3. 克隆 processor 用于新任务
            let processor = self.clone();

            // 4. 独立任务处理（包含 permit，任务完成自动释放）
            // spawn 不自动传播父 span，用 traced() 重建 span 上下文，
            // 使 process_one 内所有日志携带 trace_id/chat_id/session_id。
            let trace_id = inbound.trace_id.clone();
            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,
                &session_id_for_span,
                async move {
                    let _permit = permit; // 持有 permit 直到任务完成
                    // catch_unwind 将 panic 归一化为错误：否则工具/历史清理中的
                    // panic 只会终止任务并打 panic hook 日志，跳过错误日志与指标，
                    // 用户消息被静默吞掉。参考 channels/wechat.rs 的同类用法。
                    let result = std::panic::AssertUnwindSafe(processor.process_one(inbound))
                        .catch_unwind()
                        .await;
                    match result {
                        Ok(Ok(())) => {}
                        Ok(Err(e)) => {
                            tracing::error!(
                                error = %crate::utils::format_error_chain(&e),
                                "Message processing failed"
                            );
                            crate::observability::metrics::record_message_processing_error();
                        }
                        Err(payload) => {
                            let panic_msg = crate::utils::panic_payload_message(&payload);
                            tracing::error!(
                                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");
                                    }
                                }
                            }
                        }
                    }
                },
            ));
        }
    }

    #[tracing::instrument(skip(self, inbound), fields(trace_id = %inbound.trace_id, chat_id = %inbound.chat_id, session_id))]
    async fn process_one(&self, inbound: InboundMessage) -> Result<(), AgentError> {
        // 计算正确的 session_id（根据 channel_name 和 chat_id）
        let session_id = persistent_session_id(&inbound.channel, &inbound.chat_id);
        tracing::Span::current().record("session_id", tracing::field::display(&session_id));

        // 获取当前话题（封装了 session 创建逻辑）
        let current_topic = self
            .session_manager
            .get_current_topic(&inbound.channel, &inbound.chat_id)
            .await?;

        // 使用 ChannelInputAdapter 尝试解析命令
        let adapter = ChannelInputAdapter::new();
        let ctx = crate::command::context::AdapterContext::new(&inbound.channel)
            .with_session_id(&session_id);

        if let Ok(Some(cmd)) = adapter.try_parse(&inbound.content, ctx) {
            // 使用命令路由器处理
            let mut cmd_ctx =
                crate::command::context::CommandContext::new(&inbound.channel, &inbound.channel)
                    .with_session_id(&session_id)
                    .with_chat_id(&inbound.chat_id);
            // 只在有话题时才设置 topic_id
            if let Some(ref topic_id) = current_topic {
                cmd_ctx = cmd_ctx.with_topic_id(topic_id.as_str());
            }

            let response = self
                .command_router
                .dispatch_with_response(cmd, cmd_ctx)
                .await;

            // 发送响应给用户
            if response.success {
                // 提取响应消息
                // chat_id 保持为 inbound.chat_id（飞书 open_id）
                // session_id 放入 metadata 用于会话管理
                for msg in &response.messages {
                    if let Err(error) = self
                        .bus
                        .publish_outbound(
                            OutboundMessage::assistant(
                                inbound.channel.clone(),
                                inbound.chat_id.clone(),
                                response.metadata.get("session_id").cloned(),
                                msg.content.clone(),
                                None,
                                inbound.forwarded_metadata.clone(),
                            )
                            .with_trace_id(&inbound.trace_id),
                        )
                        .await
                    {
                        match error {
                            crate::bus::BusError::Dropped => {
                                tracing::warn!(error = %error, "Outbound dropped (bus full)");
                            }
                            crate::bus::BusError::Closed => {
                                tracing::error!(error = %error, "Failed to publish command response");
                            }
                        }
                    }
                }
            } else if let Some(error) = response.error
                && let Err(e) = self
                    .bus
                    .publish_outbound(
                        OutboundMessage::assistant(
                            inbound.channel.clone(),
                            inbound.chat_id.clone(),
                            response.metadata.get("session_id").cloned(),
                            format!("Error [{}]: {}", error.code, error.message),
                            None,
                            inbound.forwarded_metadata.clone(),
                        )
                        .with_trace_id(&inbound.trace_id),
                    )
                    .await
            {
                match e {
                    crate::bus::BusError::Dropped => {
                        tracing::warn!(error = %e, "Outbound dropped (bus full)");
                    }
                    crate::bus::BusError::Closed => {
                        tracing::error!(error = %e, "Failed to publish error response");
                    }
                }
            }
            return Ok(());
        }

        // 普通消息进入 AgentLoop
        // 构建 emitter metadata：包含 forwarded_metadata 和 topic_id（用于前端消息隔离）
        let mut emitter_metadata = inbound.forwarded_metadata.clone();
        if let Some(ref topic_id) = current_topic {
            emitter_metadata.insert("topic_id".to_string(), topic_id.clone());
        }

        // ── 异步子代理等待注入路径 ──
        // 当主 agent 正在 wait_for_subagents 中等待（已释放 serial_lock、is_waiting=true）时，
        // 新用户消息不应启动新的 agent loop，而应注入 history 并唤醒等待中的 agent。
        //
        // 流程：
        // 1. 获取 serial_lock（若 agent 正常运行则阻塞；若 agent 在 wait 中则立即获取）
        // 2. 检查 is_waiting：true → 注入 + wakeup + return；false → 释放锁走正常路径
        //
        // 安全性：is_waiting 在持锁状态下检查，wait_coordinator 清除 is_waiting 需先重获取锁，
        //        两者互斥，无 TOCTOU。
        if let Some(ref topic_id) = current_topic
            && let Some(session) = self.session_manager.get(&inbound.channel).await
        {
            let lock_key = topic_id.clone();

            // 获取 serial_lock Arc（短暂持有 session 锁）
            let serial_lock = {
                let mut g = session.lock().await;
                g.ensure_sub_done_channel(&lock_key);
                g.topic_serial_lock(&lock_key)
            };

            // 阻塞获取 serial_lock
            // - agent 正常运行：阻塞至其完成（天然串行化）
            // - agent 在 wait 中：wait 已释放锁，可立即获取
            // 每 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 = {
                let g = session.lock().await;
                g.is_waiting(&lock_key)
            };

            if is_waiting {
                // Agent 正在 wait_for_subagents 中等待 → 注入用户消息 + 唤醒
                tracing::info!(
                    topic_id = %lock_key,
                    "Topic is in waiting state, injecting user message and waking up agent"
                );

                let wakeup = {
                    let mut g = session.lock().await;
                    // 确保 session 和 chat 已加载
                    g.ensure_persistent_session(&inbound.chat_id)?;
                    g.ensure_chat_loaded(&inbound.chat_id, Some(&lock_key))?;

                    // 构造用户消息（与 prepare_and_execute_message 一致的处理流程）
                    let media_refs: Vec<String> =
                        inbound.media.iter().map(|m| m.path.clone()).collect();
                    let enriched_content =
                        enrich_user_content_with_media_refs(&inbound.content, &media_refs)?;
                    let user_message = g.create_user_message(&enriched_content, media_refs);
                    g.append_persisted_message(&inbound.chat_id, Some(&lock_key), user_message)?;

                    // 获取 wakeup 信号
                    g.wait_wakeup(&lock_key)
                };

                // 唤醒等待中的 agent（wait_coordinator 的 select! 会捕获此通知）
                wakeup.notify_one();

                // _inject_guard 在此处 drop → 释放 serial_lock
                // wait_coordinator 重获取锁后继续处理（history 已包含新用户消息）
                //
                // 跳过 handle_message / cancel 注册 / execution_completed，
                // 因为等待中的 agent 会处理这条消息。
                return Ok(());
            }
            // is_waiting=false：_inject_guard drop 释放锁，走正常 handle_message 路径
        }

        let live_emitter = Arc::new(PersistingEmittedMessageHandler::new(
            BusToolCallEmitter::new(
                self.bus.clone(),
                inbound.channel.clone(),
                inbound.chat_id.clone(),
                emitter_metadata,
                self.session_manager.store(),
                inbound.trace_id.clone(),
            ),
            self.session_manager.store(),
            &session_id,
            current_topic.clone(),
        ));

        // 保存 channel 和 chat_id 用于后续清理（因 match 中可能 move inbound）
        let channel = inbound.channel.clone();
        let chat_id = inbound.chat_id.clone();

        // 按 topic_id 注册取消信号：Agent 构建时通过 Session 消费该 receiver
        if let Some(ref topic_id) = current_topic {
            let cancel_rx = self.cancel_manager.register(topic_id).await;
            self.session_manager
                .set_agent_cancel_token(&channel, &chat_id, Some(topic_id.as_str()), cancel_rx)
                .await;
        }

        match self
            .session_manager
            .handle_message(
                &inbound.channel,
                &inbound.sender_id,
                &inbound.chat_id,
                &inbound.content,
                inbound.media,
                Some(live_emitter),
                current_topic.as_deref(),
                &inbound.trace_id,
            )
            .await
        {
            Ok(outbound_messages) => {
                for mut outbound in outbound_messages {
                    outbound.metadata.extend(inbound.forwarded_metadata.clone());
                    // 注入 topic_id 到 outbound metadata，用于前端按话题隔离消息
                    if let Some(ref topic_id) = current_topic {
                        outbound
                            .metadata
                            .insert("topic_id".to_string(), topic_id.clone());
                    }
                    // 透传 trace_id 到出站消息，保持端到端追踪贯通
                    outbound.trace_id = inbound.trace_id.clone();
                    if let Err(error) = self.bus.publish_outbound(outbound).await {
                        match error {
                            crate::bus::BusError::Dropped => {
                                tracing::warn!(error = %error, "Outbound dropped (bus full)");
                            }
                            crate::bus::BusError::Closed => {
                                tracing::error!(error = %error, "Failed to publish outbound");
                            }
                        }
                    }
                }

                // 异步生成 topic 描述（仅当描述为空且没有正在进行的生成任务时触发）
                if let Some(ref topic_id) = current_topic {
                    let store = self.session_manager.store();
                    // SQLite 是同步 I/O：放到 blocking 线程池，避免阻塞 tokio worker
                    let store_for_lookup = store.clone();
                    let topic_id_for_lookup = topic_id.clone();
                    let topic_row = tokio::task::spawn_blocking(move || {
                        store_for_lookup.get_topic(&topic_id_for_lookup)
                    })
                    .await
                    .ok()
                    .and_then(|r| r.ok())
                    .unwrap_or(None);
                    if let Some(topic) = topic_row
                        && (topic.description.is_none()
                            || topic
                                .description
                                .as_ref()
                                .map(|d| d.is_empty())
                                .unwrap_or(true))
                    {
                        // 检查并设置"生成中"守卫，防止竞态条件导致重复生成
                        let should_generate = {
                            let mut in_flight = self.description_generation_in_flight.lock();
                            if in_flight.contains(topic_id) {
                                false
                            } else {
                                in_flight.insert(topic_id.clone());
                                true
                            }
                        };

                        if should_generate {
                            let provider_config = self.provider_config.clone();
                            let topic_id_clone = topic_id.clone();
                            let store_clone = store.clone();
                            let in_flight = self.description_generation_in_flight.clone();

                            tokio::spawn(async move {
                                // 定向查询该 topic 的第一条用户消息（DB 侧 LIMIT 1，
                                // 不再全量加载整个话题历史），并放到 blocking 线程池执行
                                let store_for_query = store_clone.clone();
                                let topic_id_for_query = topic_id_clone.clone();
                                let first_user_message = tokio::task::spawn_blocking(move || {
                                    store_for_query.first_user_message_content(&topic_id_for_query)
                                })
                                .await
                                .ok()
                                .and_then(|r| r.ok())
                                .unwrap_or(None);

                                let message_content = match first_user_message {
                                    Some(content) => content,
                                    None => {
                                        tracing::warn!(topic_id = %topic_id_clone, "No user message found for topic, skipping description generation");
                                        in_flight.lock().remove(&topic_id_clone);
                                        return;
                                    }
                                };

                                let runtime_config: ProviderRuntimeConfig = provider_config.into();
                                if let Ok(provider) = create_provider(runtime_config) {
                                    match generate_topic_description(
                                        provider.as_ref(),
                                        &message_content,
                                    )
                                    .await
                                    {
                                        Ok(description) => {
                                            let store_for_update = store_clone.clone();
                                            let topic_id_for_update = topic_id_clone.clone();
                                            let description_for_update = description.clone();
                                            let update_result =
                                                tokio::task::spawn_blocking(move || {
                                                    store_for_update.update_topic_description(
                                                        &topic_id_for_update,
                                                        &description_for_update,
                                                    )
                                                })
                                                .await;
                                            match update_result {
                                                Ok(Ok(())) => {
                                                    tracing::info!(topic_id = %topic_id_clone, description = %description, "Topic description generated");
                                                }
                                                Ok(Err(e)) => {
                                                    tracing::error!(error = %e, topic_id = %topic_id_clone, "Failed to update topic description");
                                                }
                                                Err(e) => {
                                                    tracing::error!(error = %e, topic_id = %topic_id_clone, "Topic description update task panicked");
                                                }
                                            }
                                        }
                                        Err(e) => {
                                            tracing::error!(error = %e, topic_id = %topic_id_clone, "Failed to generate topic description");
                                        }
                                    }
                                }
                                // 无论成功失败，释放生成守卫
                                in_flight.lock().remove(&topic_id_clone);
                            });
                        }
                    }
                }
            }
            Err(error) => {
                tracing::error!(
                    error = %crate::utils::format_error_chain(&error),
                    "Failed to handle message"
                );
                crate::observability::metrics::record_message_processing_error();
                let mut metadata = inbound.forwarded_metadata.clone();
                metadata.insert("error_kind".to_string(), "agent_execution".to_string());
                if let Err(publish_error) = self
                    .bus
                    .publish_outbound(
                        OutboundMessage::error_notification(
                            inbound.channel,
                            inbound.chat_id,
                            None, // session_id
                            error.to_string(),
                            None,
                            metadata,
                        )
                        .with_trace_id(&inbound.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 execution error outbound");
                        }
                    }
                }
            }
        }

        // 清理取消信号注册（幂等：如果已被 cancel_by_topic() 移除则为 no-op）
        if let Some(ref topic_id) = current_topic {
            self.cancel_manager.remove_by_topic(topic_id).await;
        }

        // 发送执行完成信号，通知前端可以停止 loading 状态。
        //
        // 退出兜底（safety net）：如果当前 topic 仍有 running 状态的子代理，
        // 不发送 ExecutionCompleted。这防止 LLM 未调用 wait_for_subagents 就退出时，
        // 前端过早停止 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();
            let topic_id_for_query = topic_id.clone();
            let pending = tokio::task::spawn_blocking(move || {
                store.list_pending_subagents(&topic_id_for_query, Some("running"))
            })
            .await
            .ok()
            .and_then(|r| r.ok())
            .unwrap_or_default();
            if !pending.is_empty() {
                tracing::debug!(
                    topic_id = %topic_id,
                    pending_count = pending.len(),
                    "Skipping ExecutionCompleted: pending subagents still running"
                );
                true
            } else {
                false
            }
        } else {
            false
        };

        if !has_pending_subagents {
            let mut completion_metadata = inbound.forwarded_metadata.clone();
            if let Some(ref topic_id) = current_topic {
                completion_metadata.insert("topic_id".to_string(), topic_id.clone());
            }
            if let Err(error) = self
                .bus
                .publish_outbound(
                    OutboundMessage::execution_completed(
                        channel,
                        chat_id,
                        Some(session_id),
                        completion_metadata,
                    )
                    .with_trace_id(&inbound.trace_id),
                )
                .await
            {
                match error {
                    crate::bus::BusError::Dropped => {
                        tracing::warn!(error = %error, "Outbound dropped (bus full)");
                    }
                    crate::bus::BusError::Closed => {
                        tracing::error!(error = %error, "Failed to publish execution_completed");
                    }
                }
            }
        }

        Ok(())
    }
}