1use std::collections::BTreeMap;
5
6use serde::Serialize;
7use serde_json::{Map, Value};
8
9use super::lifecycle::{LifecycleAccumulator, WorkflowPhaseUsage};
10use super::{increment_nonempty, latency, number, string, LatencySummary};
11
12#[derive(Debug, Default, Serialize)]
13#[serde(rename_all = "camelCase")]
14pub struct WorkflowUsage {
15 pub calls: u64,
16 pub operations: Vec<WorkflowOperationUsage>,
17}
18
19#[derive(Debug, Default, Serialize)]
20#[serde(rename_all = "camelCase")]
21pub struct WorkflowOperationUsage {
22 pub tool: &'static str,
23 pub action: &'static str,
24 pub backend: &'static str,
25 pub calls: u64,
26 pub error_results: u64,
27 pub response_bytes: u64,
28 pub latency: LatencySummary,
29 pub requested_wait_ms: LatencySummary,
30 pub wait_kinds: BTreeMap<String, u64>,
31 pub wait_outcomes: BTreeMap<String, u64>,
32 pub wait_outcome_unknown: u64,
33 pub receipt_observations: u64,
35 pub prompt_delivery: BTreeMap<String, u64>,
36 pub prompt_delivery_unknown: u64,
37 pub prompt_outcome: BTreeMap<String, u64>,
38 pub prompt_outcome_unknown: u64,
39 pub prompt_activity_observed_true: u64,
40 pub prompt_activity_observed_false: u64,
41 pub prompt_activity_observed_unknown: u64,
42 pub agent_submission_accepted: u64,
43 pub agent_input_sent_true: u64,
44 pub agent_input_sent_false: u64,
45 pub agent_input_may_have_been_sent_true: u64,
46 pub agent_input_may_have_been_sent_false: u64,
47 pub phases: BTreeMap<String, WorkflowPhaseUsage>,
48 pub phase_abandoned_count: u64,
49 pub phase_trace_suppressed_count: u64,
50}
51
52#[derive(Default)]
53pub(super) struct WorkflowAccumulator {
54 calls: u64,
55 operations: BTreeMap<(&'static str, &'static str, &'static str), OperationAccumulator>,
56}
57
58#[derive(Default)]
59struct OperationAccumulator {
60 usage: WorkflowOperationUsage,
61 durations: Vec<u64>,
62 requested_wait: Vec<u64>,
63 lifecycle: LifecycleAccumulator,
64}
65
66impl WorkflowAccumulator {
67 pub fn record(&mut self, fields: &Map<String, Value>) {
68 let tool = match string(fields, "tool") {
69 "Agent" => "Agent",
70 "Execution" => "Execution",
71 _ => return,
72 };
73 let classified = fields.get("workflow_schema").and_then(Value::as_u64) == Some(1);
74 let action = if classified { action(tool, string(fields, "action")) } else { "unknown" };
75 let backend =
76 if classified { backend(string(fields, "workflow_backend")) } else { "unknown" };
77 self.calls = self.calls.saturating_add(1);
78 let operation = self.operations.entry((tool, action, backend)).or_insert_with(|| {
79 OperationAccumulator {
80 usage: WorkflowOperationUsage { tool, action, backend, ..Default::default() },
81 ..Default::default()
82 }
83 });
84 operation.usage.calls = operation.usage.calls.saturating_add(1);
85 operation.usage.error_results = operation
86 .usage
87 .error_results
88 .saturating_add(u64::from(!string(fields, "error_code").is_empty()));
89 operation.usage.response_bytes =
90 operation.usage.response_bytes.saturating_add(number(fields, "response_bytes"));
91 if let Some(duration) = fields.get("duration_ms").and_then(Value::as_u64) {
92 operation.durations.push(duration);
93 }
94 if !classified {
95 return;
96 }
97 if let Some(wait) =
98 fields.get("requested_wait_ms").and_then(Value::as_u64).filter(|wait| *wait <= 30_000)
99 {
100 operation.requested_wait.push(wait);
101 }
102 operation.lifecycle.record(fields);
103 operation.usage.record_wait(fields);
104 operation.usage.record_receipt(fields);
105 operation.usage.record_agent_input(fields);
106 }
107
108 pub fn finish(self) -> WorkflowUsage {
109 let operations = self
110 .operations
111 .into_values()
112 .map(|operation| {
113 let mut usage = operation.usage;
114 usage.latency = latency(operation.durations);
115 usage.requested_wait_ms = latency(operation.requested_wait);
116 usage.phase_abandoned_count = operation.lifecycle.abandoned;
117 usage.phase_trace_suppressed_count = operation.lifecycle.suppressed;
118 usage.phases = operation.lifecycle.finish();
119 usage
120 })
121 .collect();
122 WorkflowUsage { calls: self.calls, operations }
123 }
124}
125
126impl WorkflowOperationUsage {
127 fn record_wait(&mut self, fields: &Map<String, Value>) {
128 let kind = match string(fields, "workflow_wait_kind") {
129 "catalog" => "catalog",
130 "receipt" => "receipt",
131 "herdr" => "herdr",
132 _ => return,
133 };
134 increment_nonempty(&mut self.wait_kinds, kind);
135 if !matches!(kind, "catalog" | "herdr") {
136 return;
137 }
138 let outcome = if kind == "catalog" {
139 match string(fields, "workflow_wait_outcome") {
140 "changed" => Some("changed"),
141 "timed_out" => Some("timed_out"),
142 "resync_required" => Some("resync_required"),
143 _ => None,
144 }
145 } else {
146 None
147 };
148 match outcome {
149 Some(outcome) => increment_nonempty(&mut self.wait_outcomes, outcome),
150 None => self.wait_outcome_unknown = self.wait_outcome_unknown.saturating_add(1),
151 }
152 }
153
154 fn record_receipt(&mut self, fields: &Map<String, Value>) {
155 if string(fields, "workflow_wait_kind") != "receipt"
156 || fields.get("prompt_receipt_present").and_then(Value::as_bool) != Some(true)
157 {
158 return;
159 }
160 self.receipt_observations = self.receipt_observations.saturating_add(1);
161 match string(fields, "prompt_delivery") {
162 delivery @ ("delivered" | "uncertain") => {
163 increment_nonempty(&mut self.prompt_delivery, delivery);
164 }
165 _ => self.prompt_delivery_unknown = self.prompt_delivery_unknown.saturating_add(1),
166 }
167 match string(fields, "prompt_outcome") {
168 outcome @ ("waiting" | "awaiting_input" | "awaiting_approval" | "exited"
169 | "timed_out" | "generation_changed") => {
170 increment_nonempty(&mut self.prompt_outcome, outcome);
171 }
172 _ => self.prompt_outcome_unknown = self.prompt_outcome_unknown.saturating_add(1),
173 }
174 match fields.get("prompt_activity_observed").and_then(Value::as_bool) {
175 Some(true) => {
176 self.prompt_activity_observed_true =
177 self.prompt_activity_observed_true.saturating_add(1);
178 }
179 Some(false) => {
180 self.prompt_activity_observed_false =
181 self.prompt_activity_observed_false.saturating_add(1);
182 }
183 None => {
184 self.prompt_activity_observed_unknown =
185 self.prompt_activity_observed_unknown.saturating_add(1);
186 }
187 }
188 }
189
190 fn record_agent_input(&mut self, fields: &Map<String, Value>) {
191 if self.tool != "Agent" {
192 return;
193 }
194 if self.backend == "herdr"
195 && self.action == "prompt"
196 && string(fields, "agent_submission_status") == "accepted"
197 {
198 self.agent_submission_accepted = self.agent_submission_accepted.saturating_add(1);
199 }
200 match fields.get("agent_input_sent").and_then(Value::as_bool) {
201 Some(true) => self.agent_input_sent_true = self.agent_input_sent_true.saturating_add(1),
202 Some(false) => {
203 self.agent_input_sent_false = self.agent_input_sent_false.saturating_add(1);
204 }
205 None => {}
206 }
207 match fields.get("agent_input_may_have_been_sent").and_then(Value::as_bool) {
208 Some(true) => {
209 self.agent_input_may_have_been_sent_true =
210 self.agent_input_may_have_been_sent_true.saturating_add(1);
211 }
212 Some(false) => {
213 self.agent_input_may_have_been_sent_false =
214 self.agent_input_may_have_been_sent_false.saturating_add(1);
215 }
216 None => {}
217 }
218 }
219}
220
221fn action(tool: &str, value: &str) -> &'static str {
222 match value {
223 "list" => "list",
224 "get" => "get",
225 "read" => "read",
226 "observe" if tool == "Execution" => "observe",
227 "explain" => "explain",
228 "wait" => "wait",
229 "prompt" => "prompt",
230 _ => "unknown",
231 }
232}
233
234fn backend(value: &str) -> &'static str {
235 match value {
236 "native" => "native",
237 "herdr" => "herdr",
238 _ => "unknown",
239 }
240}
241
242#[cfg(test)]
243mod observe_tests {
244 use anyhow::Context;
245 use serde_json::json;
246
247 use super::*;
248
249 #[test]
250 fn observe_report_preserves_action_and_counts_partial_results() -> anyhow::Result<()> {
251 let mut accumulator = WorkflowAccumulator::default();
252 for (status, error_code, bytes) in
253 [("completed", "", 512), ("completed_with_issues", "execution_observation_issues", 768)]
254 {
255 let fields = json!({
256 "usage_schema":1,"workflow_schema":1,"tool":"Execution","action":"observe",
257 "workflow_backend":"native","status":status,"error_code":error_code,
258 "duration_ms":12,"response_bytes":bytes,
259 "data":{"cursor":"SECRET-CURSOR","items":[{
260 "execution_id":"SECRET-EXECUTION","page":{"text":"SECRET-OUTPUT"},
261 "errors":[{"message":"SECRET-ERROR"}]
262 }]}
263 });
264 accumulator.record(fields.as_object().context("usage fields")?);
265 }
266 let report = accumulator.finish();
267 assert_eq!(report.calls, 2);
268 assert_eq!(report.operations.len(), 1);
269 let operation = report.operations.first().context("observe operation")?;
270 assert_eq!(operation.tool, "Execution");
271 assert_eq!(operation.action, "observe");
272 assert_eq!(operation.backend, "native");
273 assert_eq!(operation.calls, 2);
274 assert_eq!(operation.error_results, 1);
275 assert_eq!(operation.response_bytes, 1280);
276 assert_eq!(
277 serde_json::to_value(&operation.requested_wait_ms)?,
278 serde_json::to_value(latency(Vec::new()))?
279 );
280 assert!(operation.wait_kinds.is_empty());
281 assert!(operation.wait_outcomes.is_empty());
282 assert_eq!(operation.wait_outcome_unknown, 0);
283 assert_eq!(operation.receipt_observations, 0);
284 assert!(operation.prompt_delivery.is_empty());
285 assert!(operation.prompt_outcome.is_empty());
286 assert_eq!(operation.agent_submission_accepted, 0);
287 assert_eq!(operation.agent_input_sent_true, 0);
288 assert_eq!(operation.agent_input_may_have_been_sent_true, 0);
289 assert!(!serde_json::to_string(&report)?.contains("SECRET-"));
290 Ok(())
291 }
292
293 #[test]
294 fn observe_report_keeps_agent_and_unversioned_actions_unknown() -> anyhow::Result<()> {
295 for (tool, schema, requested_backend, expected_backend) in [
296 ("Agent", Some(1), "native", "native"),
297 ("Agent", Some(1), "herdr", "herdr"),
298 ("Execution", None, "native", "unknown"),
299 ] {
300 let fields = json!({
301 "usage_schema":1,"workflow_schema":schema,"tool":tool,"action":"observe",
302 "workflow_backend":requested_backend,"duration_ms":9,"response_bytes":42
303 });
304 let mut accumulator = WorkflowAccumulator::default();
305 accumulator.record(fields.as_object().context("usage fields")?);
306 let report = accumulator.finish();
307 assert_eq!(report.calls, 1);
308 assert_eq!(report.operations.len(), 1);
309 let operation = report.operations.first().context("unknown operation")?;
310 assert_eq!(operation.tool, tool);
311 assert_eq!(operation.action, "unknown", "{tool} {schema:?}");
312 assert_eq!(operation.backend, expected_backend);
313 assert_eq!(operation.calls, 1);
314 assert_eq!(operation.response_bytes, 42);
315 }
316 Ok(())
317 }
318}