dynamo_data_gen/request_trace/
load.rs

1// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Source-format records and JSONL/gz loader.
5//!
6//! Local subset of Dynamo request trace rows consumed by replay.
7
8use std::collections::HashSet;
9use std::fs::File;
10use std::io::BufRead;
11use std::io::BufReader;
12use std::io::Read;
13use std::path::Path;
14use std::path::PathBuf;
15
16use anyhow::Context;
17use anyhow::Result;
18use anyhow::anyhow;
19use anyhow::bail;
20use flate2::read::MultiGzDecoder;
21use serde::de::IgnoredAny;
22use serde::{Deserialize, Serialize};
23
24use crate::{AgenticDependencyRelation, AgenticDependencyTrigger};
25
26#[derive(Debug, Clone, Deserialize)]
27pub(crate) struct RequestTraceRecord {
28    pub(crate) schema: TraceSchema,
29    pub(crate) event_type: String,
30    pub(crate) event_time_unix_ms: u64,
31    #[serde(default)]
32    pub(crate) agent_context: Option<AgentContextFields>,
33    #[serde(default)]
34    pub(crate) request: Option<RequestTraceRequestMetrics>,
35    #[serde(default)]
36    pub(crate) tool: Option<RequestTraceToolEventMetrics>,
37}
38
39#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
40pub(crate) enum TraceSchema {
41    #[serde(rename = "dynamo.request.trace.v1")]
42    RequestV1,
43}
44
45#[derive(Debug, Clone, Default, Deserialize)]
46pub(crate) struct AgentContextFields {
47    pub(crate) session_id: String,
48    #[serde(default)]
49    pub(crate) parent_session_id: Option<String>,
50}
51
52#[derive(Debug, Clone, Default, Deserialize)]
53pub(crate) struct RequestTraceRequestMetrics {
54    pub(crate) request_id: String,
55    #[serde(default)]
56    pub(crate) x_request_id: Option<String>,
57    #[serde(default)]
58    pub(crate) model: Option<String>,
59    #[serde(default)]
60    pub(crate) input_tokens: Option<u64>,
61    #[serde(default)]
62    pub(crate) output_tokens: Option<u64>,
63    #[serde(default)]
64    pub(crate) cached_tokens: Option<u64>,
65    #[serde(default)]
66    pub(crate) request_received_ms: Option<u64>,
67    #[serde(default)]
68    pub(crate) prefill_wait_time_ms: Option<f64>,
69    #[serde(default)]
70    pub(crate) prefill_time_ms: Option<f64>,
71    #[serde(default)]
72    pub(crate) ttft_ms: Option<f64>,
73    #[serde(default)]
74    pub(crate) total_time_ms: Option<f64>,
75    #[serde(default)]
76    pub(crate) avg_itl_ms: Option<f64>,
77    #[serde(default)]
78    pub(crate) kv_hit_rate: Option<f64>,
79    #[serde(default)]
80    pub(crate) kv_transfer_estimated_latency_ms: Option<f64>,
81    #[serde(default)]
82    pub(crate) queue_depth: Option<u64>,
83    #[serde(default)]
84    pub(crate) worker: Option<RequestTraceWorkerInfo>,
85    #[serde(default)]
86    pub(crate) replay: Option<RequestTraceReplayMetrics>,
87    #[serde(default)]
88    pub(crate) finish_reason_metadata: Option<RequestTraceFinishReasonMetadata>,
89}
90
91#[derive(Debug, Clone, Deserialize, Serialize)]
92pub(crate) struct RequestTraceWorkerInfo {
93    #[serde(default, skip_serializing_if = "Option::is_none")]
94    pub(crate) prefill_worker_id: Option<u64>,
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub(crate) prefill_dp_rank: Option<u32>,
97    #[serde(default, skip_serializing_if = "Option::is_none")]
98    pub(crate) decode_worker_id: Option<u64>,
99    #[serde(default, skip_serializing_if = "Option::is_none")]
100    pub(crate) decode_dp_rank: Option<u32>,
101}
102
103#[derive(Debug, Clone, Deserialize, Serialize)]
104pub(crate) struct RequestTraceFinishReasonMetadata {
105    #[serde(default, skip_serializing_if = "Option::is_none")]
106    pub(crate) finish_reason: Option<String>,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub(crate) backend_finish_reason: Option<String>,
109    #[serde(default, skip_serializing_if = "Option::is_none")]
110    pub(crate) stop_reason: Option<serde_json::Value>,
111    #[serde(default, skip_serializing_if = "Vec::is_empty")]
112    pub(crate) tool_calls: Vec<RequestTraceToolCall>,
113}
114
115#[derive(Debug, Clone, Deserialize, Serialize)]
116pub(crate) struct RequestTraceToolCall {
117    pub(crate) choice_index: u32,
118    pub(crate) tool_call_index: u32,
119    #[serde(default, skip_serializing_if = "Option::is_none")]
120    pub(crate) id: Option<String>,
121    #[serde(default, skip_serializing_if = "Option::is_none")]
122    pub(crate) name: Option<String>,
123}
124
125#[derive(Debug, Clone, Deserialize)]
126pub(crate) struct RequestTraceReplayMetrics {
127    pub(crate) trace_block_size: usize,
128    pub(crate) input_length: usize,
129    pub(crate) input_sequence_hashes: Vec<u64>,
130    /// Exporter-only edges to earlier requests. Live traces omit them, and agentic lowering then
131    /// infers cross-session edges from timestamps.
132    #[serde(default)]
133    pub(crate) dependencies: Vec<ReplayDependencyRecord>,
134}
135
136/// An edge from a request to an earlier request it could not start before.
137#[derive(Debug, Clone, Deserialize)]
138pub(crate) struct ReplayDependencyRecord {
139    pub(crate) request_id: String,
140    pub(crate) relation: AgenticDependencyRelation,
141    pub(crate) trigger: AgenticDependencyTrigger,
142}
143
144#[derive(Debug, Clone, Deserialize)]
145pub(crate) struct RequestTraceToolEventMetrics {
146    pub(crate) tool_call_id: String,
147    pub(crate) tool_class: String,
148    #[serde(default)]
149    pub(crate) claude: Option<ClaudeToolReplayMetrics>,
150    #[serde(default)]
151    pub(crate) started_at_unix_ms: Option<u64>,
152    #[serde(default)]
153    pub(crate) ended_at_unix_ms: Option<u64>,
154    #[serde(default)]
155    pub(crate) duration_ms: Option<f64>,
156    #[serde(default)]
157    pub(crate) status: Option<String>,
158    #[serde(default)]
159    pub(crate) output_bytes: Option<u64>,
160    #[serde(default)]
161    pub(crate) output_tokens: Option<u64>,
162    #[serde(default)]
163    pub(crate) error_type: Option<String>,
164}
165
166/// Claude-only evidence used to disambiguate an offline replay DAG.
167///
168/// A completed session reveals the future request that consumed a tool result;
169/// live ZMQ tool producers do not have that information. Missing metadata is
170/// expected and leaves agentic lowering on its timestamp-based fallback.
171#[derive(Debug, Clone, Deserialize)]
172pub(crate) struct ClaudeToolReplayMetrics {
173    /// Request that emitted the tool call.
174    pub(crate) source_request_id: String,
175    /// Later request that consumed the terminal result.
176    #[serde(default)]
177    pub(crate) consumer_request_id: Option<String>,
178    /// Child agent session launched by the tool, when present in the export.
179    #[serde(default)]
180    pub(crate) child_session_id: Option<String>,
181    /// Whether parent execution blocked or continued in the background.
182    pub(crate) execution_mode: String,
183}
184
185#[derive(Debug, Deserialize)]
186#[serde(untagged)]
187enum JsonLineEnvelope {
188    Object(TraceRecordEnvelope),
189    Other(IgnoredAny),
190}
191
192#[derive(Debug, Deserialize)]
193struct TraceRecordEnvelope {
194    #[serde(default)]
195    event_type: Option<String>,
196    #[serde(default)]
197    event: Option<TraceEventEnvelope>,
198}
199
200#[derive(Debug, Deserialize)]
201#[serde(untagged)]
202enum TraceEventEnvelope {
203    Object(TraceEventFields),
204    Other(IgnoredAny),
205}
206
207#[derive(Debug, Deserialize)]
208struct TraceEventFields {
209    #[serde(default)]
210    event_type: Option<String>,
211}
212
213#[derive(Debug, Deserialize)]
214struct WrappedTraceRecord {
215    event: RequestTraceRecord,
216}
217
218#[derive(Debug, Clone)]
219pub struct RequestEntry {
220    pub(crate) start_ms: i64,
221    pub(crate) end_ms: i64,
222    pub(crate) agent_context: Option<AgentContextFields>,
223    pub(crate) request: RequestTraceRequestMetrics,
224    pub(crate) replay: RequestTraceReplayMetrics,
225}
226
227#[derive(Debug, Clone)]
228pub struct ToolEntry {
229    pub(crate) session_id: String,
230    pub(crate) start_ms: i64,
231    pub(crate) end_ms: i64,
232    pub(crate) tool_call_id: String,
233    pub(crate) tool_class: String,
234    pub(crate) claude: Option<ClaudeToolReplayMetrics>,
235    pub(crate) status: String,
236    pub(crate) duration_ms: f64,
237    pub(crate) output_bytes: Option<u64>,
238    pub(crate) output_tokens: Option<u64>,
239    pub(crate) error_type: Option<String>,
240}
241
242#[derive(Debug, Default)]
243pub struct LoadedAgentTrace {
244    pub requests: Vec<RequestEntry>,
245    pub tools: Vec<ToolEntry>,
246}
247
248#[derive(Debug, Clone, Copy, PartialEq, Eq)]
249pub enum RequestTraceMode {
250    Standard,
251    Agentic,
252}
253
254impl LoadedAgentTrace {
255    pub fn mode(&self) -> Result<RequestTraceMode> {
256        if self.requests.is_empty() {
257            bail!("Dynamo request trace contains no requests");
258        }
259        let contextual_requests = self
260            .requests
261            .iter()
262            .filter(|request| request.agent_context.is_some())
263            .count();
264        match contextual_requests {
265            0 => Ok(RequestTraceMode::Standard),
266            count if count == self.requests.len() => Ok(RequestTraceMode::Agentic),
267            _ => bail!("Dynamo request trace cannot mix requests with and without agent_context"),
268        }
269    }
270
271    pub fn ensure_agentic_compatible(&self) -> Result<()> {
272        if self
273            .requests
274            .iter()
275            .any(|request| request.agent_context.is_none())
276        {
277            bail!("agentic lowering requires agent_context on every request");
278        }
279        Ok(())
280    }
281}
282
283/// `request_payload` records are skipped; replay consumes `request_end` and
284/// terminal tool events only. Errors if no `request_end` rows were found.
285pub fn load_request_trace_records(paths: &[PathBuf]) -> Result<LoadedAgentTrace> {
286    let mut loaded = LoadedAgentTrace::default();
287    let mut request_ids = HashSet::new();
288
289    for path in paths {
290        let reader = open_trace_reader(path)?;
291        for (line_index, line) in reader.lines().enumerate() {
292            let line = line
293                .with_context(|| format!("failed to read {}:{}", path.display(), line_index + 1))?;
294            if line.trim().is_empty() {
295                continue;
296            }
297            let Some(record) = parse_trace_record(&line).with_context(|| {
298                format!("failed to parse {}:{}", path.display(), line_index + 1)
299            })?
300            else {
301                continue;
302            };
303            let _schema = record.schema;
304            if record.event_type == "request_payload" {
305                continue;
306            }
307            if !matches!(
308                record.event_type.as_str(),
309                "request_end" | "tool_start" | "tool_end" | "tool_error"
310            ) {
311                bail!(
312                    "request trace schema only supports request_end/tool_* and request_payload events, got {} at {}:{}",
313                    record.event_type,
314                    path.display(),
315                    line_index + 1
316                );
317            }
318            if record.event_type == "request_end" {
319                let entry = request_entry(record).with_context(|| {
320                    format!(
321                        "invalid request_end at {}:{}",
322                        path.display(),
323                        line_index + 1
324                    )
325                })?;
326                if !request_ids.insert(entry.request.request_id.clone()) {
327                    bail!(
328                        "duplicate request_id {} at {}:{}",
329                        entry.request.request_id,
330                        path.display(),
331                        line_index + 1
332                    );
333                }
334                loaded.requests.push(entry);
335            } else if matches!(record.event_type.as_str(), "tool_end" | "tool_error") {
336                let terminal_event = record.event_type.clone();
337                if let Some(tool) = tool_entry(record, terminal_event) {
338                    loaded.tools.push(tool);
339                }
340            }
341        }
342    }
343
344    if loaded.requests.is_empty() {
345        bail!("no request_end records with replay fields found");
346    }
347
348    Ok(loaded)
349}
350
351fn open_trace_reader(path: &Path) -> Result<Box<dyn BufRead>> {
352    let file = File::open(path).with_context(|| format!("failed to open {}", path.display()))?;
353    let reader: Box<dyn Read> = if path.extension().and_then(|ext| ext.to_str()) == Some("gz") {
354        Box::new(MultiGzDecoder::new(file))
355    } else {
356        Box::new(file)
357    };
358    Ok(Box::new(BufReader::new(reader)))
359}
360
361fn parse_trace_record(line: &str) -> Result<Option<RequestTraceRecord>> {
362    let envelope = match serde_json::from_str::<JsonLineEnvelope>(line)? {
363        JsonLineEnvelope::Object(envelope) => envelope,
364        JsonLineEnvelope::Other(_) => return Ok(None),
365    };
366
367    match envelope.event {
368        Some(TraceEventEnvelope::Object(event)) => {
369            if event.event_type.as_deref() == Some("request_payload") {
370                return Ok(None);
371            }
372            Ok(Some(
373                serde_json::from_str::<WrappedTraceRecord>(line)?.event,
374            ))
375        }
376        Some(TraceEventEnvelope::Other(_)) => Ok(None),
377        None => {
378            if envelope.event_type.as_deref() == Some("request_payload") {
379                return Ok(None);
380            }
381            Ok(Some(serde_json::from_str(line)?))
382        }
383    }
384}
385
386fn request_entry(record: RequestTraceRecord) -> Result<RequestEntry> {
387    let request = record
388        .request
389        .ok_or_else(|| anyhow!("request_end record is missing request payload"))?;
390    let replay = request
391        .replay
392        .clone()
393        .ok_or_else(|| anyhow!("request payload is missing replay metrics"))?;
394    if replay.trace_block_size == 0 {
395        bail!("request replay trace_block_size must be greater than 0");
396    }
397    let expected_hashes = replay.input_length.div_ceil(replay.trace_block_size);
398    if replay.input_sequence_hashes.len() != expected_hashes {
399        bail!(
400            "input_length {} with trace_block_size {} requires exactly {} replay hashes, got {}",
401            replay.input_length,
402            replay.trace_block_size,
403            expected_hashes,
404            replay.input_sequence_hashes.len()
405        );
406    }
407    if request.request_received_ms.is_none() {
408        bail!("request trace is missing request_received_ms");
409    }
410    if request.output_tokens.is_none() {
411        bail!("request trace is missing output_tokens");
412    }
413
414    let (start_ms, end_ms) = request_times(record.event_time_unix_ms, &request);
415    Ok(RequestEntry {
416        start_ms,
417        end_ms,
418        agent_context: record.agent_context,
419        request,
420        replay,
421    })
422}
423
424fn tool_entry(record: RequestTraceRecord, terminal_event: String) -> Option<ToolEntry> {
425    let context = record.agent_context?;
426    let tool = record.tool?;
427    let end_ms = tool
428        .ended_at_unix_ms
429        .map(saturating_i64)
430        .unwrap_or_else(|| saturating_i64(record.event_time_unix_ms));
431    let start_ms = tool
432        .started_at_unix_ms
433        .map(saturating_i64)
434        .or_else(|| {
435            tool.duration_ms
436                .map(|duration_ms| end_ms.saturating_sub(duration_ms.max(0.0).round() as i64))
437        })
438        .unwrap_or(end_ms);
439    if end_ms < start_ms {
440        return None;
441    }
442    let duration_ms = tool
443        .duration_ms
444        .unwrap_or_else(|| (end_ms - start_ms).max(0) as f64);
445    let status = tool.status.unwrap_or_else(|| {
446        if terminal_event == "tool_error" {
447            "error".to_string()
448        } else {
449            "succeeded".to_string()
450        }
451    });
452    Some(ToolEntry {
453        session_id: context.session_id,
454        start_ms,
455        end_ms,
456        tool_call_id: tool.tool_call_id,
457        tool_class: tool.tool_class,
458        claude: tool.claude,
459        status,
460        duration_ms,
461        output_bytes: tool.output_bytes,
462        output_tokens: tool.output_tokens,
463        error_type: tool.error_type,
464    })
465}
466
467pub(crate) fn request_times(
468    event_time_unix_ms: u64,
469    request: &RequestTraceRequestMetrics,
470) -> (i64, i64) {
471    let total_ms = request
472        .total_time_ms
473        .map(|value| value.max(0.0).round() as u64)
474        .unwrap_or_else(|| {
475            event_time_unix_ms
476                .saturating_sub(request.request_received_ms.unwrap_or(event_time_unix_ms))
477        });
478    let end_ms = request
479        .request_received_ms
480        .map(|start| start.saturating_add(total_ms))
481        .unwrap_or(event_time_unix_ms);
482    let start_ms = request
483        .request_received_ms
484        .unwrap_or_else(|| event_time_unix_ms.saturating_sub(total_ms));
485    (saturating_i64(start_ms), saturating_i64(end_ms))
486}
487
488pub(crate) fn saturating_i64(value: u64) -> i64 {
489    value.min(i64::MAX as u64) as i64
490}
491
492#[cfg(test)]
493mod tests {
494    use std::io::Write;
495
496    use tempfile::NamedTempFile;
497
498    use super::*;
499
500    #[test]
501    fn request_times_uses_event_time_when_total_duration_is_missing() {
502        let request = RequestTraceRequestMetrics {
503            request_id: "req".to_string(),
504            output_tokens: Some(1),
505            request_received_ms: Some(1_000),
506            ..Default::default()
507        };
508
509        assert_eq!(request_times(1_250, &request), (1_000, 1_250));
510    }
511
512    #[test]
513    fn loads_context_free_request_trace() {
514        let mut file = NamedTempFile::new().unwrap();
515        writeln!(
516            file,
517            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":3,"input_sequence_hashes":[11,22]}}}}}}"#
518        )
519        .unwrap();
520
521        let loaded = load_request_trace_records(&[file.path().to_path_buf()]).unwrap();
522        assert_eq!(loaded.requests.len(), 1);
523        assert!(loaded.requests[0].agent_context.is_none());
524        assert_eq!(loaded.requests[0].start_ms, 1_000);
525        assert_eq!(loaded.requests[0].end_ms, 1_100);
526    }
527
528    #[test]
529    fn loads_wrapped_request_trace_record() {
530        let mut file = NamedTempFile::new().unwrap();
531        writeln!(
532            file,
533            r#"{{"timestamp":1,"event":{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":3,"input_sequence_hashes":[11,22]}}}}}}}}"#
534        )
535        .unwrap();
536
537        let loaded = load_request_trace_records(&[file.path().to_path_buf()]).unwrap();
538        assert_eq!(loaded.requests.len(), 1);
539        assert_eq!(loaded.requests[0].request.request_id, "req-1");
540    }
541
542    #[test]
543    fn skips_request_payload_records() {
544        let mut file = NamedTempFile::new().unwrap();
545        writeln!(
546            file,
547            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_payload","event_time_unix_ms":1050,"payload":{{"request_id":"req-1","endpoint":"openai.chat_completion","model":"test","request":{{"model":"test","messages":[{{"role":"user","content":"hi"}}]}},"payload_complete":true}}}}"#
548        )
549        .unwrap();
550        let large_payload = "x".repeat(4096);
551        writeln!(
552            file,
553            "{}",
554            serde_json::json!({
555                "timestamp": 1051,
556                "event": {
557                    "schema": "dynamo.request.trace.v1",
558                    "event_type": "request_payload",
559                    "event_time_unix_ms": 1051,
560                    "payload": {
561                        "request_id": "req-1",
562                        "endpoint": "openai.chat_completion",
563                        "model": "test",
564                        "request": {
565                            "model": "test",
566                            "messages": [{
567                                "role": "user",
568                                "content": large_payload.clone(),
569                            }],
570                        },
571                        "response": {
572                            "choices": [{
573                                "message": {
574                                    "role": "assistant",
575                                    "content": large_payload,
576                                },
577                            }],
578                        },
579                        "payload_complete": true,
580                    },
581                },
582            })
583        )
584        .unwrap();
585        writeln!(
586            file,
587            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":3,"input_sequence_hashes":[11,22]}}}}}}"#
588        )
589        .unwrap();
590
591        let loaded = load_request_trace_records(&[file.path().to_path_buf()]).unwrap();
592        assert_eq!(loaded.requests.len(), 1);
593        assert_eq!(loaded.requests[0].request.request_id, "req-1");
594    }
595
596    #[test]
597    fn rejects_unknown_trace_schema() {
598        let mut file = NamedTempFile::new().unwrap();
599        writeln!(
600            file,
601            r#"{{"schema":"dynamo.request.trace.v2","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":2,"input_sequence_hashes":[11]}}}}}}"#
602        )
603        .unwrap();
604
605        let error = load_request_trace_records(&[file.path().to_path_buf()]).unwrap_err();
606        assert!(error.to_string().contains("failed to parse"));
607        assert!(format!("{error:#}").contains("unknown variant `dynamo.request.trace.v2`"));
608    }
609
610    #[test]
611    fn request_trace_requires_arrival_time_and_output_length() {
612        let mut file = NamedTempFile::new().unwrap();
613        writeln!(
614            file,
615            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","replay":{{"trace_block_size":2,"input_length":2,"input_sequence_hashes":[11]}}}}}}"#
616        )
617        .unwrap();
618
619        let error = load_request_trace_records(&[file.path().to_path_buf()]).unwrap_err();
620        assert!(format!("{error:#}").contains("request trace is missing request_received_ms"));
621    }
622
623    #[test]
624    fn rejects_duplicate_request_ids() {
625        let mut file = NamedTempFile::new().unwrap();
626        for _ in 0..2 {
627            writeln!(
628                file,
629                r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":2,"input_sequence_hashes":[11]}}}}}}"#
630            )
631            .unwrap();
632        }
633
634        let error = load_request_trace_records(&[file.path().to_path_buf()]).unwrap_err();
635        assert!(error.to_string().contains("duplicate request_id req-1"));
636    }
637
638    #[test]
639    fn rejects_extra_replay_hashes() {
640        let mut file = NamedTempFile::new().unwrap();
641        writeln!(
642            file,
643            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"request":{{"request_id":"req-1","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":2,"input_sequence_hashes":[11,22]}}}}}}"#
644        )
645        .unwrap();
646
647        let error = load_request_trace_records(&[file.path().to_path_buf()]).unwrap_err();
648        assert!(format!("{error:#}").contains("requires exactly 1 replay hashes, got 2"));
649    }
650
651    #[test]
652    fn loads_agentic_request_trace_rows_and_tool_events() {
653        let mut file = NamedTempFile::new().unwrap();
654        writeln!(
655            file,
656            r#"{{"schema":"dynamo.request.trace.v1","event_type":"request_end","event_time_unix_ms":1100,"event_source":"dynamo","agent_context":{{"session_id":"root"}},"request":{{"request_id":"req-1","model":"test","request_received_ms":1000,"output_tokens":4,"replay":{{"trace_block_size":2,"input_length":3,"input_sequence_hashes":[11,22]}}}}}}"#
657        )
658        .unwrap();
659        writeln!(
660            file,
661            r#"{{"schema":"dynamo.request.trace.v1","event_type":"tool_end","event_time_unix_ms":1200,"event_source":"harness","agent_context":{{"session_id":"root"}},"tool":{{"tool_call_id":"tool-1","tool_class":"search","claude":{{"source_request_id":"req-1","consumer_request_id":"req-2","child_session_id":"child","execution_mode":"background"}},"started_at_unix_ms":1110,"ended_at_unix_ms":1200,"status":"succeeded","duration_ms":90}}}}"#
662        )
663        .unwrap();
664
665        let loaded = load_request_trace_records(&[file.path().to_path_buf()]).unwrap();
666        loaded.ensure_agentic_compatible().unwrap();
667        assert_eq!(loaded.requests.len(), 1);
668        assert_eq!(
669            loaded.requests[0]
670                .agent_context
671                .as_ref()
672                .expect("agent context")
673                .session_id,
674            "root"
675        );
676        assert_eq!(loaded.tools.len(), 1);
677        assert_eq!(loaded.tools[0].tool_call_id, "tool-1");
678        assert_eq!(loaded.tools[0].tool_class, "search");
679        let claude = loaded.tools[0].claude.as_ref().expect("Claude metadata");
680        assert_eq!(claude.source_request_id, "req-1");
681        assert_eq!(claude.consumer_request_id.as_deref(), Some("req-2"));
682        assert_eq!(claude.child_session_id.as_deref(), Some("child"));
683        assert_eq!(claude.execution_mode, "background");
684    }
685}
dynamo_data_gen/request_trace/load.rs

dynamo_data_gen/request_trace/
load.rs
