winx_code_agent/
report.rs

1//! Offline aggregation for privacy-safe `winx::usage` JSONL logs.
2
3use std::collections::{BTreeMap, BTreeSet, VecDeque};
4use std::fs::{self, File};
5use std::io::{BufRead as _, BufReader};
6use std::path::{Path, PathBuf};
7
8use serde::Serialize;
9use serde_json::Value;
10use time::{format_description::well_known::Rfc3339, Duration, OffsetDateTime};
11
12use crate::errors::{Result, WinxError};
13
14const MAX_REPORT_EVENTS: usize = 100_000;
15
16#[derive(Debug, Serialize)]
17#[serde(rename_all = "camelCase")]
18pub struct UsageReport {
19    pub base_path: String,
20    pub source_files: usize,
21    pub lines_seen: u64,
22    pub invalid_lines: u64,
23    pub events: usize,
24    pub first_timestamp: Option<String>,
25    pub last_timestamp: Option<String>,
26    pub tool_calls: usize,
27    pub http_requests: usize,
28    pub correlated_requests: usize,
29    pub tool_calls_without_http: usize,
30    pub http_without_tool_call: usize,
31    pub duplicate_tool_request_ids: usize,
32    pub http_statuses: BTreeMap<String, u64>,
33    pub result_statuses: BTreeMap<String, u64>,
34    pub error_codes: BTreeMap<String, u64>,
35    pub command_kinds: BTreeMap<String, u64>,
36    pub build_identities: BTreeMap<String, u64>,
37    pub workspace_coherence: BTreeMap<String, u64>,
38    pub tools: BTreeMap<String, ToolUsage>,
39    pub latency: LatencySummary,
40    pub http_overhead: LatencySummary,
41    pub response_bytes: u64,
42    pub initialize: InitializeUsage,
43    pub recovery: RecoveryUsage,
44    pub trace_audit: TraceAudit,
45}
46
47#[derive(Debug, Default, Serialize)]
48#[serde(rename_all = "camelCase")]
49pub struct ToolUsage {
50    pub calls: u64,
51    pub error_results: u64,
52    pub response_bytes: u64,
53    pub latency: LatencySummary,
54}
55
56#[derive(Clone, Debug, Default, Serialize)]
57#[serde(rename_all = "camelCase")]
58pub struct LatencySummary {
59    pub samples: usize,
60    pub average_ms: u64,
61    pub p50_ms: u64,
62    pub p95_ms: u64,
63    pub max_ms: u64,
64}
65
66#[derive(Debug, Default, Serialize)]
67#[serde(rename_all = "camelCase")]
68pub struct InitializeUsage {
69    pub calls: u64,
70    pub reused: u64,
71    pub response_bytes: u64,
72    pub transitions: BTreeMap<String, u64>,
73}
74
75#[derive(Debug, Default, Serialize)]
76#[serde(rename_all = "camelCase")]
77pub struct RecoveryUsage {
78    pub next_actions: u64,
79    pub required_reads: u64,
80    pub repeated_attempts: u64,
81    pub escalated: u64,
82    pub mutation_receipts_persisted: u64,
83}
84
85#[derive(Debug, Default, Serialize)]
86#[serde(rename_all = "camelCase")]
87pub struct TraceAudit {
88    pub healthy: bool,
89    pub violations: Vec<TraceViolation>,
90    pub workspace_coherence_rejections: u64,
91    pub audited_tool_events: u64,
92    pub legacy_unverifiable_tool_events: u64,
93}
94
95#[derive(Debug, Serialize)]
96#[serde(rename_all = "camelCase")]
97pub struct TraceViolation {
98    pub request_id: String,
99    pub rule: &'static str,
100}
101
102#[derive(Debug)]
103struct UsageEvent {
104    timestamp: Option<String>,
105    fields: serde_json::Map<String, Value>,
106}
107
108#[derive(Default)]
109struct ToolAccumulator {
110    calls: u64,
111    error_results: u64,
112    response_bytes: u64,
113    durations: Vec<u64>,
114}
115
116/// Aggregate recent usage events. `since_minutes=0` disables the time filter;
117/// `build_filter` matches either the exact display identity or package version.
118pub fn usage_report(
119    path: Option<&Path>,
120    last_events: usize,
121    since_minutes: u64,
122    build_filter: Option<&str>,
123) -> Result<UsageReport> {
124    if last_events == 0 || last_events > MAX_REPORT_EVENTS {
125        return Err(WinxError::InvalidInput(format!(
126            "--last must be between 1 and {MAX_REPORT_EVENTS}"
127        )));
128    }
129    let base = path.map_or_else(default_usage_path, Path::to_path_buf);
130    let sources = usage_sources(&base)?;
131    let cutoff = (since_minutes > 0)
132        .then(|| OffsetDateTime::now_utc() - Duration::minutes(i64_saturating(since_minutes)));
133    let mut recent = VecDeque::with_capacity(last_events.min(4096));
134    let mut lines_seen = 0_u64;
135    let mut invalid_lines = 0_u64;
136    for source in &sources {
137        let reader =
138            BufReader::new(File::open(source).map_err(|error| WinxError::FileAccessError {
139                path: source.clone(),
140                message: format!("opening usage log: {error}"),
141            })?);
142        for line in reader.lines() {
143            lines_seen = lines_seen.saturating_add(1);
144            let Ok(line) = line else {
145                invalid_lines = invalid_lines.saturating_add(1);
146                continue;
147            };
148            let Some(event) = parse_event(&line) else {
149                invalid_lines = invalid_lines.saturating_add(1);
150                continue;
151            };
152            if cutoff.is_some_and(|cutoff| {
153                event
154                    .timestamp
155                    .as_deref()
156                    .and_then(parse_timestamp)
157                    .is_some_and(|timestamp| timestamp < cutoff)
158            }) {
159                continue;
160            }
161            if build_filter.is_some_and(|filter| {
162                ![string(&event.fields, "build"), string(&event.fields, "build_version")]
163                    .contains(&filter)
164            }) {
165                continue;
166            }
167            if recent.len() == last_events {
168                recent.pop_front();
169            }
170            recent.push_back(event);
171        }
172    }
173    Ok(aggregate(&base, sources.len(), lines_seen, invalid_lines, &recent))
174}
175
176#[derive(Default)]
177struct ReportAccumulator {
178    result_statuses: BTreeMap<String, u64>,
179    error_codes: BTreeMap<String, u64>,
180    command_kinds: BTreeMap<String, u64>,
181    build_identities: BTreeMap<String, u64>,
182    workspace_coherence: BTreeMap<String, u64>,
183    http_statuses: BTreeMap<String, u64>,
184    tools: BTreeMap<String, ToolAccumulator>,
185    initialize: InitializeUsage,
186    recovery: RecoveryUsage,
187    trace_audit: TraceAudit,
188    all_durations: Vec<u64>,
189    tool_request_ids: BTreeSet<String>,
190    duplicate_tool_request_ids: usize,
191    http_request_ids: BTreeSet<String>,
192    tool_durations: BTreeMap<String, u64>,
193    http_durations: BTreeMap<String, u64>,
194    tool_calls: usize,
195    http_requests: usize,
196    response_bytes: u64,
197}
198
199impl ReportAccumulator {
200    fn record(&mut self, fields: &serde_json::Map<String, Value>) {
201        match string(fields, "event") {
202            "tool_call" => self.record_tool(fields),
203            "http_request" => self.record_http(fields),
204            _ => {}
205        }
206    }
207
208    fn record_tool(&mut self, fields: &serde_json::Map<String, Value>) {
209        self.tool_calls += 1;
210        let tool = string(fields, "tool").to_string();
211        increment_nonempty(&mut self.result_statuses, string(fields, "result_status"));
212        increment_nonempty(&mut self.error_codes, string(fields, "error_code"));
213        increment_nonempty(&mut self.command_kinds, string(fields, "command_kind"));
214        increment_nonempty(&mut self.build_identities, string(fields, "build"));
215        increment_nonempty(&mut self.workspace_coherence, string(fields, "workspace_coherence"));
216        let bytes = number(fields, "response_bytes");
217        let duration = number(fields, "duration_ms");
218        self.response_bytes = self.response_bytes.saturating_add(bytes);
219        self.all_durations.push(duration);
220
221        let accumulator = self.tools.entry(tool.clone()).or_default();
222        accumulator.calls = accumulator.calls.saturating_add(1);
223        accumulator.error_results = accumulator
224            .error_results
225            .saturating_add(u64::from(!string(fields, "error_code").is_empty()));
226        accumulator.response_bytes = accumulator.response_bytes.saturating_add(bytes);
227        accumulator.durations.push(duration);
228
229        let request_id = string(fields, "request_id");
230        if !request_id.is_empty() {
231            if !self.tool_request_ids.insert(request_id.to_string()) {
232                self.duplicate_tool_request_ids += 1;
233            }
234            self.tool_durations.insert(request_id.to_string(), duration);
235        }
236        if tool == "Initialize" {
237            self.initialize.calls = self.initialize.calls.saturating_add(1);
238            self.initialize.response_bytes = self.initialize.response_bytes.saturating_add(bytes);
239            self.initialize.reused = self
240                .initialize
241                .reused
242                .saturating_add(u64::from(boolean(fields, "initialize_reused")));
243            increment_nonempty(
244                &mut self.initialize.transitions,
245                string(fields, "initialize_transition"),
246            );
247        }
248        self.record_recovery(fields);
249        audit_tool_event(fields, &mut self.trace_audit);
250    }
251
252    fn record_recovery(&mut self, fields: &serde_json::Map<String, Value>) {
253        self.recovery.next_actions = self
254            .recovery
255            .next_actions
256            .saturating_add(u64::from(!string(fields, "next_action_tool").is_empty()));
257        self.recovery.required_reads =
258            self.recovery.required_reads.saturating_add(number(fields, "required_read_count"));
259        self.recovery.repeated_attempts = self
260            .recovery
261            .repeated_attempts
262            .saturating_add(u64::from(number(fields, "recovery_attempt") > 1));
263        self.recovery.escalated = self
264            .recovery
265            .escalated
266            .saturating_add(u64::from(string(fields, "recovery_level") == "escalated"));
267        self.recovery.mutation_receipts_persisted = self
268            .recovery
269            .mutation_receipts_persisted
270            .saturating_add(u64::from(string(fields, "mutation_receipt_state") == "persisted"));
271    }
272
273    fn record_http(&mut self, fields: &serde_json::Map<String, Value>) {
274        self.http_requests += 1;
275        let status = fields.get("status").map(value_text).unwrap_or_default();
276        increment_nonempty(&mut self.http_statuses, &status);
277        let request_id = string(fields, "request_id");
278        if !request_id.is_empty() {
279            self.http_request_ids.insert(request_id.to_string());
280            self.http_durations.insert(request_id.to_string(), number(fields, "duration_ms"));
281        }
282    }
283
284    fn finish(mut self, window: ReportWindow<'_>) -> UsageReport {
285        let correlated_requests =
286            self.tool_request_ids.intersection(&self.http_request_ids).count();
287        let overhead = self
288            .tool_durations
289            .iter()
290            .filter_map(|(id, tool)| {
291                self.http_durations.get(id).map(|http| http.saturating_sub(*tool))
292            })
293            .collect::<Vec<_>>();
294        let tools = self
295            .tools
296            .into_iter()
297            .map(|(name, accumulator)| {
298                (
299                    name,
300                    ToolUsage {
301                        calls: accumulator.calls,
302                        error_results: accumulator.error_results,
303                        response_bytes: accumulator.response_bytes,
304                        latency: latency(accumulator.durations),
305                    },
306                )
307            })
308            .collect();
309        self.trace_audit.healthy = self.trace_audit.violations.is_empty();
310
311        UsageReport {
312            base_path: window.base.to_string_lossy().into_owned(),
313            source_files: window.source_files,
314            lines_seen: window.lines_seen,
315            invalid_lines: window.invalid_lines,
316            events: window.events,
317            first_timestamp: window.first_timestamp,
318            last_timestamp: window.last_timestamp,
319            tool_calls: self.tool_calls,
320            http_requests: self.http_requests,
321            correlated_requests,
322            tool_calls_without_http: self
323                .tool_request_ids
324                .len()
325                .saturating_sub(correlated_requests),
326            http_without_tool_call: self.http_request_ids.len().saturating_sub(correlated_requests),
327            duplicate_tool_request_ids: self.duplicate_tool_request_ids,
328            http_statuses: self.http_statuses,
329            result_statuses: self.result_statuses,
330            error_codes: self.error_codes,
331            command_kinds: self.command_kinds,
332            build_identities: self.build_identities,
333            workspace_coherence: self.workspace_coherence,
334            tools,
335            latency: latency(self.all_durations),
336            http_overhead: latency(overhead),
337            response_bytes: self.response_bytes,
338            initialize: self.initialize,
339            recovery: self.recovery,
340            trace_audit: self.trace_audit,
341        }
342    }
343}
344
345struct ReportWindow<'a> {
346    base: &'a Path,
347    source_files: usize,
348    lines_seen: u64,
349    invalid_lines: u64,
350    events: usize,
351    first_timestamp: Option<String>,
352    last_timestamp: Option<String>,
353}
354
355fn aggregate(
356    base: &Path,
357    source_files: usize,
358    lines_seen: u64,
359    invalid_lines: u64,
360    events: &VecDeque<UsageEvent>,
361) -> UsageReport {
362    let mut accumulator = ReportAccumulator::default();
363    for event in events {
364        accumulator.record(&event.fields);
365    }
366    accumulator.finish(ReportWindow {
367        base,
368        source_files,
369        lines_seen,
370        invalid_lines,
371        events: events.len(),
372        first_timestamp: events.front().and_then(|event| event.timestamp.clone()),
373        last_timestamp: events.back().and_then(|event| event.timestamp.clone()),
374    })
375}
376
377fn audit_tool_event(fields: &serde_json::Map<String, Value>, audit: &mut TraceAudit) {
378    if matches!(
379        string(fields, "workspace_coherence"),
380        "mismatch" | "rejected" | "binding_mismatch" | "thread_mismatch"
381    ) {
382        audit.workspace_coherence_rejections =
383            audit.workspace_coherence_rejections.saturating_add(1);
384    }
385    if number(fields, "usage_schema") == 0 {
386        // Older logs did not record the recovery booleans/actions needed to
387        // prove these invariants. Treat them as unknown, never as violations.
388        audit.legacy_unverifiable_tool_events =
389            audit.legacy_unverifiable_tool_events.saturating_add(1);
390        return;
391    }
392    audit.audited_tool_events = audit.audited_tool_events.saturating_add(1);
393    let request_id = string(fields, "request_id");
394    let mut violation = |rule| {
395        if audit.violations.len() < 100 {
396            audit.violations.push(TraceViolation { request_id: request_id.to_string(), rule });
397        }
398    };
399    if boolean(fields, "retry_same_call") {
400        violation("retry_same_call_must_be_false");
401    }
402    if boolean(fields, "edit_applied") && boolean(fields, "retry_same_call") {
403        violation("committed_edit_must_never_be_retried");
404    }
405    if string(fields, "result_status") == "needs_read"
406        && string(fields, "next_action_tool") != "ReadFiles"
407    {
408        violation("needs_read_must_point_to_read_files");
409    }
410    if matches!(string(fields, "error_code"), "search_block_not_found" | "search_block_ambiguous")
411        && !boolean(fields, "fresh_read_required")
412    {
413        violation("search_conflict_must_invalidate_read_evidence");
414    }
415}
416
417fn usage_sources(base: &Path) -> Result<Vec<PathBuf>> {
418    let parent = base.parent().unwrap_or_else(|| Path::new("."));
419    let name = base.file_name().and_then(|name| name.to_str()).ok_or_else(|| {
420        WinxError::InvalidInput(format!("usage log path is not UTF-8: {}", base.display()))
421    })?;
422    let prefix = format!("{name}.");
423    let mut sources = fs::read_dir(parent)
424        .map_err(|error| WinxError::FileAccessError {
425            path: parent.to_path_buf(),
426            message: format!("listing usage logs: {error}"),
427        })?
428        .filter_map(std::result::Result::ok)
429        .filter_map(|entry| {
430            let entry_name = entry.file_name();
431            let entry_name = entry_name.to_str()?;
432            let metadata = entry.metadata().ok()?;
433            (metadata.is_file() && (entry_name == name || entry_name.starts_with(&prefix)))
434                .then_some(entry.path())
435        })
436        .collect::<Vec<_>>();
437    sources.sort();
438    if sources.is_empty() {
439        return Err(WinxError::FileNotFound { path: base.to_path_buf() });
440    }
441    Ok(sources)
442}
443
444fn default_usage_path() -> PathBuf {
445    std::env::var_os("WINX_USAGE_LOG").map_or_else(
446        || {
447            home::home_dir()
448                .unwrap_or_else(|| PathBuf::from("."))
449                .join(".local/state/winx/usage.jsonl")
450        },
451        PathBuf::from,
452    )
453}
454
455fn parse_event(line: &str) -> Option<UsageEvent> {
456    let value: Value = serde_json::from_str(line).ok()?;
457    (value.get("target").and_then(Value::as_str) == Some("winx::usage")).then_some(())?;
458    Some(UsageEvent {
459        timestamp: value.get("timestamp").and_then(Value::as_str).map(ToString::to_string),
460        fields: value.get("fields")?.as_object()?.clone(),
461    })
462}
463
464fn parse_timestamp(value: &str) -> Option<OffsetDateTime> {
465    OffsetDateTime::parse(value, &Rfc3339).ok()
466}
467
468fn latency(mut samples: Vec<u64>) -> LatencySummary {
469    if samples.is_empty() {
470        return LatencySummary::default();
471    }
472    samples.sort_unstable();
473    let sum = samples.iter().fold(0_u128, |sum, value| sum.saturating_add(u128::from(*value)));
474    let average = sum / samples.len() as u128;
475    LatencySummary {
476        samples: samples.len(),
477        average_ms: u64::try_from(average).unwrap_or(u64::MAX),
478        p50_ms: percentile(&samples, 50),
479        p95_ms: percentile(&samples, 95),
480        max_ms: samples.last().copied().unwrap_or_default(),
481    }
482}
483
484fn percentile(sorted: &[u64], percentile: usize) -> u64 {
485    let index = sorted.len().saturating_mul(percentile).div_ceil(100).saturating_sub(1);
486    sorted[index.min(sorted.len() - 1)]
487}
488
489fn increment_nonempty(counts: &mut BTreeMap<String, u64>, value: &str) {
490    if !value.is_empty() {
491        counts.entry(value.to_string()).and_modify(|count| *count += 1).or_insert(1);
492    }
493}
494
495fn string<'a>(fields: &'a serde_json::Map<String, Value>, key: &str) -> &'a str {
496    fields.get(key).and_then(Value::as_str).unwrap_or_default()
497}
498
499fn number(fields: &serde_json::Map<String, Value>, key: &str) -> u64 {
500    fields.get(key).and_then(Value::as_u64).unwrap_or_default()
501}
502
503fn boolean(fields: &serde_json::Map<String, Value>, key: &str) -> bool {
504    fields.get(key).and_then(Value::as_bool).unwrap_or_default()
505}
506
507fn value_text(value: &Value) -> String {
508    value.as_str().map_or_else(|| value.to_string(), ToString::to_string)
509}
510
511fn i64_saturating(value: u64) -> i64 {
512    i64::try_from(value).unwrap_or(i64::MAX)
513}
514
515#[cfg(test)]
516mod tests {
517    use std::io::Write as _;
518
519    use super::*;
520
521    #[test]
522    fn aggregates_correlated_trace_without_reading_sensitive_arguments() -> Result<()> {
523        let temp = tempfile::tempdir()?;
524        let base = temp.path().join("usage.jsonl");
525        let path = temp.path().join("usage.jsonl.2026-08-25");
526        let mut file = File::create(path)?;
527        writeln!(
528            file,
529            "{}",
530            serde_json::json!({
531                "timestamp": "2026-08-25T12:00:00Z",
532                "target": "winx::usage",
533                "fields": {
534                    "event":"tool_call", "usage_schema":1,
535                    "tool":"BashCommand", "request_id":"req",
536                    "result_status":"completed", "duration_ms":10, "response_bytes":20,
537                    "command_kind":"rust_toolchain", "build":"0.2.343+abc"
538                }
539            })
540        )?;
541        writeln!(
542            file,
543            "{}",
544            serde_json::json!({
545                "timestamp": "2026-08-25T12:00:00Z",
546                "target": "winx::usage",
547                "fields": {
548                    "event":"http_request", "request_id":"req", "status":200,
549                    "duration_ms":12
550                }
551            })
552        )?;
553        let report = usage_report(Some(&base), 100, 0, None)?;
554        assert_eq!(report.tool_calls, 1);
555        assert_eq!(report.http_requests, 1);
556        assert_eq!(report.correlated_requests, 1);
557        assert_eq!(report.http_overhead.p95_ms, 2);
558        assert_eq!(report.command_kinds["rust_toolchain"], 1);
559        Ok(())
560    }
561
562    #[test]
563    fn legacy_missing_recovery_fields_are_unknown_not_violations() -> Result<()> {
564        let temp = tempfile::tempdir()?;
565        let base = temp.path().join("usage.jsonl");
566        let mut file = File::create(&base)?;
567        writeln!(
568            file,
569            "{}",
570            serde_json::json!({
571                "timestamp": "2026-08-25T12:00:00Z",
572                "target": "winx::usage",
573                "fields": {
574                    "event": "tool_call",
575                    "tool": "MultiFileEdit",
576                    "request_id": "legacy",
577                    "result_status": "conflict",
578                    "error_code": "search_block_not_found"
579                }
580            })
581        )?;
582
583        let report = usage_report(Some(&base), 100, 0, None)?;
584        assert!(report.trace_audit.healthy);
585        assert_eq!(report.trace_audit.audited_tool_events, 0);
586        assert_eq!(report.trace_audit.legacy_unverifiable_tool_events, 1);
587        assert!(report.trace_audit.violations.is_empty());
588        Ok(())
589    }
590}
winx_code_agent/report.rs

winx_code_agent/
report.rs
