use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Instant;

use anyhow::{Context, Result};
use tokio::sync::Mutex;
use tracing::{debug, error, info, warn};

use crate::config::SearchConfig;
use crate::db::{KuzuGraph, LanceDb, SqliteDb};
use crate::knowledge::KnowledgeManager;
use crate::types::{
    LlmProvider, QueryType, SearchRequest, SearchResponse, SearchResult, SearchSource,
    SearchStrategy, TierBreakdown,
};

use super::cache::SearchCache;
use super::cost_optimizer::CostOptimizer;
use super::freshness::FreshnessManager;
use super::metrics::SearchMetrics;
use super::query_classifier::QueryClassifier;
use super::result_judge::ResultJudge;
use super::strategy_selector::StrategySelector;
use super::tier1_vector::Tier1Search;
use super::tier2_hybrid::Tier2Search;
use super::tier3_agentic::Tier3Search;

/// The main search orchestrator that coordinates the entire tiered search
/// pipeline. It owns or references every sub-component and drives the
/// following lifecycle for each query:
///
/// 1. Check the result cache.
/// 2. Classify the query type (via LLM or heuristics).
/// 3. Choose a search strategy (adaptive, budget-aware).
/// 4. Execute the strategy tier-by-tier, escalating when results are
///    insufficient.
/// 5. Cache the response and record metrics for future learning.
pub struct SearchOrchestrator {
    // -- DB layers --
    vector_db: Option<Arc<LanceDb>>,
    kuzu: Option<Arc<KuzuGraph>>,
    _sqlite: Arc<SqliteDb>,

    // -- LLM --
    llm: Arc<dyn LlmProvider>,

    // -- Sub-components --
    classifier: QueryClassifier,
    strategy_selector: StrategySelector,
    judge: ResultJudge,
    metrics: SearchMetrics,
    optimizer: CostOptimizer,
    freshness: Option<Arc<FreshnessManager>>,
    cache: SearchCache,

    // -- Knowledge Base --
    kb_manager: Option<Arc<KnowledgeManager>>,

    /// Tracks KB entry IDs already assimilated this session to prevent redundant graph writes.
    assimilated_kb_ids: Mutex<HashSet<String>>,

    // -- Config --
    config: SearchConfig,
    workspace_root: PathBuf,
}

impl SearchOrchestrator {
    /// Construct a fully-wired orchestrator.
    pub fn new(
        vector_db: Option<Arc<LanceDb>>,
        kuzu: Option<Arc<KuzuGraph>>,
        sqlite: Arc<SqliteDb>,
        llm: Arc<dyn LlmProvider>,
        config: SearchConfig,
        workspace_root: PathBuf,
    ) -> Self {
        let classifier = QueryClassifier::new(Arc::clone(&llm));
        let strategy_selector = StrategySelector::new(Arc::clone(&sqlite));
        let judge = ResultJudge::new(Arc::clone(&llm), config.min_relevance_score);
        let metrics = SearchMetrics::new(Arc::clone(&sqlite));
        let optimizer = CostOptimizer::with_config(
            config.daily_budget,
            [config.tier1_cost, config.tier2_cost, config.tier3_cost],
            Some(Arc::clone(&sqlite)),
        );
        let cache = SearchCache::new(config.cache_ttl, config.max_cache_entries);

        let freshness = vector_db
            .as_ref()
            .map(|vdb| Arc::new(FreshnessManager::new(Arc::clone(vdb))));

        info!(
            workspace = %workspace_root.display(),
            daily_budget = config.daily_budget,
            cache_ttl_secs = config.cache_ttl.as_secs(),
            "SearchOrchestrator initialized"
        );

        Self {
            vector_db,
            kuzu,
            _sqlite: sqlite,
            llm,
            classifier,
            strategy_selector,
            judge,
            metrics,
            optimizer,
            freshness,
            cache,
            kb_manager: None,
            assimilated_kb_ids: Mutex::new(HashSet::new()),
            config,
            workspace_root,
        }
    }

    /// Attach a KnowledgeManager for automatic KB blending in search.
    pub fn with_kb(mut self, kb_manager: Arc<KnowledgeManager>) -> Self {
        self.kb_manager = Some(kb_manager);
        self
    }

    // -----------------------------------------------------------------------
    // Public API
    // -----------------------------------------------------------------------

    /// Execute a search request through the full orchestration pipeline.
    pub async fn search(&self, request: &SearchRequest) -> Result<SearchResponse> {
        let start = Instant::now();
        info!(query = %request.query, "Search request received");

        // 0. Check for UTC date rollover (resets daily spend if needed).
        self.optimizer.check_date_rollover().await;

        // 1. Check cache.
        if let Some(cached) = self.cache.get(request) {
            info!(query = %request.query, "Returning cached result");
            return Ok(SearchResponse {
                query_id: cached.query_id.clone(),
                source: SearchSource::Cache,
                latency: start.elapsed(),
                ..cached
            });
        }

        // 2. Classify query.
        let query_type = self.classify_query(&request.query).await?;
        debug!(query = %request.query, query_type = ?query_type, "Query classified");

        // 3. Determine strategy (budget-aware, adaptive).
        let strategy = self.determine_strategy(&query_type).await?;
        debug!(query = %request.query, strategy = ?strategy, "Strategy selected");

        // 4. Execute strategy.
        let (results, source, breakdown, cost) =
            self.execute_strategy(request, &query_type, &strategy).await?;

        // 4b. Blend KB results if available and configured.
        let results = self.blend_kb_results(results, &request.query, request.max_results).await;

        // 5. Build response (query_id set after metrics recording).
        let mut response = SearchResponse {
            query_id: String::new(),
            results,
            source,
            latency: start.elapsed(),
            cost,
            tier_breakdown: breakdown,
        };

        // 6. Record metrics and capture the query_id for feedback linkage.
        match self
            .metrics
            .record_search(&request.query, &query_type, &strategy, &response)
            .await
        {
            Ok(id) => response.query_id = id,
            Err(e) => warn!(error = %e, "Failed to record search metrics"),
        }

        // 7. Cache the response (includes query_id).
        self.cache.put(request, response.clone());

        // 8. Record cost.
        self.optimizer.record_cost(cost);

        info!(
            query = %request.query,
            latency_ms = response.latency.as_millis() as u64,
            result_count = response.results.len(),
            source = ?response.source,
            cost,
            "Search complete"
        );

        Ok(response)
    }

    /// Invalidate a cached entry for a specific request.
    pub fn invalidate_cache(&self, request: &SearchRequest) {
        self.cache.invalidate(request);
    }

    /// Clear the entire result cache.
    pub fn clear_cache(&self) {
        self.cache.clear();
    }

    /// Get a reference to the cost optimizer for external budget queries.
    pub fn cost_optimizer(&self) -> &CostOptimizer {
        &self.optimizer
    }

    /// Get a reference to the freshness manager, if one exists.
    pub fn freshness_manager(&self) -> Option<&Arc<FreshnessManager>> {
        self.freshness.as_ref()
    }

    /// Get a reference to the search cache.
    pub fn cache(&self) -> &SearchCache {
        &self.cache
    }

    // -----------------------------------------------------------------------
    // Internal: query classification
    // -----------------------------------------------------------------------

    async fn classify_query(&self, query: &str) -> Result<QueryType> {
        match tokio::time::timeout(
            std::time::Duration::from_secs(30),
            self.classifier.classify(query),
        ).await {
            Ok(result) => result.context("Failed to classify query"),
            Err(_) => {
                warn!("Query classification timed out after 30s, defaulting to Semantic");
                Ok(QueryType::Semantic)
            }
        }
    }

    // -----------------------------------------------------------------------
    // Internal: strategy selection
    // -----------------------------------------------------------------------

    async fn determine_strategy(&self, query_type: &QueryType) -> Result<SearchStrategy> {
        let max_tier = self
            .optimizer
            .recommend_max_tier()
            .unwrap_or(1); // If budget is exhausted, allow only Tier 1 (free).

        self.strategy_selector
            .select_strategy(query_type, max_tier)
            .await
    }

    // -----------------------------------------------------------------------
    // Internal: strategy execution
    // -----------------------------------------------------------------------

    /// Dispatch to the appropriate tier(s) based on the selected strategy.
    /// Returns (results, source, tier_breakdown, cost).
    async fn execute_strategy(
        &self,
        request: &SearchRequest,
        query_type: &QueryType,
        strategy: &SearchStrategy,
    ) -> Result<(Vec<SearchResult>, SearchSource, TierBreakdown, f64)> {
        let mut breakdown = TierBreakdown::default();
        let mut total_cost = 0.0;

        match strategy {
            SearchStrategy::Tier1Only => {
                let (results, cost) = self.tier1_search(request, &mut breakdown).await?;
                Ok((results, SearchSource::Tier1Vector, breakdown, cost))
            }

            SearchStrategy::Tier1ThenTier2 => {
                let (results, cost) = self.tier1_search(request, &mut breakdown).await?;
                total_cost += cost;

                if self.is_sufficient(&request.query, &results).await? {
                    return Ok((results, SearchSource::Tier1Vector, breakdown, total_cost));
                }

                let (results, cost) = self.tier2_search(request, &mut breakdown).await?;
                total_cost += cost;
                Ok((results, SearchSource::Tier2Hybrid, breakdown, total_cost))
            }

            SearchStrategy::Tier1ThenTier2ThenTier3 => {
                // Tier 1
                let (results, cost) = self.tier1_search(request, &mut breakdown).await?;
                total_cost += cost;

                if self.is_sufficient(&request.query, &results).await? {
                    return Ok((results, SearchSource::Tier1Vector, breakdown, total_cost));
                }

                // Tier 2
                let (results, cost) = self.tier2_search(request, &mut breakdown).await?;
                total_cost += cost;

                if self.is_sufficient(&request.query, &results).await? {
                    return Ok((results, SearchSource::Tier2Hybrid, breakdown, total_cost));
                }

                // Tier 3
                let (results, cost) = self.tier3_search(request, &mut breakdown).await?;
                total_cost += cost;
                Ok((
                    results,
                    SearchSource::Tier3DeepAgentic,
                    breakdown,
                    total_cost,
                ))
            }

            SearchStrategy::Tier2Only => {
                let (results, cost) = self.tier2_search(request, &mut breakdown).await?;
                Ok((results, SearchSource::Tier2Hybrid, breakdown, cost))
            }

            SearchStrategy::Tier3Only => {
                let (results, cost) = self.tier3_search(request, &mut breakdown).await?;
                Ok((results, SearchSource::Tier3DeepAgentic, breakdown, cost))
            }

            SearchStrategy::GraphQuery => {
                let (results, cost) =
                    self.graph_search(request, query_type, &mut breakdown).await?;
                // If graph returned nothing, fall back to Tier 2.
                if results.is_empty() {
                    debug!("Graph search returned no results, falling back to Tier2");
                    let (results, tier2_cost) =
                        self.tier2_search(request, &mut breakdown).await?;
                    Ok((
                        results,
                        SearchSource::Tier2Hybrid,
                        breakdown,
                        cost + tier2_cost,
                    ))
                } else {
                    Ok((results, SearchSource::GraphQuery, breakdown, cost))
                }
            }
        }
    }

    // -----------------------------------------------------------------------
    // Internal: tier implementations
    // -----------------------------------------------------------------------

    /// Tier 1: fast vector similarity search (no LLM cost).
    /// If a FreshnessManager is available, stale results are filtered out.
    /// When more than half the results are stale, the tier reports empty so
    /// the orchestrator escalates to a higher tier.
    async fn tier1_search(
        &self,
        request: &SearchRequest,
        breakdown: &mut TierBreakdown,
    ) -> Result<(Vec<SearchResult>, f64)> {
        breakdown.tier1_tried = true;

        let vdb = match &self.vector_db {
            Some(v) => v,
            None => {
                debug!("No vector DB available, Tier 1 skipped");
                return Ok((Vec::new(), 0.0));
            }
        };

        let tier1 = Tier1Search::new(Arc::clone(vdb));

        let results = match tokio::time::timeout(
            self.config.tier1_timeout,
            tier1.search(request),
        )
        .await
        {
            Ok(Ok(r)) => r,
            Ok(Err(e)) => {
                error!(error = %e, "Tier 1 search failed");
                Vec::new()
            }
            Err(_) => {
                warn!(
                    timeout_ms = self.config.tier1_timeout.as_millis() as u64,
                    "Tier 1 search timed out"
                );
                Vec::new()
            }
        };

        // Staleness check: if freshness manager is available, filter stale results.
        let results = if let Some(ref fm) = self.freshness {
            let mut fresh = Vec::new();
            let mut stale_count = 0usize;
            for r in results {
                match tokio::time::timeout(
                    std::time::Duration::from_secs(2),
                    fm.is_stale(&r.file_path),
                ).await {
                    Ok(Ok(true)) => stale_count += 1,
                    Ok(_) => fresh.push(r),
                    Err(_) => {
                        // Timeout: assume fresh rather than blocking
                        fresh.push(r);
                    }
                }
            }
            if stale_count > 0 {
                debug!(stale_count, fresh = fresh.len(), "Staleness filter applied to Tier 1 results");
            }
            fresh
        } else {
            results
        };

        breakdown.tier1_success = !results.is_empty();
        let cost = self.optimizer.estimate_tier_cost(1);
        debug!(count = results.len(), "Tier 1 returned results");
        Ok((results, cost))
    }

    /// Tier 2: hybrid vector + grep search (minimal LLM cost).
    async fn tier2_search(
        &self,
        request: &SearchRequest,
        breakdown: &mut TierBreakdown,
    ) -> Result<(Vec<SearchResult>, f64)> {
        breakdown.tier2_tried = true;

        let vdb = match &self.vector_db {
            Some(v) => v,
            None => {
                debug!("No vector DB available, Tier 2 falls back to grep-only");
                // Without a vector DB, Tier2Search still runs grep.
                // We create a temporary LanceDb for the interface but it will
                // return no vector results. This is acceptable since the grep
                // side will still function.
                return Ok((Vec::new(), 0.0));
            }
        };

        let tier2 = Tier2Search::new(Arc::clone(vdb), self.workspace_root.clone());

        let results = match tokio::time::timeout(
            self.config.tier2_timeout,
            tier2.search(request),
        )
        .await
        {
            Ok(Ok(r)) => r,
            Ok(Err(e)) => {
                error!(error = %e, "Tier 2 search failed");
                Vec::new()
            }
            Err(_) => {
                warn!(
                    timeout_ms = self.config.tier2_timeout.as_millis() as u64,
                    "Tier 2 search timed out"
                );
                Vec::new()
            }
        };

        breakdown.tier2_success = !results.is_empty();
        let cost = self.optimizer.estimate_tier_cost(2);
        debug!(count = results.len(), "Tier 2 returned results");
        Ok((results, cost))
    }

    /// Tier 3: deep agentic search (significant LLM cost).
    async fn tier3_search(
        &self,
        request: &SearchRequest,
        breakdown: &mut TierBreakdown,
    ) -> Result<(Vec<SearchResult>, f64)> {
        breakdown.tier3_tried = true;

        // Budget check before the most expensive tier.
        let estimated_cost = self.optimizer.estimate_tier_cost(3);
        if !self.optimizer.can_afford(estimated_cost) {
            warn!("Cannot afford Tier 3 search; budget exhausted");
            return Ok((Vec::new(), 0.0));
        }

        let tier3 = Tier3Search::new(
            Arc::clone(&self.llm),
            self.vector_db.clone(),
            self.workspace_root.clone(),
        );

        let results = match tokio::time::timeout(
            self.config.tier3_timeout,
            tier3.search(request),
        )
        .await
        {
            Ok(Ok(r)) => r,
            Ok(Err(e)) => {
                error!(error = %e, "Tier 3 search failed");
                Vec::new()
            }
            Err(_) => {
                warn!(
                    timeout_ms = self.config.tier3_timeout.as_millis() as u64,
                    "Tier 3 search timed out"
                );
                Vec::new()
            }
        };

        breakdown.tier3_success = !results.is_empty();
        debug!(count = results.len(), "Tier 3 returned results");
        Ok((results, estimated_cost))
    }

    /// Graph-based search for architecture and dependency queries.
    async fn graph_search(
        &self,
        request: &SearchRequest,
        query_type: &QueryType,
        breakdown: &mut TierBreakdown,
    ) -> Result<(Vec<SearchResult>, f64)> {
        let kuzu = match &self.kuzu {
            Some(k) => k,
            None => {
                debug!("No graph DB available, graph search skipped");
                return Ok((Vec::new(), 0.0));
            }
        };

        // Translate the query into a graph query based on query type.
        let graph_query = match query_type {
            QueryType::Dependency => {
                // Try to extract a file or symbol name from the query.
                format!("MATCH (a)-[:Imports]->(b) WHERE a.path CONTAINS '{}' RETURN a, b",
                    Self::extract_subject(&request.query))
            }
            QueryType::Architecture => {
                "MATCH (a)-[:DEPENDS_ON]->(b) RETURN a, b".to_string()
            }
            _ => {
                "MATCH (a)-[:CONTAINS]->(b) RETURN a, b".to_string()
            }
        };

        debug!(graph_query = %graph_query, "Executing graph query");

        let rows = tokio::time::timeout(
            std::time::Duration::from_secs(30),
            kuzu.execute_query(&graph_query),
        )
        .await
        .map_err(|_| anyhow::anyhow!("Graph query timed out after 30s"))?
        ?;

        let results: Vec<SearchResult> = rows
            .into_iter()
            .filter_map(|row| {
                let file_path = row.get("from_path").or_else(|| row.get("file_path"))?;
                let related = row
                    .get("to_path")
                    .or_else(|| row.get("function_name"))
                    .or_else(|| row.get("type_name"))
                    .cloned()
                    .unwrap_or_default();

                Some(SearchResult {
                    file_path: file_path.clone(),
                    content: format!("Related: {}", related),
                    relevance_score: 0.7,
                    line_number: None,
                })
            })
            .take(request.max_results)
            .collect();

        // Graph searches don't neatly fit into the tier1/2/3 breakdown,
        // but we mark tier1_tried so the breakdown isn't all-false.
        breakdown.tier1_tried = true;
        breakdown.tier1_success = !results.is_empty();

        debug!(count = results.len(), "Graph search returned results");
        Ok((results, 0.0))
    }

    // -----------------------------------------------------------------------
    // Internal: result sufficiency judgment
    // -----------------------------------------------------------------------

    /// Use the ResultJudge to decide whether results are good enough
    /// to return, or whether to escalate to the next tier.
    async fn is_sufficient(&self, query: &str, results: &[SearchResult]) -> Result<bool> {
        if results.is_empty() {
            return Ok(false);
        }

        // Try the LLM judge but don't let a judge failure or timeout block escalation.
        match tokio::time::timeout(
            std::time::Duration::from_secs(15),
            self.judge.is_sufficient(query, results),
        ).await {
            Ok(Ok(sufficient)) => Ok(sufficient),
            Ok(Err(e)) => {
                warn!(error = %e, "Result judge failed, assuming insufficient");
                Ok(false)
            }
            Err(_) => {
                warn!("Result judge timed out after 15s, assuming insufficient");
                Ok(false)
            }
        }
    }

    // -----------------------------------------------------------------------
    // Internal: KB blending
    // -----------------------------------------------------------------------

    /// Blend knowledge base results into the search results if a KB manager
    /// is available and has registered sources. Also triggers async graph
    /// assimilation for new KB entries (fire-and-forget).
    async fn blend_kb_results(
        &self,
        mut results: Vec<SearchResult>,
        query: &str,
        max_results: usize,
    ) -> Vec<SearchResult> {
        let kb = match &self.kb_manager {
            Some(kb) => {
                let has = tokio::time::timeout(
                    std::time::Duration::from_secs(5),
                    kb.has_sources(),
                ).await.unwrap_or(false);
                if has { kb } else { return results; }
            }
            _ => return results,
        };

        // Default to 5 KB results per source (with timeout).
        let kb_results = match tokio::time::timeout(
            std::time::Duration::from_secs(15),
            kb.query_all_kbs(query, 5),
        ).await {
            Ok(r) => r,
            Err(_) => {
                warn!("KB query timed out after 15s during search blending");
                return results;
            }
        };
        if kb_results.is_empty() {
            return results;
        }

        let kb_search_results =
            crate::knowledge::search_adapter::kb_results_to_search_results(&kb_results);

        debug!(
            kb_results = kb_search_results.len(),
            "Blending KB results into search"
        );

        // Async graph assimilation: filter to new entries, then fire-and-forget
        let new_results = self.filter_new_kb_results(&kb_results).await;
        if !new_results.is_empty() {
            let kb_clone = Arc::clone(kb);
            tokio::spawn(async move {
                match kb_clone.assimilate_results(&new_results).await {
                    Ok(sync) => {
                        debug!(
                            nodes = sync.graph_nodes_created,
                            edges = sync.graph_edges_created,
                            "Background KB assimilation complete"
                        );
                    }
                    Err(e) => {
                        warn!(error = %e, "Background KB assimilation failed");
                    }
                }
            });
        }

        results.extend(kb_search_results);
        results.sort_by(|a, b| {
            b.relevance_score
                .partial_cmp(&a.relevance_score)
                .unwrap_or(std::cmp::Ordering::Equal)
        });
        results.truncate(max_results);

        results
    }

    /// Filter KB results to only those not yet assimilated this session.
    async fn filter_new_kb_results(
        &self,
        results: &[crate::knowledge::types::KBQueryResult],
    ) -> Vec<crate::knowledge::types::KBQueryResult> {
        let mut cache = self.assimilated_kb_ids.lock().await;
        let mut new_results = Vec::new();

        for qr in results {
            let mut new_entries = Vec::new();
            for entry in &qr.entries {
                if cache.insert(entry.id.clone()) {
                    new_entries.push(entry.clone());
                }
            }
            if !new_entries.is_empty() {
                new_results.push(crate::knowledge::types::KBQueryResult {
                    entries: new_entries,
                    source: qr.source.clone(),
                    collection: qr.collection.clone(),
                    query: qr.query.clone(),
                    total_results: qr.total_results,
                });
            }
        }

        new_results
    }

    // -----------------------------------------------------------------------
    // Internal: helpers
    // -----------------------------------------------------------------------

    /// Extract a likely subject (file name, symbol, etc.) from a query string.
    fn extract_subject(query: &str) -> String {
        // Heuristic: take the last "word" that looks like a path or identifier.
        query
            .split_whitespace()
            .rev()
            .find(|w| {
                w.contains('.')
                    || w.contains('/')
                    || w.contains('_')
                    || w.contains("::")
                    || w.chars().any(|c| c.is_uppercase())
            })
            .unwrap_or(query)
            .to_string()
    }
}

// ===========================================================================
// Tests
// ===========================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::knowledge::types::{KBQueryResult, KnowledgeEntry};

    struct MockLlm;

    #[async_trait::async_trait]
    impl LlmProvider for MockLlm {
        async fn complete(&self, _prompt: &str) -> anyhow::Result<String> {
            Ok("mock".to_string())
        }
        async fn complete_with_system(
            &self,
            _system: &str,
            _prompt: &str,
        ) -> anyhow::Result<String> {
            Ok("mock".to_string())
        }
    }

    /// Test that the assimilation cache correctly deduplicates KB entries
    /// across multiple calls to filter_new_kb_results.
    #[tokio::test]
    async fn test_assimilation_cache_prevents_duplicates() {
        let tmp_sqlite = tempfile::NamedTempFile::new().unwrap();
        let tmp_kuzu = tempfile::TempDir::new().unwrap();
        let tmp_lance = tempfile::TempDir::new().unwrap();

        let sqlite = Arc::new(SqliteDb::new(tmp_sqlite.path()).unwrap());
        let kuzu = Arc::new(KuzuGraph::new(tmp_kuzu.path()).unwrap());
        let lance = Arc::new(LanceDb::new(tmp_lance.path()).await.unwrap());
        let llm: Arc<dyn LlmProvider> = Arc::new(MockLlm);

        let config = SearchConfig::default();
        let orch = SearchOrchestrator::new(
            Some(Arc::clone(&lance)),
            Some(Arc::clone(&kuzu)),
            Arc::clone(&sqlite),
            llm,
            config,
            PathBuf::from("/tmp/test"),
        );

        let make_entry = |id: &str| KnowledgeEntry {
            id: id.to_string(),
            source: "test".into(),
            collection: "docs".into(),
            title: "Test".into(),
            content: "Content".into(),
            metadata: std::collections::HashMap::new(),
            relevance_score: 0.9,
            retrieved_at: 1707300600,
        };

        let results = vec![KBQueryResult {
            entries: vec![make_entry("kb-1"), make_entry("kb-2")],
            source: "test".into(),
            collection: "docs".into(),
            query: "test".into(),
            total_results: 2,
        }];

        // First call: both entries are new
        let new1 = orch.filter_new_kb_results(&results).await;
        assert_eq!(new1.len(), 1);
        assert_eq!(new1[0].entries.len(), 2);

        // Second call with same entries: all filtered out
        let new2 = orch.filter_new_kb_results(&results).await;
        assert!(new2.is_empty(), "Duplicate entries should be filtered out");

        // Third call with one new + one existing entry
        let results3 = vec![KBQueryResult {
            entries: vec![make_entry("kb-1"), make_entry("kb-3")],
            source: "test".into(),
            collection: "docs".into(),
            query: "test".into(),
            total_results: 2,
        }];

        let new3 = orch.filter_new_kb_results(&results3).await;
        assert_eq!(new3.len(), 1);
        assert_eq!(new3[0].entries.len(), 1);
        assert_eq!(new3[0].entries[0].id, "kb-3");
    }
}
