winx_code_agent/report/
observations.rs

1//! Call-level request and response counts, never paths or output contents.
2//! Repeated reads and observations count repeatedly. Unknown means unavailable
3//! or not applicable, never a measured zero or an execution failure.
4
5use std::collections::BTreeMap;
6
7use serde::Serialize;
8use serde_json::{Map, Value};
9
10use super::{increment_nonempty, latency, number, string, LatencySummary};
11
12#[derive(Debug, Default, Serialize)]
13#[serde(rename_all = "camelCase")]
14pub struct ReadUsage {
15    pub calls: u64,
16    pub classified_calls: u64,
17    pub unknown_calls: u64,
18    pub whole_file_requests: u64,
19    pub bounded_range_requests: u64,
20    pub open_ended_range_requests: u64,
21    pub response_bytes: u64,
22}
23
24#[derive(Debug, Default, Serialize)]
25#[serde(rename_all = "camelCase")]
26pub struct ObserveUsage {
27    pub calls: u64,
28    pub request_counts_known: u64,
29    pub request_counts_unknown: u64,
30    pub requested_items: u64,
31    pub result_counts_known: u64,
32    pub result_counts_unknown: u64,
33    pub returned_items: u64,
34    pub items_with_issues: u64,
35    /// Calls reporting `completed_with_issues`, not failed executions.
36    pub partial_results: u64,
37}
38
39/// Flattened into `ExecutionUsage`; its denominator is only `BashCommand` calls.
40#[derive(Debug, Default, Serialize)]
41#[serde(rename_all = "camelCase")]
42pub struct EffectiveWaitUsage {
43    pub effective_wait_observations: u64,
44    pub effective_wait_policies: BTreeMap<String, u64>,
45    pub effective_wait_policy_unknown: u64,
46    pub effective_wait_modes: BTreeMap<String, u64>,
47    pub effective_wait_mode_unknown: u64,
48    pub effective_wait_ms: LatencySummary,
49}
50
51#[derive(Default)]
52pub(super) struct ObservationAccumulator {
53    reads: ReadUsage,
54    observations: ObserveUsage,
55    effective_wait: EffectiveWaitUsage,
56    effective_wait_samples: Vec<u64>,
57}
58
59impl ObservationAccumulator {
60    pub fn record(&mut self, fields: &Map<String, Value>) {
61        match string(fields, "tool") {
62            "ReadFiles" => self.record_reads(fields),
63            "Execution"
64                if optional_number(fields, "workflow_schema") == Some(1)
65                    && string(fields, "action") == "observe" =>
66            {
67                self.record_observe(fields);
68            }
69            "BashCommand" => self.record_effective_wait(fields),
70            _ => {}
71        }
72    }
73
74    pub fn finish(mut self) -> (ReadUsage, ObserveUsage, EffectiveWaitUsage) {
75        self.effective_wait.effective_wait_ms = latency(self.effective_wait_samples);
76        (self.reads, self.observations, self.effective_wait)
77    }
78
79    fn record_reads(&mut self, fields: &Map<String, Value>) {
80        let reads = &mut self.reads;
81        reads.calls = reads.calls.saturating_add(1);
82        reads.response_bytes =
83            reads.response_bytes.saturating_add(number(fields, "response_bytes"));
84        let counts = (
85            optional_number(fields, "read_whole_file_requests"),
86            optional_number(fields, "read_bounded_range_requests"),
87            optional_number(fields, "read_open_ended_range_requests"),
88        );
89        if let (Some(whole), Some(bounded), Some(open)) = counts {
90            if optional_number(fields, "read_request_schema") == Some(1) {
91                reads.classified_calls = reads.classified_calls.saturating_add(1);
92                reads.whole_file_requests = reads.whole_file_requests.saturating_add(whole);
93                reads.bounded_range_requests = reads.bounded_range_requests.saturating_add(bounded);
94                reads.open_ended_range_requests =
95                    reads.open_ended_range_requests.saturating_add(open);
96                return;
97            }
98        }
99        reads.unknown_calls = reads.unknown_calls.saturating_add(1);
100    }
101
102    fn record_observe(&mut self, fields: &Map<String, Value>) {
103        let observed = &mut self.observations;
104        observed.calls = observed.calls.saturating_add(1);
105        observed.partial_results = observed
106            .partial_results
107            .saturating_add(u64::from(string(fields, "result_status") == "completed_with_issues"));
108        let classified = optional_number(fields, "observe_schema") == Some(1);
109        // Old batch_items=0 did not describe real observe batches. Never use it
110        // as a fallback for the independently versioned request counters.
111        match optional_number(fields, "observe_requested_items").filter(|_| classified) {
112            Some(count) => {
113                observed.request_counts_known = observed.request_counts_known.saturating_add(1);
114                observed.requested_items = observed.requested_items.saturating_add(count);
115            }
116            None => {
117                observed.request_counts_unknown = observed.request_counts_unknown.saturating_add(1);
118            }
119        }
120        let returned = optional_number(fields, "observe_returned_items").filter(|_| classified);
121        let issues = optional_number(fields, "observe_items_with_issues").filter(|_| classified);
122        if let (Some(returned), Some(issues)) = (returned, issues) {
123            if issues <= returned {
124                observed.result_counts_known = observed.result_counts_known.saturating_add(1);
125                observed.returned_items = observed.returned_items.saturating_add(returned);
126                observed.items_with_issues = observed.items_with_issues.saturating_add(issues);
127                return;
128            }
129        }
130        observed.result_counts_unknown = observed.result_counts_unknown.saturating_add(1);
131    }
132
133    fn record_effective_wait(&mut self, fields: &Map<String, Value>) {
134        let wait = &mut self.effective_wait;
135        wait.effective_wait_observations = wait.effective_wait_observations.saturating_add(1);
136        let policy = match string(fields, "effective_wait_policy") {
137            policy @ ("adaptive" | "until_complete" | "return_early") => Some(policy),
138            _ => None,
139        };
140        let mode = match string(fields, "effective_wait_mode") {
141            mode @ ("synchronous" | "task_inline" | "task") => Some(mode),
142            _ => None,
143        };
144        let milliseconds = optional_number(fields, "effective_wait_ms")
145            .filter(|milliseconds| *milliseconds <= 120_000)
146            .filter(|milliseconds| mode != Some("task") || *milliseconds == 0);
147        if let (Some(policy), Some(mode), Some(milliseconds)) = (policy, mode, milliseconds) {
148            increment_nonempty(&mut wait.effective_wait_policies, policy);
149            increment_nonempty(&mut wait.effective_wait_modes, mode);
150            self.effective_wait_samples.push(milliseconds);
151        } else {
152            // Includes legacy calls and actions without applicable metadata;
153            // these unknown observations are not failed executions.
154            wait.effective_wait_policy_unknown =
155                wait.effective_wait_policy_unknown.saturating_add(1);
156            wait.effective_wait_mode_unknown = wait.effective_wait_mode_unknown.saturating_add(1);
157        }
158    }
159}
160
161fn optional_number(fields: &Map<String, Value>, key: &str) -> Option<u64> {
162    fields.get(key).and_then(Value::as_u64)
163}
winx_code_agent/report/observations.rs

winx_code_agent/report/
observations.rs
