From eb6327c45414ee317f697d4733725cdbc47aadaf Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 19 Aug 2026 14:53:33 -0700 Subject: [PATCH 1/2] feat(power): wire H200 multinode AgentX power --- .github/workflows/test-process-result.yml | 7 +- benchmarks/benchmark_lib.sh | 47 ++- runners/inject_srt_power_concurrencies.py | 77 +++++ runners/launch_h200-dgxc-slurm.sh | 105 +++++- utils/agentic/aggregation/power_adapter.py | 311 +++++++++++++++++- .../agentic/aggregation/test_power_adapter.py | 246 ++++++++++++++ .../aggregation/test_power_lifecycle.py | 48 +++ utils/aggregate_power_multinode.py | 26 +- utils/test_aggregate_power_multinode.py | 60 ++++ utils/test_h200_power_official_contract.py | 129 ++++++++ utils/test_inject_srt_power_concurrencies.py | 56 ++++ 11 files changed, 1097 insertions(+), 15 deletions(-) create mode 100644 runners/inject_srt_power_concurrencies.py create mode 100644 utils/test_h200_power_official_contract.py create mode 100644 utils/test_inject_srt_power_concurrencies.py diff --git a/.github/workflows/test-process-result.yml b/.github/workflows/test-process-result.yml index 97169ae65b..eac9d2e500 100644 --- a/.github/workflows/test-process-result.yml +++ b/.github/workflows/test-process-result.yml @@ -8,10 +8,13 @@ on: - '.github/workflows/e2e-tests.yml' - '.github/workflows/test-process-result.yml' - 'benchmarks/benchmark_lib.sh' + - 'benchmarks/multi_node/srt-slurm-recipes/sglang/glm5.2/agentic/disagg-h200-2p2d-pcp8-tp8-dp8-mtp.yaml' - 'benchmarks/multi_node/srt-slurm-recipes/sglang/qwen3.5/gb200-fp8/8k1k/1p1d-tp4-tp4.yaml' - 'benchmarks/multi_node/srt-slurm-recipes/sglang/qwen3.5/gb300-fp8/8k1k/1p1d-tp4-tp4.yaml' - 'runners/launch_gb200-nv.sh' - 'runners/launch_gb300-nv.sh' + - 'runners/launch_h200-dgxc-slurm.sh' + - 'runners/inject_srt_power_concurrencies.py' - 'utils/aggregate_power.py' - 'utils/aggregate_power_multinode.py' - 'utils/agentic/aggregation/power_adapter.py' @@ -26,6 +29,8 @@ on: - 'utils/test_aggregate_power_multinode.py' - 'utils/test_gb200_power_official_contract.py' - 'utils/test_gb300_power_official_contract.py' + - 'utils/test_h200_power_official_contract.py' + - 'utils/test_inject_srt_power_concurrencies.py' - 'utils/test_process_result.py' permissions: @@ -54,4 +59,4 @@ jobs: - name: Run pytest run: | cd utils - python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/test_power_adapter.py agentic/aggregation/test_power_lifecycle.py agentic/aggregation/test_process_agentic_result.py test_gb200_power_official_contract.py test_gb300_power_official_contract.py test_process_result.py -v + python -m pytest test_aggregate_power.py test_aggregate_power_multinode.py agentic/aggregation/test_power_adapter.py agentic/aggregation/test_power_lifecycle.py agentic/aggregation/test_process_agentic_result.py test_gb200_power_official_contract.py test_gb300_power_official_contract.py test_h200_power_official_contract.py test_inject_srt_power_concurrencies.py test_process_result.py -v diff --git a/benchmarks/benchmark_lib.sh b/benchmarks/benchmark_lib.sh index a4fe70803f..9cedf29fc1 100644 --- a/benchmarks/benchmark_lib.sh +++ b/benchmarks/benchmark_lib.sh @@ -2190,11 +2190,16 @@ run_agentic_replay_and_write_outputs() ( local validation_rc local power_rc=0 local agentx_power_enabled=0 + local agentx_multinode_power_enabled=0 local agentx_monitor_stopped=1 case "${ENABLE_AGENTX_POWER:-1}" in 1|true|TRUE|yes|YES) - if [ "${IS_MULTINODE:-false}" != "true" ]; then + if [ "${IS_MULTINODE:-false}" = "true" ]; then + if [ -n "${SRT_MEASUREMENT_WINDOW_DIR:-}" ]; then + agentx_multinode_power_enabled=1 + fi + else agentx_power_enabled=1 fi ;; @@ -2207,11 +2212,42 @@ run_agentic_replay_and_write_outputs() ( fi } - if [ "$agentx_power_enabled" = "1" ]; then + _write_agentx_multinode_window() { + local state="$1" + local -a power_args + power_args=( + --result-dir "$result_dir" + --concurrency "${CONC:?CONC must be set for multinode AgentX power}" + --write-multinode-window "$state" + ) + case "${REQUIRE_POWER:-0}" in + 1|true|TRUE|yes|YES) power_args+=(--require-power) ;; + esac + ( + cd "$INFMAX_CONTAINER_WORKSPACE" + "$AIPERF_PYTHON" -m utils.agentic.aggregation.power_adapter "${power_args[@]}" + ) + } + + if [ "$agentx_power_enabled" = "1" ] || [ "$agentx_multinode_power_enabled" = "1" ]; then # AIPerf currently exports naive local datetimes while SMI emits the # same host wall clock. Capture the launch-time offset so the adapter # can attach it explicitly before normalizing the profiling window. date +%z > "$result_dir/agentic_power_timezone_offset.txt" + fi + + if [ "$agentx_multinode_power_enabled" = "1" ]; then + set +e + _write_agentx_multinode_window running + power_rc=$? + set -e + if [ "$power_rc" -ne 0 ]; then + echo "ERROR: failed to publish the AgentX formal running power window" >&2 + return "$power_rc" + fi + fi + + if [ "$agentx_power_enabled" = "1" ]; then start_gpu_monitor --output "$result_dir/gpu_metrics.csv" agentx_monitor_stopped=0 # This function runs in a subshell, so these handlers cannot replace @@ -2238,6 +2274,13 @@ run_agentic_replay_and_write_outputs() ( write_agentic_result_json "$result_dir" + if [ "$agentx_multinode_power_enabled" = "1" ] && [ "$replay_rc" -eq 0 ]; then + set +e + _write_agentx_multinode_window completed + power_rc=$? + set -e + fi + if [ "$agentx_power_enabled" = "1" ]; then local expected_num_gpus local -a power_args diff --git a/runners/inject_srt_power_concurrencies.py b/runners/inject_srt_power_concurrencies.py new file mode 100644 index 0000000000..0e16321c26 --- /dev/null +++ b/runners/inject_srt_power_concurrencies.py @@ -0,0 +1,77 @@ +#!/usr/bin/env python3 +"""Inject exact matrix concurrencies into a runtime srt-slurm recipe copy.""" + +from __future__ import annotations + +import argparse +import os +import tempfile +from pathlib import Path +from typing import Any + +import yaml + + +def _validate_concurrencies(concurrencies: list[Any]) -> list[int]: + if ( + not concurrencies + or any(isinstance(value, bool) or not isinstance(value, int) for value in concurrencies) + or any(value <= 0 for value in concurrencies) + or len(set(concurrencies)) != len(concurrencies) + ): + raise ValueError("concurrencies must be positive unique integers") + return concurrencies + + +def inject_concurrencies(recipe_path: Path, concurrencies: list[Any]) -> None: + """Atomically set benchmark.concurrencies on a disposable recipe copy.""" + values = _validate_concurrencies(concurrencies) + try: + recipe = yaml.safe_load(recipe_path.read_text(encoding="utf-8")) + except (OSError, yaml.YAMLError) as exc: + raise ValueError(f"failed to load recipe: {exc}") from exc + if not isinstance(recipe, dict) or not isinstance(recipe.get("benchmark"), dict): + raise ValueError("recipe must contain a benchmark mapping") + + recipe["benchmark"]["concurrencies"] = values + fd, temporary_name = tempfile.mkstemp( + dir=recipe_path.parent, + prefix=f".{recipe_path.name}.", + text=True, + ) + temporary_path = Path(temporary_name) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + yaml.safe_dump(recipe, handle, sort_keys=False) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temporary_path, recipe_path) + except BaseException: + temporary_path.unlink(missing_ok=True) + raise + + +def _positive_integer(raw: str) -> int: + try: + value = int(raw) + except ValueError as exc: + raise argparse.ArgumentTypeError("must be a positive integer") from exc + if value <= 0 or str(value) != raw: + raise argparse.ArgumentTypeError("must be a canonical positive integer") + return value + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("recipe", type=Path) + parser.add_argument("concurrencies", nargs="+", type=_positive_integer) + args = parser.parse_args() + try: + inject_concurrencies(args.recipe, args.concurrencies) + except ValueError as exc: + parser.error(str(exc)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/runners/launch_h200-dgxc-slurm.sh b/runners/launch_h200-dgxc-slurm.sh index a0ff8b14b1..275075d5d9 100755 --- a/runners/launch_h200-dgxc-slurm.sh +++ b/runners/launch_h200-dgxc-slurm.sh @@ -7,6 +7,12 @@ SLURM_ACCOUNT="sa-shared" HF_HUB_CACHE_MOUNT="${HF_HUB_CACHE_MOUNT:-/models/gharunners/hf-hub-cache}" AIPERF_MMAP_CACHE_HOST_PATH="${AIPERF_MMAP_CACHE_HOST_PATH:-/home/sa-shared/gharunners/ai-perf-cache}" +# Immutable producer prerequisite for the GLM-5.2 AgentX lane. This fork is +# intentionally long-lived; update the SHA only after reviewing a new fork +# commit and re-running the H200 hardware gate. +POWER_SRT_SLURM_URL="https://github.com/edwingao28/srt-slurm.git" +POWER_SRT_SLURM_PIN="a1b8c7af10c00e5ea40074aebdc0086189bbc064" + set -x source "$(dirname "${BASH_SOURCE[0]}")/slurm_utils.sh" @@ -20,6 +26,33 @@ if [[ "$IS_MULTINODE" == "true" ]]; then CONFIG_PATH="${CONFIG_FILE%%:*}" LOCAL_CONFIG_FILE="$GITHUB_WORKSPACE/benchmarks/multi_node/srt-slurm-recipes/${CONFIG_PATH#recipes/}" + # The producer pin decision is recipe-driven. Upstream-only recipes have + # no workspace mirror and remain non-power. + USES_DCGM_POWER=0 + _RECIPE_REL="${CONFIG_FILE%%:*}" + _RECIPE_SRC="$GITHUB_WORKSPACE/benchmarks/multi_node/srt-slurm-recipes/${_RECIPE_REL#recipes/}" + if [[ -n "$CONFIG_FILE" && -f "$_RECIPE_SRC" ]] && awk ' + /^telemetry:/ { t = 1; next } + t && /^[^ ]/ { t = 0 } + t && /^ provider: dcgm-power$/ { p = 1 } + t && /^ enabled: true$/ { e = 1 } + END { exit !(p && e) } + ' "$_RECIPE_SRC"; then + USES_DCGM_POWER=1 + fi + + # PR-A supports exactly the GLM-5.2 FP8 AgentX topology. Future recipes + # must earn a separate cluster smoke instead of inheriting this lane. + if [[ "$USES_DCGM_POWER" == "1" && ( + "$IS_AGENTIC" != "1" || + "$FRAMEWORK" != "dynamo-sglang" || + "$MODEL_PREFIX" != "glm5.2" || + "$PRECISION" != "fp8" + ) ]]; then + echo "Error: H200 dcgm-power is validated only for AgentX dynamo-sglang glm5.2/fp8" >&2 + exit 1 + fi + # MODEL_PATH: Override with pre-downloaded paths on H200 runner # The yaml files specify HuggingFace model IDs for portability, but we use # local paths to avoid repeated downloading on the shared H200 cluster. @@ -75,10 +108,15 @@ if [[ "$IS_MULTINODE" == "true" ]]; then fi if [[ $IS_AGENTIC == "1" && $FRAMEWORK == "dynamo-sglang" && $MODEL_PREFIX == "glm5.2" ]]; then - # v1.0.44 includes the AgentX custom benchmark integration and passes - # every logical SGLang worker's Prometheus URL to AIPerf. - git clone --branch v1.0.44 --single-branch https://github.com/NVIDIA/srt-slurm.git "$SRT_REPO_DIR" + # The pinned fork carries the v1.0.44 AgentX lifecycle plus the formal + # custom-benchmark dcgm-power contract needed by PR-A. + git clone "$POWER_SRT_SLURM_URL" "$SRT_REPO_DIR" cd "$SRT_REPO_DIR" + git checkout "$POWER_SRT_SLURM_PIN" || exit 1 + test "$(git rev-parse HEAD)" = "$POWER_SRT_SLURM_PIN" || { echo "Error: srt-slurm HEAD does not match POWER_SRT_SLURM_PIN=$POWER_SRT_SLURM_PIN" >&2; exit 1; } + if [[ "$USES_DCGM_POWER" == "1" ]]; then + git rev-parse HEAD > "$GITHUB_WORKSPACE/power-producer-sha.txt" + fi elif [[ $IS_AGENTIC == "1" && $FRAMEWORK == "vllm" && $MODEL_PREFIX == "kimik3" ]]; then git clone https://github.com/functionstackx/srt-slurm-nv.git "$SRT_REPO_DIR" cd "$SRT_REPO_DIR" @@ -162,6 +200,32 @@ if [[ "$IS_MULTINODE" == "true" ]]; then " fi + if [[ "$USES_DCGM_POWER" == "1" ]]; then + DCGM_EXPORTER_IMAGE="nvcr.io/nvidia/k8s/dcgm-exporter:4.6.0-4.8.3-distroless" + DCGM_EXPORTER_SQSH="/data/gharunners/containers/$(echo "$DCGM_EXPORTER_IMAGE" | sed 's/[\/:@#]/_/g').sqsh" + if ! unsquashfs -l "$DCGM_EXPORTER_SQSH" >/dev/null 2>&1; then + DCGM_EXPORTER_LOCK="${DCGM_EXPORTER_SQSH}.lock" + mkdir -p "$(dirname "$DCGM_EXPORTER_SQSH")" + srun --partition="$SLURM_PARTITION" --account="$SLURM_ACCOUNT" \ + --nodes=1 --ntasks=1 --time=30 --job-name="$RUNNER_NAME" \ + bash -c " + set -euo pipefail + exec 9>\"$DCGM_EXPORTER_LOCK\" + flock -w 1800 9 + if unsquashfs -l \"$DCGM_EXPORTER_SQSH\" >/dev/null 2>&1; then + exit 0 + fi + rm -f \"$DCGM_EXPORTER_SQSH\" + export ENROOT_CACHE_PATH=\${HOME}/.cache/enroot + mkdir -p \"\$ENROOT_CACHE_PATH\" + enroot import -o \"$DCGM_EXPORTER_SQSH\" docker://$DCGM_EXPORTER_IMAGE + " + fi + test -r "$DCGM_EXPORTER_SQSH" || { echo "Error: DCGM exporter squash is not readable: $DCGM_EXPORTER_SQSH" >&2; exit 1; } + unsquashfs -l "$DCGM_EXPORTER_SQSH" >/dev/null || { echo "Error: DCGM exporter squash is invalid: $DCGM_EXPORTER_SQSH" >&2; exit 1; } + sha256sum "$DCGM_EXPORTER_SQSH" > "$GITHUB_WORKSPACE/exporter-image.sha256" + fi + export ISL="$ISL" export OSL="$OSL" export EVAL_ONLY="${EVAL_ONLY:-false}" @@ -213,6 +277,11 @@ use_exclusive_sbatch_directive: false ${DEFAULT_MOUNTS_BLOCK} EOF + if [[ "$USES_DCGM_POWER" == "1" ]]; then + sed -i "/^ nginx-sqsh:/a\\ dcgm-exporter: ${DCGM_EXPORTER_SQSH}" srtslurm.yaml + grep -q "^ dcgm-exporter: " srtslurm.yaml || { echo "Error: dcgm-exporter injection failed: nginx-sqsh anchor not found in srtslurm.yaml" >&2; exit 1; } + fi + echo "Generated srtslurm.yaml:" cat srtslurm.yaml @@ -224,6 +293,12 @@ EOF cp "$LOCAL_CONFIG_FILE" "$CONFIG_PATH" fi + if [[ "$USES_DCGM_POWER" == "1" ]]; then + read -r -a POWER_CONCURRENCIES <<< "$CONC_LIST" + python "$GITHUB_WORKSPACE/runners/inject_srt_power_concurrencies.py" \ + "$CONFIG_PATH" "${POWER_CONCURRENCIES[@]}" + fi + # Export eval-related env vars for srt-slurm post-benchmark eval export INFMAX_WORKSPACE="$GITHUB_WORKSPACE" @@ -272,6 +347,30 @@ EOF echo "Found logs directory: $LOGS_DIR" + if [[ "$USES_DCGM_POWER" == "1" ]]; then + POWER_LOGS_ROOT=$(cd "$LOGS_DIR" && pwd -P) + read -r -a POWER_CONCURRENCIES <<< "$CONC_LIST" + for concurrency in "${POWER_CONCURRENCIES[@]}"; do + power_args=( + --result-dir "$POWER_LOGS_ROOT/agentic/conc_${concurrency}" + --agg-result "$GITHUB_WORKSPACE/${RESULT_FILENAME}_conc${concurrency}.json" + --power-dir "$POWER_LOGS_ROOT/power" + --logs-root "$POWER_LOGS_ROOT" + --expected-producer-sha "$POWER_SRT_SLURM_PIN" + ) + case "${REQUIRE_POWER:-0}" in + 1|true|TRUE|yes|YES) power_args+=(--require-power) ;; + esac + ( + cd "$GITHUB_WORKSPACE" + python -m utils.agentic.aggregation.power_adapter "${power_args[@]}" + ) || exit 1 + done + mkdir -p "$LOGS_DIR/power" + cp "$GITHUB_WORKSPACE/exporter-image.sha256" "$LOGS_DIR/power/exporter-image.sha256" + cp "$GITHUB_WORKSPACE/power-producer-sha.txt" "$LOGS_DIR/power/power-producer-sha.txt" + fi + cp -r "$LOGS_DIR" "$GITHUB_WORKSPACE/LOGS" bundle_server_logs "$LOGS_DIR" "$GITHUB_WORKSPACE/multinode_server_logs.tar.gz" diff --git a/utils/agentic/aggregation/power_adapter.py b/utils/agentic/aggregation/power_adapter.py index 0e98debbc9..215df15a51 100644 --- a/utils/agentic/aggregation/power_adapter.py +++ b/utils/agentic/aggregation/power_adapter.py @@ -8,22 +8,34 @@ import os import re import sys +import time from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any from utils.aggregate_power import ( + POWER_METRIC_SCHEMA_VERSION, _empty_integration, _patch_power_result, _validation_payload, _write_json_atomic, ) from utils.aggregate_power import run as run_power +from utils.aggregate_power_multinode import _ALL_POWER_METRIC_KEYS +from utils.aggregate_power_multinode import run as run_multinode_power from .process_agentic_result import _resolve_artifact_dir from .request_metrics import extract_per_record_ints, load_aggregate, load_records _UTC_OFFSET_RE = re.compile(r"^([+-])(\d{2}):?(\d{2})$") +_COMMIT_SHA_RE = re.compile(r"^[0-9a-f]{40}$") +_MULTINODE_WINDOW_STEM_RE = re.compile(r"^agentic_power_concurrency_([1-9][0-9]*)$") +_FORMAL_WINDOW_ENV = ( + "SRT_MEASUREMENT_WINDOW_DIR", + "SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE", + "SRT_MEASUREMENT_WINDOW_CONCURRENCIES", + "SRT_MEASUREMENT_WINDOW_RESULT_ROOT", +) def _captured_timezone(result_dir: Path) -> tuple[timezone | None, str | None]: @@ -222,17 +234,314 @@ def run_agentic_power( ) +def _fail_multinode_adapter(message: str, *, require_power: bool) -> int: + print(f"[agentx_power] {message}", file=sys.stderr) + return 1 if require_power else 0 + + +def _positive_concurrencies(raw: str) -> list[int] | None: + tokens = raw.split() + if not tokens or any(not token.isdecimal() for token in tokens): + return None + values = [int(token) for token in tokens] + if any(value <= 0 for value in values) or len(set(values)) != len(values): + return None + return values + + +def _multinode_window_contract( + *, + result_dir: Path, + concurrency: int, +) -> tuple[Path, Path, Path] | None: + """Resolve and validate the formal custom-benchmark window contract.""" + if isinstance(concurrency, bool) or not isinstance(concurrency, int) or concurrency <= 0: + return None + values = {name: os.environ.get(name, "") for name in _FORMAL_WINDOW_ENV} + if any(not value for value in values.values()): + return None + if values["SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE"] != "custom": + return None + measured = _positive_concurrencies( + values["SRT_MEASUREMENT_WINDOW_CONCURRENCIES"] + ) + if measured is None or concurrency not in measured: + return None + + window_dir = Path(values["SRT_MEASUREMENT_WINDOW_DIR"]) + result_root = Path(values["SRT_MEASUREMENT_WINDOW_RESULT_ROOT"]) + if ( + not window_dir.is_absolute() + or not result_root.is_absolute() + or not result_dir.is_absolute() + or not window_dir.is_dir() + or not result_root.is_dir() + or not result_dir.is_dir() + ): + return None + try: + window_dir.resolve().relative_to(result_root.resolve()) + relative_result_dir = result_dir.resolve().relative_to(result_root.resolve()) + except (OSError, ValueError): + return None + + stem = f"agentic_power_concurrency_{concurrency}" + formal_result = result_dir / f"{stem}.json" + formal_window = window_dir / f"{stem}.json" + result_path = relative_result_dir / formal_result.name + return formal_result, formal_window, result_path + + +def _window_payload( + *, + result_path: Path, + concurrency: int, + status: str, + start: float, + end: float | None, + duration: float | None, +) -> dict[str, Any]: + return { + "schema_version": 1, + "benchmark_type": "custom", + "result_path": result_path.as_posix(), + "concurrency": concurrency, + "benchmark_start_time_unix": start, + "benchmark_end_time_unix": end, + "duration": duration, + "clock_source": "head_node_unix_clock", + "status": status, + "reason": None, + } + + +def write_multinode_power_window( + *, + result_dir: Path, + concurrency: int, + state: str, + require_power: bool = False, +) -> int: + """Publish one AgentX custom-benchmark formal window on the head clock.""" + contract = _multinode_window_contract( + result_dir=result_dir, + concurrency=concurrency, + ) + if contract is None: + return _fail_multinode_adapter( + "Invalid formal measurement-window contract for multinode AgentX", + require_power=require_power, + ) + formal_result, formal_window, result_path = contract + + if state == "running": + payload = _window_payload( + result_path=result_path, + concurrency=concurrency, + status="running", + start=time.time(), + end=None, + duration=None, + ) + try: + _write_json_atomic(formal_window, payload) + except OSError as exc: + return _fail_multinode_adapter( + f"Failed to write formal measurement-window contract: {exc}", + require_power=require_power, + ) + return 0 + + if state != "completed": + return _fail_multinode_adapter( + f"Invalid formal measurement-window state: {state}", + require_power=require_power, + ) + + boundary, reasons = build_power_window(result_dir) + if boundary is None: + return _fail_multinode_adapter( + "Failed to complete formal measurement-window contract: " + + ", ".join(reasons), + require_power=require_power, + ) + formal_result_payload = {"max_concurrency": concurrency, **boundary} + completed = _window_payload( + result_path=result_path, + concurrency=concurrency, + status="completed", + start=float(boundary["benchmark_start_time_unix"]), + end=float(boundary["benchmark_end_time_unix"]), + duration=float(boundary["duration"]), + ) + try: + # The collector validates the result referenced by a completed window. + # Publish that result first so it can never observe a completed window + # that points to a missing or partially written result. + _write_json_atomic(formal_result, formal_result_payload) + _write_json_atomic(formal_window, completed) + except OSError as exc: + return _fail_multinode_adapter( + f"Failed to write formal measurement-window contract: {exc}", + require_power=require_power, + ) + return 0 + + +def _record_multinode_adapter_failure( + *, + agg_result: Path, + validation_result: Path, + reasons: list[str], +) -> None: + aggregate = json.loads(agg_result.read_text(encoding="utf-8")) + if not isinstance(aggregate, dict): + raise ValueError("AgentX aggregate must be a JSON object") + for key in _ALL_POWER_METRIC_KEYS: + aggregate.pop(key, None) + aggregate["power_metric_schema_version"] = POWER_METRIC_SCHEMA_VERSION + aggregate["power_valid"] = 0 + aggregate.pop("power_invalid_reasons", None) + _write_json_atomic(agg_result, aggregate) + _write_json_atomic( + validation_result, + { + "power_valid": False, + "reasons": reasons, + "window_source": "aiperf_multinode_custom_benchmark", + }, + ) + + +def _formal_result_for_directory(result_dir: Path) -> Path | None: + candidates = [ + path + for path in result_dir.glob("agentic_power_concurrency_*.json") + if _MULTINODE_WINDOW_STEM_RE.fullmatch(path.stem) + ] + return candidates[0] if len(candidates) == 1 else None + + +def _gpu_count(value: Any) -> int | None: + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + return None + return value + + +def run_multinode_agentic_power( + *, + result_dir: Path, + agg_result: Path, + power_dir: Path, + logs_root: Path, + expected_producer_sha: str, + require_power: bool = False, +) -> int: + """Join one AgentX aggregate to the finalized central multinode package.""" + validation_result = result_dir / "power_validation.json" + reasons: list[str] = [] + try: + aggregate = json.loads(agg_result.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + aggregate = None + reasons.append("agentic_aggregate_invalid") + if not isinstance(aggregate, dict): + if not reasons: + reasons.append("agentic_aggregate_invalid") + aggregate = {} + + prefill_gpus = _gpu_count(aggregate.get("num_prefill_gpu")) + decode_gpus = _gpu_count(aggregate.get("num_decode_gpu")) + if ( + prefill_gpus is None + or decode_gpus is None + or prefill_gpus + decode_gpus <= 0 + ): + reasons.append("agentic_gpu_topology_invalid") + bench_result = _formal_result_for_directory(result_dir) + if bench_result is None: + reasons.append("formal_benchmark_result_missing") + try: + result_dir.resolve().relative_to(logs_root.resolve()) + except (OSError, ValueError): + reasons.append("agentic_result_outside_logs_root") + if _COMMIT_SHA_RE.fullmatch(expected_producer_sha) is None: + reasons.append("expected_producer_sha_invalid") + + if reasons: + try: + if agg_result.is_file(): + _record_multinode_adapter_failure( + agg_result=agg_result, + validation_result=validation_result, + reasons=reasons, + ) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print( + f"[agentx_power] Failed to record multinode adapter failure: {exc}", + file=sys.stderr, + ) + return _fail_multinode_adapter( + "Multinode AgentX power adaptation failed: " + ", ".join(reasons), + require_power=require_power, + ) + + assert prefill_gpus is not None + assert decode_gpus is not None + assert bench_result is not None + return run_multinode_power( + power_dir=power_dir, + bench_result=bench_result, + agg_result=agg_result, + prefill_gpus=prefill_gpus, + decode_gpus=decode_gpus, + expected_producer_sha=expected_producer_sha, + logs_root=logs_root, + validation_result=validation_result, + require_power=require_power, + ) + + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--result-dir", type=Path, required=True) - parser.add_argument("--agg-result", type=Path, required=True) + parser.add_argument("--agg-result", type=Path) parser.add_argument("--expected-num-gpus", type=int) + parser.add_argument("--write-multinode-window", choices=("running", "completed")) + parser.add_argument("--concurrency", type=int) + parser.add_argument("--power-dir", type=Path) + parser.add_argument("--logs-root", type=Path) + parser.add_argument("--expected-producer-sha") parser.add_argument( "--require-power", action="store_true", default=os.environ.get("REQUIRE_POWER", "").lower() in {"1", "true", "yes"}, ) args = parser.parse_args() + if args.write_multinode_window is not None: + if args.concurrency is None: + parser.error("--concurrency is required with --write-multinode-window") + return write_multinode_power_window( + result_dir=args.result_dir, + concurrency=args.concurrency, + state=args.write_multinode_window, + require_power=args.require_power, + ) + if args.power_dir is not None: + if args.agg_result is None or args.logs_root is None or args.expected_producer_sha is None: + parser.error( + "--agg-result, --logs-root, and --expected-producer-sha are required with --power-dir" + ) + return run_multinode_agentic_power( + result_dir=args.result_dir, + agg_result=args.agg_result, + power_dir=args.power_dir, + logs_root=args.logs_root, + expected_producer_sha=args.expected_producer_sha, + require_power=args.require_power, + ) + if args.agg_result is None: + parser.error("--agg-result is required for single-node AgentX power") return run_agentic_power( result_dir=args.result_dir, agg_result=args.agg_result, diff --git a/utils/agentic/aggregation/test_power_adapter.py b/utils/agentic/aggregation/test_power_adapter.py index e4a26c627a..c9a71e1e70 100644 --- a/utils/agentic/aggregation/test_power_adapter.py +++ b/utils/agentic/aggregation/test_power_adapter.py @@ -319,3 +319,249 @@ def test_run_agentic_power_records_adapter_failure_before_returning( validation = json.loads((result_dir / "power_validation.json").read_text()) assert validation["power_valid"] is False assert "incomplete_token_accounting" in validation["reasons"] + + +def _set_multinode_window_environment( + monkeypatch: pytest.MonkeyPatch, + *, + logs_root: Path, + concurrencies: str = "8 16", +) -> tuple[Path, Path]: + window_dir = logs_root / "power" / "windows" + result_root = logs_root + window_dir.mkdir(parents=True) + monkeypatch.setenv("SRT_MEASUREMENT_WINDOW_DIR", str(window_dir)) + monkeypatch.setenv("SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE", "custom") + monkeypatch.setenv("SRT_MEASUREMENT_WINDOW_CONCURRENCIES", concurrencies) + monkeypatch.setenv("SRT_MEASUREMENT_WINDOW_RESULT_ROOT", str(result_root)) + return window_dir, result_root + + +def test_multinode_window_writer_publishes_boundary_identical_result_last( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from utils.agentic.aggregation import power_adapter + + logs_root = tmp_path / "logs" + result_dir = _write_artifacts(logs_root / "agentic" / "conc_8") + window_dir, result_root = _set_multinode_window_environment( + monkeypatch, + logs_root=logs_root, + ) + writes: list[Path] = [] + original_write = power_adapter._write_json_atomic + + def record_write(path: Path, payload: dict) -> None: + writes.append(path) + original_write(path, payload) + + monkeypatch.setattr(power_adapter, "_write_json_atomic", record_write) + monkeypatch.setattr(power_adapter.time, "time", lambda: 1_700_000_000.0) + + assert power_adapter.write_multinode_power_window( + result_dir=result_dir, + concurrency=8, + state="running", + require_power=True, + ) == 0 + + stem = "agentic_power_concurrency_8" + formal_result = result_dir / f"{stem}.json" + formal_window = window_dir / f"{stem}.json" + running = json.loads(formal_window.read_text()) + assert running == { + "schema_version": 1, + "benchmark_type": "custom", + "result_path": formal_result.relative_to(result_root).as_posix(), + "concurrency": 8, + "benchmark_start_time_unix": 1_700_000_000.0, + "benchmark_end_time_unix": None, + "duration": None, + "clock_source": "head_node_unix_clock", + "status": "running", + "reason": None, + } + assert not formal_result.exists() + + assert power_adapter.write_multinode_power_window( + result_dir=result_dir, + concurrency=8, + state="completed", + require_power=True, + ) == 0 + + result_payload = json.loads(formal_result.read_text()) + completed = json.loads(formal_window.read_text()) + assert result_payload == { + "max_concurrency": 8, + "benchmark_start_time_unix": 1_700_000_001.0, + "benchmark_end_time_unix": 1_700_000_004.0, + "duration": 3.0, + "completed": 2, + "total_input_tokens": 300, + "total_output_tokens": 150, + } + assert completed["status"] == "completed" + assert completed["benchmark_start_time_unix"] == result_payload["benchmark_start_time_unix"] + assert completed["benchmark_end_time_unix"] == result_payload["benchmark_end_time_unix"] + assert completed["duration"] == result_payload["duration"] + assert writes[-2:] == [formal_result, formal_window] + + +@pytest.mark.parametrize( + ("environment", "require_power", "expected_exit"), + [ + ({}, False, 0), + ({}, True, 1), + ( + { + "SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE": "sa-bench", + "SRT_MEASUREMENT_WINDOW_CONCURRENCIES": "8", + }, + True, + 1, + ), + ({"SRT_MEASUREMENT_WINDOW_CONCURRENCIES": "16"}, True, 1), + ], +) +def test_multinode_window_writer_fails_closed_on_invalid_formal_environment( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], + environment: dict[str, str], + require_power: bool, + expected_exit: int, +): + from utils.agentic.aggregation.power_adapter import write_multinode_power_window + + logs_root = tmp_path / "logs" + result_dir = logs_root / "agentic" / "conc_8" + result_dir.mkdir(parents=True) + window_dir = logs_root / "power" / "windows" + window_dir.mkdir(parents=True) + defaults = { + "SRT_MEASUREMENT_WINDOW_DIR": str(window_dir), + "SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE": "custom", + "SRT_MEASUREMENT_WINDOW_CONCURRENCIES": "8 16", + "SRT_MEASUREMENT_WINDOW_RESULT_ROOT": str(logs_root), + } + defaults.update(environment) + if not environment: + defaults = {} + for name in ( + "SRT_MEASUREMENT_WINDOW_DIR", + "SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE", + "SRT_MEASUREMENT_WINDOW_CONCURRENCIES", + "SRT_MEASUREMENT_WINDOW_RESULT_ROOT", + ): + monkeypatch.delenv(name, raising=False) + for name, value in defaults.items(): + monkeypatch.setenv(name, value) + + exit_code = write_multinode_power_window( + result_dir=result_dir, + concurrency=8, + state="running", + require_power=require_power, + ) + + assert exit_code == expected_exit + assert "formal measurement-window contract" in capsys.readouterr().err + + +def test_multinode_aggregation_uses_central_package_and_aggregate_topology( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + from utils.agentic.aggregation import power_adapter + + logs_root = tmp_path / "logs" + result_dir = logs_root / "agentic" / "conc_8" + result_dir.mkdir(parents=True) + bench_result = result_dir / "agentic_power_concurrency_8.json" + bench_result.write_text(json.dumps({"max_concurrency": 8})) + agg_result = tmp_path / "agg_agentx_conc8.json" + agg_result.write_text( + json.dumps({"num_prefill_gpu": 16, "num_decode_gpu": 16}), + encoding="utf-8", + ) + power_dir = logs_root / "power" + power_dir.mkdir(parents=True) + calls: list[dict] = [] + + def fake_run(**kwargs) -> int: + calls.append(kwargs) + return 0 + + monkeypatch.setattr(power_adapter, "run_multinode_power", fake_run) + + exit_code = power_adapter.run_multinode_agentic_power( + result_dir=result_dir, + agg_result=agg_result, + power_dir=power_dir, + logs_root=logs_root, + expected_producer_sha="a1b8c7af10c00e5ea40074aebdc0086189bbc064", + require_power=True, + ) + + assert exit_code == 0 + assert calls == [ + { + "power_dir": power_dir, + "bench_result": bench_result, + "agg_result": agg_result, + "prefill_gpus": 16, + "decode_gpus": 16, + "expected_producer_sha": "a1b8c7af10c00e5ea40074aebdc0086189bbc064", + "logs_root": logs_root, + "validation_result": result_dir / "power_validation.json", + "require_power": True, + } + ] + + +@pytest.mark.parametrize( + "payload", + [ + {}, + {"num_prefill_gpu": True, "num_decode_gpu": 16}, + {"num_prefill_gpu": 16.5, "num_decode_gpu": 16}, + {"num_prefill_gpu": 0, "num_decode_gpu": 0}, + {"num_prefill_gpu": -1, "num_decode_gpu": 16}, + ], +) +def test_multinode_aggregation_rejects_invalid_aggregate_topology( + tmp_path: Path, + payload: dict, +): + from utils.agentic.aggregation.power_adapter import run_multinode_agentic_power + + logs_root = tmp_path / "logs" + result_dir = logs_root / "agentic" / "conc_8" + result_dir.mkdir(parents=True) + (result_dir / "agentic_power_concurrency_8.json").write_text( + json.dumps({"max_concurrency": 8}) + ) + agg_result = tmp_path / "agg.json" + stale_metrics = { + "avg_power_w": 999, + "prefill_avg_power_w": 999, + "decode_avg_power_w": 999, + "prefill_gpu_energy_j": 999, + "decode_gpu_energy_j": 999, + } + agg_result.write_text(json.dumps({**stale_metrics, **payload})) + + assert run_multinode_agentic_power( + result_dir=result_dir, + agg_result=agg_result, + power_dir=logs_root / "power", + logs_root=logs_root, + expected_producer_sha="a" * 40, + require_power=True, + ) == 1 + aggregate = json.loads(agg_result.read_text()) + assert aggregate["power_valid"] == 0 + assert aggregate["power_metric_schema_version"] == 2 + assert stale_metrics.keys().isdisjoint(aggregate) diff --git a/utils/agentic/aggregation/test_power_lifecycle.py b/utils/agentic/aggregation/test_power_lifecycle.py index 4877ab59e9..de0e32df36 100644 --- a/utils/agentic/aggregation/test_power_lifecycle.py +++ b/utils/agentic/aggregation/test_power_lifecycle.py @@ -22,10 +22,15 @@ def _run_lifecycle( is_multinode: bool = False, enable_power: bool = True, require_power: bool = False, + formal_multinode_power: bool = False, ) -> subprocess.CompletedProcess[str]: result_dir = tmp_path / "results" result_dir.mkdir() event_log = tmp_path / "events.log" + formal_window_dir = str(tmp_path / "power/windows") if formal_multinode_power else "" + formal_benchmark_type = "custom" if formal_multinode_power else "" + formal_concurrencies = "8 16" if formal_multinode_power else "" + formal_result_root = str(tmp_path) if formal_multinode_power else "" script = f""" source {str(BENCHMARK_LIB)!r} start_gpu_monitor() {{ @@ -72,6 +77,11 @@ def _run_lifecycle( IS_MULTINODE={'true' if is_multinode else 'false'} ENABLE_AGENTX_POWER={'1' if enable_power else '0'} REQUIRE_POWER={'1' if require_power else '0'} +CONC=8 +SRT_MEASUREMENT_WINDOW_DIR={formal_window_dir!r} +SRT_MEASUREMENT_WINDOW_BENCHMARK_TYPE={formal_benchmark_type!r} +SRT_MEASUREMENT_WINDOW_CONCURRENCIES={formal_concurrencies!r} +SRT_MEASUREMENT_WINDOW_RESULT_ROOT={formal_result_root!r} set +e run_agentic_replay_and_write_outputs {str(result_dir)!r} rc=$? @@ -147,6 +157,44 @@ def test_multinode_and_explicit_opt_out_skip_local_power( assert not any(event.startswith("adapter:") for event in events) +def test_multinode_formal_window_wraps_replay_without_local_monitor(tmp_path: Path): + result = _run_lifecycle( + tmp_path, + is_multinode=True, + formal_multinode_power=True, + require_power=True, + ) + + assert result.returncode == 0, result.stderr + events = _events(tmp_path) + assert not any(event.startswith("monitor-") for event in events) + adapters = [event for event in events if event.startswith("adapter:")] + assert len(adapters) == 2 + assert "--write-multinode-window running" in adapters[0] + assert "--write-multinode-window completed" in adapters[1] + assert "--concurrency 8" in adapters[0] + assert "--require-power" in adapters[0] + assert events.index(adapters[0]) < events.index("replay") + assert events.index("aggregate") < events.index(adapters[1]) + captured_offset = (tmp_path / "results/agentic_power_timezone_offset.txt").read_text().strip() + assert re.fullmatch(r"[+-]\d{4}", captured_offset) + + +def test_multinode_formal_window_is_left_running_when_replay_is_interrupted(tmp_path: Path): + result = _run_lifecycle( + tmp_path, + replay_rc=143, + is_multinode=True, + formal_multinode_power=True, + require_power=True, + ) + + assert result.returncode == 143, result.stderr + adapters = [event for event in _events(tmp_path) if event.startswith("adapter:")] + assert len(adapters) == 1 + assert "--write-multinode-window running" in adapters[0] + + def test_shared_lifecycle_installs_idempotent_signal_cleanup(): benchmark_lib = BENCHMARK_LIB.read_text() diff --git a/utils/aggregate_power_multinode.py b/utils/aggregate_power_multinode.py index 685cd4e65f..dacbaba903 100644 --- a/utils/aggregate_power_multinode.py +++ b/utils/aggregate_power_multinode.py @@ -38,14 +38,24 @@ from dataclasses import dataclass, field from pathlib import Path, PurePosixPath -from aggregate_power import ( - POWER_METRIC_SCHEMA_VERSION, - BenchmarkData, - _append_reason, - _integrate_device, - _load_benchmark_data, - _write_json_atomic, -) +try: + from .aggregate_power import ( + POWER_METRIC_SCHEMA_VERSION, + BenchmarkData, + _append_reason, + _integrate_device, + _load_benchmark_data, + _write_json_atomic, + ) +except ImportError: # Direct execution: python utils/aggregate_power_multinode.py + from aggregate_power import ( + POWER_METRIC_SCHEMA_VERSION, + BenchmarkData, + _append_reason, + _integrate_device, + _load_benchmark_data, + _write_json_atomic, + ) # --- srt-slurm dcgm-power v1 wire contract (mirrored constants) ------------- diff --git a/utils/test_aggregate_power_multinode.py b/utils/test_aggregate_power_multinode.py index 16e77db51b..cc469e4885 100644 --- a/utils/test_aggregate_power_multinode.py +++ b/utils/test_aggregate_power_multinode.py @@ -284,6 +284,66 @@ def test_strict_mode_passes_on_valid_package(self, tmp_path): assert pkg.run(require_power=True) == 0 assert pkg.agg()["power_valid"] == 1 + def test_agentx_adapter_consumes_a_real_custom_benchmark_package(self, tmp_path): + from utils.agentic.aggregation.power_adapter import run_multinode_agentic_power + + pkg = build_package(tmp_path) + result_dir = pkg.logs_root / "agentic" / "conc_4" + result_dir.mkdir(parents=True) + stem = "agentic_power_concurrency_4" + formal_result = result_dir / f"{stem}.json" + pkg.original_result.replace(formal_result) + + old_window = pkg.windows_dir / f"{RESULT_STEM}.json" + window = json.loads(old_window.read_text()) + window.update( + { + "benchmark_type": "custom", + "result_path": f"agentic/conc_4/{stem}.json", + } + ) + old_window.unlink() + (pkg.windows_dir / f"{stem}.json").write_text(json.dumps(window, indent=2)) + + manifest_path = pkg.power_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["expected_windows"] = [ + {"benchmark_type": "custom", "concurrency": 4} + ] + manifest["window_validations"][0].update( + { + "benchmark_type": "custom", + "window_file": f"windows/{stem}.json", + } + ) + manifest_path.write_text(json.dumps(manifest, indent=2)) + pkg.agg_result.write_text( + json.dumps( + { + "hw": "h200", + "conc": 4, + "num_prefill_gpu": 2, + "num_decode_gpu": 2, + } + ) + ) + + assert run_multinode_agentic_power( + result_dir=result_dir, + agg_result=pkg.agg_result, + power_dir=pkg.power_dir, + logs_root=pkg.logs_root, + expected_producer_sha=PRODUCER_SHA, + require_power=True, + ) == 0 + + agg = pkg.agg() + assert agg["power_valid"] == 1 + assert agg["prefill_avg_power_w"] == 400.0 + assert agg["decode_avg_power_w"] == 300.0 + validation = json.loads((result_dir / "power_validation.json").read_text()) + assert validation["selected_window"]["window_file"] == f"windows/{stem}.json" + def test_trapezoid_matches_hand_computed_ramp(self, tmp_path): # node-d/0 ramps linearly 300 -> 364 W across the samples; the # trapezoid over [1000, 1060] must equal the analytic integral diff --git a/utils/test_h200_power_official_contract.py b/utils/test_h200_power_official_contract.py new file mode 100644 index 0000000000..9b0893cc31 --- /dev/null +++ b/utils/test_h200_power_official_contract.py @@ -0,0 +1,129 @@ +"""Static contract for the H200 multinode AgentX dcgm-power infrastructure.""" + +from __future__ import annotations + +import re +import subprocess +import sys +from pathlib import Path + +import yaml + +REPO_ROOT = Path(__file__).resolve().parents[1] +LAUNCHER = REPO_ROOT / "runners/launch_h200-dgxc-slurm.sh" +HELPER = REPO_ROOT / "runners/inject_srt_power_concurrencies.py" +RECIPE = REPO_ROOT / ( + "benchmarks/multi_node/srt-slurm-recipes/sglang/glm5.2/agentic/" + "disagg-h200-2p2d-pcp8-tp8-dp8-mtp.yaml" +) +WORKFLOW = REPO_ROOT / ".github/workflows/test-process-result.yml" + +PRODUCER_SHA = "a1b8c7af10c00e5ea40074aebdc0086189bbc064" +PRODUCER_URL = "https://github.com/edwingao28/srt-slurm.git" + + +def test_h200_launcher_detects_recipe_opt_in_and_rejects_unvalidated_lanes(): + launcher = LAUNCHER.read_text(encoding="utf-8") + + assert "USES_DCGM_POWER=0" in launcher + assert '_RECIPE_REL="${CONFIG_FILE%%:*}"' in launcher + assert ( + '_RECIPE_SRC="$GITHUB_WORKSPACE/benchmarks/multi_node/srt-slurm-recipes/${_RECIPE_REL#recipes/}"' + in launcher + ) + assert "/^telemetry:/ { t = 1; next }" in launcher + assert "t && /^ provider: dcgm-power$/ { p = 1 }" in launcher + assert "t && /^ enabled: true$/ { e = 1 }" in launcher + assert "USES_DCGM_POWER=1" in launcher + assert '"$IS_AGENTIC" != "1"' in launcher + assert '"$FRAMEWORK" != "dynamo-sglang"' in launcher + assert '"$MODEL_PREFIX" != "glm5.2"' in launcher + assert '"$PRECISION" != "fp8"' in launcher + + +def test_glm_agentx_uses_exact_fork_sha_and_other_clone_lanes_are_unchanged(): + launcher = LAUNCHER.read_text(encoding="utf-8") + + assert f'POWER_SRT_SLURM_URL="{PRODUCER_URL}"' in launcher + assert f'POWER_SRT_SLURM_PIN="{PRODUCER_SHA}"' in launcher + assert 'git clone "$POWER_SRT_SLURM_URL" "$SRT_REPO_DIR"' in launcher + assert 'git checkout "$POWER_SRT_SLURM_PIN" || exit 1' in launcher + assert 'test "$(git rev-parse HEAD)" = "$POWER_SRT_SLURM_PIN"' in launcher + assert 'git rev-parse HEAD > "$GITHUB_WORKSPACE/power-producer-sha.txt"' in launcher + + # These pre-existing branches are deliberately outside PR-A. + assert "https://github.com/functionstackx/srt-slurm-nv.git" in launcher + assert "df5baa93f4caf5169dea2a4236ad2cc742fe40e7" in launcher + assert "git clone --branch v1.0.38 --single-branch https://github.com/NVIDIA/srt-slurm.git" in launcher + + +def test_power_lane_provisions_exporter_and_injects_container_mapping(): + launcher = LAUNCHER.read_text(encoding="utf-8") + + assert 'DCGM_EXPORTER_IMAGE="nvcr.io/nvidia/k8s/dcgm-exporter:4.6.0-4.8.3-distroless"' in launcher + assert 'DCGM_EXPORTER_SQSH="/data/gharunners/containers/' in launcher + assert 'unsquashfs -l "$DCGM_EXPORTER_SQSH"' in launcher + assert 'sha256sum "$DCGM_EXPORTER_SQSH" > "$GITHUB_WORKSPACE/exporter-image.sha256"' in launcher + assert '"/^ nginx-sqsh:/a' in launcher + assert 'dcgm-exporter: ${DCGM_EXPORTER_SQSH}" srtslurm.yaml' in launcher + assert 'grep -q "^ dcgm-exporter: " srtslurm.yaml ||' in launcher + + +def test_launcher_injects_exact_matrix_concurrencies_and_finalizes_each_result(): + launcher = LAUNCHER.read_text(encoding="utf-8") + + assert str(HELPER.relative_to(REPO_ROOT)) in launcher + assert 'read -r -a POWER_CONCURRENCIES <<< "$CONC_LIST"' in launcher + assert '"${POWER_CONCURRENCIES[@]}"' in launcher + assert '--power-dir "$POWER_LOGS_ROOT/power"' in launcher + assert '--logs-root "$POWER_LOGS_ROOT"' in launcher + assert '--expected-producer-sha "$POWER_SRT_SLURM_PIN"' in launcher + assert '"$GITHUB_WORKSPACE/${RESULT_FILENAME}_conc${concurrency}.json"' in launcher + assert '"$POWER_LOGS_ROOT/agentic/conc_${concurrency}"' in launcher + assert 'cp "$GITHUB_WORKSPACE/exporter-image.sha256" "$LOGS_DIR/power/exporter-image.sha256"' in launcher + assert 'cp "$GITHUB_WORKSPACE/power-producer-sha.txt" "$LOGS_DIR/power/power-producer-sha.txt"' in launcher + + +def test_pr_a_keeps_glm_recipe_power_disabled_until_followup_pr_b(): + recipe = yaml.safe_load(RECIPE.read_text(encoding="utf-8")) + + assert "telemetry" not in recipe + assert "concurrencies" not in recipe["benchmark"] + assert recipe["benchmark"]["type"] == "custom" + assert recipe["infra"]["etcd_nats_dedicated_node"] is True + + +def test_process_result_ci_covers_h200_power_files(): + workflow = WORKFLOW.read_text(encoding="utf-8") + + for path in ( + "benchmarks/multi_node/srt-slurm-recipes/sglang/glm5.2/agentic/disagg-h200-2p2d-pcp8-tp8-dp8-mtp.yaml", + "runners/launch_h200-dgxc-slurm.sh", + "runners/inject_srt_power_concurrencies.py", + "utils/test_h200_power_official_contract.py", + "utils/test_inject_srt_power_concurrencies.py", + ): + assert f"- '{path}'" in workflow + pytest_command = workflow.split("- name: Run pytest", 1)[1] + assert "test_h200_power_official_contract.py" in pytest_command + assert "test_inject_srt_power_concurrencies.py" in pytest_command + + +def test_producer_pin_is_immutable_and_only_declared_once(): + launcher = LAUNCHER.read_text(encoding="utf-8") + + assert re.fullmatch(r"[0-9a-f]{40}", PRODUCER_SHA) + assert launcher.count(PRODUCER_SHA) == 1 + assert launcher.count(PRODUCER_URL) == 1 + + +def test_power_adapter_module_is_importable_from_launcher_working_directory(): + result = subprocess.run( + [sys.executable, "-m", "utils.agentic.aggregation.power_adapter", "--help"], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + + assert result.returncode == 0, result.stderr diff --git a/utils/test_inject_srt_power_concurrencies.py b/utils/test_inject_srt_power_concurrencies.py new file mode 100644 index 0000000000..990b075896 --- /dev/null +++ b/utils/test_inject_srt_power_concurrencies.py @@ -0,0 +1,56 @@ +"""Tests for runtime injection of AgentX power measurement points.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest +import yaml + + +def test_injects_positive_unique_concurrencies_without_adding_telemetry(tmp_path: Path): + from runners.inject_srt_power_concurrencies import inject_concurrencies + + recipe_path = tmp_path / "recipe.yaml" + recipe_path.write_text( + "name: agentx\nbenchmark:\n type: custom\n command: run-agentx\n", + encoding="utf-8", + ) + + inject_concurrencies(recipe_path, [8, 16, 32]) + + recipe = yaml.safe_load(recipe_path.read_text(encoding="utf-8")) + assert recipe["benchmark"]["concurrencies"] == [8, 16, 32] + assert "telemetry" not in recipe + assert recipe["benchmark"]["command"] == "run-agentx" + + +@pytest.mark.parametrize( + "concurrencies", + [[], [0], [-1], [8, 8], [True], [8, 1.5]], +) +def test_rejects_invalid_concurrency_contract( + tmp_path: Path, + concurrencies: list, +): + from runners.inject_srt_power_concurrencies import inject_concurrencies + + recipe_path = tmp_path / "recipe.yaml" + recipe_path.write_text("benchmark:\n type: custom\n", encoding="utf-8") + + with pytest.raises(ValueError, match="positive unique integers"): + inject_concurrencies(recipe_path, concurrencies) + + +@pytest.mark.parametrize( + "recipe_text", + ["[]\n", "name: missing-benchmark\n", "benchmark: []\n"], +) +def test_rejects_recipe_without_benchmark_mapping(tmp_path: Path, recipe_text: str): + from runners.inject_srt_power_concurrencies import inject_concurrencies + + recipe_path = tmp_path / "recipe.yaml" + recipe_path.write_text(recipe_text, encoding="utf-8") + + with pytest.raises(ValueError, match="benchmark mapping"): + inject_concurrencies(recipe_path, [8]) From 8073ca3fe3982146ca93ad6687b3c3aac8703f5c Mon Sep 17 00:00:00 2001 From: Wenyao Gao Date: Wed, 19 Aug 2026 19:17:45 -0700 Subject: [PATCH 2/2] fix(power): pin finalized multinode producer --- runners/launch_h200-dgxc-slurm.sh | 2 +- utils/test_h200_power_official_contract.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/runners/launch_h200-dgxc-slurm.sh b/runners/launch_h200-dgxc-slurm.sh index 275075d5d9..1a55c794eb 100755 --- a/runners/launch_h200-dgxc-slurm.sh +++ b/runners/launch_h200-dgxc-slurm.sh @@ -11,7 +11,7 @@ AIPERF_MMAP_CACHE_HOST_PATH="${AIPERF_MMAP_CACHE_HOST_PATH:-/home/sa-shared/ghar # intentionally long-lived; update the SHA only after reviewing a new fork # commit and re-running the H200 hardware gate. POWER_SRT_SLURM_URL="https://github.com/edwingao28/srt-slurm.git" -POWER_SRT_SLURM_PIN="a1b8c7af10c00e5ea40074aebdc0086189bbc064" +POWER_SRT_SLURM_PIN="e5c837f06a362dc888dfea2ee588e9f19c298270" set -x diff --git a/utils/test_h200_power_official_contract.py b/utils/test_h200_power_official_contract.py index 9b0893cc31..8aa2d18d89 100644 --- a/utils/test_h200_power_official_contract.py +++ b/utils/test_h200_power_official_contract.py @@ -18,7 +18,7 @@ ) WORKFLOW = REPO_ROOT / ".github/workflows/test-process-result.yml" -PRODUCER_SHA = "a1b8c7af10c00e5ea40074aebdc0086189bbc064" +PRODUCER_SHA = "e5c837f06a362dc888dfea2ee588e9f19c298270" PRODUCER_URL = "https://github.com/edwingao28/srt-slurm.git"