diff --git a/.github/actions/upload-aiperf-traces/action.yml b/.github/actions/upload-aiperf-traces/action.yml
new file mode 100644
index 0000000000..1291cae57b
--- /dev/null
+++ b/.github/actions/upload-aiperf-traces/action.yml
@@ -0,0 +1,105 @@
+name: Upload AIPerf Perfetto traces
+description: >-
+ Convert profile_export.jsonl files into Chrome/Perfetto traces and upload
+ them as a workflow artifact. No-ops when no AIPerf exports are present.
+
+inputs:
+ result-dir:
+ description: Directory to scan for profile_export.jsonl
+ required: true
+ artifact-name:
+ description: GitHub Actions artifact name
+ required: true
+ output-dir:
+ description: Directory to write generated traces
+ required: false
+ default: aiperf-traces
+ run-url:
+ description: Workflow run URL for the Job Summary download links
+ required: false
+ default: ""
+ retention-days:
+ description: Artifact retention in days
+ required: false
+ default: "30"
+
+outputs:
+ has_traces:
+ description: Whether any Chrome traces were generated
+ value: ${{ steps.convert.outputs.has_traces }}
+ artifact-url:
+ description: Artifact download URL when traces were uploaded
+ value: ${{ steps.upload.outputs.artifact-url }}
+
+runs:
+ using: composite
+ steps:
+ - name: Convert AIPerf exports to Chrome traces
+ id: convert
+ shell: bash
+ env:
+ RESULT_DIR: ${{ inputs.result-dir }}
+ OUTPUT_DIR: ${{ inputs.output-dir }}
+ RUN_URL: ${{ inputs.run-url }}
+ run: |
+ set -euo pipefail
+ mkdir -p "${OUTPUT_DIR}"
+ if [[ -d "${RESULT_DIR}" ]]; then
+ extra=()
+ if [[ -n "${RUN_URL}" ]]; then
+ extra+=(--run-url "${RUN_URL}")
+ fi
+ if ! python3 "${GITHUB_WORKSPACE}/.github/scripts/generate_aiperf_traces.py" \
+ "${RESULT_DIR}" \
+ --output-dir "${OUTPUT_DIR}" \
+ --index "${OUTPUT_DIR}/TRACE_INDEX.md" \
+ "${extra[@]}"; then
+ echo "::warning::AIPerf chrome-trace conversion reported errors"
+ fi
+ else
+ echo "skip chrome traces: ${RESULT_DIR} does not exist"
+ fi
+ has_traces=false
+ if [[ -n "$(find "${OUTPUT_DIR}" -type f \( -name '*.trace.json' -o -name '*.trace.json.gz' \) -print -quit)" ]]; then
+ has_traces=true
+ fi
+ echo "has_traces=${has_traces}" >> "$GITHUB_OUTPUT"
+
+ - name: Upload Perfetto traces
+ if: ${{ steps.convert.outputs.has_traces == 'true' }}
+ id: upload
+ uses: actions/upload-artifact@v7
+ with:
+ name: ${{ inputs.artifact-name }}
+ path: ${{ inputs.output-dir }}/
+ if-no-files-found: ignore
+ retention-days: ${{ inputs.retention-days }}
+
+ - name: Publish Perfetto trace download links
+ if: ${{ steps.convert.outputs.has_traces == 'true' }}
+ shell: bash
+ env:
+ ARTIFACT_URL: ${{ steps.upload.outputs.artifact-url }}
+ ARTIFACT_NAME: ${{ inputs.artifact-name }}
+ OUTPUT_DIR: ${{ inputs.output-dir }}
+ RUN_URL: ${{ inputs.run-url }}
+ run: |
+ set -euo pipefail
+ {
+ echo
+ echo "### AIPerf Chrome Traces (Perfetto)"
+ echo
+ if [[ -n "${ARTIFACT_URL}" ]]; then
+ echo "Download [\`${ARTIFACT_NAME}\`](${ARTIFACT_URL})."
+ elif [[ -n "${RUN_URL}" ]]; then
+ echo "Download artifact \`${ARTIFACT_NAME}\` from [this run](${RUN_URL})."
+ else
+ echo "Download artifact \`${ARTIFACT_NAME}\` from this workflow run."
+ fi
+ echo
+ echo "Open the \`.trace.json.gz\` file in [Perfetto UI](https://ui.perfetto.dev) (\`Open trace file\`) or gunzip it and load \`chrome://tracing\`."
+ echo
+ if [[ -f "${OUTPUT_DIR}/TRACE_INDEX.md" ]]; then
+ cat "${OUTPUT_DIR}/TRACE_INDEX.md"
+ fi
+ } >> "${GITHUB_STEP_SUMMARY}"
diff --git a/.github/scripts/aiperf_to_chrome_trace.py b/.github/scripts/aiperf_to_chrome_trace.py
new file mode 100755
index 0000000000..3093651529
--- /dev/null
+++ b/.github/scripts/aiperf_to_chrome_trace.py
@@ -0,0 +1,1177 @@
+#!/usr/bin/env python3
+"""Convert AIPerf per-request records to Chrome Trace Event JSON.
+
+The output can be opened locally with chrome://tracing or
+https://ui.perfetto.dev. It deliberately uses only the Python standard
+library so the conversion can run on a headless benchmark server.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from collections import defaultdict
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+NS_PER_MS = 1_000_000
+NS_PER_US = 1_000
+PID = 1
+CACHE_ANOMALY_THRESHOLD = 0.80
+
+
+def metric_value(metrics: dict[str, Any], name: str, default: Any = None) -> Any:
+ """Return an AIPerf metric value from its {value, unit} wrapper."""
+ entry = metrics.get(name)
+ if not isinstance(entry, dict):
+ return default
+ return entry.get("value", default)
+
+
+def load_records(path: str | Path) -> list[dict[str, Any]]:
+ """Load non-empty JSONL records and report malformed line numbers."""
+ records = []
+ with Path(path).open(encoding="utf-8") as handle:
+ for line_number, line in enumerate(handle, start=1):
+ if not line.strip():
+ continue
+ try:
+ record = json.loads(line)
+ except json.JSONDecodeError as exc:
+ raise ValueError(f"invalid JSON on line {line_number}") from exc
+ if not isinstance(record, dict):
+ raise TypeError(f"line {line_number} is not a JSON object")
+ records.append(record)
+ return records
+
+
+def _number(value: Any) -> float | None:
+ return float(value) if isinstance(value, (int, float)) else None
+
+
+def _session_value(metadata: dict[str, Any], session_key: str) -> str:
+ value = metadata.get(session_key)
+ if value is None:
+ value = (
+ metadata.get("root_correlation_id")
+ or metadata.get("x_correlation_id")
+ or metadata.get("conversation_id")
+ or "unknown"
+ )
+ return str(value)
+
+
+def _short_id(value: str, length: int = 12) -> str:
+ return value if len(value) <= length else f"{value[:length]}…"
+
+
+def _event(
+ name: str,
+ *,
+ ts_us: int,
+ tid: int,
+ category: str,
+ args: dict[str, Any] | None = None,
+ duration_us: int | None = None,
+ color: str | None = None,
+) -> dict[str, Any]:
+ event: dict[str, Any] = {
+ "name": name,
+ "cat": category,
+ "ph": "X" if duration_us is not None else "i",
+ "pid": PID,
+ "tid": tid,
+ "ts": ts_us,
+ "args": args or {},
+ }
+ if duration_us is not None:
+ event["dur"] = duration_us
+ else:
+ event["s"] = "t"
+ if color is not None:
+ event["cname"] = color
+ return event
+
+
+def _record_sort_key(record: dict[str, Any]) -> tuple[float, int]:
+ metadata = record.get("metadata", {})
+ start = _number(metadata.get("request_start_ns"))
+ return (
+ start if start is not None else float("inf"),
+ int(metadata.get("turn_index", 0)),
+ )
+
+
+def _cache_observation(metrics: dict[str, Any]) -> tuple[float | None, bool | None]:
+ input_tokens = _number(metric_value(metrics, "input_sequence_length"))
+ cache_tokens = _number(metric_value(metrics, "usage_prompt_cache_read_tokens"))
+ if input_tokens is None or cache_tokens is None or input_tokens <= 0:
+ return None, None
+ ratio = cache_tokens / input_tokens
+ return ratio, ratio < CACHE_ANOMALY_THRESHOLD
+
+
+def _tree_key(metadata: dict[str, Any]) -> str:
+ return str(
+ metadata.get("root_correlation_id")
+ or metadata.get("x_correlation_id")
+ or metadata.get("conversation_id")
+ or "unknown"
+ )
+
+
+def _request_session_key(metadata: dict[str, Any], tree: str) -> str:
+ return str(
+ metadata.get("x_correlation_id") or metadata.get("conversation_id") or tree
+ )
+
+
+def _pack_request_rows(
+ items: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+) -> tuple[dict[int, int], int]:
+ """Pack overlapping requests into non-overlapping rows."""
+ order = sorted(
+ range(len(items)), key=lambda index: (items[index][3], items[index][4])
+ )
+ row_ends: list[float] = []
+ rows: dict[int, int] = {}
+ for index in order:
+ start_ns, end_ns = items[index][3], items[index][4]
+ for row, row_end in enumerate(row_ends):
+ if start_ns >= row_end:
+ row_ends[row] = end_ns
+ rows[id(items[index][0])] = row
+ break
+ else:
+ rows[id(items[index][0])] = len(row_ends)
+ row_ends.append(end_ns)
+ return rows, len(row_ends)
+
+
+def _assign_tree_slots(
+ spans: dict[str, tuple[float, float]], concurrency: int
+) -> dict[str, int]:
+ """Assign non-overlapping tree lifetimes to a fixed number of slots."""
+ if concurrency < 1:
+ raise ValueError("concurrency must be at least 1")
+ slot_ends: list[float] = []
+ assignments: dict[str, int] = {}
+ for tree, (start_ns, end_ns) in sorted(spans.items(), key=lambda item: item[1][0]):
+ available = next(
+ (slot for slot, slot_end in enumerate(slot_ends) if start_ns >= slot_end),
+ None,
+ )
+ if available is None:
+ if len(slot_ends) < concurrency:
+ available = len(slot_ends)
+ slot_ends.append(end_ns)
+ else:
+ # Keep the requested number of tracks for an incomplete or
+ # inconsistent export whose tree spans appear over capacity.
+ available = min(range(concurrency), key=slot_ends.__getitem__)
+ slot_ends[available] = max(slot_ends[available], end_ns)
+ else:
+ slot_ends[available] = end_ns
+ assignments[tree] = available
+ return assignments
+
+
+def _build_slot_layout(
+ usable: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ concurrency: int,
+) -> tuple[dict[int, tuple[int, int]], dict[str, tuple[int, float, float]], int]:
+ """Return request placement, tree spans, and the number of used slots."""
+ trees: dict[str, list[tuple[dict[str, Any], dict[str, Any], str, float, float]]] = (
+ defaultdict(list)
+ )
+ tree_sessions: dict[
+ str, dict[str, list[tuple[dict[str, Any], dict[str, Any], str, float, float]]]
+ ] = defaultdict(lambda: defaultdict(list))
+ for item in usable:
+ tree = _tree_key(item[1])
+ session = _request_session_key(item[1], tree)
+ trees[tree].append(item)
+ tree_sessions[tree][session].append(item)
+
+ spans = {
+ tree: (min(item[3] for item in items), max(item[4] for item in items))
+ for tree, items in trees.items()
+ }
+ assignments = _assign_tree_slots(spans, concurrency)
+ placement: dict[int, tuple[int, int]] = {}
+ tree_layout: dict[str, tuple[int, float, float]] = {}
+ used_slots = max(assignments.values(), default=-1) + 1
+
+ for tree, members in tree_sessions.items():
+ slot = assignments[tree]
+ root_session = (
+ tree
+ if tree in members
+ else min(
+ members,
+ key=lambda session: (
+ min(item[3] for item in members[session]),
+ min((item[1].get("agent_depth") or 0) for item in members[session]),
+ ),
+ )
+ )
+ root_items = members.get(root_session, [])
+ root_rows, root_row_count = _pack_request_rows(root_items)
+ for item in root_items:
+ placement[id(item[0])] = (slot, root_rows[id(item[0])])
+
+ child_row_ends: list[float] = []
+ children = sorted(
+ (session for session in members if session != root_session),
+ key=lambda session: min(item[3] for item in members[session]),
+ )
+ for session in children:
+ child_items = members[session]
+ child_rows, child_row_count = _pack_request_rows(child_items)
+ child_start = min(item[3] for item in child_items)
+ child_end = max(item[4] for item in child_items)
+ row_base = 0
+ for row in range(len(child_row_ends) + 1):
+ if all(
+ child_row_ends[row + offset] <= child_start
+ for offset in range(child_row_count)
+ if row + offset < len(child_row_ends)
+ ):
+ row_base = row
+ break
+ child_row_ends.extend(
+ [0.0] * (row_base + child_row_count - len(child_row_ends))
+ )
+ for offset in range(child_row_count):
+ child_row_ends[row_base + offset] = child_end
+ for item in child_items:
+ placement[id(item[0])] = (
+ slot,
+ root_row_count + row_base + child_rows[id(item[0])],
+ )
+
+ tree_layout[tree] = (slot, spans[tree][0], spans[tree][1])
+ return placement, tree_layout, used_slots
+
+
+def _append_wait_prefill_analysis(
+ events: list[dict[str, Any]],
+ usable: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ placement: dict[int, tuple[int, int]],
+ *,
+ inspect_slot: int,
+ wait_tid: int | None,
+ overlap_tid: int,
+ origin_ns: float,
+) -> None:
+ """Add wait/TTFB and wait∩prefill events for one fixed slot."""
+ for record, metadata, _, start_ns, end_ns in usable:
+ slot, _ = placement[id(record)]
+ if slot != inspect_slot:
+ continue
+ metrics = record.get("metrics", {})
+ if not isinstance(metrics, dict):
+ continue
+ waiting_ms = _number(metric_value(metrics, "http_req_waiting"))
+ if waiting_ms is None or waiting_ms <= 0:
+ continue
+ sending_ms = _number(metric_value(metrics, "http_req_sending")) or 0.0
+ wait_start_ns = start_ns + sending_ms * NS_PER_MS
+ wait_end_ns = min(end_ns, wait_start_ns + waiting_ms * NS_PER_MS)
+ request_number = metadata.get("session_num", metadata.get("turn_index", "?"))
+ if wait_tid is not None:
+ events.append(
+ _event(
+ f"wait/TTFB #{request_number} | {waiting_ms:.1f} ms",
+ ts_us=round((wait_start_ns - origin_ns) / NS_PER_US),
+ tid=wait_tid,
+ category="http",
+ args={
+ "request_num": request_number,
+ "wait_start_ns": int(wait_start_ns),
+ "wait_end_ns": int(wait_end_ns),
+ "wait_ttfb_ms": waiting_ms,
+ "root_correlation_id": _tree_key(metadata),
+ },
+ duration_us=max(
+ 0, round((wait_end_ns - wait_start_ns) / NS_PER_US)
+ ),
+ )
+ )
+
+ ttft_ms = _number(metric_value(metrics, "time_to_first_token"))
+ if ttft_ms is None or ttft_ms < 0:
+ continue
+ prefill_end_ns = min(end_ns, start_ns + ttft_ms * NS_PER_MS)
+ overlap_start_ns = max(start_ns, wait_start_ns)
+ overlap_end_ns = min(prefill_end_ns, wait_end_ns)
+ overlap_ns = overlap_end_ns - overlap_start_ns
+ if overlap_ns <= 0:
+ continue
+ prefill_ms = max(0.0, (prefill_end_ns - start_ns) / NS_PER_MS)
+ overlap_ms = overlap_ns / NS_PER_MS
+ events.append(
+ _event(
+ f"wait∩prefill #{request_number} | {overlap_ms:.1f} ms",
+ ts_us=round((overlap_start_ns - origin_ns) / NS_PER_US),
+ tid=overlap_tid,
+ category="overlap",
+ args={
+ "request_num": request_number,
+ "overlap_ms": overlap_ms,
+ "prefill_ms": prefill_ms,
+ "prefill_covered_pct": (
+ 100.0 * overlap_ms / prefill_ms if prefill_ms else 0.0
+ ),
+ "wait_ttfb_ms": waiting_ms,
+ "time_to_first_token_ms": ttft_ms,
+ "root_correlation_id": _tree_key(metadata),
+ },
+ duration_us=round(overlap_ns / NS_PER_US),
+ )
+ )
+
+
+def _flow_event(
+ name: str,
+ phase: str,
+ *,
+ tid: int,
+ ts_us: int,
+ flow_id: str,
+ args: dict[str, Any],
+) -> dict[str, Any]:
+ return {
+ "name": name,
+ "cat": "flow",
+ "ph": phase,
+ "pid": PID,
+ "tid": tid,
+ "ts": ts_us,
+ "id": flow_id,
+ "args": args,
+ }
+
+
+def _append_ttft_summary(
+ events: list[dict[str, Any]],
+ usable: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ placement: dict[int, tuple[int, int]],
+ *,
+ summary_tid: int,
+ origin_ns: float,
+) -> None:
+ """Add one TTFT overview track and flow links to slot prefill rows."""
+ for record, metadata, _, start_ns, end_ns in usable:
+ metrics = record.get("metrics", {})
+ if not isinstance(metrics, dict):
+ metrics = {}
+ ttft_ms = _number(metric_value(metrics, "time_to_first_token"))
+ if ttft_ms is None or ttft_ms < 0:
+ continue
+ slot, row = placement[id(record)]
+ first_token_ns = min(end_ns, start_ns + ttft_ms * NS_PER_MS)
+ prefill_us = max(0, round((first_token_ns - start_ns) / NS_PER_US))
+ start_us = round((start_ns - origin_ns) / NS_PER_US)
+ request_number = metadata.get("session_num", metadata.get("turn_index", "?"))
+ request_id = str(metadata.get("x_request_id") or id(record))
+ flow_id = f"ttft:{request_id}"
+ input_tokens = metric_value(metrics, "input_sequence_length")
+ output_tokens = metric_value(metrics, "output_sequence_length")
+ cache_ratio, cache_anomaly = _cache_observation(metrics)
+ args = {
+ "request_num": request_number,
+ "x_request_id": metadata.get("x_request_id"),
+ "root_correlation_id": _tree_key(metadata),
+ "slot_index": slot,
+ "slot_row": row,
+ "time_to_first_token_ms": ttft_ms,
+ "input_tokens": input_tokens,
+ "usage_prompt_tokens": metric_value(metrics, "usage_prompt_tokens"),
+ "prompt_cache_read_tokens": metric_value(
+ metrics, "usage_prompt_cache_read_tokens"
+ ),
+ "cache_read_ratio_pct": (
+ 100.0 * cache_ratio if cache_ratio is not None else None
+ ),
+ "cache_anomaly": cache_anomaly,
+ "output_tokens": output_tokens,
+ "usage_completion_tokens": metric_value(metrics, "usage_completion_tokens"),
+ "target_track": f"slot {slot:02d} row {row}",
+ }
+ events.append(
+ _event(
+ f"slot {slot:02d} TTFT #{request_number} | {ttft_ms:.1f} ms",
+ ts_us=start_us,
+ tid=summary_tid,
+ category="ttft",
+ args=args,
+ duration_us=prefill_us,
+ color="terrible" if cache_anomaly else None,
+ )
+ )
+ events.append(
+ _flow_event(
+ "TTFT → slot prefill",
+ "s",
+ tid=summary_tid,
+ ts_us=start_us,
+ flow_id=flow_id,
+ args=args,
+ )
+ )
+ events.append(
+ _flow_event(
+ "TTFT → slot prefill",
+ "f",
+ tid=10_000 + slot * 100 + row + 1,
+ ts_us=start_us,
+ flow_id=flow_id,
+ args=args,
+ )
+ )
+
+
+def _append_decode_summary(
+ events: list[dict[str, Any]],
+ usable: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ placement: dict[int, tuple[int, int]],
+ *,
+ summary_tid: int,
+ origin_ns: float,
+) -> None:
+ """Add one decode overview track and flow links to slot decode rows."""
+ for record, metadata, _, start_ns, end_ns in usable:
+ metrics = record.get("metrics", {})
+ if not isinstance(metrics, dict):
+ metrics = {}
+ ttft_ms = _number(metric_value(metrics, "time_to_first_token"))
+ if ttft_ms is None or ttft_ms < 0:
+ continue
+ slot, row = placement[id(record)]
+ first_token_ns = min(end_ns, start_ns + ttft_ms * NS_PER_MS)
+ decode_us = max(0, round((end_ns - first_token_ns) / NS_PER_US))
+ request_number = metadata.get("session_num", metadata.get("turn_index", "?"))
+ decode_ms = _number(metric_value(metrics, "full_decode_duration"))
+ output_tokens = metric_value(metrics, "output_sequence_length")
+ args = {
+ "request_num": request_number,
+ "x_request_id": metadata.get("x_request_id"),
+ "root_correlation_id": _tree_key(metadata),
+ "slot_index": slot,
+ "slot_row": row,
+ "decode_ms": decode_ms,
+ "decode_interval_ms": decode_us / 1_000.0,
+ "output_tokens": output_tokens,
+ "target_track": f"slot {slot:02d} row {row}",
+ }
+ start_us = round((first_token_ns - origin_ns) / NS_PER_US)
+ request_id = str(metadata.get("x_request_id") or id(record))
+ flow_id = f"decode:{request_id}"
+ events.append(
+ _event(
+ f"slot {slot:02d} decode #{request_number} | {decode_us / 1_000:.1f} ms",
+ ts_us=start_us,
+ tid=summary_tid,
+ category="decode",
+ args=args,
+ duration_us=decode_us,
+ )
+ )
+ events.append(
+ _flow_event(
+ "decode → slot decode",
+ "s",
+ tid=summary_tid,
+ ts_us=start_us,
+ flow_id=flow_id,
+ args=args,
+ )
+ )
+ events.append(
+ _flow_event(
+ "decode → slot decode",
+ "f",
+ tid=10_000 + slot * 100 + row + 1,
+ ts_us=start_us,
+ flow_id=flow_id,
+ args=args,
+ )
+ )
+
+
+def _append_prefill_decode_summary(
+ events: list[dict[str, Any]],
+ usable: list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ placement: dict[int, tuple[int, int]],
+ *,
+ summary_tid: int,
+ origin_ns: float,
+) -> None:
+ """Add prefill and decode intervals to one combined overview track."""
+ for record, metadata, _, start_ns, end_ns in usable:
+ metrics = record.get("metrics", {})
+ if not isinstance(metrics, dict):
+ metrics = {}
+ ttft_ms = _number(metric_value(metrics, "time_to_first_token"))
+ if ttft_ms is None or ttft_ms < 0:
+ continue
+ slot, row = placement[id(record)]
+ first_token_ns = min(end_ns, start_ns + ttft_ms * NS_PER_MS)
+ prefill_us = max(0, round((first_token_ns - start_ns) / NS_PER_US))
+ decode_us = max(0, round((end_ns - first_token_ns) / NS_PER_US))
+ request_number = metadata.get("session_num", metadata.get("turn_index", "?"))
+ input_tokens = metric_value(metrics, "input_sequence_length")
+ output_tokens = metric_value(metrics, "output_sequence_length")
+ cache_ratio, cache_anomaly = _cache_observation(metrics)
+ args = {
+ "request_num": request_number,
+ "x_request_id": metadata.get("x_request_id"),
+ "root_correlation_id": _tree_key(metadata),
+ "slot_index": slot,
+ "slot_row": row,
+ "time_to_first_token_ms": ttft_ms,
+ "input_tokens": input_tokens,
+ "output_tokens": output_tokens,
+ "prompt_cache_read_tokens": metric_value(
+ metrics, "usage_prompt_cache_read_tokens"
+ ),
+ "cache_read_ratio_pct": (
+ 100.0 * cache_ratio if cache_ratio is not None else None
+ ),
+ "cache_anomaly": cache_anomaly,
+ }
+ start_us = round((start_ns - origin_ns) / NS_PER_US)
+ first_token_us = round((first_token_ns - origin_ns) / NS_PER_US)
+ events.append(
+ _event(
+ f"slot {slot:02d} prefill #{request_number} | {ttft_ms:.1f} ms",
+ ts_us=start_us,
+ tid=summary_tid,
+ category="prefill",
+ args=args,
+ duration_us=prefill_us,
+ color="terrible" if cache_anomaly else "yellow",
+ )
+ )
+ events.append(
+ _event(
+ f"slot {slot:02d} decode #{request_number} | {decode_us / 1_000:.1f} ms",
+ ts_us=first_token_us,
+ tid=summary_tid,
+ category="decode",
+ args=args,
+ duration_us=decode_us,
+ color="thread_state_runnable",
+ )
+ )
+
+
+def _usable_records(
+ records: Iterable[dict[str, Any]],
+ *,
+ phase: str | None,
+ session_key: str,
+ session_id: str | None,
+) -> list[tuple[dict[str, Any], dict[str, Any], str, float, float]]:
+ usable = []
+ for record in records:
+ metadata = record.get("metadata")
+ if not isinstance(metadata, dict):
+ continue
+ if phase is not None and metadata.get("benchmark_phase") != phase:
+ continue
+ current_session = _session_value(metadata, session_key)
+ if session_id is not None and current_session != session_id:
+ continue
+ start = _number(metadata.get("request_start_ns"))
+ end = _number(metadata.get("request_end_ns"))
+ if start is None:
+ continue
+ if end is None:
+ end = start
+ end = max(start, end)
+ usable.append((record, metadata, current_session, start, end))
+ return sorted(usable, key=lambda item: (item[3], int(item[1].get("turn_index", 0))))
+
+
+def build_trace(
+ records: Iterable[dict[str, Any]],
+ *,
+ phase: str | None = "profiling",
+ session_key: str = "root_correlation_id",
+ session_id: str | None = None,
+ concurrency: int | None = None,
+ inspect_slot: int | None = None,
+ inspect_all_slots: bool = False,
+) -> dict[str, Any]:
+ """Build a Chrome Trace Event document from AIPerf records.
+
+ When ``concurrency`` is provided, root trees are assigned to a fixed
+ number of reusable slots, matching AIPerf's agentic replay swim-lane
+ layout. Child x-correlation sessions are packed into rows below the root
+ row within each slot.
+ """
+ if (inspect_slot is not None or inspect_all_slots) and concurrency is None:
+ raise ValueError("slot analysis requires concurrency")
+ if inspect_slot is not None and inspect_all_slots:
+ raise ValueError("choose inspect_slot or inspect_all_slots, not both")
+ if (
+ inspect_slot is not None
+ and concurrency is not None
+ and not 0 <= inspect_slot < concurrency
+ ):
+ raise ValueError("inspect_slot must be within the configured concurrency slots")
+ usable = _usable_records(
+ records,
+ phase=phase,
+ session_key=session_key,
+ session_id=session_id,
+ )
+ if not usable:
+ filters = f"phase={phase!r}"
+ if session_id is not None:
+ filters += f", session={session_id!r}"
+ raise ValueError(f"no usable AIPerf records found ({filters})")
+
+ origin_ns = min(item[3] for item in usable)
+ events: list[dict[str, Any]] = [
+ {
+ "name": "process_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": 0,
+ "args": {"name": "AIPerf request timeline"},
+ }
+ ]
+
+ grouped: dict[
+ str, list[tuple[dict[str, Any], dict[str, Any], str, float, float]]
+ ] = defaultdict(list)
+ for item in usable:
+ grouped[item[2]].append(item)
+
+ slot_layout = None
+ tree_layout: dict[str, tuple[int, float, float]] = {}
+ if concurrency is not None:
+ placement, tree_layout, _used_slots = _build_slot_layout(usable, concurrency)
+ grouped_by_track: dict[
+ tuple[int, int],
+ list[tuple[dict[str, Any], dict[str, Any], str, float, float]],
+ ] = defaultdict(list)
+ rows_by_slot: defaultdict[int, int] = defaultdict(lambda: 1)
+ for item in usable:
+ slot, row = placement[id(item[0])]
+ grouped_by_track[(slot, row)].append(item)
+ rows_by_slot[slot] = max(rows_by_slot[slot], row + 1)
+
+ summary_tid = 9_000
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": summary_tid,
+ "args": {"name": "all slots • TTFT"},
+ }
+ )
+ events.append(
+ {
+ "name": "thread_sort_index",
+ "ph": "M",
+ "pid": PID,
+ "tid": summary_tid,
+ "args": {"sort_index": -1},
+ }
+ )
+ decode_summary_tid = 9_001
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": decode_summary_tid,
+ "args": {"name": "all slots • decode"},
+ }
+ )
+ events.append(
+ {
+ "name": "thread_sort_index",
+ "ph": "M",
+ "pid": PID,
+ "tid": decode_summary_tid,
+ "args": {"sort_index": -2},
+ }
+ )
+ combined_summary_tid = 9_002
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": combined_summary_tid,
+ "args": {"name": "all slots • prefill+decode"},
+ }
+ )
+ events.append(
+ {
+ "name": "thread_sort_index",
+ "ph": "M",
+ "pid": PID,
+ "tid": combined_summary_tid,
+ "args": {"sort_index": -3},
+ }
+ )
+
+ if inspect_all_slots:
+ analysis_slots = set(range(concurrency))
+ elif inspect_slot is not None:
+ analysis_slots = {inspect_slot}
+ else:
+ analysis_slots = set()
+ special_rows_by_slot: dict[int, tuple[int | None, int]] = {}
+ for slot in analysis_slots:
+ special_row_start = rows_by_slot[slot]
+ if inspect_all_slots:
+ rows_by_slot[slot] += 1
+ special_rows_by_slot[slot] = (None, special_row_start)
+ else:
+ rows_by_slot[slot] += 2
+ special_rows_by_slot[slot] = (
+ special_row_start,
+ special_row_start + 1,
+ )
+
+ for slot in range(concurrency):
+ overview_tid = 10_000 + slot * 100
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": overview_tid,
+ "args": {"name": f"slot {slot:02d} overview", "slot_index": slot},
+ }
+ )
+ events.append(
+ {
+ "name": "thread_sort_index",
+ "ph": "M",
+ "pid": PID,
+ "tid": overview_tid,
+ "args": {"sort_index": slot * 100},
+ }
+ )
+ for row in range(rows_by_slot[slot]):
+ tid = overview_tid + row + 1
+ special_rows = special_rows_by_slot.get(slot)
+ if (
+ special_rows
+ and special_rows[0] is not None
+ and row == special_rows[0]
+ ):
+ track_name = f"slot {slot:02d} wait/TTFB"
+ elif special_rows and row == special_rows[1]:
+ track_name = f"slot {slot:02d} wait∩prefill"
+ else:
+ track_name = f"slot {slot:02d} row {row}"
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": tid,
+ "args": {
+ "name": track_name,
+ "slot_index": slot,
+ "slot_row": row,
+ },
+ }
+ )
+ events.append(
+ {
+ "name": "thread_sort_index",
+ "ph": "M",
+ "pid": PID,
+ "tid": tid,
+ "args": {"sort_index": slot * 100 + row + 1},
+ }
+ )
+
+ for slot, (wait_row, overlap_row) in special_rows_by_slot.items():
+ _append_wait_prefill_analysis(
+ events,
+ usable,
+ placement,
+ inspect_slot=slot,
+ wait_tid=(
+ 10_000 + slot * 100 + wait_row + 1 if wait_row is not None else None
+ ),
+ overlap_tid=10_000 + slot * 100 + overlap_row + 1,
+ origin_ns=origin_ns,
+ )
+ _append_ttft_summary(
+ events,
+ usable,
+ placement,
+ summary_tid=summary_tid,
+ origin_ns=origin_ns,
+ )
+ _append_decode_summary(
+ events,
+ usable,
+ placement,
+ summary_tid=decode_summary_tid,
+ origin_ns=origin_ns,
+ )
+ _append_prefill_decode_summary(
+ events,
+ usable,
+ placement,
+ summary_tid=combined_summary_tid,
+ origin_ns=origin_ns,
+ )
+
+ for tree, (slot, start_ns, end_ns) in tree_layout.items():
+ events.append(
+ _event(
+ f"tree {_short_id(tree)}",
+ ts_us=round((start_ns - origin_ns) / NS_PER_US),
+ tid=10_000 + slot * 100,
+ category="session-tree",
+ args={
+ "root_correlation_id": tree,
+ "slot_index": slot,
+ "tree_start_ns": int(start_ns),
+ "tree_end_ns": int(end_ns),
+ },
+ duration_us=round((end_ns - start_ns) / NS_PER_US),
+ )
+ )
+
+ track_entries = [
+ (
+ f"slot {slot:02d} row {row}",
+ 10_000 + slot * 100 + row + 1,
+ grouped_by_track[(slot, row)],
+ slot,
+ row,
+ )
+ for slot in range(concurrency)
+ for row in range(rows_by_slot[slot])
+ if grouped_by_track[(slot, row)]
+ ]
+ sessions = sorted(tree_layout)
+ session_count = len(tree_layout)
+ slot_count = concurrency
+ slot_layout = placement
+ else:
+ sessions = sorted({item[2] for item in usable})
+ tids = {value: index + 1 for index, value in enumerate(sessions)}
+ track_entries = [
+ (session, tids[session], grouped[session], None, None)
+ for session in sessions
+ ]
+ session_count = len(sessions)
+ slot_count = None
+
+ if concurrency is None:
+ for session_index, session in enumerate(sessions, start=1):
+ events.append(
+ {
+ "name": "thread_name",
+ "ph": "M",
+ "pid": PID,
+ "tid": tids[session],
+ "args": {
+ "name": f"{session_key}={_short_id(session)}",
+ "session_id": session,
+ "session_index": session_index,
+ },
+ }
+ )
+
+ for session, tid, track_items, slot_index, slot_row in track_entries:
+ previous_end_ns: float | None = None
+ for record, metadata, _, start_ns, end_ns in track_items:
+ metrics = record.get("metrics", {})
+ if not isinstance(metrics, dict):
+ metrics = {}
+
+ relative_start_us = round((start_ns - origin_ns) / NS_PER_US)
+ request_duration_us = round((end_ns - start_ns) / NS_PER_US)
+ ttft_ms = _number(metric_value(metrics, "time_to_first_token"))
+ decode_ms = _number(metric_value(metrics, "full_decode_duration"))
+ sending_ms = _number(metric_value(metrics, "http_req_sending"))
+ waiting_ms = _number(metric_value(metrics, "http_req_waiting"))
+ input_tokens = metric_value(metrics, "input_sequence_length")
+ usage_prompt_tokens = metric_value(metrics, "usage_prompt_tokens")
+ prompt_cache_read_tokens = metric_value(
+ metrics, "usage_prompt_cache_read_tokens"
+ )
+ output_tokens = metric_value(metrics, "output_sequence_length")
+ usage_completion_tokens = metric_value(metrics, "usage_completion_tokens")
+ cache_ratio, cache_anomaly = _cache_observation(metrics)
+
+ request_number = metadata.get(
+ "session_num", metadata.get("turn_index", "?")
+ )
+ request_session = (
+ _request_session_key(metadata, _tree_key(metadata))
+ if concurrency is not None
+ else session
+ )
+ request_args = {
+ "session_id": request_session,
+ "root_correlation_id": _tree_key(metadata),
+ "session_num": request_number,
+ "turn_index": metadata.get("turn_index"),
+ "x_request_id": metadata.get("x_request_id"),
+ "conversation_id": metadata.get("conversation_id"),
+ "source_kind": metadata.get("source_kind"),
+ "parent_correlation_id": metadata.get("parent_correlation_id"),
+ "phase": metadata.get("benchmark_phase"),
+ "start_ns": int(start_ns),
+ "end_ns": int(end_ns),
+ "time_to_first_token_ms": ttft_ms,
+ "full_decode_duration_ms": decode_ms,
+ "input_tokens": input_tokens,
+ "usage_prompt_tokens": usage_prompt_tokens,
+ "prompt_cache_read_tokens": prompt_cache_read_tokens,
+ "cache_read_ratio_pct": (
+ 100.0 * cache_ratio if cache_ratio is not None else None
+ ),
+ "cache_anomaly": cache_anomaly,
+ "output_tokens": output_tokens,
+ "usage_completion_tokens": usage_completion_tokens,
+ "error": record.get("error"),
+ }
+ if slot_index is not None:
+ request_args["slot_index"] = slot_index
+ request_args["slot_row"] = slot_row
+ input_label = (
+ f" | input={int(input_tokens)} tok" if input_tokens is not None else ""
+ )
+ output_label = (
+ f" | output={int(output_tokens)} tok"
+ if output_tokens is not None
+ else ""
+ )
+ events.append(
+ _event(
+ f"request #{request_number}{input_label}{output_label}",
+ ts_us=relative_start_us,
+ tid=tid,
+ category="request",
+ args=request_args,
+ duration_us=request_duration_us,
+ color="terrible" if cache_anomaly else None,
+ )
+ )
+
+ if previous_end_ns is not None and start_ns > previous_end_ns:
+ events.append(
+ _event(
+ "session idle",
+ ts_us=round((previous_end_ns - origin_ns) / NS_PER_US),
+ tid=tid,
+ category="session",
+ args={"idle_ms": (start_ns - previous_end_ns) / NS_PER_MS},
+ duration_us=round((start_ns - previous_end_ns) / NS_PER_US),
+ )
+ )
+
+ if sending_ms is not None and sending_ms > 0:
+ events.append(
+ _event(
+ f"HTTP send #{request_number}",
+ ts_us=relative_start_us,
+ tid=tid,
+ category="http",
+ args={"duration_ms": sending_ms},
+ duration_us=round(sending_ms * 1_000),
+ )
+ )
+
+ if ttft_ms is not None and ttft_ms >= 0:
+ first_token_ns = min(end_ns, start_ns + ttft_ms * NS_PER_MS)
+ first_token_us = round((first_token_ns - origin_ns) / NS_PER_US)
+ prefill_us = max(0, round((first_token_ns - start_ns) / NS_PER_US))
+ decode_us = max(0, request_duration_us - prefill_us)
+ events.append(
+ _event(
+ f"prefill #{request_number}",
+ ts_us=relative_start_us,
+ tid=tid,
+ category="inference",
+ args={"time_to_first_token_ms": ttft_ms},
+ duration_us=prefill_us,
+ color="terrible" if cache_anomaly else None,
+ )
+ )
+ events.append(
+ _event(
+ f"decode #{request_number}",
+ ts_us=first_token_us,
+ tid=tid,
+ category="inference",
+ args={"full_decode_duration_ms": decode_ms},
+ duration_us=decode_us,
+ )
+ )
+ events.append(
+ _event(
+ f"first token #{request_number}",
+ ts_us=first_token_us,
+ tid=tid,
+ category="milestone",
+ args={"time_to_first_token_ms": ttft_ms},
+ )
+ )
+
+ if waiting_ms is not None and waiting_ms > 0:
+ wait_start_ns = start_ns + (sending_ms or 0) * NS_PER_MS
+ events.append(
+ _event(
+ f"HTTP wait/TTFB #{request_number}",
+ ts_us=round((wait_start_ns - origin_ns) / NS_PER_US),
+ tid=tid,
+ category="http",
+ args={"duration_ms": waiting_ms},
+ duration_us=round(waiting_ms * 1_000),
+ )
+ )
+
+ ack_ns = _number(metadata.get("request_ack_ns"))
+ if ack_ns is not None:
+ events.append(
+ _event(
+ f"server ack #{request_number}",
+ ts_us=round((ack_ns - origin_ns) / NS_PER_US),
+ tid=tid,
+ category="milestone",
+ args={"ack_ns": int(ack_ns)},
+ )
+ )
+ events.append(
+ _event(
+ f"response complete #{request_number}",
+ ts_us=round((end_ns - origin_ns) / NS_PER_US),
+ tid=tid,
+ category="milestone",
+ args={"end_ns": int(end_ns)},
+ )
+ )
+ previous_end_ns = max(previous_end_ns or end_ns, end_ns)
+
+ return {
+ "displayTimeUnit": "ms",
+ "traceEvents": events,
+ "metadata": {
+ "format": "AIPerf profile_export.jsonl → Chrome Trace Event",
+ "phase": phase,
+ "session_key": session_key,
+ "session_id": session_id,
+ "origin_ns": int(origin_ns),
+ "record_count": len(usable),
+ "session_count": session_count,
+ "slot_count": slot_count,
+ "tree_count": len(tree_layout) if slot_layout is not None else None,
+ "inspect_slot": inspect_slot,
+ },
+ }
+
+
+def list_sessions(
+ records: Iterable[dict[str, Any]],
+ *,
+ phase: str | None,
+ session_key: str,
+) -> list[tuple[str, int]]:
+ counts: defaultdict[str, int] = defaultdict(int)
+ for record in records:
+ metadata = record.get("metadata", {})
+ if not isinstance(metadata, dict):
+ continue
+ if phase is not None and metadata.get("benchmark_phase") != phase:
+ continue
+ counts[_session_value(metadata, session_key)] += 1
+ return sorted(counts.items(), key=lambda item: (-item[1], item[0]))
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("input", type=Path, help="AIPerf profile_export.jsonl")
+ parser.add_argument(
+ "-o",
+ "--output",
+ type=Path,
+ help="output trace JSON (default: .trace.json)",
+ )
+ parser.add_argument(
+ "--phase",
+ default="profiling",
+ help="benchmark phase to include; use 'all' to include every phase",
+ )
+ parser.add_argument(
+ "--session-key",
+ choices=("root_correlation_id", "x_correlation_id", "conversation_id"),
+ default="root_correlation_id",
+ help="metadata field used as a trace lane/session",
+ )
+ parser.add_argument(
+ "--session-id",
+ help="only export one session ID; use --list-sessions to discover IDs",
+ )
+ parser.add_argument(
+ "--concurrency",
+ type=int,
+ help="use fixed reusable root-tree slots (for example, 48)",
+ )
+ parser.add_argument(
+ "--inspect-slot",
+ type=int,
+ help="add wait/TTFB and wait∩prefill tracks after this slot",
+ )
+ parser.add_argument(
+ "--inspect-all-slots",
+ action="store_true",
+ help="add wait/TTFB and wait∩prefill tracks after every slot",
+ )
+ parser.add_argument(
+ "--list-sessions",
+ action="store_true",
+ help="list session IDs and record counts, then exit",
+ )
+ args = parser.parse_args()
+
+ records = load_records(args.input)
+ phase = None if args.phase == "all" else args.phase
+ if args.list_sessions:
+ for session, count in list_sessions(
+ records, phase=phase, session_key=args.session_key
+ ):
+ print(f"{count:6d} {session}")
+ return
+
+ output = args.output or args.input.with_name(f"{args.input.stem}.trace.json")
+ trace = build_trace(
+ records,
+ phase=phase,
+ session_key=args.session_key,
+ session_id=args.session_id,
+ concurrency=args.concurrency,
+ inspect_slot=args.inspect_slot,
+ inspect_all_slots=args.inspect_all_slots,
+ )
+ output.write_text(json.dumps(trace, separators=(",", ":")), encoding="utf-8")
+ print(
+ f"wrote {output} "
+ f"({trace['metadata']['record_count']} requests, "
+ f"{trace['metadata']['session_count']} sessions)"
+ )
+
+
+if __name__ == "__main__":
+ main()
diff --git a/.github/scripts/atomesh/pd_server_atom.sh b/.github/scripts/atomesh/pd_server_atom.sh
index f78eca6aac..108bd68b0b 100644
--- a/.github/scripts/atomesh/pd_server_atom.sh
+++ b/.github/scripts/atomesh/pd_server_atom.sh
@@ -852,6 +852,23 @@ print(f"[aiperf] dashboard json: {dst}")
PY
}
+write_aiperf_chrome_trace() {
+ local out_dir="$1"
+ local generator="${ATOMESH_SCRIPT_DIR}/../generate_aiperf_traces.py"
+ local jsonl="${out_dir}/profile_export.jsonl"
+ if [[ ! -f "${jsonl}" ]]; then
+ echo "[aiperf] skip chrome trace: ${jsonl} was not produced"
+ return 0
+ fi
+ if [[ ! -f "${generator}" ]]; then
+ echo "[aiperf] skip chrome trace: ${generator} not found"
+ return 0
+ fi
+ echo "[aiperf] converting ${jsonl} to Perfetto/Chrome trace"
+ python3 "${generator}" "${out_dir}" \
+ || echo "[aiperf] WARNING: chrome trace conversion failed for ${out_dir}" >&2
+}
+
run_aiperf_agentic_benchmark() {
ensure_aiperf
@@ -929,6 +946,7 @@ run_aiperf_agentic_benchmark() {
return 1
fi
write_aiperf_dashboard_json "${aiperf_json}" "${dashboard_json}" "${conc}"
+ write_aiperf_chrome_trace "${out_dir}"
done
}
diff --git a/.github/scripts/generate_aiperf_traces.py b/.github/scripts/generate_aiperf_traces.py
new file mode 100755
index 0000000000..a123956205
--- /dev/null
+++ b/.github/scripts/generate_aiperf_traces.py
@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""Convert AIPerf profile_export.jsonl artifacts into Perfetto/Chrome traces.
+
+Used by benchmark workflows after a client run so CI can upload a dedicated
+artifact and print download links. Stdlib only: summarize jobs run on
+ubuntu-latest without extra Python deps.
+"""
+
+from __future__ import annotations
+
+import argparse
+import gzip
+import importlib.util
+import json
+import re
+import shutil
+import sys
+from pathlib import Path
+from typing import Any
+
+RECORDS_FILENAME = "profile_export.jsonl"
+AIPERF_SUMMARY_FILENAME = "profile_export_aiperf.json"
+CONVERTER_PATH = Path(__file__).resolve().with_name("aiperf_to_chrome_trace.py")
+DIR_CONCURRENCY_RE = re.compile(r"-c(\d+)$")
+
+
+def load_converter(path: Path | None = None) -> Any:
+ converter = path or CONVERTER_PATH
+ if not converter.is_file():
+ raise FileNotFoundError(f"AIPerf chrome-trace converter not found: {converter}")
+ spec = importlib.util.spec_from_file_location("aiperf_to_chrome_trace", converter)
+ if spec is None or spec.loader is None:
+ raise ImportError(f"cannot load converter from {converter}")
+ module = importlib.util.module_from_spec(spec)
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+def _positive_int(value: Any) -> int | None:
+ if isinstance(value, bool):
+ return None
+ if isinstance(value, int) and value >= 1:
+ return value
+ if isinstance(value, float) and value.is_integer() and value >= 1:
+ return int(value)
+ if isinstance(value, str) and value.isdigit():
+ parsed = int(value)
+ return parsed if parsed >= 1 else None
+ return None
+
+
+def concurrency_from_aiperf_summary(path: Path) -> int | None:
+ """Prefer the configured profiling concurrency from AIPerf's JSON summary."""
+ try:
+ data = json.loads(path.read_text(encoding="utf-8"))
+ except (OSError, json.JSONDecodeError):
+ return None
+ if not isinstance(data, dict):
+ return None
+ nested = []
+ for key in ("input_config", "config", "metadata"):
+ value = data.get(key)
+ if isinstance(value, dict):
+ nested.append(value)
+ for blob in (data, *nested):
+ for key in ("concurrency", "max_concurrency", "configured_concurrency"):
+ parsed = _positive_int(blob.get(key))
+ if parsed is not None:
+ return parsed
+ return None
+
+
+def infer_concurrency(jsonl: Path) -> int | None:
+ summary = jsonl.with_name(AIPERF_SUMMARY_FILENAME)
+ if summary.is_file():
+ parsed = concurrency_from_aiperf_summary(summary)
+ if parsed is not None:
+ return parsed
+ match = DIR_CONCURRENCY_RE.search(jsonl.parent.name)
+ if match:
+ return int(match.group(1))
+ return None
+
+
+def trace_stem(jsonl: Path, concurrency: int | None) -> str:
+ if concurrency is None:
+ return f"{jsonl.stem}.trace.json"
+ return f"{jsonl.stem}.c{concurrency}.slots.trace.json"
+
+
+def unique_trace_name(jsonl: Path, concurrency: int | None) -> str:
+ """Prefix the artifact directory so merged CI uploads do not collide."""
+ return f"{jsonl.parent.name}.{trace_stem(jsonl, concurrency)}"
+
+
+def iter_profile_exports(root: Path) -> list[Path]:
+ return sorted(path for path in root.rglob(RECORDS_FILENAME) if path.is_file())
+
+
+def write_gzip(src: Path, dst: Path) -> None:
+ dst.parent.mkdir(parents=True, exist_ok=True)
+ with src.open("rb") as reader, gzip.open(dst, "wb", compresslevel=6) as writer:
+ shutil.copyfileobj(reader, writer)
+
+
+def find_existing_trace(jsonl: Path, concurrency: int | None) -> Path | None:
+ stem = jsonl.with_name(trace_stem(jsonl, concurrency))
+ for candidate in (Path(str(stem) + ".gz"), stem):
+ if candidate.is_file():
+ return candidate
+ return None
+
+
+def _result_payload(
+ jsonl: Path,
+ destination: Path,
+ concurrency: int | None,
+ metadata: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+ metadata = metadata or {}
+ return {
+ "input": str(jsonl),
+ "output": str(destination),
+ "concurrency": concurrency,
+ "record_count": metadata.get("record_count"),
+ "session_count": metadata.get("session_count"),
+ "slot_count": metadata.get("slot_count"),
+ "tree_count": metadata.get("tree_count"),
+ "phase": metadata.get("phase"),
+ "bytes": destination.stat().st_size,
+ }
+
+
+def _gzip_destination(path: Path) -> Path:
+ return path if str(path).endswith(".gz") else Path(str(path) + ".gz")
+
+
+def convert_export(
+ jsonl: Path,
+ *,
+ converter: Any,
+ output_dir: Path | None,
+ gzip_output: bool,
+) -> dict[str, Any]:
+ concurrency = infer_concurrency(jsonl)
+ if output_dir is not None:
+ destination = output_dir / unique_trace_name(jsonl, concurrency)
+ else:
+ destination = jsonl.with_name(trace_stem(jsonl, concurrency))
+ destination.parent.mkdir(parents=True, exist_ok=True)
+
+ existing = find_existing_trace(jsonl, concurrency)
+ if existing is not None:
+ if output_dir is None:
+ return _result_payload(jsonl, existing, concurrency)
+ if gzip_output and existing.suffix != ".gz":
+ destination = _gzip_destination(destination)
+ write_gzip(existing, destination)
+ else:
+ if gzip_output:
+ destination = _gzip_destination(destination)
+ shutil.copy2(existing, destination)
+ return _result_payload(jsonl, destination, concurrency)
+
+ records = converter.load_records(jsonl)
+ trace = converter.build_trace(
+ records,
+ phase=None,
+ concurrency=concurrency,
+ inspect_all_slots=concurrency is not None,
+ )
+ destination.write_text(json.dumps(trace, separators=(",", ":")), encoding="utf-8")
+ if gzip_output:
+ gzip_path = _gzip_destination(destination)
+ write_gzip(destination, gzip_path)
+ destination.unlink()
+ destination = gzip_path
+ return _result_payload(jsonl, destination, concurrency, trace.get("metadata", {}))
+
+
+def write_index(
+ results: list[dict[str, Any]],
+ index_path: Path,
+ *,
+ artifact_url: str | None,
+ run_url: str | None,
+) -> None:
+ lines = [
+ "### AIPerf Chrome Traces",
+ "",
+ (
+ "Download the `*.trace.json.gz` files from this run, then open them in "
+ "[Perfetto UI](https://ui.perfetto.dev) (`Open trace file`) or "
+ "`chrome://tracing` (gunzip first for Chrome)."
+ ),
+ "",
+ ]
+ if artifact_url:
+ lines.append(f"- Artifact download: {artifact_url}")
+ if run_url:
+ lines.append(f"- Workflow run (all artifacts): {run_url}")
+ if artifact_url or run_url:
+ lines.append("")
+ if not results:
+ lines.append(
+ "No `profile_export.jsonl` files were found; no traces were generated."
+ )
+ index_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+ return
+ lines.extend(
+ [
+ "| Trace | Concurrency | Requests | Trees | Size (bytes) |",
+ "| --- | ---: | ---: | ---: | ---: |",
+ ]
+ )
+ for item in results:
+ lines.append(
+ "| `{name}` | {conc} | {records} | {trees} | {size} |".format(
+ name=Path(item["output"]).name,
+ conc=item["concurrency"] if item["concurrency"] is not None else "--",
+ records=(
+ item["record_count"] if item["record_count"] is not None else "--"
+ ),
+ trees=item["tree_count"] if item["tree_count"] is not None else "--",
+ size=item["bytes"],
+ )
+ )
+ index_path.parent.mkdir(parents=True, exist_ok=True)
+ index_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument(
+ "result_dir", type=Path, help="Directory containing AIPerf artifacts"
+ )
+ parser.add_argument(
+ "--output-dir",
+ type=Path,
+ help="Write traces here (default: next to each profile_export.jsonl)",
+ )
+ parser.add_argument(
+ "--index",
+ type=Path,
+ help="Markdown index path (default: /TRACE_INDEX.md)",
+ )
+ parser.add_argument(
+ "--converter", type=Path, help="Override aiperf_to_chrome_trace.py path"
+ )
+ parser.add_argument(
+ "--run-url", default=None, help="Workflow run URL for the index"
+ )
+ parser.add_argument(
+ "--artifact-url", default=None, help="Artifact download URL for the index"
+ )
+ parser.add_argument(
+ "--no-gzip", action="store_true", help="Leave traces uncompressed"
+ )
+ args = parser.parse_args()
+
+ root = args.result_dir
+ if not root.is_dir():
+ print(f"skip chrome traces: result directory not found: {root}")
+ return 0
+
+ converter = load_converter(args.converter)
+ exports = iter_profile_exports(root)
+ results: list[dict[str, Any]] = []
+ failures = 0
+ for jsonl in exports:
+ try:
+ results.append(
+ convert_export(
+ jsonl,
+ converter=converter,
+ output_dir=args.output_dir,
+ gzip_output=not args.no_gzip,
+ )
+ )
+ print(f"wrote {results[-1]['output']}")
+ except (OSError, TypeError, ValueError) as exc:
+ failures += 1
+ print(f"WARNING: failed to convert {jsonl}: {exc}", file=sys.stderr)
+
+ index_path = args.index
+ if index_path is None:
+ base = args.output_dir if args.output_dir is not None else root
+ index_path = base / "TRACE_INDEX.md"
+ write_index(
+ results,
+ index_path,
+ artifact_url=args.artifact_url,
+ run_url=args.run_url,
+ )
+ print(
+ f"converted {len(results)}/{len(exports)} AIPerf export(s); index={index_path}"
+ )
+ return 0 if failures == 0 else 1
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/.github/workflows/atomesh-benchmark.yaml b/.github/workflows/atomesh-benchmark.yaml
index c02afa1389..1204a5c181 100644
--- a/.github/workflows/atomesh-benchmark.yaml
+++ b/.github/workflows/atomesh-benchmark.yaml
@@ -1290,6 +1290,15 @@ jobs:
cat "${SUMMARY_PATH}"
+ - name: Upload AIPerf Perfetto traces
+ if: ${{ always() && matrix.benchmark.kind == 'aiperf_agentic' }}
+ uses: ./.github/actions/upload-aiperf-traces
+ with:
+ result-dir: atomesh-results/${{ matrix.id }}
+ artifact-name: aiperf-chrome-traces-${{ matrix.id }}
+ output-dir: aiperf-traces/${{ matrix.id }}
+ run-url: https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}
+
- name: Upload model benchmark artifacts
if: always()
uses: actions/upload-artifact@v7
@@ -1345,6 +1354,15 @@ jobs:
echo "No successful ATOMesh benchmark result JSONs were produced; dashboard publish will be skipped." >> "$GITHUB_STEP_SUMMARY"
fi
+ - name: Upload AIPerf Perfetto traces
+ if: always()
+ uses: ./.github/actions/upload-aiperf-traces
+ with:
+ result-dir: atomesh-results
+ artifact-name: aiperf-chrome-traces-${{ github.run_id }}
+ output-dir: aiperf-traces
+ run-url: https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }}
+
- name: Upload benchmark summary
if: ${{ steps.summary.outputs.has_results == 'true' }}
uses: actions/upload-artifact@v7