aptu_coder/
metrics.rs

1// SPDX-FileCopyrightText: 2026 aptu-coder contributors
2// SPDX-License-Identifier: Apache-2.0
3//! Metrics collection and daily-rotating JSONL emission.
4//!
5//! Provides a channel-based pipeline: callers emit [`MetricEvent`] values via [`MetricsSender`],
6//! and [`MetricsWriter`] drains the channel and appends events to a daily-rotated JSONL file
7//! under the XDG data directory (`~/.local/share/aptu-coder/metrics-YYYY-MM-DD.jsonl`).
8//! Files older than 30 days are deleted on startup.
9
10use aptu_coder_core::lang::language_for_extension;
11use serde::{Deserialize, Serialize};
12use std::path::{Path, PathBuf};
13use std::time::{SystemTime, UNIX_EPOCH};
14use tokio::io::AsyncWriteExt;
15use tokio::sync::mpsc;
16
17/// A single metric event emitted by a tool invocation.
18#[derive(Debug, Clone, Default, Serialize, Deserialize)]
19#[serde(default)]
20pub struct MetricEvent {
21    pub ts: u64,
22    pub tool: &'static str,
23    pub duration_ms: u64,
24    pub output_chars: usize,
25    pub param_path_depth: usize,
26    pub max_depth: Option<u32>,
27    pub result: &'static str,
28    pub error_type: Option<String>,
29    #[serde(default, skip_serializing_if = "Option::is_none")]
30    pub error_subtype: Option<String>,
31    #[serde(default)]
32    pub session_id: Option<String>,
33    #[serde(default)]
34    pub seq: Option<u32>,
35    #[serde(default)]
36    #[serde(skip_serializing_if = "Option::is_none")]
37    pub cache_hit: Option<bool>,
38    #[serde(default, skip_serializing_if = "Option::is_none")]
39    pub cache_tier: Option<&'static str>,
40    /// Set to Some(true) when an L2 disk cache write fails (dir, tempfile, write, or rename).
41    /// Drives the cache_write_failures_total OTEL counter.
42    #[serde(default, skip_serializing_if = "Option::is_none")]
43    pub cache_write_failure: Option<bool>,
44    #[serde(default, skip_serializing_if = "Option::is_none")]
45    pub exit_code: Option<i32>,
46    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
47    pub timed_out: bool,
48    #[serde(default, skip_serializing_if = "Option::is_none")]
49    pub output_truncated: Option<bool>,
50    /// True when `output_chars > 30_000`; fires for the top ~0.33% of exec_command calls
51    /// (p99.7 of 27,981 observed calls). Early-warning signal for responses approaching
52    /// the per-stream byte-cap threshold.
53    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
54    pub chars_threshold_breach: bool,
55    /// File extension of the analyzed path, lowercased. `Some("rs")` for known extensions,
56    /// `Some("other")` for unrecognized extensions, `None` when the path has no extension.
57    /// Only populated for `analyze_file` and `analyze_module`.
58    #[serde(default, skip_serializing_if = "Option::is_none")]
59    pub file_ext: Option<&'static str>,
60    /// Name of the filter rule that matched and transformed exec_command output.
61    /// `None` when no filter fired or for non-`exec_command` tools.
62    #[serde(default, skip_serializing_if = "Option::is_none")]
63    pub filter_applied: Option<String>,
64    /// Human-readable programming language name derived from the file extension
65    /// (e.g., `Some("Rust")` for `.rs` files). `None` when the path has no extension
66    /// or the extension is not recognized. Only populated for `analyze_file` and
67    /// `analyze_module`.
68    #[serde(default, skip_serializing_if = "Option::is_none")]
69    pub language: Option<String>,
70}
71
72/// Sender half of the metrics channel; cloned and passed to tools for event emission.
73#[derive(Clone)]
74pub struct MetricsSender(pub tokio::sync::mpsc::UnboundedSender<MetricEvent>);
75
76impl MetricsSender {
77    pub fn send(&self, event: MetricEvent) {
78        let _ = self.0.send(event);
79    }
80}
81
82/// Receiver half of the metrics channel; drains events and writes them to daily-rotated JSONL files.
83pub struct MetricsWriter {
84    rx: tokio::sync::mpsc::UnboundedReceiver<MetricEvent>,
85    base_dir: PathBuf,
86    dir_created: bool,
87}
88
89/// Accumulated metrics for a single tool.
90#[derive(Default, Debug)]
91struct ToolMetrics {
92    count: u64,
93    duration_ms: u64,
94    output_chars: u64,
95}
96
97impl MetricsWriter {
98    pub fn new(
99        rx: tokio::sync::mpsc::UnboundedReceiver<MetricEvent>,
100        base_dir: Option<PathBuf>,
101    ) -> Self {
102        let dir = base_dir.unwrap_or_else(xdg_metrics_dir);
103        Self {
104            rx,
105            base_dir: dir,
106            dir_created: false,
107        }
108    }
109
110    /// Accumulate a metric event into tool_counts and export_session_id.
111    fn accumulate_event(
112        tool_counts: &mut std::collections::HashMap<&'static str, ToolMetrics>,
113        export_session_id: &mut Option<String>,
114        event: &MetricEvent,
115    ) {
116        let entry = tool_counts.entry(event.tool).or_default();
117        entry.count += 1;
118        entry.duration_ms += event.duration_ms;
119        // output_chars is capped at 50 KB per stream (stdout + stderr each), so usize -> u64 is lossless.
120        entry.output_chars += event.output_chars as u64;
121        if export_session_id.is_none() {
122            *export_session_id = event.session_id.clone();
123        }
124    }
125
126    /// Write accumulated batch to file. Fire-and-forget semantics: errors are logged but not propagated.
127    async fn flush_batch(file: &mut tokio::fs::File, batch: Vec<MetricEvent>) {
128        for event in batch {
129            // Record to OTel metrics if available
130            record_otel_metrics(&event);
131
132            // Always write to JSONL as fallback
133            if let Ok(mut json) = serde_json::to_string(&event) {
134                json.push('\n');
135                let _ = file.write_all(json.as_bytes()).await;
136            }
137        }
138        let _ = file.flush().await;
139    }
140
141    /// Check for date transition and rotate metrics file if needed.
142    /// Returns the current file path and updates state if rotation occurred.
143    fn rotate_metrics_file(
144        base_dir: &std::path::Path,
145        current_date: &mut String,
146        current_file: &mut Option<PathBuf>,
147        dir_created: &mut bool,
148    ) -> PathBuf {
149        let new_date = current_date_str();
150        if new_date != *current_date {
151            *current_date = new_date;
152            *current_file = None;
153            *dir_created = false;
154        }
155
156        if current_file.is_none() {
157            *current_file = Some(base_dir.join(format!("metrics-{}.jsonl", current_date)));
158        }
159
160        current_file
161            .as_ref()
162            .expect("current_file is guaranteed Some after check above")
163            .clone()
164    }
165
166    /// Receive and accumulate a batch of events from the channel.
167    async fn receive_batch(
168        rx: &mut tokio::sync::mpsc::UnboundedReceiver<MetricEvent>,
169        tool_counts: &mut std::collections::HashMap<&'static str, ToolMetrics>,
170        export_session_id: &mut Option<String>,
171    ) -> Option<Vec<MetricEvent>> {
172        let mut batch = Vec::new();
173        if let Some(event) = rx.recv().await {
174            Self::accumulate_event(tool_counts, export_session_id, &event);
175            batch.push(event);
176            for _ in 0..99 {
177                match rx.try_recv() {
178                    Ok(e) => {
179                        Self::accumulate_event(tool_counts, export_session_id, &e);
180                        batch.push(e);
181                    }
182                    Err(
183                        mpsc::error::TryRecvError::Empty | mpsc::error::TryRecvError::Disconnected,
184                    ) => break,
185                }
186            }
187            Some(batch)
188        } else {
189            None
190        }
191    }
192
193    /// Ensure metrics directory exists for the given path.
194    async fn ensure_metrics_dir(path: &std::path::Path, dir_created: &mut bool) {
195        if !*dir_created
196            && let Some(parent) = path.parent()
197            && !parent.as_os_str().is_empty()
198        {
199            match tokio::fs::create_dir_all(parent).await {
200                Ok(()) => {
201                    *dir_created = true;
202                }
203                Err(e) => {
204                    tracing::warn!(
205                        error = %e,
206                        path = %parent.display(),
207                        "metrics: failed to create directory; will retry next batch"
208                    );
209                }
210            }
211        }
212    }
213
214    pub async fn run(mut self) {
215        cleanup_old_files(&self.base_dir).await;
216        let mut current_date = current_date_str();
217        let mut current_file: Option<PathBuf> = None;
218
219        // Accumulate per-tool metrics for export on shutdown (issue #773)
220        let mut tool_counts: std::collections::HashMap<&'static str, ToolMetrics> =
221            std::collections::HashMap::new();
222        let mut export_session_id: Option<String> = None;
223
224        loop {
225            let Some(batch) =
226                Self::receive_batch(&mut self.rx, &mut tool_counts, &mut export_session_id).await
227            else {
228                break;
229            };
230
231            let path = Self::rotate_metrics_file(
232                &self.base_dir,
233                &mut current_date,
234                &mut current_file,
235                &mut self.dir_created,
236            );
237
238            Self::ensure_metrics_dir(&path, &mut self.dir_created).await;
239
240            // Open file once per batch
241            let file = tokio::fs::OpenOptions::new()
242                .create(true)
243                .append(true)
244                .open(&path)
245                .await;
246
247            if let Ok(mut file) = file {
248                Self::flush_batch(&mut file, batch).await;
249            }
250        }
251
252        // Export metrics summary on shutdown (issue #773)
253        if let Ok(export_path) = std::env::var("APTU_CODER_METRICS_EXPORT_FILE") {
254            if !std::path::Path::new(&export_path).is_absolute() {
255                tracing::warn!(
256                    path = %export_path,
257                    "metrics: APTU_CODER_METRICS_EXPORT_FILE must be an absolute path; skipping export"
258                );
259            } else {
260                let mut tool_calls = Vec::new();
261                let mut total_duration_ms = 0u64;
262                let mut total_output_chars_sum = 0u64;
263                // Sort by tool name for deterministic JSON output
264                let mut sorted_tools: Vec<_> = tool_counts.iter().collect();
265                sorted_tools.sort_by_key(|&(name, _)| name);
266                for (tool_name, metrics) in sorted_tools {
267                    tool_calls.push(serde_json::json!({
268                        "tool": tool_name,
269                        "call_count": metrics.count,
270                        "total_duration_ms": metrics.duration_ms,
271                        "total_output_chars": metrics.output_chars
272                    }));
273                    total_duration_ms += metrics.duration_ms;
274                    total_output_chars_sum += metrics.output_chars;
275                }
276                let summary = serde_json::json!({
277                    "session_id": export_session_id.unwrap_or_default(),
278                    "tool_calls": tool_calls,
279                    "total_duration_ms": total_duration_ms,
280                    "total_output_chars": total_output_chars_sum
281                });
282                if let Ok(json_str) = serde_json::to_string(&summary)
283                    && let Err(e) = tokio::fs::write(&export_path, json_str).await
284                {
285                    tracing::warn!(
286                        error = %e,
287                        path = %export_path,
288                        "metrics: failed to write export file"
289                    );
290                }
291            }
292        }
293    }
294}
295
296/// Returns the current UNIX timestamp in milliseconds.
297#[must_use]
298pub fn unix_ms() -> u64 {
299    SystemTime::now()
300        .duration_since(UNIX_EPOCH)
301        .unwrap_or_default()
302        .as_millis()
303        .try_into()
304        .unwrap_or(u64::MAX)
305}
306
307/// Counts the number of path segments in a file path.
308#[must_use]
309pub fn path_component_count(path: &str) -> usize {
310    Path::new(path).components().count()
311}
312
313/// Returns the lowercased file extension of `path` as a `&'static str`.
314///
315/// - Returns `Some(ext)` for extensions recognized by [`language_for_extension`] (e.g. `"rs"`).
316/// - Returns `Some("other")` for paths that have an extension but it is not in the known set.
317/// - Returns `None` for paths with no extension or an empty extension.
318#[must_use]
319pub fn path_file_ext(path: &str) -> Option<&'static str> {
320    let ext_os = Path::new(path).extension()?;
321    let ext_str = ext_os.to_str()?;
322    if ext_str.is_empty() {
323        return None;
324    }
325    // language_for_extension does case-insensitive lookup; if found, return the
326    // canonical (lowercased) extension key from EXTENSION_MAP via supported_extensions().
327    if language_for_extension(ext_str).is_some() {
328        aptu_coder_core::lang::supported_extensions()
329            .into_iter()
330            .find(|e| e.eq_ignore_ascii_case(ext_str))
331    } else {
332        Some("other")
333    }
334}
335
336/// Derive a human-readable language name from a file path.
337///
338/// - Returns `Some("Rust")` for paths with a recognized extension.
339/// - Returns `None` for paths with no extension or an unrecognized extension.
340#[must_use]
341pub fn path_language(path: &str) -> Option<String> {
342    let ext_os = Path::new(path).extension()?;
343    let ext_str = ext_os.to_str()?;
344    if ext_str.is_empty() {
345        return None;
346    }
347    language_for_extension(ext_str).map(std::borrow::ToOwned::to_owned)
348}
349
350fn xdg_metrics_dir() -> PathBuf {
351    if let Ok(xdg_data_home) = std::env::var("XDG_DATA_HOME")
352        && !xdg_data_home.is_empty()
353    {
354        return PathBuf::from(xdg_data_home).join("aptu-coder");
355    }
356
357    if let Ok(home) = std::env::var("HOME") {
358        PathBuf::from(home)
359            .join(".local")
360            .join("share")
361            .join("aptu-coder")
362    } else {
363        PathBuf::from(".")
364    }
365}
366
367async fn cleanup_old_files(base_dir: &Path) {
368    let now_days = u32::try_from(unix_ms() / 86_400_000).unwrap_or(u32::MAX);
369
370    let Ok(mut entries) = tokio::fs::read_dir(base_dir).await else {
371        return;
372    };
373
374    loop {
375        match entries.next_entry().await {
376            Ok(Some(entry)) => {
377                let path = entry.path();
378                let file_name = match path.file_name() {
379                    Some(n) => n.to_string_lossy().into_owned(),
380                    None => continue,
381                };
382
383                // Expected format: metrics-YYYY-MM-DD.jsonl
384                if !file_name.starts_with("metrics-")
385                    || std::path::Path::new(&*file_name)
386                        .extension()
387                        .is_none_or(|e| !e.eq_ignore_ascii_case("jsonl"))
388                {
389                    continue;
390                }
391                let date_part = &file_name[8..file_name.len() - 6];
392                if date_part.len() != 10
393                    || date_part.as_bytes().get(4) != Some(&b'-')
394                    || date_part.as_bytes().get(7) != Some(&b'-')
395                {
396                    continue;
397                }
398                let Ok(year) = date_part[0..4].parse::<u32>() else {
399                    continue;
400                };
401                let Ok(month) = date_part[5..7].parse::<u32>() else {
402                    continue;
403                };
404                let Ok(day) = date_part[8..10].parse::<u32>() else {
405                    continue;
406                };
407                if month == 0 || month > 12 || day == 0 || day > 31 {
408                    continue;
409                }
410
411                let file_days = date_to_days_since_epoch(year, month, day);
412                if now_days > file_days && (now_days - file_days) > 30 {
413                    let _ = tokio::fs::remove_file(&path).await;
414                }
415            }
416            Ok(None) => break,
417            Err(e) => {
418                tracing::warn!("error reading metrics directory entry: {e}");
419            }
420        }
421    }
422}
423
424fn date_to_days_since_epoch(y: u32, m: u32, d: u32) -> u32 {
425    // Shift year so March is month 0
426    let (y, m) = if m <= 2 { (y - 1, m + 9) } else { (y, m - 3) };
427    let era = y / 400;
428    let yoe = y - era * 400;
429    let doy = (153 * m + 2) / 5 + d - 1;
430    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
431    // Compute the proleptic Gregorian day number, then subtract the Unix epoch offset.
432    // The subtraction must wrap the full expression; applying .saturating_sub to `doe`
433    // alone would underflow for recent dates where doe < 719_468.
434    (era * 146_097 + doe).saturating_sub(719_468)
435}
436
437/// Returns the current UTC date as a string in YYYY-MM-DD format.
438#[must_use]
439pub fn current_date_str() -> String {
440    let days = u32::try_from(unix_ms() / 86_400_000).unwrap_or(u32::MAX);
441    let z = days + 719_468;
442    let era = z / 146_097;
443    let doe = z - era * 146_097;
444    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146_096) / 365;
445    let y = yoe + era * 400;
446    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
447    let mp = (5 * doy + 2) / 153;
448    let d = doy - (153 * mp + 2) / 5 + 1;
449    let m = if mp < 10 { mp + 3 } else { mp - 9 };
450    let y = if m <= 2 { y + 1 } else { y };
451    format!("{y:04}-{m:02}-{d:02}")
452}
453
454/// Migrate legacy metrics directory from `code-analyze-mcp` to `aptu-coder`.
455///
456/// - If the old directory exists and the new one does not, rename it and log info.
457/// - If both exist, log a warning and do nothing.
458/// - If neither exists, do nothing.
459///
460/// Returns `Ok(())` on success, propagating any I/O errors.
461pub fn migrate_legacy_metrics_dir() -> std::io::Result<()> {
462    let home =
463        std::env::var("HOME").map_err(|e| std::io::Error::new(std::io::ErrorKind::NotFound, e))?;
464    migrate_legacy_metrics_dir_impl(&home)
465}
466
467#[allow(dead_code)]
468fn migrate_legacy_metrics_dir_impl(home: &str) -> std::io::Result<()> {
469    let old_dir = PathBuf::from(home).join(".local/share/code-analyze-mcp");
470    let new_dir = PathBuf::from(home).join(".local/share/aptu-coder");
471
472    let old_exists = old_dir.is_dir();
473    let new_exists = new_dir.is_dir();
474
475    if old_exists && !new_exists {
476        std::fs::rename(&old_dir, &new_dir)?;
477        tracing::info!(
478            "Migrated legacy metrics directory from {:?} to {:?}",
479            old_dir,
480            new_dir
481        );
482    } else if old_exists && new_exists {
483        tracing::warn!("Both legacy and new metrics directories exist; not migrating");
484    }
485    // If old does not exist, nothing to do.
486    Ok(())
487}
488
489#[cfg(test)]
490mod tests {
491    use super::*;
492    use std::fs;
493    use std::sync::{Mutex, OnceLock};
494    use tempfile::TempDir;
495
496    /// Serializes tests that mutate `APTU_CODER_METRICS_EXPORT_FILE` to prevent parallel
497    /// pollution. Recovers from poison caused by panicking tests.
498    fn metrics_export_lock() -> std::sync::MutexGuard<'static, ()> {
499        static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
500        let m = LOCK.get_or_init(|| Mutex::new(()));
501        m.lock().unwrap_or_else(|e| e.into_inner())
502    }
503
504    #[test]
505    fn test_migrate_legacy_only_old_exists() {
506        // Arrange
507        let tmp_home = TempDir::new().unwrap();
508        let home_str = tmp_home.path().to_str().unwrap();
509        let old_path = tmp_home.path().join(".local/share/code-analyze-mcp");
510        let new_path = tmp_home.path().join(".local/share/aptu-coder");
511        fs::create_dir_all(&old_path).unwrap();
512        assert!(!new_path.exists());
513
514        // Act
515        let result = migrate_legacy_metrics_dir_impl(home_str);
516
517        // Assert
518        assert!(result.is_ok());
519        assert!(!old_path.exists(), "old dir should be moved");
520        assert!(new_path.is_dir(), "new dir should exist");
521    }
522
523    #[test]
524    fn test_migrate_legacy_both_exist() {
525        // Arrange
526        let tmp_home = TempDir::new().unwrap();
527        let home_str = tmp_home.path().to_str().unwrap();
528        let old_path = tmp_home.path().join(".local/share/code-analyze-mcp");
529        let new_path = tmp_home.path().join(".local/share/aptu-coder");
530        fs::create_dir_all(&old_path).unwrap();
531        fs::create_dir_all(&new_path).unwrap();
532
533        // Act
534        let result = migrate_legacy_metrics_dir_impl(home_str);
535
536        // Assert
537        assert!(result.is_ok());
538        assert!(old_path.is_dir(), "old dir should remain");
539        assert!(new_path.is_dir(), "new dir should remain");
540    }
541
542    #[test]
543    fn test_migrate_legacy_neither_exists() {
544        // Arrange
545        let tmp_home = TempDir::new().unwrap();
546        let home_str = tmp_home.path().to_str().unwrap();
547        let old_path = tmp_home.path().join(".local/share/code-analyze-mcp");
548        let new_path = tmp_home.path().join(".local/share/aptu-coder");
549
550        // Act
551        let result = migrate_legacy_metrics_dir_impl(home_str);
552
553        // Assert
554        assert!(result.is_ok());
555        assert!(!old_path.exists(), "old dir should not exist");
556        assert!(!new_path.exists(), "new dir should not exist");
557    }
558
559    #[test]
560    fn test_date_to_days_since_epoch_known_dates() {
561        assert_eq!(date_to_days_since_epoch(1970, 1, 1), 0);
562        assert_eq!(date_to_days_since_epoch(2020, 1, 1), 18_262);
563        assert_eq!(date_to_days_since_epoch(2000, 2, 29), 11_016);
564    }
565
566    #[test]
567    fn test_current_date_str_format() {
568        let s = current_date_str();
569        assert_eq!(s.len(), 10);
570        assert_eq!(s.as_bytes()[4], b'-');
571        assert_eq!(s.as_bytes()[7], b'-');
572        let year: u32 = s[0..4].parse().expect("year must be numeric");
573        assert!(year >= 2020 && year <= 2100);
574    }
575
576    #[tokio::test]
577    async fn test_metrics_writer_batching() {
578        let dir = TempDir::new().unwrap();
579        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<MetricEvent>();
580        let writer = MetricsWriter::new(rx, Some(dir.path().to_path_buf()));
581        let make_event = || MetricEvent {
582            ts: unix_ms(),
583            tool: "analyze_directory",
584            duration_ms: 1,
585            output_chars: 10,
586            param_path_depth: 1,
587            max_depth: None,
588            result: "ok",
589            error_type: None,
590            error_subtype: None,
591            session_id: None,
592            seq: None,
593            cache_hit: None,
594            cache_write_failure: None,
595            exit_code: None,
596            timed_out: false,
597            cache_tier: None,
598            output_truncated: None,
599            chars_threshold_breach: false,
600            file_ext: None,
601            filter_applied: None,
602            language: None,
603        };
604        tx.send(make_event()).unwrap();
605        tx.send(make_event()).unwrap();
606        tx.send(make_event()).unwrap();
607        drop(tx);
608        writer.run().await;
609        let entries: Vec<_> = std::fs::read_dir(dir.path())
610            .unwrap()
611            .filter_map(|e| e.ok())
612            .filter(|e| {
613                e.path()
614                    .extension()
615                    .and_then(|x| x.to_str())
616                    .map(|x| x.eq_ignore_ascii_case("jsonl"))
617                    .unwrap_or(false)
618            })
619            .collect();
620        assert_eq!(entries.len(), 1);
621        let content = std::fs::read_to_string(entries[0].path()).unwrap();
622        let lines: Vec<&str> = content.lines().collect();
623        assert_eq!(lines.len(), 3);
624    }
625
626    #[tokio::test]
627    async fn test_cleanup_old_files_deletes_old_keeps_recent() {
628        let dir = TempDir::new().unwrap();
629        let old_file = dir.path().join("metrics-1970-01-01.jsonl");
630        let today = current_date_str();
631        let recent_file = dir.path().join(format!("metrics-{}.jsonl", today));
632        std::fs::write(&old_file, "old\n").unwrap();
633        std::fs::write(&recent_file, "recent\n").unwrap();
634        cleanup_old_files(dir.path()).await;
635        assert!(!old_file.exists());
636        assert!(recent_file.exists());
637    }
638
639    #[test]
640    fn test_metric_event_serialization() {
641        let event = MetricEvent {
642            ts: 1_700_000_000_000,
643            tool: "analyze_directory",
644            duration_ms: 42,
645            output_chars: 100,
646            param_path_depth: 3,
647            max_depth: Some(2),
648            result: "ok",
649            error_type: None,
650            error_subtype: None,
651            session_id: None,
652            seq: None,
653            cache_hit: None,
654            cache_write_failure: None,
655            exit_code: None,
656            timed_out: false,
657            cache_tier: None,
658            output_truncated: None,
659            chars_threshold_breach: false,
660            file_ext: None,
661            filter_applied: None,
662            language: None,
663        };
664        let json = serde_json::to_string(&event).unwrap();
665        assert!(json.contains("analyze_directory"));
666        assert!(json.contains(r#""result":"ok""#));
667        assert!(json.contains(r#""output_chars":100"#));
668        // Verify error_subtype is omitted when None (backward compat)
669        assert!(!json.contains("error_subtype"));
670    }
671
672    #[test]
673    fn test_metric_event_serialization_error() {
674        let event = MetricEvent {
675            ts: 1_700_000_000_000,
676            tool: "analyze_directory",
677            duration_ms: 5,
678            output_chars: 0,
679            param_path_depth: 3,
680            max_depth: Some(3),
681            result: "error",
682            error_type: Some("invalid_params".to_string()),
683            error_subtype: None,
684            session_id: None,
685            seq: None,
686            cache_hit: None,
687            cache_write_failure: None,
688            exit_code: None,
689            timed_out: false,
690            cache_tier: None,
691            output_truncated: None,
692            chars_threshold_breach: false,
693            file_ext: None,
694            filter_applied: None,
695            language: None,
696        };
697        let json = serde_json::to_string(&event).unwrap();
698        assert!(json.contains(r#""result":"error""#));
699        assert!(json.contains(r#""error_type":"invalid_params""#));
700        assert!(json.contains(r#""output_chars":0"#));
701        // Verify error_subtype is omitted when None (backward compat)
702        assert!(!json.contains("error_subtype"));
703    }
704
705    #[test]
706    fn test_metric_event_error_subtype_some_serializes() {
707        let event = MetricEvent {
708            ts: 1_700_000_000_000,
709            tool: "edit_replace",
710            duration_ms: 10,
711            output_chars: 0,
712            param_path_depth: 2,
713            max_depth: None,
714            result: "error",
715            error_type: Some("invalid_params".to_string()),
716            error_subtype: Some("not_found".to_string()),
717            session_id: None,
718            seq: None,
719            cache_hit: None,
720            cache_write_failure: None,
721            exit_code: None,
722            timed_out: false,
723            cache_tier: None,
724            output_truncated: None,
725            chars_threshold_breach: false,
726            file_ext: None,
727            filter_applied: None,
728            language: None,
729        };
730        let json = serde_json::to_string(&event).unwrap();
731        assert!(json.contains(r#""error_subtype":"not_found""#));
732    }
733
734    #[test]
735    fn test_metric_event_error_subtype_ambiguous() {
736        let event = MetricEvent {
737            ts: 1_700_000_000_000,
738            tool: "edit_replace",
739            duration_ms: 10,
740            output_chars: 0,
741            param_path_depth: 2,
742            max_depth: None,
743            result: "error",
744            error_type: Some("invalid_params".to_string()),
745            error_subtype: Some("ambiguous".to_string()),
746            session_id: None,
747            seq: None,
748            cache_hit: None,
749            cache_write_failure: None,
750            exit_code: None,
751            timed_out: false,
752            cache_tier: None,
753            output_truncated: None,
754            chars_threshold_breach: false,
755            file_ext: None,
756            filter_applied: None,
757            language: None,
758        };
759        let json = serde_json::to_string(&event).unwrap();
760        assert!(json.contains(r#""error_subtype":"ambiguous""#));
761    }
762
763    #[test]
764    fn test_metric_event_new_fields_round_trip() {
765        let event = MetricEvent {
766            ts: 1_700_000_000_000,
767            tool: "analyze_file",
768            duration_ms: 100,
769            output_chars: 500,
770            param_path_depth: 2,
771            max_depth: Some(3),
772            result: "ok",
773            error_type: None,
774            error_subtype: None,
775            session_id: Some("1742468880123-42".to_string()),
776            seq: Some(5),
777            cache_hit: None,
778            cache_write_failure: None,
779            exit_code: None,
780            timed_out: false,
781            cache_tier: None,
782            output_truncated: None,
783            chars_threshold_breach: false,
784            file_ext: None,
785            filter_applied: None,
786            language: None,
787        };
788        let serialized = serde_json::to_string(&event).unwrap();
789        let json_str = r#"{"ts":1700000000000,"tool":"analyze_file","duration_ms":100,"output_chars":500,"param_path_depth":2,"max_depth":3,"result":"ok","error_type":null,"session_id":"1742468880123-42","seq":5}"#;
790        assert_eq!(serialized, json_str);
791    }
792
793    #[test]
794    fn test_path_file_ext_known() {
795        // Arrange / Act / Assert: known extension returns the lowercased extension key
796        assert_eq!(path_file_ext("src/main.rs"), Some("rs"));
797    }
798
799    #[test]
800    fn test_path_file_ext_unknown() {
801        // Arrange / Act / Assert: unrecognized extension returns Some("other")
802        assert_eq!(path_file_ext("file.xyz"), Some("other"));
803    }
804
805    #[test]
806    fn test_path_file_ext_no_ext() {
807        // Arrange / Act / Assert: path with no extension returns None
808        assert_eq!(path_file_ext("Makefile"), None);
809    }
810
811    #[test]
812    fn test_path_file_ext_case_insensitive() {
813        // Arrange / Act / Assert: uppercase extension is normalized to lowercase key
814        assert_eq!(path_file_ext("src/main.RS"), Some("rs"));
815    }
816
817    #[test]
818    fn test_path_file_ext_multi_dot() {
819        // Arrange / Act / Assert: multi-dot filename uses the last extension
820        assert_eq!(path_file_ext("file.test.rs"), Some("rs"));
821    }
822
823    #[test]
824    fn test_path_language_known_ext() {
825        // Arrange / Act / Assert: known extension returns Some(language name)
826        assert_eq!(path_language("src/main.rs"), Some("rust".to_string()));
827    }
828
829    #[test]
830    fn test_path_language_unknown_ext() {
831        // Arrange / Act / Assert: unknown extension returns None
832        assert_eq!(path_language("file.xyz"), None);
833    }
834
835    #[test]
836    fn test_path_language_no_ext() {
837        // Arrange / Act / Assert: path without extension returns None
838        assert_eq!(path_language("Makefile"), None);
839    }
840
841    #[tokio::test]
842    async fn test_metrics_export_file_created() {
843        let _guard = metrics_export_lock();
844        // Arrange: create temp dir and set export env var
845        let dir = TempDir::new().unwrap();
846        let export_file = dir.path().join("metrics_export.json");
847        let export_path_str = export_file.to_string_lossy().to_string();
848
849        unsafe {
850            std::env::set_var("APTU_CODER_METRICS_EXPORT_FILE", &export_path_str);
851        }
852
853        // Create metrics writer and send events
854        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<MetricEvent>();
855        let writer = MetricsWriter::new(rx, Some(dir.path().to_path_buf()));
856
857        // Act: send a few events with session_id
858        tx.send(MetricEvent {
859            ts: unix_ms(),
860            tool: "analyze_directory",
861            duration_ms: 100,
862            output_chars: 50,
863            param_path_depth: 1,
864            max_depth: None,
865            result: "ok",
866            error_type: None,
867            error_subtype: None,
868            session_id: Some("test-session-123".to_string()),
869            seq: None,
870            cache_hit: None,
871            cache_write_failure: None,
872            exit_code: None,
873            timed_out: false,
874            cache_tier: None,
875            output_truncated: None,
876            chars_threshold_breach: false,
877            file_ext: None,
878            filter_applied: None,
879            language: None,
880        })
881        .unwrap();
882        tx.send(MetricEvent {
883            ts: unix_ms(),
884            tool: "analyze_file",
885            duration_ms: 50,
886            output_chars: 100,
887            param_path_depth: 2,
888            max_depth: Some(3),
889            result: "ok",
890            error_type: None,
891            error_subtype: None,
892            session_id: Some("test-session-123".to_string()),
893            seq: None,
894            cache_hit: None,
895            cache_write_failure: None,
896            exit_code: None,
897            timed_out: false,
898            cache_tier: None,
899            output_truncated: None,
900            chars_threshold_breach: false,
901            file_ext: None,
902            filter_applied: None,
903            language: None,
904        })
905        .unwrap();
906        drop(tx);
907        writer.run().await;
908
909        // Assert: export file should exist with correct JSON structure
910        assert!(
911            export_file.exists(),
912            "export file should be created at {:?}",
913            export_file
914        );
915        let content = std::fs::read_to_string(&export_file).unwrap();
916        let json: serde_json::Value = serde_json::from_str(&content).unwrap();
917
918        assert_eq!(
919            json["session_id"], "test-session-123",
920            "export should contain correct session_id"
921        );
922        assert!(
923            json["tool_calls"].is_array(),
924            "export should contain tool_calls array"
925        );
926        let tool_calls = json["tool_calls"].as_array().unwrap();
927        assert_eq!(tool_calls.len(), 2, "should have 2 tool calls");
928        assert!(
929            json["total_duration_ms"].is_number(),
930            "export should contain total_duration_ms"
931        );
932        assert_eq!(
933            json["total_duration_ms"], 150,
934            "total_duration_ms should be sum of all durations"
935        );
936        assert_eq!(
937            json["tool_calls"][0]["total_output_chars"], 50,
938            "first tool call should have total_output_chars=50"
939        );
940        assert_eq!(
941            json["tool_calls"][1]["total_output_chars"], 100,
942            "second tool call should have total_output_chars=100"
943        );
944        assert_eq!(
945            json["total_output_chars"], 150,
946            "total_output_chars should be sum of all output_chars"
947        );
948
949        // Cleanup
950        unsafe {
951            std::env::remove_var("APTU_CODER_METRICS_EXPORT_FILE");
952        }
953    }
954
955    #[tokio::test]
956    async fn test_metrics_export_env_var_unset() {
957        let _guard = metrics_export_lock();
958        // Arrange: ensure env var is not set
959        unsafe {
960            std::env::remove_var("APTU_CODER_METRICS_EXPORT_FILE");
961        }
962        let dir = TempDir::new().unwrap();
963        // Use a unique marker to ensure we don't pick up files from other tests
964        let marker = "metrics_export_unset_test";
965
966        // Create metrics writer and send events
967        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<MetricEvent>();
968        let writer = MetricsWriter::new(rx, Some(dir.path().to_path_buf()));
969
970        // Act: send events and run writer
971        tx.send(MetricEvent {
972            ts: unix_ms(),
973            tool: "analyze_directory",
974            duration_ms: 100,
975            output_chars: 50,
976            param_path_depth: 1,
977            max_depth: None,
978            result: "ok",
979            error_type: None,
980            error_subtype: None,
981            session_id: Some("test-session-456".to_string()),
982            seq: None,
983            cache_hit: None,
984            cache_write_failure: None,
985            exit_code: None,
986            timed_out: false,
987            cache_tier: None,
988            output_truncated: None,
989            chars_threshold_breach: false,
990            file_ext: None,
991            filter_applied: None,
992            language: None,
993        })
994        .unwrap();
995        drop(tx);
996        writer.run().await;
997
998        // Assert: no export file should be created
999        let entries: Vec<_> = std::fs::read_dir(dir.path())
1000            .unwrap()
1001            .filter_map(|e| e.ok())
1002            .filter(|e| {
1003                e.path()
1004                    .file_name()
1005                    .and_then(|n| n.to_str())
1006                    .map(|n| n.contains(marker))
1007                    .unwrap_or(false)
1008            })
1009            .collect();
1010        assert_eq!(
1011            entries.len(),
1012            0,
1013            "no export file should be created when env var is unset"
1014        );
1015    }
1016
1017    #[tokio::test]
1018    async fn test_metrics_export_relative_path_rejected() {
1019        let _guard = metrics_export_lock();
1020        // Arrange: set export env var to a relative path
1021        let relative_path = "relative/path/metrics.json";
1022        unsafe {
1023            std::env::set_var("APTU_CODER_METRICS_EXPORT_FILE", relative_path);
1024        }
1025
1026        let dir = TempDir::new().unwrap();
1027        // Use a unique marker to ensure we don't pick up files from other tests
1028        let marker = "metrics_export_relative_test";
1029
1030        // Create metrics writer and send events
1031        let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<MetricEvent>();
1032        let writer = MetricsWriter::new(rx, Some(dir.path().to_path_buf()));
1033
1034        // Act: send events and run writer
1035        tx.send(MetricEvent {
1036            ts: unix_ms(),
1037            tool: "analyze_directory",
1038            duration_ms: 100,
1039            output_chars: 50,
1040            param_path_depth: 1,
1041            max_depth: None,
1042            result: "ok",
1043            error_type: None,
1044            error_subtype: None,
1045            session_id: Some(marker.to_string()),
1046            seq: None,
1047            cache_hit: None,
1048            cache_write_failure: None,
1049            exit_code: None,
1050            timed_out: false,
1051            cache_tier: None,
1052            output_truncated: None,
1053            chars_threshold_breach: false,
1054            file_ext: None,
1055            filter_applied: None,
1056            language: None,
1057        })
1058        .unwrap();
1059        drop(tx);
1060        writer.run().await;
1061
1062        // Assert: no export file should be created for relative path
1063        let entries: Vec<_> = std::fs::read_dir(dir.path())
1064            .unwrap()
1065            .filter_map(|e| e.ok())
1066            .filter(|e| {
1067                e.path()
1068                    .file_name()
1069                    .and_then(|n| n.to_str())
1070                    .map(|n| n.contains("metrics.json"))
1071                    .unwrap_or(false)
1072            })
1073            .collect();
1074        assert_eq!(
1075            entries.len(),
1076            0,
1077            "no export file should be created for relative path"
1078        );
1079
1080        // Cleanup
1081        unsafe {
1082            std::env::remove_var("APTU_CODER_METRICS_EXPORT_FILE");
1083        }
1084    }
1085}
1086
1087/// Record a metric event to OTel metrics if the global meter provider is available.
1088///
1089/// Records:
1090/// - Histogram: mcp.server.operation.duration (in milliseconds)
1091/// - Counter: mcp.server.tool.calls (incremented by 1)
1092///
1093/// Labels: gen_ai.tool.name, error.type (or "none" if no error)
1094///
1095/// Instruments are initialized once via OnceLock to avoid rebuilding them on every call.
1096fn record_otel_metrics(event: &MetricEvent) {
1097    use opentelemetry::metrics::{Counter, Histogram};
1098    use opentelemetry::{KeyValue, global};
1099    use std::sync::OnceLock;
1100
1101    static DURATION_HISTOGRAM: OnceLock<Histogram<f64>> = OnceLock::new();
1102    static CALL_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
1103    static CACHE_HITS_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
1104    static CACHE_WRITE_FAILURES_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
1105
1106    let histogram = DURATION_HISTOGRAM.get_or_init(|| {
1107        global::meter("aptu-coder")
1108            .f64_histogram("mcp.server.operation.duration")
1109            .with_unit("s")
1110            .with_boundaries(vec![
1111                0.01, 0.02, 0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0,
1112            ])
1113            .build()
1114    });
1115
1116    let counter = CALL_COUNTER.get_or_init(|| {
1117        global::meter("aptu-coder")
1118            .u64_counter("mcp.server.tool.calls")
1119            .build()
1120    });
1121
1122    let cache_hits_counter = CACHE_HITS_COUNTER.get_or_init(|| {
1123        global::meter("aptu-coder")
1124            .u64_counter("mcp.server.tool.cache_hits_total")
1125            .with_description("Number of tool responses served from cache (l1_memory or l2_disk)")
1126            .build()
1127    });
1128
1129    let cache_write_failures_counter = CACHE_WRITE_FAILURES_COUNTER.get_or_init(|| {
1130        global::meter("aptu-coder")
1131            .u64_counter("mcp.server.tool.cache_write_failures_total")
1132            .with_description(
1133                "Number of L2 disk cache write failures (dir, tempfile, write, rename)",
1134            )
1135            .build()
1136    });
1137
1138    let error_type = event.error_type.as_deref().unwrap_or("success");
1139    let attributes = [
1140        KeyValue::new("gen_ai.tool.name", event.tool.to_string()),
1141        KeyValue::new("error.type", error_type.to_string()),
1142        KeyValue::new("mcp.method.name", "tools/call"),
1143        KeyValue::new("mcp.protocol.version", "2025-11-25"),
1144        KeyValue::new("network.transport", "pipe"),
1145    ];
1146
1147    histogram.record(event.duration_ms as f64 / 1000.0, &attributes);
1148    counter.add(1, &attributes);
1149
1150    if event.cache_hit == Some(true) {
1151        let tier = event.cache_tier.unwrap_or("unknown");
1152        cache_hits_counter.add(
1153            1,
1154            &[
1155                KeyValue::new("gen_ai.tool.name", event.tool.to_string()),
1156                KeyValue::new("cache_tier", tier.to_string()),
1157            ],
1158        );
1159    }
1160
1161    if event.cache_write_failure == Some(true) {
1162        cache_write_failures_counter.add(
1163            1,
1164            &[KeyValue::new("gen_ai.tool.name", event.tool.to_string())],
1165        );
1166    }
1167}
aptu_coder/metrics.rs

aptu_coder/
metrics.rs
