aisimulate_core/replay/loadgen/

dynamo.rs

// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Dynamo request-trace-v1 loading without a dependency on Dynamo crates.

use std::collections::{HashMap, HashSet};
use std::fs::File;
use std::io::{BufRead, BufReader, Read};
use std::path::PathBuf;

use anyhow::{Context, Result, anyhow, bail, ensure};
use flate2::read::MultiGzDecoder;
use serde::Deserialize;

use super::trace::assign_dependency_component_play_ids;
use super::{
    AGENTIC_MOONCAKE_SCHEMA, AGENTIC_MOONCAKE_VERSION, AgenticDependency,
    AgenticDependencyRelation, AgenticDependencyTrigger, AgenticHashIdScope, AgenticMooncakeHeader,
    AgenticMooncakeRow, AgenticSourceProvenance, AgenticTrace, MooncakeRow, Trace,
};

#[derive(Debug, Clone, PartialEq)]
pub enum DynamoRequestTrace {
    Standard(Trace),
    Agentic(AgenticTrace),
}

#[derive(Debug, Clone, Deserialize)]
struct Record {
    schema: String,
    event_type: String,
    event_time_unix_ms: u64,
    #[serde(default)]
    agent_context: Option<AgentContext>,
    #[serde(default)]
    request: Option<RequestMetrics>,
    #[serde(default)]
    tool: Option<ToolMetrics>,
}

#[derive(Debug, Clone, Deserialize)]
struct AgentContext {
    session_id: String,
    #[serde(default)]
    parent_session_id: Option<String>,
}

#[derive(Debug, Clone, Deserialize)]
struct RequestMetrics {
    request_id: String,
    #[serde(default)]
    model: Option<String>,
    #[serde(default)]
    output_tokens: Option<u64>,
    #[serde(default)]
    request_received_ms: Option<u64>,
    #[serde(default)]
    total_time_ms: Option<f64>,
    replay: ReplayMetrics,
}

#[derive(Debug, Clone, Deserialize)]
struct ReplayMetrics {
    trace_block_size: usize,
    input_length: usize,
    input_sequence_hashes: Vec<u64>,
}

#[derive(Debug, Clone, Deserialize)]
struct ToolMetrics {
    tool_call_id: String,
    tool_class: String,
    #[serde(default)]
    claude: Option<ClaudeToolMetrics>,
    #[serde(default)]
    started_at_unix_ms: Option<u64>,
    #[serde(default)]
    ended_at_unix_ms: Option<u64>,
    #[serde(default)]
    duration_ms: Option<f64>,
}

#[derive(Debug, Clone, Deserialize)]
struct ClaudeToolMetrics {
    source_request_id: String,
    #[serde(default)]
    consumer_request_id: Option<String>,
    #[serde(default)]
    child_session_id: Option<String>,
    execution_mode: String,
}

#[derive(Debug, Clone)]
struct RequestEntry {
    start_ms: i64,
    end_ms: i64,
    agent_context: Option<AgentContext>,
    request: RequestMetrics,
}

#[derive(Debug, Clone)]
struct ToolEntry {
    session_id: String,
    tool_call_id: String,
    tool_class: String,
    claude: Option<ClaudeToolMetrics>,
}

#[derive(Debug, Default)]
struct LoadedEntries {
    requests: Vec<RequestEntry>,
    tools: Vec<ToolEntry>,
}

impl DynamoRequestTrace {
    pub fn from_request_trace_files(
        paths: &[PathBuf],
        expected_block_size: Option<usize>,
    ) -> Result<Self> {
        ensure!(!paths.is_empty(), "Dynamo trace requires at least one path");
        let loaded = load_entries(paths)?;
        let mut entries = loaded.requests;
        let contextual = entries
            .iter()
            .filter(|entry| entry.agent_context.is_some())
            .count();
        if contextual != 0 && contextual != entries.len() {
            bail!("Dynamo request trace cannot mix requests with and without agent_context");
        }
        let block_size = entries[0].request.replay.trace_block_size;
        ensure!(block_size > 0, "embedded trace block size must be positive");
        if entries
            .iter()
            .any(|entry| entry.request.replay.trace_block_size != block_size)
        {
            bail!("mixed replay trace_block_size values are not supported");
        }
        if let Some(expected) = expected_block_size {
            ensure!(
                expected == block_size,
                "trace_block_size {expected} does not match embedded Dynamo request trace block size {block_size}"
            );
        }
        entries.sort_by(|left, right| {
            (left.start_ms, left.end_ms, &left.request.request_id).cmp(&(
                right.start_ms,
                right.end_ms,
                &right.request.request_id,
            ))
        });
        if contextual == 0 {
            lower_standard(entries, block_size).map(Self::Standard)
        } else {
            lower_agentic(entries, loaded.tools, block_size).map(Self::Agentic)
        }
    }
}

fn open_reader(path: &PathBuf) -> Result<Box<dyn BufRead>> {
    let file = File::open(path).with_context(|| format!("failed to open {}", path.display()))?;
    let reader: Box<dyn Read> = if path.extension().and_then(|value| value.to_str()) == Some("gz") {
        Box::new(MultiGzDecoder::new(file))
    } else {
        Box::new(file)
    };
    Ok(Box::new(BufReader::new(reader)))
}

fn parse_record(line: &str) -> Result<Option<Record>> {
    let value: serde_json::Value = serde_json::from_str(line)?;
    let Some(object) = value.as_object() else {
        return Ok(None);
    };
    let event = object.get("event").unwrap_or(&value);
    let event_type = event
        .get("event_type")
        .and_then(serde_json::Value::as_str)
        .or_else(|| object.get("event_type").and_then(serde_json::Value::as_str));
    if matches!(event_type, Some("request_payload" | "tool_start")) {
        return Ok(None);
    }
    ensure!(
        matches!(event_type, Some("request_end" | "tool_end" | "tool_error")),
        "request trace supports request_end, terminal tool events, request_payload, and tool_start; got {event_type:?}"
    );
    let record: Record = serde_json::from_value(event.clone())?;
    ensure!(
        record.schema == "dynamo.request.trace.v1",
        "unsupported Dynamo request trace schema {:?}",
        record.schema
    );
    ensure!(
        Some(record.event_type.as_str()) == event_type,
        "record event_type changed while decoding"
    );
    Ok(Some(record))
}

fn load_entries(paths: &[PathBuf]) -> Result<LoadedEntries> {
    let mut loaded = LoadedEntries::default();
    let mut request_ids = HashSet::new();
    for path in paths {
        for (line_index, line) in open_reader(path)?.lines().enumerate() {
            let line = line
                .with_context(|| format!("failed to read {}:{}", path.display(), line_index + 1))?;
            if line.trim().is_empty() {
                continue;
            }
            let Some(record) = parse_record(&line).with_context(|| {
                format!("failed to parse {}:{}", path.display(), line_index + 1)
            })?
            else {
                continue;
            };
            if record.event_type == "request_end" {
                let request = record
                    .request
                    .context("request_end is missing request metrics")?;
                ensure!(
                    !request.request_id.trim().is_empty(),
                    "request_id must be nonempty"
                );
                ensure!(
                    request_ids.insert(request.request_id.clone()),
                    "duplicate request_id {:?}",
                    request.request_id
                );
                let (start_ms, end_ms) = request_times(record.event_time_unix_ms, &request)?;
                loaded.requests.push(RequestEntry {
                    start_ms,
                    end_ms,
                    agent_context: record.agent_context,
                    request,
                });
            } else if let Some(tool) = tool_entry(record)? {
                loaded.tools.push(tool);
            }
        }
    }
    ensure!(
        !loaded.requests.is_empty(),
        "Dynamo trace contains no request_end records"
    );
    Ok(loaded)
}

fn request_times(event_time_unix_ms: u64, request: &RequestMetrics) -> Result<(i64, i64)> {
    let total_ms = request
        .total_time_ms
        .map(|value| {
            ensure!(
                value.is_finite() && value >= 0.0,
                "request duration must be finite and nonnegative"
            );
            Ok(value.round() as u64)
        })
        .transpose()?;
    let end_ms = match (request.request_received_ms, total_ms) {
        (Some(start), Some(duration)) => start.saturating_add(duration),
        _ => event_time_unix_ms,
    };
    let start_ms = request
        .request_received_ms
        .unwrap_or_else(|| event_time_unix_ms.saturating_sub(total_ms.unwrap_or(0)));
    Ok((saturating_i64(start_ms), saturating_i64(end_ms)))
}

fn tool_entry(record: Record) -> Result<Option<ToolEntry>> {
    let Some(context) = record.agent_context else {
        return Ok(None);
    };
    let Some(tool) = record.tool else {
        return Ok(None);
    };
    ensure!(
        !context.session_id.trim().is_empty(),
        "tool session_id must be nonempty"
    );
    ensure!(
        !tool.tool_call_id.trim().is_empty(),
        "tool_call_id must be nonempty"
    );
    ensure!(
        !tool.tool_class.trim().is_empty(),
        "tool_class must be nonempty"
    );
    if let Some(duration) = tool.duration_ms {
        ensure!(
            duration.is_finite() && duration >= 0.0,
            "tool duration must be finite and nonnegative"
        );
    }
    let end_ms = saturating_i64(tool.ended_at_unix_ms.unwrap_or(record.event_time_unix_ms));
    let start_ms = tool
        .started_at_unix_ms
        .map(saturating_i64)
        .or_else(|| {
            tool.duration_ms
                .map(|duration| end_ms.saturating_sub(duration.round() as i64))
        })
        .unwrap_or(end_ms);
    ensure!(end_ms >= start_ms, "tool end time precedes start time");
    Ok(Some(ToolEntry {
        session_id: context.session_id,
        tool_call_id: tool.tool_call_id,
        tool_class: tool.tool_class,
        claude: tool.claude,
    }))
}

fn saturating_i64(value: u64) -> i64 {
    value.min(i64::MAX as u64) as i64
}

fn lower_standard(entries: Vec<RequestEntry>, block_size: usize) -> Result<Trace> {
    let first_start = entries
        .iter()
        .map(|entry| entry.start_ms)
        .min()
        .ok_or_else(|| anyhow!("Dynamo trace contains no requests"))?;
    let rows = entries
        .into_iter()
        .map(|entry| -> Result<MooncakeRow> {
            Ok(MooncakeRow {
                request_id: Some(entry.request.request_id),
                input_length: Some(entry.request.replay.input_length),
                output_length: Some(
                    usize::try_from(
                        entry
                            .request
                            .output_tokens
                            .context("missing output_tokens")?,
                    )
                    .context("output_tokens does not fit usize")?,
                ),
                hash_ids: Some(entry.request.replay.input_sequence_hashes),
                timestamp: Some((entry.start_ms - first_start) as f64),
                ..Default::default()
            })
        })
        .collect::<Result<Vec<_>>>()?;
    Trace::from_mooncake_rows(rows, block_size)
}

fn lower_agentic(
    entries: Vec<RequestEntry>,
    tools: Vec<ToolEntry>,
    block_size: usize,
) -> Result<AgenticTrace> {
    let first_start = entries
        .iter()
        .map(|entry| entry.start_ms)
        .min()
        .ok_or_else(|| anyhow!("Dynamo trace contains no requests"))?;
    let id_to_index = entries
        .iter()
        .enumerate()
        .map(|(index, entry)| (entry.request.request_id.clone(), index))
        .collect::<HashMap<_, _>>();
    let mut by_session: HashMap<String, Vec<usize>> = HashMap::new();
    let mut parent_by_session: HashMap<String, String> = HashMap::new();
    for (index, entry) in entries.iter().enumerate() {
        let context = entry
            .agent_context
            .as_ref()
            .context("agentic request is missing agent_context")?;
        ensure!(
            !context.session_id.trim().is_empty(),
            "session_id must be nonempty"
        );
        by_session
            .entry(context.session_id.clone())
            .or_default()
            .push(index);
        if let Some(parent) = context.parent_session_id.as_ref() {
            match parent_by_session.get(&context.session_id) {
                Some(existing) if existing != parent => bail!(
                    "session {:?} has conflicting parent_session_id values {:?} and {:?}",
                    context.session_id,
                    existing,
                    parent
                ),
                Some(_) => {}
                None => {
                    parent_by_session.insert(context.session_id.clone(), parent.clone());
                }
            }
        }
    }
    for indices in by_session.values_mut() {
        indices.sort_by_key(|index| {
            let entry = &entries[*index];
            (
                entry.start_ms,
                entry.end_ms,
                entry.request.request_id.clone(),
            )
        });
    }
    let mut dependencies = vec![Vec::<AgenticDependency>::new(); entries.len()];
    for indices in by_session.values() {
        for pair in indices.windows(2) {
            push_dependency(
                &mut dependencies[pair[1]],
                dependency_between(
                    &entries,
                    pair[0],
                    pair[1],
                    AgenticDependencyTrigger::Completion,
                    AgenticDependencyRelation::Sequence,
                ),
            );
        }
    }

    let mut explicit_tool_by_child: HashMap<String, &ToolEntry> = HashMap::new();
    for tool in &tools {
        let Some(claude) = tool.claude.as_ref() else {
            continue;
        };
        ensure!(
            matches!(claude.execution_mode.as_str(), "blocking" | "background"),
            "tool {:?} ({}) has unsupported execution_mode {:?}",
            tool.tool_call_id,
            tool.tool_class,
            claude.execution_mode
        );
        for request_id in [
            Some(claude.source_request_id.as_str()),
            claude.consumer_request_id.as_deref(),
        ]
        .into_iter()
        .flatten()
        {
            let request_index = id_to_index.get(request_id).with_context(|| {
                format!(
                    "tool {:?} references unknown request_id {:?}",
                    tool.tool_call_id, request_id
                )
            })?;
            let request_session = &entries[*request_index]
                .agent_context
                .as_ref()
                .expect("validated agent context")
                .session_id;
            ensure!(
                request_session == &tool.session_id,
                "tool {:?} request {:?} belongs to session {:?}, expected {:?}",
                tool.tool_call_id,
                request_id,
                request_session,
                tool.session_id
            );
        }
        let Some(child_session) = claude.child_session_id.as_ref() else {
            continue;
        };
        if !by_session.contains_key(child_session) {
            continue;
        }
        ensure!(
            explicit_tool_by_child
                .insert(child_session.clone(), tool)
                .is_none(),
            "multiple tool events reference child session {:?}",
            child_session
        );
    }

    for (child_session, parent_session) in &parent_by_session {
        let child_indices = by_session
            .get(child_session)
            .expect("child session must have requests");
        let parent_indices = by_session.get(parent_session).with_context(|| {
            format!(
                "child session {:?} references unknown parent session {:?}",
                child_session, parent_session
            )
        })?;
        let first_child = child_indices[0];
        let last_child = *child_indices
            .iter()
            .max_by_key(|index| {
                let entry = &entries[**index];
                (entry.end_ms, entry.start_ms, &entry.request.request_id)
            })
            .expect("child session must be nonempty");

        if let Some(tool) = explicit_tool_by_child.get(child_session) {
            let claude = tool.claude.as_ref().expect("explicit tool has metadata");
            let parent_spawn = id_to_index[&claude.source_request_id];
            ensure!(
                parent_indices.contains(&parent_spawn),
                "tool {:?} source request {:?} is not in parent session {:?}",
                tool.tool_call_id,
                claude.source_request_id,
                parent_session
            );
            push_dependency(
                &mut dependencies[first_child],
                dependency_between(
                    &entries,
                    parent_spawn,
                    first_child,
                    AgenticDependencyTrigger::Dispatch,
                    AgenticDependencyRelation::Spawn,
                ),
            );
            if let Some(consumer) = claude.consumer_request_id.as_ref() {
                let parent_join = id_to_index[consumer];
                ensure!(
                    parent_indices.contains(&parent_join),
                    "tool {:?} consumer request {:?} is not in parent session {:?}",
                    tool.tool_call_id,
                    consumer,
                    parent_session
                );
                push_dependency(
                    &mut dependencies[parent_join],
                    dependency_between(
                        &entries,
                        last_child,
                        parent_join,
                        AgenticDependencyTrigger::Completion,
                        AgenticDependencyRelation::Join,
                    ),
                );
            }
            continue;
        }

        if let Some(parent_spawn) = parent_indices
            .iter()
            .copied()
            .filter(|index| entries[*index].start_ms <= entries[first_child].start_ms)
            .max_by_key(|index| entries[*index].start_ms)
        {
            push_dependency(
                &mut dependencies[first_child],
                dependency_between(
                    &entries,
                    parent_spawn,
                    first_child,
                    AgenticDependencyTrigger::Dispatch,
                    AgenticDependencyRelation::Spawn,
                ),
            );
        }
        if let Some(parent_join) = parent_indices
            .iter()
            .copied()
            .filter(|index| entries[*index].start_ms >= entries[last_child].end_ms)
            .min_by_key(|index| entries[*index].start_ms)
        {
            push_dependency(
                &mut dependencies[parent_join],
                dependency_between(
                    &entries,
                    last_child,
                    parent_join,
                    AgenticDependencyTrigger::Completion,
                    AgenticDependencyRelation::Join,
                ),
            );
        }
    }

    let mut rows = entries
        .iter()
        .enumerate()
        .map(|(index, entry)| -> Result<AgenticMooncakeRow> {
            let context = entry
                .agent_context
                .as_ref()
                .expect("validated agent context");
            let request_dependencies = dependencies[index].clone();
            Ok(AgenticMooncakeRow {
                request_id: entry.request.request_id.clone(),
                play_id: "dynamo-request-trace".to_string(),
                session_id: context.session_id.clone(),
                model: entry
                    .request
                    .model
                    .clone()
                    .unwrap_or_else(|| "unknown".to_string()),
                input_length: Some(entry.request.replay.input_length),
                output_length: Some(
                    usize::try_from(
                        entry
                            .request
                            .output_tokens
                            .context("missing output_tokens")?,
                    )
                    .context("output_tokens does not fit usize")?,
                ),
                output_token_ids: None,
                hash_ids: Some(entry.request.replay.input_sequence_hashes.clone()),
                not_before_ms: if request_dependencies.is_empty() {
                    (entry.start_ms - first_start) as f64
                } else {
                    0.0
                },
                priority: None,
                strict_priority: None,
                policy_class: None,
                dependencies: request_dependencies,
            })
        })
        .collect::<Result<Vec<_>>>()?;
    assign_dependency_component_play_ids(&mut rows, "dynamo-play");
    AgenticTrace::from_agentic_mooncake_rows(
        AgenticMooncakeHeader {
            schema: AGENTIC_MOONCAKE_SCHEMA.to_string(),
            version: AGENTIC_MOONCAKE_VERSION,
            block_size,
            hash_id_scope: AgenticHashIdScope::Local,
            source: AgenticSourceProvenance {
                format: "dynamo.request.trace.v1".to_string(),
                digest: format!("requests:{};tools:{}", rows.len(), tools.len()),
            },
        },
        rows,
    )
}

fn dependency_between(
    entries: &[RequestEntry],
    source: usize,
    target: usize,
    trigger: AgenticDependencyTrigger,
    relation: AgenticDependencyRelation,
) -> AgenticDependency {
    let source_time = match trigger {
        AgenticDependencyTrigger::Dispatch => entries[source].start_ms,
        AgenticDependencyTrigger::Completion => entries[source].end_ms,
    };
    AgenticDependency {
        request_id: entries[source].request.request_id.clone(),
        trigger,
        delay_ms: entries[target].start_ms.saturating_sub(source_time) as f64,
        relation,
    }
}

fn push_dependency(dependencies: &mut Vec<AgenticDependency>, dependency: AgenticDependency) {
    if dependencies.iter().any(|existing| {
        existing.request_id == dependency.request_id && existing.trigger == dependency.trigger
    }) {
        return;
    }
    dependencies.push(dependency);
}

#[cfg(test)]
mod tests {
    use std::io::Write;

    use serde_json::json;
    use tempfile::NamedTempFile;

    use super::*;

    fn request(id: &str, start_ms: u64, session_id: Option<&str>) -> serde_json::Value {
        let mut value = json!({
            "schema": "dynamo.request.trace.v1",
            "event_type": "request_end",
            "event_time_unix_ms": start_ms + 10,
            "request": {
                "request_id": id,
                "output_tokens": 2,
                "request_received_ms": start_ms,
                "total_time_ms": 10,
                "replay": {
                    "trace_block_size": 4,
                    "input_length": 4,
                    "input_sequence_hashes": [11]
                }
            }
        });
        if let Some(session_id) = session_id {
            value["agent_context"] = json!({"session_id": session_id});
        }
        value
    }

    fn child_request(
        id: &str,
        start_ms: u64,
        session_id: &str,
        parent_session_id: &str,
    ) -> serde_json::Value {
        let mut value = request(id, start_ms, Some(session_id));
        value["agent_context"]["parent_session_id"] = json!(parent_session_id);
        value
    }

    fn child_tool(
        source_request_id: &str,
        consumer_request_id: Option<&str>,
        child_session_id: &str,
        execution_mode: &str,
    ) -> serde_json::Value {
        json!({
            "schema": "dynamo.request.trace.v1",
            "event_type": "tool_end",
            "event_time_unix_ms": 115,
            "agent_context": {"session_id": "parent"},
            "tool": {
                "tool_call_id": "tool-1",
                "tool_class": "agent",
                "started_at_unix_ms": 110,
                "ended_at_unix_ms": 115,
                "claude": {
                    "source_request_id": source_request_id,
                    "consumer_request_id": consumer_request_id,
                    "child_session_id": child_session_id,
                    "execution_mode": execution_mode
                }
            }
        })
    }

    fn dependencies<'a>(trace: &'a AgenticTrace, request_id: &str) -> &'a [AgenticDependency] {
        trace
            .nodes()
            .iter()
            .find(|node| node.request_id() == request_id)
            .expect("request must exist")
            .dependencies()
    }

    fn trace_file(rows: &[serde_json::Value]) -> NamedTempFile {
        let mut file = NamedTempFile::new().unwrap();
        for row in rows {
            writeln!(file, "{}", serde_json::to_string(row).unwrap()).unwrap();
        }
        file
    }

    #[test]
    fn loads_standard_multi_file_trace() {
        let first = trace_file(&[request("a", 100, None)]);
        let second = trace_file(&[request("b", 120, None)]);
        let loaded = DynamoRequestTrace::from_request_trace_files(
            &[first.path().to_path_buf(), second.path().to_path_buf()],
            Some(4),
        )
        .unwrap();
        let DynamoRequestTrace::Standard(trace) = loaded else {
            panic!("expected standard trace");
        };
        assert_eq!(trace.sessions.len(), 2);
        assert_eq!(trace.sessions[0].first_arrival_timestamp_ms, Some(0.0));
        assert_eq!(trace.sessions[1].first_arrival_timestamp_ms, Some(20.0));
    }

    #[test]
    fn loads_agentic_trace_as_dependency_graph() {
        let file = trace_file(&[
            request("a", 100, Some("session")),
            request("b", 120, Some("session")),
        ]);
        let loaded =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap();
        let DynamoRequestTrace::Agentic(trace) = loaded else {
            panic!("expected agentic trace");
        };
        assert_eq!(trace.node_count(), 2);
        assert_eq!(trace.play_count(), 1);
    }

    #[test]
    fn missing_duration_uses_request_end_event_time() {
        let mut first = request("a", 100, Some("session"));
        first["event_time_unix_ms"] = json!(150);
        first["request"]
            .as_object_mut()
            .unwrap()
            .remove("total_time_ms");
        let file = trace_file(&[first, request("b", 160, Some("session"))]);

        let DynamoRequestTrace::Agentic(trace) =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap()
        else {
            panic!("expected agentic trace");
        };

        let dependency = &dependencies(&trace, "b")[0];
        assert_eq!(dependency.request_id, "a");
        assert_eq!(dependency.delay_ms, 10.0);
    }

    #[test]
    fn blocking_child_preserves_parent_sequence_spawn_and_join() {
        let file = trace_file(&[
            request("parent-source", 100, Some("parent")),
            child_tool(
                "parent-source",
                Some("parent-consumer"),
                "child",
                "blocking",
            ),
            child_request("child-first", 120, "child", "parent"),
            child_request("child-last", 150, "child", "parent"),
            request("parent-consumer", 200, Some("parent")),
        ]);

        let DynamoRequestTrace::Agentic(trace) =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap()
        else {
            panic!("expected agentic trace");
        };

        assert!(
            dependencies(&trace, "child-first")
                .iter()
                .any(|dependency| {
                    dependency.request_id == "parent-source"
                        && dependency.trigger == AgenticDependencyTrigger::Dispatch
                        && dependency.relation == AgenticDependencyRelation::Spawn
                })
        );
        let consumer = dependencies(&trace, "parent-consumer");
        assert!(consumer.iter().any(|dependency| {
            dependency.request_id == "parent-source"
                && dependency.relation == AgenticDependencyRelation::Sequence
        }));
        assert!(consumer.iter().any(|dependency| {
            dependency.request_id == "child-last"
                && dependency.trigger == AgenticDependencyTrigger::Completion
                && dependency.relation == AgenticDependencyRelation::Join
        }));
    }

    #[test]
    fn background_child_launches_without_implicit_parent_join() {
        let file = trace_file(&[
            request("parent-source", 100, Some("parent")),
            child_tool("parent-source", None, "child", "background"),
            child_request("child", 120, "child", "parent"),
            request("parent-next", 130, Some("parent")),
        ]);

        let DynamoRequestTrace::Agentic(trace) =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap()
        else {
            panic!("expected agentic trace");
        };

        assert!(dependencies(&trace, "child").iter().any(|dependency| {
            dependency.request_id == "parent-source"
                && dependency.trigger == AgenticDependencyTrigger::Dispatch
                && dependency.relation == AgenticDependencyRelation::Spawn
        }));
        assert_eq!(dependencies(&trace, "parent-next").len(), 1);
        assert_eq!(
            dependencies(&trace, "parent-next")[0].request_id,
            "parent-source"
        );
    }

    #[test]
    fn timestamp_fallback_infers_spawn_and_last_child_join() {
        let file = trace_file(&[
            request("parent-source", 100, Some("parent")),
            child_request("child-first", 120, "child", "parent"),
            child_request("child-last", 150, "child", "parent"),
            request("parent-consumer", 200, Some("parent")),
        ]);

        let DynamoRequestTrace::Agentic(trace) =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap()
        else {
            panic!("expected agentic trace");
        };

        assert!(
            dependencies(&trace, "child-first")
                .iter()
                .any(|dependency| {
                    dependency.request_id == "parent-source"
                        && dependency.relation == AgenticDependencyRelation::Spawn
                })
        );
        assert!(
            dependencies(&trace, "parent-consumer")
                .iter()
                .any(|dependency| {
                    dependency.request_id == "child-last"
                        && dependency.relation == AgenticDependencyRelation::Join
                })
        );
    }

    #[test]
    fn independent_agent_sessions_become_independent_plays() {
        let file = trace_file(&[
            request("a", 100, Some("session-a")),
            request("b", 120, Some("session-b")),
        ]);
        let loaded =
            DynamoRequestTrace::from_request_trace_files(&[file.path().to_path_buf()], Some(4))
                .unwrap();
        let DynamoRequestTrace::Agentic(trace) = loaded else {
            panic!("expected agentic trace");
        };
        assert_eq!(trace.node_count(), 2);
        assert_eq!(trace.play_count(), 2);
    }
}