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
10// Re-export types from metrics_export so the lib.rs re-export chain stays intact.
11pub use crate::metrics_export::MetricsWriter;
12pub use crate::metrics_export::migrate_legacy_metrics_dir;
13// Re-export helpers used by tool handlers via crate::metrics::*
14pub(crate) use crate::metrics_export::{
15    path_component_count, path_file_ext, path_language, unix_ms,
16};
17
18use opentelemetry::metrics::{Counter, Histogram};
19use opentelemetry::{KeyValue, global};
20use serde::{Deserialize, Serialize};
21use std::sync::OnceLock;
22
23/// A single metric event emitted by a tool invocation.
24#[derive(Debug, Clone, Default, Serialize, Deserialize)]
25#[serde(default)]
26pub struct MetricEvent {
27    pub ts: u64,
28    pub tool: &'static str,
29    pub duration_ms: u64,
30    pub output_chars: usize,
31    pub param_path_depth: usize,
32    pub max_depth: Option<u32>,
33    pub result: &'static str,
34    #[serde(default, skip_serializing_if = "Option::is_none")]
35    pub error_type: Option<String>,
36    #[serde(default, skip_serializing_if = "Option::is_none")]
37    pub error_subtype: Option<String>,
38    #[serde(default)]
39    pub session_id: Option<String>,
40    #[serde(default)]
41    pub seq: Option<u32>,
42    #[serde(default)]
43    #[serde(skip_serializing_if = "Option::is_none")]
44    pub cache_hit: Option<bool>,
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    pub cache_tier: Option<&'static str>,
47    /// Set to Some(true) when an L2 disk cache write fails (dir, tempfile, write, or rename).
48    /// Drives the cache_write_failures_total OTEL counter.
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub cache_write_failure: Option<bool>,
51    #[serde(default, skip_serializing_if = "Option::is_none")]
52    pub exit_code: Option<i32>,
53    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
54    pub timed_out: bool,
55    #[serde(default, skip_serializing_if = "Option::is_none")]
56    pub output_truncated: Option<bool>,
57    /// True when `output_chars > 30_000`; fires for the top ~0.33% of exec_command calls
58    /// (p99.7 of 27,981 observed calls). Early-warning signal for responses approaching
59    /// the per-stream byte-cap threshold.
60    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
61    pub chars_threshold_breach: bool,
62    /// File extension of the analyzed path, lowercased. `Some("rs")` for known extensions,
63    /// `Some("other")` for unrecognized extensions, `None` when the path has no extension.
64    /// Only populated for `analyze_file` and `analyze_module`.
65    #[serde(default, skip_serializing_if = "Option::is_none")]
66    pub file_ext: Option<&'static str>,
67    /// Name of the filter rule that matched and transformed exec_command output.
68    /// `None` when no filter fired or for non-`exec_command` tools.
69    #[serde(default, skip_serializing_if = "Option::is_none")]
70    pub filter_applied: Option<String>,
71    /// Human-readable programming language name derived from the file extension
72    /// (e.g., `Some("Rust")` for `.rs` files). `None` when the path has no extension
73    /// or the extension is not recognized. Only populated for `analyze_file` and
74    /// `analyze_module`.
75    #[serde(default, skip_serializing_if = "Option::is_none")]
76    pub language: Option<String>,
77    /// Whether the tool call used a `git_ref` parameter. Populated by `analyze_directory`
78    /// and `analyze_symbol`.
79    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
80    pub git_ref_used: bool,
81    /// Whether the tool call used `summary=true` or was auto-summarized.
82    /// Populated by `analyze_directory`, `analyze_file`, and `analyze_symbol`.
83    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
84    pub summary_mode: bool,
85    /// Whether the tool call used pagination (`cursor` was provided).
86    /// Populated by `analyze_directory`, `analyze_file`, and `analyze_symbol`.
87    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
88    pub is_paginated: bool,
89    /// Whether the `fields` parameter was provided to `analyze_file`.
90    /// Populated by `analyze_file`.
91    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
92    pub fields_projected: bool,
93    /// Symbol matching mode used by `analyze_symbol` (e.g., "exact", "insensitive").
94    /// `None` when not an `analyze_symbol` call.
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub match_mode: Option<String>,
97    /// Call graph traversal depth for `analyze_symbol` (default 1).
98    #[serde(default, skip_serializing_if = "Option::is_none")]
99    pub follow_depth: Option<u32>,
100    /// Whether `import_lookup=true` was set on `analyze_symbol`.
101    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
102    pub import_lookup: bool,
103    /// Whether `def_use=true` was set on `analyze_symbol`.
104    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
105    pub def_use: bool,
106    /// Whether `impl_only=true` was set on `analyze_symbol`.
107    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
108    pub impl_only: bool,
109    /// Whether `stdin` was provided to `exec_command`.
110    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
111    pub stdin_provided: bool,
112    /// Configured timeout in milliseconds for `exec_command`. `None` means no limit.
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub timeout_configured_ms: Option<i64>,
115    /// Drain timeout in milliseconds for `exec_command`. `None` means default (500ms).
116    #[serde(default, skip_serializing_if = "Option::is_none")]
117    pub drain_timeout_ms: Option<i64>,
118    /// Whether a `working_dir` parameter was provided. Populated by `edit_overwrite`,
119    /// `edit_replace`, and `exec_command`.
120    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
121    pub working_dir_used: bool,
122}
123
124/// Fluent builder for MetricEvent. Reduces repetitive struct literal boilerplate.
125#[derive(Debug, Default)]
126pub(crate) struct MetricEventBuilder {
127    ts: u64,
128    tool: &'static str,
129    duration_ms: u64,
130    output_chars: usize,
131    param_path_depth: usize,
132    max_depth: Option<u32>,
133    result: &'static str,
134    error_type: Option<String>,
135    error_subtype: Option<String>,
136    session_id: Option<String>,
137    seq: Option<u32>,
138    cache_hit: Option<bool>,
139    cache_write_failure: Option<bool>,
140    cache_tier: Option<&'static str>,
141    exit_code: Option<i32>,
142    timed_out: bool,
143    output_truncated: Option<bool>,
144    chars_threshold_breach: bool,
145    file_ext: Option<&'static str>,
146    filter_applied: Option<String>,
147    language: Option<String>,
148    git_ref_used: bool,
149    summary_mode: bool,
150    is_paginated: bool,
151    fields_projected: bool,
152    match_mode: Option<String>,
153    follow_depth: Option<u32>,
154    import_lookup: bool,
155    def_use: bool,
156    impl_only: bool,
157    stdin_provided: bool,
158    timeout_configured_ms: Option<i64>,
159    drain_timeout_ms: Option<i64>,
160    working_dir_used: bool,
161}
162
163#[allow(clippy::too_many_arguments)]
164impl MetricEventBuilder {
165    #[must_use]
166    pub(crate) fn new(tool: &'static str, result: &'static str, duration_ms: u64) -> Self {
167        Self {
168            ts: unix_ms(),
169            tool,
170            result,
171            duration_ms,
172            ..Self::default()
173        }
174    }
175
176    #[must_use]
177    pub(crate) fn output_chars(mut self, v: usize) -> Self {
178        self.output_chars = v;
179        self
180    }
181    #[must_use]
182    pub(crate) fn param_path_depth(mut self, v: usize) -> Self {
183        self.param_path_depth = v;
184        self
185    }
186    #[must_use]
187    pub(crate) fn max_depth(mut self, v: Option<u32>) -> Self {
188        self.max_depth = v;
189        self
190    }
191    #[must_use]
192    pub(crate) fn error_type(mut self, v: Option<String>) -> Self {
193        self.error_type = v;
194        self
195    }
196    #[must_use]
197    pub(crate) fn error_subtype(mut self, v: Option<String>) -> Self {
198        self.error_subtype = v;
199        self
200    }
201    #[must_use]
202    pub(crate) fn session_id(mut self, v: Option<String>) -> Self {
203        self.session_id = v;
204        self
205    }
206    #[must_use]
207    pub(crate) fn seq(mut self, v: Option<u32>) -> Self {
208        self.seq = v;
209        self
210    }
211    #[must_use]
212    pub(crate) fn cache_hit(mut self, v: Option<bool>) -> Self {
213        self.cache_hit = v;
214        self
215    }
216    #[must_use]
217    pub(crate) fn cache_tier(mut self, v: Option<&'static str>) -> Self {
218        self.cache_tier = v;
219        self
220    }
221    #[must_use]
222    pub(crate) fn cache_write_failure(mut self, v: Option<bool>) -> Self {
223        self.cache_write_failure = v;
224        self
225    }
226    #[must_use]
227    pub(crate) fn exit_code(mut self, v: Option<i32>) -> Self {
228        self.exit_code = v;
229        self
230    }
231    #[must_use]
232    pub(crate) fn timed_out(mut self, v: bool) -> Self {
233        self.timed_out = v;
234        self
235    }
236    #[must_use]
237    pub(crate) fn output_truncated(mut self, v: Option<bool>) -> Self {
238        self.output_truncated = v;
239        self
240    }
241    #[must_use]
242    pub(crate) fn chars_threshold_breach(mut self, v: bool) -> Self {
243        self.chars_threshold_breach = v;
244        self
245    }
246    #[must_use]
247    pub(crate) fn file_ext(mut self, v: Option<&'static str>) -> Self {
248        self.file_ext = v;
249        self
250    }
251    #[must_use]
252    pub(crate) fn filter_applied(mut self, v: Option<String>) -> Self {
253        self.filter_applied = v;
254        self
255    }
256    #[must_use]
257    pub(crate) fn language(mut self, v: Option<String>) -> Self {
258        self.language = v;
259        self
260    }
261    #[must_use]
262    pub(crate) fn git_ref_used(mut self, v: bool) -> Self {
263        self.git_ref_used = v;
264        self
265    }
266    #[must_use]
267    pub(crate) fn summary_mode(mut self, v: bool) -> Self {
268        self.summary_mode = v;
269        self
270    }
271    #[must_use]
272    #[allow(clippy::wrong_self_convention)]
273    pub(crate) fn is_paginated(mut self, v: bool) -> Self {
274        self.is_paginated = v;
275        self
276    }
277    #[must_use]
278    pub(crate) fn fields_projected(mut self, v: bool) -> Self {
279        self.fields_projected = v;
280        self
281    }
282    #[must_use]
283    pub(crate) fn match_mode(mut self, v: Option<String>) -> Self {
284        self.match_mode = v;
285        self
286    }
287    #[must_use]
288    pub(crate) fn follow_depth(mut self, v: Option<u32>) -> Self {
289        self.follow_depth = v;
290        self
291    }
292    #[must_use]
293    pub(crate) fn import_lookup(mut self, v: bool) -> Self {
294        self.import_lookup = v;
295        self
296    }
297    #[must_use]
298    pub(crate) fn def_use(mut self, v: bool) -> Self {
299        self.def_use = v;
300        self
301    }
302    #[must_use]
303    pub(crate) fn impl_only(mut self, v: bool) -> Self {
304        self.impl_only = v;
305        self
306    }
307    #[must_use]
308    pub(crate) fn stdin_provided(mut self, v: bool) -> Self {
309        self.stdin_provided = v;
310        self
311    }
312    #[must_use]
313    pub(crate) fn timeout_configured_ms(mut self, v: Option<i64>) -> Self {
314        self.timeout_configured_ms = v;
315        self
316    }
317    #[must_use]
318    pub(crate) fn drain_timeout_ms(mut self, v: Option<i64>) -> Self {
319        self.drain_timeout_ms = v;
320        self
321    }
322    #[must_use]
323    pub(crate) fn working_dir_used(mut self, v: bool) -> Self {
324        self.working_dir_used = v;
325        self
326    }
327    #[must_use]
328    pub(crate) fn build(self) -> MetricEvent {
329        MetricEvent {
330            ts: self.ts,
331            tool: self.tool,
332            duration_ms: self.duration_ms,
333            output_chars: self.output_chars,
334            param_path_depth: self.param_path_depth,
335            max_depth: self.max_depth,
336            result: self.result,
337            error_type: self.error_type,
338            error_subtype: self.error_subtype,
339            session_id: self.session_id,
340            seq: self.seq,
341            cache_hit: self.cache_hit,
342            cache_write_failure: self.cache_write_failure,
343            cache_tier: self.cache_tier,
344            exit_code: self.exit_code,
345            timed_out: self.timed_out,
346            output_truncated: self.output_truncated,
347            chars_threshold_breach: self.chars_threshold_breach,
348            file_ext: self.file_ext,
349            filter_applied: self.filter_applied,
350            language: self.language,
351            git_ref_used: self.git_ref_used,
352            summary_mode: self.summary_mode,
353            is_paginated: self.is_paginated,
354            fields_projected: self.fields_projected,
355            match_mode: self.match_mode,
356            follow_depth: self.follow_depth,
357            import_lookup: self.import_lookup,
358            def_use: self.def_use,
359            impl_only: self.impl_only,
360            stdin_provided: self.stdin_provided,
361            timeout_configured_ms: self.timeout_configured_ms,
362            drain_timeout_ms: self.drain_timeout_ms,
363            working_dir_used: self.working_dir_used,
364        }
365    }
366}
367
368/// Sender half of the metrics channel; cloned and passed to tools for event emission.
369#[derive(Clone)]
370pub struct MetricsSender(pub tokio::sync::mpsc::UnboundedSender<MetricEvent>);
371
372impl MetricsSender {
373    pub fn send(&self, event: MetricEvent) {
374        let _ = self.0.send(event);
375    }
376}
377
378/// Accumulated metrics for a single tool.
379#[derive(Default, Debug)]
380pub(crate) struct ToolMetrics {
381    pub(crate) count: u64,
382    pub(crate) duration_ms: u64,
383    pub(crate) output_chars: u64,
384}
385
386/// RAII guard that releases an exclusive lock on a metrics .lock file when dropped.
387/// Lock release happens implicitly when the underlying `std::fs::File` is closed.
388#[allow(dead_code)]
389pub(crate) struct MetricsLockGuard(pub(crate) std::fs::File);
390
391/// Record a metric event to OTel metrics if the global meter provider is available.
392///
393/// Records:
394/// - Histogram: mcp.server.operation.duration (in milliseconds)
395/// - Counter: mcp.server.tool.calls (incremented by 1)
396///
397/// Labels: gen_ai.tool.name, error.type (or "none" if no error)
398///
399/// Instruments are initialized once via OnceLock to avoid rebuilding them on every call.
400pub(crate) fn record_otel_metrics(event: &MetricEvent) {
401    // Skip OTEL recording for "received" events (duration_ms=0 would pollute latency histograms)
402    if event.result == "received" {
403        return;
404    }
405
406    static DURATION_HISTOGRAM: OnceLock<Histogram<f64>> = OnceLock::new();
407    static CALL_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
408    static CACHE_HITS_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
409    static CACHE_WRITE_FAILURES_COUNTER: OnceLock<Counter<u64>> = OnceLock::new();
410
411    let histogram = DURATION_HISTOGRAM.get_or_init(|| {
412        global::meter("aptu-coder")
413            .f64_histogram("mcp.server.operation.duration")
414            .with_unit("s")
415            .with_boundaries(vec![
416                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,
417            ])
418            .build()
419    });
420
421    let counter = CALL_COUNTER.get_or_init(|| {
422        global::meter("aptu-coder")
423            .u64_counter("mcp.server.tool.calls")
424            .build()
425    });
426
427    let cache_hits_counter = CACHE_HITS_COUNTER.get_or_init(|| {
428        global::meter("aptu-coder")
429            .u64_counter("mcp.server.tool.cache_hits_total")
430            .with_description("Number of tool responses served from cache (l1_memory or l2_disk)")
431            .build()
432    });
433
434    let cache_write_failures_counter = CACHE_WRITE_FAILURES_COUNTER.get_or_init(|| {
435        global::meter("aptu-coder")
436            .u64_counter("mcp.server.tool.cache_write_failures_total")
437            .with_description(
438                "Number of L2 disk cache write failures (dir, tempfile, write, rename)",
439            )
440            .build()
441    });
442
443    let error_type = event.error_type.as_deref().unwrap_or("success");
444    let attributes = [
445        KeyValue::new("gen_ai.tool.name", event.tool),
446        KeyValue::new("error.type", error_type.to_string()),
447        KeyValue::new("mcp.method.name", "tools/call"),
448        KeyValue::new("mcp.protocol.version", "2025-11-25"),
449        KeyValue::new("network.transport", "pipe"),
450    ];
451
452    histogram.record(event.duration_ms as f64 / 1000.0, &attributes);
453    counter.add(1, &attributes);
454
455    if event.cache_hit == Some(true) {
456        let tier = event.cache_tier.unwrap_or("unknown");
457        cache_hits_counter.add(
458            1,
459            &[
460                KeyValue::new("gen_ai.tool.name", event.tool),
461                KeyValue::new("cache_tier", tier),
462            ],
463        );
464    }
465
466    if event.cache_write_failure == Some(true) {
467        cache_write_failures_counter.add(1, &[KeyValue::new("gen_ai.tool.name", event.tool)]);
468    }
469}
470
471#[cfg(test)]
472mod tests {
473    use super::*;
474
475    #[test]
476    fn test_metric_event_serialization() {
477        let event = MetricEvent {
478            ts: 1_700_000_000_000,
479            tool: "analyze_directory",
480            duration_ms: 100,
481            output_chars: 500,
482            param_path_depth: 1,
483            max_depth: None,
484            result: "ok",
485            error_type: None,
486            error_subtype: None,
487            session_id: Some("1742468880123-42".to_string()),
488            seq: Some(5),
489            cache_hit: None,
490            cache_write_failure: None,
491            cache_tier: None,
492            exit_code: Some(0),
493            timed_out: false,
494            output_truncated: None,
495            chars_threshold_breach: false,
496            file_ext: None,
497            ..Default::default()
498        };
499        let serialized = serde_json::to_string(&event).unwrap();
500        assert!(serialized.contains(r#""ts":1700000000000"#));
501        assert!(serialized.contains(r#""tool":"analyze_directory""#));
502        assert!(serialized.contains(r#""session_id":"1742468880123-42""#));
503        assert!(serialized.contains(r#""exit_code":0"#));
504    }
505
506    #[test]
507    fn test_metric_event_serialization_error() {
508        let event = MetricEvent {
509            ts: 1_700_000_000_000,
510            tool: "edit_replace",
511            duration_ms: 10,
512            output_chars: 0,
513            param_path_depth: 2,
514            max_depth: None,
515            result: "error",
516            error_type: Some("invalid_params".to_string()),
517            session_id: None,
518            seq: None,
519            cache_hit: None,
520            cache_write_failure: None,
521            exit_code: None,
522            timed_out: false,
523            cache_tier: None,
524            output_truncated: None,
525            chars_threshold_breach: false,
526            file_ext: None,
527            ..Default::default()
528        };
529        let json = serde_json::to_string(&event).unwrap();
530        assert!(json.contains(r#""error_type":"invalid_params""#));
531    }
532
533    #[test]
534    fn test_metric_event_error_subtype_some_serializes() {
535        let event = MetricEvent {
536            ts: 1_700_000_000_000,
537            tool: "edit_replace",
538            duration_ms: 10,
539            output_chars: 0,
540            param_path_depth: 2,
541            max_depth: None,
542            result: "error",
543            error_type: Some("invalid_params".to_string()),
544            error_subtype: Some("not_found".to_string()),
545            session_id: None,
546            seq: None,
547            cache_hit: None,
548            cache_write_failure: None,
549            exit_code: None,
550            timed_out: false,
551            cache_tier: None,
552            output_truncated: None,
553            chars_threshold_breach: false,
554            file_ext: None,
555            ..Default::default()
556        };
557        let json = serde_json::to_string(&event).unwrap();
558        assert!(json.contains(r#""error_subtype":"not_found""#));
559    }
560
561    #[test]
562    fn test_metric_event_error_subtype_ambiguous() {
563        let event = MetricEvent {
564            ts: 1_700_000_000_000,
565            tool: "edit_replace",
566            duration_ms: 10,
567            output_chars: 0,
568            param_path_depth: 2,
569            max_depth: None,
570            result: "error",
571            error_type: Some("invalid_params".to_string()),
572            error_subtype: Some("ambiguous".to_string()),
573            session_id: None,
574            seq: None,
575            cache_hit: None,
576            cache_write_failure: None,
577            exit_code: None,
578            timed_out: false,
579            cache_tier: None,
580            output_truncated: None,
581            chars_threshold_breach: false,
582            file_ext: None,
583            ..Default::default()
584        };
585        let json = serde_json::to_string(&event).unwrap();
586        assert!(json.contains(r#""error_subtype":"ambiguous""#));
587    }
588
589    #[test]
590    fn test_metric_event_new_fields_round_trip() {
591        let event = MetricEvent {
592            ts: 1_700_000_000_000,
593            tool: "analyze_file",
594            duration_ms: 100,
595            output_chars: 500,
596            param_path_depth: 2,
597            max_depth: Some(3),
598            result: "ok",
599            error_type: None,
600            error_subtype: None,
601            session_id: Some("1742468880123-42".to_string()),
602            seq: Some(5),
603            cache_hit: None,
604            cache_write_failure: None,
605            exit_code: None,
606            timed_out: false,
607            cache_tier: None,
608            output_truncated: None,
609            chars_threshold_breach: false,
610            file_ext: None,
611            filter_applied: None,
612            language: None,
613            git_ref_used: false,
614            summary_mode: false,
615            is_paginated: false,
616            fields_projected: false,
617            match_mode: None,
618            follow_depth: None,
619            import_lookup: false,
620            def_use: false,
621            impl_only: false,
622            stdin_provided: false,
623            timeout_configured_ms: None,
624            drain_timeout_ms: None,
625            working_dir_used: false,
626        };
627        let serialized = serde_json::to_string(&event).unwrap();
628        let json_str = r#"{"ts":1700000000000,"tool":"analyze_file","duration_ms":100,"output_chars":500,"param_path_depth":2,"max_depth":3,"result":"ok","session_id":"1742468880123-42","seq":5}"#;
629        assert_eq!(serialized, json_str);
630    }
631}
aptu_coder/metrics.rs

aptu_coder/
metrics.rs
