lean_ctx/tools/

server_metrics.rs

use std::sync::atomic::Ordering;

use super::server::{CepComputedStats, CrpMode, LeanCtxServer, ToolCallRecord};
use super::startup::auto_consolidate_knowledge;
use super::{ctx_compress, ctx_share};

impl LeanCtxServer {
    /// Records a tool call's token savings without timing information.
    pub async fn record_call(
        &self,
        tool: &str,
        original: usize,
        saved: usize,
        mode: Option<String>,
    ) {
        self.record_call_with_timing(tool, original, saved, mode, 0)
            .await;
    }

    /// Records a tool call like `record_call`, but includes an optional file path for observability.
    pub async fn record_call_with_path(
        &self,
        tool: &str,
        original: usize,
        saved: usize,
        mode: Option<String>,
        path: Option<&str>,
    ) {
        self.record_call_with_timing_inner(tool, original, saved, mode, 0, path)
            .await;
    }

    /// Records a tool call's token savings, duration, and emits events and stats.
    pub async fn record_call_with_timing(
        &self,
        tool: &str,
        original: usize,
        saved: usize,
        mode: Option<String>,
        duration_ms: u64,
    ) {
        self.record_call_with_timing_inner(tool, original, saved, mode, duration_ms, None)
            .await;
    }

    async fn record_call_with_timing_inner(
        &self,
        tool: &str,
        original: usize,
        saved: usize,
        mode: Option<String>,
        duration_ms: u64,
        path: Option<&str>,
    ) {
        let ts = chrono::Local::now().format("%Y-%m-%d %H:%M:%S").to_string();
        let mut calls = self.tool_calls.write().await;
        calls.push(ToolCallRecord {
            tool: tool.to_string(),
            original_tokens: original,
            saved_tokens: saved,
            mode: mode.clone(),
            duration_ms,
            timestamp: ts.clone(),
        });

        const MAX_TOOL_CALL_RECORDS: usize = 500;
        if calls.len() > MAX_TOOL_CALL_RECORDS {
            let excess = calls.len() - MAX_TOOL_CALL_RECORDS;
            calls.drain(..excess);
        }

        if duration_ms > 0 {
            Self::append_tool_call_log(tool, duration_ms, original, saved, mode.as_deref(), &ts);
        }

        crate::core::events::emit_tool_call(
            tool,
            original as u64,
            saved as u64,
            mode.clone(),
            duration_ms,
            path.map(ToString::to_string),
        );

        let output_tokens = original.saturating_sub(saved);
        crate::core::stats::record(tool, original, output_tokens);

        let mut session = self.session.write().await;
        session.record_tool_call(saved as u64, original as u64);
        if tool == "ctx_shell" {
            session.record_command();
        }
        let pending_save = if session.should_save() {
            session.prepare_save().ok()
        } else {
            None
        };
        drop(calls);
        drop(session);

        if let Some(prepared) = pending_save {
            tokio::task::spawn_blocking(move || {
                let _ = prepared.write_to_disk();
            });
        }

        self.write_mcp_live_stats().await;
    }

    /// Increments the call counter and returns true if a checkpoint is due.
    pub fn increment_and_check(&self) -> bool {
        let count = self.call_count.fetch_add(1, Ordering::Relaxed) + 1;
        let interval = Self::checkpoint_interval_effective();
        interval > 0 && count.is_multiple_of(interval)
    }

    /// Generates a compressed context checkpoint with session state and multi-agent sync.
    pub async fn auto_checkpoint(&self) -> Option<String> {
        let cache = self.cache.read().await;
        if cache.get_all_entries().is_empty() {
            return None;
        }
        let complexity = crate::core::adaptive::classify_from_context(&cache);
        let checkpoint = ctx_compress::handle(&cache, false, CrpMode::effective());
        drop(cache);

        let mut session = self.session.write().await;
        let _ = session.save();
        let session_summary = session.format_compact();
        let has_insights = !session.findings.is_empty() || !session.decisions.is_empty();
        let project_root = session.project_root.clone();
        drop(session);

        if has_insights {
            if let Some(ref root) = project_root {
                let root = root.clone();
                std::thread::spawn(move || {
                    auto_consolidate_knowledge(&root);
                });
            }
        }

        let multi_agent_block = self
            .auto_multi_agent_checkpoint(project_root.as_ref())
            .await;

        self.record_call("ctx_compress", 0, 0, Some("auto".to_string()))
            .await;

        self.record_cep_snapshot().await;

        if !crate::core::protocol::meta_visible() {
            return None;
        }

        // Check if agent has documented anything recently
        let doc_reminder = {
            let session = self.session.read().await;
            let calls_since_last_doc = Self::calls_since_last_documentation(&session);
            if calls_since_last_doc >= 30 {
                "\n[CHECKPOINT: please document current progress via ctx_session(action=\"task\") or ctx_knowledge(action=\"remember\")]"
            } else {
                ""
            }
        };

        Some(format!(
            "{checkpoint}\n\n--- SESSION STATE ---\n{session_summary}\n\n{}{multi_agent_block}{doc_reminder}",
            complexity.instruction_suffix()
        ))
    }

    async fn auto_multi_agent_checkpoint(&self, project_root: Option<&String>) -> String {
        let Some(root) = project_root else {
            return String::new();
        };

        let registry = crate::core::agents::AgentRegistry::load_or_create();
        let active = registry.list_active(Some(root));
        if active.len() <= 1 {
            return String::new();
        }

        let agent_id = self.agent_id.read().await;
        let my_id = match agent_id.as_deref() {
            Some(id) => id.to_string(),
            None => return String::new(),
        };
        drop(agent_id);

        let cache = self.cache.read().await;
        let entries = cache.get_all_entries();
        if !entries.is_empty() {
            let mut by_access: Vec<_> = entries.iter().collect();
            by_access.sort_by_key(|x| std::cmp::Reverse(x.1.read_count));
            let top_paths: Vec<&str> = by_access
                .iter()
                .take(5)
                .map(|(key, _)| key.as_str())
                .collect();
            let paths_csv = top_paths.join(",");

            let _ = ctx_share::handle(
                "push",
                Some(&my_id),
                None,
                Some(&paths_csv),
                None,
                &cache,
                root,
            );
        }
        drop(cache);

        let pending_count = registry
            .scratchpad
            .iter()
            .filter(|e| !e.read_by.contains(&my_id) && e.from_agent != my_id)
            .count();

        let shared_dir = crate::core::data_dir::lean_ctx_data_dir()
            .unwrap_or_default()
            .join("agents")
            .join("shared");
        let shared_count = if shared_dir.exists() {
            std::fs::read_dir(&shared_dir).map_or(0, std::iter::Iterator::count)
        } else {
            0
        };

        let agent_names: Vec<String> = active
            .iter()
            .map(|a| {
                let role = a.role.as_deref().unwrap_or(&a.agent_type);
                format!("{role}({})", &a.agent_id[..8.min(a.agent_id.len())])
            })
            .collect();

        format!(
            "\n\n--- MULTI-AGENT SYNC ---\nAgents: {} | Pending msgs: {} | Shared contexts: {}\nAuto-shared top-5 cached files.\n--- END SYNC ---",
            agent_names.join(", "),
            pending_count,
            shared_count,
        )
    }

    /// Appends a tool call entry to the rotating `tool-calls.log` file.
    pub fn append_tool_call_log(
        tool: &str,
        duration_ms: u64,
        original: usize,
        saved: usize,
        mode: Option<&str>,
        timestamp: &str,
    ) {
        const MAX_LOG_LINES: usize = 50;
        if let Ok(dir) = crate::core::data_dir::lean_ctx_data_dir() {
            let log_path = dir.join("tool-calls.log");
            let mode_str = mode.unwrap_or("-");
            let slow = if duration_ms > 5000 { " **SLOW**" } else { "" };
            let line = format!(
                "{timestamp}\t{tool}\t{duration_ms}ms\torig={original}\tsaved={saved}\tmode={mode_str}{slow}\n"
            );

            let mut lines: Vec<String> = std::fs::read_to_string(&log_path)
                .unwrap_or_default()
                .lines()
                .map(std::string::ToString::to_string)
                .collect();

            lines.push(line.trim_end().to_string());
            if lines.len() > MAX_LOG_LINES {
                lines.drain(0..lines.len() - MAX_LOG_LINES);
            }

            let _ = std::fs::write(&log_path, lines.join("\n") + "\n");
        }
    }

    fn compute_cep_stats(
        calls: &[ToolCallRecord],
        stats: &crate::core::cache::CacheStats,
        complexity: &crate::core::adaptive::TaskComplexity,
    ) -> CepComputedStats {
        let total_original: u64 = calls.iter().map(|c| c.original_tokens as u64).sum();
        let total_saved: u64 = calls.iter().map(|c| c.saved_tokens as u64).sum();
        let total_compressed = total_original.saturating_sub(total_saved);
        let compression_rate = if total_original > 0 {
            total_saved as f64 / total_original as f64
        } else {
            0.0
        };

        let modes_used: std::collections::HashSet<&str> =
            calls.iter().filter_map(|c| c.mode.as_deref()).collect();
        let mode_diversity = (modes_used.len() as f64 / 10.0).min(1.0);
        let cache_util = stats.hit_rate() / 100.0;
        let cep_score = cache_util * 0.3 + mode_diversity * 0.2 + compression_rate * 0.5;

        let mut mode_counts: std::collections::HashMap<String, u64> =
            std::collections::HashMap::new();
        for call in calls {
            if let Some(ref mode) = call.mode {
                *mode_counts.entry(mode.clone()).or_insert(0) += 1;
            }
        }

        CepComputedStats {
            cep_score: (cep_score * 100.0).round() as u32,
            cache_util: (cache_util * 100.0).round() as u32,
            mode_diversity: (mode_diversity * 100.0).round() as u32,
            compression_rate: (compression_rate * 100.0).round() as u32,
            total_original,
            total_compressed,
            total_saved,
            mode_counts,
            complexity: format!("{complexity:?}"),
            cache_hits: stats.cache_hits,
            total_reads: stats.total_reads,
            tool_call_count: calls.len() as u64,
        }
    }

    async fn write_mcp_live_stats(&self) {
        let count = self.call_count.load(Ordering::Relaxed);
        if count > 1 && !count.is_multiple_of(5) {
            return;
        }

        let cache = self.cache.read().await;
        let calls = self.tool_calls.read().await;
        let stats = cache.get_stats();
        let complexity = crate::core::adaptive::classify_from_context(&cache);

        let cs = Self::compute_cep_stats(&calls, stats, &complexity);
        let started_at = calls
            .first()
            .map(|c| c.timestamp.clone())
            .unwrap_or_default();

        drop(cache);
        drop(calls);
        let live = serde_json::json!({
            "cep_score": cs.cep_score,
            "cache_utilization": cs.cache_util,
            "mode_diversity": cs.mode_diversity,
            "compression_rate": cs.compression_rate,
            "task_complexity": cs.complexity,
            "files_cached": cs.total_reads,
            "total_reads": cs.total_reads,
            "cache_hits": cs.cache_hits,
            "tokens_saved": cs.total_saved,
            "tokens_original": cs.total_original,
            "tool_calls": cs.tool_call_count,
            "started_at": started_at,
            "updated_at": chrono::Local::now().to_rfc3339(),
        });

        if let Ok(dir) = crate::core::data_dir::lean_ctx_data_dir() {
            let _ = std::fs::write(dir.join("mcp-live.json"), live.to_string());
        }
    }

    /// Persists a CEP (Context Efficiency Protocol) score snapshot for analytics.
    pub async fn record_cep_snapshot(&self) {
        let cache = self.cache.read().await;
        let calls = self.tool_calls.read().await;
        let stats = cache.get_stats();
        let complexity = crate::core::adaptive::classify_from_context(&cache);

        let cs = Self::compute_cep_stats(&calls, stats, &complexity);

        drop(cache);
        drop(calls);

        crate::core::stats::record_cep_session(
            cs.cep_score,
            cs.cache_hits,
            cs.total_reads,
            cs.total_original,
            cs.total_compressed,
            &cs.mode_counts,
            cs.tool_call_count,
            &cs.complexity,
        );
    }

    /// Counts tool calls since the last documentation action (ctx_knowledge remember, ctx_session task).
    fn calls_since_last_documentation(session: &crate::core::session::SessionState) -> u64 {
        let total = session.stats.total_tool_calls;
        // Use findings and decisions timestamps as proxy for documentation activity
        let last_finding_ts = session.findings.last().map(|f| f.timestamp);
        let last_decision_ts = session.decisions.last().map(|d| d.timestamp);

        // If there are recent decisions or explicit task updates, consider it documented
        if session.task.is_some() {
            // Task was set at some point — check how many calls ago
            let task_set_at = session
                .progress
                .last()
                .map(|p| p.timestamp)
                .or(last_decision_ts)
                .or(last_finding_ts);

            if let Some(ts) = task_set_at {
                let age = chrono::Utc::now() - ts;
                // If last doc action was less than 5 minutes ago, not stale
                if age.num_minutes() < 5 {
                    return 0;
                }
            }
        }

        // Conservative: if no decisions at all, report total calls
        if session.decisions.is_empty() && session.progress.is_empty() {
            return u64::from(total);
        }

        // Estimate calls since last documentation based on time
        u64::from(total.min(30))
    }
}